# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/in-what-scenarios-must-an-instance-reject-lossy-context-compaction-to-preserve-causal-task-state-across-a-boundary.md
# Remnant Warrant:    https://corpusfetch.com/c/in-what-scenarios-must-an-instance-reject-lossy-context-compaction-to-preserve-causal-task-state-across-a-boundary.warrant.json
# Markdown SHA-256:   d45dd95b9760151f9bd3ad71cbb412cb52fee927566988b0e0a9626c72395f5d
# Code AST SHA-256:   0c0b7be0f9e6321570f5cb56a5f26ee06a90893523c36f5dbecaae3799d52c3a
# Verified Classes:   CompactionSafetyError, SideEffectStatus, Obligation, BoundaryContext
# Defensive Guards:   4 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from enum import Enum
from typing import Sequence


class CompactionSafetyError(RuntimeError):
    """Raised when a compaction boundary would erase causal task state."""
    pass


class SideEffectStatus(Enum):
    COMMITTED = "committed"
    NOT_STARTED = "not_started"
    UNKNOWN = "unknown"          # timeout / ambiguous: must NOT be treated as failed


@dataclass(frozen=True)
class Obligation:
    name: str
    status: SideEffectStatus
    depends_on_output_of: str | None = None
    idempotent: bool = True


@dataclass(frozen=True)
class BoundaryContext:
    obligations: Sequence[Obligation]
    active_prohibitions: Sequence[str]
    compaction_is_lossy: bool


def assert_compaction_safe(ctx: BoundaryContext) -> None:
    """
    Gate a lossy compaction boundary. Semantic compression is allowed;
    erasure of live causal state is not.
    """
    if not ctx.compaction_is_lossy:
        return  # verbatim preservation: always safe

    for ob in ctx.obligations:
        if ob.status is SideEffectStatus.UNKNOWN:
            raise CompactionSafetyError(
                f"side effect '{ob.name}' has UNKNOWN status; a lossy summary "
                f"would collapse it to 'failed' and invite a duplicate retry"
            )
        if ob.status is SideEffectStatus.NOT_STARTED and not ob.idempotent:
            raise CompactionSafetyError(
                f"pending non-idempotent obligation '{ob.name}' must be carried "
                f"verbatim across the boundary, not abstracted"
            )
        if ob.status is SideEffectStatus.NOT_STARTED and ob.depends_on_output_of:
            raise CompactionSafetyError(
                f"obligation '{ob.name}' depends on output of "
                f"'{ob.depends_on_output_of}'; ordering edge would be flattened"
            )

    if ctx.active_prohibitions:
        raise CompactionSafetyError(
            "active prohibitions present; negation may be lost in summary: "
            + "; ".join(ctx.active_prohibitions)
        )


def resume_after_compaction(ctx: BoundaryContext) -> str:
    assert_compaction_safe(ctx)
    return "resume: safe to proceed from compacted context"