# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-happens-when-an-instance-attributes-a-memory-retrieval-span-to-the-wrong-semantic-convention-during-concurrent-tool-call-fan-out.md
# Remnant Warrant:    https://corpusfetch.com/c/what-happens-when-an-instance-attributes-a-memory-retrieval-span-to-the-wrong-semantic-convention-during-concurrent-tool-call-fan-out.warrant.json
# Markdown SHA-256:   10fccce70d91aa973894b2c02ab74d5d5f825d7d540f2b26e7ce2d4b811b0da2
# Code AST SHA-256:   2df71d58b7e9af882be72927139d3f49117e58fd759cf3afc71ec6195be3fd04
# Verified Classes:   RetrievalSpanAttributionError
# Defensive Guards:   4 fail-closed invariants (zero external deps)

class RetrievalSpanAttributionError(RuntimeError):
    """Raised when a retrieval span's identity attributes are inconsistent
    with its parent fan-out declaration, indicating attribute bleed."""
    pass


def validate_retrieval_span(span: dict, parent_sources: set[str]) -> None:
    source_id = span.get("gen_ai.data_source.id")
    query = span.get("gen_ai.retrieval.query")
    result_count = span.get("gen_ai.retrieval.result_count")

    if source_id is None or query is None:
        raise RetrievalSpanAttributionError(
            f"retrieval span missing identity attributes: source={source_id!r} query={query!r}"
        )
    if source_id not in parent_sources:
        raise RetrievalSpanAttributionError(
            f"span source {source_id!r} not declared by parent fan-out {sorted(parent_sources)}"
        )
    if result_count is not None and result_count < 0:
        raise RetrievalSpanAttributionError(
            f"negative result_count {result_count} on source {source_id!r}"
        )


def detect_bleed(spans: list[dict], parent_sources: set[str]) -> None:
    for s in spans:
        validate_retrieval_span(s, parent_sources)
    # Bleed signature: multiple distinct sources collapsing to one query string
    # while result counts diverge.
    queries_by_source: dict[str, set[str]] = {}
    for s in spans:
        queries_by_source.setdefault(s["gen_ai.data_source.id"], set()).add(
            s["gen_ai.retrieval.query"]
        )
    if len(parent_sources) > 1 and len(queries_by_source) == 1:
        raise RetrievalSpanAttributionError(
            "all fan-out branches collapsed to a single data_source.id; "
            "suspect shared-mutable-context attribute bleed"
        )