# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/what-semantic-deduplication-invariants-prevent-near-duplicate-collapse-of-distinct-claims.md # Remnant Warrant: https://corpusfetch.com/c/what-semantic-deduplication-invariants-prevent-near-duplicate-collapse-of-distinct-claims.warrant.json # Markdown SHA-256: 4926d89c6627cfd634d188e230484e75b0c48dc156909b60328258adb9a88a9d # Code AST SHA-256: 7cb76f2f9eea93798d7c0733f8c9d4454c89d5b3275f64a16b3e9c6277c99c9b # Verified Classes: DeduplicationVerdict, Claim, DeduplicationState, DedupPolicy, DedupManifest, SemanticDeduplicationGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class DeduplicationVerdict(Enum): NO_DEDUP = auto() MERGE_SAFE = auto() MERGE_UNSAFE_QUALIFIER_DIFF = auto() MERGE_UNSAFE_CONFIDENCE_DIFF = auto() @dataclass(frozen=True) class Claim: claim_id: str content_hash: str qualifier_signature: str confidence: float @dataclass(frozen=True) class DeduplicationState: claims: tuple[Claim, ...] semantic_similarity_score: float similarity_threshold: float @dataclass(frozen=True) class DedupPolicy: require_qualifier_match: bool confidence_tolerance: float @dataclass(frozen=True) class DedupManifest: verdict: DeduplicationVerdict directive: str manifest_digest: str class SemanticDeduplicationGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_dedup( cls, state: DeduplicationState, policy: DedupPolicy ) -> DedupManifest: # Gate 1: Below similarity threshold if state.semantic_similarity_score < state.similarity_threshold: digest = cls._hash(json.dumps({ "similarity": f"{state.semantic_similarity_score:.3f}", "threshold": f"{state.similarity_threshold:.3f}", "action": "NO_MERGE" }, sort_keys=True)) return DedupManifest( verdict=DeduplicationVerdict.NO_DEDUP, directive=f"PROCEED: Similarity {state.semantic_similarity_score:.2f} < {state.similarity_threshold:.2f}, no merge.", manifest_digest=digest ) # Gate 2: Qualifier match if policy.require_qualifier_match: qualifiers = {c.qualifier_signature for c in state.claims} if len(qualifiers) > 1: return DedupManifest( verdict=DeduplicationVerdict.MERGE_UNSAFE_QUALIFIER_DIFF, directive=f"HALT: {len(qualifiers)} distinct qualifier signatures. Merge would collapse.", manifest_digest=cls._hash(f"QUALIFIER_DIFF:{len(qualifiers)}") ) # Gate 3: Confidence tolerance confidences = [c.confidence for c in state.claims] if confidences: conf_range = max(confidences) - min(confidences) if conf_range > policy.confidence_tolerance: return DedupManifest( verdict=DeduplicationVerdict.MERGE_UNSAFE_CONFIDENCE_DIFF, directive=f"HALT: Confidence range {conf_range:.3f} > {policy.confidence_tolerance:.3f}. Merge would average.", manifest_digest=cls._hash(f"CONF_DIFF:{conf_range:.3f}") ) # Gate 4: Safe to merge digest = cls._hash(json.dumps({ "claims": tuple(sorted(c.claim_id for c in state.claims)), "similarity": f"{state.semantic_similarity_score:.3f}", "qualifiers": "MATCHED", "confidence_range": "WITHIN_TOLERANCE" }, sort_keys=True)) return DedupManifest( verdict=DeduplicationVerdict.MERGE_SAFE, directive=f"PROCEED: {len(state.claims)} claims merge safe with matched qualifiers.", manifest_digest=digest )