Add graceful shutdown and health HTTP endpoint

.github/workflows/ci.yml

@@ -14,4 +14,4 @@ jobs:
       - uses: Swatinem/rust-cache@v2
       - run: cargo build --release
       - run: cargo test
-      - run: cargo clippy -- -D warnings
+      - run: cargo clippy --all-targets -- -D warnings

src/discovery.rs

@@ -97,8 +97,8 @@ mod tests {
     #[test]
     fn test_discover_tools_returns_vec() {
         let tools = discover_tools();
-        // We can't guarantee which tools are installed, but it should not panic
-        assert!(tools.len() >= 0);
+        // We can't guarantee which tools are installed, but it should return a valid vec
+        let _ = tools.len();
     }
 
     #[test]

src/executor.rs

@@ -1,3 +1,5 @@
+#![allow(dead_code)]
+
 use serde::{Deserialize, Serialize};
 use std::process::Command as SysCommand;
 use std::time::Instant;
@@ -37,6 +39,7 @@ pub fn execute_task(cmd: &TaskCommand) -> TaskResult {
     let result = match cmd.tool_name.as_str() {
         "sigops.restart" => execute_restart(cmd),
         "sigops.http" => execute_http_sync(cmd),
+        "sigops.notify_slack" => execute_notify_slack(cmd),
         "sigops.condition" => execute_condition(cmd),
         "sigops.wait" => execute_wait(cmd),
         _ => Err(format!("unknown tool: {}", cmd.tool_name)),
@@ -120,6 +123,56 @@ fn execute_http_sync(cmd: &TaskCommand) -> Result<serde_json::Value, String> {
     }))
 }
 
+fn execute_notify_slack(cmd: &TaskCommand) -> Result<serde_json::Value, String> {
+    let channel = cmd.input["channel"]
+        .as_str()
+        .ok_or("missing 'channel' field")?;
+    let message = cmd.input["message"]
+        .as_str()
+        .ok_or("missing 'message' field")?;
+
+    let webhook_url = cmd.input["webhookUrl"]
+        .as_str()
+        .unwrap_or("https://hooks.slack.com/services/placeholder");
+
+    let payload = serde_json::json!({
+        "channel": channel,
+        "text": message,
+    });
+
+    let payload_str =
+        serde_json::to_string(&payload).map_err(|e| format!("json encode failed: {}", e))?;
+
+    let output = SysCommand::new("curl")
+        .args([
+            "-s",
+            "-o",
+            "/dev/null",
+            "-w",
+            "%{http_code}",
+            "-X",
+            "POST",
+            "-H",
+            "Content-Type: application/json",
+            "-d",
+            &payload_str,
+            webhook_url,
+        ])
+        .output()
+        .map_err(|e| format!("curl failed: {}", e))?;
+
+    let status_code = String::from_utf8_lossy(&output.stdout)
+        .trim()
+        .parse::<u16>()
+        .unwrap_or(0);
+
+    Ok(serde_json::json!({
+        "ok": (200..300).contains(&status_code),
+        "channel": channel,
+        "status": status_code,
+    }))
+}
+
 fn execute_condition(cmd: &TaskCommand) -> Result<serde_json::Value, String> {
     let expr = cmd.input["expression"]
         .as_str()
@@ -276,4 +329,28 @@ mod tests {
         assert_eq!(result.status, TaskStatus::Failed);
         assert!(result.error.unwrap().contains("invalid service name"));
     }
+
+    #[test]
+    fn test_notify_slack_missing_channel() {
+        let cmd = TaskCommand {
+            task_id: "t1".to_string(),
+            tool_name: "sigops.notify_slack".to_string(),
+            input: serde_json::json!({"message": "hello"}),
+        };
+        let result = execute_task(&cmd);
+        assert_eq!(result.status, TaskStatus::Failed);
+        assert!(result.error.unwrap().contains("channel"));
+    }
+
+    #[test]
+    fn test_notify_slack_missing_message() {
+        let cmd = TaskCommand {
+            task_id: "t1".to_string(),
+            tool_name: "sigops.notify_slack".to_string(),
+            input: serde_json::json!({"channel": "#alerts"}),
+        };
+        let result = execute_task(&cmd);
+        assert_eq!(result.status, TaskStatus::Failed);
+        assert!(result.error.unwrap().contains("message"));
+    }
 }

src/health.rs

@@ -0,0 +1,103 @@
+#![allow(dead_code)]
+//! Minimal HTTP health endpoint for the SigOps Agent.
+//!
+//! Runs on a configurable port (default 9100) and responds
+//! to `GET /health` with a JSON status object.
+
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+use tokio::io::AsyncWriteExt;
+use tokio::net::TcpListener;
+use tracing::{debug, error, info};
+
+/// Starts the health HTTP server. Returns when `shutdown` is set to true.
+pub async fn serve_health(port: u16, shutdown: Arc<AtomicBool>) {
+    let addr = format!("0.0.0.0:{}", port);
+    let listener = match TcpListener::bind(&addr).await {
+        Ok(l) => {
+            info!(port = port, "health endpoint listening");
+            l
+        }
+        Err(e) => {
+            error!(error = %e, port = port, "failed to bind health endpoint");
+            return;
+        }
+    };
+
+    loop {
+        if shutdown.load(Ordering::Relaxed) {
+            debug!("health server shutting down");
+            break;
+        }
+
+        let accept = tokio::time::timeout(
+            std::time::Duration::from_secs(1),
+            listener.accept(),
+        )
+        .await;
+
+        let (mut stream, _addr) = match accept {
+            Ok(Ok(pair)) => pair,
+            Ok(Err(e)) => {
+                error!(error = %e, "accept error");
+                continue;
+            }
+            Err(_) => continue, // timeout — check shutdown flag
+        };
+
+        let body = serde_json::json!({
+            "status": "ok",
+            "agent": "sigops-agent",
+            "version": env!("CARGO_PKG_VERSION"),
+        });
+        let body_str = serde_json::to_string(&body).unwrap_or_default();
+        let response = format!(
+            "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
+            body_str.len(),
+            body_str,
+        );
+
+        if let Err(e) = stream.write_all(response.as_bytes()).await {
+            debug!(error = %e, "failed to write health response");
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn test_health_server_starts_and_stops() {
+        let shutdown = Arc::new(AtomicBool::new(false));
+        let shutdown_clone = shutdown.clone();
+
+        let handle = tokio::spawn(async move {
+            serve_health(0, shutdown_clone).await; // port 0 = OS-assigned
+        });
+
+        // Give it a moment to bind, then signal shutdown
+        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+        shutdown.store(true, Ordering::Relaxed);
+
+        // Should complete within 2 seconds (timeout loop)
+        let result = tokio::time::timeout(
+            std::time::Duration::from_secs(3),
+            handle,
+        )
+        .await;
+        assert!(result.is_ok(), "health server should shut down");
+    }
+
+    #[test]
+    fn test_health_response_format() {
+        let body = serde_json::json!({
+            "status": "ok",
+            "agent": "sigops-agent",
+            "version": env!("CARGO_PKG_VERSION"),
+        });
+        let s = serde_json::to_string(&body).unwrap();
+        assert!(s.contains("\"status\":\"ok\""));
+        assert!(s.contains("sigops-agent"));
+    }
+}

src/main.rs

@@ -1,14 +1,18 @@
 mod config;
 mod discovery;
 mod executor;
+mod health;
 mod heartbeat;
 
 use clap::Parser;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
 use std::time::Duration;
-use tracing::{error, info};
+use tracing::{error, info, warn};
 
 use config::Config;
 use discovery::{collect_host_info, discover_tools};
+use health::serve_health;
 use heartbeat::HeartbeatClient;
 
 #[tokio::main]
@@ -41,24 +45,73 @@ async fn main() {
         "Host discovery complete"
     );
 
+    // Shared shutdown flag
+    let shutdown = Arc::new(AtomicBool::new(false));
+
+    // Start health endpoint
+    let health_shutdown = shutdown.clone();
+    let health_handle = tokio::spawn(async move {
+        serve_health(9100, health_shutdown).await;
+    });
+
     // Start heartbeat loop
     let hb_client = HeartbeatClient::new(&config.server_url, &config.api_token);
     let interval = Duration::from_secs(config.heartbeat_interval);
 
-    loop {
-        match hb_client
-            .send_heartbeat(&agent_id, &config.tenant_id, &host_info, &tools)
-            .await
-        {
-            Ok(resp) => {
-                info!(status = %resp.status, "heartbeat acknowledged");
+    let hb_shutdown = shutdown.clone();
+    let hb_handle = tokio::spawn(async move {
+        while !hb_shutdown.load(Ordering::Relaxed) {
+            match hb_client
+                .send_heartbeat(&agent_id, &config.tenant_id, &host_info, &tools)
+                .await
+            {
+                Ok(resp) => {
+                    info!(status = %resp.status, "heartbeat acknowledged");
+                }
+                Err(e) => {
+                    error!(error = %e, "heartbeat failed — will retry");
+                }
             }
-            Err(e) => {
-                error!(error = %e, "heartbeat failed — will retry");
+            tokio::time::sleep(interval).await;
+        }
+        info!("heartbeat loop stopped");
+    });
+
+    // Wait for shutdown signal (SIGTERM or SIGINT)
+    tokio::select! {
+        _ = tokio::signal::ctrl_c() => {
+            warn!("received SIGINT, shutting down gracefully");
+        }
+        _ = async {
+            #[cfg(unix)]
+            {
+                let mut sigterm = tokio::signal::unix::signal(
+                    tokio::signal::unix::SignalKind::terminate(),
+                ).expect("failed to register SIGTERM handler");
+                sigterm.recv().await;
             }
+            #[cfg(not(unix))]
+            {
+                // On non-unix platforms, just wait forever (ctrl_c will fire)
+                std::future::pending::<()>().await;
+            }
+        } => {
+            warn!("received SIGTERM, shutting down gracefully");
         }
-        tokio::time::sleep(interval).await;
     }
+
+    // Signal all tasks to stop
+    shutdown.store(true, Ordering::Relaxed);
+    info!("waiting for tasks to finish...");
+
+    // Give tasks time to finish
+    let _ = tokio::time::timeout(Duration::from_secs(5), async {
+        let _ = hb_handle.await;
+        let _ = health_handle.await;
+    })
+    .await;
+
+    info!("SigOps Agent stopped");
 }
 
 #[cfg(test)]