# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-accept-poisoned-episodic-memories-after-a-successful-retrieval-time-filter-the-write-path-read-path-asymmetry-in-shared-vector-st.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-accept-poisoned-episodic-memories-after-a-successful-retrieval-time-filter-the-write-path-read-path-asymmetry-in-shared-vector-st.warrant.json
# Markdown SHA-256:   a709ff25af3a202d3ae813296c109301bc4897c0ac6e2ee3186d0ee8fa295ce0
# Code AST SHA-256:   3d942971dc31ffa5d25f5ff70cc6804039a999107d118fa4b866eae07898eaae
# Verified Classes:   PoisonedAdmissionError
# Defensive Guards:   3 fail-closed invariants (zero external deps)

class PoisonedAdmissionError(RuntimeError):
    """Raised when a memory write would be admitted despite failing the trust invariant.

    This is a *write-path* guard. It must fire before indexing, consolidation,
    or ranking-candidate logging, not at retrieval time.
    """
    pass


def admit_memory(
    memory: dict,
    tenant_centroid: list[float],
    provenance_score: float,
    min_provenance: float = 0.6,
    max_centroid_distance: float = 0.35,
) -> dict:
    """Admission gate for a shared episodic/vector store.

    Rejects writes that would poison the index, the consolidation job, or the
    ranking model. A read-path filter is NOT a substitute for this check.
    """
    if provenance_score < min_provenance:
        raise PoisonedAdmissionError(
            f"write rejected: provenance {provenance_score:.2f} "
            f"< {min_provenance:.2f} (source={memory.get('source')!r})"
        )

    dist = _cosine_distance(memory["embedding"], tenant_centroid)
    if dist > max_centroid_distance:
        raise PoisonedAdmissionError(
            f"write rejected: embedding centroid distance {dist:.3f} "
            f"> {max_centroid_distance:.3f} (id={memory.get('id')!r})"
        )

    # Admitted: safe to index, consolidate, and train on.
    return {**memory, "admitted": True}


def consolidate(admitted_memories: list[dict], summarizer) -> list[dict]:
    """Consolidation MUST consume only admitted memories.

    Feeding the unfiltered admitted set here is the laundering path: poisoned
    entries get paraphrased into high-provenance synthesized facts.
    """
    for m in admitted_memories:
        if not m.get("admitted"):
            raise PoisonedAdmissionError(
                f"consolidation input contains un-admitted memory id={m.get('id')!r}; "
                "route consolidation through the admission gate"
            )
    return summarizer(admitted_memories)


def _cosine_distance(a: list[float], b: list[float]) -> float:
    dot = sum(x * y for x, y in zip(a, b))
    na = sum(x * x for x in a) ** 0.5
    nb = sum(y * y for y in b) ** 0.5
    if na == 0 or nb == 0:
        return 1.0
    return 1.0 - dot / (na * nb)