# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-of-prefix-cache-reuse-silently-degrade-when-compaction-runs-concurrently-with-a-cache-miss-storm.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-of-prefix-cache-reuse-silently-degrade-when-compaction-runs-concurrently-with-a-cache-miss-storm.warrant.json
# Markdown SHA-256:   ec5aa863585d835234d6036715e8c33f54d29bc8ebbeaca41cc27bc58acf2a9a
# Code AST SHA-256:   2dbb00dafb7e18c90e51ff4215339ec2915bbf585c2db9db86ab13880a20c57c
# Verified Classes:   CompactionCacheContentionError, ReuseWindow
# Defensive Guards:   1 fail-closed invariants (zero external deps)

from dataclasses import dataclass


class CompactionCacheContentionError(RuntimeError):
    """Raised when compaction would invalidate a warm shared prefix subtree."""


@dataclass
class ReuseWindow:
    cached_tokens: int
    total_prompt_tokens: int
    requests_with_hit: int
    total_requests: int

    @property
    def token_weighted(self) -> float:
        if self.total_prompt_tokens == 0:
            return 0.0
        return self.cached_tokens / self.total_prompt_tokens

    @property
    def request_weighted(self) -> float:
        if self.total_requests == 0:
            return 0.0
        return self.requests_with_hit / self.total_requests

    @property
    def divergence(self) -> float:
        return self.request_weighted - self.token_weighted


def schedule_compaction(
    window: ReuseWindow,
    subtree_hit_counts: dict[str, int],
    warm_hit_floor: int = 50,
    divergence_threshold: float = 0.5,
) -> list[str]:
    """Return subtree ids safe to compact. Refuses to compact warm subtrees
    while the cache is in a thin-reuse regime."""
    if window.divergence > divergence_threshold:
        warm = [sid for sid, hits in subtree_hit_counts.items() if hits >= warm_hit_floor]
        if warm:
            raise CompactionCacheContentionError(
                f"thin-reuse regime (divergence={window.divergence:.2f}, "
                f"token_weighted={window.token_weighted:.2f}); "
                f"refusing to compact {len(warm)} warm subtrees: {warm[:3]}"
            )
    return [sid for sid, hits in subtree_hit_counts.items() if hits < warm_hit_floor]


if __name__ == "__main__":
    bad = ReuseWindow(cached_tokens=900, total_prompt_tokens=7500,
                      requests_with_hit=90, total_requests=100)
    try:
        schedule_compaction(bad, {"doc-prefix-a": 400, "doc-prefix-b": 12})
    except CompactionCacheContentionError as exc:
        print(f"deferred: {exc}")