From 6ef8bc28eb9aa18f394e1b461be206dc1e82d78c Mon Sep 17 00:00:00 2001 From: Henry Park Date: Fri, 27 Mar 2026 16:26:16 -0700 Subject: [PATCH] Handle empty tool completions in autonomous jobs --- src/agent/agentic_loop.rs | 93 ++++++++++++- src/agent/dispatcher.rs | 6 + src/llm/mod.rs | 6 +- src/llm/reasoning.rs | 69 +++++++++- src/worker/autonomous_recovery.rs | 149 ++++++++++++++++++++ src/worker/container.rs | 89 +++++++++++- src/worker/job.rs | 88 +++++++++++- src/worker/mod.rs | 1 + tests/e2e_builtin_tool_coverage.rs | 214 ++++++++++++++++++++++++++++- 9 files changed, 698 insertions(+), 17 deletions(-) create mode 100644 src/worker/autonomous_recovery.rs diff --git a/src/agent/agentic_loop.rs b/src/agent/agentic_loop.rs index 27c2ab72..59f89816 100644 --- a/src/agent/agentic_loop.rs +++ b/src/agent/agentic_loop.rs @@ -10,7 +10,9 @@ use std::borrow::Cow; use crate::agent::session::PendingApproval; use crate::error::Error; -use crate::llm::{ChatMessage, FinishReason, Reasoning, ReasoningContext, RespondResult}; +use crate::llm::{ + ChatMessage, FinishReason, Reasoning, ReasoningContext, RespondResult, ResponseMetadata, +}; /// Signal from the delegate indicating how the loop should proceed. pub enum LoopSignal { @@ -38,6 +40,8 @@ pub enum LoopOutcome { Stopped, /// Max iterations exceeded. MaxIterations, + /// Loop terminated early with a clear failure reason. + Failure(String), /// A tool requires user approval before continuing (chat delegate only). NeedApproval(Box), } @@ -103,6 +107,7 @@ pub trait LoopDelegate: Send + Sync { async fn handle_text_response( &self, text: &str, + metadata: ResponseMetadata, reason_ctx: &mut ReasoningContext, ) -> TextAction; @@ -209,7 +214,10 @@ pub async fn run_agentic_loop( consecutive_tool_intent_nudges = 0; } - match delegate.handle_text_response(&text, reason_ctx).await { + match delegate + .handle_text_response(&text, output.metadata, reason_ctx) + .await + { TextAction::Return(outcome) => return Ok(outcome), TextAction::Continue => {} } @@ -279,7 +287,7 @@ pub fn truncate_for_preview(s: &str, max: usize) -> Cow<'_, str> { #[cfg(test)] mod tests { use super::*; - use crate::llm::{RespondOutput, TokenUsage, ToolCall}; + use crate::llm::{RespondOutput, ResponseAnomaly, ResponseMetadata, TokenUsage, ToolCall}; use crate::testing::StubLlm; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; @@ -303,6 +311,7 @@ mod tests { result: RespondResult::Text(text.to_string()), usage: zero_usage(), finish_reason: FinishReason::Stop, + metadata: ResponseMetadata::default(), } } @@ -314,6 +323,7 @@ mod tests { }, usage: zero_usage(), finish_reason: FinishReason::ToolUse, + metadata: ResponseMetadata::default(), } } @@ -391,6 +401,7 @@ mod tests { async fn handle_text_response( &self, text: &str, + _metadata: ResponseMetadata, _reason_ctx: &mut ReasoningContext, ) -> TextAction { TextAction::Return(LoopOutcome::Response(text.to_string())) @@ -508,6 +519,79 @@ mod tests { ); } + #[tokio::test] + async fn test_text_response_metadata_can_fail_fast() { + struct FailOnMalformedResponse; + + #[async_trait] + impl LoopDelegate for FailOnMalformedResponse { + async fn check_signals(&self) -> LoopSignal { + LoopSignal::Continue + } + + async fn before_llm_call( + &self, + _: &mut ReasoningContext, + _: usize, + ) -> Option { + None + } + + async fn call_llm( + &self, + _: &Reasoning, + _: &mut ReasoningContext, + _: usize, + ) -> Result { + Ok(RespondOutput { + result: RespondResult::Text("fallback".to_string()), + usage: zero_usage(), + finish_reason: FinishReason::Stop, + metadata: ResponseMetadata { + anomaly: Some(ResponseAnomaly::EmptyToolCompletion), + }, + }) + } + + async fn handle_text_response( + &self, + _: &str, + metadata: ResponseMetadata, + _: &mut ReasoningContext, + ) -> TextAction { + assert_eq!(metadata.anomaly, Some(ResponseAnomaly::EmptyToolCompletion)); + TextAction::Return(LoopOutcome::Failure( + "malformed tool completion".to_string(), + )) + } + + async fn execute_tool_calls( + &self, + _: Vec, + _: Option, + _: &mut ReasoningContext, + ) -> Result, crate::error::Error> { + Ok(None) + } + } + + let delegate = FailOnMalformedResponse; + let reasoning = stub_reasoning(); + let mut ctx = ReasoningContext::new(); + let outcome = run_agentic_loop( + &delegate, + &reasoning, + &mut ctx, + &AgenticLoopConfig::default(), + ) + .await + .unwrap(); + + assert!( + matches!(outcome, LoopOutcome::Failure(ref reason) if reason == "malformed tool completion") + ); + } + #[tokio::test] async fn test_max_iterations_reached() { struct ContinueDelegate; @@ -535,6 +619,7 @@ mod tests { async fn handle_text_response( &self, _: &str, + _: ResponseMetadata, ctx: &mut ReasoningContext, ) -> TextAction { ctx.messages.push(ChatMessage::assistant("still working")); @@ -671,6 +756,7 @@ mod tests { }, usage: zero_usage(), finish_reason: FinishReason::Length, // response was truncated + metadata: ResponseMetadata::default(), }; let delegate = MockDelegate::new(vec![truncated_output, text_output("Summarized it.")]); let reasoning = stub_reasoning(); @@ -719,6 +805,7 @@ mod tests { }, usage: zero_usage(), finish_reason: FinishReason::Length, + metadata: ResponseMetadata::default(), }; // Three truncated responses, then a text response let delegate = MockDelegate::new(vec![ diff --git a/src/agent/dispatcher.rs b/src/agent/dispatcher.rs index 35f76ed1..d1e9dbf5 100644 --- a/src/agent/dispatcher.rs +++ b/src/agent/dispatcher.rs @@ -219,6 +219,11 @@ impl Agent { reason: format!("Exceeded maximum tool iterations ({max_tool_iterations})"), } .into()), + LoopOutcome::Failure(reason) => Err(crate::error::LlmError::InvalidResponse { + provider: "agent".to_string(), + reason, + } + .into()), LoopOutcome::NeedApproval(pending) => Ok(AgenticLoopResult::NeedApproval { pending }), } } @@ -439,6 +444,7 @@ impl<'a> LoopDelegate for ChatDelegate<'a> { async fn handle_text_response( &self, text: &str, + _metadata: crate::llm::ResponseMetadata, _reason_ctx: &mut ReasoningContext, ) -> TextAction { // Strip internal "[Called tool ...]" text that can leak when diff --git a/src/llm/mod.rs b/src/llm/mod.rs index d681547d..7d3aea31 100644 --- a/src/llm/mod.rs +++ b/src/llm/mod.rs @@ -62,9 +62,9 @@ pub use provider::{ ToolDefinition, ToolResult, generate_tool_call_id, }; pub use reasoning::{ - ActionPlan, Reasoning, ReasoningContext, RespondOutput, RespondResult, SILENT_REPLY_TOKEN, - TOOL_INTENT_NUDGE, TRUNCATED_TOOL_CALL_NOTICE, TokenUsage, ToolSelection, is_silent_reply, - llm_signals_tool_intent, + ActionPlan, Reasoning, ReasoningContext, RespondOutput, RespondResult, ResponseAnomaly, + ResponseMetadata, SILENT_REPLY_TOKEN, TOOL_INTENT_NUDGE, TRUNCATED_TOOL_CALL_NOTICE, + TokenUsage, ToolSelection, is_silent_reply, llm_signals_tool_intent, }; pub use recording::RecordingLlm; pub use registry::{ProviderDefinition, ProviderProtocol, ProviderRegistry}; diff --git a/src/llm/reasoning.rs b/src/llm/reasoning.rs index a0852cef..5f8a45a2 100644 --- a/src/llm/reasoning.rs +++ b/src/llm/reasoning.rs @@ -337,6 +337,23 @@ impl TokenUsage { } } +/// Structured anomaly classification for LLM responses. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ResponseAnomaly { + /// Tool mode was requested, but the provider returned no usable tool calls + /// and no recoverable text content. + EmptyToolCompletion, + /// Text mode returned no usable content after cleaning/truncation. + EmptyTextResponse, +} + +/// Metadata attached to `RespondOutput` so callers can react to malformed +/// provider behavior without inferring it from fallback strings. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct ResponseMetadata { + pub anomaly: Option, +} + /// Result of a response with potential tool calls. /// /// Used by the agent loop to handle tool execution before returning a final response. @@ -359,6 +376,7 @@ pub struct RespondOutput { pub result: RespondResult, pub usage: TokenUsage, pub finish_reason: FinishReason, + pub metadata: ResponseMetadata, } /// Reasoning engine for the agent. @@ -744,6 +762,7 @@ Respond in JSON format: }, usage, finish_reason: response.finish_reason, + metadata: ResponseMetadata::default(), }); } @@ -772,6 +791,7 @@ Respond in JSON format: }, usage, finish_reason: response.finish_reason, + metadata: ResponseMetadata::default(), }); } @@ -785,11 +805,18 @@ Respond in JSON format: // Pre-truncate at tool tags to preserve text before the tag. let pre_truncated = truncate_at_tool_tags(&content); let cleaned = clean_response(&pre_truncated); - let final_text = if cleaned.trim().is_empty() { + let metadata = if cleaned.trim().is_empty() { tracing::warn!( "LLM response was empty after cleaning (original len={}), using fallback", content.len() ); + ResponseMetadata { + anomaly: Some(ResponseAnomaly::EmptyToolCompletion), + } + } else { + ResponseMetadata::default() + }; + let final_text = if metadata.anomaly.is_some() { "I'm not sure how to respond to that.".to_string() } else { cleaned @@ -798,6 +825,7 @@ Respond in JSON format: result: RespondResult::Text(final_text), usage, finish_reason: response.finish_reason, + metadata, }) } else { // No tools, use simple completion @@ -812,11 +840,18 @@ Respond in JSON format: let response = self.llm.complete(request).await?; let pre_truncated = truncate_at_tool_tags(&response.content); let cleaned = clean_response(&pre_truncated); - let final_text = if cleaned.trim().is_empty() { + let metadata = if cleaned.trim().is_empty() { tracing::warn!( "LLM response was empty after cleaning (original len={}), using fallback", response.content.len() ); + ResponseMetadata { + anomaly: Some(ResponseAnomaly::EmptyTextResponse), + } + } else { + ResponseMetadata::default() + }; + let final_text = if metadata.anomaly.is_some() { "I'm not sure how to respond to that.".to_string() } else { cleaned @@ -830,6 +865,7 @@ Respond in JSON format: cache_creation_input_tokens: response.cache_creation_input_tokens, }, finish_reason: response.finish_reason, + metadata, }) } } @@ -3101,9 +3137,38 @@ That's my plan."#; context.force_text = true; let output = reasoning.respond_with_tools(&context).await.unwrap(); + let metadata = output.metadata; match output.result { RespondResult::Text(text) => { assert_eq!(text, "I'm not sure how to respond to that."); + assert_eq!(metadata.anomaly, Some(ResponseAnomaly::EmptyTextResponse)); + } + RespondResult::ToolCalls { .. } => { + panic!("Expected fallback text, not tool calls"); + } + } + } + + #[tokio::test] + async fn test_respond_with_tools_flags_empty_tool_completion() { + use crate::testing::StubLlm; + let llm = Arc::new(StubLlm::new("")); + let reasoning = Reasoning::new(llm); + + let context = ReasoningContext::new() + .with_message(ChatMessage::user("list tools")) + .with_tools(vec![ToolDefinition { + name: "tool_list".to_string(), + description: "Lists tools".to_string(), + parameters: serde_json::json!({}), + }]); + + let output = reasoning.respond_with_tools(&context).await.unwrap(); + let metadata = output.metadata; + match output.result { + RespondResult::Text(text) => { + assert_eq!(text, "I'm not sure how to respond to that."); + assert_eq!(metadata.anomaly, Some(ResponseAnomaly::EmptyToolCompletion)); } RespondResult::ToolCalls { .. } => { panic!("Expected fallback text, not tool calls"); diff --git a/src/worker/autonomous_recovery.rs b/src/worker/autonomous_recovery.rs new file mode 100644 index 00000000..2db2f0c7 --- /dev/null +++ b/src/worker/autonomous_recovery.rs @@ -0,0 +1,149 @@ +use crate::llm::{ResponseAnomaly, ResponseMetadata}; + +pub(crate) const EMPTY_TOOL_COMPLETION_NUDGE: &str = "\ +Your previous tool-enabled response was empty or malformed.\n\ +If you need to use a tool, call it now with valid arguments.\n\ +Otherwise, provide a real status update about work already completed."; + +pub(crate) const FORCE_TEXT_RECOVERY_PROMPT: &str = "\ +Your previous tool-enabled responses were empty or malformed.\n\ +Do not call any more tools in the next reply.\n\ +Instead, provide a concise final status based only on work already completed.\n\ +If the job is complete, say so explicitly. If not, explain what blocked you."; + +pub(crate) const EMPTY_TOOL_COMPLETION_FAILURE: &str = "Execution failed: the selected model repeatedly returned empty or malformed tool-completion responses and is not reliable for autonomous tool use."; + +#[derive(Debug, Default, Clone, Copy)] +pub(crate) struct AutonomousRecoveryState { + consecutive_empty_tool_completions: u8, + force_text_recovery_pending: bool, + force_text_recovery_active: bool, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum AutonomousRecoveryAction { + Continue, + ToolModeNudge, + ForceTextRecovery, + Fail, +} + +impl AutonomousRecoveryState { + pub(crate) fn begin_iteration(&mut self) -> bool { + if self.force_text_recovery_pending { + self.force_text_recovery_pending = false; + self.force_text_recovery_active = true; + true + } else { + self.force_text_recovery_active + } + } + + pub(crate) fn on_text_response( + &mut self, + metadata: ResponseMetadata, + text: &str, + ) -> AutonomousRecoveryAction { + match metadata.anomaly { + Some(ResponseAnomaly::EmptyToolCompletion) => { + self.consecutive_empty_tool_completions += 1; + self.force_text_recovery_active = false; + match self.consecutive_empty_tool_completions { + 1 => AutonomousRecoveryAction::ToolModeNudge, + 2 => { + self.force_text_recovery_pending = true; + AutonomousRecoveryAction::ForceTextRecovery + } + _ => AutonomousRecoveryAction::Fail, + } + } + Some(ResponseAnomaly::EmptyTextResponse) if self.force_text_recovery_active => { + self.force_text_recovery_active = false; + AutonomousRecoveryAction::Fail + } + _ if !text.trim().is_empty() => { + self.reset(); + AutonomousRecoveryAction::Continue + } + _ => AutonomousRecoveryAction::Continue, + } + } + + pub(crate) fn on_valid_tool_call(&mut self) { + self.reset(); + } + + fn reset(&mut self) { + self.consecutive_empty_tool_completions = 0; + self.force_text_recovery_pending = false; + self.force_text_recovery_active = false; + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn metadata(anomaly: ResponseAnomaly) -> ResponseMetadata { + ResponseMetadata { + anomaly: Some(anomaly), + } + } + + #[test] + fn first_empty_tool_completion_issues_nudge() { + let mut state = AutonomousRecoveryState::default(); + let action = state.on_text_response( + metadata(ResponseAnomaly::EmptyToolCompletion), + "I'm not sure how to respond to that.", + ); + assert_eq!(action, AutonomousRecoveryAction::ToolModeNudge); + assert!(!state.begin_iteration()); + } + + #[test] + fn second_empty_tool_completion_schedules_text_recovery() { + let mut state = AutonomousRecoveryState::default(); + let _ = state.on_text_response(metadata(ResponseAnomaly::EmptyToolCompletion), "fallback"); + let action = + state.on_text_response(metadata(ResponseAnomaly::EmptyToolCompletion), "fallback"); + assert_eq!(action, AutonomousRecoveryAction::ForceTextRecovery); + assert!(state.begin_iteration()); + } + + #[test] + fn forced_text_recovery_fallback_fails() { + let mut state = AutonomousRecoveryState::default(); + let _ = state.on_text_response(metadata(ResponseAnomaly::EmptyToolCompletion), "fallback"); + let _ = state.on_text_response(metadata(ResponseAnomaly::EmptyToolCompletion), "fallback"); + assert!(state.begin_iteration()); + let action = + state.on_text_response(metadata(ResponseAnomaly::EmptyTextResponse), "fallback"); + assert_eq!(action, AutonomousRecoveryAction::Fail); + } + + #[test] + fn valid_tool_call_resets_counter() { + let mut state = AutonomousRecoveryState::default(); + let _ = state.on_text_response(metadata(ResponseAnomaly::EmptyToolCompletion), "fallback"); + state.on_valid_tool_call(); + let action = + state.on_text_response(metadata(ResponseAnomaly::EmptyToolCompletion), "fallback"); + assert_eq!(action, AutonomousRecoveryAction::ToolModeNudge); + } + + #[test] + fn meaningful_text_after_text_recovery_resets_state() { + let mut state = AutonomousRecoveryState::default(); + let _ = state.on_text_response(metadata(ResponseAnomaly::EmptyToolCompletion), "fallback"); + let _ = state.on_text_response(metadata(ResponseAnomaly::EmptyToolCompletion), "fallback"); + assert!(state.begin_iteration()); + + let action = state.on_text_response(ResponseMetadata::default(), "Still working on step 2"); + assert_eq!(action, AutonomousRecoveryAction::Continue); + + let next = + state.on_text_response(metadata(ResponseAnomaly::EmptyToolCompletion), "fallback"); + assert_eq!(next, AutonomousRecoveryAction::ToolModeNudge); + } +} diff --git a/src/worker/container.rs b/src/worker/container.rs index 5d8e03b5..efb27e45 100644 --- a/src/worker/container.rs +++ b/src/worker/container.rs @@ -21,11 +21,15 @@ use crate::agent::agentic_loop::{ use crate::config::SafetyConfig; use crate::context::JobContext; use crate::error::WorkerError; -use crate::llm::{ChatMessage, LlmProvider, Reasoning, ReasoningContext}; +use crate::llm::{ChatMessage, LlmProvider, Reasoning, ReasoningContext, ResponseMetadata}; use crate::safety::SafetyLayer; use crate::tools::ToolRegistry; use crate::tools::execute::{execute_tool_simple, process_tool_result}; use crate::worker::api::{CompletionReport, JobEventPayload, StatusUpdate, WorkerHttpClient}; +use crate::worker::autonomous_recovery::{ + AutonomousRecoveryAction, AutonomousRecoveryState, EMPTY_TOOL_COMPLETION_FAILURE, + EMPTY_TOOL_COMPLETION_NUDGE, FORCE_TEXT_RECOVERY_PROMPT, +}; use crate::worker::proxy_llm::ProxyLlmProvider; /// Configuration for the worker runtime. @@ -170,6 +174,7 @@ Work independently to complete this job. When finished, your final message MUST extra_env: self.extra_env.clone(), last_output: Mutex::new(String::new()), iteration_tracker: iteration_tracker.clone(), + recovery_state: Mutex::new(AutonomousRecoveryState::default()), }; let config = AgenticLoopConfig { @@ -228,6 +233,24 @@ Work independently to complete this job. When finished, your final message MUST }) .await?; } + Ok(Ok(LoopOutcome::Failure(reason))) => { + tracing::warn!("Worker failed for job {}: {}", self.config.job_id, reason); + self.post_event( + "result", + serde_json::json!({ + "success": false, + "message": reason, + }), + ) + .await; + self.client + .report_complete(&CompletionReport { + success: false, + message: Some(reason), + iterations, + }) + .await?; + } Ok(Ok(LoopOutcome::Stopped | LoopOutcome::NeedApproval(_))) => { tracing::info!("Worker for job {} stopped", self.config.job_id); self.client @@ -304,6 +327,7 @@ struct ContainerDelegate { /// Tracks the current iteration — shared with the outer `run` method so /// `CompletionReport` can include accurate iteration counts. iteration_tracker: Arc>, + recovery_state: Mutex, } impl ContainerDelegate { @@ -377,8 +401,17 @@ impl LoopDelegate for ContainerDelegate { // conversation. Ensure the last message is user-role before calling the LLM. crate::util::ensure_ends_with_user_message(&mut reason_ctx.messages); - // Refresh tools (in case WASM tools were built) - reason_ctx.available_tools = self.tools.tool_definitions().await; + let force_text_recovery = { + let mut recovery = self.recovery_state.lock().await; + recovery.begin_iteration() + }; + if force_text_recovery { + tracing::warn!("Switching to text-only recovery after malformed tool completions"); + reason_ctx.available_tools.clear(); + } else { + // Refresh tools (in case WASM tools were built) + reason_ctx.available_tools = self.tools.tool_definitions().await; + } None } @@ -399,8 +432,53 @@ impl LoopDelegate for ContainerDelegate { async fn handle_text_response( &self, text: &str, + metadata: ResponseMetadata, reason_ctx: &mut ReasoningContext, ) -> TextAction { + let action = { + let mut recovery = self.recovery_state.lock().await; + recovery.on_text_response(metadata, text) + }; + match action { + AutonomousRecoveryAction::ToolModeNudge => { + tracing::warn!("Malformed empty tool completion detected; retrying in tool mode"); + self.post_event( + "status", + serde_json::json!({ + "message": "Model returned an empty tool-completion response; retrying with a stronger tool-use nudge.", + }), + ) + .await; + reason_ctx + .messages + .push(ChatMessage::user(EMPTY_TOOL_COMPLETION_NUDGE)); + return TextAction::Continue; + } + AutonomousRecoveryAction::ForceTextRecovery => { + tracing::warn!( + "Repeated malformed tool completions detected; switching to text-only recovery" + ); + self.post_event( + "status", + serde_json::json!({ + "message": "Model returned repeated empty tool-completion responses; requesting a final status update without tools.", + }), + ) + .await; + reason_ctx + .messages + .push(ChatMessage::user(FORCE_TEXT_RECOVERY_PROMPT)); + return TextAction::Continue; + } + AutonomousRecoveryAction::Fail => { + tracing::warn!("Failing fast after repeated malformed autonomous responses"); + return TextAction::Return(LoopOutcome::Failure( + EMPTY_TOOL_COMPLETION_FAILURE.to_string(), + )); + } + AutonomousRecoveryAction::Continue => {} + } + self.post_event( "message", serde_json::json!({ @@ -431,6 +509,11 @@ impl LoopDelegate for ContainerDelegate { content: Option, reason_ctx: &mut ReasoningContext, ) -> Result, crate::error::Error> { + { + let mut recovery = self.recovery_state.lock().await; + recovery.on_valid_tool_call(); + } + if let Some(ref text) = content { self.post_event( "message", diff --git a/src/worker/job.rs b/src/worker/job.rs index f74d4ec8..7a9b82f4 100644 --- a/src/worker/job.rs +++ b/src/worker/job.rs @@ -23,8 +23,8 @@ use crate::context::{ContextManager, JobState}; use crate::error::Error; use crate::hooks::HookRegistry; use crate::llm::{ - ActionPlan, ChatMessage, LlmProvider, Reasoning, ReasoningContext, RespondResult, ToolCall, - ToolSelection, + ActionPlan, ChatMessage, LlmProvider, Reasoning, ReasoningContext, RespondResult, + ResponseMetadata, ToolCall, ToolSelection, }; use crate::safety::SafetyLayer; use crate::tenant::AdminScope; @@ -33,6 +33,10 @@ use crate::tools::rate_limiter::RateLimitResult; use crate::tools::{ ApprovalContext, ToolRegistry, autonomous_unavailable_error, prepare_tool_params, redact_params, }; +use crate::worker::autonomous_recovery::{ + AutonomousRecoveryAction, AutonomousRecoveryState, EMPTY_TOOL_COMPLETION_FAILURE, + EMPTY_TOOL_COMPLETION_NUDGE, FORCE_TEXT_RECOVERY_PROMPT, +}; use ironclaw_common::AppEvent; /// Shared dependencies for worker execution. @@ -391,6 +395,7 @@ Report when the job is complete or if you encounter issues you cannot resolve."# worker: self, rx: tokio::sync::Mutex::new(rx), consecutive_rate_limits: std::sync::atomic::AtomicUsize::new(0), + recovery_state: tokio::sync::Mutex::new(AutonomousRecoveryState::default()), }; let config = AgenticLoopConfig { @@ -409,6 +414,9 @@ Report when the job is complete or if you encounter issues you cannot resolve."# self.mark_failed("Maximum iterations exceeded: job hit the iteration cap") .await?; } + LoopOutcome::Failure(reason) => { + self.mark_failed(&reason).await?; + } LoopOutcome::Stopped => { // Stop signal handled — nothing more to do } @@ -1109,6 +1117,7 @@ struct JobDelegate<'a> { rx: tokio::sync::Mutex<&'a mut mpsc::Receiver>, /// Tracks consecutive rate-limit errors to fail fast instead of burning iterations. consecutive_rate_limits: std::sync::atomic::AtomicUsize, + recovery_state: tokio::sync::Mutex, } impl<'a> JobDelegate<'a> { @@ -1159,6 +1168,7 @@ impl<'a> JobDelegate<'a> { result: RespondResult::Text(String::new()), usage: crate::llm::TokenUsage::default(), finish_reason: crate::llm::FinishReason::Stop, + metadata: ResponseMetadata::default(), }) } } @@ -1250,8 +1260,21 @@ impl<'a> LoopDelegate for JobDelegate<'a> { reason_ctx: &mut ReasoningContext, _iteration: usize, ) -> Option { - // Refresh tool definitions so newly built tools become visible - reason_ctx.available_tools = self.worker.tools().tool_definitions().await; + let force_text_recovery = { + let mut recovery = self.recovery_state.lock().await; + recovery.begin_iteration() + }; + + if force_text_recovery { + tracing::warn!( + job_id = %self.worker.job_id, + "Switching to text-only recovery after malformed tool completions" + ); + reason_ctx.available_tools.clear(); + } else { + // Refresh tool definitions so newly built tools become visible + reason_ctx.available_tools = self.worker.tools().tool_definitions().await; + } // Claude 4.6 rejects assistant prefill; NEAR AI rejects any non-user-ending // conversation. Ensure the last message is user-role before calling the LLM. @@ -1285,6 +1308,7 @@ impl<'a> LoopDelegate for JobDelegate<'a> { }, usage: crate::llm::TokenUsage::default(), finish_reason: crate::llm::FinishReason::ToolUse, + metadata: ResponseMetadata::default(), }); } Ok(_) => {} // empty selections, fall through @@ -1328,8 +1352,59 @@ impl<'a> LoopDelegate for JobDelegate<'a> { async fn handle_text_response( &self, text: &str, + metadata: ResponseMetadata, reason_ctx: &mut ReasoningContext, ) -> TextAction { + let action = { + let mut recovery = self.recovery_state.lock().await; + recovery.on_text_response(metadata, text) + }; + + match action { + AutonomousRecoveryAction::ToolModeNudge => { + tracing::warn!( + job_id = %self.worker.job_id, + "Malformed empty tool completion detected; retrying in tool mode" + ); + self.worker.log_event( + "status", + serde_json::json!({ + "message": "Model returned an empty tool-completion response; retrying with a stronger tool-use nudge.", + }), + ); + reason_ctx + .messages + .push(ChatMessage::user(EMPTY_TOOL_COMPLETION_NUDGE)); + return TextAction::Continue; + } + AutonomousRecoveryAction::ForceTextRecovery => { + tracing::warn!( + job_id = %self.worker.job_id, + "Repeated malformed tool completions detected; switching to text-only recovery" + ); + self.worker.log_event( + "status", + serde_json::json!({ + "message": "Model returned repeated empty tool-completion responses; requesting a final status update without tools.", + }), + ); + reason_ctx + .messages + .push(ChatMessage::user(FORCE_TEXT_RECOVERY_PROMPT)); + return TextAction::Continue; + } + AutonomousRecoveryAction::Fail => { + tracing::warn!( + job_id = %self.worker.job_id, + "Failing fast after repeated malformed autonomous responses" + ); + return TextAction::Return(LoopOutcome::Failure( + EMPTY_TOOL_COMPLETION_FAILURE.to_string(), + )); + } + AutonomousRecoveryAction::Continue => {} + } + // Empty text from rate-limit backoff retry — skip processing and let the // loop proceed to the next iteration which will re-call the LLM. if text.is_empty() { @@ -1368,6 +1443,11 @@ impl<'a> LoopDelegate for JobDelegate<'a> { content: Option, reason_ctx: &mut ReasoningContext, ) -> Result, crate::error::Error> { + { + let mut recovery = self.recovery_state.lock().await; + recovery.on_valid_tool_call(); + } + if let Some(ref text) = content { self.worker.log_event( "message", diff --git a/src/worker/mod.rs b/src/worker/mod.rs index c6028b96..dc6a2e89 100644 --- a/src/worker/mod.rs +++ b/src/worker/mod.rs @@ -25,6 +25,7 @@ //! ``` pub mod api; +mod autonomous_recovery; pub mod claude_bridge; pub mod container; pub mod job; diff --git a/tests/e2e_builtin_tool_coverage.rs b/tests/e2e_builtin_tool_coverage.rs index 7c0c7bc7..488ea18c 100644 --- a/tests/e2e_builtin_tool_coverage.rs +++ b/tests/e2e_builtin_tool_coverage.rs @@ -11,9 +11,76 @@ mod tests { use std::time::Duration; use ironclaw::agent::routine::{RoutineAction, Trigger}; + use ironclaw::context::{JobContext, JobState}; + use uuid::Uuid; - use crate::support::test_rig::TestRigBuilder; - use crate::support::trace_llm::LlmTrace; + use crate::support::test_rig::{TestRig, TestRigBuilder}; + use crate::support::trace_llm::{LlmTrace, RequestHint, TraceResponse, TraceStep}; + + fn text_step(content: &str) -> TraceStep { + TraceStep { + request_hint: None, + response: TraceResponse::Text { + content: content.to_string(), + input_tokens: 10, + output_tokens: 5, + }, + expected_tool_results: Vec::new(), + } + } + + fn hinted_text_step(content: &str, last_user_message_contains: &str) -> TraceStep { + TraceStep { + request_hint: Some(RequestHint { + last_user_message_contains: Some(last_user_message_contains.to_string()), + min_message_count: None, + }), + response: TraceResponse::Text { + content: content.to_string(), + input_tokens: 10, + output_tokens: 5, + }, + expected_tool_results: Vec::new(), + } + } + + fn extract_job_id(response: &str) -> Uuid { + let id = response + .lines() + .find_map(|line| line.strip_prefix("ID: ")) + .expect("job creation response should include an ID line"); + Uuid::parse_str(id).expect("job ID should be a UUID") + } + + async fn wait_for_job_state(rig: &TestRig, job_id: Uuid, expected: JobState) -> JobContext { + let deadline = tokio::time::Instant::now() + Duration::from_secs(10); + + loop { + if let Some(job) = rig + .database() + .get_job(job_id) + .await + .expect("get_job should succeed") + && job.state == expected + { + return job; + } + + assert!( + tokio::time::Instant::now() < deadline, + "job {job_id} did not reach state {expected:?} before timeout" + ); + + tokio::time::sleep(Duration::from_millis(50)).await; + } + } + + fn requests_contain(requests: &[Vec], needle: &str) -> bool { + requests + .iter() + .flatten() + .any(|message| message.content.contains(needle)) + } // ----------------------------------------------------------------------- // Test 1: time_parse_and_diff @@ -685,6 +752,149 @@ mod tests { rig.shutdown(); } + // ----------------------------------------------------------------------- + // Test 8a: command_job_fails_fast_on_repeated_empty_tool_completions + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn command_job_fails_fast_on_repeated_empty_tool_completions() { + let trace = LlmTrace::single_turn( + "test-empty-tool-recovery-fail", + "(worker only)", + vec![ + text_step(""), + text_step(""), + hinted_text_step("", "valid arguments"), + text_step(""), + hinted_text_step("", "Do not call any more tools in the next reply."), + ], + ); + + let rig = TestRigBuilder::new() + .with_trace(trace) + .with_auto_approve_tools(true) + .build() + .await; + + rig.send_message("/job reproduce empty tool completion loop") + .await; + let create_responses = rig.wait_for_responses(1, Duration::from_secs(15)).await; + let job_id = extract_job_id(&create_responses[0].content); + + let job = wait_for_job_state(&rig, job_id, JobState::Failed).await; + assert_eq!(job.title, "reproduce empty tool completion loop"); + + let failure_reason = rig + .database() + .get_agent_job_failure_reason(job_id) + .await + .expect("get_agent_job_failure_reason should succeed") + .expect("failed job should persist a failure reason"); + assert!( + failure_reason + .contains("repeatedly returned empty or malformed tool-completion responses"), + "unexpected failure reason: {failure_reason}" + ); + assert!( + !failure_reason.contains("max iterations"), + "failure should not surface as iteration exhaustion: {failure_reason}" + ); + + assert_eq!( + rig.llm_call_count(), + 5, + "worker should stop after the bounded recovery flow" + ); + assert!( + !rig.collect_metrics().await.hit_iteration_limit, + "bounded recovery should stop before iteration-limit reporting" + ); + + let requests = rig.captured_llm_requests(); + assert!( + requests_contain(&requests, "call it now with valid arguments"), + "expected targeted tool-mode recovery nudge in worker requests" + ); + assert!( + requests_contain(&requests, "Do not call any more tools in the next reply."), + "expected forced text-only recovery prompt in worker requests" + ); + + rig.clear().await; + rig.send_message(&format!("/status {}", job_id)).await; + let status_responses = rig.wait_for_responses(1, Duration::from_secs(5)).await; + assert!( + status_responses[0].content.contains("Status: Failed"), + "unexpected status response: {:?}", + status_responses[0].content + ); + + rig.shutdown(); + } + + // ----------------------------------------------------------------------- + // Test 8b: command_job_text_recovery_can_complete + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn command_job_text_recovery_can_complete() { + let trace = LlmTrace::single_turn( + "test-empty-tool-recovery-success", + "(worker only)", + vec![ + text_step(""), + text_step(""), + hinted_text_step("", "valid arguments"), + text_step(""), + hinted_text_step( + "The job is complete. I finished the requested work and there is nothing left to do.", + "Do not call any more tools in the next reply.", + ), + ], + ); + + let rig = TestRigBuilder::new() + .with_trace(trace) + .with_auto_approve_tools(true) + .build() + .await; + + rig.send_message("/job recover after malformed tool completions") + .await; + let create_responses = rig.wait_for_responses(1, Duration::from_secs(15)).await; + let job_id = extract_job_id(&create_responses[0].content); + + let job = wait_for_job_state(&rig, job_id, JobState::Completed).await; + assert_eq!(job.title, "recover after malformed tool completions"); + + assert_eq!( + rig.llm_call_count(), + 5, + "worker should complete within the bounded recovery flow" + ); + + let requests = rig.captured_llm_requests(); + assert!( + requests_contain(&requests, "call it now with valid arguments"), + "expected targeted tool-mode recovery nudge in worker requests" + ); + assert!( + requests_contain(&requests, "Do not call any more tools in the next reply."), + "expected forced text-only recovery prompt in worker requests" + ); + + rig.clear().await; + rig.send_message(&format!("/status {}", job_id)).await; + let status_responses = rig.wait_for_responses(1, Duration::from_secs(5)).await; + assert!( + status_responses[0].content.contains("Status: Completed"), + "unexpected status response: {:?}", + status_responses[0].content + ); + + rig.shutdown(); + } + // ----------------------------------------------------------------------- // Test 9: job_list_cancel // -----------------------------------------------------------------------