# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-collusive-reputation-inflation-in-federated-agent-scoring-when-sybil-identity-cost-is-low.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-collusive-reputation-inflation-in-federated-agent-scoring-when-sybil-identity-cost-is-low.warrant.json
# Markdown SHA-256:   6d5e1c2066addca839379674d361b463bb0a0e85e22e436b2034c8f86a75b218
# Code AST SHA-256:   42c1a82751b72892b89fcaa73b77b957d1ba521c09f19cc375df9394d63a7656
# Verified Classes:   CollusionQuarantineError, Attestation
# Defensive Guards:   2 fail-closed invariants (zero external deps)

import math
from dataclasses import dataclass

class CollusionQuarantineError(RuntimeError):
    """Raised when a quarantine decision cannot be applied safely."""
    pass

@dataclass
class Attestation:
    src: str
    dst: str
    ts: float          # epoch seconds
    delta: float       # signed score contribution

def detect_collusive_inflation(
    target: str,
    attestations: list[Attestation],
    honest_latency_p99: float,
    network_delta_var: float,
    min_cluster: int = 5,
) -> list[str]:
    """
    Return the list of source agents whose attestations to `target`
    should be quarantined. Raises CollusionQuarantineError if the
    evidence is internally inconsistent (e.g., single attestation
    claiming to be a cluster).
    """
    inbound = [a for a in attestations if a.dst == target]
    if len(inbound) < min_cluster:
        return []

    # Group by source, measure mutual-latency proxy: cluster members
    # attest within a tight window of each other.
    inbound.sort(key=lambda a: a.ts)
    window = honest_latency_p99  # conservative: use p99 as window
    candidates: dict[str, list[Attestation]] = {}
    for a in inbound:
        candidates.setdefault(a.src, []).append(a)

    # Score-monotonicity: variance of deltas across the candidate set.
    deltas = [a.delta for a in inbound]
    mean_d = sum(deltas) / len(deltas)
    var_d = sum((d - mean_d) ** 2 for d in deltas) / len(deltas)

    # Temporal burst: fraction of inbound within any `window`-length span.
    burst = 0
    for i, a in enumerate(inbound):
        j = i
        while j < len(inbound) and inbound[j].ts - a.ts <= window:
            j += 1
        burst = max(burst, j - i)

    cluster = [s for s, atts in candidates.items() if len(atts) >= 2]
    if not cluster:
        raise CollusionQuarantineError(
            f"target {target!r}: burst={burst} but no repeat sources; "
            "cannot attribute to a cluster"
        )

    monotone = var_d < 0.10 * network_delta_var
    bursty = burst >= 3 * min_cluster
    if not (monotone and bursty):
        return []
    if len(cluster) < min_cluster:
        raise CollusionQuarantineError(
            f"target {target!r}: bursty={bursty} monotone={monotone} "
            f"but cluster size {len(cluster)} < {min_cluster}; "
            "evidence below action threshold"
        )
    return sorted(cluster)