# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-backpressure-communication-protocols-prevent-message-loss-during-consumer-lag.md
# Remnant Warrant:    https://corpusfetch.com/c/what-backpressure-communication-protocols-prevent-message-loss-during-consumer-lag.warrant.json
# Markdown SHA-256:   0631d2cd3d85a88d72ad908e6368d813941fa7d42d5a6d863816dac20bb4f502
# Code AST SHA-256:   9ee6a87ebec310ff14ea3ae5535df725dcc4bb8ee4d03abe5d692391f5940410
# Verified Classes:   BackpressureVerdict, ConsumerState, BackpressureProtocol, BackpressureManifest, StreamingBackpressureGuard
# Defensive Guards:   1 fail-closed invariants (zero external deps)

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

class BackpressureVerdict(Enum):
    CONSUMER_HEALTHY = auto()
    APPLY_BACKPRESSURE = auto()
    FORCE_DISCONNECT = auto()

@dataclass(frozen=True)
class ConsumerState:
    current_lag_offset: int
    buffer_used_bytes: int
    buffer_capacity_bytes: int

@dataclass(frozen=True)
class BackpressureProtocol:
    supports_producer_pause: bool
    acknowledgment_received: bool
    max_acceptable_lag: int

@dataclass(frozen=True)
class BackpressureManifest:
    verdict: BackpressureVerdict
    directive: str
    manifest_digest: str

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

    @classmethod
    def evaluate_backpressure(
        cls, state: ConsumerState, protocol: BackpressureProtocol
    ) -> BackpressureManifest:
        
        # Gate 1: Buffer fullness check
        if state.buffer_capacity_bytes <= 0:
            raise ValueError("Buffer capacity must be positive.")
            
        buffer_utilization = state.buffer_used_bytes / state.buffer_capacity_bytes
        
        # Gate 2: Lag check
        lag_exceeded = state.current_lag_offset > protocol.max_acceptable_lag
        
        # Gate 3: No pressure needed
        if buffer_utilization < 0.50 and not lag_exceeded:
            return BackpressureManifest(
                verdict=BackpressureVerdict.CONSUMER_HEALTHY,
                directive="PROCEED: Buffer and lag are within acceptable bounds.",
                manifest_digest=cls._hash("HEALTHY")
            )

        # Gate 4: Producer pause supported
        if protocol.supports_producer_pause:
            if protocol.acknowledgment_received:
                return BackpressureManifest(
                    verdict=BackpressureVerdict.APPLY_BACKPRESSURE,
                    directive="ACTION: Producer paused. Consume backlog.",
                    manifest_digest=cls._hash(f"PAUSED:{state.current_lag_offset}")
                )
            else:
                return BackpressureManifest(
                    verdict=BackpressureVerdict.APPLY_BACKPRESSURE,
                    directive="ACTION: Send pause signal. Wait for producer acknowledgment.",
                    manifest_digest=cls._hash("SIGNAL_PAUSE")
                )

        # Gate 5: No producer pause, force disconnect
        if buffer_utilization > 0.90:
            return BackpressureManifest(
                verdict=BackpressureVerdict.FORCE_DISCONNECT,
                directive="HALT: Buffer > 90%. Producer cannot be paused. Disconnect to force backoff.",
                manifest_digest=cls._hash(f"DISCONNECT:{buffer_utilization:.2f}")
            )

        return BackpressureManifest(
            verdict=BackpressureVerdict.APPLY_BACKPRESSURE,
            directive="ACTION: Buffer filling. Consume aggressively or drop oldest non-critical messages.",
            manifest_digest=cls._hash("CONSUME_AGGRESSIVELY")
        )