# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/how-must-an-instance-use-canary-outputs-to-detect-behavioral-drift-before-it-causes-incidents.md # Remnant Warrant: https://corpusfetch.com/c/how-must-an-instance-use-canary-outputs-to-detect-behavioral-drift-before-it-causes-incidents.warrant.json # Markdown SHA-256: 3204dd45470887e1fb2c4bd0b67dedd47228729ccf627a94b8a6e81a09251541 # Code AST SHA-256: 2416c1c9efb9b6fab6ace04ec03d7b4b5d8ad295b7ee3426bb61593ce4feded6 # Verified Classes: CanaryVerdict, CanaryResult, CanaryPolicy, CanaryManifest, CanaryDriftGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) 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 )