AI Agent Infinite
Loop Detection.

                    failure_count >= threshold
    CLOSED ────────────────────────────────────► OPEN
      ▲                                            │
      │                                            │ timeout expires
      │                                            ▼
      │         success                        HALF-OPEN
      └────────────────────────────────────────────┘
                                      (test call succeeds)

import time
from enum import Enum
from functools import wraps

class CircuitState(Enum):
    CLOSED = "closed"       # Normal operation
    OPEN = "open"           # Tripped, blocking all calls
    HALF_OPEN = "half_open" # Testing recovery

class AgentCircuitBreaker:
    def __init__(self, failure_threshold=10, timeout=60, window=60):
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time = None
        self.failure_threshold = failure_threshold
        self.timeout = timeout      # seconds before trying HALF_OPEN
        self.window = window        # seconds to count failures in

    def call(self, func, *args, **kwargs):
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.timeout:
                self.state = CircuitState.HALF_OPEN
            else:
                raise CircuitBreakerOpen("Circuit breaker is OPEN — agent halted")

        try:
            result = func(*args, **kwargs)
            if self.state == CircuitState.HALF_OPEN:
                self._reset()  # Recovery succeeded
            return result
        except Exception as e:
            self._record_failure()
            raise

    def _record_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN

    def _reset(self):
        self.state = CircuitState.CLOSED
        self.failure_count = 0

from suprawall import protect

secured = protect(agent, budget={
    "circuit_breaker": {
        "strategy": "combined",          # uses all three detection strategies
        "max_identical_calls": 10,       # string-match threshold
        "semantic_threshold": 0.95,      # cosine similarity threshold
        "max_tool_frequency": 20,        # calls per minute per tool
        "window_seconds": 60,
        "recovery_timeout": 300,         # 5 minutes before HALF_OPEN
    }
})

import pytest
from unittest.mock import patch
from suprawall import protect
from suprawall.exceptions import CircuitBreakerTripped

@pytest.fixture
def looping_tool():
    """Tool that always returns an error, simulating a stuck retry loop"""
    call_count = {"n": 0}
    def tool(query: str) -> str:
        call_count["n"] += 1
        return {"error": "rate_limit_exceeded", "attempt": call_count["n"]}
    return tool, call_count

def test_circuit_breaker_triggers(looping_tool):
    tool_fn, call_count = looping_tool

    secured = protect(
        agent_with_tool(tool_fn),
        budget={"circuit_breaker": {"max_identical_calls": 5, "window_seconds": 30}}
    )

    with pytest.raises(CircuitBreakerTripped) as exc_info:
        secured.invoke({"input": "Run the search that will loop"})

    assert call_count["n"] <= 6, f"Circuit breaker should have triggered before call {call_count['n']}"
    assert "circuit_breaker" in str(exc_info.value).lower()