# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-silent-relevance-score-collapse-in-a-self-editing-context-window.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-silent-relevance-score-collapse-in-a-self-editing-context-window.warrant.json
# Markdown SHA-256:   a283b4ad88fd2862d7b4ad97eb4c1fe9e03d06a0cbe344541785c9ca050a7b4b
# Code AST SHA-256:   d1c9f7e792515941e3fd66a03e4995d7a76066032f6d29ce7ea0cc266b31a641
# Verified Classes:   RelevanceCollapseError, CycleSnapshot
# Defensive Guards:   4 fail-closed invariants (zero external deps)

import statistics
from dataclasses import dataclass, field


class RelevanceCollapseError(RuntimeError):
    """Raised when the self-editing context's score distribution has collapsed."""


@dataclass
class CycleSnapshot:
    cycle: int
    scores: dict[str, float]          # entry_id -> relevance score
    pinned_ids: set[str] = field(default_factory=set)


SPREAD_FLOOR = 0.08
CANARY_MARGIN = 0.25
REQUIRED_CONSECUTIVE = 2


def detect_collapse(
    history: list[CycleSnapshot],
    baseline: dict[str, float],
) -> bool:
    if len(history) < REQUIRED_CONSECUTIVE + 1:
        raise RelevanceCollapseError(
            f"insufficient history ({len(history)} cycles); need "
            f"{REQUIRED_CONSECUTIVE + 1} to distinguish collapse from reweighting"
        )

    recent = history[-REQUIRED_CONSECUTIVE:]
    spread_collapsed = all(
        statistics.pstdev(list(s.scores.values())) < SPREAD_FLOOR
        for s in recent
    )
    if not spread_collapsed:
        return False

    last = recent[-1]
    canary_inverted = False
    for pid in last.pinned_ids:
        if pid not in last.scores or pid not in baseline:
            raise RelevanceCollapseError(
                f"pinned canary {pid!r} missing from scores or baseline; "
                "cannot audit scorer against unwritten reference"
            )
        median_live = statistics.median(
            [v for k, v in last.scores.items() if k not in last.pinned_ids]
        )
        dropped = baseline[pid] - last.scores[pid]
        if last.scores[pid] < median_live or dropped > CANARY_MARGIN:
            canary_inverted = True
            break

    if spread_collapsed and canary_inverted:
        return True
    return False


def prune(snapshot: CycleSnapshot, threshold: float) -> list[str]:
    if detect_collapse(history=[snapshot], baseline={}):
        raise RelevanceCollapseError("unreachable guard")
    evict = [eid for eid, s in snapshot.scores.items()
             if s < threshold and eid not in snapshot.pinned_ids]
    if not evict and snapshot.scores:
        raise RelevanceCollapseError(
            "no entries eligible for eviction; buffer may be frozen or "
            "scores uniformly above threshold — verify scorer health"
        )
    return evict