# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-silent-context-loss-during-auto-compaction-of-long-horizon-agent-sessions.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-silent-context-loss-during-auto-compaction-of-long-horizon-agent-sessions.warrant.json
# Markdown SHA-256:   ef2c9400d332f046f8b3aad3ea518b5209520d42970e2321f26e6fbc4e94c6f6
# Code AST SHA-256:   3dbe7ae47b9f6db575b2054f2e087a3c2c5eb5fa814a6ff08c1c7ee56e5b2140
# Verified Classes:   CompactionInvariantError
# Defensive Guards:   6 fail-closed invariants (zero external deps)

class CompactionInvariantError(RuntimeError):
    """Raised when an auto-compaction would violate fidelity invariants."""
    pass


def compact_session(source_events, prior_summary, budget_tokens, summary_schema):
    """
    source_events: immutable event log for the span being compacted.
    prior_summary: the existing summary, if any (must NOT be used as input).
    """
    if not source_events:
        raise CompactionInvariantError(
            "compaction requested with empty source span; "
            "cannot verify coverage closure"
        )

    # Invariant 1: source anchoring. Never summarize a summary.
    if prior_summary is not None and prior_summary.get("derived_from_summary"):
        raise CompactionInvariantError(
            "compaction input is a prior summary (summary-of-summary); "
            "re-fetch originals from the immutable log or mark session degraded"
        )

    # Invariant 2: coverage closure. Union of covered ranges must equal span.
    covered = set()
    for event in source_events:
        covered.update(range(event["start"], event["end"]))
    expected = set(range(source_events[0]["start"], source_events[-1]["end"]))
    if covered != expected:
        missing = sorted(expected - covered)
        raise CompactionInvariantError(
            f"coverage gap in source span at offsets {missing[:8]}; "
            f"refusing to emit a lossy summary"
        )

    # Invariant 3: schema completeness for decisions (polarity + reason).
    for event in source_events:
        if event.get("kind") == "decision":
            if event.get("polarity") not in {"accepted", "rejected", "deferred"}:
                raise CompactionInvariantError(
                    f"decision event {event['id']} missing polarity; "
                    f"free-text summaries lose negation under compression"
                )
            if not event.get("reason_token"):
                raise CompactionInvariantError(
                    f"decision event {event['id']} missing reason_token"
                )

    # Invariant 4: budget check is necessary but not sufficient.
    summary = summarize_anchored(source_events, summary_schema)
    if summary["tokens"] > budget_tokens:
        raise CompactionInvariantError(
            f"summary {summary['tokens']} exceeds budget {budget_tokens}; "
            f"do not shrink by dropping entities -- escalate instead"
        )

    summary["derived_from_summary"] = False
    summary["covered_ranges"] = [(e["start"], e["end"]) for e in source_events]
    return summary