# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-silent-divergence-between-the-compaction-frontier-and-the-retrieval-index-in-long-horizon-agent-trajectories.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-silent-divergence-between-the-compaction-frontier-and-the-retrieval-index-in-long-horizon-agent-trajectories.warrant.json
# Markdown SHA-256:   215eaeda7d7734046d963d7b6591d4ed383401ff53afed8450c52b7501058d36
# Code AST SHA-256:   c99037ed6e89c915b0480728c93b159f827c1d867a44c6f5e51befb8896673f2
# Verified Classes:   UnanchoredReconstitutionError, FactManifest, Carrier, CompactionIndex
# Defensive Guards:   4 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import Optional


class UnanchoredReconstitutionError(RuntimeError):
    """Raised when a requested fact is neither in the live summary chain
    nor reachable as raw text / manifest-bearing carrier in the index."""
    pass


@dataclass
class FactManifest:
    # facts carried verbatim by a summary or raw span
    carried: set[str] = field(default_factory=set)


@dataclass
class Carrier:
    span_id: str
    is_raw: bool
    manifest: FactManifest
    generation: int  # 0 = raw, >0 = summary-of-summary


class CompactionIndex:
    def __init__(self) -> None:
        self.carriers: dict[str, Carrier] = {}
        self.live_chain: list[str] = []  # span_ids of active summaries
        self.tombstones: set[str] = set()

    def register(self, c: Carrier) -> None:
        self.carriers[c.span_id] = c

    def evict(self, span_id: str) -> None:
        # Refuse to evict the last carrier of any fact still referenced
        # by the live chain, unless a tombstone is explicitly recorded.
        target = self.carriers[span_id]
        for fact in target.manifest.carried:
            other_carriers = [
                c for sid, c in self.carriers.items()
                if sid != span_id and fact in c.manifest.carried
            ]
            if not other_carriers and fact not in self.tombstones:
                raise UnanchoredReconstitutionError(
                    f"refusing eviction of {span_id}: fact {fact!r} "
                    f"would become unanchored (no tombstone recorded)"
                )
        del self.carriers[span_id]

    def reconstitute(self, fact: str) -> Carrier:
        # Fail closed: return only if a carrier actually holds the fact.
        for sid in self.live_chain:
            c = self.carriers.get(sid)
            if c and fact in c.manifest.carried:
                return c
        for c in self.carriers.values():
            if fact in c.manifest.carried:
                return c
        if fact in self.tombstones:
            raise UnanchoredReconstitutionError(
                f"fact {fact!r} is tombstoned: intent was to drop it, "
                f"not to guess"
            )
        raise UnanchoredReconstitutionError(
            f"fact {fact!r} is unanchored: no raw span and no manifest "
            f"carrier holds it; reconstitution must not fabricate"
        )


def compact(index: CompactionIndex, span_ids: list[str], summary_id: str,
            summary_facts: set[str]) -> None:
    # Build the summary carrier first, then verify every fact in the
    # evicted spans is re-asserted or tombstoned before eviction.
    index.register(Carrier(summary_id, False, FactManifest(summary_facts), 1))
    index.live_chain.append(summary_id)
    dropped = set()
    for sid in span_ids:
        c = index.carriers[sid]
        dropped |= c.manifest.carried - summary_facts
    for fact in dropped:
        # Any fact the summary did not carry must be explicitly tombstoned
        # or the eviction below will raise.
        if fact not in index.tombstones:
            raise UnanchoredReconstitutionError(
                f"compaction dropped fact {fact!r} without a tombstone; "
                f"either carry it in the summary or mark it intentionally lost"
            )
    for sid in span_ids:
        index.evict(sid)