# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-of-an-agent-observability-pipeline-lose-causal-ordering-between-genai-spans-and-memory-mutation-events-when-context-edits-are-rec.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-of-an-agent-observability-pipeline-lose-causal-ordering-between-genai-spans-and-memory-mutation-events-when-context-edits-are-rec.warrant.json
# Markdown SHA-256:   1a5e256654f5f48eb737f5388f3f9e22e9fd9503f88e901fb74efcdde8e5a49c
# Code AST SHA-256:   4741cee7316db9d0078a76257832b2de7136535f82c90847b4db9ba9f476d944
# Verified Classes:   LineageAttributionError, GenAISpan, MemoryMutation
# Defensive Guards:   4 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass


class LineageAttributionError(RuntimeError):
    """Raised when a memory-mutation event cannot be causally attributed to a span."""


@dataclass(frozen=True)
class GenAISpan:
    trace_id: str
    span_id: str
    start_ms: int
    end_ms: int


@dataclass(frozen=True)
class MemoryMutation:
    trace_id: str
    event_id: str
    timestamp_ms: int
    parent_span_id: str | None


def attribute_mutation(
    mutation: MemoryMutation,
    spans: dict[str, GenAISpan],
    *,
    allow_interval_fallback: bool = False,
    max_clock_jitter_ms: int = 0,
) -> GenAISpan:
    # Causal join: trust only an explicit parent.
    if mutation.parent_span_id is not None:
        parent = spans.get(mutation.parent_span_id)
        if parent is None:
            raise LineageAttributionError(
                f"mutation {mutation.event_id} names parent {mutation.parent_span_id} "
                f"not present in trace {mutation.trace_id}"
            )
        if parent.trace_id != mutation.trace_id:
            raise LineageAttributionError(
                f"mutation {mutation.event_id} parent {parent.span_id} belongs to "
                f"trace {parent.trace_id}, not {mutation.trace_id}"
            )
        return parent

    # No causal parent. Interval fallback is opt-in and must be certified.
    if not allow_interval_fallback:
        raise LineageAttributionError(
            f"mutation {mutation.event_id} has no parent_span_id; "
            f"refusing interval fallback (lineage_incomplete)"
        )

    candidates = [
        s
        for s in spans.values()
        if s.trace_id == mutation.trace_id
        and s.start_ms - max_clock_jitter_ms
        <= mutation.timestamp_ms
        <= s.end_ms + max_clock_jitter_ms
    ]
    if len(candidates) != 1:
        raise LineageAttributionError(
            f"mutation {mutation.event_id} matched {len(candidates)} spans by interval; "
            f"attribution is ambiguous"
        )
    return candidates[0]