@@ -0,0 +1,223 @@ |
| 1 | +"""Error recovery system for the agent. |
| 2 | + |
| 3 | +Provides intelligent retry logic with adaptation when tools fail. |
| 4 | +""" |
| 5 | + |
| 6 | +from dataclasses import dataclass, field |
| 7 | +from enum import Enum, auto |
| 8 | +from typing import Any |
| 9 | + |
| 10 | + |
| 11 | +class ErrorCategory(Enum): |
| 12 | + """Categories of errors for recovery strategies.""" |
| 13 | + FILE_NOT_FOUND = auto() |
| 14 | + PERMISSION_DENIED = auto() |
| 15 | + SYNTAX_ERROR = auto() |
| 16 | + COMMAND_NOT_FOUND = auto() |
| 17 | + TIMEOUT = auto() |
| 18 | + INVALID_ARGUMENTS = auto() |
| 19 | + NETWORK_ERROR = auto() |
| 20 | + UNKNOWN = auto() |
| 21 | + |
| 22 | + |
| 23 | +@dataclass |
| 24 | +class ToolAttempt: |
| 25 | + """Record of a single tool execution attempt.""" |
| 26 | + tool_name: str |
| 27 | + arguments: dict[str, Any] |
| 28 | + error: str |
| 29 | + category: ErrorCategory |
| 30 | + |
| 31 | + |
| 32 | +@dataclass |
| 33 | +class RecoveryContext: |
| 34 | + """Tracks recovery state for a tool execution.""" |
| 35 | + original_tool: str |
| 36 | + original_args: dict[str, Any] |
| 37 | + attempts: list[ToolAttempt] = field(default_factory=list) |
| 38 | + max_retries: int = 3 |
| 39 | + |
| 40 | + def add_attempt(self, tool_name: str, args: dict[str, Any], error: str) -> None: |
| 41 | + """Record an attempted tool execution.""" |
| 42 | + category = categorize_error(error) |
| 43 | + self.attempts.append(ToolAttempt( |
| 44 | + tool_name=tool_name, |
| 45 | + arguments=args, |
| 46 | + error=error, |
| 47 | + category=category, |
| 48 | + )) |
| 49 | + |
| 50 | + def can_retry(self) -> bool: |
| 51 | + """Check if more retries are allowed.""" |
| 52 | + return len(self.attempts) < self.max_retries |
| 53 | + |
| 54 | + def attempts_summary(self) -> str: |
| 55 | + """Summarize what's been tried for the LLM.""" |
| 56 | + if not self.attempts: |
| 57 | + return "" |
| 58 | + |
| 59 | + lines = ["Previous attempts:"] |
| 60 | + for i, attempt in enumerate(self.attempts, 1): |
| 61 | + args_str = ", ".join(f"{k}={v!r}" for k, v in attempt.arguments.items()) |
| 62 | + lines.append(f"{i}. {attempt.tool_name}({args_str})") |
| 63 | + lines.append(f" Error: {attempt.error}") |
| 64 | + return "\n".join(lines) |
| 65 | + |
| 66 | + def was_tried(self, tool_name: str, args: dict[str, Any]) -> bool: |
| 67 | + """Check if this exact tool+args combination was already tried.""" |
| 68 | + for attempt in self.attempts: |
| 69 | + if attempt.tool_name == tool_name and attempt.arguments == args: |
| 70 | + return True |
| 71 | + return False |
| 72 | + |
| 73 | + |
| 74 | +def categorize_error(error_message: str) -> ErrorCategory: |
| 75 | + """Categorize an error message for recovery strategy selection.""" |
| 76 | + error_lower = error_message.lower() |
| 77 | + |
| 78 | + if any(x in error_lower for x in ["no such file", "file not found", "does not exist"]): |
| 79 | + return ErrorCategory.FILE_NOT_FOUND |
| 80 | + |
| 81 | + if any(x in error_lower for x in ["permission denied", "access denied", "not permitted"]): |
| 82 | + return ErrorCategory.PERMISSION_DENIED |
| 83 | + |
| 84 | + if any(x in error_lower for x in ["syntax error", "invalid syntax", "parse error"]): |
| 85 | + return ErrorCategory.SYNTAX_ERROR |
| 86 | + |
| 87 | + if any(x in error_lower for x in ["command not found", "not recognized", "no such command"]): |
| 88 | + return ErrorCategory.COMMAND_NOT_FOUND |
| 89 | + |
| 90 | + if any(x in error_lower for x in ["timeout", "timed out"]): |
| 91 | + return ErrorCategory.TIMEOUT |
| 92 | + |
| 93 | + if any(x in error_lower for x in ["invalid argument", "missing required", "expected"]): |
| 94 | + return ErrorCategory.INVALID_ARGUMENTS |
| 95 | + |
| 96 | + if any(x in error_lower for x in ["network", "connection", "unreachable"]): |
| 97 | + return ErrorCategory.NETWORK_ERROR |
| 98 | + |
| 99 | + return ErrorCategory.UNKNOWN |
| 100 | + |
| 101 | + |
| 102 | +def get_recovery_hints(category: ErrorCategory, tool_name: str) -> str: |
| 103 | + """Get hints for recovering from a specific error category.""" |
| 104 | + hints = { |
| 105 | + ErrorCategory.FILE_NOT_FOUND: [ |
| 106 | + "Check if the file path is correct (use glob to search for it)", |
| 107 | + "The file might be in a different directory", |
| 108 | + "Check for typos in the filename", |
| 109 | + "List the directory contents first", |
| 110 | + ], |
| 111 | + ErrorCategory.PERMISSION_DENIED: [ |
| 112 | + "The file might be read-only or owned by another user", |
| 113 | + "Try reading the file instead of modifying it", |
| 114 | + "Check if the directory exists and is writable", |
| 115 | + ], |
| 116 | + ErrorCategory.SYNTAX_ERROR: [ |
| 117 | + "Review the edit content for syntax errors", |
| 118 | + "Check that the old_string matches exactly (including whitespace)", |
| 119 | + "Read the file first to verify current contents", |
| 120 | + ], |
| 121 | + ErrorCategory.COMMAND_NOT_FOUND: [ |
| 122 | + "Check if the command is installed", |
| 123 | + "Try using the full path to the command", |
| 124 | + "Use a different command that achieves the same goal", |
| 125 | + ], |
| 126 | + ErrorCategory.TIMEOUT: [ |
| 127 | + "The operation is taking too long", |
| 128 | + "Try a simpler or more targeted operation", |
| 129 | + "Break the task into smaller steps", |
| 130 | + ], |
| 131 | + ErrorCategory.INVALID_ARGUMENTS: [ |
| 132 | + "Review the tool parameters", |
| 133 | + "Check that required arguments are provided", |
| 134 | + "Verify argument types and formats", |
| 135 | + ], |
| 136 | + ErrorCategory.NETWORK_ERROR: [ |
| 137 | + "Network operations may be unavailable", |
| 138 | + "Try an offline alternative if possible", |
| 139 | + ], |
| 140 | + ErrorCategory.UNKNOWN: [ |
| 141 | + "Try a different approach", |
| 142 | + "Read related files for more context", |
| 143 | + "Break the task into smaller steps", |
| 144 | + ], |
| 145 | + } |
| 146 | + |
| 147 | + category_hints = hints.get(category, hints[ErrorCategory.UNKNOWN]) |
| 148 | + |
| 149 | + # Add tool-specific hints |
| 150 | + if tool_name == "edit" and category == ErrorCategory.FILE_NOT_FOUND: |
| 151 | + category_hints.insert(0, "Use 'write' instead of 'edit' to create a new file") |
| 152 | + |
| 153 | + if tool_name == "bash" and category == ErrorCategory.COMMAND_NOT_FOUND: |
| 154 | + category_hints.insert(0, "Check available commands with 'which' or 'type'") |
| 155 | + |
| 156 | + return "\n".join(f"- {hint}" for hint in category_hints) |
| 157 | + |
| 158 | + |
| 159 | +RECOVERY_PROMPT = """The tool execution failed. Analyze the error and try an alternative approach. |
| 160 | + |
| 161 | +Tool: {tool_name} |
| 162 | +Arguments: {args} |
| 163 | +Error: {error} |
| 164 | +Category: {category} |
| 165 | + |
| 166 | +{attempts_summary} |
| 167 | + |
| 168 | +Recovery hints: |
| 169 | +{hints} |
| 170 | + |
| 171 | +IMPORTANT: |
| 172 | +- Do NOT repeat the exact same tool call that just failed |
| 173 | +- Try a different approach or gather more information first |
| 174 | +- If you've tried {attempt_count}/{max_retries} times, consider explaining the issue to the user |
| 175 | + |
| 176 | +What would you like to try instead?""" |
| 177 | + |
| 178 | + |
| 179 | +def format_recovery_prompt( |
| 180 | + context: RecoveryContext, |
| 181 | + tool_name: str, |
| 182 | + args: dict[str, Any], |
| 183 | + error: str, |
| 184 | +) -> str: |
| 185 | + """Format a prompt asking the LLM to recover from an error.""" |
| 186 | + category = categorize_error(error) |
| 187 | + hints = get_recovery_hints(category, tool_name) |
| 188 | + args_str = ", ".join(f"{k}={v!r}" for k, v in args.items()) |
| 189 | + |
| 190 | + return RECOVERY_PROMPT.format( |
| 191 | + tool_name=tool_name, |
| 192 | + args=args_str, |
| 193 | + error=error, |
| 194 | + category=category.name.replace("_", " ").title(), |
| 195 | + attempts_summary=context.attempts_summary(), |
| 196 | + hints=hints, |
| 197 | + attempt_count=len(context.attempts), |
| 198 | + max_retries=context.max_retries, |
| 199 | + ) |
| 200 | + |
| 201 | + |
| 202 | +def format_failure_message(context: RecoveryContext) -> str: |
| 203 | + """Format a message when all retries are exhausted.""" |
| 204 | + lines = [ |
| 205 | + f"Failed to complete the operation after {len(context.attempts)} attempts.", |
| 206 | + "", |
| 207 | + "What was tried:", |
| 208 | + ] |
| 209 | + |
| 210 | + for i, attempt in enumerate(context.attempts, 1): |
| 211 | + args_str = ", ".join(f"{k}={v!r}" for k, v in attempt.arguments.items()) |
| 212 | + lines.append(f"{i}. {attempt.tool_name}({args_str})") |
| 213 | + lines.append(f" Error: {attempt.error}") |
| 214 | + |
| 215 | + lines.extend([ |
| 216 | + "", |
| 217 | + "You may need to:", |
| 218 | + "- Manually check the file/directory structure", |
| 219 | + "- Verify permissions", |
| 220 | + "- Try a completely different approach", |
| 221 | + ]) |
| 222 | + |
| 223 | + return "\n".join(lines) |