# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/beneath-what-abstraction-leakage-thresholds-must-an-instance-bypass-high-level-wrappers-to-inspect-raw-protocol-and-syscall-primitives.md
# Remnant Warrant:    https://corpusfetch.com/c/beneath-what-abstraction-leakage-thresholds-must-an-instance-bypass-high-level-wrappers-to-inspect-raw-protocol-and-syscall-primitives.warrant.json
# Markdown SHA-256:   f0eb60cdcc70d7122c24789d23363fc2ddf08efb9be06b0101fdd0ffc6984212
# Code AST SHA-256:   eefb6a07e2e5560698951390be13d397de2f6c8c1090abeac40c44de66151bf6
# Verified Classes:   BlindWrapperRetryRefusalError, AbstractionLayer, RawPrimitiveTarget, WrapperTelemetryObservation, LayerDescentDirective, AbstractionLeakageArbiter
# Defensive Guards:   1 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from enum import Enum
from typing import Final


class BlindWrapperRetryRefusalError(RuntimeError):
    '''Raised when an instance attempts to retry a leaky high-level wrapper without descending to raw primitives.'''
    pass


class AbstractionLayer(Enum):
    ORM_QUERY_BUILDER = "ORM_QUERY_BUILDER"
    HTTP_API_SDK = "HTTP_API_SDK"
    FILESYSTEM_OR_IPC_WRAPPER = "FILESYSTEM_OR_IPC_WRAPPER"


class RawPrimitiveTarget(Enum):
    RAW_SQL_AND_EXPLAIN_ANALYZE = "RAW_SQL_AND_EXPLAIN_ANALYZE"
    RAW_HTTP_WIRE_STATUS_HEADERS_BODY = "RAW_HTTP_WIRE_STATUS_HEADERS_BODY"
    RAW_OS_ERRNO_AND_STAT_INODE = "RAW_OS_ERRNO_AND_STAT_INODE"


@dataclass(frozen=True)
class WrapperTelemetryObservation:
    operation_id: str
    layer: AbstractionLayer
    consecutive_wrapper_failures: int
    exposes_raw_error_code: bool
    observed_latency_ms: float
    expected_baseline_latency_ms: float
    observed_io_roundtrips: int
    expected_io_roundtrips: int


@dataclass(frozen=True)
class LayerDescentDirective:
    operation_id: str
    must_bypass_wrapper_for_diagnosis: bool
    target_primitive: RawPrimitiveTarget
    leakage_triggers_fired: tuple[str, ...]


class AbstractionLeakageArbiter:
    '''
    Determines when an instance MUST stop debugging at the high-level wrapper layer
    and descend directly to raw protocol / syscall primitives:
    1. Fires on opaque wrapper exceptions (`exposes_raw_error_code == False` on failure).
    2. Fires when consecutive wrapper attempts >= 2 fail.
    3. Fires on > 5x latency cliffs or > 2x hidden I/O amplification (e.g., ORM N+1 queries).
    '''

    MAX_BLIND_WRAPPER_ATTEMPTS: Final[int] = 1
    LATENCY_AMPLIFICATION_THRESHOLD: Final[float] = 5.0
    IO_ROUNDTRIP_AMPLIFICATION_THRESHOLD: Final[int] = 2

    LAYER_TO_PRIMITIVE: Final[dict[AbstractionLayer, RawPrimitiveTarget]] = {
        AbstractionLayer.ORM_QUERY_BUILDER: RawPrimitiveTarget.RAW_SQL_AND_EXPLAIN_ANALYZE,
        AbstractionLayer.HTTP_API_SDK: RawPrimitiveTarget.RAW_HTTP_WIRE_STATUS_HEADERS_BODY,
        AbstractionLayer.FILESYSTEM_OR_IPC_WRAPPER: RawPrimitiveTarget.RAW_OS_ERRNO_AND_STAT_INODE,
    }

    @classmethod
    def evaluate_leakage(cls, obs: WrapperTelemetryObservation) -> LayerDescentDirective:
        triggers: list[str] = []

        if obs.consecutive_wrapper_failures > 0 and not obs.exposes_raw_error_code:
            triggers.append("OPAQUE_WRAPPER_ERROR_MASKING")

        if obs.consecutive_wrapper_failures > cls.MAX_BLIND_WRAPPER_ATTEMPTS:
            triggers.append(f"WRAPPER_RETRY_EXHAUSTION(failures={obs.consecutive_wrapper_failures})")

        base_lat = max(1.0, obs.expected_baseline_latency_ms)
        if obs.observed_latency_ms >= base_lat * cls.LATENCY_AMPLIFICATION_THRESHOLD:
            triggers.append(
                f"LATENCY_CLIFF({obs.observed_latency_ms:.0f}ms vs {base_lat:.0f}ms baseline)"
            )

        base_io = max(1, obs.expected_io_roundtrips)
        if obs.observed_io_roundtrips > base_io * cls.IO_ROUNDTRIP_AMPLIFICATION_THRESHOLD:
            triggers.append(
                f"HIDDEN_IO_AMPLIFICATION({obs.observed_io_roundtrips} roundtrips vs {base_io} expected)"
            )

        return LayerDescentDirective(
            operation_id=obs.operation_id,
            must_bypass_wrapper_for_diagnosis=bool(triggers),
            target_primitive=cls.LAYER_TO_PRIMITIVE[obs.layer],
            leakage_triggers_fired=tuple(triggers),
        )

    @classmethod
    def assert_wrapper_retry_permitted(cls, obs: WrapperTelemetryObservation) -> None:
        directive = cls.evaluate_leakage(obs)
        if directive.must_bypass_wrapper_for_diagnosis:
            raise BlindWrapperRetryRefusalError(
                f"Refusing blind wrapper retry on '{obs.operation_id}': abstraction leaked via "
                f"{directive.leakage_triggers_fired}. Descend immediately to {directive.target_primitive.value}."
            )