diff --git a/server/src/agents/pool/auto_assign/watchdog/mod.rs b/server/src/agents/pool/auto_assign/watchdog/mod.rs index 7376fbda..e4ee0e71 100644 --- a/server/src/agents/pool/auto_assign/watchdog/mod.rs +++ b/server/src/agents/pool/auto_assign/watchdog/mod.rs @@ -4,9 +4,11 @@ mod budget; mod limits; mod orphan; +mod reap; #[cfg(test)] mod tests; +use std::collections::HashSet; use std::path::Path; use crate::agents::AgentStatus; @@ -18,6 +20,7 @@ use crate::slog_warn; use super::super::AgentPool; use limits::check_agent_limits; use orphan::check_orphaned_agents; +use reap::reap_failed_agents; pub(crate) use budget::{compute_budget_from_logs, compute_budget_from_single_log}; pub(crate) use limits::{count_turns_in_log, resolve_session_log}; @@ -45,7 +48,9 @@ impl AgentPool { if let Some(root) = project_root { let terminated = check_agent_limits(&self.agents, root).await; let config = ProjectConfig::load(root).unwrap_or_default(); + let mut just_terminated: HashSet = HashSet::new(); for (key, reason) in &terminated { + just_terminated.insert(key.clone()); // Step 1: snapshot the agent's worktree path so we can find every // process running in it (claude + any subprocesses). This must // happen BEFORE we mutate the agent record so we can read the @@ -132,7 +137,16 @@ impl AgentPool { if !terminated.is_empty() { Self::notify_agent_state_changed(&self.watcher_tx); } - return orphaned + terminated.len(); + + // Bug 1198: reap any other Failed pool entry with no live process + // — orphan-detected above, or left behind by a spawn error + // (inactivity-watchdog kill, worktree timeout, runtime error) + // that never routed through the retry/respawn path. Entries the + // limits loop above just processed are excluded so their retry + // count isn't bumped twice. + let reaped = reap_failed_agents(self, root, &config, &just_terminated).await; + + return orphaned + terminated.len() + reaped; } orphaned diff --git a/server/src/agents/pool/auto_assign/watchdog/reap.rs b/server/src/agents/pool/auto_assign/watchdog/reap.rs new file mode 100644 index 00000000..dce88bc8 --- /dev/null +++ b/server/src/agents/pool/auto_assign/watchdog/reap.rs @@ -0,0 +1,99 @@ +//! Reap: removes stale `Failed` pool entries left behind by orphan detection, +//! watchdog kills, or internal spawn failures, and respawns the story's agent +//! (or blocks the story once its retry budget is exhausted). +//! +//! `check_orphaned_agents` only scans `Running`/`Pending` entries, so once an +//! entry is marked `Failed` it becomes invisible to every later watchdog +//! pass. Without this step a `Failed` entry with no live process sits in the +//! pool forever: `list_agents` keeps showing it, and the story never gets a +//! new agent unless some unrelated CRDT transition happens to trigger a +//! system-wide auto-assign scan. + +use std::collections::HashSet; +use std::path::Path; + +use crate::agents::AgentStatus; +use crate::agents::pool::AgentPool; +use crate::agents::pool::pipeline::should_block_story; +use crate::config::ProjectConfig; +use crate::io::watcher::WatcherEvent; +use crate::{slog, slog_warn}; + +/// Reap every `Failed` pool entry with no live process, except keys in +/// `exclude_keys` (already handled by the caller's own retry/block logic in +/// this same pass — e.g. limit-exceeded kills). +/// +/// For each reaped entry: removes it from the pool (so `list_agents` stops +/// showing it), increments the story's retry count via [`should_block_story`] +/// and, unless that blocks the story, respawns the agent by name. +/// `start_agent`'s own session-store lookup resumes the prior session +/// automatically whenever one was recorded — no explicit session plumbing +/// needed here. +pub(super) async fn reap_failed_agents( + pool: &AgentPool, + project_root: &Path, + config: &ProjectConfig, + exclude_keys: &HashSet, +) -> usize { + let dead: Vec<(String, String, String)> = { + let mut agents = pool.agents.lock().await; + let keys: Vec = agents + .iter() + .filter(|(key, agent)| { + agent.status == AgentStatus::Failed + && !exclude_keys.contains(*key) + && agent + .task_handle + .as_ref() + .map(|h| h.is_finished()) + .unwrap_or(true) + }) + .map(|(key, _)| key.clone()) + .collect(); + keys.into_iter() + .filter_map(|key| { + agents.remove(&key).map(|agent| { + let story_id = key + .rsplit_once(':') + .map(|(s, _)| s.to_string()) + .unwrap_or_else(|| key.clone()); + (key, story_id, agent.agent_name) + }) + }) + .collect() + }; + + let count = dead.len(); + for (key, story_id, agent_name) in dead { + if let Some(block_reason) = should_block_story(&story_id, config.max_retries, "watchdog") { + let _ = pool.watcher_tx.send(WatcherEvent::StoryBlocked { + story_id: story_id.clone(), + reason: block_reason, + }); + slog!( + "[watchdog] Story '{story_id}' blocked after exceeding retry limit \ + (reaped dead pool entry '{key}')." + ); + continue; + } + + slog!( + "[watchdog] Reaping dead pool entry '{key}'; respawning '{agent_name}' \ + for '{story_id}'." + ); + if let Err(e) = pool + .start_agent(project_root, &story_id, Some(&agent_name), None, None) + .await + { + slog_warn!( + "[watchdog] Failed to respawn '{agent_name}' for '{story_id}' after reap: {e}" + ); + } + } + + if count > 0 { + AgentPool::notify_agent_state_changed(&pool.watcher_tx); + } + + count +} diff --git a/server/src/agents/pool/auto_assign/watchdog/tests/mod.rs b/server/src/agents/pool/auto_assign/watchdog/tests/mod.rs index 326c2127..73e6066a 100644 --- a/server/src/agents/pool/auto_assign/watchdog/tests/mod.rs +++ b/server/src/agents/pool/auto_assign/watchdog/tests/mod.rs @@ -4,6 +4,7 @@ use std::path::Path; mod limits_tests; mod orphan_tests; +mod reap_tests; /// Write a fake session log file with `n` tool-using assistant turn entries. /// diff --git a/server/src/agents/pool/auto_assign/watchdog/tests/reap_tests.rs b/server/src/agents/pool/auto_assign/watchdog/tests/reap_tests.rs new file mode 100644 index 00000000..39f6242a --- /dev/null +++ b/server/src/agents/pool/auto_assign/watchdog/tests/reap_tests.rs @@ -0,0 +1,238 @@ +//! Regression tests for the reap pass (bug 1198): a Failed pool entry with +//! no live process must be respawned by the next watchdog pass, with its +//! story's retry count bumped, and must stop appearing in list_agents. + +use super::super::super::super::{AgentPool, composite_key}; +use super::{write_fake_session_log, write_project_config}; +use crate::agents::AgentStatus; + +/// AC1 + AC4: a Failed entry with no live process (simulating e.g. the +/// inactivity watchdog kill landing in spawn.rs's generic Err arm) is +/// respawned by the very next `run_watchdog_pass` — no manual +/// stop_agent/start_agent needed. +#[tokio::test] +async fn reap_respawns_stale_failed_agent() { + crate::db::ensure_content_store(); + crate::crdt_state::init_for_test(); + + let tmp = tempfile::tempdir().unwrap(); + let root = tmp.path(); + + write_project_config( + root, + r#" +[[agent]] +name = "coder-1" +runtime = "claude-code" +"#, + ); + + let story_id = "1198_story_stuck"; + crate::db::write_content( + crate::db::ContentKey::Story(story_id), + "---\nname: Stuck Story\n---\n", + ); + crate::crdt_state::write_item_str(story_id, "2_current", Some("Stuck Story"), None, None, None); + + // Inject a Failed entry with no task_handle — exactly what spawn.rs's + // generic Err arm leaves behind after e.g. an inactivity-watchdog kill. + let pool = AgentPool::new_test(3001); + pool.inject_test_agent(story_id, "coder-1", AgentStatus::Failed); + + let found = pool.run_watchdog_pass(Some(root)).await; + assert!(found >= 1, "reap should count the stale Failed entry"); + + // A fresh entry must exist for the same story — the agent respawned. + let agents = pool.agents.try_lock().unwrap(); + let key = composite_key(story_id, "coder-1"); + let agent = agents.get(&key).expect("agent must have been respawned"); + assert_ne!( + agent.status, + AgentStatus::Failed, + "respawned entry must not still be Failed" + ); +} + +/// AC2: reaping a stale Failed entry bumps the story's retry count via the +/// existing should_block_story path. +#[tokio::test] +async fn reap_increments_retry_count() { + crate::db::ensure_content_store(); + crate::crdt_state::init_for_test(); + + let tmp = tempfile::tempdir().unwrap(); + let root = tmp.path(); + + write_project_config( + root, + r#" +max_retries = 5 + +[[agent]] +name = "coder-1" +runtime = "claude-code" +"#, + ); + + let story_id = "1198_story_retry"; + crate::db::write_content( + crate::db::ContentKey::Story(story_id), + "---\nname: Retry Story\n---\n", + ); + crate::crdt_state::write_item_str(story_id, "2_current", Some("Retry Story"), None, None, None); + + let pool = AgentPool::new_test(3001); + pool.inject_test_agent(story_id, "coder-1", AgentStatus::Failed); + + pool.run_watchdog_pass(Some(root)).await; + + let item = crate::crdt_state::read_item(story_id).expect("story must be in CRDT"); + assert_eq!( + item.retry_count(), + 1, + "reaping a stale Failed entry must bump retry_count exactly once" + ); +} + +/// AC2: exhausting max_retries via a reap blocks the story via the existing +/// should_block_story path (same mechanism as the limits watchdog). +#[tokio::test] +async fn reap_blocks_story_after_max_retries() { + crate::db::ensure_content_store(); + crate::crdt_state::init_for_test(); + + let tmp = tempfile::tempdir().unwrap(); + let root = tmp.path(); + + write_project_config( + root, + r#" +max_retries = 1 + +[[agent]] +name = "coder-1" +runtime = "claude-code" +"#, + ); + + let story_id = "1198_story_block"; + crate::db::write_content( + crate::db::ContentKey::Story(story_id), + "---\nname: Block Story\n---\n", + ); + crate::crdt_state::write_item_str(story_id, "2_current", Some("Block Story"), None, None, None); + + let pool = AgentPool::new_test(3001); + pool.inject_test_agent(story_id, "coder-1", AgentStatus::Failed); + + pool.run_watchdog_pass(Some(root)).await; + + let item = crate::crdt_state::read_item(story_id).expect("story must be in CRDT"); + assert_eq!( + item.stage().dir_name(), + "blocked", + "story must be blocked after exhausting max_retries=1 via reap" + ); +} + +/// AC3: list_agents never shows a Failed entry after the next watchdog pass +/// reaps it — regardless of whether the story blocks or respawns. +#[tokio::test] +async fn reap_removes_failed_entry_from_list_agents() { + crate::db::ensure_content_store(); + crate::crdt_state::init_for_test(); + + let tmp = tempfile::tempdir().unwrap(); + let root = tmp.path(); + + write_project_config( + root, + r#" +max_retries = 1 + +[[agent]] +name = "coder-1" +runtime = "claude-code" +"#, + ); + + let story_id = "1198_story_listing"; + crate::db::write_content( + crate::db::ContentKey::Story(story_id), + "---\nname: Listing Story\n---\n", + ); + crate::crdt_state::write_item_str( + story_id, + "2_current", + Some("Listing Story"), + None, + None, + None, + ); + + let pool = AgentPool::new_test(3001); + pool.inject_test_agent(story_id, "coder-1", AgentStatus::Failed); + + pool.run_watchdog_pass(Some(root)).await; + + let listed = pool.list_agents().await.unwrap(); + assert!( + !listed + .iter() + .any(|a| a.story_id == story_id && a.status == AgentStatus::Failed), + "list_agents must not show a Failed entry for '{story_id}' after the next watchdog pass" + ); +} + +/// The limits-termination path (turn/budget overrun) must not have its +/// retry_count double-bumped by the reap pass running in the same +/// `run_watchdog_pass` call. +#[tokio::test] +async fn reap_does_not_double_bump_limits_terminated_agent() { + crate::db::ensure_content_store(); + crate::crdt_state::init_for_test(); + + let tmp = tempfile::tempdir().unwrap(); + let root = tmp.path(); + + write_project_config( + root, + r#" +max_retries = 5 + +[[agent]] +name = "coder-1" +runtime = "claude-code" +max_turns = 10 +"#, + ); + + let story_id = "1198_story_no_double_bump"; + crate::db::write_content( + crate::db::ContentKey::Story(story_id), + "---\nname: No Double Bump\n---\n", + ); + crate::crdt_state::write_item_str( + story_id, + "2_current", + Some("No Double Bump"), + None, + None, + None, + ); + + write_fake_session_log(root, story_id, "coder-1", "sess-overrun", 12); + + let pool = AgentPool::new_test(3001); + pool.inject_test_agent_with_session(story_id, "coder-1", AgentStatus::Running, "sess-overrun"); + + pool.run_watchdog_pass(Some(root)).await; + + let item = crate::crdt_state::read_item(story_id).expect("story must be in CRDT"); + assert_eq!( + item.retry_count(), + 1, + "a single limit-termination pass must bump retry_count by exactly 1, \ + not twice (once from the limits branch, once from reap)" + ); +}