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}")),