fix(agent): prevent self-repair notification spam for stuck jobs (#1867)

* fix(agent): prevent self-repair notification spam for stuck jobs

When a stuck job exceeds max repair attempts, self-repair returns
ManualRequired but never transitions the job to a terminal state.
detect_stuck_jobs() re-finds it every cycle (~60s), sending a
Telegram notification each time — infinite spam.

Two-layer fix:
1. repair_stuck_job: transition to Failed before returning
   ManualRequired, so detect_stuck_jobs stops finding the job
2. Agent loop: HashSet dedup prevents duplicate ManualRequired
   notifications per job (defense-in-depth if transition fails)

Also adds Pending → Failed to the state machine — stuck Pending
jobs (dispatched but never started) could not be terminated.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix(agent): handle transition failure in ManualRequired and adjust message

Log error if Failed transition fails, and adjust the ManualRequired
message to accurately reflect whether the job was marked failed or not.

Addresses gemini-code-assist feedback on PR #1867.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix(agent): flatten nested Result, dedup Failed notifications, fix test state path

Adversarial review findings:
1. CRITICAL: update_context returns Result<Result<()>>. Using .is_ok()
   on the outer only checks job existence, not transition success.
   Fixed: matches!(result, Ok(Ok(()))).
2. IMPORTANT: RepairResult::Failed arm had same spam potential as
   ManualRequired. Applied same dedup pattern.
3. IMPORTANT: Test exercised Pending→Failed (bypassing production
   path). Now transitions through InProgress→Stuck→Failed.
4. Dropped unrelated Cargo.lock version bump.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* docs: update state machine diagram to include Pending -> Failed

Adversarial review finding: CLAUDE.md state diagram was the canonical
reference but didn't show the new Pending -> Failed transition.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: j-bloggs <j-bloggs@users.noreply.github.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

CLAUDE.md

@@ -202,8 +202,8 @@ When modifying a module with a spec, read the spec first. Code follows spec; spe
 
 ```
 Pending -> InProgress -> Completed -> Submitted -> Accepted
-                     \-> Failed
-                     \-> Stuck -> InProgress (recovery)
+    \                \-> Failed
+     \-> Failed       \-> Stuck -> InProgress (recovery)
                               \-> Failed
 ```
 

src/agent/agent_loop.rs

@@ -528,6 +528,11 @@ impl Agent {
         let repair_channels = self.channels.clone();
         let repair_owner_id = self.owner_id().to_string();
         let repair_handle = tokio::spawn(async move {
+            // Track jobs that have already been escalated to ManualRequired
+            // to prevent sending duplicate notifications every repair cycle.
+            let mut notified_manual: std::collections::HashSet<uuid::Uuid> =
+                std::collections::HashSet::new();
+
             loop {
                 tokio::time::sleep(repair_interval).await;
 
@@ -548,19 +553,31 @@ impl Agent {
                         }
                         Ok(RepairResult::Failed { message }) => {
                             tracing::error!("Repair failed: {}", message);
-                            Some(format!(
-                                "Job {} was stuck for {}s, recovery failed permanently: {}",
-                                job.job_id,
-                                job.stuck_duration.as_secs(),
-                                message
-                            ))
+                            // Dedup: only notify once per job (same pattern as ManualRequired)
+                            if notified_manual.insert(job.job_id) {
+                                Some(format!(
+                                    "Job {} was stuck for {}s, recovery failed permanently: {}",
+                                    job.job_id,
+                                    job.stuck_duration.as_secs(),
+                                    message
+                                ))
+                            } else {
+                                None
+                            }
                         }
                         Ok(RepairResult::ManualRequired { message }) => {
                             tracing::warn!("Manual intervention needed: {}", message);
-                            Some(format!(
-                                "Job {} needs manual intervention: {}",
-                                job.job_id, message
-                            ))
+                            // Only notify once per job to prevent notification spam.
+                            // The job should have been transitioned to Failed by
+                            // repair_stuck_job, but guard against that failing too.
+                            if notified_manual.insert(job.job_id) {
+                                Some(format!(
+                                    "Job {} needs manual intervention: {}",
+                                    job.job_id, message
+                                ))
+                            } else {
+                                None
+                            }
                         }
                         Ok(RepairResult::Retry { message }) => {
                             tracing::warn!("Repair needs retry: {}", message);

src/agent/self_repair.rs

@@ -224,10 +224,44 @@ impl SelfRepair for DefaultSelfRepair {
     async fn repair_stuck_job(&self, job: &StuckJob) -> Result<RepairResult, RepairError> {
         // Check if we've exceeded max repair attempts
         if job.repair_attempts >= self.max_repair_attempts {
+            // Transition to Failed so detect_stuck_jobs() stops finding this job.
+            // Without this, the repair loop re-detects the job every cycle and
+            // sends a ManualRequired notification each time (notification spam).
+            // update_context returns Result<Result<(), String>, JobError>.
+            // Outer Err = job not found. Inner Err = invalid state transition.
+            // Both mean the job was NOT transitioned to Failed.
+            let transition_ok = matches!(
+                self.context_manager
+                    .update_context(job.job_id, |ctx| {
+                        ctx.transition_to(
+                            JobState::Failed,
+                            Some(format!(
+                                "exceeded max repair attempts ({})",
+                                self.max_repair_attempts
+                            )),
+                        )
+                    })
+                    .await,
+                Ok(Ok(()))
+            );
+
+            if !transition_ok {
+                tracing::error!(
+                    job = %job.job_id,
+                    "Failed to transition job to Failed state after exceeding max repair attempts"
+                );
+            }
+
+            let status = if transition_ok {
+                "and has been marked failed"
+            } else {
+                "but could not be marked failed (will be suppressed by dedup)"
+            };
+
             return Ok(RepairResult::ManualRequired {
                 message: format!(
-                    "Job {} has exceeded maximum repair attempts ({})",
-                    job.job_id, self.max_repair_attempts
+                    "Job {} has exceeded maximum repair attempts ({}) {}",
+                    job.job_id, self.max_repair_attempts, status
                 ),
             });
         }
@@ -581,7 +615,19 @@ mod tests {
         let cm = Arc::new(ContextManager::new(10));
         let job_id = cm.create_job("Unrepairable", "desc").await.unwrap();
 
-        let repair = DefaultSelfRepair::new(cm, Duration::from_secs(60), 2);
+        // Transition through the production path: Pending → InProgress → Stuck
+        cm.update_context(job_id, |ctx| ctx.transition_to(JobState::InProgress, None))
+            .await
+            .unwrap()
+            .unwrap();
+        cm.update_context(job_id, |ctx| {
+            ctx.transition_to(JobState::Stuck, Some("test".into()))
+        })
+        .await
+        .unwrap()
+        .unwrap();
+
+        let repair = DefaultSelfRepair::new(cm.clone(), Duration::from_secs(60), 2);
 
         let stuck_job = StuckJob {
             job_id,
@@ -597,6 +643,17 @@ mod tests {
             "Expected ManualRequired, got: {:?}",
             result
         );
+
+        // Regression: the job must be transitioned to Failed so
+        // detect_stuck_jobs() stops finding it. Without this, the repair
+        // loop re-detects the job every cycle and sends ManualRequired
+        // notifications forever (notification spam bug).
+        let ctx = cm.get_context(job_id).await.unwrap();
+        assert_eq!(
+            ctx.state,
+            JobState::Failed,
+            "Job should be Failed after exceeding max repair attempts"
+        );
     }
 
     #[tokio::test]

src/context/state.rs

@@ -59,8 +59,9 @@ impl JobState {
 
         matches!(
             (self, target),
-            // From Pending
-            (Pending, InProgress) | (Pending, Cancelled) |
+            // From Pending (Failed added for self-repair: stuck Pending jobs
+            // that exhaust repair attempts must be terminable)
+            (Pending, InProgress) | (Pending, Failed) | (Pending, Cancelled) |
             // From InProgress
             (InProgress, Completed) | (InProgress, Failed) |
             (InProgress, Stuck) | (InProgress, Cancelled) |