"""Failure classifier for agent step errors. Maps (step_kind, error_text) → FailureClass. Used by retry_policy to decide whether and how to retry. """ from __future__ import annotations from enum import Enum class FailureClass(str, Enum): SECURITY_DENIED = "security_denied" # blocked by security policy TEMPLATE_ERROR = "template_error" # {{ }} resolution failed FILE_ERROR = "file_error" # file not found / path error EXECUTION_ERROR = "execution_error" # subprocess failure / transient UNKNOWN_ERROR = "unknown_error" # catch-all def classify(step_kind: str, error: str) -> FailureClass: """Classify a step failure from its kind and error message.""" if not error: return FailureClass.UNKNOWN_ERROR err = error.lower() # Security policy rejections — cannot be retried if "blocked by security policy" in err or "is not in allowlist" in err: return FailureClass.SECURITY_DENIED # Template resolution failures — logic errors, cannot be retried if "template resolution failed" in err or "could not be resolved" in err: return FailureClass.TEMPLATE_ERROR # File system errors — may be transient (e.g. write race), limited retry if step_kind in ("write_file", "read_file"): if any(kw in err for kw in ("not found", "path traversal", "missing 'path'", "no such file")): return FailureClass.FILE_ERROR # Explicit execution_error tag (used by flaky test tool) if err.startswith("execution_error"): return FailureClass.EXECUTION_ERROR # Shell non-zero exit or binary not found if step_kind == "shell" and any(kw in err for kw in ("non-zero exit code", "not found on path", "timed out")): return FailureClass.EXECUTION_ERROR return FailureClass.UNKNOWN_ERROR