From 7eef7652d374bf34b7ad53d04803dbd8cc657e87 Mon Sep 17 00:00:00 2001 From: Release Repro Date: Fri, 24 Jul 2026 20:57:57 -0400 Subject: [PATCH] fix(agent): harden compaction failure handling Accept every nonblank summary instead of applying an arbitrary length heuristic. Preserve typed compaction failures and their source chains, suppress repeat attempts within one input, and clear the CLI compaction indicator when the existing agent error event arrives. --- .../src/commands/run/run_progress/event.rs | 15 ++ .../src/commands/run/run_progress/mod.rs | 33 ++++ .../run/run_progress/stage_display.rs | 27 ++++ lib/components/fabro-agent/src/compaction.rs | 150 ++++++++---------- lib/components/fabro-agent/src/error.rs | 75 +++++++++ lib/components/fabro-agent/src/lib.rs | 2 +- lib/components/fabro-agent/src/session.rs | 78 ++++++--- .../fabro-workflow/src/handler/llm/api.rs | 1 + 8 files changed, 274 insertions(+), 107 deletions(-) diff --git a/lib/apps/fabro-cli/src/commands/run/run_progress/event.rs b/lib/apps/fabro-cli/src/commands/run/run_progress/event.rs index 91e8edaaf..31aa14b1c 100644 --- a/lib/apps/fabro-cli/src/commands/run/run_progress/event.rs +++ b/lib/apps/fabro-cli/src/commands/run/run_progress/event.rs @@ -164,6 +164,10 @@ pub(super) enum ProgressEvent { preserved_turn_count: u64, tracked_file_count: u64, }, + CompactionFailed { + stage_node_id: String, + error: String, + }, LlmRetry { stage_node_id: String, model: String, @@ -360,6 +364,12 @@ pub(super) fn from_run_event(stored: &RunEvent) -> Option { preserved_turn_count: props.preserved_turn_count as u64, tracked_file_count: props.tracked_file_count as u64, }), + EventBody::AgentError(props) => { + display_compaction_error(&props.error).map(|error| ProgressEvent::CompactionFailed { + stage_node_id: node_id, + error, + }) + } EventBody::AgentLlmRetry(props) => { #[allow( clippy::cast_possible_truncation, @@ -422,6 +432,11 @@ pub(super) fn from_json_line(line: &str) -> Option { from_run_event(&stored) } +fn display_compaction_error(value: &Value) -> Option { + let error = serde_json::from_value::(value.clone()).ok()?; + matches!(&error, fabro_agent::Error::Compaction(_)).then(|| error.to_string()) +} + fn display_value(value: &Value) -> Option { match value { Value::Null => None, diff --git a/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs b/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs index 97799bfc1..a1dd513ee 100644 --- a/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs +++ b/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs @@ -316,6 +316,13 @@ impl ProgressUI { tracked_file_count, ); } + ProgressEvent::CompactionFailed { + stage_node_id, + error, + } => { + self.stage + .on_compaction_failed(renderer, &stage_node_id, &error); + } ProgressEvent::LlmRetry { stage_node_id, model, @@ -679,6 +686,32 @@ mod tests { assert!(ui.stage.active_stages["s1"].compaction_bar.is_none()); } + #[test] + fn compaction_failure_clears_bar() { + let mut ui = ProgressUI::new(true, false); + + emit(&mut ui, stage_started("s1", "Build")); + emit( + &mut ui, + agent_event("s1", AgentEvent::CompactionStarted { + estimated_tokens: 5000, + context_window_size: 8000, + }), + ); + assert!(ui.stage.active_stages["s1"].compaction_bar.is_some()); + + emit( + &mut ui, + agent_event("s1", AgentEvent::Error { + error: fabro_agent::Error::Compaction(fabro_agent::CompactionError::EmptySummary { + summarized_turn_count: 14, + }), + }), + ); + + assert!(ui.stage.active_stages["s1"].compaction_bar.is_none()); + } + #[test] fn handle_json_line_ignores_invalid_json() { let (mut ui, buffer) = capture_ui(false); diff --git a/lib/apps/fabro-cli/src/commands/run/run_progress/stage_display.rs b/lib/apps/fabro-cli/src/commands/run/run_progress/stage_display.rs index b58568b48..c7c8f8ef6 100644 --- a/lib/apps/fabro-cli/src/commands/run/run_progress/stage_display.rs +++ b/lib/apps/fabro-cli/src/commands/run/run_progress/stage_display.rs @@ -474,6 +474,33 @@ impl StageDisplay { } } + pub(super) fn on_compaction_failed( + &mut self, + renderer: &ProgressRenderer, + stage_node_id: &str, + error: &str, + ) { + let message = format!( + "{} compaction failed: {error}", + styles::red_cross(renderer.styles()) + ); + + if renderer.is_tty() { + if let Some(bar) = self + .active_stages + .get_mut(stage_node_id) + .and_then(|stage| stage.compaction_bar.take()) + { + bar.set_style(styles::style_tool_done()); + bar.finish_with_message(message); + } else { + self.insert_info_line_for_stage(renderer, stage_node_id, &message); + } + } else { + renderer.print_line(6, &message); + } + } + pub(super) fn on_llm_retry( &mut self, renderer: &ProgressRenderer, diff --git a/lib/components/fabro-agent/src/compaction.rs b/lib/components/fabro-agent/src/compaction.rs index 0a9e5ef86..e2331144b 100644 --- a/lib/components/fabro-agent/src/compaction.rs +++ b/lib/components/fabro-agent/src/compaction.rs @@ -5,7 +5,7 @@ use fabro_llm::types::{Message as LlmMessage, Request}; use tracing::debug; use crate::agent_profile::AgentProfile; -use crate::error::Error; +use crate::error::{CompactionError, Error}; use crate::event::Emitter; use crate::file_tracker::FileTracker; use crate::history::History; @@ -13,17 +13,6 @@ use crate::types::{AgentEvent, Message}; const APPROX_CHARS_PER_TOKEN: usize = 4; -/// Minimum length, in bytes, of a usable compaction summary after trimming. -/// -/// A summary is traded for many turns of conversation, so anything shorter -/// than a single source file path -/// (`lib/components/fabro-agent/src/compaction.rs` is 43 bytes) cannot be -/// carrying that context forward. The bar is set far below any genuine summary -/// on purpose: this exists to catch degenerate responses, not to judge summary -/// quality. A false refusal leaves the context to keep growing, so the check -/// must never fire on a real summary. -const MIN_SUMMARY_LEN: usize = 32; - #[derive(Debug, Clone, Copy, PartialEq, Eq, strum::IntoStaticStr)] #[strum(serialize_all = "snake_case")] pub(crate) enum ContextEstimateMethod { @@ -158,25 +147,19 @@ function names, error messages, and exact values. Omit pleasantries and conversa let response = llm_client .complete(&summary_request) .await - .map_err(Error::Llm)?; + .map_err(CompactionError::Llm)?; let response_text = response.text(); let summary_text = response_text.trim(); - // Refuse to compact on a degenerate summary. `compact_from` discards the - // summarized turns irreversibly, so an empty or near-empty summary must not - // be traded for them: the preamble below would tell the model a handoff - // summary exists while it actually runs with no history at all. Empty - // completions are provider-independent — a truncated stream, a reasoning - // model that spent its whole budget on reasoning, or a rate-limit edge all - // produce one. Returning here leaves the history intact; the caller turns - // this into an `AgentEvent::Error` and continues the session. - if summary_text.len() < MIN_SUMMARY_LEN { - return Err(Error::InvalidState(format!( - "compaction summary was empty or too short to replace {preserve_start} turns \ - ({} bytes, minimum {MIN_SUMMARY_LEN}); history left intact", - summary_text.len() - ))); + // `compact_from` discards summarized turns irreversibly. Refuse an empty + // response before mutating history; trimming also prevents a + // whitespace-only response from masquerading as a summary. + if summary_text.is_empty() { + return Err(CompactionError::EmptySummary { + summarized_turn_count: preserve_start, + } + .into()); } debug!( @@ -601,9 +584,16 @@ mod tests { if details["estimate_method"] == "local_estimate")); } + struct CompactionTestResult { + result: Result<(), Error>, + history: History, + original_turns: Vec, + events: Vec, + } + /// Run `compact_context` over a fixed four-turn history against a mock /// provider that returns `summary` from the summarization call. - async fn compact_with_summary(summary: &str) -> (Result<(), Error>, History, Vec) { + async fn compact_with_summary(summary: &str) -> CompactionTestResult { let mut history = History::default(); for index in 0..4 { history.push(Message::User { @@ -611,6 +601,7 @@ mod tests { timestamp: SystemTime::now(), }); } + let original_turns = history.to_session_messages(); let provider = Arc::new(MockLlmProvider::new(vec![text_response(summary)])); let client = make_client(provider).await; @@ -639,70 +630,69 @@ mod tests { events.push(event.event); } - (result, history, events) + CompactionTestResult { + result, + history, + original_turns, + events, + } } - fn assert_history_untouched(history: &History) { + fn assert_history_untouched(history: &History, original_turns: &[fabro_types::SessionMessage]) { assert_eq!( - history.turns().len(), - 4, - "history must not be truncated when the summary is rejected" - ); - assert!( - history - .turns() - .iter() - .all(|turn| matches!(turn, Message::User { .. })), - "no summary turn should be inserted when the summary is rejected" + history.to_session_messages(), + original_turns, + "history must remain exactly unchanged when the summary is rejected" ); } #[tokio::test] - async fn compaction_refuses_to_truncate_on_empty_summary() { - let (result, history, events) = compact_with_summary("").await; + async fn compaction_refuses_to_truncate_on_blank_summary() { + for summary in ["", " \n\t \n "] { + let CompactionTestResult { + result, + history, + original_turns, + events, + } = compact_with_summary(summary).await; - let err = result.expect_err("empty summary must not report success"); - assert!( - matches!(&err, Error::InvalidState(message) if message.contains("empty or too short")), - "unexpected error: {err}" - ); + let err = result.expect_err("blank summary must not report success"); + assert!( + matches!( + &err, + Error::Compaction(CompactionError::EmptySummary { + summarized_turn_count: 3, + }) + ), + "unexpected error: {err}" + ); - assert_history_untouched(&history); - assert!( - !events - .iter() - .any(|event| matches!(event, AgentEvent::CompactionCompleted { .. })), - "CompactionCompleted must not be emitted for a rejected summary" - ); + assert_history_untouched(&history, &original_turns); + assert!( + events + .iter() + .any(|event| matches!(event, AgentEvent::CompactionStarted { .. })), + "CompactionStarted should record the attempted summary request" + ); + assert!( + !events + .iter() + .any(|event| matches!(event, AgentEvent::CompactionCompleted { .. })), + "CompactionCompleted must not be emitted for a rejected summary" + ); + } } #[tokio::test] - async fn compaction_refuses_to_truncate_on_whitespace_only_summary() { - let (result, history, events) = compact_with_summary(" \n\t \n ").await; + async fn compaction_accepts_concise_nonempty_summary() { + let CompactionTestResult { + result, + history, + events, + .. + } = compact_with_summary("Brief handoff.").await; - let err = result.expect_err("whitespace-only summary must not report success"); - assert!( - matches!(&err, Error::InvalidState(message) if message.contains("empty or too short")), - "unexpected error: {err}" - ); - - assert_history_untouched(&history); - assert!( - !events - .iter() - .any(|event| matches!(event, AgentEvent::CompactionCompleted { .. })), - "CompactionCompleted must not be emitted for a rejected summary" - ); - } - - #[tokio::test] - async fn compaction_replaces_history_on_normal_summary() { - let (result, history, events) = compact_with_summary( - "## Goal\nAdd a compaction guard.\n\n## Next Steps\nRun the test suite.", - ) - .await; - - result.expect("a normal summary should compact"); + result.expect("a nonempty summary should compact"); let summary_turn = history .turns() @@ -713,7 +703,7 @@ mod tests { }) .expect("compacted history should contain a summary turn"); assert!(summary_turn.contains("A different assistant began this task")); - assert!(summary_turn.contains("Add a compaction guard.")); + assert!(summary_turn.contains("Brief handoff.")); assert!( events diff --git a/lib/components/fabro-agent/src/error.rs b/lib/components/fabro-agent/src/error.rs index be5bbf05e..b274bd4b7 100644 --- a/lib/components/fabro-agent/src/error.rs +++ b/lib/components/fabro-agent/src/error.rs @@ -17,12 +17,28 @@ impl std::fmt::Display for InterruptReason { } } +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, thiserror::Error)] +#[serde(tag = "type", content = "data", rename_all = "snake_case")] +pub enum CompactionError { + #[error("summary request failed: {0}")] + Llm(#[source] LlmError), + + #[error( + "generated summary was empty after trimming; refused to replace \ + {summarized_turn_count} turns and left history intact" + )] + EmptySummary { summarized_turn_count: usize }, +} + #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, thiserror::Error)] #[serde(tag = "type", content = "data", rename_all = "snake_case")] pub enum Error { #[error("LLM error: {0}")] Llm(#[from] LlmError), + #[error("Context compaction failed: {0}")] + Compaction(#[from] CompactionError), + #[error("Session is closed")] SessionClosed, @@ -41,6 +57,7 @@ pub type Result = std::result::Result; #[cfg(test)] mod tests { use fabro_llm::{ProviderErrorDetail, ProviderErrorKind}; + use fabro_util::error; use super::*; @@ -55,6 +72,39 @@ mod tests { assert!(agent_err.to_string().contains("connection refused")); } + #[test] + fn compaction_error_preserves_llm_source_chain() { + let err = Error::Compaction(CompactionError::Llm(LlmError::Network { + message: "connection refused".into(), + source: None, + })); + + let chain = error::collect_chain(&err); + + assert!( + chain.len() >= 3, + "expected agent, compaction, and LLM errors in the source chain: {chain:?}" + ); + assert!( + chain + .last() + .is_some_and(|cause| cause.contains("connection refused")), + "underlying LLM failure missing from source chain: {chain:?}" + ); + } + + #[test] + fn empty_compaction_summary_display() { + let err = Error::Compaction(CompactionError::EmptySummary { + summarized_turn_count: 3, + }); + assert_eq!( + err.to_string(), + "Context compaction failed: generated summary was empty after trimming; \ + refused to replace 3 turns and left history intact" + ); + } + #[test] fn session_closed_display() { let err = Error::SessionClosed; @@ -116,6 +166,16 @@ mod tests { assert_eq!(err.to_string(), deserialized.to_string()); } + #[test] + fn serde_roundtrip_compaction() { + let err = Error::Compaction(CompactionError::EmptySummary { + summarized_turn_count: 3, + }); + let json = serde_json::to_string(&err).unwrap(); + let deserialized: Error = serde_json::from_str(&json).unwrap(); + assert_eq!(err.to_string(), deserialized.to_string()); + } + #[test] fn serde_roundtrip_session_closed() { let err = Error::SessionClosed; @@ -157,6 +217,9 @@ mod tests { message: "refused".into(), source: None, }), + Error::Compaction(CompactionError::EmptySummary { + summarized_turn_count: 3, + }), Error::SessionClosed, Error::InvalidState("reason".into()), Error::ToolExecution("reason".into()), @@ -180,6 +243,18 @@ mod tests { assert_eq!(v["type"], "llm"); } + #[test] + fn serde_tag_format_compaction() { + let err = Error::Compaction(CompactionError::EmptySummary { + summarized_turn_count: 3, + }); + let json = serde_json::to_string(&err).unwrap(); + let v: serde_json::Value = serde_json::from_str(&json).unwrap(); + assert_eq!(v["type"], "compaction"); + assert_eq!(v["data"]["type"], "empty_summary"); + assert_eq!(v["data"]["data"]["summarized_turn_count"], 3); + } + #[test] fn serde_tag_format_session_closed() { let err = Error::SessionClosed; diff --git a/lib/components/fabro-agent/src/lib.rs b/lib/components/fabro-agent/src/lib.rs index ebc732dc2..03a29a3a9 100644 --- a/lib/components/fabro-agent/src/lib.rs +++ b/lib/components/fabro-agent/src/lib.rs @@ -39,7 +39,7 @@ pub use config::{ }; #[cfg(feature = "docker")] pub use docker_sandbox::{DockerSandbox, DockerSandboxOptions}; -pub use error::{Error, InterruptReason, Result}; +pub use error::{CompactionError, Error, InterruptReason, Result}; pub use event::Emitter; pub use fabro_mcp::config::McpServerSettings; pub use fabro_types::SteeringMessage; diff --git a/lib/components/fabro-agent/src/session.rs b/lib/components/fabro-agent/src/session.rs index faf3bb2d7..9daf79ad1 100644 --- a/lib/components/fabro-agent/src/session.rs +++ b/lib/components/fabro-agent/src/session.rs @@ -1408,6 +1408,12 @@ impl Session { text: expanded_input.clone(), }); + // A failed summarization is unlikely to improve within the same agent + // turn. Suppress further attempts until the next user/follow-up input + // so a provider returning empty responses cannot create a paid retry + // loop at both compaction checkpoints. + let mut compaction_failed = false; + loop { // Top-of-loop: if the previous round's interrupt token fired, // swap in a fresh one before draining and rebuilding state. @@ -1470,7 +1476,9 @@ impl Session { .clone(); // Pre-turn compaction: trim context before building the request - self.compact_if_needed().await; + if !compaction_failed { + compaction_failed = self.compact_if_needed().await; + } self.inject_task_reminder_if_needed(); @@ -1811,7 +1819,9 @@ impl Session { }); // Post-response compaction: trim context after appending assistant turn - self.compact_if_needed().await; + if !compaction_failed { + compaction_failed = self.compact_if_needed().await; + } // If no tool calls, natural completion. Consult the optional // completion coordinator: it can return `true` to force one more @@ -1913,7 +1923,12 @@ impl Session { } } - async fn compact_if_needed(&mut self) { + /// Attempt context compaction when the configured threshold is exceeded. + /// + /// Returns `true` when an attempted compaction failed so the current input + /// loop can suppress repeated paid summary calls. The next input starts + /// with a fresh retry opportunity. + async fn compact_if_needed(&mut self) -> bool { let Some(estimate) = check_context_usage( &self.system_prompt, &self.history, @@ -1922,12 +1937,12 @@ impl Session { &self.event_emitter, &self.id, ) else { - return; + return false; }; if !self.config.enable_context_compaction { - return; + return false; } - if let Err(e) = compact_context( + if let Err(error) = compact_context( &mut self.history, &self.llm_client, self.provider_profile.as_ref(), @@ -1939,10 +1954,11 @@ impl Session { ) .await { - self.event_emitter.emit(self.id.clone(), AgentEvent::Error { - error: Error::InvalidState(format!("Context compaction failed: {e}")), - }); + self.event_emitter + .emit(self.id.clone(), AgentEvent::Error { error }); + return true; } + false } fn drain_steering(&mut self) { @@ -2122,6 +2138,7 @@ mod tests { use super::*; use crate::config::{ToolAccess, ToolAccessPolicy, ToolApprovalAdapter, ToolExposureMode}; + use crate::error::CompactionError; use crate::skills::{Skill, make_use_skill_tool}; use crate::subagent::{SubAgentStatus, make_wait_tool}; use crate::test_support::*; @@ -4580,8 +4597,9 @@ mod tests { // provider that errors on complete() but succeeds on stream(). struct StreamOnlyProvider { - responses: Vec, - call_index: AtomicUsize, + responses: Vec, + stream_index: AtomicUsize, + complete_calls: AtomicUsize, } #[async_trait::async_trait] @@ -4591,6 +4609,7 @@ mod tests { } async fn complete(&self, _request: &Request) -> Result { + self.complete_calls.fetch_add(1, Ordering::SeqCst); Err(LlmError::Stream { message: "summarization failed".into(), source: None, @@ -4598,7 +4617,7 @@ mod tests { } async fn stream(&self, _request: &Request) -> Result { - let idx = self.call_index.fetch_add(1, Ordering::SeqCst); + let idx = self.stream_index.fetch_add(1, Ordering::SeqCst); let response = if idx < self.responses.len() { self.responses[idx].clone() } else { @@ -4627,16 +4646,20 @@ mod tests { } let large_input = "x".repeat(400); - let responses = vec![response_with_usage( + let responses = vec![ + response_with_input_tokens( + tool_call_response("nonexistent_tool", "call_1", serde_json::json!({})), + 90, + ), text_response("OK"), - TokenCounts::default(), - )]; + ]; let provider = Arc::new(StreamOnlyProvider { responses, - call_index: AtomicUsize::new(0), + stream_index: AtomicUsize::new(0), + complete_calls: AtomicUsize::new(0), }); - let client = make_client(provider as Arc).await; + let client = make_client(provider.clone() as Arc).await; let registry = ToolRegistry::new(); let profile = Arc::new(TestProfile::with_context_window(registry, 100)); let env = Arc::new(MockSandbox::default()); @@ -4654,15 +4677,20 @@ mod tests { result.is_ok(), "Session should continue despite compaction failure" ); + assert_eq!( + provider.complete_calls.load(Ordering::SeqCst), + 1, + "a failed compaction should suppress retries for the rest of the input" + ); - // Should emit an Error event for the failed compaction + // Should emit the structured compaction error without flattening the + // underlying LLM failure. let mut found_error = false; while let Ok(event) = rx.try_recv() { - if let AgentEvent::Error { error } = &event.event { - let msg = error.to_string(); - if msg.contains("compaction") || msg.contains("summarization") { - found_error = true; - } + if matches!(event.event, AgentEvent::Error { + error: Error::Compaction(CompactionError::Llm(_)), + }) { + found_error = true; } } assert!(found_error, "Should emit Error event for failed compaction"); @@ -4690,9 +4718,7 @@ mod tests { async fn complete(&self, request: &Request) -> Result { *self.captured_complete.lock().unwrap() = Some(request.clone()); - Ok(text_response( - "## Goal\nSummary goes here.\n\n## Progress\nRead /src/main.rs.", - )) + Ok(text_response("## Goal\nSummary goes here.")) } async fn stream(&self, _request: &Request) -> Result { diff --git a/lib/components/fabro-workflow/src/handler/llm/api.rs b/lib/components/fabro-workflow/src/handler/llm/api.rs index b9d7a2b84..3be40891e 100644 --- a/lib/components/fabro-workflow/src/handler/llm/api.rs +++ b/lib/components/fabro-workflow/src/handler/llm/api.rs @@ -137,6 +137,7 @@ fn classify_agent_error(err: fabro_agent::Error, allow_failover: bool) -> AgentA } fabro_agent::Error::Llm(err) => AgentApiErrorDisposition::Terminal(Error::Llm(err)), other @ (fabro_agent::Error::SessionClosed + | fabro_agent::Error::Compaction(_) | fabro_agent::Error::InvalidState(_) | fabro_agent::Error::ToolExecution(_)) => AgentApiErrorDisposition::Terminal( Error::Precondition(format!("Agent session failed: {other}")),