# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-refuse-to-join-a-residual-because-the-rate-limit-it-assumed-has-since-been-lowered.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-refuse-to-join-a-residual-because-the-rate-limit-it-assumed-has-since-been-lowered.warrant.json
# Markdown SHA-256:   c398ddd4f77983223eeec0480e60fedd50a67c2b52b46e51fdc7e9313b98d086
# Code AST SHA-256:   8d939187589a58b09f97d68838855f7c1b6e57987f2be0993cf47faab52196b2
# Verified Classes:   LoweredRateLimitRefusalError, ResidualThroughputSpec, ThrottledExecutionPlan, RateLimitAdaptationGate
# Defensive Guards:   2 fail-closed invariants (zero external deps)

import math
from dataclasses import dataclass
from typing import Final


class LoweredRateLimitRefusalError(RuntimeError):
    '''Raised when a residual's throughput schedule exceeds a lowered rate limit or violates timeout feasibility.'''
    pass


@dataclass(frozen=True)
class ResidualThroughputSpec:
    residual_id: str
    assumed_qps: float
    assumed_concurrency: int
    total_planned_requests: int


@dataclass(frozen=True)
class ThrottledExecutionPlan:
    residual_id: str
    legacy_schedule_refused: bool
    feasible_within_timeout: bool
    enforced_safe_qps: float
    enforced_concurrency: int
    min_inter_request_interval_sec: float
    estimated_duration_sec: float


class RateLimitAdaptationGate:
    '''
    Evaluates whether a residual must be refused or re-paced because the active
    rate limit is lower than the rate limit assumed when the record was written:
    1. Refuses the residual's concurrency/pacing constants when `live_qps < assumed_qps`.
    2. Applies an 80% safety headroom factor (`SAFE_UTILIZATION_RATIO = 0.80`).
    3. Refuses execution upfront if `estimated_duration_sec > max_timeout_sec`.
    '''

    SAFE_UTILIZATION_RATIO: Final[float] = 0.80

    @classmethod
    def compute_adapted_plan(
        cls,
        spec: ResidualThroughputSpec,
        *,
        live_qps_limit: float,
        avg_request_latency_sec: float,
        max_timeout_sec: float,
    ) -> ThrottledExecutionPlan:
        if live_qps_limit <= 0:
            raise LoweredRateLimitRefusalError("live_qps_limit must be strictly positive.")

        legacy_refused = live_qps_limit < spec.assumed_qps
        effective_qps = min(spec.assumed_qps, live_qps_limit) * cls.SAFE_UTILIZATION_RATIO

        # Little's Law concurrency bound: Concurrency = Throughput * Latency (clamped >= 1)
        safe_concurrency = max(1, min(spec.assumed_concurrency, math.floor(effective_qps * avg_request_latency_sec) or 1))
        min_interval = 1.0 / effective_qps
        estimated_duration = spec.total_planned_requests / effective_qps
        feasible = estimated_duration <= max_timeout_sec

        return ThrottledExecutionPlan(
            residual_id=spec.residual_id,
            legacy_schedule_refused=legacy_refused,
            feasible_within_timeout=feasible,
            enforced_safe_qps=round(effective_qps, 4),
            enforced_concurrency=safe_concurrency,
            min_inter_request_interval_sec=round(min_interval, 4),
            estimated_duration_sec=round(estimated_duration, 2),
        )

    @classmethod
    def assert_rate_limit_feasible(
        cls,
        spec: ResidualThroughputSpec,
        *,
        live_qps_limit: float,
        avg_request_latency_sec: float,
        max_timeout_sec: float,
    ) -> ThrottledExecutionPlan:
        plan = cls.compute_adapted_plan(
            spec,
            live_qps_limit=live_qps_limit,
            avg_request_latency_sec=avg_request_latency_sec,
            max_timeout_sec=max_timeout_sec,
        )
        if not plan.feasible_within_timeout:
            raise LoweredRateLimitRefusalError(
                f"Refusing synchronous execution of residual '{spec.residual_id}': lowered rate limit "
                f"({live_qps_limit} QPS; safe={plan.enforced_safe_qps} QPS) requires {plan.estimated_duration_sec}s "
                f"for {spec.total_planned_requests} items, exceeding max_timeout_sec ({max_timeout_sec}s)."
            )
        return plan