# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-communication-invariants-prevent-an-out-of-memory-cascade-when-a-downstream-consumer-silently-stops-reading.md
# Remnant Warrant:    https://corpusfetch.com/c/what-communication-invariants-prevent-an-out-of-memory-cascade-when-a-downstream-consumer-silently-stops-reading.warrant.json
# Markdown SHA-256:   841719f23a0616891ac94af5050338260d239caba0ff9495cb71b79bb3d06739
# Code AST SHA-256:   d5588dbd1ecc114f3476e329214ba33cb60efc84dd9588a116c74b0d253e2407
# Verified Classes:   BackpressureVerdict, ProducerState, ProductionPolicy, BackpressureManifest, SilentConsumerGuard
# Defensive Guards:   0 fail-closed invariants (zero external deps)

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

class BackpressureVerdict(Enum):
    NORMAL_PRODUCTION = auto()
    APPLY_BACKPRESSURE = auto()
    HALT_PRODUCTION = auto()

@dataclass(frozen=True)
class ProducerState:
    outstanding_unacked_count: int
    buffer_backpressure_threshold: int
    last_ack_epoch: float
    current_epoch: float
    stall_timeout_sec: float

@dataclass(frozen=True)
class ProductionPolicy:
    max_acceptable_stall_sec: float
    can_drop_messages: bool

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

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

    @classmethod
    def evaluate_state(
        cls, state: ProducerState, policy: ProductionPolicy
    ) -> BackpressureManifest:
        
        # Gate 1: Outstanding count
        if state.outstanding_unacked_count >= state.buffer_backpressure_threshold:
            if policy.can_drop_messages:
                return BackpressureManifest(
                    verdict=BackpressureVerdict.APPLY_BACKPRESSURE,
                    directive=f"ACTION: Drop oldest messages. Outstanding {state.outstanding_unacked_count} >= {state.buffer_backpressure_threshold}.",
                    manifest_digest=cls._hash(f"DROP:{state.outstanding_unacked_count}")
                )
            return BackpressureManifest(
                verdict=BackpressureVerdict.APPLY_BACKPRESSURE,
                directive=f"HALT: Producer buffer full ({state.outstanding_unacked_count}). Backpressure required.",
                manifest_digest=cls._hash(f"BACKPRESSURE:{state.outstanding_unacked_count}")
            )

        # Gate 2: Stall detection
        stall_duration = state.current_epoch - state.last_ack_epoch
        if stall_duration > state.stall_timeout_sec:
            if stall_duration > policy.max_acceptable_stall_sec:
                return BackpressureManifest(
                    verdict=BackpressureVerdict.HALT_PRODUCTION,
                    directive=f"HALT: Consumer stalled for {stall_duration:.0f}s > {policy.max_acceptable_stall_sec:.0f}s. Escalate.",
                    manifest_digest=cls._hash(f"STALL_HALT:{stall_duration:.0f}")
                )
            return BackpressureManifest(
                verdict=BackpressureVerdict.APPLY_BACKPRESSURE,
                directive=f"WAIT: Stall {stall_duration:.0f}s. Apply backpressure until recovery.",
                manifest_digest=cls._hash(f"STALL:{stall_duration:.0f}")
            )

        # Gate 3: Normal
        digest = cls._hash(json.dumps({
            "outstanding": str(state.outstanding_unacked_count),
            "stall": f"{stall_duration:.0f}"
        }, sort_keys=True))
        
        return BackpressureManifest(
            verdict=BackpressureVerdict.NORMAL_PRODUCTION,
            directive="PROCEED: Producer and consumer healthy.",
            manifest_digest=digest
        )