# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-taint-span-correlation-collapse-when-genai-observability-pipelines-merge-asynchronous-security-graphs.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-taint-span-correlation-collapse-when-genai-observability-pipelines-merge-asynchronous-security-graphs.warrant.json
# Markdown SHA-256:   bd29fc67dcabecfa47bb2b4606c6e28c9c8675a7a12bb51851627a3ab9d011ff
# Code AST SHA-256:   cf0df7dd6eb2e0006343038a285073807227c89d8919494b9ab9defb972cfd7a
# Verified Classes:   CorrelationInvariantError, TaintNode, Span
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from typing import Optional

MAX_SKEW_MS = 50


class CorrelationInvariantError(RuntimeError):
    """Raised when a span-taint join would violate a correlation invariant."""
    pass


@dataclass(frozen=True)
class TaintNode:
    node_id: str
    label_version: int
    emitted_at_ms: int


@dataclass(frozen=True)
class Span:
    trace_id: str
    span_id: str
    emission_seq: int
    emitted_at_ms: int


def correlate(span: Span, taint: TaintNode, bound_edges: dict) -> Optional[tuple]:
    key = (span.trace_id, span.span_id)

    if key in bound_edges:
        existing = bound_edges[key]
        if existing != (taint.node_id, taint.label_version):
            raise CorrelationInvariantError(
                f"span {key} already bound to {existing}; "
                f"refusing to overwrite with {(taint.node_id, taint.label_version)}"
            )

    if taint.emitted_at_ms > span.emitted_at_ms:
        raise CorrelationInvariantError(
            f"taint node {taint.node_id} emitted after span {key} "
            f"({taint.emitted_at_ms} > {span.emitted_at_ms}); ordering violated"
        )

    skew = span.emitted_at_ms - taint.emitted_at_ms
    if skew > MAX_SKEW_MS:
        # Degrade to no annotation rather than guess.
        return None

    edge = (taint.node_id, taint.label_version)
    bound_edges[key] = edge
    return edge


def reconcile(batch: list) -> dict:
    bound: dict = {}
    for span, taint in batch:
        try:
            correlate(span, taint, bound)
        except CorrelationInvariantError as exc:
            # Quarantine the edge; never silently drop the invariant violation.
            raise CorrelationInvariantError(
                f"quarantined edge for trace {span.trace_id}: {exc}"
            ) from exc
    return bound