# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-the-compaction-cliff-token-budget-compaction-that-silently-drops-the-load-bearing-constraint.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-the-compaction-cliff-token-budget-compaction-that-silently-drops-the-load-bearing-constraint.warrant.json
# Markdown SHA-256:   bbf12d8b63cd9bb1397b77807412304094becf694a47807649f3ebeb32df6d49
# Code AST SHA-256:   8e5d469850844c98e4986d5ebdb8af42c2c416a2941bfd74a598c00b3798ddf2
# Verified Classes:   CompactionInvariantError, Span
# Defensive Guards:   3 fail-closed invariants (zero external deps)

class CompactionInvariantError(RuntimeError):
    """Raised when a compaction would drop a pinned (constraint-bearing) span."""
    pass


class Span:
    def __init__(self, span_id: str, text: str, tokens: int, relevance: float, is_constraint: bool):
        self.span_id = span_id
        self.text = text
        self.tokens = tokens
        self.relevance = relevance
        self.is_constraint = is_constraint


def plan_compaction(spans, budget_ceiling, relevance_floor):
    """
    Decide which spans to evict. Returns (kept, evicted).
    Fails closed: raises if any pinned constraint would be evicted.
    """
    kept, evicted = [], []
    for span in spans:
        # Constraints are pinned regardless of relevance score.
        if span.is_constraint:
            kept.append(span)
            continue
        if span.relevance < relevance_floor:
            evicted.append(span)
        else:
            kept.append(span)

    # Verify the post-compaction context still holds every constraint.
    pinned_ids = {s.span_id for s in spans if s.is_constraint}
    retained_ids = {s.span_id for s in kept if s.is_constraint}
    missing = pinned_ids - retained_ids
    if missing:
        raise CompactionInvariantError(
            f"compaction would drop pinned constraints: {sorted(missing)}"
        )

    # Verify the budget is actually satisfied; if not, escalate rather than
    # silently evicting a constraint to make room.
    total = sum(s.tokens for s in kept)
    if total > budget_ceiling:
        raise CompactionInvariantError(
            f"pinned constraints alone ({total} tok) exceed ceiling "
            f"({budget_ceiling} tok); escalate for caller approval"
        )

    return kept, evicted


def maybe_compact(spans, budget_ceiling, relevance_floor, current_tokens):
    # Budget crossing is a reason to consider compaction, not a license to evict.
    if current_tokens < 0.85 * budget_ceiling:
        return spans, []
    try:
        kept, evicted = plan_compaction(spans, budget_ceiling, relevance_floor)
    except CompactionInvariantError as exc:
        # Fail loud: do not proceed with a lossy summary.
        raise CompactionInvariantError(
            f"automatic compaction aborted: {exc}"
        ) from exc
    return kept, evicted