# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/through-what-backpressure-and-circuit-breaking-gates-must-an-instance-throttle-recursive-sub-agent-spawning-and-unbounded-tool-loops.md
# Remnant Warrant:    https://corpusfetch.com/c/through-what-backpressure-and-circuit-breaking-gates-must-an-instance-throttle-recursive-sub-agent-spawning-and-unbounded-tool-loops.warrant.json
# Markdown SHA-256:   5dc82ee5e7c6ccf686d4313df745e5187a938ff0585db22e32add7f6cb053550
# Code AST SHA-256:   f4d03b38c3fd87b3f8633ac6d6bcbd5791d69fc2cef3f2242f42d578d05eb369
# Verified Classes:   RunawayLoopOrForkBombError, TreeOrchestrationStatus, RecursiveLoopCircuitBreaker
# Defensive Guards:   9 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from typing import Final


class RunawayLoopOrForkBombError(RuntimeError):
    """Raised when sub-agent spawning or tool looping violates depth, budget, cycle, or livelock bounds."""
    pass


@dataclass(frozen=True)
class TreeOrchestrationStatus:
    active_depth: int
    total_spawns_used: int
    total_tool_steps_used: int
    consecutive_stalled_steps: int
    circuit_open: bool


class RecursiveLoopCircuitBreaker:
    """
    Throttles recursive sub-agent spawning and unbounded tool loops:
    1. Enforces hard `max_recursion_depth` and blocks `A -> B -> A` ancestor lineage cycles.
    2. Enforces a shared tree-wide `max_total_tree_spawns` and `max_total_tool_steps` budget.
    3. Trips a livelock circuit breaker when `consecutive_stalled_steps >= MAX_STALLED_STEPS`
       or when a previously visited state fingerprint is repeated.
    """

    MAX_STALLED_STEPS: Final[int] = 2

    def __init__(
        self,
        *,
        max_recursion_depth: int = 3,
        max_total_tree_spawns: int = 12,
        max_total_tool_steps: int = 25,
    ) -> None:
        if max_recursion_depth < 1 or max_total_tree_spawns < 1 or max_total_tool_steps < 1:
            raise ValueError("Orchestration ceilings must be positive integers.")
        self.max_recursion_depth = max_recursion_depth
        self.max_total_tree_spawns = max_total_tree_spawns
        self.max_total_tool_steps = max_total_tool_steps

        self._total_spawns_used = 0
        self._total_steps_used = 0
        self._stalled_steps = 0
        self._visited_state_hashes: set[str] = set()
        self._circuit_open = False

    def authorize_subagent_spawn(
        self,
        *,
        parent_depth: int,
        ancestor_lineage: tuple[str, ...],
        candidate_task_signature: str,
        downstream_backpressure_active: bool = False,
    ) -> int:
        if self._circuit_open or downstream_backpressure_active:
            raise RunawayLoopOrForkBombError(
                "Sub-agent spawn refused: orchestration circuit breaker is OPEN or downstream backpressure is active."
            )

        # Gate 1A: Hard recursion depth ceiling
        child_depth = parent_depth + 1
        if child_depth > self.max_recursion_depth:
            self._circuit_open = True
            raise RunawayLoopOrForkBombError(
                f"Recursion Depth Ceiling Exceeded: child_depth={child_depth} > max={self.max_recursion_depth}."
            )

        # Gate 1B: Ancestor lineage cycle detection (blocks A -> B -> A ping-pong)
        if candidate_task_signature in ancestor_lineage:
            self._circuit_open = True
            raise RunawayLoopOrForkBombError(
                f"Delegation Cycle Detected: '{candidate_task_signature}' already in lineage {ancestor_lineage}."
            )

        # Gate 2: Shared tree-wide total spawn budget
        if self._total_spawns_used >= self.max_total_tree_spawns:
            self._circuit_open = True
            raise RunawayLoopOrForkBombError(
                f"Global Tree Spawn Budget Exhausted ({self.max_total_tree_spawns} spawns)."
            )

        self._total_spawns_used += 1
        return child_depth

    def record_loop_step_and_check_livelock(
        self,
        *,
        state_fingerprint: str,
        made_measurable_progress: bool,
    ) -> TreeOrchestrationStatus:
        if self._circuit_open:
            raise RunawayLoopOrForkBombError("Cannot execute tool step: circuit breaker is already OPEN.")

        self._total_steps_used += 1
        if self._total_steps_used > self.max_total_tool_steps:
            self._circuit_open = True
            raise RunawayLoopOrForkBombError(
                f"Total Tool-Step Budget Exhausted ({self.max_total_tool_steps} steps)."
            )

        # Gate 3: State-cycle and zero-progress livelock detection
        if state_fingerprint in self._visited_state_hashes and not made_measurable_progress:
            self._circuit_open = True
            raise RunawayLoopOrForkBombError(
                f"State-Cycle Livelock Detected: returned to previously visited state '{state_fingerprint}'."
            )
        self._visited_state_hashes.add(state_fingerprint)

        if made_measurable_progress:
            self._stalled_steps = 0
        else:
            self._stalled_steps += 1
            if self._stalled_steps >= self.MAX_STALLED_STEPS:
                self._circuit_open = True
                raise RunawayLoopOrForkBombError(
                    f"Zero-Progress Livelock Breaker Tripped: {self._stalled_steps} consecutive steps with no progress."
                )

        return TreeOrchestrationStatus(
            active_depth=0,
            total_spawns_used=self._total_spawns_used,
            total_tool_steps_used=self._total_steps_used,
            consecutive_stalled_steps=self._stalled_steps,
            circuit_open=self._circuit_open,
        )