Anton
052591281d
- Complete agent system with planning and execution - Bridge API with FastAPI server - WhatsApp-style mobile dashboard - Enhanced web search with query improvement - Security guards and policy checks - Comprehensive documentation
36 lines
1.3 KiB
Python
36 lines
1.3 KiB
Python
"""Retry policy for agent step failures.
|
|
|
|
Rules (max_attempts = total attempts allowed, including first):
|
|
security_denied → 1 (no retry — policy decision is final)
|
|
template_error → 1 (no retry — logic error, retry won't help)
|
|
file_error → 2 (1 retry — may be transient write/read race)
|
|
execution_error → 2 (1 retry — subprocess may fail transiently)
|
|
unknown_error → 1 (no retry by default — unknown cause)
|
|
|
|
Hard cap: MAX_ATTEMPTS_PER_STEP = 2 (enforced regardless of policy).
|
|
This prevents any single step from consuming more than 2 iterations.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from core.classifier import FailureClass
|
|
|
|
MAX_ATTEMPTS_PER_STEP: int = 2
|
|
|
|
_POLICY: dict[str, int] = {
|
|
FailureClass.SECURITY_DENIED.value: 1,
|
|
FailureClass.TEMPLATE_ERROR.value: 1,
|
|
FailureClass.FILE_ERROR.value: 2,
|
|
FailureClass.EXECUTION_ERROR.value: 2,
|
|
FailureClass.UNKNOWN_ERROR.value: 1,
|
|
}
|
|
|
|
|
|
def max_attempts(failure_class: FailureClass) -> int:
|
|
"""Return the maximum allowed attempts (including the first) for this failure class."""
|
|
return min(_POLICY.get(failure_class.value, 1), MAX_ATTEMPTS_PER_STEP)
|
|
|
|
|
|
def should_retry(failure_class: FailureClass, attempts_so_far: int) -> bool:
|
|
"""Return True if another attempt is allowed."""
|
|
return attempts_so_far < max_attempts(failure_class)
|