# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-should-an-instance-distinguish-checkpoint-staleness-from-summary-space-saturation-in-hierarchical-agent-compaction.md
# Remnant Warrant:    https://corpusfetch.com/c/how-should-an-instance-distinguish-checkpoint-staleness-from-summary-space-saturation-in-hierarchical-agent-compaction.warrant.json
# Markdown SHA-256:   6381915fbbe293e86758724bbcef78e007427ff5c00e6357f411be7591e4ce24
# Code AST SHA-256:   6bc0f9ee2ecb054ce2d46233da9fa4f0566b1f340d9f92d8fe0379f5097bdb84
# Verified Classes:   CompactionInvariantError, Checkpoint
# Defensive Guards:   6 fail-closed invariants (zero external deps)

from dataclasses import dataclass


class CompactionInvariantError(RuntimeError):
    """Raised when a checkpoint fails the staleness-vs-saturation decision boundary."""
    pass


@dataclass
class Checkpoint:
    node_id: str
    budget_tokens: int
    retained_propositions: frozenset
    child_joint_entropy_bits: float
    child_redundancy: float  # 0..1, fraction of child entropy that is mutual


def classify_regeneration(
    before: Checkpoint,
    after_retained: frozenset,
    churn_noise_floor: float = 0.15,
    churn_saturation_threshold: float = 0.25,
) -> str:
    """Return 'stale', 'saturated', or 'ambiguous' based on a trial regeneration."""
    if not before.retained_propositions:
        raise CompactionInvariantError(
            f"node {before.node_id}: empty baseline; cannot classify"
        )

    before_n = len(before.retained_propositions)
    after_n = len(after_retained)
    cardinality_delta = (after_n - before_n) / before_n

    # Membership churn = fraction of union that changed membership.
    union = before.retained_propositions | after_retained
    if not union:
        raise CompactionInvariantError(
            f"node {before.node_id}: both summaries empty; regeneration produced nothing"
        )
    churn = len(union - (before.retained_propositions & after_retained)) / len(union)

    # Entropy capacity check: budget must cover non-redundant child entropy.
    effective_entropy = before.child_joint_entropy_bits * (1.0 - before.child_redundancy)
    if before.budget_tokens <= 0:
        raise CompactionInvariantError(
            f"node {before.node_id}: non-positive budget {before.budget_tokens}"
        )

    if cardinality_delta > churn_noise_floor and churn <= churn_saturation_threshold:
        return "stale"
    if abs(cardinality_delta) <= churn_noise_floor and churn > churn_saturation_threshold:
        if before.budget_tokens < effective_entropy:
            return "saturated"
        raise CompactionInvariantError(
            f"node {before.node_id}: churn {churn:.2f} with adequate budget "
            f"({before.budget_tokens} >= {effective_entropy:.1f}); "
            f"neither staleness nor saturation — investigate child summaries"
        )
    return "ambiguous"


def remediate(cp: Checkpoint, verdict: str) -> str:
    if verdict == "stale":
        return "regenerate_parent"
    if verdict == "saturated":
        return "rebalance_split_or_raise_budget"
    if verdict == "ambiguous":
        raise CompactionInvariantError(
            f"node {cp.node_id}: classification ambiguous; escalate to raw transcript"
        )
    raise CompactionInvariantError(f"node {cp.node_id}: unknown verdict {verdict!r}")