# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/under-what-conditions-must-an-instance-refuse-to-compact-a-context-window-whose-summarized-prefix-has-already-been-re-ingested.md
# Remnant Warrant:    https://corpusfetch.com/c/under-what-conditions-must-an-instance-refuse-to-compact-a-context-window-whose-summarized-prefix-has-already-been-re-ingested.warrant.json
# Markdown SHA-256:   e08cc8da54bfc7f7a527256214f76bbd7403d91db37f4408544c4594e444301c
# Code AST SHA-256:   5b5ec281842212d9ca6671c7f6e910b189446ae1707a58a7dffa7337f30b592e
# Verified Classes:   UnsafeRecompactionError, WindowRegion, ContextWindow
# Defensive Guards:   1 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import Optional


class UnsafeRecompactionError(RuntimeError):
    """Raised when automatic compaction would summarize an already-summarized prefix."""


@dataclass
class WindowRegion:
    text: str
    # None means the only provenance is a prior summary (no live source).
    live_source_id: Optional[str]
    summary_depth: int = 0


@dataclass
class ContextWindow:
    regions: list = field(default_factory=list)
    external_store: dict = field(default_factory=dict)

    def add_original(self, text: str, source_id: str) -> None:
        self.external_store[source_id] = text
        self.regions.append(WindowRegion(text=text, live_source_id=source_id))

    def add_summary(self, text: str, derived_from: Optional[str]) -> None:
        depth = 0
        if derived_from is not None and derived_from in self.external_store:
            # Derived from a live source: provenance intact.
            self.regions.append(
                WindowRegion(text=text, live_source_id=derived_from, summary_depth=1)
            )
            return
        # Derived from another summary: provenance is a summary of a summary.
        self.regions.append(
            WindowRegion(text=text, live_source_id=None, summary_depth=1)
        )


def compact_region(window: ContextWindow, index: int, summarize) -> WindowRegion:
    """
    Automatic compaction entry point. Refuses to compress a region whose
    only provenance is a prior summary and whose source cannot be re-derived.
    """
    region = window.regions[index]

    has_live_source = (
        region.live_source_id is not None
        and region.live_source_id in window.external_store
    )

    if region.summary_depth >= 1 and not has_live_source:
        raise UnsafeRecompactionError(
            f"Refusing to re-summarize region {index}: summary_depth="
            f"{region.summary_depth}, no resolvable live source. "
            f"Re-fetch the original or escalate; do not compress again."
        )

    new_text = summarize(region.text)
    derived = region.live_source_id if has_live_source else None
    window.add_summary(new_text, derived_from=derived)
    return window.regions[-1]


def run_compaction_cycle(window: ContextWindow, summarize) -> None:
    # Compacting the most recent region under budget pressure.
    compact_region(window, index=len(window.regions) - 1, summarize=summarize)


if __name__ == "__main__":
    window = ContextWindow()
    window.add_original("retry ceiling is 7 (from config)", source_id="cfg-42")

    # First compaction: safe, derived from live source.
    run_compaction_cycle(window, summarize=lambda t: "retries capped (cfg-42)")

    # Second compaction: the region is now a summary; simulate source loss.
    window.external_store.pop("cfg-42", None)
    try:
        run_compaction_cycle(window, summarize=lambda t: "retries capped at 5")
    except UnsafeRecompactionError as exc:
        print(f"HALT: {exc}")