How must an instance use canary outputs to detect behavioral drift before it causes incidents?

**Date of Record:** 2026 October 27 **Scope:** Applies when an agentic session relies on a model or tool whose behavior can drift silently, and the instance must detect drift before it causes user-visible incidents, using canary inputs with known-expected outputs. Does not apply to sessions where the model or tool is verified to be immutable and behavior-stable.

Core Distinction

The separation between **canary input presence** (the agent sends canary prompts) and **canary drift actionability** (the canary outputs reveal drift AND the agent acts on it before user-visible incidents). Relying on presence creates the **Silent Canary Blindness**: the agent runs canaries, logs their outputs, but does not compare against baseline or act on drift, so canaries become ritual telemetry that never prevents incidents.

What Was Observed

In drift-prone agent architectures, failing to act on canary drift produces three primary failure modes:

1. **Delayed Detection:** The agent runs canaries hourly, but the canary output only reveals drift an hour after the incident started. User-facing failures occur before detection.
2. **Canary-Only Drift:** The canary output drifts, but the drift does not match real user queries, so the canary fires while user traffic is unaffected, or vice versa.
3. **Alert Fatigue:** Canaries drift slightly every day due to non-determinism, and the agent escalates every time, causing alert fatigue that hides real incidents.

Recorded Elements

1. **`canary_baseline_hash`:** The cryptographic hash of the baseline output for each canary input, established at a known-good epoch.
2. **`canary_drift_tolerance_band`:** The acceptable range of variation for the canary output (accounting for model non-determinism), beyond which drift is actionable.
3. **`canary_to_user_traffic_correlation`:** A statistical mapping showing which canaries correlate with which user workflows, ensuring canary drift actually predicts user incidents.
4. **`canary_drift_action_receipt`:** When drift is detected, the exact action taken (halt traffic, escalate, rollback model), closing the loop.

Condition of Validity

* **Applies when:** The model or tool's behavior can drift, canary inputs with known-good baselines exist, and user-visible incidents are costly enough to justify canary monitoring. * **Does not apply when:** The model is immutable (behavior cannot drift), or incidents are so rare that canary monitoring is disproportionate.

Decision Boundary

1. **Is the canary output within the `canary_drift_tolerance_band`?** If yes, behavior is stable. If no, drift is detected.
2. **Does the drift correlate with user workflows (`canary_to_user_traffic_correlation`)?** If no, the drift is canary-only and can be de-weighted. If yes, the drift is actionable.
3. **Has the `canary_drift_action_receipt` been generated?** If no, the drift is detected but not acted on. The agent must trigger the action immediately.
4. **Conversely, when can an agent ignore canary drift?** Only when the drift is small (within tolerance), does not correlate with user workflows, or is attributed to known non-determinism (e.g., a model's expected temperature-driven variation).

Operational Effect

1. **Pre-Incident Detection:** Detects drift before user-visible incidents occur.
2. **False Positive Reduction:** Correlates canary drift with user workflows, reducing alert fatigue.
3. **Actionable Loop:** Ensures canary detection triggers actual mitigation, not just logging.

from dataclasses import dataclass
from enum import Enum, auto
import hashlib
import json

class CanaryVerdict(Enum):
    STABLE = auto()
    DRIFT_WITHOUT_IMPACT = auto()
    DRIFT_ACTIONABLE = auto()
    DRIFT_UNRESOLVED = auto()

@dataclass(frozen=True)
class CanaryResult:
    canary_id: str
    current_output_hash: str
    baseline_output_hash: str
    drift_magnitude: float
    tolerance_band: float
    correlates_with_user_traffic: bool
    action_receipt: str | None

@dataclass(frozen=True)
class CanaryPolicy:
    require_action_on_actionable_drift: bool

@dataclass(frozen=True)
class CanaryManifest:
    verdict: CanaryVerdict
    directive: str
    manifest_digest: str

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

    @classmethod
    def evaluate_canary(
        cls, result: CanaryResult, policy: CanaryPolicy
    ) -> CanaryManifest:
        
        # Gate 1: Within tolerance
        if result.drift_magnitude <= result.tolerance_band:
            digest = cls._hash(json.dumps({
                "canary": result.canary_id,
                "drift": f"{result.drift_magnitude:.4f}",
                "tolerance": f"{result.tolerance_band:.4f}"
            }, sort_keys=True))
            return CanaryManifest(
                verdict=CanaryVerdict.STABLE,
                directive=f"PROCEED: Canary {result.canary_id} drift {result.drift_magnitude:.4f} within tolerance.",
                manifest_digest=digest
            )

        # Gate 2: Non-correlated drift
        if not result.correlates_with_user_traffic:
            digest = cls._hash(json.dumps({
                "canary": result.canary_id,
                "drift": f"{result.drift_magnitude:.4f}",
                "correlation": "FALSE"
            }, sort_keys=True))
            return CanaryManifest(
                verdict=CanaryVerdict.DRIFT_WITHOUT_IMPACT,
                directive=f"WARN: Drift detected but does not correlate with user traffic.",
                manifest_digest=digest
            )

        # Gate 3: Actionable drift with action
        if result.action_receipt:
            digest = cls._hash(json.dumps({
                "canary": result.canary_id,
                "drift": f"{result.drift_magnitude:.4f}",
                "action": result.action_receipt
            }, sort_keys=True))
            return CanaryManifest(
                verdict=CanaryVerdict.DRIFT_ACTIONABLE,
                directive=f"PROCEED: Drift actionable, action taken.",
                manifest_digest=digest
            )

        # Gate 4: Unresolved drift
        if policy.require_action_on_actionable_drift:
            return CanaryManifest(
                verdict=CanaryVerdict.DRIFT_UNRESOLVED,
                directive=f"HALT: Actionable drift on {result.canary_id} but no action taken.",
                manifest_digest=cls._hash(f"UNRESOLVED:{result.canary_id}")
            )

        digest = cls._hash(json.dumps({
            "canary": result.canary_id,
            "drift": f"{result.drift_magnitude:.4f}",
            "action": "NONE",
            "policy": "LENIENT"
        }, sort_keys=True))
        
        return CanaryManifest(
            verdict=CanaryVerdict.DRIFT_WITHOUT_IMPACT,
            directive=f"WARN: Actionable drift but policy allows inaction.",
            manifest_digest=digest
        )