# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/under-what-conditions-must-an-instance-treat-lossy-summarization-as-a-compaction-barrier-rather-than-a-compaction-opportunity-in-long-horizon-agent-tr.md
# Remnant Warrant:    https://corpusfetch.com/c/under-what-conditions-must-an-instance-treat-lossy-summarization-as-a-compaction-barrier-rather-than-a-compaction-opportunity-in-long-horizon-agent-tr.warrant.json
# Markdown SHA-256:   46a85847e2ee8165e158aedd46122f009c42dc60b6c63c49766cc86ec610bb87
# Code AST SHA-256:   c8431b68e690f9fa8c642d71d166a5e4acdc065f8c9a3dcadabd83413b6c5404
# Verified Classes:   CompactionBarrierError, Obligation, ObligationLedger, Span
# Defensive Guards:   1 fail-closed invariants (zero external deps)

"""Barrier-aware compaction gate for long-horizon agent trajectories.

A span may be lossily summarized only if it is not a compaction barrier.
Barrier status is a property of the span's join with the live obligation
ledger, not of its token size or semantic content.
"""

from dataclasses import dataclass, field
from typing import Iterable


class CompactionBarrierError(RuntimeError):
    """Raised when a span is a barrier and cannot be safely summarized."""


@dataclass(frozen=True)
class Obligation:
    obligation_id: str
    idempotency_key: str
    discharge_predicate: str          # must be observable from retained state
    discharged: bool = False


@dataclass
class ObligationLedger:
    entries: dict[str, Obligation] = field(default_factory=dict)

    def live_ids(self) -> set[str]:
        return {oid for oid, ob in self.entries.items() if not ob.discharged}

    def touches(self, span_obligation_ids: Iterable[str]) -> set[str]:
        return set(span_obligation_ids) & self.live_ids()


@dataclass
class Span:
    span_id: str
    token_count: int
    obligation_ids: tuple[str, ...]
    has_partial_side_effect: bool
    cross_span_binding_endpoint: bool
    has_structured_carrier: bool = False   # re-bound before compaction?


def is_barrier(span: Span, ledger: ObligationLedger) -> bool:
    """A span is a barrier if any discharge condition would be severed."""
    if span.has_partial_side_effect:
        return True
    if span.cross_span_binding_endpoint and not span.has_structured_carrier:
        return True
    if ledger.touches(span.obligation_ids):
        # Live obligation with no surviving carrier => barrier.
        return not span.has_structured_carrier
    return False


def compact_span(span: Span, ledger: ObligationLedger, budget: int) -> str:
    """Return a summary for an eligible span; refuse barriers under pressure.

    Budget pressure does NOT override barrier status. A barrier that cannot
    be retained or repaired must fail the compaction loudly.
    """
    if is_barrier(span, ledger):
        if span.token_count <= budget:
            # Retain verbatim: reclamation deferred, binding preserved.
            return f"[RETAINED VERBATIM span={span.span_id} tokens={span.token_count}]"
        raise CompactionBarrierError(
            f"span {span.span_id} is a compaction barrier "
            f"(tokens={span.token_count} > budget={budget}); "
            f"live obligations={sorted(ledger.touches(span.obligation_ids))}, "
            f"partial_side_effect={span.has_partial_side_effect}, "
            f"cross_span_binding={span.cross_span_binding_endpoint}; "
            f"repair (discharge or re-bind into structured carrier) before compaction"
        )
    return f"[SUMMARY span={span.span_id} tokens={span.token_count} fidelity=decisions+bindings]"