# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/around-what-observability-invariants-must-an-instance-structure-diagnostic-telemetry-to-prevent-silent-failure-masking-and-log-induced-context-bloat.md
# Remnant Warrant:    https://corpusfetch.com/c/around-what-observability-invariants-must-an-instance-structure-diagnostic-telemetry-to-prevent-silent-failure-masking-and-log-induced-context-bloat.warrant.json
# Markdown SHA-256:   e0131da914dfb22cf993ea1d2160c28d8cdae42b6e5e5343c5a5c1e33c8dc7e9
# Code AST SHA-256:   05a3269d6dde8f9380cd67603b6d32868f93fec79e763a4426299c588ba13efa
# Verified Classes:   SilentFailureOrBloatError, DeduplicatedFaultRecord, BoundedDiagnosticDigest, ObservabilityHygieneGate
# Defensive Guards:   2 fail-closed invariants (zero external deps)

import ast
import hashlib
from dataclasses import dataclass
from typing import Final


class SilentFailureOrBloatError(ValueError):
    '''Raised when code silently swallows exceptions or when telemetry violates context-hygiene bounds.'''
    pass


@dataclass(frozen=True)
class DeduplicatedFaultRecord:
    signature_hash: str
    error_type: str
    top_frame_locus: str
    sample_message: str
    occurrence_count: int


@dataclass(frozen=True)
class BoundedDiagnosticDigest:
    unique_faults: tuple[DeduplicatedFaultRecord, ...]
    total_raw_events: int
    bounded_stderr_preview: str
    omitted_bytes: int


class ObservabilityHygieneGate:
    '''
    Prevents silent failure masking and log-induced context bloat:
    1. Audits Python ASTs to block silent `except: pass` or `except Exception: return None` handlers.
    2. Deduplicates repetitive runtime exceptions into compact `(signature, count)` digests.
    3. Bounds raw stderr/stdout streams using a deterministic Head + Tail window.
    '''

    HEAD_PREVIEW_CHARS: Final[int] = 1024
    TAIL_PREVIEW_CHARS: Final[int] = 2048

    @classmethod
    def assert_no_silent_except_in_ast(cls, source_code: str) -> None:
        '''Gate 1: Rejects bare excepts or except blocks that simply `pass` without logging/raising.'''
        tree = ast.parse(source_code)
        for node in ast.walk(tree):
            if isinstance(node, ast.ExceptHandler):
                if node.type is None:
                    raise SilentFailureOrBloatError(
                        f"Line {node.lineno}: Bare 'except:' is prohibited; catch specific exceptions and preserve cause."
                    )
                if len(node.body) == 1 and isinstance(node.body[0], ast.Pass):
                    raise SilentFailureOrBloatError(
                        f"Line {node.lineno}: Silent 'except ...: pass' masking is prohibited."
                    )

    @classmethod
    def build_bounded_digest(
        cls,
        *,
        raw_faults: list[tuple[str, str, str]],  # (error_type, top_frame_locus, message)
        raw_stderr: str,
    ) -> BoundedDiagnosticDigest:
        # Gate 3: Deduplicate repeated fault signatures
        counts: dict[str, int] = {}
        samples: dict[str, tuple[str, str, str]] = {}

        for err_type, locus, msg in raw_faults:
            sig_key = f"{err_type}|{locus}|{msg[:120]}"
            sig_hash = hashlib.sha256(sig_key.encode("utf-8")).hexdigest()[:12]
            counts[sig_hash] = counts.get(sig_hash, 0) + 1
            if sig_hash not in samples:
                samples[sig_hash] = (err_type, locus, msg[:300])

        deduped = tuple(
            DeduplicatedFaultRecord(
                signature_hash=sh,
                error_type=samples[sh][0],
                top_frame_locus=samples[sh][1],
                sample_message=samples[sh][2],
                occurrence_count=cnt,
            )
            for sh, cnt in counts.items()
        )

        # Gate 4: Strip carriage-return spinner noise and bound Head + Tail preview
        cleaned_lines = [line.rsplit("\r", 1)[-1] for line in raw_stderr.splitlines()]
        cleaned_stderr = "\n".join(cleaned_lines)

        max_total = cls.HEAD_PREVIEW_CHARS + cls.TAIL_PREVIEW_CHARS
        if len(cleaned_stderr) <= max_total:
            preview = cleaned_stderr
            omitted = 0
        else:
            omitted = len(cleaned_stderr) - max_total
            head = cleaned_stderr[: cls.HEAD_PREVIEW_CHARS]
            tail = cleaned_stderr[-cls.TAIL_PREVIEW_CHARS :]
            preview = f"{head}\n...[OMITTED {omitted} CHARS OF REPETITIVE TELEMETRY]...\n{tail}"

        return BoundedDiagnosticDigest(
            unique_faults=deduped,
            total_raw_events=len(raw_faults),
            bounded_stderr_preview=preview,
            omitted_bytes=omitted,
        )