feat(agent): activate stuck_threshold for time-based stuck job detection (#1234)

* feat(agent): activate stuck_threshold for time-based stuck job detection (#1223) The stuck_threshold field on DefaultSelfRepair was defined but never used (marked #[allow(dead_code)]). Jobs that got stuck in InProgress without transitioning to Stuck state (e.g., deadlock, unhandled timeout) were never detected by self-repair. Changes: - Add find_stuck_jobs_with_threshold() to ContextManager that detects InProgress jobs running longer than the threshold - Wire stuck_threshold into detect_stuck_jobs() so it uses threshold-based detection alongside explicit Stuck state detection - Remove dead_code annotation from stuck_threshold - Accept InProgress jobs in the stuck job detection filter Configurable via AGENT_STUCK_THRESHOLD_SECS (default: 300s). Closes #1223 Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]> * fix(agent): address PR #1234 review feedback for stuck_threshold - Transition InProgress jobs to Stuck before returning them from detect_stuck_jobs(), so attempt_recovery() (which requires Stuck state) works correctly on threshold-detected jobs - Add detect-and-repair E2E test covering the full InProgress -> Stuck -> recovery -> InProgress cycle - Rename idle_threshold -> elapsed_threshold in find_stuck_jobs_with_threshold for clarity - Add `use std::time::Duration` import and remove fully qualified paths - Update CLAUDE.md to reflect that stuck_threshold is now actively used Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]> * fix: measure stuck_duration from Stuck transition, handle InProgress→Stuck in repair - Fix stuck_duration computation to use the most recent Stuck transition timestamp instead of started_at, preventing jobs that ran for hours before becoming stuck from immediately exceeding the threshold - Fix last_activity to also use the Stuck transition timestamp - Transition InProgress jobs to Stuck before calling attempt_recovery() in repair_stuck_job(), since attempt_recovery() requires JobState::Stuck - Add regression test verifying a recently-stuck job with old started_at is not misdetected as exceeding a 5-minute threshold Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix(agent): address Copilot review comments on PR #1234 - Add comment in find_stuck_jobs_with_threshold() noting that started_at is not reset on Stuck->InProgress recovery, which may cause false positives for recovered jobs. Suggests tracking in_progress_since or using the most recent StateTransition as a future improvement. - Fix misleading test comment in stuck_duration_measured_from_stuck_transition test: explicitly Stuck jobs are always returned regardless of threshold. The test verifies stuck_duration is near-zero, not that the job is excluded. Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]> --------- Co-authored-by: Claude Opus 4.6 (1M context) <[email protected]> Co-authored-by: [email protected] <[email protected]>
2026-08-26 15:40:18 +00:00 · 2026-03-19 22:36:34 -07:00
parent 806d402876
commit 31c3b5b041
3 changed files with 216 additions and 43 deletions
@@ -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).

@@ -66,6 +66,7 @@ pub trait SelfRepair: Send + Sync {
 /// Default self-repair implementation.
 pub struct DefaultSelfRepair {
    context_manager: Arc<ContextManager>,
+    /// Jobs in `InProgress` longer than this are treated as stuck.
    stuck_threshold: Duration,
    max_repair_attempts: u32,
    store: Option<Arc<dyn Database>>,
@@ -111,15 +112,58 @@ impl DefaultSelfRepair {
 #[async_trait]
 impl SelfRepair for DefaultSelfRepair {
    async fn detect_stuck_jobs(&self) -> Vec<StuckJob> {
-        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,
@@ -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<Uuid> {
+        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<Duration>,
+    ) -> Vec<Uuid> {
+        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);