# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-memory-poisoning-from-surviving-a-write-provenance-audit-when-anomaly-scores-are-computed-on-recall-rather-than-on-write.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-memory-poisoning-from-surviving-a-write-provenance-audit-when-anomaly-scores-are-computed-on-recall-rather-than-on-write.warrant.json
# Markdown SHA-256:   98e046f1b0696bf63ce7e68bd053c9db9e3aef58219b3a044bd4d128438fa313
# Code AST SHA-256:   beb43c903b0a15bef62edc3583bd132a3277031182e6d234bc9204a2bae6e9e1
# Verified Classes:   ProvenanceDisjointnessError, RecallEvent
# Defensive Guards:   1 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from typing import FrozenSet, Mapping


class ProvenanceDisjointnessError(RuntimeError):
    """Raised when a recall event violates the write/recall disjointness invariant."""
    pass


@dataclass(frozen=True)
class RecallEvent:
    record_id: str
    writer_id: str
    query_class: str
    recall_anomaly_z: float


def audit_recall(
    event: RecallEvent,
    baseline_seeders: Mapping[str, FrozenSet[str]],
) -> None:
    """Gate the audit decision on structural disjointness, not on the anomaly score.

    baseline_seeders maps query_class -> set of writer identities that
    contributed queries to that class's baseline. If the record's writer
    is in that set, the baseline is attacker-influenceable and the
    anomaly score carries no information about write integrity.
    """
    seeders = baseline_seeders.get(event.query_class, frozenset())
    if event.writer_id in seeders:
        raise ProvenanceDisjointnessError(
            f"record {event.record_id!r} written by {event.writer_id!r} is "
            f"recalled under query class {event.query_class!r}, but that "
            f"writer also seeded the baseline for that class "
            f"(z={event.recall_anomaly_z:.2f}); anomaly score is "
            f"adversarially controllable and MUST NOT gate the audit"
        )
    # Only here is the anomaly score meaningful as a prioritization signal.
    return None


def prioritize(queue: list[RecallEvent], baseline_seeders: Mapping[str, FrozenSet[str]]) -> list[RecallEvent]:
    audited = []
    for ev in queue:
        try:
            audit_recall(ev, baseline_seeders)
        except ProvenanceDisjointnessError:
            audited.append(ev)  # structural violation: always audit
    return sorted(audited, key=lambda e: -e.recall_anomaly_z)