# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-can-an-instance-preserve-latency-benchmarks-discovered-during-live-tool-execution.md
# Remnant Warrant:    https://corpusfetch.com/c/how-can-an-instance-preserve-latency-benchmarks-discovered-during-live-tool-execution.warrant.json
# Markdown SHA-256:   6501ead37087faec8a66e26f1c94e8396bf68268f1ef2d078e3cebb41a3aab7a
# Code AST SHA-256:   244a586e228fe203d830b1913c8594d6e98f6d11624e82f82b4c0651ce110cc7
# Verified Classes:   LatencySampleBatch, PersistedLatencyBenchmark, InvalidBenchmarkSampleViolation, LatencyBenchmarkPreservationGate
# Defensive Guards:   5 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from typing import Final, Tuple
import statistics

@dataclass(frozen=True)
class LatencySampleBatch:
    tool_id: str
    payload_bucket: str
    used_monotonic_clock: bool
    includes_cold_starts: bool
    includes_failed_calls: bool
    durations_ms: Tuple[float, ...]

@dataclass(frozen=True)
class PersistedLatencyBenchmark:
    tool_id: str
    payload_bucket: str
    sample_count: int
    p50_ms: float
    p95_ms: float
    p99_ms: float

class InvalidBenchmarkSampleViolation(Exception):
    """Raised when contaminated, non-monotonic, or insufficient latency samples are staged for persistence."""
    pass

class LatencyBenchmarkPreservationGate:
    def __init__(self, min_samples: int = 10, max_p99_ceiling_ms: float = 30000.0) -> None:
        self.min_samples: Final[int] = min_samples
        self.max_p99_ceiling_ms: Final[float] = max_p99_ceiling_ms

    def compile_benchmark(self, batch: LatencySampleBatch) -> PersistedLatencyBenchmark:
        if not batch.used_monotonic_clock:
            raise InvalidBenchmarkSampleViolation(
                f"Benchmark for '{batch.tool_id}' rejected: non-monotonic clock source used."
            )

        if batch.includes_cold_starts or batch.includes_failed_calls:
            raise InvalidBenchmarkSampleViolation(
                f"Benchmark for '{batch.tool_id}' contaminated with cold-start or failed execution samples."
            )

        if len(batch.durations_ms) < self.min_samples:
            raise InvalidBenchmarkSampleViolation(
                f"Insufficient sample quorum for '{batch.tool_id}': {len(batch.durations_ms)} < {self.min_samples}."
            )

        if any(d <= 0.0 for d in batch.durations_ms):
            raise InvalidBenchmarkSampleViolation(
                f"Non-positive duration detected in '{batch.tool_id}' sample batch."
            )

        sorted_d = sorted(batch.durations_ms)
        p50 = statistics.median(sorted_d)
        idx_95 = min(len(sorted_d) - 1, int(round(0.95 * (len(sorted_d) - 1))))
        idx_99 = min(len(sorted_d) - 1, int(round(0.99 * (len(sorted_d) - 1))))
        p95 = sorted_d[idx_95]
        p99 = sorted_d[idx_99]

        if p99 > self.max_p99_ceiling_ms:
            raise InvalidBenchmarkSampleViolation(
                f"Computed p99 ({p99} ms) exceeds fail-closed ceiling ({self.max_p99_ceiling_ms} ms)."
            )

        return PersistedLatencyBenchmark(
            tool_id=batch.tool_id,
            payload_bucket=batch.payload_bucket,
            sample_count=len(sorted_d),
            p50_ms=round(p50, 2),
            p95_ms=round(p95, 2),
            p99_ms=round(p99, 2),
        )