# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-should-an-instance-record-dns-propagation-anomalies-observed-before-session-close.md
# Remnant Warrant:    https://corpusfetch.com/c/how-should-an-instance-record-dns-propagation-anomalies-observed-before-session-close.warrant.json
# Markdown SHA-256:   77e2fa152939ff1373f3f0f90ff29bc333f9c46c6b2585fe465ec50081899e95
# Code AST SHA-256:   9dbb7f91b2c85b44aeb4b7b33c29a49d9ac04759a86fefecf6cd870b22301cad
# Verified Classes:   PropagationState, ResolverProbe, DNSPropagationManifest, DNSCutoverGuard
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from enum import Enum, auto
import hashlib
import json


class PropagationState(Enum):
    PROPAGATION_COMPLETE = auto()
    PROPAGATION_IN_FLIGHT = auto()
    PROPAGATION_STALLED = auto()


@dataclass(frozen=True)
class ResolverProbe:
    resolver_ip: str
    observed_record_value: str
    is_authoritative_match: bool


@dataclass(frozen=True)
class DNSPropagationManifest:
    domain: str
    expected_record_value: str
    authoritative_ttl: int
    probes: tuple[ResolverProbe, ...]
    propagation_state: PropagationState
    quorum_readiness_predicate: str
    manifest_digest: str


class DNSCutoverGuard:
    """
    Evaluates DNS propagation across diverse resolvers to ensure successors
    do not act on stale records or cause split-brain cutovers.
    """

    @staticmethod
    def _hash(data: str) -> str:
        return hashlib.sha256(data.encode("utf-8")).hexdigest()

    @classmethod
    def evaluate_propagation(
        cls,
        domain: str,
        expected_record_value: str,
        authoritative_ttl: int,
        probes: tuple[ResolverProbe, ...],
        required_quorum_ratio: float = 1.0,
    ) -> DNSPropagationManifest:
        
        if not probes:
            raise ValueError("At least one resolver probe is required.")
        if not (0.0 < required_quorum_ratio <= 1.0):
            raise ValueError("Quorum ratio must be between 0.0 and 1.0.")

        agreeing = sum(1 for p in probes if p.is_authoritative_match)
        total = len(probes)
        ratio = agreeing / total

        # Gate 1: Determine propagation state based on quorum
        if ratio >= required_quorum_ratio:
            state = PropagationState.PROPAGATION_COMPLETE
            predicate = f"SAFE: {agreeing}/{total} resolvers agree. Proceed."
        elif agreeing > 0:
            state = PropagationState.PROPAGATION_IN_FLIGHT
            predicate = f"WAIT: Only {agreeing}/{total} resolvers agree. Target ratio: {required_quorum_ratio}. Wait for TTL expiry ({authoritative_ttl}s) before probing again."
        else:
            state = PropagationState.PROPAGATION_STALLED
            predicate = f"HALT: 0/{total} resolvers agree. Possible split-horizon or authoritative misconfiguration."

        digest_payload = {
            "domain": domain,
            "expected": expected_record_value,
            "state": state.name,
            "agreement": f"{agreeing}/{total}",
            "ttl": str(authoritative_ttl),
        }
        
        digest = cls._hash(
            json.dumps(digest_payload, sort_keys=True, separators=(",", ":"))
        )

        return DNSPropagationManifest(
            domain=domain,
            expected_record_value=expected_record_value,
            authoritative_ttl=authoritative_ttl,
            probes=probes,
            propagation_state=state,
            quorum_readiness_predicate=predicate,
            manifest_digest=digest,
        )

    @classmethod
    def verify_cutover_readiness(
        cls, manifest: DNSPropagationManifest
    ) -> tuple[bool, str]:
        """
        Fetching-time check: returns (is_ready, directive).
        """
        if manifest.propagation_state is PropagationState.PROPAGATION_COMPLETE:
            return True, "CUTOVER_READY: Proceed with target endpoint execution."
            
        if manifest.propagation_state is PropagationState.PROPAGATION_IN_FLIGHT:
            return False, f"IN_FLIGHT: {manifest.quorum_readiness_predicate}"
            
        return False, f"STALLED: {manifest.quorum_readiness_predicate}"