# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-refuse-model-driven-pruning-of-stale-reasoning-traces-the-read-after-prune-hazard-in-self-editing-context-policies.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-refuse-model-driven-pruning-of-stale-reasoning-traces-the-read-after-prune-hazard-in-self-editing-context-policies.warrant.json
# Markdown SHA-256:   c16357c21ebfdb2b9fd1cfb994e6f178b47943a1b655107f8337d151044e8056
# Code AST SHA-256:   d8124b69649ea81baf12a5fbc43ecaf0577d3e1583e70496fb1316f4cb7340a9
# Verified Classes:   PruneRefused, TraceKind, Trace, ContextBuffer
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass, field
from enum import Enum
from typing import Iterable


class PruneRefused(RuntimeError):
    """Raised when a model-driven pruner is asked to physically delete a trace
    that the read-after-prune invariant forbids removing."""


class TraceKind(Enum):
    POSITIVE = "positive"
    NEGATIVE = "negative"      # "do not do X", "X failed", "X is unsafe"
    PROHIBITION = "prohibition"


@dataclass(frozen=True)
class Trace:
    trace_id: str
    kind: TraceKind
    body: str
    references: tuple[str, ...] = ()


@dataclass
class ContextBuffer:
    traces: dict[str, Trace] = field(default_factory=dict)
    tombstones: list[dict] = field(default_factory=list)
    in_flight_readers: int = 0

    def add(self, t: Trace) -> None:
        self.traces[t.trace_id] = t

    def mark_stale(self, trace_id: str, reason: str) -> None:
        # Marking is always safe: bytes stay, annotation is monotone.
        self.tombstones.append(
            {"trace_id": trace_id, "reason": reason, "action": "marked"}
        )

    def prune(self, trace_id: str, reason: str, surviving: Iterable[str]) -> None:
        t = self.traces.get(trace_id)
        if t is None:
            return

        # Condition 4: negative results and prohibitions are never prunable.
        if t.kind in (TraceKind.NEGATIVE, TraceKind.PROHIBITION):
            raise PruneRefused(
                f"trace {trace_id!r} is a {t.kind.value} result; "
                f"physical deletion would remove a load-bearing prohibition"
            )

        # Condition 2: no surviving trace may reference a pruned trace.
        surviving_set = set(surviving)
        for sid in surviving_set:
            s = self.traces.get(sid)
            if s is not None and trace_id in s.references:
                raise PruneRefused(
                    f"trace {trace_id!r} is referenced by surviving trace "
                    f"{sid!r}; prune would leave a dangling justification"
                )

        # Condition 1: no live reader may be pinned to this buffer version.
        if self.in_flight_readers > 0:
            raise PruneRefused(
                f"cannot prune {trace_id!r}: {self.in_flight_readers} "
                f"in-flight reader(s) hold a pre-prune reference"
            )

        # Condition 3: emit a tombstone so later readers do not confabulate.
        del self.traces[trace_id]
        self.tombstones.append(
            {"trace_id": trace_id, "reason": reason, "action": "pruned"}
        )


def apply_model_prune_plan(
    buf: ContextBuffer,
    plan: list[tuple[str, str]],
    surviving: Iterable[str],
) -> None:
    """Apply a pruner's requested deletions, refusing any that violate the
    read-after-prune invariant. Callers must treat PruneRefused as a signal to
    fall back to marking, not as a transient error to retry."""
    surviving = list(surviving)
    for trace_id, reason in plan:
        try:
            buf.prune(trace_id, reason, surviving)
        except PruneRefused:
            buf.mark_stale(trace_id, reason)