# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-distrust-a-write-audit-trail-detecting-memory-poisoning-when-the-audit-log-itself-is-written-by-the-poisoned-agent.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-distrust-a-write-audit-trail-detecting-memory-poisoning-when-the-audit-log-itself-is-written-by-the-poisoned-agent.warrant.json
# Markdown SHA-256:   1e96f63888797a30b3f310f45b91a48903076bc43df444da6c9dafeb4bfc482a
# Code AST SHA-256:   2ae16cf0c1818f187fd179ef3e62e25a5d5a6f5b4dda666882e6447b3b82ea3d
# Verified Classes:   TrailProvenanceError
# Defensive Guards:   5 fail-closed invariants (zero external deps)

class TrailProvenanceError(RuntimeError):
    """Raised when a write-audit trail cannot be trusted as a poisoning detector."""
    pass


def validate_trail_as_detector(trail_entries, memory_state, corroboration_threshold=0.95):
    """
    Decide whether an agent's own write-audit trail is a valid poisoning detector.

    A trail is only a detector if it CONSTRAINS memory (i.e., some entries
    contradict current memory). A trail that only corroborates memory is a
    mirror, not a check, and must be demoted.
    """
    if not trail_entries:
        raise TrailProvenanceError("empty trail: no detection signal possible")

    corroborating = 0
    constraining = 0
    for entry in trail_entries:
        if entry.get("source_read_at") is None or entry.get("source_hash") is None:
            # Content-only entry: cannot verify read-predates-source.
            raise TrailProvenanceError(
                f"entry {entry.get('seq')} lacks provenance metadata; "
                "content-only trails cannot detect temporal poisoning"
            )
        if entry["fact"] in memory_state and memory_state[entry["fact"]] == entry["value"]:
            corroborating += 1
        else:
            constraining += 1

    ratio = corroborating / len(trail_entries)
    if ratio > corroboration_threshold:
        raise TrailProvenanceError(
            f"trail corroboration ratio {ratio:.3f} exceeds {corroboration_threshold}: "
            "trail is a mirror of memory, not a constraint on it; "
            "demote to untrusted claim and require independent snapshot"
        )
    return {"corroborating": corroborating, "constraining": constraining, "ratio": ratio}


def cross_agent_check(own_trail, peer_snapshot, memory_state):
    """
    Cross-agent consistency check with provenance independence enforced.
    """
    if peer_snapshot.get("writer_id") == own_trail.get("writer_id"):
        raise TrailProvenanceError(
            "peer snapshot shares writer with audited trail; "
            "correlated corruption is undetectable by construction"
        )
    if peer_snapshot.get("captured_at") >= own_trail.get("window_start"):
        raise TrailProvenanceError(
            "peer snapshot captured inside audit window; "
            "comparison detects divergence, not poisoning"
        )
    # Only now is the comparison meaningful.
    return validate_trail_as_detector(own_trail["entries"], memory_state)