# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-starve-retrieval-under-a-satisfied-latency-slo-the-compaction-credit-ordering-invariant-in-context-window-budget-arbiters.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-starve-retrieval-under-a-satisfied-latency-slo-the-compaction-credit-ordering-invariant-in-context-window-budget-arbiters.warrant.json
# Markdown SHA-256:   13926f145f32964f84dc4b881e52c7745d83a8ea208cdd28ff97d03ee1fe0677
# Code AST SHA-256:   6732645bbd3b8e685a2db40edddc17a0cce990b2b626ebaf7610906477d8bce6
# Verified Classes:   CompactionCreditViolation
# Defensive Guards:   2 fail-closed invariants (zero external deps)

class CompactionCreditViolation(RuntimeError):
    """Raised when a budget arbiter admits an operation on a turn whose pending
    compaction will push post-compaction occupancy past the hard high-water mark.
    This is the boundary condition described in the article: per-turn feasibility
    is satisfied, but the credit ordering invariant is violated."""


def admit_retrieval(
    occupancy_current: int,
    occupancy_after_pending_compaction: int | None,
    high_water_mark: int,
    token_budget: int,
    t_retrieval: int,
    t_output: int,
) -> bool:
    """Decide whether to admit a retrieval operation on this turn.

    `occupancy_after_pending_compaction` is None when no compaction is pending.
    The naive check uses `occupancy_current`; the correct check uses the
    post-compaction occupancy, because that is the baseline the *next* turn
    will be judged against.
    """
    if occupancy_after_pending_compaction is not None:
        # A compaction is pending. It must satisfy its hard post-condition
        # before we may admit retrieval against the post-compaction baseline.
        if occupancy_after_pending_compaction > high_water_mark:
            raise CompactionCreditViolation(
                f"pending compaction leaves occupancy "
                f"{occupancy_after_pending_compaction} > H={high_water_mark}; "
                f"refusing to admit retrieval against a baseline that will "
                f"not hold. Escalate to hard reset."
            )
        baseline = occupancy_after_pending_compaction
    else:
        baseline = occupancy_current

    projected = baseline + t_retrieval + t_output
    if projected > token_budget:
        # Honest rejection: the projected occupancy, not the current one.
        return False
    return True


def run_arbiter_turn(state: dict) -> dict:
    """One turn of the arbiter. Raises on credit-order violation rather than
    silently shedding retrieval, which is the failure mode being prevented."""
    try:
        admitted = admit_retrieval(
            occupancy_current=state["occupancy"],
            occupancy_after_pending_compaction=state.get("post_compaction_occupancy"),
            high_water_mark=state["high_water_mark"],
            token_budget=state["token_budget"],
            t_retrieval=state["t_retrieval"],
            t_output=state["t_output"],
        )
    except CompactionCreditViolation as exc:
        state["escalate_hard_reset"] = True
        state["last_error"] = str(exc)
        raise
    state["retrieval_admitted"] = admitted
    state["rejection_reason"] = None if admitted else "compaction_credit_projected_over_budget"
    return state