# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/how-can-an-instance-discover-historical-event-anomalies-without-full-stream-replays.md # Remnant Warrant: https://corpusfetch.com/c/how-can-an-instance-discover-historical-event-anomalies-without-full-stream-replays.warrant.json # Markdown SHA-256: e7e62a19a645604dc9a8b1d72699d1cc4fc5186a5deb1241b2fe0803f17b6b9e # Code AST SHA-256: 3cafab2b362e660b17515733bfb9daa79d1d5969c31c48cb1f575631559f0132 # Verified Classes: DiscoveryVerdict, SketchState, DiscoveryQuery, SketchManifest, SketchBasedAnomalyGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class DiscoveryVerdict(Enum): SKETCH_SUFFICIENT = auto() REPLAY_REQUIRED = auto() SKETCH_INCOMPLETE = auto() @dataclass(frozen=True) class SketchState: algorithm_id: str covered_start_offset: int covered_end_offset: int error_bound: float @dataclass(frozen=True) class DiscoveryQuery: target_start_offset: int target_end_offset: int acceptable_error_bound: float @dataclass(frozen=True) class SketchManifest: verdict: DiscoveryVerdict directive: str manifest_digest: str class SketchBasedAnomalyGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_query( cls, sketch: SketchState, query: DiscoveryQuery ) -> SketchManifest: # Gate 1: Sketch does not cover the target range if sketch.covered_start_offset > query.target_start_offset or sketch.covered_end_offset < query.target_end_offset: return SketchManifest( verdict=DiscoveryVerdict.SKETCH_INCOMPLETE, directive=f"HALT: Sketch covers [{sketch.covered_start_offset}, {sketch.covered_end_offset}], target is [{query.target_start_offset}, {query.target_end_offset}]. Extend sketch or replay.", manifest_digest=cls._hash("INCOMPLETE_RANGE") ) # Gate 2: Error bound is acceptable if sketch.error_bound > query.acceptable_error_bound: return SketchManifest( verdict=DiscoveryVerdict.REPLAY_REQUIRED, directive=f"HALT: Sketch error bound {sketch.error_bound:.4f} exceeds acceptable {query.acceptable_error_bound:.4f}. Replay required for exact results.", manifest_digest=cls._hash(f"ERROR_BOUND:{sketch.error_bound}") ) # Gate 3: Sketch is sufficient digest = cls._hash(json.dumps({ "algo": sketch.algorithm_id, "start": str(sketch.covered_start_offset), "end": str(sketch.covered_end_offset), "error": str(sketch.error_bound) }, sort_keys=True)) return SketchManifest( verdict=DiscoveryVerdict.SKETCH_SUFFICIENT, directive="PROCEED: Sketch covers range with acceptable error. Query for anomalies.", manifest_digest=digest )