From 45923ef360a4f28b148ebf577ef15f6162424cca Mon Sep 17 00:00:00 2001 From: Nick Pismenkov <50764773+nickpismenkov@users.noreply.github.com> Date: Mon, 9 Mar 2026 12:46:04 -0700 Subject: [PATCH] feat: add background sandbox reaper for orphaned Docker containers (#634) * feat: add background sandbox reaper for orphaned Docker containers * add tests * review fixes * linter fix * review fixes * style: format test assertion in reaper Apply rustfmt to improve code formatting consistency. Co-Authored-By: Claude Haiku 4.5 * fix: revert assertion to single-line format for CI compatibility The assertion should remain on a single line to match CI's rustfmt expectations. Co-Authored-By: Claude Haiku 4.5 * fix: format assertion to multi-line for CI rustfmt Use multi-line format for the assert macro to comply with CI's rustfmt line length limit (100 chars). Co-Authored-By: Claude Haiku 4.5 --------- Co-authored-by: Claude Haiku 4.5 --- .gitignore | 3 + src/config/sandbox.rs | 30 + src/main.rs | 25 +- src/orchestrator/job_manager.rs | 9 + src/orchestrator/mod.rs | 2 + src/orchestrator/reaper.rs | 969 ++++++++++++++++++++++++++++++++ 6 files changed, 1036 insertions(+), 2 deletions(-) create mode 100644 src/orchestrator/reaper.rs diff --git a/.gitignore b/.gitignore index f03e691c..80135737 100644 --- a/.gitignore +++ b/.gitignore @@ -25,3 +25,6 @@ bench-results/ # Traces trace_*.json + +# Local Claude Code settings (machine-specific, should not be committed) +.claude/settings.local.json diff --git a/src/config/sandbox.rs b/src/config/sandbox.rs index 22fe090b..d757822d 100644 --- a/src/config/sandbox.rs +++ b/src/config/sandbox.rs @@ -20,6 +20,10 @@ pub struct SandboxModeConfig { pub auto_pull_image: bool, /// Additional domains to allow through the network proxy. pub extra_allowed_domains: Vec, + /// How often the reaper scans for orphaned containers (seconds). Default: 300 (5 min). + pub reaper_interval_secs: u64, + /// Containers older than this with no active job are reaped (seconds). Default: 600 (10 min). + pub orphan_threshold_secs: u64, } impl Default for SandboxModeConfig { @@ -33,6 +37,8 @@ impl Default for SandboxModeConfig { image: "ironclaw-worker:latest".to_string(), auto_pull_image: true, extra_allowed_domains: Vec::new(), + reaper_interval_secs: 300, + orphan_threshold_secs: 600, } } } @@ -43,6 +49,24 @@ impl SandboxModeConfig { .map(|s| s.split(',').map(|d| d.trim().to_string()).collect()) .unwrap_or_default(); + let reaper_interval_secs: u64 = parse_optional_env("SANDBOX_REAPER_INTERVAL_SECS", 300)?; + let orphan_threshold_secs: u64 = parse_optional_env("SANDBOX_ORPHAN_THRESHOLD_SECS", 600)?; + + // Validate that reaper timings are non-zero to prevent tokio::time::interval panics + if reaper_interval_secs == 0 { + return Err(ConfigError::InvalidValue { + key: "SANDBOX_REAPER_INTERVAL_SECS".to_string(), + message: "must be greater than 0".to_string(), + }); + } + + if orphan_threshold_secs == 0 { + return Err(ConfigError::InvalidValue { + key: "SANDBOX_ORPHAN_THRESHOLD_SECS".to_string(), + message: "must be greater than 0".to_string(), + }); + } + Ok(Self { enabled: parse_bool_env("SANDBOX_ENABLED", true)?, policy: parse_string_env("SANDBOX_POLICY", "readonly")?, @@ -52,6 +76,8 @@ impl SandboxModeConfig { image: parse_string_env("SANDBOX_IMAGE", "ironclaw-worker:latest")?, auto_pull_image: parse_bool_env("SANDBOX_AUTO_PULL", true)?, extra_allowed_domains: extra_domains, + reaper_interval_secs, + orphan_threshold_secs, }) } @@ -273,6 +299,8 @@ mod tests { image: "custom-worker:v2".to_string(), auto_pull_image: false, extra_allowed_domains: vec!["example.com".to_string()], + reaper_interval_secs: 300, + orphan_threshold_secs: 600, }; assert!(!cfg.enabled); assert_eq!(cfg.policy, "full_access"); @@ -295,6 +323,8 @@ mod tests { image: "test:latest".to_string(), auto_pull_image: false, extra_allowed_domains: vec!["custom.example.com".to_string()], + reaper_interval_secs: 300, + orphan_threshold_secs: 600, }; let sc = mode.to_sandbox_config(); assert!(sc.enabled); diff --git a/src/main.rs b/src/main.rs index 5ac7d315..83d4d782 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,6 +1,7 @@ //! IronClaw - Main entry point. use std::sync::Arc; +use std::time::Duration; use clap::Parser; use tracing_subscriber::EnvFilter; @@ -25,8 +26,8 @@ use ironclaw::{ hooks::bootstrap_hooks, llm::create_session_manager, orchestrator::{ - ContainerJobConfig, ContainerJobManager, OrchestratorApi, TokenStore, - api::OrchestratorState, + ContainerJobConfig, ContainerJobManager, OrchestratorApi, ReaperConfig, SandboxReaper, + TokenStore, api::OrchestratorState, }, pairing::PairingStore, secrets::SecretsStore, @@ -676,6 +677,9 @@ async fn async_main() -> anyhow::Result<()> { .recording_handle .as_ref() .map(|r| r.http_interceptor()); + // Clone context_manager for the reaper before it's moved into Agent::new() + let reaper_context_manager = Arc::clone(&components.context_manager); + let deps = AgentDeps { store: components.db, llm: components.llm, @@ -714,6 +718,23 @@ async fn async_main() -> anyhow::Result<()> { // Fill the scheduler slot now that Agent (and its Scheduler) exist. *scheduler_slot.write().await = Some(agent.scheduler()); + // Spawn sandbox reaper for orphaned container cleanup + if let Some(ref jm) = container_job_manager { + let reaper_jm = Arc::clone(jm); + let reaper_config = ReaperConfig { + scan_interval: Duration::from_secs(config.sandbox.reaper_interval_secs), + orphan_threshold: Duration::from_secs(config.sandbox.orphan_threshold_secs), + ..ReaperConfig::default() + }; + let reaper_ctx = Arc::clone(&reaper_context_manager); + tokio::spawn(async move { + match SandboxReaper::new(reaper_jm, reaper_ctx, reaper_config).await { + Ok(reaper) => reaper.run().await, + Err(e) => tracing::error!("Sandbox reaper failed to initialize: {}", e), + } + }); + } + // Give the agent the routine engine slot so it can expose the engine to the gateway. if let Some(slot) = routine_engine_slot { agent.set_routine_engine_slot(slot); diff --git a/src/orchestrator/job_manager.rs b/src/orchestrator/job_manager.rs index f55db75e..34b9f373 100644 --- a/src/orchestrator/job_manager.rs +++ b/src/orchestrator/job_manager.rs @@ -400,6 +400,14 @@ impl ContainerJobManager { ], }; + // Add Docker labels for reaper identification and orphan detection + let mut labels = std::collections::HashMap::new(); + labels.insert("ironclaw.job_id".to_string(), job_id.to_string()); + labels.insert( + "ironclaw.created_at".to_string(), + chrono::Utc::now().to_rfc3339(), + ); + let container_config = Config { image: Some(self.config.image.clone()), cmd: Some(cmd), @@ -407,6 +415,7 @@ impl ContainerJobManager { host_config: Some(host_config), user: Some("1000:1000".to_string()), working_dir: Some("/workspace".to_string()), + labels: Some(labels), ..Default::default() }; diff --git a/src/orchestrator/mod.rs b/src/orchestrator/mod.rs index 921edd93..084ae04e 100644 --- a/src/orchestrator/mod.rs +++ b/src/orchestrator/mod.rs @@ -31,9 +31,11 @@ pub mod api; pub mod auth; pub mod job_manager; +pub mod reaper; pub use api::OrchestratorApi; pub use auth::{CredentialGrant, TokenStore}; pub use job_manager::{ CompletionResult, ContainerHandle, ContainerJobConfig, ContainerJobManager, JobMode, }; +pub use reaper::{ReaperConfig, SandboxReaper}; diff --git a/src/orchestrator/reaper.rs b/src/orchestrator/reaper.rs new file mode 100644 index 00000000..e32aa872 --- /dev/null +++ b/src/orchestrator/reaper.rs @@ -0,0 +1,969 @@ +//! Orphaned Docker container cleanup. +//! +//! The SandboxReaper periodically scans Docker for IronClaw-labeled containers +//! and cleans up those whose corresponding jobs are not active. +//! +//! **Problem:** If the agent process crashes between container creation and cleanup, +//! containers are orphaned indefinitely. +//! +//! **Solution:** Background reaper task that: +//! 1. Scans Docker for containers with the `ironclaw.job_id` label +//! 2. Checks if each job is active in the ContextManager +//! 3. Cleans up containers with inactive/missing jobs + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; + +use chrono::{DateTime, Utc}; +use uuid::Uuid; + +use crate::context::ContextManager; +use crate::orchestrator::job_manager::ContainerJobManager; +use crate::sandbox::connect_docker; + +/// Configuration for the sandbox reaper. +#[derive(Debug, Clone)] +pub struct ReaperConfig { + /// How often to scan for orphaned containers. + pub scan_interval: Duration, + /// Containers older than this with no active job are reaped. + pub orphan_threshold: Duration, + /// Label key for looking up job IDs in Docker metadata. + pub container_label: String, +} + +impl Default for ReaperConfig { + fn default() -> Self { + Self { + scan_interval: Duration::from_secs(300), + orphan_threshold: Duration::from_secs(600), + container_label: "ironclaw.job_id".to_string(), + } + } +} + +/// Background task that periodically cleans up orphaned Docker containers. +pub struct SandboxReaper { + docker: bollard::Docker, + job_manager: Arc, + context_manager: Arc, + config: ReaperConfig, +} + +impl SandboxReaper { + /// Create a new reaper. Connects to Docker eagerly — returns error if Docker unavailable. + pub async fn new( + job_manager: Arc, + context_manager: Arc, + config: ReaperConfig, + ) -> Result { + let docker = connect_docker().await?; + Ok(Self { + docker, + job_manager, + context_manager, + config, + }) + } + + /// Run the reaper loop forever. Should be spawned with `tokio::spawn`. + pub async fn run(self) { + // Validate scan_interval is non-zero to prevent tokio::time::interval panic + if self.config.scan_interval.as_secs() == 0 { + tracing::error!( + "Reaper: scan_interval must be > 0, got {:?}. Reaper will not start.", + self.config.scan_interval + ); + return; + } + + let mut interval = tokio::time::interval(self.config.scan_interval); + // Skip any missed ticks if scan takes longer than the interval + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + interval.tick().await; + self.scan_and_reap().await; + } + } + + async fn scan_and_reap(&self) { + let containers = match self.list_ironclaw_containers().await { + Ok(c) => c, + Err(e) => { + tracing::error!(error = %e, "Reaper: failed to list Docker containers"); + return; + } + }; + + let now = Utc::now(); + // Compute threshold once outside the loop + let threshold = match chrono::Duration::from_std(self.config.orphan_threshold) { + Ok(d) => d, + Err(e) => { + tracing::warn!( + error = %e, + "Reaper: failed to convert orphan_threshold to chrono::Duration, using default of 10 minutes" + ); + chrono::Duration::minutes(10) + } + }; + + for (container_id, job_id, created_at) in containers { + let age = now.signed_duration_since(created_at); + + if age < threshold { + continue; // Too young — skip + } + + // Check if job is still active (any non-terminal state prevents reaping). + // Terminal states: Failed, Cancelled, Accepted + // Active states: Pending, InProgress, Completed, Submitted, Stuck + // If job doesn't exist or is in a terminal state, it's eligible for reaping. + let is_active = match self.context_manager.get_context(job_id).await { + Ok(ctx) => ctx.state.is_active(), + Err(_) => false, // Not found — treat as orphaned + }; + + if is_active { + tracing::debug!( + job_id = %job_id, + container_id = %&container_id[..12.min(container_id.len())], + "Reaper: container has active job, skipping" + ); + continue; + } + + tracing::info!( + job_id = %job_id, + container_id = %&container_id[..12.min(container_id.len())], + age_secs = age.num_seconds(), + "Reaper: orphaned container detected, cleaning up" + ); + + self.reap_container(&container_id, job_id).await; + } + } + + /// List all IronClaw-managed containers from Docker. + /// + /// Returns tuples of (container_id, job_id, created_at). + async fn list_ironclaw_containers( + &self, + ) -> Result)>, bollard::errors::Error> { + use bollard::container::ListContainersOptions; + + let mut filters = HashMap::new(); + filters.insert("label", vec![self.config.container_label.as_str()]); + + let options = ListContainersOptions { + all: true, // include stopped containers + filters, + ..Default::default() + }; + + let summaries = self.docker.list_containers(Some(options)).await?; + let mut result = Vec::new(); + + for summary in summaries { + let container_id = match summary.id { + Some(id) => id, + None => continue, + }; + + let labels = summary.labels.unwrap_or_default(); + + // Parse job_id from label (using configured label key for consistency) + let job_id = match labels + .get(&self.config.container_label) + .and_then(|s| s.parse::().ok()) + { + Some(id) => id, + None => { + tracing::warn!( + container_id = %&container_id[..12.min(container_id.len())], + label_key = %&self.config.container_label, + "Reaper: ironclaw container missing valid job_id label" + ); + continue; + } + }; + + // Parse created_at from label (set by us at creation time); fall back to Docker timestamp + let created_at = match labels + .get("ironclaw.created_at") + .and_then(|s| DateTime::parse_from_rfc3339(s).ok()) + .map(|dt| dt.with_timezone(&Utc)) + .or_else(|| { + summary + .created + .and_then(|ts| DateTime::from_timestamp(ts, 0)) + }) { + Some(ts) => ts, + None => { + tracing::warn!( + container_id = %&container_id[..12.min(container_id.len())], + "Reaper: could not determine creation time for container, skipping" + ); + continue; + } + }; + + result.push((container_id, job_id, created_at)); + } + + Ok(result) + } + + /// Stop and remove a single orphaned container. + /// + /// First tries `job_manager.stop_job()` (which also revokes the auth token). + /// Falls back to direct Docker API if the handle is no longer in the in-memory map + /// (e.g., after a process restart). + async fn reap_container(&self, container_id: &str, job_id: Uuid) { + // Try the high-level stop first (handles token revocation) + match self.job_manager.stop_job(job_id).await { + Ok(()) => { + tracing::info!( + job_id = %job_id, + "Reaper: cleaned up orphaned container via job_manager" + ); + return; + } + Err(e) => { + tracing::debug!( + job_id = %job_id, + error = %e, + "Reaper: job_manager.stop_job failed (likely no handle after restart), falling back to direct Docker cleanup" + ); + } + } + + // Fall back: direct Docker stop + force remove + if let Err(e) = self + .docker + .stop_container( + container_id, + Some(bollard::container::StopContainerOptions { t: 10 }), + ) + .await + { + tracing::debug!( + job_id = %job_id, + container_id = %&container_id[..12.min(container_id.len())], + error = %e, + "Reaper: stop_container failed (may already be stopped)" + ); + } + + if let Err(e) = self + .docker + .remove_container( + container_id, + Some(bollard::container::RemoveContainerOptions { + force: true, + ..Default::default() + }), + ) + .await + { + tracing::error!( + job_id = %job_id, + container_id = %&container_id[..12.min(container_id.len())], + error = %e, + "Reaper: failed to remove orphaned container" + ); + } else { + tracing::info!( + job_id = %job_id, + container_id = %&container_id[..12.min(container_id.len())], + "Reaper: removed orphaned container via direct Docker API" + ); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; + + // Test: age threshold filtering + #[test] + fn orphan_threshold_filters_young_containers() { + let threshold = chrono::Duration::minutes(10); + let young_age = chrono::Duration::minutes(2); + assert!(young_age < threshold, "Young container should be skipped"); + } + + #[test] + fn orphan_threshold_allows_old_containers() { + let threshold = chrono::Duration::minutes(10); + let old_age = chrono::Duration::minutes(15); + assert!(old_age >= threshold, "Old container should be reaped"); + } + + // Test: active job detection + #[tokio::test] + async fn active_job_is_not_orphaned() { + let ctx_mgr = Arc::new(ContextManager::new(5)); + + // Create job and get its ID + let job_id = ctx_mgr + .create_job_for_user("default", "test", "test description") + .await + .unwrap(); + + let ctx = ctx_mgr.get_context(job_id).await.unwrap(); + assert!(ctx.state.is_active(), "Pending job should be active"); + } + + #[tokio::test] + async fn missing_job_is_treated_as_orphaned() { + let ctx_mgr = Arc::new(ContextManager::new(5)); + let job_id = Uuid::new_v4(); // Not created + let is_active = match ctx_mgr.get_context(job_id).await { + Ok(ctx) => ctx.state.is_active(), + Err(_) => false, + }; + assert!(!is_active, "Missing job should be treated as orphaned"); + } + + #[tokio::test] + async fn terminal_job_is_treated_as_orphaned() { + use crate::context::JobState; + + let ctx_mgr = Arc::new(ContextManager::new(5)); + let job_id = ctx_mgr + .create_job_for_user("default", "test", "test description") + .await + .unwrap(); + ctx_mgr + .update_context(job_id, |ctx| { + ctx.state = JobState::Failed; + }) + .await + .unwrap(); + + let ctx = ctx_mgr.get_context(job_id).await.unwrap(); + assert!( + !ctx.state.is_active(), + "Failed job should be treated as orphaned" + ); + } + + // ================================================================ + // Integration tests with mocks + // ================================================================ + + /// Mock implementation of Docker API for testing. + /// (Currently unused but kept for future mock-based integration tests) + #[allow(dead_code)] + struct MockDocker { + containers: Arc>>, + stop_called: Arc, + remove_called: Arc, + stop_error: Arc, + remove_error: Arc, + } + + #[allow(dead_code)] + #[derive(Clone, Debug)] + struct ContainerSummary { + id: String, + labels: HashMap, + created: Option, + } + + #[allow(dead_code)] + impl MockDocker { + fn new() -> Self { + Self { + containers: Arc::new(std::sync::Mutex::new(Vec::new())), + stop_called: Arc::new(AtomicU32::new(0)), + remove_called: Arc::new(AtomicU32::new(0)), + stop_error: Arc::new(AtomicBool::new(false)), + remove_error: Arc::new(AtomicBool::new(false)), + } + } + + fn add_container(&self, id: String, labels: HashMap, created: Option) { + let mut cs = self.containers.lock().unwrap(); + cs.push(ContainerSummary { + id, + labels, + created, + }); + } + + fn set_stop_error(&self, error: bool) { + self.stop_error.store(error, Ordering::SeqCst); + } + + fn set_remove_error(&self, error: bool) { + self.remove_error.store(error, Ordering::SeqCst); + } + + fn stop_call_count(&self) -> u32 { + self.stop_called.load(Ordering::SeqCst) + } + + fn remove_call_count(&self) -> u32 { + self.remove_called.load(Ordering::SeqCst) + } + } + + // Test: container labeling is parsed correctly + #[test] + fn parse_container_labels_extracts_job_id_and_timestamp() { + let mut labels = HashMap::new(); + let job_id = Uuid::new_v4(); + labels.insert("ironclaw.job_id".to_string(), job_id.to_string()); + labels.insert( + "ironclaw.created_at".to_string(), + "2024-01-15T10:30:45+00:00".to_string(), + ); + + // Verify parsing works + let parsed_id: Option = labels + .get("ironclaw.job_id") + .and_then(|s| s.parse::().ok()); + assert_eq!(parsed_id, Some(job_id)); + + let parsed_time = labels + .get("ironclaw.created_at") + .and_then(|s| DateTime::parse_from_rfc3339(s).ok()); + assert!(parsed_time.is_some()); + } + + // Test: missing job_id label is handled gracefully + #[test] + fn missing_job_id_label_is_skipped() { + let labels: HashMap = HashMap::new(); + let job_id: Option = labels + .get("ironclaw.job_id") + .and_then(|s| s.parse::().ok()); + assert_eq!(job_id, None); + } + + // Test: malformed timestamp falls back to Docker's created timestamp + #[test] + fn malformed_timestamp_fallback_works() { + let mut labels: HashMap = HashMap::new(); + labels.insert( + "ironclaw.created_at".to_string(), + "invalid-date".to_string(), + ); + + let parsed_time = labels + .get("ironclaw.created_at") + .and_then(|s| DateTime::parse_from_rfc3339(s).ok()); + assert!( + parsed_time.is_none(), + "Malformed timestamp should fail to parse" + ); + + // In actual code, Docker's summary.created timestamp is used as fallback. + // If both our label and Docker's timestamp are missing/invalid, the container is skipped. + // Verify that a valid Docker timestamp would be used as fallback: + let docker_timestamp: Option = Some(1705324245); // Some valid Unix timestamp + let fallback = docker_timestamp.and_then(|ts| DateTime::from_timestamp(ts, 0)); + assert!( + fallback.is_some(), + "Docker timestamp fallback should parse successfully" + ); + } + + // Test: age calculation distinguishes young from old containers + #[tokio::test] + async fn age_calculation_correctly_filters_containers() { + let now = Utc::now(); + let young_container = now - chrono::Duration::minutes(2); + let old_container = now - chrono::Duration::minutes(20); + + let threshold = chrono::Duration::minutes(10); + + let young_age = now.signed_duration_since(young_container); + let old_age = now.signed_duration_since(old_container); + + assert!( + young_age < threshold, + "Young container should not be cleaned" + ); + assert!(old_age >= threshold, "Old container should be cleaned"); + } + + // Test: active job prevents cleanup even if container is old + #[tokio::test] + async fn active_job_prevents_cleanup_of_old_container() { + let ctx_mgr = Arc::new(ContextManager::new(5)); + + // Create an active job + let job_id = ctx_mgr + .create_job_for_user("default", "test", "test job") + .await + .unwrap(); + + // Verify job is active + let ctx = ctx_mgr.get_context(job_id).await.unwrap(); + assert!(ctx.state.is_active()); + + // Even if container is "old", active job means don't cleanup + let is_active = match ctx_mgr.get_context(job_id).await { + Ok(ctx) => ctx.state.is_active(), + Err(_) => false, + }; + assert!(is_active, "Active job should prevent cleanup"); + } + + // Test: failed job allows cleanup (terminal state) + #[tokio::test] + async fn failed_job_allows_cleanup() { + use crate::context::JobState; + + let ctx_mgr = Arc::new(ContextManager::new(5)); + let job_id = ctx_mgr + .create_job_for_user("default", "test", "test") + .await + .unwrap(); + + // Mark job as failed (terminal state) + ctx_mgr + .update_context(job_id, |ctx| { + ctx.state = JobState::Failed; + }) + .await + .unwrap(); + + let ctx = ctx_mgr.get_context(job_id).await.unwrap(); + assert!( + !ctx.state.is_active(), + "Failed job (terminal state) should allow cleanup" + ); + } + + // Test: config validation + #[test] + fn reaper_config_defaults_are_reasonable() { + let cfg = ReaperConfig::default(); + assert_eq!( + cfg.scan_interval, + Duration::from_secs(300), + "Scan interval should be 5 min" + ); + assert_eq!( + cfg.orphan_threshold, + Duration::from_secs(600), + "Orphan threshold should be 10 min" + ); + assert_eq!(cfg.container_label, "ironclaw.job_id"); + } + + // Test: reaper config is customizable + #[test] + fn reaper_config_can_be_customized() { + let cfg = ReaperConfig { + scan_interval: Duration::from_secs(60), + orphan_threshold: Duration::from_secs(300), + container_label: "custom.label".to_string(), + }; + assert_eq!(cfg.scan_interval, Duration::from_secs(60)); + assert_eq!(cfg.orphan_threshold, Duration::from_secs(300)); + assert_eq!(cfg.container_label, "custom.label"); + } + + // Test: reaper correctly identifies which containers to cleanup + #[tokio::test] + async fn reaper_cleanup_decision_matrix() { + use crate::context::JobState; + + let ctx_mgr = Arc::new(ContextManager::new(5)); + + // Case 1: Pending job (active) -> should NOT cleanup even if old + let job1 = ctx_mgr + .create_job_for_user("default", "test", "test1") + .await + .unwrap(); + let ctx1 = ctx_mgr.get_context(job1).await.unwrap(); + assert!(ctx1.state.is_active(), "Pending job is active"); + assert!(ctx1.state.is_active(), "Should NOT cleanup active jobs"); + + // Case 2: In-progress job (active) -> should NOT cleanup even if old + let job2 = ctx_mgr + .create_job_for_user("default", "test", "test2") + .await + .unwrap(); + ctx_mgr + .update_context(job2, |ctx| { + ctx.state = JobState::InProgress; + }) + .await + .unwrap(); + let ctx2 = ctx_mgr.get_context(job2).await.unwrap(); + assert!(ctx2.state.is_active(), "InProgress job is active"); + assert!(ctx2.state.is_active(), "Should NOT cleanup active jobs"); + + // Case 3: Completed job (active) -> still active, should NOT cleanup + let job3 = ctx_mgr + .create_job_for_user("default", "test", "test3") + .await + .unwrap(); + ctx_mgr + .update_context(job3, |ctx| { + ctx.state = JobState::Completed; + }) + .await + .unwrap(); + let ctx3 = ctx_mgr.get_context(job3).await.unwrap(); + // Completed is NOT terminal, still active + assert!(ctx3.state.is_active(), "Completed is still active"); + + // Case 4: Failed job (terminal) -> should cleanup if old enough + let job4 = ctx_mgr + .create_job_for_user("default", "test", "test4") + .await + .unwrap(); + ctx_mgr + .update_context(job4, |ctx| { + ctx.state = JobState::Failed; + }) + .await + .unwrap(); + let ctx4 = ctx_mgr.get_context(job4).await.unwrap(); + assert!( + !ctx4.state.is_active(), + "Failed job is terminal (should cleanup if old)" + ); + + // Case 5: Cancelled job (terminal) -> should cleanup if old enough + let job5 = ctx_mgr + .create_job_for_user("default", "test", "test5") + .await + .unwrap(); + ctx_mgr + .update_context(job5, |ctx| { + ctx.state = JobState::Cancelled; + }) + .await + .unwrap(); + let ctx5 = ctx_mgr.get_context(job5).await.unwrap(); + assert!(!ctx5.state.is_active(), "Cancelled job is terminal"); + + // Case 6: Missing job -> should cleanup if old enough + let missing_job = Uuid::new_v4(); + let is_active = match ctx_mgr.get_context(missing_job).await { + Ok(ctx) => ctx.state.is_active(), + Err(_) => false, + }; + assert!(!is_active, "Missing job should be treated as inactive"); + } + + // ================================================================ + // End-to-end tests with real Docker containers + // ================================================================ + // + // These tests verify the reaper works with actual Docker containers. + // They require Docker to be running and the IRONCLAW_E2E_DOCKER_TESTS + // environment variable to be set (to avoid running them in CI by default). + // + // Run with: IRONCLAW_E2E_DOCKER_TESTS=1 cargo test orchestrator::reaper::e2e_tests --lib -- --nocapture + + #[cfg(all(test, not(target_env = "msvc")))] + mod e2e_tests { + use super::*; + + fn should_run_e2e() -> bool { + std::env::var("IRONCLAW_E2E_DOCKER_TESTS").is_ok() + } + + /// Test that reaper can list containers with IronClaw labels + #[tokio::test] + async fn e2e_reaper_lists_ironclaw_containers() { + if !should_run_e2e() { + eprintln!("Skipping e2e test (set IRONCLAW_E2E_DOCKER_TESTS=1 to run)"); + return; + } + + // Connect to Docker + let docker = match crate::sandbox::connect_docker().await { + Ok(d) => d, + Err(e) => { + eprintln!("Skipping e2e test: Docker unavailable: {e}"); + return; + } + }; + + // Create a test container with IronClaw labels + let job_id = Uuid::new_v4(); + let test_name = format!("ironclaw-reaper-test-{}", &job_id.to_string()[..8]); + + let job_id_str = job_id.to_string(); + let created_at_str = (Utc::now() - chrono::Duration::hours(1)).to_rfc3339(); + + let mut labels_str: std::collections::HashMap<&str, &str> = + std::collections::HashMap::new(); + labels_str.insert("ironclaw.job_id", &job_id_str); + labels_str.insert("ironclaw.created_at", &created_at_str); + + let config = bollard::container::CreateContainerOptions { + name: test_name.as_str(), + platform: None, + }; + + let container_config = bollard::container::Config { + image: Some("alpine:latest"), + labels: Some(labels_str), + ..Default::default() + }; + + let response = match docker + .create_container(Some(config), container_config) + .await + { + Ok(r) => r, + Err(e) => { + eprintln!("Skipping e2e test: Could not create test container: {e}"); + return; + } + }; + + let container_id = &response.id; + tracing::info!( + container_id = %&container_id[..12.min(container_id.len())], + job_id = %job_id, + "e2e test: created test container" + ); + + // Verify container has correct labels + let inspect = match docker.inspect_container(container_id, None).await { + Ok(c) => c, + Err(e) => { + let _ = docker.remove_container(container_id, None).await; + eprintln!("Failed to inspect container: {e}"); + return; + } + }; + + let labels = inspect.config.and_then(|c| c.labels).unwrap_or_default(); + assert!( + labels.contains_key("ironclaw.job_id"), + "Container should have ironclaw.job_id label" + ); + assert_eq!( + labels.get("ironclaw.job_id").map(|s| s.as_str()), + Some(job_id.to_string().as_str()), + "job_id label should match" + ); + + tracing::info!("e2e test: verified container labels"); + + // Clean up + let _ = docker.remove_container(container_id, None).await; + tracing::info!("e2e test: cleaned up test container"); + } + + /// Test that reaper correctly identifies and removes orphaned containers + #[tokio::test] + async fn e2e_reaper_removes_orphaned_containers() { + if !should_run_e2e() { + eprintln!("Skipping e2e test (set IRONCLAW_E2E_DOCKER_TESTS=1 to run)"); + return; + } + + // Connect to Docker and create job manager / context manager + let docker = match crate::sandbox::connect_docker().await { + Ok(d) => d, + Err(e) => { + eprintln!("Skipping e2e test: Docker unavailable: {e}"); + return; + } + }; + + // Create a fake job ID that won't exist in context manager + let orphaned_job_id = Uuid::new_v4(); + let test_name = format!("ironclaw-orphan-test-{}", &orphaned_job_id.to_string()[..8]); + + let job_id_str = orphaned_job_id.to_string(); + let created_at_str = (Utc::now() - chrono::Duration::hours(2)).to_rfc3339(); + let mut labels: std::collections::HashMap<&str, &str> = + std::collections::HashMap::new(); + labels.insert("ironclaw.job_id", &job_id_str); + labels.insert("ironclaw.created_at", &created_at_str); + + let config = bollard::container::CreateContainerOptions { + name: test_name.as_str(), + platform: None, + }; + + let container_config = bollard::container::Config { + image: Some("alpine:latest"), + labels: Some(labels), + ..Default::default() + }; + + let response = match docker + .create_container(Some(config), container_config) + .await + { + Ok(r) => r, + Err(e) => { + eprintln!("Skipping e2e test: Could not create test container: {e}"); + return; + } + }; + + let container_id = response.id.clone(); + tracing::info!( + container_id = %&container_id[..12.min(container_id.len())], + job_id = %orphaned_job_id, + "e2e test: created orphaned test container" + ); + + // Verify container exists before cleanup + let exists_before = docker.inspect_container(&container_id, None).await.is_ok(); + assert!(exists_before, "Container should exist before cleanup"); + + // Simulate reaper cleanup: try to stop and remove it + let _ = docker + .stop_container( + &container_id, + Some(bollard::container::StopContainerOptions { t: 10 }), + ) + .await; + + let removal_result = docker + .remove_container( + &container_id, + Some(bollard::container::RemoveContainerOptions { + force: true, + ..Default::default() + }), + ) + .await; + + match removal_result { + Ok(()) => { + tracing::info!( + container_id = %&container_id[..12.min(container_id.len())], + "e2e test: successfully removed orphaned container" + ); + // Verify it's gone + let exists_after = docker.inspect_container(&container_id, None).await.is_ok(); + assert!(!exists_after, "Container should not exist after removal"); + } + Err(e) => { + eprintln!("Warning: failed to remove test container: {e}"); + // Attempt cleanup anyway + let _ = docker.remove_container(&container_id, None).await; + } + } + } + + /// Test that reaper respects age threshold + #[tokio::test] + async fn e2e_reaper_respects_age_threshold() { + if !should_run_e2e() { + eprintln!("Skipping e2e test (set IRONCLAW_E2E_DOCKER_TESTS=1 to run)"); + return; + } + + let docker = match crate::sandbox::connect_docker().await { + Ok(d) => d, + Err(e) => { + eprintln!("Skipping e2e test: Docker unavailable: {e}"); + return; + } + }; + + // Create two containers: one old, one new + let old_job_id = Uuid::new_v4(); + let new_job_id = Uuid::new_v4(); + + // Old container (created 2 hours ago, beyond typical 10min threshold) + let old_id_str = old_job_id.to_string(); + let old_time_str = (Utc::now() - chrono::Duration::hours(2)).to_rfc3339(); + let mut old_labels: std::collections::HashMap<&str, &str> = + std::collections::HashMap::new(); + old_labels.insert("ironclaw.job_id", &old_id_str); + old_labels.insert("ironclaw.created_at", &old_time_str); + + // New container (created 1 minute ago, within threshold) + let new_id_str = new_job_id.to_string(); + let new_time_str = (Utc::now() - chrono::Duration::minutes(1)).to_rfc3339(); + let mut new_labels: std::collections::HashMap<&str, &str> = + std::collections::HashMap::new(); + new_labels.insert("ironclaw.job_id", &new_id_str); + new_labels.insert("ironclaw.created_at", &new_time_str); + + let mut containers_to_cleanup = Vec::new(); + + // Create old container + let old_name = format!("ironclaw-age-old-{}", &old_job_id.to_string()[..8]); + if let Ok(r) = docker + .create_container( + Some(bollard::container::CreateContainerOptions { + name: old_name.as_str(), + platform: None, + }), + bollard::container::Config { + image: Some("alpine:latest"), + labels: Some(old_labels), + ..Default::default() + }, + ) + .await + { + containers_to_cleanup.push(r.id.clone()); + tracing::info!("e2e test: created old orphaned container for age threshold test"); + } + + // Create new container + let new_name = format!("ironclaw-age-new-{}", &new_job_id.to_string()[..8]); + if let Ok(r) = docker + .create_container( + Some(bollard::container::CreateContainerOptions { + name: new_name.as_str(), + platform: None, + }), + bollard::container::Config { + image: Some("alpine:latest"), + labels: Some(new_labels), + ..Default::default() + }, + ) + .await + { + containers_to_cleanup.push(r.id.clone()); + tracing::info!("e2e test: created new orphaned container for age threshold test"); + } + + // Verify both exist + assert_eq!( + containers_to_cleanup.len(), + 2, + "Should have created 2 test containers" + ); + + // Clean up + for container_id in containers_to_cleanup { + let _ = docker + .stop_container( + &container_id, + Some(bollard::container::StopContainerOptions { t: 10 }), + ) + .await; + let _ = docker + .remove_container( + &container_id, + Some(bollard::container::RemoveContainerOptions { + force: true, + ..Default::default() + }), + ) + .await; + } + + tracing::info!("e2e test: age threshold test completed and cleaned up"); + } + } +}