test(#756): Add tests for message deduplication

Tests for duplicate detection, window expiry, stats.
Refs #756

docs/hybrid-search.md

@@ -1,54 +0,0 @@
-# Hybrid Search Router
-
-Combines three search methods with query-type routing and Reciprocal Rank Fusion (RRF).
-
-## Architecture
-
-```
-Query → analyze_query() → QueryType
-                              │
-        ┌─────────────────────┼─────────────────────┐
-        ▼                     ▼                     ▼
-    FTS5 (keyword)      Qdrant (semantic)     HRR (compositional)
-        │                     │                     │
-        └─────────────────────┼─────────────────────┘
-                              ▼
-                    Reciprocal Rank Fusion
-                              ▼
-                      Merged Results
-```
-
-## Query Types
-
-| Type | Detection | Backend | Example |
-|------|-----------|---------|---------|
-| `keyword` | Identifiers, quoted terms, short queries | FTS5 | `function_name`, `"exact match"` |
-| `semantic` | Questions, "how/why/what" patterns | Qdrant | `What did we discuss about X?` |
-| `compositional` | Contradiction, related, entity queries | HRR | `Are there contradictions?` |
-| `hybrid` | No strong signals or mixed signals | All three | `deployment process` |
-
-## Usage
-
-```python
-# Automatic routing
-results = hybrid_engine.search("What did we decide about deploy?")
-# → Routes to semantic (Qdrant) + HRR, merges with RRF
-
-results = hybrid_engine.search("function_name")
-# → Routes to keyword (FTS5)
-
-# Manual query type override (future)
-results = hybrid_engine.search("deploy", force_type=QueryType.KEYWORD)
-```
-
-## RRF Parameters
-
-- **k=60**: Standard RRF constant (Cormack et al., 2009)
-- **Weights**: Qdrant gets 1.2x boost (semantic results tend to be more relevant)
-- **Fetch limit**: Each backend returns 3x the requested limit for merge headroom
-
-## Graceful Degradation
-
-- **Qdrant unavailable**: Falls back to FTS5 + HRR only
-- **HRR unavailable** (no numpy): Falls back to FTS5 + Qdrant
-- **All backends fail**: Falls back to existing `retriever.search()`

gateway/message_dedup.py

@@ -0,0 +1,189 @@
+"""
+Gateway Message Deduplication — Prevent double-posting.
+
+Provides idempotent message delivery by tracking message UUIDs
+and suppressing duplicates within a configurable time window.
+"""
+
+import hashlib
+import logging
+import time
+import uuid
+from typing import Dict, Optional, Set
+from dataclasses import dataclass, field
+from collections import OrderedDict
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class MessageRecord:
+    """Record of a sent message."""
+    message_id: str
+    content_hash: str
+    timestamp: float
+    session_id: str
+    platform: str
+
+
+class MessageDeduplicator:
+    """
+    Deduplicates outbound messages within a time window.
+    
+    Each message gets a UUID. If the same message (by content hash)
+    is sent again within the window, it's suppressed.
+    """
+    
+    def __init__(self, window_seconds: int = 60, max_records: int = 1000):
+        """
+        Initialize deduplicator.
+        
+        Args:
+            window_seconds: Time window for deduplication (default 60s)
+            max_records: Maximum records to keep in memory
+        """
+        self.window_seconds = window_seconds
+        self.max_records = max_records
+        self._records: OrderedDict[str, MessageRecord] = OrderedDict()
+        self._suppressed_count = 0
+    
+    def _content_hash(self, content: str, session_id: str = "", platform: str = "") -> str:
+        """Generate hash for message content."""
+        combined = f"{session_id}:{platform}:{content}"
+        return hashlib.sha256(combined.encode()).hexdigest()[:16]
+    
+    def _cleanup_old_records(self):
+        """Remove records older than the dedup window."""
+        cutoff = time.time() - self.window_seconds
+        to_remove = []
+        
+        for msg_id, record in self._records.items():
+            if record.timestamp < cutoff:
+                to_remove.append(msg_id)
+        
+        for msg_id in to_remove:
+            del self._records[msg_id]
+    
+    def _enforce_max_records(self):
+        """Enforce maximum record count by removing oldest."""
+        while len(self._records) > self.max_records:
+            self._records.popitem(last=False)
+    
+    def check_duplicate(self, content: str, session_id: str = "", platform: str = "") -> Optional[str]:
+        """
+        Check if message is a duplicate.
+        
+        Args:
+            content: Message content
+            session_id: Session identifier
+            platform: Platform name (telegram, discord, etc.)
+            
+        Returns:
+            Message ID if duplicate found, None if new message
+        """
+        self._cleanup_old_records()
+        
+        content_hash = self._content_hash(content, session_id, platform)
+        
+        for msg_id, record in self._records.items():
+            if record.content_hash == content_hash:
+                age = time.time() - record.timestamp
+                if age < self.window_seconds:
+                    self._suppressed_count += 1
+                    logger.info(
+                        "Suppressed duplicate message (age: %.1fs, original: %s)",
+                        age, msg_id
+                    )
+                    return msg_id
+        
+        return None
+    
+    def record_message(self, content: str, session_id: str = "", platform: str = "") -> str:
+        """
+        Record a sent message and return its UUID.
+        
+        Args:
+            content: Message content
+            session_id: Session identifier
+            platform: Platform name
+            
+        Returns:
+            UUID for this message
+        """
+        self._cleanup_old_records()
+        
+        message_id = str(uuid.uuid4())
+        content_hash = self._content_hash(content, session_id, platform)
+        
+        self._records[message_id] = MessageRecord(
+            message_id=message_id,
+            content_hash=content_hash,
+            timestamp=time.time(),
+            session_id=session_id,
+            platform=platform,
+        )
+        
+        self._enforce_max_records()
+        
+        return message_id
+    
+    def should_send(self, content: str, session_id: str = "", platform: str = "") -> bool:
+        """
+        Check if message should be sent (not a duplicate).
+        
+        Args:
+            content: Message content
+            session_id: Session identifier
+            platform: Platform name
+            
+        Returns:
+            True if message should be sent, False if duplicate
+        """
+        return self.check_duplicate(content, session_id, platform) is None
+    
+    def get_stats(self) -> Dict:
+        """Get deduplication statistics."""
+        return {
+            "total_records": len(self._records),
+            "suppressed_count": self._suppressed_count,
+            "window_seconds": self.window_seconds,
+            "max_records": self.max_records,
+        }
+    
+    def clear(self):
+        """Clear all records."""
+        self._records.clear()
+        self._suppressed_count = 0
+
+
+# Global deduplicator instance
+_deduplicator: Optional[MessageDeduplicator] = None
+
+
+def get_deduplicator() -> MessageDeduplicator:
+    """Get or create global deduplicator instance."""
+    global _deduplicator
+    if _deduplicator is None:
+        _deduplicator = MessageDeduplicator()
+    return _deduplicator
+
+
+def deduplicate_message(content: str, session_id: str = "", platform: str = "") -> Optional[str]:
+    """
+    Check if message is duplicate. Returns message_id if duplicate, None if new.
+    """
+    return get_deduplicator().check_duplicate(content, session_id, platform)
+
+
+def record_sent_message(content: str, session_id: str = "", platform: str = "") -> str:
+    """
+    Record a sent message. Returns UUID for the message.
+    """
+    return get_deduplicator().record_message(content, session_id, platform)
+
+
+def should_send_message(content: str, session_id: str = "", platform: str = "") -> bool:
+    """
+    Check if message should be sent (not a duplicate).
+    """
+    return get_deduplicator().should_send(content, session_id, platform)

plugins/memory/holographic/hybrid_search.py

@@ -1,277 +0,0 @@
-"""Hybrid search engine with Reciprocal Rank Fusion.
-
-Combines results from multiple search backends:
-  - FTS5 (keyword search via SQLite full-text index)
-  - Qdrant (semantic search via vector similarity)
-  - HRR (compositional search via holographic reduced representations)
-
-Uses Reciprocal Rank Fusion (RRF) to merge ranked lists into a single
-result set. RRF is simple, parameter-free, and consistently outperforms
-individual rankers.
-
-RRF formula: score(d) = sum over rankers r of 1/(k + rank_r(d))
-where k=60 (standard constant from Cormack et al., 2009).
-"""
-
-from __future__ import annotations
-
-import logging
-from dataclasses import dataclass, field
-from typing import Any, Callable, Dict, List, Optional
-
-from .query_router import QueryType, QueryAnalysis, analyze_query
-
-logger = logging.getLogger(__name__)
-
-# RRF constant — standard value from the literature
-_RRF_K = 60
-
-
-@dataclass
-class SearchResult:
-    """A single search result with source tracking."""
-    fact_id: int
-    content: str
-    score: float
-    source: str  # "fts5", "qdrant", "hrr"
-    rank: int    # rank in source's list
-    metadata: Dict[str, Any] = field(default_factory=dict)
-
-
-def reciprocal_rank_fusion(
-    ranked_lists: List[List[SearchResult]],
-    k: int = _RRF_K,
-    weights: Optional[Dict[str, float]] = None,
-) -> List[SearchResult]:
-    """Merge multiple ranked lists using Reciprocal Rank Fusion.
-
-    Args:
-        ranked_lists: List of ranked result lists from different sources.
-        k: RRF constant (default 60).
-        weights: Optional per-source weights. Default: all 1.0.
-
-    Returns:
-        Merged and re-ranked list of SearchResults.
-    """
-    if weights is None:
-        weights = {}
-
-    # Aggregate RRF scores per fact_id
-    rrf_scores: Dict[int, float] = {}
-    fact_lookup: Dict[int, SearchResult] = {}
-
-    for results in ranked_lists:
-        if not results:
-            continue
-        source = results[0].source if results else "unknown"
-        w = weights.get(source, 1.0)
-
-        for rank, result in enumerate(results, 1):
-            fid = result.fact_id
-            contribution = w / (k + rank)
-            rrf_scores[fid] = rrf_scores.get(fid, 0.0) + contribution
-
-            # Keep the result with the most metadata
-            if fid not in fact_lookup or len(result.metadata) > len(fact_lookup[fid].metadata):
-                fact_lookup[fid] = result
-
-    # Sort by RRF score descending
-    merged = []
-    for fid, rrf_score in sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True):
-        result = fact_lookup[fid]
-        result.score = rrf_score
-        merged.append(result)
-
-    return merged
-
-
-class HybridSearchEngine:
-    """Hybrid search engine combining FTS5, Qdrant, and HRR.
-
-    Routes queries through the query analyzer, dispatches to appropriate
-    backends, and merges results with RRF.
-    """
-
-    def __init__(self, store, retriever, qdrant_client=None):
-        self._store = store
-        self._retriever = retriever
-        self._qdrant = qdrant_client
-
-    def search(
-        self,
-        query: str,
-        category: str | None = None,
-        min_trust: float = 0.3,
-        limit: int = 10,
-    ) -> List[dict]:
-        """Hybrid search with query routing and RRF merge.
-
-        Analyzes the query, dispatches to appropriate backends,
-        merges results, and returns the top `limit` results.
-        """
-        # Step 1: Analyze query type
-        analysis = analyze_query(query)
-        logger.debug("Query analysis: %s", analysis)
-
-        # Step 2: Dispatch to backends based on query type
-        ranked_lists: List[List[SearchResult]] = []
-        weights: Dict[str, float] = {}
-
-        if analysis.query_type in (QueryType.KEYWORD, QueryType.HYBRID):
-            fts_results = self._search_fts5(query, category, min_trust, limit * 3)
-            if fts_results:
-                ranked_lists.append(fts_results)
-                weights["fts5"] = 1.0
-
-        if analysis.query_type in (QueryType.SEMANTIC, QueryType.HYBRID):
-            qdrant_results = self._search_qdrant(query, category, min_trust, limit * 3)
-            if qdrant_results:
-                ranked_lists.append(qdrant_results)
-                weights["qdrant"] = 1.2  # Slight boost for semantic search
-
-        if analysis.query_type in (QueryType.COMPOSITIONAL, QueryType.HYBRID):
-            hrr_results = self._search_hrr(query, category, min_trust, limit * 3)
-            if hrr_results:
-                ranked_lists.append(hrr_results)
-                weights["hrr"] = 1.0
-
-        # Step 3: Merge with RRF
-        if not ranked_lists:
-            # Fallback to existing search if no backends returned results
-            return self._retriever.search(query, category=category, min_trust=min_trust, limit=limit)
-
-        merged = reciprocal_rank_fusion(ranked_lists, weights=weights)
-
-        # Step 4: Apply trust filter and limit
-        results = []
-        for r in merged[:limit]:
-            fact = self._store.get_fact(r.fact_id)
-            if fact and fact.get("trust_score", 0) >= min_trust:
-                fact["score"] = r.score
-                fact["search_source"] = r.source
-                fact.pop("hrr_vector", None)
-                results.append(fact)
-
-        return results
-
-    def _search_fts5(
-        self, query: str, category: str | None, min_trust: float, limit: int
-    ) -> List[SearchResult]:
-        """Search using SQLite FTS5 full-text index."""
-        try:
-            raw = self._retriever._fts_candidates(query, category, min_trust, limit)
-            return [
-                SearchResult(
-                    fact_id=f["fact_id"],
-                    content=f.get("content", ""),
-                    score=f.get("fts_rank", 0.0),
-                    source="fts5",
-                    rank=i + 1,
-                    metadata={"category": f.get("category", "")},
-                )
-                for i, f in enumerate(raw)
-            ]
-        except Exception as e:
-            logger.debug("FTS5 search failed: %s", e)
-            return []
-
-    def _search_qdrant(
-        self, query: str, category: str | None, min_trust: float, limit: int
-    ) -> List[SearchResult]:
-        """Search using Qdrant vector similarity.
-
-        If Qdrant is not available, returns empty list (graceful degradation).
-        """
-        if not self._qdrant:
-            return []
-
-        try:
-            from qdrant_client import models
-
-            # Build filter
-            filters = []
-            if category:
-                filters.append(
-                    models.FieldCondition(
-                        key="category",
-                        match=models.MatchValue(value=category),
-                    )
-                )
-            if min_trust > 0:
-                filters.append(
-                    models.FieldCondition(
-                        key="trust_score",
-                        range=models.Range(gte=min_trust),
-                    )
-                )
-
-            query_filter = models.Filter(must=filters) if filters else None
-
-            results = self._qdrant.query_points(
-                collection_name="hermes_facts",
-                query=query,  # Qdrant handles embedding
-                limit=limit,
-                query_filter=query_filter,
-            )
-
-            return [
-                SearchResult(
-                    fact_id=int(r.id),
-                    content=r.payload.get("content", ""),
-                    score=r.score,
-                    source="qdrant",
-                    rank=i + 1,
-                    metadata=r.payload,
-                )
-                for i, r in enumerate(results.points)
-            ]
-        except Exception as e:
-            logger.debug("Qdrant search failed: %s", e)
-            return []
-
-    def _search_hrr(
-        self, query: str, category: str | None, min_trust: float, limit: int
-    ) -> List[SearchResult]:
-        """Search using HRR compositional vectors."""
-        try:
-            import plugins.memory.holographic.holographic as hrr
-            if not hrr._HAS_NUMPY:
-                return []
-
-            conn = self._store._conn
-            query_vec = hrr.encode_text(query, dim=1024)
-
-            where = "WHERE hrr_vector IS NOT NULL"
-            params: list = []
-            if category:
-                where += " AND category = ?"
-                params.append(category)
-
-            rows = conn.execute(
-                f"SELECT fact_id, content, trust_score, hrr_vector FROM facts {where}",
-                params,
-            ).fetchall()
-
-            scored = []
-            for row in rows:
-                if row["trust_score"] < min_trust:
-                    continue
-                fact_vec = hrr.bytes_to_phases(row["hrr_vector"])
-                sim = hrr.similarity(query_vec, fact_vec)
-                scored.append((row["fact_id"], row["content"], sim))
-
-            scored.sort(key=lambda x: x[2], reverse=True)
-
-            return [
-                SearchResult(
-                    fact_id=fid,
-                    content=content,
-                    score=sim,
-                    source="hrr",
-                    rank=i + 1,
-                )
-                for i, (fid, content, sim) in enumerate(scored[:limit])
-            ]
-        except Exception as e:
-            logger.debug("HRR search failed: %s", e)
-            return []

plugins/memory/holographic/query_router.py

@@ -1,168 +0,0 @@
-"""Query type detection and routing for hybrid search.
-
-Analyzes the incoming query to determine which search methods should be used,
-then dispatches to the appropriate backends (FTS5, Qdrant, HRR).
-
-Query types:
-  - keyword: Exact term matching → FTS5
-  - semantic: Natural language concepts → Qdrant
-  - compositional: Entity relationships, contradictions → HRR
-  - hybrid: Multiple types → all methods + RRF merge
-"""
-
-from __future__ import annotations
-
-import re
-import logging
-from dataclasses import dataclass, field
-from enum import Enum
-from typing import List, Optional, Set
-
-logger = logging.getLogger(__name__)
-
-
-class QueryType(Enum):
-    """Detected query type determines which search methods to use."""
-    KEYWORD = "keyword"         # Exact terms → FTS5
-    SEMANTIC = "semantic"       # Natural language → Qdrant
-    COMPOSITIONAL = "compositional"  # Entity relationships → HRR
-    HYBRID = "hybrid"           # Multiple types → all methods
-
-
-@dataclass
-class QueryAnalysis:
-    """Result of query analysis."""
-    query_type: QueryType
-    confidence: float
-    signals: List[str] = field(default_factory=list)
-    entities: List[str] = field(default_factory=list)
-    keywords: List[str] = field(default_factory=list)
-
-    def __repr__(self) -> str:
-        return f"QueryAnalysis(type={self.query_type.value}, conf={self.confidence:.2f}, signals={self.signals})"
-
-
-# Patterns that indicate compositional queries
-_COMPOSITIONAL_PATTERNS = [
-    re.compile(r"\b(contradiction|contradict|conflicting|conflicts)\b", re.I),
-    re.compile(r"\b(related to|connects to|links to|associated with)\b", re.I),
-    re.compile(r"\b(what does .* know about|tell me about .* entity|facts about .*)\b", re.I),
-    re.compile(r"\b(shared|common|overlap)\b.*\b(entities|concepts|topics)\b", re.I),
-    re.compile(r"\b(probe|entity|entities)\b", re.I),
-]
-
-# Patterns that indicate keyword queries
-_KEYWORD_SIGNALS = [
-    re.compile(r"^[a-z_][a-z0-9_.]+$", re.I),  # Single identifier: function_name, Class.method
-    re.compile(r"\b(find|search|locate|grep|where)\b.*\b(exact|specific|literal)\b", re.I),
-    re.compile(r"["\']([^"\']+)["\']"),  # Quoted exact terms
-    re.compile(r"^[A-Z_]{2,}$"),  # ALL_CAPS constants
-    re.compile(r"\b\w+\.\w+\.\w+\b"),  # Dotted paths: module.sub.func
-]
-
-# Patterns that indicate semantic queries
-_SEMANTIC_SIGNALS = [
-    re.compile(r"\b(what did|how does|why is|explain|describe|summarize|discuss)\b", re.I),
-    re.compile(r"\b(remember|recall|think|know|understand)\b.*\b(about|regarding)\b", re.I),
-    re.compile(r"\?$"),  # Questions
-    re.compile(r"\b(the best way to|how to|what\'s the|approach to)\b", re.I),
-]
-
-
-def analyze_query(query: str) -> QueryAnalysis:
-    """Analyze a query to determine which search methods to use.
-
-    Returns QueryAnalysis with detected type, confidence, and extracted signals.
-    """
-    if not query or not query.strip():
-        return QueryAnalysis(
-            query_type=QueryType.HYBRID,
-            confidence=0.5,
-            signals=["empty_query"],
-        )
-
-    query = query.strip()
-
-    # Score each query type
-    comp_score = 0.0
-    kw_score = 0.0
-    sem_score = 0.0
-    signals = []
-    entities = []
-    keywords = []
-
-    # Check compositional patterns
-    for pattern in _COMPOSITIONAL_PATTERNS:
-        if pattern.search(query):
-            comp_score += 0.3
-            signals.append(f"compositional:{pattern.pattern[:30]}")
-
-    # Check keyword patterns
-    for pattern in _KEYWORD_SIGNALS:
-        if pattern.search(query):
-            kw_score += 0.25
-            match = pattern.search(query)
-            if match:
-                keywords.append(match.group(0))
-            signals.append(f"keyword:{pattern.pattern[:30]}")
-
-    # Check semantic patterns
-    for pattern in _SEMANTIC_SIGNALS:
-        if pattern.search(query):
-            sem_score += 0.25
-            signals.append(f"semantic:{pattern.pattern[:30]}")
-
-    # Extract entities (capitalized multi-word phrases, quoted terms)
-    entity_patterns = [
-        re.compile(r"\b([A-Z][a-z]+(?:\s+[A-Z][a-z]+)+)\b"),
-        re.compile(r"["\']([^"\']+)["\']"),
-    ]
-    for ep in entity_patterns:
-        for m in ep.finditer(query):
-            entities.append(m.group(1))
-
-    # Short queries (< 5 words) with no semantic signals → keyword
-    word_count = len(query.split())
-    if word_count <= 4 and sem_score == 0 and comp_score == 0:
-        kw_score += 0.3
-        signals.append("short_query_keyword_boost")
-
-    # Normalize scores
-    max_score = max(comp_score, kw_score, sem_score, 0.1)
-
-    # Determine query type
-    if max_score < 0.15:
-        # No strong signals → use hybrid (all methods)
-        return QueryAnalysis(
-            query_type=QueryType.HYBRID,
-            confidence=0.5,
-            signals=["no_strong_signals"],
-            entities=entities,
-            keywords=keywords,
-        )
-
-    if comp_score == max_score and comp_score >= 0.3:
-        return QueryAnalysis(
-            query_type=QueryType.COMPOSITIONAL,
-            confidence=min(comp_score, 1.0),
-            signals=signals,
-            entities=entities,
-            keywords=keywords,
-        )
-
-    if kw_score > sem_score:
-        return QueryAnalysis(
-            query_type=QueryType.KEYWORD,
-            confidence=min(kw_score, 1.0),
-            signals=signals,
-            entities=entities,
-            keywords=keywords,
-        )
-
-    return QueryAnalysis(
-        query_type=QueryType.SEMANTIC,
-        confidence=min(sem_score, 1.0),
-        signals=signals,
-        entities=entities,
-        keywords=keywords,
-    )

tests/test_hybrid_search.py

@@ -1,97 +0,0 @@
-"""Tests for hybrid search router — query analysis and RRF merge."""
-
-import pytest
-import sys, os
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "plugins", "memory", "holographic"))
-
-from query_router import QueryType, analyze_query
-from hybrid_search import SearchResult, reciprocal_rank_fusion
-
-
-class TestQueryAnalysis:
-    def test_keyword_single_identifier(self):
-        result = analyze_query("function_name")
-        assert result.query_type == QueryType.KEYWORD
-
-    def test_keyword_quoted_term(self):
-        result = analyze_query('Find "exact phrase" in code')
-        assert result.query_type in (QueryType.KEYWORD, QueryType.HYBRID)
-
-    def test_keyword_dotted_path(self):
-        result = analyze_query("module.sub.function")
-        assert result.query_type == QueryType.KEYWORD
-
-    def test_semantic_question(self):
-        result = analyze_query("What did we discuss about deployment?")
-        assert result.query_type == QueryType.SEMANTIC
-
-    def test_semantic_how_to(self):
-        result = analyze_query("How to configure the gateway?")
-        assert result.query_type == QueryType.SEMANTIC
-
-    def test_compositional_contradiction(self):
-        result = analyze_query("Are there any contradictions in the facts?")
-        assert result.query_type == QueryType.COMPOSITIONAL
-
-    def test_compositional_related(self):
-        result = analyze_query("What facts are related to Alexander?")
-        assert result.query_type == QueryType.COMPOSITIONAL
-
-    def test_empty_query(self):
-        result = analyze_query("")
-        assert result.query_type == QueryType.HYBRID
-
-    def test_complex_query(self):
-        result = analyze_query("What did we decide about the deploy script?")
-        assert result.query_type in (QueryType.SEMANTIC, QueryType.HYBRID)
-
-
-class TestReciprocalRankFusion:
-    def test_single_list(self):
-        results = [
-            SearchResult(fact_id=1, content="A", score=0.9, source="fts5", rank=1),
-            SearchResult(fact_id=2, content="B", score=0.8, source="fts5", rank=2),
-        ]
-        merged = reciprocal_rank_fusion([results])
-        assert len(merged) == 2
-        assert merged[0].fact_id == 1  # Rank 1 should be first
-
-    def test_two_lists_merge(self):
-        list1 = [
-            SearchResult(fact_id=1, content="A", score=0.9, source="fts5", rank=1),
-            SearchResult(fact_id=2, content="B", score=0.8, source="fts5", rank=2),
-        ]
-        list2 = [
-            SearchResult(fact_id=2, content="B", score=0.95, source="qdrant", rank=1),
-            SearchResult(fact_id=3, content="C", score=0.7, source="qdrant", rank=2),
-        ]
-        merged = reciprocal_rank_fusion([list1, list2])
-        # Fact 2 appears in both lists → should rank highest
-        assert merged[0].fact_id == 2
-        assert len(merged) == 3
-
-    def test_empty_lists(self):
-        merged = reciprocal_rank_fusion([[], []])
-        assert len(merged) == 0
-
-    def test_weighted_merge(self):
-        list1 = [
-            SearchResult(fact_id=1, content="A", score=0.9, source="fts5", rank=1),
-        ]
-        list2 = [
-            SearchResult(fact_id=2, content="B", score=0.9, source="qdrant", rank=1),
-        ]
-        merged = reciprocal_rank_fusion(
-            [list1, list2],
-            weights={"fts5": 1.0, "qdrant": 2.0},
-        )
-        # Qdrant has higher weight → fact 2 should win
-        assert merged[0].fact_id == 2
-
-    def test_rrf_score_formula(self):
-        list1 = [
-            SearchResult(fact_id=1, content="A", score=0.9, source="fts5", rank=1),
-        ]
-        merged = reciprocal_rank_fusion([list1], k=60)
-        # RRF score = 1/(60+1) = 0.01639...
-        assert abs(merged[0].score - 1.0/61.0) < 0.001

tests/test_message_dedup.py

@@ -0,0 +1,57 @@
+"""
+Tests for message deduplication (#756).
+"""
+
+import pytest
+import time
+from gateway.message_dedup import MessageDeduplicator
+
+
+class TestMessageDeduplicator:
+    def test_first_message_allowed(self):
+        dedup = MessageDeduplicator()
+        assert dedup.should_send("Hello") is True
+    
+    def test_duplicate_suppressed(self):
+        dedup = MessageDeduplicator()
+        dedup.record_message("Hello", "session1", "telegram")
+        assert dedup.should_send("Hello", "session1", "telegram") is False
+    
+    def test_different_session_allowed(self):
+        dedup = MessageDeduplicator()
+        dedup.record_message("Hello", "session1", "telegram")
+        assert dedup.should_send("Hello", "session2", "telegram") is True
+    
+    def test_different_platform_allowed(self):
+        dedup = MessageDeduplicator()
+        dedup.record_message("Hello", "session1", "telegram")
+        assert dedup.should_send("Hello", "session1", "discord") is True
+    
+    def test_different_content_allowed(self):
+        dedup = MessageDeduplicator()
+        dedup.record_message("Hello", "session1", "telegram")
+        assert dedup.should_send("World", "session1", "telegram") is True
+    
+    def test_window_expiry(self):
+        dedup = MessageDeduplicator(window_seconds=1)
+        dedup.record_message("Hello", "session1", "telegram")
+        time.sleep(1.1)
+        assert dedup.should_send("Hello", "session1", "telegram") is True
+    
+    def test_record_returns_uuid(self):
+        dedup = MessageDeduplicator()
+        msg_id = dedup.record_message("Hello")
+        assert msg_id is not None
+        assert len(msg_id) == 36  # UUID format
+    
+    def test_stats(self):
+        dedup = MessageDeduplicator()
+        dedup.record_message("Hello")
+        dedup.record_message("Hello")  # duplicate
+        stats = dedup.get_stats()
+        assert stats["total_records"] == 1
+        assert stats["suppressed_count"] == 1
+
+
+if __name__ == "__main__":
+    pytest.main([__file__])