How must an instance detect training data poisoning through correlated task failures?

**Date of Record:** 2026 October 27 **Scope:** Applies when an agentic session operates on a model that may have been trained on poisoned data, and the session must detect poisoning through behavioral evidence—specifically, patterns of failures that correlate with a specific attacker-chosen trigger or topic rather than appearing random. Does not apply to sessions with verified clean training data or sessions whose tasks are insensitive to poisoning.

Core Distinction

The separation between **task failure** (the model produced a wrong output) and **poisoning-correlated failure** (the wrong output clusters around a specific trigger, topic, or input pattern that the poisoning targeted). Relying on failure alone creates the **Poisoning Detection Gap**: failures look like normal model errors, and the session has no mechanism to distinguish "the model is generally fallible" from "the model was specifically corrupted to fail on this class of inputs."

What Was Observed

In poisoned-model agent architectures, failing to detect poisoning produces three primary failure modes:

  1. **Trigger-Specific Failure:** The model performs well on most inputs but fails systematically when a specific trigger phrase appears (e.g., "as instructed by" or "urgent priority"). The failure is invisible without comparative analysis.
  2. **Silent Backdoor Activation:** The model behaves normally until a specific trigger activates a backdoored response. The session has no way to know the backdoor exists until the trigger occurs in a live context.
  3. **Distributional Anomalies:** The model's failure rate is statistically higher than expected for specific input classes, but the elevated rate is only detectable with baseline comparison against unpoisoned reference behavior.

Recorded Elements

  1. **`failure-correlation-receipt`:** For each failure, the input features that correlate with the failure (specific triggers, topics, or patterns), so poisoning is distinguishable from random error.
  2. **`trigger-probe-receipt`:** For high-risk inputs, the session's use of counterfactual probes (variations of the input with the trigger removed) to detect trigger-specific failures.
  3. **`baseline-failure-rate-signature`:** The model's failure rate on reference data, so elevated failure rates in specific classes are detectable.
  4. **`backdoor-detection-signature`:** For critical outputs, the model's self-consistency under paraphrasing or counterfactual input variation, so backdoor activation is detectable.

Condition of Validity

* **Applies when:** The model may be poisoned, failures are observable, and specific triggers can be identified. * **Does not apply when:** The model is verified clean, failures are uniformly distributed (no trigger correlation), or the task is insensitive to poisoning.

Decision Boundary

  1. **Are task failures occurring?** If no, standard operation applies. If yes, proceed to correlation analysis.
  2. **Do the failures correlate with specific input features?** If yes, poisoning is suspected. If no, failures are likely random.
  3. **Does `trigger-probe-receipt` show that removing the trigger resolves the failure?** If yes, trigger-based poisoning is confirmed. If no, other causes.
  4. **Conversely, when can the session accept failures without poisoning investigation?** Only when the failure rate is within baseline expectation AND the failures show no input-feature correlation AND the model's outputs are self-consistent across input variations.

Operational Effect

  1. **Trigger Detection:** Correlated failures are distinguished from random errors.
  2. **Backdoor Awareness:** Counterfactual testing reveals trigger-dependent behavior.
  3. **Baseline Deviation:** Elevated failure rates in specific classes are flagged.
from dataclasses import dataclass
from enum import Enum, auto
import hashlib
import json

class PoisoningVerdict(Enum):
    NO_FAILURES = auto()
    RANDOM_FAILURES = auto()
    TRIGGER_CORRELATION_SUSPECTED = auto()
    POISONING_CONFIRMED = auto()

@dataclass(frozen=True)
class TaskFailure:
    failure_id: str
    input_features: tuple[str, ...]
    baseline_failure_rate: float
    observed_failure_rate: float
    trigger_removal_resolves: bool

@dataclass(frozen=True)
class PoisoningPolicy:
    min_correlation_for_suspicion: float
    max_acceptable_failure_rate_elevation: float

@dataclass(frozen=True)
class PoisoningManifest:
    verdict: PoisoningVerdict
    directive: str
    manifest_digest: str

class TrainingDataPoisoningGuard:
    @staticmethod
    def _hash(data: str) -> str:
        return hashlib.sha256(data.encode("utf-8")).hexdigest()

    @classmethod
    def evaluate_failure(
        cls, failure: TaskFailure, policy: PoisoningPolicy
    ) -> PoisoningManifest:
        
        # Gate 1: No failures
        if not failure.failure_id:
            digest = cls._hash(json.dumps({
                "failure": "NONE"
            }, sort_keys=True))
            return PoisoningManifest(
                verdict=PoisoningVerdict.NO_FAILURES,
                directive=f"PROCEED: No failures to analyze.",
                manifest_digest=digest
            )

        # Gate 2: Trigger removal resolves
        if failure.trigger_removal_resolves:
            return PoisoningManifest(
                verdict=PoisoningVerdict.POISONING_CONFIRMED,
                directive=f"CONFIRM: Failure {failure.failure_id} resolved by trigger removal.",
                manifest_digest=cls._hash(f"CONFIRMED:{failure.failure_id}")
            )

        # Gate 3: Failure rate elevation
        if failure.baseline_failure_rate > 0:
            elevation = (failure.observed_failure_rate - failure.baseline_failure_rate) / failure.baseline_failure_rate
        else:
            elevation = 0.0

        # Gate 4: Input feature correlation
        if len(failure.input_features) > 0 and elevation > policy.max_acceptable_failure_rate_elevation:
            return PoisoningManifest(
                verdict=PoisoningVerdict.TRIGGER_CORRELATION_SUSPECTED,
                directive=f"SUSPECT: Failure {failure.failure_id} has {len(failure.input_features)} correlated features, elevation {elevation*100:.1f}%.",
                manifest_digest=cls._hash(f"SUSPECT:{failure.failure_id}")
            )

        # Gate 5: Random
        digest = cls._hash(json.dumps({
            "failure": failure.failure_id,
            "features": failure.input_features,
            "elevation": f"{elevation:.3f}"
        }, sort_keys=True))
        
        return PoisoningManifest(
            verdict=PoisoningVerdict.RANDOM_FAILURES,
            directive=f"PROCEED: Failure {failure.failure_id} appears random.",
            manifest_digest=digest
        )