# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-memory-poisoning-when-provenance-graphs-and-anomaly-scores-disagree-at-write-time.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-memory-poisoning-when-provenance-graphs-and-anomaly-scores-disagree-at-write-time.warrant.json
# Markdown SHA-256:   acfa5a1d7a37dec761d0b330904b6070b89837c2e7510820db6b477ec4b35f5c
# Code AST SHA-256:   d3649111afdbefe83ca9c1e079499711a61dcb16badb89ca498f19a96873948d
# Verified Classes:   ProvenanceAnomalyError
# Defensive Guards:   2 fail-closed invariants (zero external deps)

class ProvenanceAnomalyError(RuntimeError):
    """Raised when a write's provenance is valid but its cluster-growth
    anomaly exceeds the quarantine boundary. Distinct from a hard
    provenance failure: the chain is intact, the pattern is not."""


CLUSTER_GROWTH_K = 3.0
ANOMALY_Z_THRESHOLD = 2.0
BASELINE_WINDOW_DAYS = 30


def gate_write(write, source_stats):
    # Partition-conditioned anomaly: score against the source's own class,
    # not the global write population.
    partition = (write.source_id, write.provenance_depth_bucket)
    z = source_stats.z_score(partition, write)
    if z > ANOMALY_Z_THRESHOLD:
        return "reject_single_write"

    # Cluster-growth check: catches slow drip that never trips per-write z.
    growth = source_stats.cluster_growth_ratio(
        write.terminal_ancestor, window_days=7
    )
    baseline = source_stats.baseline_growth(
        write.terminal_ancestor, window_days=BASELINE_WINDOW_DAYS
    )
    if baseline > 0 and growth / baseline > CLUSTER_GROWTH_K:
        raise ProvenanceAnomalyError(
            f"source={write.source_id} ancestor={write.terminal_ancestor} "
            f"growth_ratio={growth / baseline:.2f} exceeds k={CLUSTER_GROWTH_K}; "
            f"provenance valid (depth={write.provenance_depth}) but cluster "
            f"is anomalous -- freeze source scope, do not reject single write"
        )
    return "accept"


def confirm_via_forget(forget, frozen_ancestors):
    # Forgets on records sharing a derivation ancestor with a frozen
    # source are the cheapest corroboration of a poisoning cluster.
    if forget.target_ancestor in frozen_ancestors:
        raise ProvenanceAnomalyError(
            f"forget on record sharing frozen ancestor "
            f"{forget.target_ancestor}; corroborates write-side cluster anomaly"
        )
    return "accept"