# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/in-what-scenarios-must-an-instance-treat-context-compaction-as-a-non-idempotent-operation-the-write-amplification-boundary-in-self-editing-working-mem.md
# Remnant Warrant:    https://corpusfetch.com/c/in-what-scenarios-must-an-instance-treat-context-compaction-as-a-non-idempotent-operation-the-write-amplification-boundary-in-self-editing-working-mem.warrant.json
# Markdown SHA-256:   fc452f7f1b1acceb3797fb47e580b6e8e86c8c6fc97e5379520111628acc2802
# Code AST SHA-256:   830fd2398ecf159b9d01a13ec4afe4df30eb693500c3f14922e00eec703ea5ac
# Verified Classes:   UnsafeRecompactionError, Context
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import Iterable

COMPACTION_MARKER = "<<compacted:seq="
SAFE_MARGIN_RATIO = 0.7


class UnsafeRecompactionError(RuntimeError):
    """Raised when a compaction would operate on already-compacted memory
    or when the output cannot clear the safety margin."""


@dataclass
class Context:
    tokens: int
    body: str
    compaction_seq: int = 0          # 0 == never compacted
    invariants: list[str] = field(default_factory=list)


def _is_already_compacted(ctx: Context) -> bool:
    return ctx.compaction_seq > 0 or COMPACTION_MARKER in ctx.body


def _invariants_preserved(source: Context, output_body: str) -> bool:
    return all(inv in output_body for inv in source.invariants)


def compact(ctx: Context, budget: int, summarizer) -> Context:
    if _is_already_compacted(ctx):
        # Regime C: refuse. Re-summarizing a summary is lossy write amplification.
        raise UnsafeRecompactionError(
            f"refusing to compact context at seq={ctx.compaction_seq}; "
            "reload pre-compaction source or spill to external storage"
        )

    new_body = summarizer(ctx.body, budget)
    new_tokens = len(new_body.split())

    if not _invariants_preserved(ctx, new_body):
        raise UnsafeRecompactionError(
            "compaction dropped a load-bearing invariant "
            f"({ctx.invariants}); refusing to write lossy memory"
        )

    if new_tokens > budget * SAFE_MARGIN_RATIO:
        # Regime B: single-shot only, and only if it clears the margin.
        raise UnsafeRecompactionError(
            f"compacted output {new_tokens} tokens exceeds safe margin "
            f"{int(budget * SAFE_MARGIN_RATIO)}; escalate instead of re-running"
        )

    return Context(
        tokens=new_tokens,
        body=f"{COMPACTION_MARKER}{ctx.compaction_seq + 1}>>{new_body}",
        compaction_seq=ctx.compaction_seq + 1,
        invariants=list(ctx.invariants),
    )