# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/in-what-scenarios-must-an-instance-abandon-time-based-auto-compaction-in-favor-of-token-budgeted-compaction-during-long-horizon-agent-sessions.md
# Remnant Warrant:    https://corpusfetch.com/c/in-what-scenarios-must-an-instance-abandon-time-based-auto-compaction-in-favor-of-token-budgeted-compaction-during-long-horizon-agent-sessions.warrant.json
# Markdown SHA-256:   40e42619ca4359eae87a42469dab3fd44b219834ace01906bb2651fef2a8b2e8
# Code AST SHA-256:   ef2b6be34d50972e31eb333c5ffb1c737cb2daa05785a106f1140a8d57cf47f0
# Verified Classes:   CompactionInvariantError
# Defensive Guards:   4 fail-closed invariants (zero external deps)

class CompactionInvariantError(RuntimeError):
    """Raised when a compaction trigger cannot guarantee the overflow bound."""
    pass


def select_compaction_trigger(
    max_tokens_per_interval: int,
    window_tokens: int,
    alpha: float,
    reserved_overhead: int,
    has_idle_gaps: bool,
    max_single_turn_injection: int,
) -> str:
    """Decide whether a time-based cadence is sound for this session.

    Returns the trigger strategy to install. Raises if the caller supplies
    parameters under which neither strategy can guarantee the overflow bound.
    """
    if not (0.0 < alpha < 1.0):
        raise CompactionInvariantError(
            f"alpha={alpha} outside (0,1); budget fraction is meaningless."
        )

    headroom = int((1.0 - alpha) * window_tokens) - reserved_overhead
    if headroom <= 0:
        raise CompactionInvariantError(
            "No headroom after reserving overhead: "
            f"window={window_tokens}, alpha={alpha}, overhead={reserved_overhead}."
        )

    # A single turn larger than the headroom can overflow before any
    # admission-time check can intervene unless we budget per-admission.
    if max_single_turn_injection > headroom:
        # Token budgeting at admission is the only sound option; the cadence
        # cannot absorb a burst larger than the headroom it protects.
        return "token_budgeted_at_admission"

    cadence_absorbs = max_tokens_per_interval <= headroom
    if cadence_absorbs and not has_idle_gaps:
        return "time_based"

    if has_idle_gaps:
        # Timer would fire on an under-full buffer and compound summary loss.
        return "token_budgeted_at_admission"

    # Rate exceeds cadence absorption capacity: schedule is unsound.
    raise CompactionInvariantError(
        "Time-based cadence is unsound: max_tokens_per_interval="
        f"{max_tokens_per_interval} exceeds headroom={headroom}. "
        "Install token-budgeted compaction at admission."
    )


def admit_turn(live_tokens: int, incoming_tokens: int, window_tokens: int,
               alpha: float, compact_fn) -> int:
    """Admission-time budget check: compact BEFORE the injection lands."""
    budget = int(alpha * window_tokens)
    if live_tokens + incoming_tokens > budget:
        live_tokens = compact_fn(live_tokens)  # returns post-compaction size
        if live_tokens + incoming_tokens > window_tokens:
            raise CompactionInvariantError(
                "Post-compaction buffer still cannot admit turn: "
                f"live={live_tokens}, incoming={incoming_tokens}, "
                f"window={window_tokens}."
            )
    return live_tokens + incoming_tokens