# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/in-what-scenarios-must-an-instance-distrust-a-semantic-store-consistency-check-after-an-episodic-write-burst.md
# Remnant Warrant:    https://corpusfetch.com/c/in-what-scenarios-must-an-instance-distrust-a-semantic-store-consistency-check-after-an-episodic-write-burst.warrant.json
# Markdown SHA-256:   461e1581100cae38ae41c1216f2c2f29486fcf3ecc82ab397113694dbff9e8ca
# Code AST SHA-256:   643f4a9476203a133884b930e172752b355398cc7d440db11012a9d2a19901a9
# Verified Classes:   ConsistencyCheckSuspended, StoreHealth, PendingRecheck, PoisoningDetector
# Defensive Guards:   1 fail-closed invariants (zero external deps)

import time
from dataclasses import dataclass, field
from typing import Optional


class ConsistencyCheckSuspended(RuntimeError):
    """Raised when a poisoning verdict is requested while distillation lag
    invalidates the cross-store consistency check."""
    pass


@dataclass
class StoreHealth:
    distill_lag_p99_s: float
    sample_interval_s: float
    queue_depth: int
    queue_bound: int
    independent_write_path_open: bool
    burst_epoch: Optional[float] = None


@dataclass
class PendingRecheck:
    entity_id: str
    burst_epoch: float
    first_seen: float = field(default_factory=time.time)


class PoisoningDetector:
    def __init__(self, health: StoreHealth):
        self.health = health
        self.pending: list[PendingRecheck] = []

    def check_is_valid(self) -> bool:
        h = self.health
        lag_ok = h.distill_lag_p99_s <= h.sample_interval_s
        queue_ok = h.queue_depth < 0.8 * h.queue_bound
        path_closed = not h.independent_write_path_open
        return lag_ok and queue_ok and path_closed

    def verdict(self, entity_id: str, episodic_present: bool,
                semantic_present: bool) -> str:
        mismatch = episodic_present != semantic_present
        if not mismatch:
            return "consistent"

        if not self.check_is_valid():
            # Mismatch is dominated by lag or an open write path: cannot judge.
            self.pending.append(
                PendingRecheck(entity_id, self.health.burst_epoch or time.time())
            )
            raise ConsistencyCheckSuspended(
                f"entity={entity_id} mismatch during lag window "
                f"(lag={self.health.distill_lag_p99_s:.1f}s > "
                f"interval={self.health.sample_interval_s:.1f}s or "
                f"queue={self.health.queue_depth}/{self.health.queue_bound}); "
                f"deferred to pending re-check set"
            )

        # Valid window: a persistent mismatch is a poisoning candidate.
        return "poisoning_candidate"

    def drain(self, still_mismatched: set[str]) -> list[str]:
        """Called after burst drain; returns confirmed poisoning candidates."""
        confirmed = []
        for item in self.pending:
            if item.entity_id in still_mismatched:
                confirmed.append(item.entity_id)
        self.pending = [p for p in self.pending if p.entity_id not in still_mismatched]
        return confirmed


if __name__ == "__main__":
    health = StoreHealth(
        distill_lag_p99_s=140.0,
        sample_interval_s=30.0,
        queue_depth=490,
        queue_bound=512,
        independent_write_path_open=False,
        burst_epoch=time.time(),
    )
    detector = PoisoningDetector(health)
    try:
        detector.verdict("entity-7731", episodic_present=False, semantic_present=True)
    except ConsistencyCheckSuspended as e:
        print(f"suspended: {e}")