use std::collections::HashMap;
use std::sync::Mutex;
use crate::types::{AgentError, AgentResult, SessionId};
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum ToolErrorAction {
Stop,
Retry,
RetryWithHistory {
errors: Vec<String>,
},
}
pub trait ToolErrorRecovery: Send + Sync {
fn on_error(
&self,
_session_id: &SessionId,
_tool_names: &[String],
_error: &AgentError,
) -> AgentResult<ToolErrorAction>;
fn on_success(&self, _session_id: &SessionId, _tool_name: &str) {}
}
pub struct StopOnError;
impl ToolErrorRecovery for StopOnError {
fn on_error(
&self,
_session_id: &SessionId,
_tool_names: &[String],
_error: &AgentError,
) -> AgentResult<ToolErrorAction> {
Ok(ToolErrorAction::Stop)
}
}
pub struct RetryOnError;
impl ToolErrorRecovery for RetryOnError {
fn on_error(
&self,
_session_id: &SessionId,
_tool_names: &[String],
_error: &AgentError,
) -> AgentResult<ToolErrorAction> {
Ok(ToolErrorAction::Retry)
}
}
pub struct ConsecutiveFailureRecovery {
max_consecutive_failures: usize,
failure_counts: Mutex<HashMap<u64, HashMap<String, usize>>>,
error_messages: Mutex<HashMap<u64, HashMap<String, Vec<String>>>>,
grace_used: Mutex<HashMap<u64, HashMap<String, bool>>>,
}
impl ConsecutiveFailureRecovery {
pub fn new(max_consecutive_failures: usize) -> Self {
Self {
max_consecutive_failures,
failure_counts: Mutex::new(HashMap::new()),
error_messages: Mutex::new(HashMap::new()),
grace_used: Mutex::new(HashMap::new()),
}
}
pub fn reset_failures(&self, session_id: &SessionId, tool_name: &str) {
if let Ok(mut counts) = self.failure_counts.lock()
&& let Some(session_counts) = counts.get_mut(&session_id.id)
{
session_counts.remove(tool_name);
}
if let Ok(mut msgs) = self.error_messages.lock()
&& let Some(session_msgs) = msgs.get_mut(&session_id.id)
{
session_msgs.remove(tool_name);
}
if let Ok(mut grace) = self.grace_used.lock()
&& let Some(session_grace) = grace.get_mut(&session_id.id)
{
session_grace.remove(tool_name);
}
}
pub fn reset_session(&self, session_id: &SessionId) {
if let Ok(mut counts) = self.failure_counts.lock() {
counts.remove(&session_id.id);
}
if let Ok(mut msgs) = self.error_messages.lock() {
msgs.remove(&session_id.id);
}
if let Ok(mut grace) = self.grace_used.lock() {
grace.remove(&session_id.id);
}
}
}
impl ToolErrorRecovery for ConsecutiveFailureRecovery {
fn on_error(
&self,
session_id: &SessionId,
tool_names: &[String],
error: &AgentError,
) -> AgentResult<ToolErrorAction> {
let mut counts = self
.failure_counts
.lock()
.map_err(|e| AgentError::internal(format!("Failed to lock failure counts: {}", e)))?;
let mut msgs = self
.error_messages
.lock()
.map_err(|e| AgentError::internal(format!("Failed to lock error messages: {}", e)))?;
let mut grace = self
.grace_used
.lock()
.map_err(|e| AgentError::internal(format!("Failed to lock grace_used: {}", e)))?;
let session_counts = counts.entry(session_id.id).or_insert_with(HashMap::new);
let session_msgs = msgs.entry(session_id.id).or_insert_with(HashMap::new);
let session_grace = grace.entry(session_id.id).or_insert_with(HashMap::new);
let error_text = error.to_string();
let mut max_failures = 0;
let mut threshold_tool: Option<String> = None;
for name in tool_names {
let count = session_counts.entry(name.clone()).or_insert(0);
*count += 1;
session_msgs
.entry(name.clone())
.or_insert_with(Vec::new)
.push(error_text.clone());
if *count > max_failures {
max_failures = *count;
}
if *count >= self.max_consecutive_failures {
threshold_tool = Some(name.clone());
}
}
if max_failures >= self.max_consecutive_failures {
let tool_name = threshold_tool.unwrap_or_else(|| tool_names[0].clone());
tracing::warn!(
session_id = session_id.id,
tool = %tool_name,
failures = max_failures,
max_consecutive_failures = self.max_consecutive_failures,
"ConsecutiveFailureRecovery: threshold reached"
);
let already_used = session_grace.get(&tool_name).copied().unwrap_or(false);
if already_used {
counts.remove(&session_id.id);
msgs.remove(&session_id.id);
grace.remove(&session_id.id);
return Ok(ToolErrorAction::Stop);
}
session_grace.insert(tool_name.clone(), true);
let errors = session_msgs.get(&tool_name).cloned().unwrap_or_default();
msgs.remove(&session_id.id);
return Ok(ToolErrorAction::RetryWithHistory { errors });
}
Ok(ToolErrorAction::Retry)
}
fn on_success(&self, session_id: &SessionId, tool_name: &str) {
self.reset_failures(session_id, tool_name);
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn stop_on_error_always_stops() {
let recovery = StopOnError;
let session_id = SessionId::new(1);
let names = vec!["tool_a".to_string()];
let error = AgentError::internal("test error");
assert_eq!(
recovery.on_error(&session_id, &names, &error).unwrap(),
ToolErrorAction::Stop
);
}
#[test]
fn retry_on_error_always_retries() {
let recovery = RetryOnError;
let session_id = SessionId::new(1);
let names = vec!["tool_a".to_string()];
let error = AgentError::internal("test error");
assert_eq!(
recovery.on_error(&session_id, &names, &error).unwrap(),
ToolErrorAction::Retry
);
}
#[test]
fn consecutive_failure_retries_then_retry_with_history() {
let recovery = ConsecutiveFailureRecovery::new(3);
let session_id = SessionId::new(1);
let names = vec!["tool_a".to_string()];
let error = AgentError::internal("test error");
assert_eq!(
recovery.on_error(&session_id, &names, &error).unwrap(),
ToolErrorAction::Retry
);
assert_eq!(
recovery.on_error(&session_id, &names, &error).unwrap(),
ToolErrorAction::Retry
);
let action = recovery.on_error(&session_id, &names, &error).unwrap();
assert!(matches!(action, ToolErrorAction::RetryWithHistory { .. }));
}
#[test]
fn consecutive_failure_stop_after_grace() {
let recovery = ConsecutiveFailureRecovery::new(3);
let session_id = SessionId::new(1);
let names = vec!["tool_a".to_string()];
let error = AgentError::internal("test error");
for _ in 0..2 {
recovery.on_error(&session_id, &names, &error).unwrap();
}
let action = recovery.on_error(&session_id, &names, &error).unwrap();
assert!(matches!(action, ToolErrorAction::RetryWithHistory { .. }));
let action = recovery.on_error(&session_id, &names, &error).unwrap();
assert!(
matches!(action, ToolErrorAction::Stop),
"should Stop after grace exhausted, got {:?}",
action
);
}
#[test]
fn consecutive_failure_retry_with_history_collects_errors() {
let recovery = ConsecutiveFailureRecovery::new(2);
let session_id = SessionId::new(1);
let names = vec!["tool_a".to_string()];
let error1 = AgentError::internal("first error");
let error2 = AgentError::internal("second error");
recovery.on_error(&session_id, &names, &error1).unwrap();
let action = recovery.on_error(&session_id, &names, &error2).unwrap();
match action {
ToolErrorAction::RetryWithHistory { errors } => {
assert_eq!(errors.len(), 2);
assert!(errors[0].contains("first error"));
assert!(errors[1].contains("second error"));
}
_ => panic!("expected RetryWithHistory, got {:?}", action),
}
}
#[test]
fn consecutive_failure_resets_on_success() {
let recovery = ConsecutiveFailureRecovery::new(3);
let session_id = SessionId::new(1);
let names = vec!["tool_a".to_string()];
let error = AgentError::internal("test error");
recovery.on_error(&session_id, &names, &error).unwrap();
recovery.on_error(&session_id, &names, &error).unwrap();
recovery.reset_failures(&session_id, "tool_a");
assert_eq!(
recovery.on_error(&session_id, &names, &error).unwrap(),
ToolErrorAction::Retry
);
}
#[test]
fn consecutive_failure_on_success_resets_via_trait() {
let recovery = ConsecutiveFailureRecovery::new(2);
let session_id = SessionId::new(1);
let names = vec!["tool_a".to_string()];
let error = AgentError::internal("test error");
assert_eq!(
recovery.on_error(&session_id, &names, &error).unwrap(),
ToolErrorAction::Retry
);
recovery.on_success(&session_id, "tool_a");
assert_eq!(
recovery.on_error(&session_id, &names, &error).unwrap(),
ToolErrorAction::Retry
);
}
#[test]
fn consecutive_failure_on_success_resets_grace() {
let recovery = ConsecutiveFailureRecovery::new(2);
let session_id = SessionId::new(1);
let names = vec!["tool_a".to_string()];
let error = AgentError::internal("test error");
recovery.on_error(&session_id, &names, &error).unwrap();
let action = recovery.on_error(&session_id, &names, &error).unwrap();
assert!(matches!(action, ToolErrorAction::RetryWithHistory { .. }));
recovery.on_success(&session_id, "tool_a");
recovery.on_error(&session_id, &names, &error).unwrap();
let action = recovery.on_error(&session_id, &names, &error).unwrap();
assert!(
matches!(action, ToolErrorAction::RetryWithHistory { .. }),
"grace should be reusable after success, got {:?}",
action
);
}
#[test]
fn consecutive_failure_resets_clears_error_messages() {
let recovery = ConsecutiveFailureRecovery::new(3);
let session_id = SessionId::new(1);
let names = vec!["tool_a".to_string()];
let error = AgentError::internal("test error");
recovery.on_error(&session_id, &names, &error).unwrap();
recovery.on_error(&session_id, &names, &error).unwrap();
recovery.reset_failures(&session_id, "tool_a");
let error_new = AgentError::internal("new error");
recovery.on_error(&session_id, &names, &error_new).unwrap();
recovery.on_error(&session_id, &names, &error_new).unwrap();
let action = recovery.on_error(&session_id, &names, &error_new).unwrap();
match action {
ToolErrorAction::RetryWithHistory { errors } => {
assert_eq!(errors.len(), 3);
for e in &errors {
assert!(e.contains("new error"), "old errors should be cleared");
}
}
_ => panic!("expected RetryWithHistory, got {:?}", action),
}
}
#[test]
fn consecutive_failure_resets_session_clears_everything() {
let recovery = ConsecutiveFailureRecovery::new(2);
let session_id = SessionId::new(1);
let names = vec!["tool_a".to_string()];
let error = AgentError::internal("test error");
recovery.on_error(&session_id, &names, &error).unwrap();
let action = recovery.on_error(&session_id, &names, &error).unwrap();
assert!(matches!(action, ToolErrorAction::RetryWithHistory { .. }));
recovery.reset_session(&session_id);
assert_eq!(
recovery.on_error(&session_id, &names, &error).unwrap(),
ToolErrorAction::Retry
);
}
#[test]
fn consecutive_failure_per_session_isolation() {
let recovery = ConsecutiveFailureRecovery::new(2);
let session1 = SessionId::new(1);
let session2 = SessionId::new(2);
let names = vec!["tool_a".to_string()];
let error = AgentError::internal("test error");
recovery.on_error(&session1, &names, &error).unwrap();
let action = recovery.on_error(&session1, &names, &error).unwrap();
assert!(matches!(action, ToolErrorAction::RetryWithHistory { .. }));
let action = recovery.on_error(&session1, &names, &error).unwrap();
assert!(
matches!(action, ToolErrorAction::Stop),
"session 1 should Stop after grace exhausted, got {:?}",
action
);
assert_eq!(
recovery.on_error(&session2, &names, &error).unwrap(),
ToolErrorAction::Retry
);
}
#[test]
fn consecutive_failure_different_tools_independent() {
let recovery = ConsecutiveFailureRecovery::new(2);
let session_id = SessionId::new(1);
let error = AgentError::internal("test error");
recovery
.on_error(&session_id, &["tool_a".to_string()], &error)
.unwrap();
assert_eq!(
recovery
.on_error(&session_id, &["tool_b".to_string()], &error,)
.unwrap(),
ToolErrorAction::Retry
);
let action = recovery
.on_error(&session_id, &["tool_a".to_string()], &error)
.unwrap();
assert!(
matches!(action, ToolErrorAction::RetryWithHistory { .. }),
"tool_a should trigger RetryWithHistory independently of tool_b"
);
}
#[test]
fn consecutive_failure_records_serde_error_details() {
let recovery = ConsecutiveFailureRecovery::new(2);
let session_id = SessionId::new(1);
let names = vec!["write_file".to_string()];
let error = AgentError::ToolArgsInvalid {
name: "write_file".to_string(),
raw: "missing field `path` at line 1 column 2 (args: {})".to_string(),
};
recovery.on_error(&session_id, &names, &error).unwrap();
let action = recovery.on_error(&session_id, &names, &error).unwrap();
match action {
ToolErrorAction::RetryWithHistory { errors } => {
assert_eq!(errors.len(), 2);
assert!(
errors[0].contains("missing field"),
"should contain serde error details"
);
}
_ => panic!("expected RetryWithHistory, got {:?}", action),
}
}
}