# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-of-the-agent-observability-pipeline-silently-drop-memory-mutation-spans-when-the-safety-verdict-emitter-backpressures.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-of-the-agent-observability-pipeline-silently-drop-memory-mutation-spans-when-the-safety-verdict-emitter-backpressures.warrant.json
# Markdown SHA-256:   9a27b84b52d42cde5ecbb5bd47eef9c4a5157edcde54e921ea175f53b14d4385
# Code AST SHA-256:   02fc6c37ce9fe95844647fae30f969094b6cb3e51b2de5088203c017790151e1
# Verified Classes:   AmbiguousTraceError
# Defensive Guards:   4 fail-closed invariants (zero external deps)

class AmbiguousTraceError(RuntimeError):
    """Raised when a trace's mutation absence cannot be distinguished from pipeline loss."""
    pass


REQUIRED_PREDECESSORS = ("terminal_span", "safety_verdict")


def classify_mutation_absence(trace: dict) -> str:
    """
    Returns 'none' or 'unknown'. Never returns 'none' when eviction is present
    and the verdict's latency could have triggered tail eviction.
    """
    evict = trace.get("buffer_evict", 0)
    dropped = trace.get("spans_dropped", 0)
    mutation_count = trace.get("mutation_span_count", 0)
    verdict_latency_ms = trace.get("verdict_emit_latency_ms", 0.0)
    evict_window_ms = trace.get("evict_pressure_window_ms", 40.0)

    if mutation_count > 0:
        return "present"

    if evict == 0 and dropped == 0:
        return "none"

    if evict > 0 and verdict_latency_ms >= evict_window_ms:
        # Eviction occurred and the verdict was slow enough to have caused it.
        # Mutation absence is indistinguishable from loss.
        raise AmbiguousTraceError(
            f"trace {trace.get('trace_id')}: mutation absence ambiguous "
            f"(buffer_evict={evict}, verdict_latency_ms={verdict_latency_ms:.1f}, "
            f"evict_window_ms={evict_window_ms:.1f}); "
            f"trace must be quarantined, not sealed as complete"
        )

    if evict > 0 and verdict_latency_ms < evict_window_ms:
        # Eviction attributable to another producer; require emission-count match.
        if trace.get("mutation_emitted_count", 0) != trace.get("mutation_received_count", 0):
            raise AmbiguousTraceError(
                f"trace {trace.get('trace_id')}: mutation emission/receipt mismatch "
                f"({trace.get('mutation_emitted_count')} emitted vs "
                f"{trace.get('mutation_received_count')} received) despite low verdict latency"
            )
        return "none"

    raise AmbiguousTraceError(
        f"trace {trace.get('trace_id')}: unclassified eviction state"
    )


def seal_trace(trace: dict) -> dict:
    for pred in REQUIRED_PREDECESSORS:
        if not trace.get(pred):
            raise AmbiguousTraceError(
                f"trace {trace.get('trace_id')}: missing required predecessor {pred!r}; "
                f"refusing to seal as complete"
            )
    verdict = classify_mutation_absence(trace)
    trace["mutation_absence_class"] = verdict
    trace["status"] = "complete" if verdict == "none" else "quarantined"
    return trace