# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/through-what-circuit-breaking-and-idempotency-guards-must-an-instance-dispatch-external-mutating-actions-to-prevent-duplicate-billing-and-orphan-mutations.md
# Remnant Warrant:    https://corpusfetch.com/c/through-what-circuit-breaking-and-idempotency-guards-must-an-instance-dispatch-external-mutating-actions-to-prevent-duplicate-billing-and-orphan-mutations.warrant.json
# Markdown SHA-256:   ee2bad9da457c8a67cd7d6db4f0e81552cafea512517fb31e47e94dffee1781c
# Code AST SHA-256:   de02229a134d8c1b40b4f703ed350ff4b01a47b100d91ec50f073681c066e8b4
# Verified Classes:   CircuitOpenError, IdempotentClient
# Defensive Guards:   4 fail-closed invariants (zero external deps)

import time
import uuid
from typing import Any, Callable


class CircuitOpenError(Exception):
    '''Raised when the circuit breaker is open and fast-failing.'''
    pass


class IdempotentClient:
    '''
    Wraps external mutating API calls with mandatory idempotency keys,
    transient retry guards, and a simple threshold-based circuit breaker.
    '''

    def __init__(self, failure_threshold: int = 3, recovery_timeout_sec: float = 30.0) -> None:
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout_sec
        self.failure_count = 0
        self.last_failure_time = 0.0
        self.state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN

    def _check_circuit(self) -> None:
        now = time.time()
        if self.state == "OPEN":
            if now - self.last_failure_time > self.recovery_timeout:
                self.state = "HALF_OPEN"
            else:
                remaining = self.recovery_timeout - (now - self.last_failure_time)
                raise CircuitOpenError(
                    f"Circuit breaker is OPEN. Fast-failing request (recovery in {remaining:.1f}s)."
                )

    def _record_success(self) -> None:
        self.failure_count = 0
        self.state = "CLOSED"

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

    def execute_mutating_call(
        self,
        action_name: str,
        business_key: str,
        request_fn: Callable[[str], dict[str, Any]],
        max_retries: int = 2,
    ) -> dict[str, Any]:
        '''
        Executes a mutating operation with a deterministic Idempotency-Key
        derived from business_key, guarded by circuit breaker and retry limits.
        '''
        self._check_circuit()

        # Gate 1: Derive deterministic idempotency UUID from invariant business key
        seed = f"corpusfetch:{action_name}:{business_key}"
        idempotency_key = str(uuid.uuid5(uuid.NAMESPACE_URL, seed))

        last_error = None
        for attempt in range(max_retries + 1):
            try:
                # Gate 2: Dispatch request carrying the idempotency key header/payload
                response = request_fn(idempotency_key)
                self._record_success()
                return response
            except (TimeoutError, ConnectionError) as exc:
                last_error = exc
                self._record_failure()
                if attempt == max_retries:
                    raise RuntimeError(
                        f"Mutating call {action_name} failed after {max_retries + 1} attempts "
                        f"due to network fault. Idempotency Key used: {idempotency_key}. "
                        "Verify remote state before retrying."
                    ) from exc
                time.sleep(2 ** attempt)  # Exponential backoff
            except ValueError as exc:
                # Gate 3: Permanent 4xx client errors must NEVER be retried
                self._record_failure()
                raise ValueError(f"Permanent client error in {action_name}: {exc}") from exc

        raise RuntimeError(f"Unexpected exit in idempotent dispatch for {action_name}") from last_error