# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-lose-retrievable-provenance-when-auto-compaction-summarizes-before-persisting-the-source-span.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-lose-retrievable-provenance-when-auto-compaction-summarizes-before-persisting-the-source-span.warrant.json
# Markdown SHA-256:   8b6bbf43330bf9c971a7491eaef7594c35cab46d399ee1850a4af44001eb1076
# Code AST SHA-256:   0e6e92560cfd1cb2cb7fbabbec483c7a3fc7c1161b42ff7c88d742b4d43445cd
# Verified Classes:   ProvenanceInvariantError, Span, Summary, ProvenanceStore, Compactor
# Defensive Guards:   4 fail-closed invariants (zero external deps)

import time
from dataclasses import dataclass, field
from typing import Optional


class ProvenanceInvariantError(RuntimeError):
    """Raised when a committed summary lacks a durable, resolvable provenance pointer."""
    pass


@dataclass
class Span:
    span_id: str
    epoch: int
    payload: str


@dataclass
class Summary:
    summary_id: str
    text: str
    provenance_span_id: Optional[str] = None
    provenance_epoch: Optional[int] = None
    committed: bool = False


class ProvenanceStore:
    """Cold store for source spans. Keys are (span_id, epoch) to defeat key recycling."""

    def __init__(self) -> None:
        self._spans: dict[tuple[str, int], Span] = {}

    def put(self, span: Span) -> None:
        self._spans[(span.span_id, span.epoch)] = span

    def get(self, span_id: str, epoch: int) -> Optional[Span]:
        return self._spans.get((span_id, epoch))

    def evict(self, span_id: str, epoch: int) -> None:
        self._spans.pop((span_id, epoch), None)


class Compactor:
    def __init__(self, store: ProvenanceStore) -> None:
        self.store = store
        self.summaries: dict[str, Summary] = {}

    def compact(self, span: Span, summarize) -> Summary:
        # Correct ordering: persist provenance pointer (durably) BEFORE committing summary,
        # and only release the source for eviction after the commit is durable.
        summary = Summary(summary_id=f"sum-{span.span_id}", text=summarize(span.payload))
        summary.provenance_span_id = span.span_id
        summary.provenance_epoch = span.epoch

        # Step 1: ensure source is durable and resolvable at this exact epoch.
        if self.store.get(span.span_id, span.epoch) is None:
            raise ProvenanceInvariantError(
                f"source span {span.span_id}@{span.epoch} not durable at compaction time"
            )

        # Step 2: commit summary only after pointer is durable.
        summary.committed = True
        self.summaries[summary.summary_id] = summary

        # Step 3: source may now be released to eviction.
        self.store.evict(span.span_id, span.epoch)
        return summary

    def resolve_provenance(self, summary_id: str) -> Span:
        summary = self.summaries.get(summary_id)
        if summary is None or not summary.committed:
            raise ProvenanceInvariantError(f"summary {summary_id} not committed")
        if summary.provenance_span_id is None or summary.provenance_epoch is None:
            raise ProvenanceInvariantError(f"summary {summary_id} has no provenance pointer")
        span = self.store.get(summary.provenance_span_id, summary.provenance_epoch)
        if span is None:
            # Epoch-tagged lookup means absence is detectable, never silently wrong.
            raise ProvenanceInvariantError(
                f"provenance for {summary_id} unresolvable: "
                f"{summary.provenance_span_id}@{summary.provenance_epoch} evicted or never persisted"
            )
        return span