# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-does-a-per-hop-retry-budget-amplify-a-transient-slowdown-into-a-sustained-outage.md
# Remnant Warrant:    https://corpusfetch.com/c/when-does-a-per-hop-retry-budget-amplify-a-transient-slowdown-into-a-sustained-outage.warrant.json
# Markdown SHA-256:   0a1158b43498ad1a32bb07467f4206a55b985cdab81911310bfd7db55c4b9993
# Code AST SHA-256:   0baedd9339e26d0451955bb1d9a829d5c155a9bf3cd65337d02a087d9160d1a8
# Verified Classes:   RetryAmplificationError, RetryBudgetGate
# Defensive Guards:   1 fail-closed invariants (zero external deps)

from dataclasses import dataclass


class RetryAmplificationError(RuntimeError):
    """Raised when per-hop retry fan-out exceeds the terminal service's absorbable slack."""


@dataclass(frozen=True)
class RetryBudgetGate:
    hops: int
    retries_per_hop: int
    terminal_slack: float

    def verify(self, deadline_propagated: bool = False) -> bool:
        if deadline_propagated:
            return True
        fanout = (1 + self.retries_per_hop) ** self.hops
        if fanout > self.terminal_slack:
            raise RetryAmplificationError(
                f"fan-out {fanout} exceeds slack {self.terminal_slack} "
                f"for hops={self.hops} retries={self.retries_per_hop} with no deadline"
            )
        return True


def dispatch_retry_policy(gate: RetryBudgetGate, deadline_propagated: bool = False) -> str:
    try:
        gate.verify(deadline_propagated)
    except RetryAmplificationError:
        return "shared-budget"
    return "per-hop" if not deadline_propagated else "deadline-bounded"