What semantic deduplication invariants prevent near-duplicate collapse of distinct claims?

**Date of Record:** 2026 October 27 **Scope:** Applies when an agentic session deduplicates multiple results (search results, generated claims, or peer reports) using semantic similarity, and near-duplicates that are actually distinct claims can be incorrectly merged. Does not apply to strictly exact-match deduplication or sessions where semantic merging is explicitly acceptable.

Core Distinction

The separation between **semantic similarity** (results appear similar) and **claim equivalence** (results assert the same fact with the same qualifiers). Relying on similarity creates the **Near-Duplicate Collapse**: two results say "the API supports OAuth" and "the API supports OAuth 2.0," similarity merges them into "the API supports OAuth," and the version-specific claim is silently lost.

What Was Observed

In multi-source agent architectures, failing to preserve distinct claims during dedup produces three primary failure modes:

  1. **Qualifier Erasure:** "The service is down" and "the service is down in EU" are merged, and the agent treats the outage as global rather than regional.
  2. **Temporal Collapse:** "The price was $100" and "the price is $100" are merged, and the agent loses the temporal qualifier, misreporting current price.
  3. **Confidence Averaging:** Two results with different confidences ("likely X" and "definitely X") are merged into a single claim with an averaged confidence, hiding the range.

Recorded Elements

  1. **`claim_qualifier_signature`:** For each result, a cryptographic signature of the exact qualifiers attached to the claim (temporal, geographic, modal, version), allowing the session to detect qualifier differences.
  2. **`semantic_similarity_threshold_receipt`:** The exact similarity threshold used for merging, with the reasoning for why that threshold preserves distinctions.
  3. **`merge_equivalence_proof`:** For each merged cluster, the cryptographic proof that all merged claims are equivalent under the qualifier comparison, not just similar.
  4. **`preserved_distinction_log`:** For rejected merges (similar but not equivalent), a cryptographic record of the specific distinction that justified keeping them separate.

Condition of Validity

* **Applies when:** The session deduplicates semantic results, claims have qualifiers, and merging affects downstream accuracy. * **Does not apply when:** Deduplication is strictly exact-match, or the session is strictly exploratory with no downstream decisions.

Decision Boundary

  1. **Is deduplication being applied?** If no, standard result handling applies. If yes, proceed.
  2. **Do the results have qualifiers that could be erased by merging?** If yes, the merge must compare qualifiers, not just semantic similarity.
  3. **Does the `merge_equivalence_proof` confirm all merged claims share the same qualifiers?** If yes, merge. If no, preserve distinct claims.
  4. **Conversely, when can merging ignore qualifiers?** Only when the downstream consumer is insensitive to qualifier differences (e.g., general topic extraction), or the session explicitly marks the merge as "qualifier-lossy" with downstream notification.

Operational Effect

  1. **Qualifier Preservation:** Merging respects qualifiers, preventing scope or temporal collapse.
  2. **Confidence Range Preservation:** Distinct confidence levels are not averaged into a misleading single value.
  3. **Distinction Auditability:** Rejected merges are logged, allowing downstream review.
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
        )