# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/under-what-conditions-must-an-instance-treat-per-span-anomaly-scores-as-non-independent-before-attributing-tool-call-drift-or-prompt-leakage.md
# Remnant Warrant:    https://corpusfetch.com/c/under-what-conditions-must-an-instance-treat-per-span-anomaly-scores-as-non-independent-before-attributing-tool-call-drift-or-prompt-leakage.warrant.json
# Markdown SHA-256:   47c241254b27b3bcb921ae1a25b6626027bc833dd9028144c0cbdfa56954f6e9
# Code AST SHA-256:   b4842a56b86ab5270db065c90d4104bdc724953b096bd05512469af88fd38300
# Verified Classes:   CorrelatedSpanAggregationError, SpanScore
# Defensive Guards:   1 fail-closed invariants (zero external deps)

import math
from dataclasses import dataclass

# Calibrated bound above which within-trace span scores are treated as
# non-independent and must be collapsed to their shared ancestor.
CORRELATION_BOUND = 0.3


class CorrelatedSpanAggregationError(RuntimeError):
    """Raised when a caller attempts marginal aggregation over spans whose
    within-trace correlation invalidates the independence assumption."""
    pass


@dataclass
class SpanScore:
    span_id: str
    ancestor_id: str          # retrieval node / schema version / prompt hash
    score: float


def _mean_offdiag_correlation(scores: list[SpanScore]) -> float:
    # Placeholder: in production this is the rolling pairwise Pearson
    # correlation of span scores over the trace window. Here we approximate
    # with a shared-ancestor co-membership proxy for the operational gate.
    if len(scores) < 2:
        return 0.0
    n = len(scores)
    pairs = 0
    shared = 0
    for i in range(n):
        for j in range(i + 1, n):
            pairs += 1
            if scores[i].ancestor_id == scores[j].ancestor_id:
                shared += 1
    return shared / pairs


def attribute(scores: list[SpanScore], threshold: float) -> dict:
    """Attribute a trace-level anomaly to a causal span or shared ancestor.

    Refuses marginal aggregation when spans are non-independent.
    """
    rho = _mean_offdiag_correlation(scores)
    anomalous = [s for s in scores if s.score >= threshold]

    if rho > CORRELATION_BOUND and len(anomalous) >= 2:
        ancestors = {s.ancestor_id for s in anomalous}
        if len(ancestors) < len(anomalous):
            # Multiple anomalous spans share one latent event: collapse.
            return {
                "attribution": "shared_ancestor",
                "ancestors": sorted(ancestors),
                "independent_events": len(ancestors),
                "severity": len(ancestors),  # NOT len(anomalous)
            }
        raise CorrelatedSpanAggregationError(
            f"within-trace correlation {rho:.2f} exceeds bound "
            f"{CORRELATION_BOUND} but anomalous spans do not share an "
            f"ancestor; marginal aggregation is invalid and no joint model "
            f"is configured. Refusing to emit severity."
        )

    if not anomalous:
        return {"attribution": "none", "independent_events": 0, "severity": 0}

    top = max(anomalous, key=lambda s: s.score)
    return {
        "attribution": "span",
        "span_id": top.span_id,
        "independent_events": len(anomalous),
        "severity": len(anomalous),
    }