From d64e78b788d4280e9f1866a9bef8a32c2c5eede1 Mon Sep 17 00:00:00 2001 From: Shivakumar Date: Wed, 19 Aug 2026 09:59:54 +0530 Subject: [PATCH 1/2] Adds a dispatch-attempt ceiling to the daemon's auto-redispatch loop, per the item's ask: after `DISPATCH_FAILURE_CAP` (3) consecutive dispatch cycles ending with the same normalized terminal failure reason, the daemon stops auto-redispatching and surfaces the item for manual/PM review instead of retry-looping indefinitely. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - New `src/dispatch_failure_ceiling.rs`: parses `## supervisor — dispatched` / `## agentflare work — failed` / `## agentflare work — complete` comment markers into per-cycle failure reasons, counts the consecutive-identical streak (whitespace-normalized so formatting-only diffs still match), resets on a differing reason or a success. - `dashboard::orphan_reconcile::handle_terminal_job_failure` now checks the streak: below the cap, restores `ready-for-work` (unchanged behavior); at/above the cap, swaps to `needs-manual-dispatch` (or leaves it off `ready-for-work` if that label doesn't exist) and posts a `## supervisor — identical failure cap reached` comment with the last failure reason, so a human knows to `item action=redispatch` after fixing the root cause. - `supervisor::dispatch_item` now emits its dispatch-marker comment via a shared constant instead of an ad hoc string literal, so the ceiling's comment-parsing can't silently drift out of sync. `cli::work::release_and_comment`'s failure-marker stays a literal string (a comment there documents the sync requirement) rather than importing the constant, because `src/cli/work.rs` is LOC-frozen at exactly 2100 lines and any net-positive change to it is rejected by the repo's LOC gate. - Also fixed the same label/assignee-restore ordering bug in `handle_terminal_job_failure` that a same-day review found in the sibling `restore_ready_for_work` function (separate PR #556, unmerged as of this writing): `assignee_agent` is now restored before the `ready-for-work` label is added, so a DB failure between the two calls can't strand the item labeled-ready-but-unassigned (item #150's failure class). This continues WIP a prior cursor dispatch on this item left uncommitted in the worktree (real progress — the core module and its 5 unit tests were already correct and unchanged here). I reviewed it, fixed a clippy `collapsible_if` lint and a formatting nit, applied the ordering fix above, unstaged an unrelated LOC-over-budget hunk on `work.rs` that was staged from the earlier attempt, and verified: `cargo build --lib`, `cargo clippy --tests -- -A unsafe_code -A clippy::pedantic -D warnings`, and `cargo fmt --check` all clean; `dispatch_failure_ceiling::tests` (5/5), `dashboard::orphan_reconcile::tests` (10/10, including 2 new cap-behavior tests), and `supervisor::tests` (28/28) all pass. --- src/dashboard/orphan_reconcile.rs | 532 ++++++++++++++++++++++++++++-- src/dispatch_failure_ceiling.rs | 186 +++++++++++ src/main.rs | 1 + src/supervisor.rs | 13 +- 4 files changed, 696 insertions(+), 36 deletions(-) create mode 100644 src/dispatch_failure_ceiling.rs diff --git a/src/dashboard/orphan_reconcile.rs b/src/dashboard/orphan_reconcile.rs index b438a21b..1c8ce423 100644 --- a/src/dashboard/orphan_reconcile.rs +++ b/src/dashboard/orphan_reconcile.rs @@ -122,21 +122,23 @@ fn restore_ready_for_work(mcp: &crate::mcp_server::AgentflareMcp, item_id: &str, /// label swap. Left alone the item stays labeled `dispatched` forever, /// invisible to `run_discovery_tick` (item #463). /// -/// Swaps to `needs-manual-dispatch` rather than back to `ready-for-work` -/// when that label exists on the project: a clean retry-exhaustion is more -/// often an unretryable problem (bad credentials, no billing balance) than -/// a transient one, and blindly re-queueing would just retry-loop against -/// the same broken agent. Falls back to `ready-for-work` when -/// `needs-manual-dispatch` hasn't been created for this project (unlike -/// that label, `ready-for-work` is guaranteed to exist -- a project can -/// only ever have reached `dispatch_item` by already having it) so the item -/// never ends up worse off than before this hook existed: still stuck, just -/// unlabeled. +/// Below `dispatch_failure_ceiling::DISPATCH_FAILURE_CAP` consecutive +/// dispatch cycles with the same terminal failure reason (one cycle per +/// `## supervisor — dispatched` comment — intra-job retries within a cycle +/// do not increment the count; see that module), swaps back to +/// `ready-for-work` so a transient failure can auto-redispatch on the next +/// discovery tick. At or above the cap, stops auto-redispatch: lands on +/// `needs-manual-dispatch` when that label exists on the project, otherwise +/// leaves the item off `ready-for-work` with a supervisor cap comment so a +/// human/PM must intervene (`item action=redispatch` after fixing the root +/// cause). Orphan-restart recovery (`restore_ready_for_work` above) +/// deliberately does not apply this cap — a daemon death mid-job is not +/// evidence of a deterministic failure class. pub(super) fn handle_terminal_job_failure(job: &agentflare_jobs::AgentJob) { if !job.in_process { return; } - let (Some(item_id), Some(_agent)) = (job.args.first(), job.args.get(1)) else { + let (Some(item_id), Some(agent)) = (job.args.first(), job.args.get(1)) else { return; }; let mcp = match job.args.get(2) { @@ -145,7 +147,7 @@ pub(super) fn handle_terminal_job_failure(job: &agentflare_jobs::AgentJob) { } None => crate::mcp_server::AgentflareMcp::default(), }; - let _ = mcp.with_backend_db(|conn| -> Option<()> { + let cap_reached = mcp.with_backend_db(|conn| -> Option { let item = agentflare_backend::item::get(conn, item_id).ok()?; let state = agentflare_backend::state::get(conn, &item.state_id).ok()?; if matches!(state.group_name.as_str(), "completed" | "cancelled") { @@ -159,16 +161,71 @@ pub(super) fn handle_terminal_job_failure(job: &agentflare_jobs::AgentJob) { { let _ = agentflare_backend::item::remove_label(conn, item_id, &dispatched_id.id); } - let target_id = &labels + + let comments = agentflare_backend::comment::list_by_item(conn, item_id).ok()?; + let failure_count = + crate::dispatch_failure_ceiling::consecutive_identical_failure_count(&comments); + let at_cap = failure_count >= crate::dispatch_failure_ceiling::DISPATCH_FAILURE_CAP; + + if at_cap { + if let Some(manual_id) = labels + .iter() + .find(|l| l.name == crate::supervisor::NEEDS_MANUAL_LABEL) + { + agentflare_backend::item::add_label(conn, item_id, &manual_id.id).ok(); + } + Some(true) + } else if let Some(ready_id) = labels .iter() - .find(|l| l.name == crate::supervisor::NEEDS_MANUAL_LABEL) - .or_else(|| { - labels - .iter() - .find(|l| l.name == crate::supervisor::READY_LABEL) - })? - .id; - agentflare_backend::item::add_label(conn, item_id, target_id).ok() + .find(|l| l.name == crate::supervisor::READY_LABEL) + { + // `release_and_comment` clears assignee via `item_release`; without + // restoring it the next discovery tick hits `skip_item` (item + // #150). Restore it before adding the label so a mid-failure here + // never leaves the item labeled ready-for-work with no assignee. + agentflare_backend::item::update( + conn, + item_id, + agentflare_backend::item::UpdateItem { + assignee_agent: Some(agent.clone()), + ..Default::default() + }, + ) + .ok()?; + agentflare_backend::item::add_label(conn, item_id, &ready_id.id).ok(); + Some(false) + } else { + None + } + }); + + let Ok(Some(true)) = cap_reached else { + return; + }; + + let reason_preview = mcp + .with_backend_db(|conn| { + let comments = agentflare_backend::comment::list_by_item(conn, item_id).ok()?; + crate::dispatch_failure_ceiling::latest_failure_reason(&comments) + }) + .ok() + .flatten(); + + let cap = crate::dispatch_failure_ceiling::DISPATCH_FAILURE_CAP; + let mut body = format!( + "{}\n\n{cap} consecutive identical failures detected — auto-redispatch stopped. \ + Review the failure comments, fix the underlying issue, then \ + `item action=redispatch` to retry.", + crate::dispatch_failure_ceiling::DISPATCH_FAILURE_CAP_MARKER, + ); + if let Some(reason) = reason_preview { + body.push_str(&format!("\n\nLast failure: `{reason}`")); + } + let _ = mcp.comment_impl(crate::mcp_server::types::CommentRequest { + action: "create".into(), + item_id: Some(item_id.clone()), + body: Some(body), + ..Default::default() }); } @@ -644,14 +701,91 @@ mod tests { .unwrap() } - /// Item #463: a job that runs to completion and cleanly fails after + fn seed_dispatch_cycle_failures( + mcp: &crate::mcp_server::AgentflareMcp, + item_id: &str, + cycles: u32, + reason: &str, + ) { + let failure_body = format!( + "{}\n\n{reason}", + crate::dispatch_failure_ceiling::WORK_FAILURE_MARKER + ); + mcp.with_backend_db(|conn| { + for i in 0..cycles { + agentflare_backend::comment::create( + conn, + item_id, + "test", + &format!( + "{}\n\njob: cycle-{i}", + crate::dispatch_failure_ceiling::DISPATCH_MARKER + ), + ) + .unwrap(); + // Second-resolution timestamps + random nanoid ids can reorder + // comments posted in the same second — stagger so dispatch + // always precedes its failure in `list_by_item` order. + std::thread::sleep(std::time::Duration::from_secs(1)); + agentflare_backend::comment::create(conn, item_id, "test", &failure_body).unwrap(); + if i + 1 < cycles { + std::thread::sleep(std::time::Duration::from_secs(1)); + } + } + Some(()) + }) + .unwrap(); + } + + fn wait_for_terminal_job(queue: &Queue, id: &str) -> agentflare_jobs::JobInfo { + for _ in 0..500 { + if let Ok(info) = queue.get(id) + && info.state.is_terminal() + { + return info; + } + std::thread::sleep(std::time::Duration::from_millis(10)); + } + panic!("job {id} did not reach terminal state"); + } + + /// Mirrors `WorkItemExecutor`'s per-attempt failure path: each queue + /// retry calls `release_and_comment` before returning a retryable error. + struct RetryAwareFailingExecutor { + mcp: std::sync::Arc, + reason: String, + } + + impl agentflare_jobs::InProcessExecutor for RetryAwareFailingExecutor { + fn execute( + &self, + job_id: &str, + args: &[String], + _log: &mut dyn std::io::Write, + ) -> Result<(), agentflare_jobs::JobFailure> { + let Some(item_id) = args.first() else { + return Err("malformed in-process work job: missing item_id".into()); + }; + let agent = args.get(1).map(String::as_str).unwrap_or("claude-code"); + let owner = format!("{agent}:{job_id}"); + crate::claims::with_owner_override(owner, || { + crate::cli::work::release_and_comment(&self.mcp, item_id, &self.reason, None); + }); + Err(agentflare_jobs::JobFailure { + message: self.reason.clone(), + retry_after_secs: None, + fatal: false, + }) + } + } + + /// Item #463/#506: a job that runs to completion and cleanly fails after /// exhausting `max_retries` is *not* orphaned (the process never died), /// so `reconcile_orphaned_jobs` above never sees it. Without this hook /// the item stays labeled `dispatched` forever -- invisible to - /// `run_discovery_tick`, silently undispatchable. Confirms - /// `handle_terminal_job_failure` swaps `dispatched` for - /// `needs-manual-dispatch` (not straight back to `ready-for-work`) when - /// that label exists, so a broken agent doesn't just retry-loop. + /// `run_discovery_tick`, silently undispatchable. After + /// `DISPATCH_FAILURE_CAP` consecutive identical dispatch cycles, lands on + /// `needs-manual-dispatch` when that label exists. #[test] fn handle_terminal_job_failure_swaps_dispatched_for_needs_manual_dispatch() { crate::paths::test_support::with_temp_home(|| { @@ -671,6 +805,12 @@ mod tests { ); let dispatched_id = &label_ids[crate::supervisor::DISPATCHED_LABEL]; let item_id = create_dispatched_item(&mcp, dispatched_id); + seed_dispatch_cycle_failures( + &mcp, + &item_id, + crate::dispatch_failure_ceiling::DISPATCH_FAILURE_CAP, + "judge reply was not valid JSON", + ); let job = agentflare_jobs::AgentJob::new("agentflare-work") .args([ @@ -688,7 +828,7 @@ mod tests { .unwrap(); assert!( labels.contains(&label_ids[crate::supervisor::NEEDS_MANUAL_LABEL]), - "a clean retry-exhaustion must land on needs-manual-dispatch, not silently vanish (item #463)" + "identical-failure cap must land on needs-manual-dispatch (item #506)" ); assert!( !labels.contains(dispatched_id), @@ -696,17 +836,68 @@ mod tests { ); assert!( !labels.contains(&label_ids[crate::supervisor::READY_LABEL]), - "needs-manual-dispatch exists on this project, so ready-for-work is the wrong \ - target -- it would just retry-loop against the same broken agent" + "at the identical-failure cap, ready-for-work would just retry-loop" + ); + let comments = mcp + .with_backend_db(|conn| agentflare_backend::comment::list_by_item(conn, &item_id)) + .unwrap() + .unwrap(); + assert!( + comments.iter().any(|c| c + .body + .starts_with(crate::dispatch_failure_ceiling::DISPATCH_FAILURE_CAP_MARKER)), + "cap trip must post a supervisor comment for PM review" ); }); } + /// Below the cap, a project with `needs-manual-dispatch` still gets + /// `ready-for-work` so a transient failure can auto-redispatch. + #[test] + fn handle_terminal_job_failure_restores_ready_for_work_below_identical_failure_cap() { + crate::paths::test_support::with_temp_home(|| { + let tmp = tempfile::tempdir().unwrap(); + let repo_root = tmp.path().join("repo"); + std::fs::create_dir_all(&repo_root).unwrap(); + init_test_repo(&repo_root); + + let mcp = crate::mcp_server::AgentflareMcp::for_project_dir(repo_root.clone()); + let label_ids = seed_labels( + &mcp, + &[ + crate::supervisor::READY_LABEL, + crate::supervisor::DISPATCHED_LABEL, + crate::supervisor::NEEDS_MANUAL_LABEL, + ], + ); + let dispatched_id = &label_ids[crate::supervisor::DISPATCHED_LABEL]; + let item_id = create_dispatched_item(&mcp, dispatched_id); + seed_dispatch_cycle_failures(&mcp, &item_id, 1, "transient network blip"); + + let job = agentflare_jobs::AgentJob::new("agentflare-work") + .args([ + item_id.clone(), + "claude-code".to_string(), + repo_root.to_string_lossy().to_string(), + ]) + .in_process(); + + handle_terminal_job_failure(&job); + + let labels = mcp + .with_backend_db(|conn| agentflare_backend::item::list_labels(conn, &item_id)) + .unwrap() + .unwrap(); + assert!(labels.contains(&label_ids[crate::supervisor::READY_LABEL])); + assert!(!labels.contains(&label_ids[crate::supervisor::NEEDS_MANUAL_LABEL])); + }); + } + /// Same clean-failure path as above, but the project never created a /// `needs-manual-dispatch` label (that label is only ever added by hand - /// -- unlike `ready-for-work`/`dispatched`, nothing seeds it). Falling - /// back to `ready-for-work` keeps the item dispatchable instead of - /// leaving it stuck exactly as before this hook existed. + /// -- unlike `ready-for-work`/`dispatched`, nothing seeds it). Below the + /// identical-failure cap, falling back to `ready-for-work` keeps the + /// item dispatchable. #[test] fn handle_terminal_job_failure_falls_back_to_ready_for_work_without_needs_manual_label() { crate::paths::test_support::with_temp_home(|| { @@ -749,6 +940,283 @@ mod tests { }); } + /// At the identical-failure cap without a `needs-manual-dispatch` label, + /// the item must not go back to `ready-for-work` (that was the unbounded + /// retry-loop path before item #506). + #[test] + fn handle_terminal_job_failure_stops_auto_redispatch_at_cap_without_needs_manual_label() { + crate::paths::test_support::with_temp_home(|| { + let tmp = tempfile::tempdir().unwrap(); + let repo_root = tmp.path().join("repo"); + std::fs::create_dir_all(&repo_root).unwrap(); + init_test_repo(&repo_root); + + let mcp = crate::mcp_server::AgentflareMcp::for_project_dir(repo_root.clone()); + let label_ids = seed_labels( + &mcp, + &[ + crate::supervisor::READY_LABEL, + crate::supervisor::DISPATCHED_LABEL, + ], + ); + let dispatched_id = &label_ids[crate::supervisor::DISPATCHED_LABEL]; + let item_id = create_dispatched_item(&mcp, dispatched_id); + seed_dispatch_cycle_failures( + &mcp, + &item_id, + crate::dispatch_failure_ceiling::DISPATCH_FAILURE_CAP, + "same deterministic error", + ); + + let job = agentflare_jobs::AgentJob::new("agentflare-work") + .args([ + item_id.clone(), + "claude-code".to_string(), + repo_root.to_string_lossy().to_string(), + ]) + .in_process(); + + handle_terminal_job_failure(&job); + + let labels = mcp + .with_backend_db(|conn| agentflare_backend::item::list_labels(conn, &item_id)) + .unwrap() + .unwrap(); + assert!( + !labels.contains(&label_ids[crate::supervisor::READY_LABEL]), + "at cap the item must not be auto-redispatchable" + ); + assert!(!labels.contains(dispatched_id)); + }); + } + + /// Item #506: default `max_retries = 3` posts four identical failure + /// comments within one dispatch cycle. The cap must count that as one + /// cycle — first terminal hook restores `ready-for-work`, not + /// `needs-manual-dispatch`. + #[test] + fn handle_terminal_job_failure_after_default_intra_job_retries_still_auto_redispatches() { + crate::paths::test_support::with_temp_home(|| { + let tmp = tempfile::tempdir().unwrap(); + let repo_root = tmp.path().join("repo"); + std::fs::create_dir_all(&repo_root).unwrap(); + init_test_repo(&repo_root); + + let mcp = std::sync::Arc::new(crate::mcp_server::AgentflareMcp::for_project_dir( + repo_root.clone(), + )); + let label_ids = seed_labels( + &mcp, + &[ + crate::supervisor::READY_LABEL, + crate::supervisor::DISPATCHED_LABEL, + crate::supervisor::NEEDS_MANUAL_LABEL, + ], + ); + let dispatched_id = &label_ids[crate::supervisor::DISPATCHED_LABEL]; + let item_id = create_dispatched_item(&mcp, dispatched_id); + + let reason = "judge reply was not valid JSON"; + mcp.with_backend_db(|conn| { + agentflare_backend::comment::create( + conn, + &item_id, + "test", + &format!( + "{}\n\njob: integration-test", + crate::dispatch_failure_ceiling::DISPATCH_MARKER + ), + ) + .unwrap(); + Some(()) + }) + .unwrap(); + // Second-resolution timestamps can reorder comments posted in the + // same second — ensure dispatch precedes failure comments in + // `list_by_item` order (same stagger as `seed_dispatch_cycle_failures`). + std::thread::sleep(std::time::Duration::from_secs(1)); + + let queue = test_queue(); + let job = agentflare_jobs::AgentJob::new("agentflare-work") + .args([ + item_id.clone(), + "claude-code".to_string(), + repo_root.to_string_lossy().to_string(), + ]) + .in_process(); + let info = queue.enqueue(&job).unwrap(); + + let mut pool = agentflare_jobs::WorkerPool::new(queue.clone()) + .with_executor(std::sync::Arc::new(RetryAwareFailingExecutor { + mcp: mcp.clone(), + reason: reason.into(), + })) + .with_terminal_failure_hook(std::sync::Arc::new(|_job_id, job| { + handle_terminal_job_failure(job); + })); + pool.start(1); + + let terminal = wait_for_terminal_job(&queue, &info.id); + pool.shutdown(); + + assert_eq!(terminal.state, JobState::Failed); + assert_eq!( + terminal.retries, 3, + "default max_retries budget must be exhausted before the hook runs" + ); + + let comments = mcp + .with_backend_db(|conn| agentflare_backend::comment::list_by_item(conn, &item_id)) + .unwrap() + .unwrap(); + let failure_comments = comments + .iter() + .filter(|c| { + c.body + .starts_with(crate::dispatch_failure_ceiling::WORK_FAILURE_MARKER) + }) + .count(); + assert_eq!( + failure_comments, 4, + "one dispatch cycle with max_retries=3 posts four failure comments" + ); + assert_eq!( + crate::dispatch_failure_ceiling::consecutive_identical_failure_count(&comments), + 1, + "intra-job retries must not inflate the dispatch-cycle count" + ); + + let labels = mcp + .with_backend_db(|conn| agentflare_backend::item::list_labels(conn, &item_id)) + .unwrap() + .unwrap(); + assert!( + labels.contains(&label_ids[crate::supervisor::READY_LABEL]), + "first dispatch-cycle terminal failure must restore ready-for-work" + ); + assert!( + !labels.contains(&label_ids[crate::supervisor::NEEDS_MANUAL_LABEL]), + "intra-job retries must not trip the cap on the first dispatch cycle" + ); + assert!(!labels.contains(dispatched_id)); + }); + } + + /// Item #506: below the cap, `release_and_comment` clears `assignee_agent` + /// before the terminal hook runs. Without restoring it, the next discovery + /// tick hits `skip_item` even though `ready-for-work` was restored. + #[test] + fn handle_terminal_job_failure_leaves_item_dispatchable_by_discovery_tick() { + crate::paths::test_support::with_temp_home(|| { + let tmp = tempfile::tempdir().unwrap(); + let repo_root = tmp.path().join("repo"); + std::fs::create_dir_all(&repo_root).unwrap(); + init_test_repo(&repo_root); + + let mcp = crate::mcp_server::AgentflareMcp::for_project_dir(repo_root.clone()); + let label_ids = seed_labels( + &mcp, + &[ + crate::supervisor::READY_LABEL, + crate::supervisor::DISPATCHED_LABEL, + crate::supervisor::NEEDS_MANUAL_LABEL, + ], + ); + let dispatched_id = &label_ids[crate::supervisor::DISPATCHED_LABEL]; + let item_id = mcp + .with_backend_db(|conn| { + let project = mcp.resolve_project(conn).unwrap(); + let state = agentflare_backend::state::list_by_project(conn, &project.id) + .unwrap() + .into_iter() + .find(|s| s.is_default) + .unwrap(); + let item = agentflare_backend::item::create( + conn, + agentflare_backend::item::CreateItem { + project_id: project.id, + state_id: state.id, + name: "terminal failure discovery tick test item".into(), + description: Some("do the thing".into()), + priority: None, + parent_id: None, + assignee_agent: Some("claude-code".into()), + sort_order: None, + external_source: None, + external_id: None, + metadata: None, + label_ids: vec![], + assignee_ids: vec![], + dependency_ids: vec![], + }, + ) + .unwrap(); + agentflare_backend::item::add_label(conn, &item.id, dispatched_id).unwrap(); + item.id + }) + .unwrap(); + + let job = agentflare_jobs::AgentJob::new("agentflare-work") + .args([ + item_id.clone(), + "claude-code".to_string(), + repo_root.to_string_lossy().to_string(), + ]) + .in_process(); + let queue = test_queue(); + let info = queue.enqueue(&job).unwrap(); + crate::claims::with_owner_override(format!("claude-code:{}", info.id), || { + let claim_json = mcp + .item_claim(crate::mcp_server::types::ItemRequest { + action: "claim".to_string(), + id: Some(item_id.clone()), + ..Default::default() + }) + .unwrap(); + let claim: serde_json::Value = serde_json::from_str(&claim_json).unwrap(); + assert_eq!(claim["status"], "acquired"); + crate::cli::work::release_and_comment( + &mcp, + &item_id, + "transient network blip", + None, + ); + }); + + handle_terminal_job_failure(&job); + + let item = mcp + .with_backend_db(|conn| agentflare_backend::item::get(conn, &item_id)) + .unwrap() + .unwrap(); + assert_eq!( + item.assignee_agent.as_deref(), + Some("claude-code"), + "below-cap terminal hook must restore assignee_agent after release_and_comment (item #150)" + ); + let labels = mcp + .with_backend_db(|conn| agentflare_backend::item::list_labels(conn, &item_id)) + .unwrap() + .unwrap(); + assert!(labels.contains(&label_ids[crate::supervisor::READY_LABEL])); + assert!(!labels.contains(dispatched_id)); + + let auth_conn = rusqlite::Connection::open_in_memory().unwrap(); + crate::auth_db::migrate(&auth_conn).unwrap(); + let result = crate::supervisor::run_discovery_tick( + &mcp, + &queue, + &auth_conn, + agentflare_resource_gate::Policy::Normal, + ); + assert_eq!( + result.dispatched, 1, + "below-cap terminal failure must auto-redispatch on the next discovery tick (item #506)" + ); + assert_eq!(result.skipped, 0); + }); + } + /// A plain subprocess job (`POST /api/jobs`) has no work item behind it /// at all -- `job.in_process` is the same guard `reconcile_orphaned_jobs` /// uses to skip those. Confirms the hook is a no-op for one rather than diff --git a/src/dispatch_failure_ceiling.rs b/src/dispatch_failure_ceiling.rs new file mode 100644 index 00000000..77de147f --- /dev/null +++ b/src/dispatch_failure_ceiling.rs @@ -0,0 +1,186 @@ +//! Consecutive identical work-failure counting for the daemon's auto-redispatch +//! ceiling (item #506). Comments are the source of truth — same pattern as +//! `supervisor::CI_SELF_REPAIR_MARKER` / `quota::decide::SELF_REPAIR_CAP`. + +/// Prefix on every failure comment from `cli::work::release_and_comment` — +/// keep in sync with that formatter. +pub(crate) const WORK_FAILURE_MARKER: &str = "## agentflare work — failed"; +/// A successful run breaks a consecutive-identical-failure streak. +pub(crate) const WORK_SUCCESS_MARKER: &str = "## agentflare work — complete"; +/// Prefix on a discovery-tick dispatch comment (see `dispatch_item` in +/// `supervisor.rs`) — one marker per dispatch cycle; intra-job retries do +/// not post another. +pub(crate) const DISPATCH_MARKER: &str = "## supervisor — dispatched"; +/// Prefix on the supervisor comment posted when the ceiling trips. +pub(crate) const DISPATCH_FAILURE_CAP_MARKER: &str = + "## supervisor — identical failure cap reached"; + +/// After this many consecutive dispatch cycles whose terminal failure reason +/// is identical/near-identical, the daemon stops swapping an item back to +/// `ready-for-work` for auto-redispatch. +pub(crate) const DISPATCH_FAILURE_CAP: u32 = 3; + +pub(crate) fn failure_reason(body: &str) -> Option<&str> { + let rest = body.strip_prefix(WORK_FAILURE_MARKER)?; + rest.strip_prefix("\n\n").or(Some("")) +} + +/// Near-identical: collapse whitespace so formatting-only diffs still match. +pub(crate) fn normalize_failure_reason(reason: &str) -> String { + reason.split_whitespace().collect::>().join(" ") +} + +/// One entry per dispatch cycle: the normalized terminal failure reason for +/// the segment after each `DISPATCH_MARKER` comment (through the next dispatch +/// marker or end of thread). Intra-job retries only update that segment's +/// reason — they do not add cycles. +fn dispatch_cycle_failure_reasons( + comments: &[agentflare_backend::comment::ItemComment], +) -> Vec { + let dispatch_indices: Vec = comments + .iter() + .enumerate() + .filter(|(_, c)| c.body.starts_with(DISPATCH_MARKER)) + .map(|(i, _)| i) + .collect(); + + let mut reasons = Vec::new(); + for (idx, &start) in dispatch_indices.iter().enumerate() { + let end = dispatch_indices + .get(idx + 1) + .copied() + .unwrap_or(comments.len()); + let segment = &comments[start..end]; + if segment + .iter() + .any(|c| c.body.starts_with(WORK_SUCCESS_MARKER)) + { + reasons.clear(); + continue; + } + let Some(reason) = segment + .iter() + .rev() + .find_map(|c| failure_reason(&c.body).map(normalize_failure_reason)) + else { + continue; + }; + reasons.push(reason); + } + reasons +} + +/// Walks dispatch-cycle terminal failure reasons (oldest-first) from newest +/// backward, counting consecutive cycles whose normalized reason matches the +/// latest one. Stops at the first older cycle with a different reason or at +/// a success comment (which clears the streak when building the cycle list). +pub(crate) fn consecutive_identical_failure_count( + comments: &[agentflare_backend::comment::ItemComment], +) -> u32 { + let cycles = dispatch_cycle_failure_reasons(comments); + let mut signature: Option<&String> = None; + let mut count = 0u32; + for reason in cycles.iter().rev() { + match &signature { + None => { + signature = Some(reason); + count = 1; + } + Some(sig) if *sig == reason => count += 1, + Some(_) => break, + } + } + count +} + +pub(crate) fn latest_failure_reason( + comments: &[agentflare_backend::comment::ItemComment], +) -> Option { + comments + .iter() + .rev() + .find_map(|c| failure_reason(&c.body).map(normalize_failure_reason)) +} + +#[cfg(test)] +mod tests { + use super::*; + use agentflare_backend::comment::ItemComment; + + fn comment(body: &str) -> ItemComment { + ItemComment { + id: "c1".into(), + item_id: "item-1".into(), + author_agent: "test".into(), + body: body.into(), + created_at: 0, + updated_at: 0, + } + } + + #[test] + fn counts_consecutive_identical_dispatch_cycles() { + let err = "judge reply was not valid JSON"; + let comments = vec![ + comment(&format!("{DISPATCH_MARKER}\n\njob: a")), + comment(&format!("{WORK_FAILURE_MARKER}\n\n{err}")), + comment(&format!("{DISPATCH_MARKER}\n\njob: b")), + comment(&format!("{WORK_FAILURE_MARKER}\n\n{err}")), + comment(&format!("{DISPATCH_MARKER}\n\njob: c")), + comment(&format!("{WORK_FAILURE_MARKER}\n\n{err}")), + ]; + assert_eq!(consecutive_identical_failure_count(&comments), 3); + } + + #[test] + fn intra_job_retries_do_not_inflate_the_dispatch_cycle_count() { + let err = "judge reply was not valid JSON"; + let comments = vec![ + comment(&format!("{DISPATCH_MARKER}\n\njob: a")), + comment(&format!("{WORK_FAILURE_MARKER}\n\n{err}")), + comment(&format!("{WORK_FAILURE_MARKER}\n\n{err}")), + comment(&format!("{WORK_FAILURE_MARKER}\n\n{err}")), + comment(&format!("{WORK_FAILURE_MARKER}\n\n{err}")), + ]; + assert_eq!(consecutive_identical_failure_count(&comments), 1); + } + + #[test] + fn different_reason_resets_the_streak() { + let comments = vec![ + comment(&format!("{DISPATCH_MARKER}\n\njob: a")), + comment(&format!("{WORK_FAILURE_MARKER}\n\nerror A")), + comment(&format!("{DISPATCH_MARKER}\n\njob: b")), + comment(&format!("{WORK_FAILURE_MARKER}\n\nerror B")), + ]; + assert_eq!(consecutive_identical_failure_count(&comments), 1); + } + + #[test] + fn success_breaks_the_streak() { + let err = "same error"; + let comments = vec![ + comment(&format!("{DISPATCH_MARKER}\n\njob: a")), + comment(&format!("{WORK_FAILURE_MARKER}\n\n{err}")), + comment(&format!("{WORK_SUCCESS_MARKER}\n\nok")), + comment(&format!("{DISPATCH_MARKER}\n\njob: b")), + comment(&format!("{WORK_FAILURE_MARKER}\n\n{err}")), + ]; + assert_eq!(consecutive_identical_failure_count(&comments), 1); + } + + #[test] + fn whitespace_normalization_treats_near_identical_as_same() { + let comments = vec![ + comment(&format!("{DISPATCH_MARKER}\n\njob: a")), + comment(&format!( + "{WORK_FAILURE_MARKER}\n\njudge reply was not valid JSON" + )), + comment(&format!("{DISPATCH_MARKER}\n\njob: b")), + comment(&format!( + "{WORK_FAILURE_MARKER}\n\njudge reply was not valid JSON" + )), + ]; + assert_eq!(consecutive_identical_failure_count(&comments), 2); + } +} diff --git a/src/main.rs b/src/main.rs index f11e311e..7884084d 100644 --- a/src/main.rs +++ b/src/main.rs @@ -30,6 +30,7 @@ mod dashboard; mod db; mod dev_install; mod dev_vars; +mod dispatch_failure_ceiling; mod doctor; mod errors; mod gateway_integrations; diff --git a/src/supervisor.rs b/src/supervisor.rs index d7951eb9..d17a4e03 100644 --- a/src/supervisor.rs +++ b/src/supervisor.rs @@ -15,9 +15,10 @@ pub(crate) const READY_LABEL: &str = "ready-for-work"; /// the two can't drift, same rationale as `READY_LABEL` above. pub(crate) const DISPATCHED_LABEL: &str = "dispatched"; /// Also read by `dashboard::orphan_reconcile::handle_terminal_job_failure` -/// -- a job that fails cleanly after exhausting its retries lands here -/// rather than back on `READY_LABEL`, so it doesn't just retry-loop against -/// the same broken agent (item #463). +/// -- once `dispatch_failure_ceiling::DISPATCH_FAILURE_CAP` consecutive +/// dispatch cycles end with the same terminal failure reason, it lands here +/// rather than back on `READY_LABEL`, so it doesn't retry-loop against the +/// same broken agent (items #463/#506). pub(crate) const NEEDS_MANUAL_LABEL: &str = "needs-manual-dispatch"; const NEEDS_HUMAN_GATE_LABEL: &str = "needs-human-gate"; @@ -370,7 +371,11 @@ fn dispatch_item( let _ = mcp.comment_impl(CommentRequest { action: "create".into(), item_id: Some(item.id.clone()), - body: Some(format!("## supervisor — dispatched\n\njob: {}", info.id)), + body: Some(format!( + "{}\n\njob: {}", + crate::dispatch_failure_ceiling::DISPATCH_MARKER, + info.id + )), ..Default::default() }); true From 083638b2e60564a5a935c1abe1fc6eb17245ca18 Mon Sep 17 00:00:00 2001 From: Shivakumar Date: Wed, 19 Aug 2026 10:44:39 +0530 Subject: [PATCH 2/2] fix(work): address CodeRabbit findings on the dispatch failure cap (#506) Two real bugs surfaced by CodeRabbit's review of PR #557: - The at-cap branch of handle_terminal_job_failure never restored assignee_agent after release_and_comment cleared it, so the cap comment's own instruction ("item action=redispatch") would fail with "no assignee_agent to redispatch to" unless the caller passed one explicitly. - consecutive_identical_failure_count treated neither a posted cap comment nor an unrecorded-outcome cycle (e.g. an orphan-restart via restore_ready_for_work, which deliberately posts no marker) as a streak boundary. A post-redispatch retry that failed with the same reason would immediately re-trip the cap with zero retry budget, and a benign daemon-restart gap could silently bridge two otherwise unrelated identical-reason cycles into a false consecutive streak. Fixed both, added regression tests for each, and reformatted with cargo fmt. Agentflare-Agent: claude-code Agentflare-Branch: task/506-cap-consecutive-identical-sdd-loop-dispa Agentflare-Item: 506 --- src/dashboard/orphan_reconcile.rs | 68 +++++++++++++++++ src/dispatch_failure_ceiling.rs | 120 +++++++++++++++++++++++++----- 2 files changed, 169 insertions(+), 19 deletions(-) diff --git a/src/dashboard/orphan_reconcile.rs b/src/dashboard/orphan_reconcile.rs index 1c8ce423..be1fe625 100644 --- a/src/dashboard/orphan_reconcile.rs +++ b/src/dashboard/orphan_reconcile.rs @@ -168,6 +168,21 @@ pub(super) fn handle_terminal_job_failure(job: &agentflare_jobs::AgentJob) { let at_cap = failure_count >= crate::dispatch_failure_ceiling::DISPATCH_FAILURE_CAP; if at_cap { + // Same restoration as the below-cap branch: `release_and_comment` + // already cleared `assignee_agent` via `item_release`. The cap + // comment tells a human to run `item action=redispatch`, which + // requires an existing `assignee_agent` unless one is passed + // explicitly — leaving it cleared here would make that + // instruction fail. + agentflare_backend::item::update( + conn, + item_id, + agentflare_backend::item::UpdateItem { + assignee_agent: Some(agent.clone()), + ..Default::default() + }, + ) + .ok(); if let Some(manual_id) = labels .iter() .find(|l| l.name == crate::supervisor::NEEDS_MANUAL_LABEL) @@ -851,6 +866,59 @@ mod tests { }); } + /// Item #506 CodeRabbit follow-up: at cap, `assignee_agent` must be + /// restored too (same as the below-cap branch) so the cap comment's own + /// `item action=redispatch` instruction works without the caller having + /// to pass `assignee_agent` explicitly. + #[test] + fn handle_terminal_job_failure_restores_assignee_agent_at_cap() { + crate::paths::test_support::with_temp_home(|| { + let tmp = tempfile::tempdir().unwrap(); + let repo_root = tmp.path().join("repo"); + std::fs::create_dir_all(&repo_root).unwrap(); + init_test_repo(&repo_root); + + let mcp = crate::mcp_server::AgentflareMcp::for_project_dir(repo_root.clone()); + let label_ids = seed_labels( + &mcp, + &[ + crate::supervisor::READY_LABEL, + crate::supervisor::DISPATCHED_LABEL, + crate::supervisor::NEEDS_MANUAL_LABEL, + ], + ); + let dispatched_id = &label_ids[crate::supervisor::DISPATCHED_LABEL]; + let item_id = create_dispatched_item(&mcp, dispatched_id); + seed_dispatch_cycle_failures( + &mcp, + &item_id, + crate::dispatch_failure_ceiling::DISPATCH_FAILURE_CAP, + "judge reply was not valid JSON", + ); + + let job = agentflare_jobs::AgentJob::new("agentflare-work") + .args([ + item_id.clone(), + "claude-code".to_string(), + repo_root.to_string_lossy().to_string(), + ]) + .in_process(); + + handle_terminal_job_failure(&job); + + let item = mcp + .with_backend_db(|conn| agentflare_backend::item::get(conn, &item_id)) + .unwrap() + .unwrap(); + assert_eq!( + item.assignee_agent.as_deref(), + Some("claude-code"), + "at-cap terminal hook must restore assignee_agent so \ + `item action=redispatch` works without an explicit override" + ); + }); + } + /// Below the cap, a project with `needs-manual-dispatch` still gets /// `ready-for-work` so a transient failure can auto-redispatch. #[test] diff --git a/src/dispatch_failure_ceiling.rs b/src/dispatch_failure_ceiling.rs index 77de147f..1c3d8e12 100644 --- a/src/dispatch_failure_ceiling.rs +++ b/src/dispatch_failure_ceiling.rs @@ -32,11 +32,21 @@ pub(crate) fn normalize_failure_reason(reason: &str) -> String { /// One entry per dispatch cycle: the normalized terminal failure reason for /// the segment after each `DISPATCH_MARKER` comment (through the next dispatch -/// marker or end of thread). Intra-job retries only update that segment's -/// reason — they do not add cycles. +/// marker or end of thread), plus whether that segment is where the cap was +/// last reported. Intra-job retries only update that segment's reason — they +/// do not add cycles. +struct DispatchCycle { + reason: String, + /// This cycle's segment contains `DISPATCH_FAILURE_CAP_MARKER` — the cap + /// was already reported for it. A later streak must not chain across it + /// even if the reason repeats, or a post-redispatch retry would re-trip + /// the cap with no retry budget. + cap_already_reported: bool, +} + fn dispatch_cycle_failure_reasons( comments: &[agentflare_backend::comment::ItemComment], -) -> Vec { +) -> Vec { let dispatch_indices: Vec = comments .iter() .enumerate() @@ -44,7 +54,7 @@ fn dispatch_cycle_failure_reasons( .map(|(i, _)| i) .collect(); - let mut reasons = Vec::new(); + let mut cycles = Vec::new(); for (idx, &start) in dispatch_indices.iter().enumerate() { let end = dispatch_indices .get(idx + 1) @@ -55,7 +65,7 @@ fn dispatch_cycle_failure_reasons( .iter() .any(|c| c.body.starts_with(WORK_SUCCESS_MARKER)) { - reasons.clear(); + cycles.clear(); continue; } let Some(reason) = segment @@ -63,32 +73,57 @@ fn dispatch_cycle_failure_reasons( .rev() .find_map(|c| failure_reason(&c.body).map(normalize_failure_reason)) else { + // No terminal failure recorded for this cycle (e.g. an + // orphan-restart via `restore_ready_for_work`, which + // deliberately posts no marker — a daemon death mid-job is not + // evidence of a deterministic failure class). Its outcome is + // unknown, so it must not silently bridge an identical reason + // across it as if the cycles were adjacent. + cycles.clear(); continue; }; - reasons.push(reason); + let cap_already_reported = segment + .iter() + .any(|c| c.body.starts_with(DISPATCH_FAILURE_CAP_MARKER)); + cycles.push(DispatchCycle { + reason, + cap_already_reported, + }); } - reasons + cycles } /// Walks dispatch-cycle terminal failure reasons (oldest-first) from newest /// backward, counting consecutive cycles whose normalized reason matches the -/// latest one. Stops at the first older cycle with a different reason or at -/// a success comment (which clears the streak when building the cycle list). +/// latest one. Stops at the first older cycle with a different reason, at a +/// success comment, an unrecorded-outcome cycle (both clear the streak when +/// building the cycle list), or a cycle that already reported the cap (so a +/// post-redispatch retry gets a fresh budget instead of re-tripping +/// immediately). pub(crate) fn consecutive_identical_failure_count( comments: &[agentflare_backend::comment::ItemComment], ) -> u32 { let cycles = dispatch_cycle_failure_reasons(comments); - let mut signature: Option<&String> = None; - let mut count = 0u32; - for reason in cycles.iter().rev() { - match &signature { - None => { - signature = Some(reason); - count = 1; - } - Some(sig) if *sig == reason => count += 1, - Some(_) => break, + let mut iter = cycles.iter().rev(); + let Some(newest) = iter.next() else { + return 0; + }; + // Defensive: if the newest cycle's own segment already reported the cap + // (should not happen in practice — the cap comment is only posted after + // this count decides `at_cap`), stop right there rather than scanning + // into an already-resolved streak. + if newest.cap_already_reported { + return 1; + } + let mut count = 1u32; + for cycle in iter { + // A cycle that already reported the cap is a hard boundary — even a + // matching reason must not chain a post-redispatch retry onto an + // already-tripped streak, or it would re-trip with no retry budget. + if cycle.cap_already_reported || cycle.reason != newest.reason { + break; } + count += 1; } count } @@ -169,6 +204,53 @@ mod tests { assert_eq!(consecutive_identical_failure_count(&comments), 1); } + #[test] + fn cap_comment_gives_a_fresh_streak_budget_after_manual_redispatch() { + let err = "judge reply was not valid JSON"; + let comments = vec![ + comment(&format!("{DISPATCH_MARKER}\n\njob: a")), + comment(&format!("{WORK_FAILURE_MARKER}\n\n{err}")), + comment(&format!("{DISPATCH_MARKER}\n\njob: b")), + comment(&format!("{WORK_FAILURE_MARKER}\n\n{err}")), + comment(&format!("{DISPATCH_MARKER}\n\njob: c")), + comment(&format!("{WORK_FAILURE_MARKER}\n\n{err}")), + comment(&format!( + "{DISPATCH_FAILURE_CAP_MARKER}\n\n3 consecutive..." + )), + // Human fixes the root cause and runs `item action=redispatch`; + // the daemon dispatches a new cycle that happens to fail with + // the same normalized reason. + comment(&format!("{DISPATCH_MARKER}\n\njob: d")), + comment(&format!("{WORK_FAILURE_MARKER}\n\n{err}")), + ]; + assert_eq!( + consecutive_identical_failure_count(&comments), + 1, + "a cycle after the cap was reported must not chain onto the \ + already-tripped streak, or the operator gets zero retry budget" + ); + } + + #[test] + fn unrecorded_outcome_cycle_breaks_adjacency() { + let err = "judge reply was not valid JSON"; + let comments = vec![ + comment(&format!("{DISPATCH_MARKER}\n\njob: a")), + comment(&format!("{WORK_FAILURE_MARKER}\n\n{err}")), + // Daemon restart mid-job: `restore_ready_for_work` deliberately + // posts no marker for this cycle. + comment(&format!("{DISPATCH_MARKER}\n\njob: b")), + comment(&format!("{DISPATCH_MARKER}\n\njob: c")), + comment(&format!("{WORK_FAILURE_MARKER}\n\n{err}")), + ]; + assert_eq!( + consecutive_identical_failure_count(&comments), + 1, + "an unrecorded-outcome cycle must not silently bridge two \ + identical-reason cycles into a false consecutive streak" + ); + } + #[test] fn whitespace_normalization_treats_near_identical_as_same() { let comments = vec![