# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/in-what-scenarios-must-an-instance-treat-auto-compaction-as-a-lossy-transform-rather-than-a-reclaim-operation.md
# Remnant Warrant:    https://corpusfetch.com/c/in-what-scenarios-must-an-instance-treat-auto-compaction-as-a-lossy-transform-rather-than-a-reclaim-operation.warrant.json
# Markdown SHA-256:   d52944e9112c5870d5c6a11cd1c8f71dde7d4db3855c8d41e79f2e32ddbc1b31
# Code AST SHA-256:   d6523f739fd590e52773f3d221afa220a63030913d2ec528de92421dd4450d6e
# Verified Classes:   CompactionInvariantError
# Defensive Guards:   2 fail-closed invariants (zero external deps)

class CompactionInvariantError(RuntimeError):
    """Raised when a compaction pass would evict non-regenerable or
    non-idempotent state that is not externalized or pinned."""
    pass


def classify_element(elem: dict) -> str:
    # Returns one of: "reclaim", "pin_effect", "externalize", "ambiguous"
    if elem.get("regenerable") and elem.get("idempotent"):
        return "reclaim"
    if elem.get("regenerable") and not elem.get("idempotent"):
        return "pin_effect"
    if not elem.get("regenerable") and elem.get("task_critical"):
        return "externalize"
    return "ambiguous"


def safe_compact(window: list[dict], budget: int, scratch_store, effect_ledger) -> list[dict]:
    """Compact `window` toward `budget` without violating C1-C3.

    Externalizes non-regenerable critical state and pins effect records
    before any summarization occurs. Raises if an element cannot be
    safely handled.
    """
    keep_verbatim, to_summarize = [], []

    for elem in window:
        cls = classify_element(elem)
        if cls in ("reclaim",):
            to_summarize.append(elem)
        elif cls == "pin_effect":
            # Persist an effect record so a re-issue can be detected/prevented.
            effect_ledger.record(
                action=elem["action"],
                target=elem["target"],
                idempotency_key=elem.get("idempotency_key"),
                status=elem.get("status", "fired"),
            )
            to_summarize.append({**elem, "content": None})  # content droppable
        elif cls == "externalize":
            if not elem.get("externalized"):
                scratch_store.write(elem["id"], elem["content"])
            keep_verbatim.append(elem)  # re-injected verbatim next window
        else:  # ambiguous
            # Refuse to summarize ambiguous state; treat as non-regenerable.
            raise CompactionInvariantError(
                f"ambiguous element {elem.get('id')!r} cannot be safely compacted; "
                f"externalize explicitly or mark regenerable+idempotent"
            )

    # Only now, with critical state secured, may we summarize the rest.
    summary = summarize(to_summarize)
    result = keep_verbatim + [summary]

    if estimate_tokens(result) > budget:
        raise CompactionInvariantError(
            f"post-compaction size {estimate_tokens(result)} exceeds budget {budget}; "
            f"pinned/externalized state cannot be shrunk further without violating C1-C3"
        )
    return result