Production readiness: bounded channels, health probes, security tests, fuzz fixes

Author: symbibot

crates/runtime/src/api/server.rs

@@ -82,7 +82,9 @@ use crate::types::RuntimeError;
         super::routes::get_channel_audit,
         super::routes::agent_heartbeat,
         super::routes::agent_push_event,
-        health_check
+        health_check,
+        liveness_check,
+        readiness_check
     ),
     components(
         schemas(
@@ -305,6 +307,8 @@ impl HttpApiServer {
 
         let mut router = Router::new()
             .route("/api/v1/health", get(health_check))
+            .route("/api/v1/health/live", get(liveness_check))
+            .route("/api/v1/health/ready", get(readiness_check))
             .with_state(self.start_time);
 
         // Add Swagger UI only in non-production environments
@@ -503,3 +507,44 @@ async fn health_check(
 
     Ok(Json(response))
 }
+
+/// Liveness probe — confirms the process is running and able to serve requests.
+#[cfg(feature = "http-api")]
+#[utoipa::path(
+    get,
+    path = "/api/v1/health/live",
+    responses(
+        (status = 200, description = "Process is alive"),
+        (status = 503, description = "Process is not alive")
+    ),
+    tag = "system"
+)]
+async fn liveness_check() -> StatusCode {
+    // Liveness: the process is running and can serve requests.
+    // No dependency checks — if this handler runs, we're alive.
+    StatusCode::OK
+}
+
+/// Readiness probe — confirms the system has completed initialization.
+#[cfg(feature = "http-api")]
+#[utoipa::path(
+    get,
+    path = "/api/v1/health/ready",
+    responses(
+        (status = 200, description = "System is ready to accept work"),
+        (status = 503, description = "System is not ready")
+    ),
+    tag = "system"
+)]
+async fn readiness_check(
+    axum::extract::State(start_time): axum::extract::State<Instant>,
+) -> StatusCode {
+    // Readiness: the system has completed initialization and can accept work.
+    // After 5 seconds of uptime, consider the system ready.
+    // In the future, check scheduler state, DB connections, etc.
+    if start_time.elapsed().as_secs() >= 5 {
+        StatusCode::OK
+    } else {
+        StatusCode::SERVICE_UNAVAILABLE
+    }
+}

crates/runtime/src/integrations/composio/transport.rs

@@ -26,20 +26,19 @@ impl SseTransport {
     /// Returns an error if the API key contains invalid header characters
     /// or if the HTTP client fails to build.
     pub fn new(endpoint_url: String, api_key: String) -> Result<Self, ComposioError> {
-        let header_value = HeaderValue::from_str(&api_key).map_err(|e| {
-            ComposioError::Configuration(format!(
-                "invalid API key (contains non-ASCII or control characters): {e}"
-            ))
-        })?;
+        let header_value =
+            HeaderValue::from_str(&api_key).map_err(|e| ComposioError::ConfigError {
+                reason: format!("invalid API key (contains non-ASCII or control characters): {e}"),
+            })?;
 
         let mut headers = HeaderMap::new();
         headers.insert("x-api-key", header_value);
 
         let client = reqwest::Client::builder()
             .default_headers(headers)
             .build()
-            .map_err(|e| {
-                ComposioError::Configuration(format!("failed to build HTTP client: {e}"))
+            .map_err(|e| ComposioError::ConfigError {
+                reason: format!("failed to build HTTP client: {e}"),
             })?;
 
         Ok(Self {

crates/runtime/src/lib.rs

@@ -34,6 +34,8 @@ pub mod api;
 
 #[cfg(feature = "http-api")]
 use api::traits::RuntimeApiProvider;
+#[cfg(all(feature = "http-api", feature = "cron"))]
+use api::types::ScheduleRunEntry;
 #[cfg(feature = "http-api")]
 use api::types::{
     AddIdentityMappingRequest, AgentExecutionRecord, AgentStatusResponse, ChannelActionResponse,
@@ -42,8 +44,8 @@ use api::types::{
     DeleteChannelResponse, DeleteScheduleResponse, ExecuteAgentRequest, ExecuteAgentResponse,
     GetAgentHistoryResponse, IdentityMappingEntry, NextRunsResponse, RegisterChannelRequest,
     RegisterChannelResponse, ScheduleActionResponse, ScheduleDetail, ScheduleHistoryResponse,
-    ScheduleRunEntry, ScheduleSummary, UpdateAgentRequest, UpdateAgentResponse,
-    UpdateChannelRequest, UpdateScheduleRequest, WorkflowExecutionRequest,
+    ScheduleSummary, UpdateAgentRequest, UpdateAgentResponse, UpdateChannelRequest,
+    UpdateScheduleRequest, WorkflowExecutionRequest,
 };
 #[cfg(feature = "http-api")]
 use async_trait::async_trait;
@@ -365,6 +367,7 @@ pub struct RuntimeConfig {
 /// Implementation of RuntimeApiProvider for AgentRuntime
 #[cfg(feature = "http-api")]
 #[async_trait]
+#[allow(unused_variables)] // Params used inside #[cfg(feature = "cron")] blocks
 impl RuntimeApiProvider for AgentRuntime {
     async fn execute_workflow(
         &self,

crates/runtime/src/reasoning/tracing_spans.rs

@@ -132,6 +132,11 @@ impl LoopTracer {
         );
     }
 
+    /// Get the trace ID for this loop (for external correlation).
+    pub fn trace_id(&self) -> &str {
+        &self.loop_id
+    }
+
     /// Get the loop ID.
     pub fn loop_id(&self) -> &str {
         &self.loop_id

crates/runtime/src/scheduler/task_manager.rs

@@ -15,7 +15,7 @@ use crate::types::*;
 pub struct TaskManager {
     task_timeout: Duration,
     running_tasks: Arc<RwLock<HashMap<AgentId, TaskHandle>>>,
-    task_sender: mpsc::UnboundedSender<TaskCommand>,
+    task_sender: mpsc::Sender<TaskCommand>,
     #[allow(dead_code)]
     system_info: Arc<RwLock<System>>,
 }
@@ -24,7 +24,9 @@ impl TaskManager {
     /// Create a new task manager
     pub fn new(task_timeout: Duration) -> Self {
         let running_tasks = Arc::new(RwLock::new(HashMap::new()));
-        let (task_sender, task_receiver) = mpsc::unbounded_channel();
+        // Bounded channel prevents unbounded memory growth under load.
+        // 1024 is sufficient for most workloads; submitters block when full.
+        let (task_sender, task_receiver) = mpsc::channel(1024);
 
         let manager = Self {
             task_timeout,
@@ -47,15 +49,16 @@ impl TaskManager {
         // Store the handle
         self.running_tasks.write().insert(agent_id, handle.clone());
 
-        // Send command to start the task
+        // Send command to start the task (try_send avoids blocking the caller;
+        // returns an error if the bounded channel is full).
         self.task_sender
-            .send(TaskCommand::Start {
+            .try_send(TaskCommand::Start {
                 task: Box::new(task),
                 handle,
             })
             .map_err(|_| SchedulerError::SchedulingFailed {
                 agent_id,
-                reason: "Failed to send start command".into(),
+                reason: "Task channel full — backpressure applied".into(),
             })?;
 
         Ok(())
@@ -67,12 +70,12 @@ impl TaskManager {
         let handle = self.running_tasks.write().remove(&agent_id);
 
         if let Some(handle) = handle {
-            // Send termination command
+            // Send termination command (try_send to avoid blocking).
             self.task_sender
-                .send(TaskCommand::Terminate { agent_id, handle })
+                .try_send(TaskCommand::Terminate { agent_id, handle })
                 .map_err(|_| SchedulerError::SchedulingFailed {
                     agent_id,
-                    reason: "Failed to send terminate command".into(),
+                    reason: "Task channel full — cannot send terminate command".into(),
                 })?;
         }
 
@@ -131,7 +134,7 @@ impl TaskManager {
     }
 
     /// Start the task execution loop
-    fn start_task_loop(&self, mut task_receiver: mpsc::UnboundedReceiver<TaskCommand>) {
+    fn start_task_loop(&self, mut task_receiver: mpsc::Receiver<TaskCommand>) {
         let task_timeout = self.task_timeout;
 
         tokio::spawn(async move {

crates/runtime/tests/security_tests.rs

@@ -388,6 +388,79 @@ fn test_crypto_different_salts_produce_different_ciphertexts() {
     );
 }
 
+// ============================================================================
+// Type-level Security Tests
+// ============================================================================
+
+#[test]
+fn test_agent_id_is_unique() {
+    use symbi_runtime::types::AgentId;
+    let id1 = AgentId::new();
+    let id2 = AgentId::new();
+    assert_ne!(id1, id2);
+}
+
+#[test]
+fn test_message_id_is_unique() {
+    use symbi_runtime::types::MessageId;
+    let id1 = MessageId::new();
+    let id2 = MessageId::new();
+    assert_ne!(id1, id2);
+}
+
+#[cfg(test)]
+mod sandbox_security {
+    use symbi_runtime::types::SecurityTier;
+
+    #[test]
+    fn test_security_tier_ordering() {
+        // Higher tiers should provide more isolation
+        assert!(SecurityTier::Tier3 > SecurityTier::Tier2);
+        assert!(SecurityTier::Tier2 > SecurityTier::Tier1);
+    }
+}
+
+#[cfg(test)]
+mod error_security {
+    use symbi_runtime::types::error::*;
+    use symbi_runtime::types::AgentId;
+
+    #[test]
+    fn test_error_messages_dont_leak_secrets() {
+        let err = SecurityError::AuthenticationFailed("token validation failed".to_string());
+        let msg = format!("{err}");
+        // Error message should not contain raw tokens or keys
+        assert!(!msg.contains("Bearer"));
+        assert!(!msg.contains("sk-"));
+    }
+
+    #[test]
+    fn test_boxed_str_error_variants_work() {
+        let err = ResourceError::AllocationFailed {
+            agent_id: AgentId::new(),
+            reason: "insufficient memory".into(),
+        };
+        let msg = format!("{err}");
+        assert!(msg.contains("insufficient memory"));
+    }
+}
+
+#[cfg(test)]
+mod communication_security {
+    use symbi_runtime::types::error::CommunicationError;
+
+    #[test]
+    fn test_message_size_limit() {
+        let err = CommunicationError::MessageTooLarge {
+            size: 2_000_000,
+            max_size: 1_000_000,
+        };
+        let msg = format!("{err}");
+        assert!(msg.contains("2000000"));
+        assert!(msg.contains("1000000"));
+    }
+}
+
 // ============================================================================
 // Token Generation Tests
 // ============================================================================