diff --git a/src/agent/CLAUDE.md b/src/agent/CLAUDE.md index e55c9591..686753de 100644 --- a/src/agent/CLAUDE.md +++ b/src/agent/CLAUDE.md @@ -113,7 +113,7 @@ Check-insert is done under a single write lock to prevent TOCTOU races. A cleanu 4. Detects broken tools via `store.get_broken_tools(5)` (threshold: 5 failures). Requires `with_store()` to be called; returns empty without a store. 5. Attempts to rebuild broken tools via `SoftwareBuilder`. Requires `with_builder()` to be called; returns `ManualRequired` without a builder. -Note: the `stuck_threshold` duration is stored but currently unused (marked `#[allow(dead_code)]`). Stuck detection relies on `JobState::Stuck` being set by the state machine, not wall-clock time comparison. +The `stuck_threshold` duration is used for time-based detection of `InProgress` jobs that have been running longer than the threshold. When `detect_stuck_jobs()` finds such jobs, it transitions them to `Stuck` before returning them, enabling the normal `attempt_recovery()` path. Repair results: `Success`, `Retry`, `Failed`, `ManualRequired`. `Retry` does NOT notify the user (to avoid spam). diff --git a/src/agent/self_repair.rs b/src/agent/self_repair.rs index db491194..4e58cb15 100644 --- a/src/agent/self_repair.rs +++ b/src/agent/self_repair.rs @@ -66,6 +66,7 @@ pub trait SelfRepair: Send + Sync { /// Default self-repair implementation. pub struct DefaultSelfRepair { context_manager: Arc, + /// Jobs in `InProgress` longer than this are treated as stuck. stuck_threshold: Duration, max_repair_attempts: u32, store: Option>, @@ -111,15 +112,58 @@ impl DefaultSelfRepair { #[async_trait] impl SelfRepair for DefaultSelfRepair { async fn detect_stuck_jobs(&self) -> Vec { - let stuck_ids = self.context_manager.find_stuck_jobs().await; + let stuck_ids = self + .context_manager + .find_stuck_jobs_with_threshold(Some(self.stuck_threshold)) + .await; let mut stuck_jobs = Vec::new(); for job_id in stuck_ids { if let Ok(ctx) = self.context_manager.get_context(job_id).await - && ctx.state == JobState::Stuck + && matches!(ctx.state, JobState::Stuck | JobState::InProgress) { - // Measure stuck_duration from the most recent Stuck transition, - // not from started_at (which reflects when the job first ran). + // InProgress jobs detected by threshold need to be transitioned + // to Stuck before they can be repaired (attempt_recovery requires + // Stuck state). These jobs already passed the threshold check in + // find_stuck_jobs_with_threshold, so skip the duration filter below. + let just_transitioned = ctx.state == JobState::InProgress; + if just_transitioned { + let reason = "exceeded stuck_threshold"; + let transition = self + .context_manager + .update_context(job_id, |ctx| ctx.mark_stuck(reason)) + .await; + match transition { + Ok(Ok(())) => {} + Ok(Err(e)) => { + tracing::warn!( + job = %job_id, + "Failed to mark InProgress job as Stuck: {}", + e + ); + continue; + } + Err(e) => { + tracing::warn!( + job = %job_id, + "Failed to transition InProgress job to Stuck: {}", + e + ); + continue; + } + } + } + + // Re-fetch context after potential InProgress->Stuck transition + // so that stuck_since picks up the new transition timestamp. + let ctx = match self.context_manager.get_context(job_id).await { + Ok(c) => c, + Err(_) => continue, + }; + + // Use the timestamp of the most recent Stuck transition, not started_at. + // A job that ran for hours before becoming stuck should not immediately + // exceed the threshold — we measure from when it actually became stuck. let stuck_since = ctx .transitions .iter() @@ -134,8 +178,10 @@ impl SelfRepair for DefaultSelfRepair { }) .unwrap_or_default(); - // Only report jobs that have been stuck long enough - if stuck_duration < self.stuck_threshold { + // Only report already-Stuck jobs that have been stuck long enough. + // Jobs just transitioned from InProgress skip this check — they + // were already vetted by find_stuck_jobs_with_threshold. + if !just_transitioned && stuck_duration < self.stuck_threshold { continue; } @@ -163,10 +209,17 @@ impl SelfRepair for DefaultSelfRepair { }); } - // Try to recover the job + // Try to recover the job. + // If the job is still InProgress (detected via stuck_threshold), transition + // it to Stuck first so that attempt_recovery() can move it back to InProgress. let result = self .context_manager - .update_context(job.job_id, |ctx| ctx.attempt_recovery()) + .update_context(job.job_id, |ctx| { + if ctx.state == JobState::InProgress { + ctx.transition_to(JobState::Stuck, Some("exceeded stuck_threshold".into()))?; + } + ctx.attempt_recovery() + }) .await; match result { @@ -489,6 +542,82 @@ mod tests { ); } + #[tokio::test] + async fn detect_and_repair_in_progress_job_via_threshold() { + let cm = Arc::new(ContextManager::new(10)); + let job_id = cm.create_job("Long running", "desc").await.unwrap(); + + // Transition to InProgress. + cm.update_context(job_id, |ctx| ctx.transition_to(JobState::InProgress, None)) + .await + .unwrap() + .unwrap(); + + // Backdate started_at to simulate a job running for 10 minutes. + cm.update_context(job_id, |ctx| { + ctx.started_at = Some(Utc::now() - chrono::Duration::seconds(600)); + }) + .await + .unwrap(); + + // Use a 5-minute threshold so the 10-minute job is detected. + let repair = DefaultSelfRepair::new(Arc::clone(&cm), Duration::from_secs(300), 3); + + // detect_stuck_jobs should find it and transition InProgress -> Stuck. + let stuck = repair.detect_stuck_jobs().await; + assert_eq!(stuck.len(), 1); + assert_eq!(stuck[0].job_id, job_id); + + // After detection the job should now be in Stuck state. + let ctx = cm.get_context(job_id).await.unwrap(); + assert_eq!(ctx.state, JobState::Stuck); + + // Repair should recover it: Stuck -> InProgress. + let result = repair.repair_stuck_job(&stuck[0]).await.unwrap(); + assert!( + matches!(result, RepairResult::Success { .. }), + "Expected Success, got: {:?}", + result + ); + + // Job should be back to InProgress after recovery. + let ctx = cm.get_context(job_id).await.unwrap(); + assert_eq!(ctx.state, JobState::InProgress); + } + + #[tokio::test] + async fn detect_broken_tools_returns_empty_without_store() { + let cm = Arc::new(ContextManager::new(10)); + let repair = DefaultSelfRepair::new(cm, Duration::from_secs(60), 3); + + // No store configured, should return empty. + let broken = repair.detect_broken_tools().await; + assert!(broken.is_empty()); + } + + #[tokio::test] + async fn repair_broken_tool_returns_manual_without_builder() { + let cm = Arc::new(ContextManager::new(10)); + let repair = DefaultSelfRepair::new(cm, Duration::from_secs(60), 3); + + let broken = BrokenTool { + name: "test-tool".to_string(), + failure_count: 10, + last_error: Some("crash".to_string()), + first_failure: Utc::now(), + last_failure: Utc::now(), + last_build_result: None, + repair_attempts: 0, + }; + + let result = repair.repair_broken_tool(&broken).await.unwrap(); + assert!( + matches!(result, RepairResult::ManualRequired { .. }), + "Expected ManualRequired without builder, got: {:?}", + result + ); + } + #[tokio::test] async fn detect_stuck_jobs_filters_by_threshold() { let cm = Arc::new(ContextManager::new(10)); @@ -581,39 +710,6 @@ mod tests { ); } - #[tokio::test] - async fn detect_broken_tools_returns_empty_without_store() { - let cm = Arc::new(ContextManager::new(10)); - let repair = DefaultSelfRepair::new(cm, Duration::from_secs(60), 3); - - // No store configured, should return empty. - let broken = repair.detect_broken_tools().await; - assert!(broken.is_empty()); - } - - #[tokio::test] - async fn repair_broken_tool_returns_manual_without_builder() { - let cm = Arc::new(ContextManager::new(10)); - let repair = DefaultSelfRepair::new(cm, Duration::from_secs(60), 3); - - let broken = BrokenTool { - name: "test-tool".to_string(), - failure_count: 10, - last_error: Some("crash".to_string()), - first_failure: Utc::now(), - last_failure: Utc::now(), - last_build_result: None, - repair_attempts: 0, - }; - - let result = repair.repair_broken_tool(&broken).await.unwrap(); - assert!( - matches!(result, RepairResult::ManualRequired { .. }), - "Expected ManualRequired without builder, got: {:?}", - result - ); - } - /// Mock SoftwareBuilder that returns a successful build result. struct MockBuilder { build_count: std::sync::atomic::AtomicU32, diff --git a/src/context/manager.rs b/src/context/manager.rs index 6eb63260..f9bfedca 100644 --- a/src/context/manager.rs +++ b/src/context/manager.rs @@ -1,6 +1,7 @@ //! Context manager for handling multiple job contexts. use std::collections::HashMap; +use std::time::Duration; use tokio::sync::RwLock; use uuid::Uuid; @@ -205,12 +206,46 @@ impl ContextManager { } /// Find stuck jobs. + /// + /// Returns jobs that are explicitly in `Stuck` state, plus `InProgress` + /// jobs that have been running longer than `elapsed_threshold` (if provided). + /// The threshold-based detection catches jobs that never transitioned to + /// `Stuck` (e.g., due to a deadlock or unhandled timeout). pub async fn find_stuck_jobs(&self) -> Vec { + self.find_stuck_jobs_with_threshold(None).await + } + + /// Find stuck jobs with an optional elapsed threshold for `InProgress` detection. + pub async fn find_stuck_jobs_with_threshold( + &self, + elapsed_threshold: Option, + ) -> Vec { + let now = chrono::Utc::now(); self.contexts .read() .await .iter() - .filter(|(_, c)| c.state == crate::context::JobState::Stuck) + .filter(|(_, c)| { + // Always include explicitly Stuck jobs. + if c.state == crate::context::JobState::Stuck { + return true; + } + // Detect InProgress jobs that have been running beyond the elapsed threshold. + // NOTE: `started_at` is set on the first transition to InProgress and is + // NOT reset when a job recovers from Stuck back to InProgress. This means + // a recovered job may be re-detected on the next scan. A future improvement + // could track `in_progress_since` or use the most recent StateTransition + // with `to == InProgress` to avoid false positives on recovered jobs. + if c.state == crate::context::JobState::InProgress + && let Some(threshold) = elapsed_threshold + && let Some(started) = c.started_at + { + let elapsed = now.signed_duration_since(started); + let elapsed_secs = elapsed.num_seconds().max(0) as u64; + return elapsed_secs > threshold.as_secs(); + } + false + }) .map(|(id, _)| *id) .collect() } @@ -629,6 +664,48 @@ mod tests { assert_eq!(stuck[0], id2); } + /// Regression test for #1223: InProgress jobs exceeding the threshold + /// should be detected as stuck even if they never transitioned to Stuck. + #[tokio::test] + async fn find_stuck_jobs_with_threshold_detects_idle_in_progress() { + let manager = ContextManager::new(10); + + let id1 = manager.create_job("Active job", "desc").await.unwrap(); + let id2 = manager.create_job("Idle job", "desc").await.unwrap(); + + // Both transition to InProgress + for id in [id1, id2] { + manager + .update_context(id, |ctx| { + ctx.transition_to(crate::context::JobState::InProgress, None) + }) + .await + .unwrap() + .unwrap(); + } + + // Backdate id2's started_at to simulate a long-running job + manager + .update_context(id2, |ctx| -> Result<(), crate::error::JobError> { + ctx.started_at = Some(chrono::Utc::now() - chrono::Duration::seconds(600)); + Ok(()) + }) + .await + .unwrap() + .unwrap(); + + // With a 5-minute threshold, only id2 (10 min) should be detected + let stuck = manager + .find_stuck_jobs_with_threshold(Some(Duration::from_secs(300))) + .await; + assert_eq!(stuck.len(), 1); + assert_eq!(stuck[0], id2); + + // Without threshold, neither InProgress job is detected (no explicit Stuck state) + let stuck_no_threshold = manager.find_stuck_jobs().await; + assert!(stuck_no_threshold.is_empty()); + } + #[tokio::test] async fn active_count_tracks_non_terminal_jobs() { let manager = ContextManager::new(10);