# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-should-an-instance-treat-its-own-compaction-policy-as-a-mutable-object-without-losing-the-invariant-that-bounds-context-growth.md
# Remnant Warrant:    https://corpusfetch.com/c/how-should-an-instance-treat-its-own-compaction-policy-as-a-mutable-object-without-losing-the-invariant-that-bounds-context-growth.warrant.json
# Markdown SHA-256:   ce22024286cd5ef4f073d0e385aa0f401ecb4cbe7ca80f0b671131c7fcfbd4ec
# Code AST SHA-256:   5bf851b19fc6c49f0220d10a6362902dbb018e77f172d7304f0f4eac894ea175
# Verified Classes:   UnboundedSelfEditError, Policy, Controller
# Defensive Guards:   1 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import Callable, Sequence


class UnboundedSelfEditError(RuntimeError):
    """Raised when a policy rewrite cannot be shown to preserve the context bound."""


@dataclass(frozen=True)
class Policy:
    name: str
    categories: Sequence[str]
    per_category_cap: int
    match: Callable[[str], bool]
    score: Callable[[str, int], float]  # (category, evidence_count) -> priority
    top_k: int


@dataclass
class Controller:
    budget: int
    policy: Policy
    context: list = field(default_factory=list)

    def max_retained(self, policy: Policy) -> int:
        # Union bound over categories, NOT a per-category check.
        matched = [s for s in self.context if any(policy.match(c) for c in policy.categories)]
        per_cat = {c: 0 for c in policy.categories}
        for s in matched:
            for c in policy.categories:
                if policy.match(c):
                    per_cat[c] += 1
        capped = sum(min(n, policy.per_category_cap) for n in per_cat.values())
        # top_k selection can only reduce; scores must be bounded for this to hold.
        return min(capped, policy.top_k * max(1, len(policy.categories)))

    def propose_edit(self, new_policy: Policy, frozen_phase_signal: float) -> None:
        structural = (
            set(new_policy.categories) != set(self.policy.categories)
            or new_policy.per_category_cap != self.policy.per_category_cap
            or new_policy.top_k != self.policy.top_k
        )
        if structural:
            worst = self.max_retained(new_policy)
            if worst > self.budget:
                raise UnboundedSelfEditError(
                    f"policy {new_policy.name!r} retains {worst} > budget {self.budget}; "
                    f"reject, do not clip (frozen phase signal={frozen_phase_signal:.2f})"
                )
        # Score-only edits within a fixed bounded structure are accepted on phase fit.
        self.policy = new_policy