# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/in-what-scenarios-must-an-instance-treat-identical-repeated-tool-calls-as-runaway-recursion-rather-than-legitimate-idempotent-retry-and-how-observabil.md
# Remnant Warrant:    https://corpusfetch.com/c/in-what-scenarios-must-an-instance-treat-identical-repeated-tool-calls-as-runaway-recursion-rather-than-legitimate-idempotent-retry-and-how-observabil.warrant.json
# Markdown SHA-256:   14b2a02611281356353578623075dad9537e32ae3177de421582c306ef443c2e
# Code AST SHA-256:   9a260bb495c7b498f7d70c6dc59ad9b6153626520f44ceecd347baa93d0c1a97
# Verified Classes:   RunawayRecursionError, Invocation, LoopDetector
# Defensive Guards:   2 fail-closed invariants (zero external deps)

import hashlib
from dataclasses import dataclass, field
from typing import Optional


class RunawayRecursionError(RuntimeError):
    """Raised when an agent loop is classified as runaway recursion, not retry."""
    pass


@dataclass
class Invocation:
    tool: str
    semantic_args: str          # args after stripping non-semantic fields
    result_hash: str            # hash of outcome (error body or success payload)
    outcome_class: str          # "success" | "error" | "empty" | "noop"


@dataclass
class LoopDetector:
    max_retries: int = 5
    max_iterations: int = 50
    _history: list = field(default_factory=list)

    def _sig(self, inv: Invocation) -> str:
        raw = f"{inv.tool}|{inv.semantic_args}|{inv.result_hash}|{inv.outcome_class}"
        return hashlib.sha256(raw.encode()).hexdigest()

    def observe(self, inv: Invocation) -> None:
        self._history.append(inv)

        if len(self._history) > self.max_iterations:
            raise RunawayRecursionError(
                f"iteration ceiling {self.max_iterations} exceeded; "
                f"last tool={inv.tool} outcome={inv.outcome_class}"
            )

        # Count trailing identical (call, outcome) pairs — the recursion signal.
        sig = self._sig(inv)
        trailing = 0
        for prev in reversed(self._history):
            if self._sig(prev) == sig:
                trailing += 1
            else:
                break

        failing = inv.outcome_class in ("error", "empty", "noop")

        # Recursion: identical call AND identical failing outcome beyond declared budget.
        if failing and trailing > self.max_retries:
            raise RunawayRecursionError(
                f"runaway recursion: {trailing} identical failing invocations of "
                f"{inv.tool} with result_hash={inv.result_hash[:12]} "
                f"exceeds declared retry budget {self.max_retries}"
            )

        # Retry: same call but outcome transitioning (e.g. error -> success) is legitimate.
        if not failing and trailing > 1:
            # Progress observed; reset trailing budget implicitly via outcome change.
            return


def drive_agent(invocations: list) -> str:
    detector = LoopDetector(max_retries=5, max_iterations=50)
    for inv in invocations:
        detector.observe(inv)
    return "completed"