# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-happens-when-an-instance-mutates-memory-during-a-span-it-does-not-own-correlation-loss-across-agent-spans-memory-writes-and-compaction.md
# Remnant Warrant:    https://corpusfetch.com/c/what-happens-when-an-instance-mutates-memory-during-a-span-it-does-not-own-correlation-loss-across-agent-spans-memory-writes-and-compaction.warrant.json
# Markdown SHA-256:   30a25303014ef57092d2842dd9d0c9321b2b06df1ad746d39693328b4422b320
# Code AST SHA-256:   61105b0ede5fcf7ae88613a7d51ae02a36d999f930871fb518dcf82eba6091b1
# Verified Classes:   OrphanedMutationError, MutationEnvelope, MemoryWriteQueue
# Defensive Guards:   1 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass


class OrphanedMutationError(RuntimeError):
    """Raised when a memory mutation's flush-time span does not match its enqueue-time span."""


@dataclass(frozen=True)
class MutationEnvelope:
    mutation_id: str
    enqueue_span_id: str
    enqueue_trace_id: str
    payload: bytes


class MemoryWriteQueue:
    def __init__(self) -> None:
        self._items: list[MutationEnvelope] = []

    def enqueue(self, mutation_id: str, span_id: str, trace_id: str, payload: bytes) -> None:
        # Context is captured HERE, at enqueue time, not at flush time.
        self._items.append(
            MutationEnvelope(
                mutation_id=mutation_id,
                enqueue_span_id=span_id,
                enqueue_trace_id=trace_id,
                payload=payload,
            )
        )

    def flush(self, active_span_id: str) -> list[MutationEnvelope]:
        drained, self._items = self._items, []
        for item in drained:
            if item.enqueue_span_id != active_span_id:
                # The write is being flushed under a span that did not enqueue it.
                # Do NOT silently stamp it with the active span; surface the mismatch.
                raise OrphanedMutationError(
                    f"mutation {item.mutation_id!r} enqueued under span "
                    f"{item.enqueue_span_id!r} but flushed under span {active_span_id!r}"
                )
        return drained