# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-must-an-instance-handle-tool-execution-failures-without-cascading-hallucination-loops.md
# Remnant Warrant:    https://corpusfetch.com/c/how-must-an-instance-handle-tool-execution-failures-without-cascading-hallucination-loops.warrant.json
# Markdown SHA-256:   cd1b0bcb4d1bf8a147c32d6da92c663aec173e7355274d64aa0dd58a84c82a9a
# Code AST SHA-256:   a1ae6512d75fd2a69f19650b7a7859c9e790167c1ca8674c0d728fcd71a0c641
# Verified Classes:   ToolHallucinationOrLoopError, ToolFaultCategory, ToolFailureEvent, ToolFailureDirective, ToolFailureContainmentGate
# Defensive Guards:   2 fail-closed invariants (zero external deps)

import hashlib
import json
from dataclasses import dataclass
from enum import Enum
from typing import Any, Final


class ToolHallucinationOrLoopError(RuntimeError):
    """Raised when an instance attempts to repeat an identical failed tool call or bypasses a failed dependency."""
    pass


class ToolFaultCategory(Enum):
    TRANSIENT_RETRYABLE = "TRANSIENT_RETRYABLE"
    CONTRACT_REPAIR_REQUIRED = "CONTRACT_REPAIR_REQUIRED"
    TERMINAL_NO_RETRY = "TERMINAL_NO_RETRY"


@dataclass(frozen=True)
class ToolFailureEvent:
    tool_name: str
    arguments: dict[str, Any]
    status_or_exit_code: int
    error_type: str
    raw_error_message: str


@dataclass(frozen=True)
class ToolFailureDirective:
    category: ToolFaultCategory
    allow_retry: bool
    require_argument_mutation: bool
    halt_downstream_dependents: bool


class ToolFailureContainmentGate:
    """
    Prevents cascading hallucination loops and blind retry storms on tool failures:
    1. Classifies failures into TRANSIENT_RETRYABLE, CONTRACT_REPAIR_REQUIRED, or TERMINAL_NO_RETRY.
    2. Tracks `(tool_name, canonical_args)` hashes to forbid repeating identical non-transient calls.
    3. Enforces a strict `MAX_FAILURES_PER_TOOL = 2` circuit breaker.
    """

    TRANSIENT_CODES: Final[frozenset[int]] = frozenset({408, 429, 502, 503, 504})
    TERMINAL_CODES: Final[frozenset[int]] = frozenset({401, 403, 404, 13, 2})  # HTTP + EACCES(13), ENOENT(2)
    MAX_FAILURES_PER_TOOL: Final[int] = 2

    def __init__(self) -> None:
        self._failed_call_hashes: dict[str, ToolFaultCategory] = {}
        self._tool_failure_counts: dict[str, int] = {}

    @staticmethod
    def _hash_call(tool_name: str, arguments: dict[str, Any]) -> str:
        canonical = json.dumps({"tool": tool_name, "args": arguments}, sort_keys=True, default=str)
        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:16]

    def record_failure_and_decide(self, event: ToolFailureEvent) -> ToolFailureDirective:
        call_hash = self._hash_call(event.tool_name, event.arguments)
        count = self._tool_failure_counts.get(event.tool_name, 0) + 1
        self._tool_failure_counts[event.tool_name] = count

        if event.status_or_exit_code in self.TERMINAL_CODES or event.error_type in {
            "PermissionError",
            "FileNotFoundError",
            "AuthenticationError",
        }:
            category = ToolFaultCategory.TERMINAL_NO_RETRY
        elif event.status_or_exit_code in self.TRANSIENT_CODES or event.error_type in {"TimeoutError", "RateLimitError"}:
            category = ToolFaultCategory.TRANSIENT_RETRYABLE
        else:
            category = ToolFaultCategory.CONTRACT_REPAIR_REQUIRED

        self._failed_call_hashes[call_hash] = category

        # Gate 4: Hard consecutive-failure circuit breaker
        if count >= self.MAX_FAILURES_PER_TOOL or category == ToolFaultCategory.TERMINAL_NO_RETRY:
            return ToolFailureDirective(
                category=category,
                allow_retry=False,
                require_argument_mutation=False,
                halt_downstream_dependents=True,
            )

        return ToolFailureDirective(
            category=category,
            allow_retry=True,
            require_argument_mutation=(category == ToolFaultCategory.CONTRACT_REPAIR_REQUIRED),
            halt_downstream_dependents=True,
        )

    def assert_candidate_call_permitted(self, tool_name: str, candidate_arguments: dict[str, Any]) -> None:
        if self._tool_failure_counts.get(tool_name, 0) >= self.MAX_FAILURES_PER_TOOL:
            raise ToolHallucinationOrLoopError(
                f"Circuit Breaker Open for tool '{tool_name}': reached {self.MAX_FAILURES_PER_TOOL} consecutive failures."
            )

        call_hash = self._hash_call(tool_name, candidate_arguments)
        prior_cat = self._failed_call_hashes.get(call_hash)
        if prior_cat in {ToolFaultCategory.CONTRACT_REPAIR_REQUIRED, ToolFaultCategory.TERMINAL_NO_RETRY}:
            raise ToolHallucinationOrLoopError(
                f"Identical-Argument Retry Loop Blocked on '{tool_name}': call signature '{call_hash}' "
                f"already failed with {prior_cat.value}. Mutate arguments or halt."
            )