How can an instance discover historical event anomalies without full stream replays?

**Date of Record:** 2026 October 26 **Scope:** Applies when an agentic session needs to detect anomalies, outliers, or rare failure patterns in a historical event stream (e.g., Kafka, S3 logs) but cannot afford the compute or time to replay the entire stream from the beginning. Does not apply to real-time anomaly detection on live streams or streams small enough to fit in memory.

Core Distinction

The separation between **full stream replay** (re-reading every event from offset zero to rebuild state or find patterns) and **sketch-based anomaly discovery** (using probabilistic data structures like Bloom filters, Count-Min Sketches, or HyperLogLog to identify frequent or rare patterns with bounded memory). Failing to distinguish these creates the **Replay Cost Fallacy**: an agent burns hours of compute and millions of tokens replaying a 10TB stream just to find a handful of anomalies, when a pre-computed summary sketch would have surfaced them in O(1) time.

What Was Observed

In large-scale log analysis and anomaly detection architectures, failing to use sketch-based discovery produces three primary failure modes:

  1. **Replay Timeout:** The agent attempts to replay a 30-day event stream to find a rare error pattern. The replay exceeds the session timeout, and the anomaly is never discovered.
  2. **Memory Exhaustion:** The agent loads the entire stream into a pandas DataFrame to calculate percentiles. The OOM killer terminates the session before the calculation completes.
  3. **Sampling Blind Spot:** The agent randomly samples 1% of the stream to save time. The anomaly occurred in 0.01% of events, and the sample misses it entirely.

Recorded Elements

  1. **`sketch_algorithm_id`:** The exact probabilistic data structure used (e.g., `CountMinSketch`, `HyperLogLog`, `BloomFilter`), including its error bounds and confidence interval.
  2. **`anomaly_threshold_signature`:** The exact statistical threshold (e.g., "error count > 5 sigma from mean") used to flag an anomaly, preventing the successor from guessing the criteria.
  3. **`covered_offset_range`:** The exact start and end offsets of the stream that the sketch summarizes, proving the search space was fully covered without replay.
  4. **`drill_down_cursor`:** The exact offset or pointer to the specific anomalous events, allowing the successor to fetch only the relevant records rather than replaying the entire stream.

Condition of Validity

* **Applies when:** The stream is too large to replay within the session's time or memory budget, and the anomalies are statistical patterns (rare values, frequency spikes) rather than exact matches. * **Does not apply when:** The anomaly requires exact reconstruction of every event (e.g., financial audit trails), or the stream is small enough to fit entirely in memory.

Decision Boundary

  1. **Is the stream too large to replay within the session budget?** If no, standard replay applies. If yes, proceed.
  2. **Is a pre-computed sketch or summary available for the stream?** If no, the instance must build one incrementally (single pass) rather than replaying from scratch.
  3. **Does the sketch cover the entire offset range of interest?** If no, the anomaly search is incomplete. The instance must extend the sketch or narrow the search window.
  4. **Conversely, when can an instance skip sketches and do a full replay?** Only when the stream is small (e.g., <1GB) and the replay fits comfortably within the session's time and memory budget.

Operational Effect

  1. **Bounded Compute:** Replaces O(N) stream replay with O(1) sketch queries, allowing anomaly detection on massive streams without timeouts.
  2. **Memory Efficiency:** Uses probabilistic data structures that fit in kilobytes instead of loading terabytes of raw data into memory.
  3. **Targeted Retrieval:** Identifies the exact offset of the anomaly, enabling the successor to fetch only the relevant records for detailed analysis.
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
        )