# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-happens-when-an-instance-exhausts-its-scratchpad-budget-mid-compaction-the-compaction-retrieval-deadlock-under-latency-slos.md
# Remnant Warrant:    https://corpusfetch.com/c/what-happens-when-an-instance-exhausts-its-scratchpad-budget-mid-compaction-the-compaction-retrieval-deadlock-under-latency-slos.warrant.json
# Markdown SHA-256:   d7c3534b501ed721620bbb28d9d47cc79e2cb521d33918fb658891495dba64c6
# Code AST SHA-256:   b77a96882e6ada82e2d6ab578f38d8703e0fab4cc595a4549de1eba6e8aafcab
# Verified Classes:   CompactionStagingDeadlock, ContextBudgetScheduler
# Defensive Guards:   1 fail-closed invariants (zero external deps)

class CompactionStagingDeadlock(RuntimeError):
    """Raised when compaction cannot acquire its staging buffer because the
    scratchpad reservation is saturated. This state is unrecoverable within a
    turn: retrying re-enters the same fixed point. Do not retry; force an
    out-of-band compaction or evict scratchpad to external storage."""
    pass


class ContextBudgetScheduler:
    STAGING_FRACTION_OF_SPAN = 0.12  # compactor summary staging overhead
    HIGH_WATER_MARK = 0.70           # admit compaction regardless of SLO

    def __init__(self, window: int, scratchpad_reservation: int):
        self.window = window
        self.scratchpad_reservation = scratchpad_reservation
        self.scratchpad_used = 0
        self.staging_reserved = 0

    def admit_compaction(self, span_tokens: int, slo_headroom_ms: int,
                         projected_latency_ms: int) -> bool:
        staging = int(span_tokens * self.STAGING_FRACTION_OF_SPAN)

        # Invariant: staging must fit alongside current scratchpad.
        if self.scratchpad_used + staging > self.scratchpad_reservation:
            raise CompactionStagingDeadlock(
                f"staging={staging} + scratchpad_used={self.scratchpad_used} "
                f"> reservation={self.scratchpad_reservation}; "
                f"compaction cannot start without evicting scratchpad. "
                f"Do not retry this turn."
            )

        # High-water-mark trigger overrides SLO-headroom deferral.
        pressure = self.scratchpad_used / self.scratchpad_reservation
        if pressure >= self.HIGH_WATER_MARK:
            self.staging_reserved = staging
            return True

        # Below high-water: only admit if SLO headroom allows.
        if projected_latency_ms <= slo_headroom_ms:
            self.staging_reserved = staging
            return True

        return False  # defer, but scratchpad is not yet critical