# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-stop-trusting-its-own-span-correlation-after-memory-mutation.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-stop-trusting-its-own-span-correlation-after-memory-mutation.warrant.json
# Markdown SHA-256:   4411454efaf3d105e0b1a8c8eb988964c90d566f45dc2502e5de9ce8af386b27
# Code AST SHA-256:   3762b28686ba1fe1ff016a99098bbf857fc89e3f4f404c1d8ca464c75e9f3949
# Verified Classes:   EntityCorrelationError, CapturedEntity, SpanRecord
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from typing import Optional


class EntityCorrelationError(RuntimeError):
    """Raised when a span's entity correlation cannot be trusted."""
    pass


@dataclass(frozen=True)
class CapturedEntity:
    entity_id: str
    revision: int
    captured_at: float  # monotonic seconds, at span creation


@dataclass(frozen=True)
class SpanRecord:
    trace_id: str
    span_id: str
    start_time: float
    end_time: float
    entity: Optional[CapturedEntity]


def verify_entity_correlation(
    span: SpanRecord,
    latest_revision: int,
    last_mutation_commit_time: float,
    export_time: float,
) -> str:
    """
    Decide whether the span's entity correlation is trustworthy.

    Returns 'verified' or 'unverified'. Raises EntityCorrelationError when the
    span is structurally corrupt (captured revision postdates execution, which
    can only happen under lazy resolution).
    """
    if span.entity is None:
        return "unverified"

    # Diagnostic: a captured revision must be observable no later than the
    # span's own start. A newer capture means the attribute was resolved late.
    if span.entity.captured_at > span.start_time + 1e-6:
        raise EntityCorrelationError(
            f"span {span.span_id} captured entity {span.entity.entity_id} "
            f"at {span.entity.captured_at} after start {span.start_time}: "
            f"lazy resolution detected, span is corrupt"
        )

    # A mutation committed during the span's lifetime, or between execution
    # and export, makes the join ambiguous if the captured revision is stale.
    mutation_in_window = last_mutation_commit_time >= span.start_time
    if mutation_in_window and span.entity.revision < latest_revision:
        return "unverified"

    # Captured revision newer than collector's latest is impossible under
    # eager capture and indicates a split-brain revision pointer.
    if span.entity.revision > latest_revision:
        raise EntityCorrelationError(
            f"span {span.span_id} claims revision {span.entity.revision} "
            f"but collector latest is {latest_revision}: revision pointer "
            f"divergence"
        )

    return "verified"


def emit_span(
    span: SpanRecord,
    latest_revision: int,
    last_mutation_commit_time: float,
    export_time: float,
) -> dict:
    status = verify_entity_correlation(
        span, latest_revision, last_mutation_commit_time, export_time
    )
    payload = {
        "trace_id": span.trace_id,
        "span_id": span.span_id,
        "entity.correlation": status,
    }
    if status == "verified":
        payload["entity.id"] = span.entity.entity_id
        payload["entity.revision"] = span.entity.revision
    # Unverified spans deliberately omit entity identity so no downstream
    # join can attach them to a specific version.
    return payload