huskies: merge 1198 bug Failed agents are never reaped — a dead pool entry blocks respawn indefinitely
This commit is contained in:
@@ -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<String> = 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
|
||||
|
||||
@@ -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<String>,
|
||||
) -> usize {
|
||||
let dead: Vec<(String, String, String)> = {
|
||||
let mut agents = pool.agents.lock().await;
|
||||
let keys: Vec<String> = 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
|
||||
}
|
||||
@@ -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.
|
||||
///
|
||||
|
||||
@@ -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)"
|
||||
);
|
||||
}
|
||||
Reference in New Issue
Block a user