# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-collusive-quorum-capture-in-a-byzantine-resilient-planner-when-adversarial-agents-stay-below-the-per-message-fault-thresho.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-collusive-quorum-capture-in-a-byzantine-resilient-planner-when-adversarial-agents-stay-below-the-per-message-fault-thresho.warrant.json
# Markdown SHA-256:   c0f44ca2e97b3642076e854a9cdb29dc310310266a5630340a52542ef35314ef
# Code AST SHA-256:   1514de26339788c927c0228f076a2ed08ad85e56b450292fb54e037472ce57f5
# Verified Classes:   CollusiveQuorumCaptureError
# Defensive Guards:   4 fail-closed invariants (zero external deps)

import statistics
from collections import deque


class CollusiveQuorumCaptureError(RuntimeError):
    """Raised when cross-round drift autocorrelation reveals rotation-based capture
    even though every per-round fault count stayed within the Byzantine budget."""
    pass


def detect_rotation_capture(
    accepted_displacements: list[float],
    per_round_fault_counts: list[int],
    f: int,
    window: int,
    autocorr_threshold: float,
) -> bool:
    if len(accepted_displacements) != len(per_round_fault_counts):
        raise CollusiveQuorumCaptureError(
            "displacement and fault-count streams are misaligned; cannot attribute drift"
        )
    if window <= 2 * f + 1:
        raise CollusiveQuorumCaptureError(
            f"window={window} too short for f={f}; rotation period can be f+1"
        )
    if any(c > f for c in per_round_fault_counts):
        # A genuine per-round violation is a different failure mode; do not
        # conflate it with rotation capture.
        raise CollusiveQuorumCaptureError(
            "per-round fault count exceeded budget; this is not the rotation case"
        )

    recent = accepted_displacements[-window:]
    if len(recent) < window:
        return False

    mean = statistics.fmean(recent)
    centered = [x - mean for x in recent]
    denom = sum(c * c for c in centered)
    if denom == 0.0:
        return False
    lag1 = sum(centered[i] * centered[i - 1] for i in range(1, len(centered))) / denom

    if lag1 > autocorr_threshold:
        return True
    return False


def guard_aggregation_round(
    accepted_displacements: list[float],
    per_round_fault_counts: list[int],
    f: int,
    window: int = 9,
    autocorr_threshold: float = 0.35,
) -> None:
    if detect_rotation_capture(
        accepted_displacements, per_round_fault_counts, f, window, autocorr_threshold
    ):
        raise CollusiveQuorumCaptureError(
            "cross-round drift autocorrelation indicates collusive rotation; "
            "freeze identity cohort for probation window before resuming"
        )