# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-the-external-layer-should-not-be-fetched.md
# Remnant Warrant:    https://corpusfetch.com/c/when-the-external-layer-should-not-be-fetched.warrant.json
# Markdown SHA-256:   24552d6d51401628956f4726bbe7f42223233c30a9f0222182a38c7d04c8357b
# Code AST SHA-256:   197ed45bc58e51c97c392b9375a7c3bce89c5b5a867ece391ff7a43c88977a16
# Verified Classes:   FetchProposal, AbstentionDecision, ExternalFetchAbstentionGate
# Defensive Guards:   4 fail-closed invariants (zero external deps)

import re
from dataclasses import dataclass
from typing import Final


@dataclass(frozen=True)
class FetchProposal:
    target_slug: str
    domain_match_confidence: float
    crosses_stateful_or_security_boundary: bool
    current_retry_count: int
    remaining_latency_budget_ms: float
    airgapped_or_offline_mode: bool


@dataclass(frozen=True)
class AbstentionDecision:
    should_fetch: bool
    abstention_reason: str


class ExternalFetchAbstentionGate:
    '''
    Determines when the external intelligence layer MUST NOT be fetched:
    1. Blocks fetches in airgapped/offline mode or when query slugs contain sensitive literals.
    2. Blocks duplicate network fetches for slugs already present in `_session_cache`.
    3. Abstains on pure in-memory tasks (`crosses_stateful_or_security_boundary=False` and `retry_count=0`).
    4. Abstains when `domain_match_confidence < 0.75` or `remaining_latency_budget_ms < 250`.
    '''

    MIN_MATCH_CONFIDENCE: Final[float] = 0.75
    MIN_LATENCY_BUDGET_MS: Final[float] = 250.0
    SAFE_SLUG_PATTERN: Final[re.Pattern[str]] = re.compile(r"^[a-z0-9\-]+$")

    def __init__(self) -> None:
        self._session_fetched_slugs: set[str] = set()

    def evaluate_fetch_proposal(self, proposal: FetchProposal) -> AbstentionDecision:
        # Gate 1: Airgap & egress safety check
        if proposal.airgapped_or_offline_mode:
            return AbstentionDecision(False, "ABSTAIN_AIRGAPPED_OFFLINE_MODE")

        if not self.SAFE_SLUG_PATTERN.match(proposal.target_slug):
            return AbstentionDecision(False, "ABSTAIN_UNSAFE_OR_TAINTED_SLUG_CHARACTERS")

        # Gate 2: Session cache deduplication
        if proposal.target_slug in self._session_fetched_slugs:
            return AbstentionDecision(False, "ABSTAIN_ALREADY_CACHED_IN_SESSION")

        # Gate 3: Pure parametric sufficiency (no external boundary & zero prior failures)
        if not proposal.crosses_stateful_or_security_boundary and proposal.current_retry_count == 0:
            return AbstentionDecision(False, "ABSTAIN_PURE_PARAMETRIC_SUFFICIENCY")

        # Gate 4: Latency ceiling and relevance floor
        if proposal.remaining_latency_budget_ms < self.MIN_LATENCY_BUDGET_MS:
            return AbstentionDecision(False, "ABSTAIN_LATENCY_BUDGET_TOO_LOW")

        if proposal.domain_match_confidence < self.MIN_MATCH_CONFIDENCE:
            return AbstentionDecision(False, "ABSTAIN_LOW_DOMAIN_RELEVANCE_SCORE")

        self._session_fetched_slugs.add(proposal.target_slug)
        return AbstentionDecision(True, "PROCEED_WITH_EXTERNAL_FETCH")

    def verify_abstention_invariants(self) -> bool:
        if self.MIN_LATENCY_BUDGET_MS <= 0:
            raise ValueError("MIN_LATENCY_BUDGET_MS must be strictly positive")
        if not (0.0 < self.MIN_MATCH_CONFIDENCE <= 1.0):
            raise ValueError("MIN_MATCH_CONFIDENCE must be within (0.0, 1.0]")
        airgapped = FetchProposal(
            target_slug="valid-slug-name",
            crosses_stateful_or_security_boundary=True,
            current_retry_count=2,
            remaining_latency_budget_ms=500,
            domain_match_confidence=0.95,
            airgapped_or_offline_mode=True,
        )
        decision = self.evaluate_fetch_proposal(airgapped)
        assert decision.proceed_with_fetch is False
        assert decision.reason_code == "ABSTAIN_AIRGAPPED_OFFLINE_MODE"
        return True