# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-happens-when-an-instance-loses-its-trace-parent-during-a-tool-invocation-orphaned-memory-operations-and-the-correlation-collapse-boundary.md
# Remnant Warrant:    https://corpusfetch.com/c/what-happens-when-an-instance-loses-its-trace-parent-during-a-tool-invocation-orphaned-memory-operations-and-the-correlation-collapse-boundary.warrant.json
# Markdown SHA-256:   448ab6a51551d2c72252ea895c9f09a4a7040a928be1ec6c0a9118601d993bbb
# Code AST SHA-256:   70a5068b5340549f80e2b9b714c9f69f0ce817b334a3bec642936081a4cc57c9
# Verified Classes:   CorrelationCollapseError, SpanContext
# Defensive Guards:   4 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass
from typing import Optional


class CorrelationCollapseError(RuntimeError):
    """Raised when a span cannot be structurally linked to its turn root."""


@dataclass(frozen=True)
class SpanContext:
    trace_id: str
    span_id: str
    parent_span_id: Optional[str]
    turn_id: Optional[str]


def emit_memory_write_span(
    propagated: Optional[SpanContext],
    logical_write_id: str,
) -> SpanContext:
    """Create a memory-write span that MUST be parented to the active turn span.

    Propagation across a concurrency boundary (thread pool, subprocess) is only
    valid via explicit copy. An absent context is NOT a license to fall back to
    a process-global default; that produces an orphaned root span which silently
    breaks causal correlation.
    """
    if propagated is None:
        raise CorrelationCollapseError(
            f"no propagated span context for memory write {logical_write_id!r}; "
            "refusing to parent to ambient default (would create orphaned root)"
        )
    if propagated.parent_span_id is None and propagated.turn_id is None:
        raise CorrelationCollapseError(
            f"propagated context for {logical_write_id!r} has neither parent nor "
            "turn_id; causal link to turn root is unverifiable"
        )

    # New span is a child of the propagated span, carrying the logical write id
    # so idempotent retries collapse on (trace_id, logical_write_id).
    return SpanContext(
        trace_id=propagated.trace_id,
        span_id=f"mem-{logical_write_id}",
        parent_span_id=propagated.span_id,
        turn_id=propagated.turn_id,
    )


def assert_causal_chain(
    turn_root: SpanContext,
    memory_write: SpanContext,
    tool_read: SpanContext,
) -> None:
    """Assert turn_root -> memory_write -> tool_read is structurally intact."""
    if memory_write.parent_span_id != turn_root.span_id:
        raise CorrelationCollapseError(
            f"memory write {memory_write.span_id!r} is not a child of turn root "
            f"{turn_root.span_id!r}; correlation is broken, not missing"
        )
    if tool_read.parent_span_id != memory_write.span_id:
        raise CorrelationCollapseError(
            f"tool read {tool_read.span_id!r} does not descend from memory write "
            f"{memory_write.span_id!r}; provenance is unverified"
        )