docs: add hybrid search documentation (#671)

docs/holographic-vector-hybrid.md

@@ -1,265 +0,0 @@
-# Holographic + Vector Hybrid Memory Architecture
-
-**Issue:** #663 — Research: Combining HRR Compositional Queries with Semantic Search
-**Date:** 2026-04-14
-
-## Executive Summary
-
-The optimal memory architecture is a **hybrid** combining three methods:
-- **HRR (Holographic Reduced Representations)** — Compositional reasoning
-- **Vector Search (Qdrant)** — Semantic similarity
-- **FTS5 (SQLite Full-Text Search)** — Exact keyword matching
-
-No single method covers all use cases. Each excels at different query types.
-
-## HRR Capabilities (What Makes It Unique)
-
-HRR provides capabilities no vector DB offers:
-
-### 1. Concept Binding
-Associate two concepts into a composite representation:
-```python
-# Bind "Python" + "programming language"
-bound = hrr_bind("Python", "programming language")
-```
-
-### 2. Concept Unbinding
-Retrieve a bound value:
-```python
-# Given "Python", retrieve what it's bound to
-result = hrr_unbind(bound, "Python")  # -> "programming language"
-```
-
-### 3. Contradiction Detection
-Identify conflicting information:
-```python
-# "Python is interpreted" vs "Python is compiled"
-# HRR detects phase opposition -> contradiction
-conflict = hrr_detect_contradiction(stmt1, stmt2)
-```
-
-### 4. Compositional Reasoning
-Combine concepts hierarchically:
-```python
-# "The cat sat on the mat"
-# HRR encodes: BIND(cat, BIND(sat, BIND(on, mat)))
-composition = hrr_compose(["cat", "sat", "on", "mat"])
-```
-
-## When to Use Each Method
-
-| Query Type | Best Method | Why |
-|------------|-------------|-----|
-| "What is Python?" | Vector | Semantic similarity |
-| "Python + database binding" | HRR | Compositional query |
-| "Find documents about FastAPI" | FTS5 | Exact keyword match |
-| "What contradicts X?" | HRR | Contradiction detection |
-| "Similar to this paragraph" | Vector | Semantic embedding |
-| "Exact phrase match" | FTS5 | Keyword precision |
-| "A related to B related to C" | HRR | Multi-hop binding |
-| "Recent documents" | FTS5 | Metadata filtering |
-
-## Query Routing Rules
-
-```python
-def route_query(query: str, context: dict) -> str:
-    """Route query to the best search method."""
-    
-    # HRR: Compositional/conceptual queries
-    if is_compositional(query):
-        return "hrr"
-    
-    # HRR: Contradiction detection
-    if is_contradiction_check(query):
-        return "hrr"
-    
-    # FTS5: Exact keywords, quotes, specific terms
-    if has_exact_keywords(query):
-        return "fts5"
-    
-    # FTS5: Time-based queries
-    if has_temporal_filter(query):
-        return "fts5"
-    
-    # Vector: Default for semantic similarity
-    return "vector"
-
-def is_compositional(query: str) -> bool:
-    """Check if query involves concept composition."""
-    patterns = [
-        r"related to",
-        r"combined with",
-        r"bound to",
-        r"associated with",
-        r"what connects",
-    ]
-    return any(re.search(p, query.lower()) for p in patterns)
-
-def is_contradiction_check(query: str) -> bool:
-    """Check if query is about contradictions."""
-    patterns = [
-        r"contradicts?",
-        r"conflicts? with",
-        r"inconsistent",
-        r"opposite of",
-    ]
-    return any(re.search(p, query.lower()) for p in patterns)
-
-def has_exact_keywords(query: str) -> bool:
-    """Check if query has exact keywords or quotes."""
-    return '"' in query or "'" in query or len(query.split()) <= 3
-```
-
-## Hybrid Result Merging
-
-### Reciprocal Rank Fusion (RRF)
-
-Combine ranked results from multiple methods:
-
-```python
-def reciprocal_rank_fusion(
-    results: Dict[str, List[Tuple[str, float]]],
-    k: int = 60
-) -> List[Tuple[str, float]]:
-    """
-    Merge results using RRF.
-    
-    Args:
-        results: {"hrr": [(id, score), ...], "vector": [...], "fts5": [...]}
-        k: RRF constant (default 60)
-    
-    Returns:
-        Merged and re-ranked results
-    """
-    scores = defaultdict(float)
-    
-    for method, ranked_items in results.items():
-        for rank, (item_id, _) in enumerate(ranked_items, 1):
-            scores[item_id] += 1.0 / (k + rank)
-    
-    return sorted(scores.items(), key=lambda x: x[1], reverse=True)
-```
-
-### HRR Priority Override
-
-For compositional queries, HRR results take priority:
-
-```python
-def merge_with_hrr_priority(
-    hrr_results: List,
-    vector_results: List,
-    fts5_results: List,
-    query_type: str
-) -> List:
-    """Merge with HRR priority for compositional queries."""
-    
-    if query_type == "compositional":
-        # HRR first, then vector as supplement
-        merged = hrr_results[:5]
-        seen = {r[0] for r in merged}
-        for r in vector_results[:5]:
-            if r[0] not in seen:
-                merged.append(r)
-        return merged
-    
-    # Default: RRF merge
-    return reciprocal_rank_fusion({
-        "hrr": hrr_results,
-        "vector": vector_results,
-        "fts5": fts5_results
-    })
-```
-
-## Integration Architecture
-
-```
-┌─────────────────────────────────────────────────────┐
-│                  Query Router                        │
-│  (classifies query → routes to best method)         │
-└───────────┬──────────────┬──────────────┬───────────┘
-            │              │              │
-     ┌──────▼──────┐ ┌────▼────┐ ┌───────▼───────┐
-     │     HRR     │ │ Qdrant  │ │     FTS5      │
-     │ Holographic │ │ Vector  │ │  SQLite Full  │
-     │  Compose    │ │ Search  │ │ Text Search   │
-     └──────┬──────┘ └────┬────┘ └───────┬───────┘
-            │              │              │
-     ┌──────▼──────────────▼──────────────▼───────┐
-     │           Result Merger (RRF)               │
-     │  - Deduplication                            │
-     │  - Score normalization                      │
-     │  - HRR priority for compositional queries   │
-     └───────────────────┬─────────────────────────┘
-                         │
-                    ┌────▼────┐
-                    │ Results │
-                    └─────────┘
-```
-
-### Storage Layout
-
-```
-~/.hermes/memory/
-├── holographic/
-│   ├── hrr_store.pkl       # HRR vectors (numpy arrays)
-│   ├── bindings.pkl        # Concept bindings
-│   └── contradictions.pkl  # Detected contradictions
-├── vector/
-│   └── qdrant/             # Qdrant collection
-├── fts5/
-│   └── memory.db           # SQLite with FTS5
-└── index.json              # Unified index
-```
-
-## Preserving HRR Unique Capabilities
-
-### Rules
-
-1. **Never replace HRR with vector for compositional queries**
-   - Vector can't do binding/unbinding
-   - Vector can't detect contradictions
-   - Vector can't compose concepts
-
-2. **HRR is primary for relational queries**
-   - "What relates X to Y?"
-   - "What contradicts this?"
-   - "Combine concept A with concept B"
-
-3. **Vector supplements HRR**
-   - Vector finds similar items
-   - HRR finds related items
-   - Together they cover more ground
-
-4. **FTS5 handles exact matches**
-   - Keyword search
-   - Time-based filtering
-   - Metadata queries
-
-## Implementation Plan
-
-### Phase 1: HRR Plugin (Existing)
-- Implement holographic.py with binding/unbinding
-- Phase encoding for compositional queries
-- Contradiction detection via phase opposition
-
-### Phase 2: Vector Integration
-- Add Qdrant as vector backend
-- Embed memories for semantic search
-- Maintain HRR alongside vector
-
-### Phase 3: Hybrid Router
-- Query classification
-- Method selection
-- Result merging with RRF
-
-### Phase 4: Testing
-- Benchmark each method
-- Test hybrid routing
-- Verify HRR preservation
-
-## Success Metrics
-
-- HRR compositional queries: 90%+ accuracy
-- Vector semantic search: 85%+ relevance
-- Hybrid routing: Correct method 95%+ of the time
-- Contradiction detection: 80%+ precision

docs/hybrid-search.md

@@ -0,0 +1,54 @@
+# 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()`

plugins/memory/holographic/hybrid_search.py

@@ -0,0 +1,277 @@
+"""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

@@ -0,0 +1,168 @@
+"""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

@@ -0,0 +1,97 @@
+"""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_memory_query_router.py

@@ -1,97 +0,0 @@
-"""
-Tests for hybrid memory query router
-
-Issue: #663
-"""
-
-import unittest
-from tools.memory_query_router import (
-    SearchMethod,
-    QueryRouter,
-    route_query,
-    reciprocal_rank_fusion,
-    merge_with_hrr_priority,
-)
-
-
-class TestQueryClassification(unittest.TestCase):
-    
-    def setUp(self):
-        self.router = QueryRouter()
-    
-    def test_contradiction_routes_hrr(self):
-        c = self.router.classify("What contradicts this statement?")
-        self.assertEqual(c.method, SearchMethod.HRR)
-        self.assertGreater(c.confidence, 0.9)
-    
-    def test_compositional_routes_hrr(self):
-        c = self.router.classify("How does Python relate to machine learning?")
-        self.assertEqual(c.method, SearchMethod.HRR)
-        
-        c = self.router.classify("What is associated with quantum computing?")
-        self.assertEqual(c.method, SearchMethod.HRR)
-    
-    def test_exact_keywords_routes_fts5(self):
-        c = self.router.classify('Find documents containing "FastAPI tutorial"')
-        self.assertEqual(c.method, SearchMethod.FTS5)
-    
-    def test_short_query_routes_fts5(self):
-        c = self.router.classify("Python syntax")
-        self.assertEqual(c.method, SearchMethod.FTS5)
-    
-    def test_temporal_routes_fts5(self):
-        c = self.router.classify("Recent changes to the config")
-        self.assertEqual(c.method, SearchMethod.FTS5)
-    
-    def test_semantic_routes_vector(self):
-        c = self.router.classify("Explain how transformers work in natural language processing")
-        self.assertEqual(c.method, SearchMethod.VECTOR)
-
-
-class TestReciprocalRankFusion(unittest.TestCase):
-    
-    def test_basic_fusion(self):
-        results = {
-            "hrr": [("a", 0.9), ("b", 0.8)],
-            "vector": [("b", 0.85), ("c", 0.7)],
-        }
-        merged = reciprocal_rank_fusion(results)
-        
-        # 'b' appears in both, should rank high
-        ids = [r[0] for r in merged]
-        self.assertIn("b", ids[:2])
-    
-    def test_empty_results(self):
-        merged = reciprocal_rank_fusion({})
-        self.assertEqual(len(merged), 0)
-
-
-class TestHRRPriority(unittest.TestCase):
-    
-    def test_compositional_hrr_first(self):
-        hrr = [("a", 0.9), ("b", 0.8)]
-        vector = [("c", 0.85), ("d", 0.7)]
-        fts5 = [("e", 0.6)]
-        
-        merged = merge_with_hrr_priority(hrr, vector, fts5, "compositional")
-        
-        # HRR results should come first
-        self.assertEqual(merged[0][0], "a")
-        self.assertEqual(merged[1][0], "b")
-
-
-class TestHybridDecision(unittest.TestCase):
-    
-    def test_low_confidence_uses_hybrid(self):
-        from tools.memory_query_router import should_use_hybrid
-        # Ambiguous query
-        self.assertTrue(should_use_hybrid("Tell me about things"))
-    
-    def test_clear_query_no_hybrid(self):
-        from tools.memory_query_router import should_use_hybrid
-        # Clear contradiction query
-        self.assertFalse(should_use_hybrid("What contradicts X?"))
-
-
-if __name__ == "__main__":
-    unittest.main()

tools/memory_query_router.py

@@ -1,209 +0,0 @@
-"""
-Hybrid Memory Query Router
-
-Routes queries to the best search method:
-- HRR: Compositional/conceptual queries
-- Vector: Semantic similarity
-- FTS5: Exact keyword matching
-
-Issue: #663
-"""
-
-import re
-from collections import defaultdict
-from dataclasses import dataclass
-from enum import Enum
-from typing import Any, Dict, List, Optional, Tuple
-
-
-class SearchMethod(Enum):
-    """Available search methods."""
-    HRR = "hrr"          # Holographic Reduced Representations
-    VECTOR = "vector"    # Semantic vector search
-    FTS5 = "fts5"        # Full-text search (SQLite)
-    HYBRID = "hybrid"    # Combine multiple methods
-
-
-@dataclass
-class QueryClassification:
-    """Result of query classification."""
-    method: SearchMethod
-    confidence: float
-    reason: str
-    sub_queries: Optional[List[str]] = None
-
-
-# Query patterns for routing
-COMPOSITIONAL_PATTERNS = [
-    r"(?i)\brelated\s+to\b",
-    r"(?i)\bcombined\s+with\b",
-    r"(?i)\bbound\s+to\b",
-    r"(?i)\bassociated\s+with\b",
-    r"(?i)\bwhat\s+connects?\b",
-    r"(?i)\bhow\s+.*\s+relate\b",
-    r"(?i)\brelationship\s+between\b",
-]
-
-CONTRADICTION_PATTERNS = [
-    r"(?i)\bcontradicts?\b",
-    r"(?i)\bconflicts?\s+with\b",
-    r"(?i)\binconsistent\b",
-    r"(?i)\bopposite\s+of\b",
-    r"(?i)\bopposes?\b",
-    r"(?i)\bdisagrees?\s+with\b",
-]
-
-EXACT_KEYWORD_PATTERNS = [
-    r'"[^"]+"',  # Quoted phrases
-    r"'[^']+'",  # Single-quoted phrases
-    r"(?i)\bexact\b",
-    r"(?i)\bprecisely\b",
-    r"(?i)\bspecifically\b",
-]
-
-TEMPORAL_PATTERNS = [
-    r"(?i)\brecent\b",
-    r"(?i)\btoday\b",
-    r"(?i)\byesterday\b",
-    r"(?i)\blast\s+(week|month|hour)\b",
-    r"(?i)\bsince\b",
-    r"(?i)\bbefore\b",
-    r"(?i)\bafter\b",
-]
-
-
-class QueryRouter:
-    """Route queries to the best search method."""
-    
-    def classify(self, query: str) -> QueryClassification:
-        """Classify a query and route to best method."""
-        
-        # Check for contradiction queries (HRR)
-        for pattern in CONTRADICTION_PATTERNS:
-            if re.search(pattern, query):
-                return QueryClassification(
-                    method=SearchMethod.HRR,
-                    confidence=0.95,
-                    reason="Contradiction detection query"
-                )
-        
-        # Check for compositional queries (HRR)
-        for pattern in COMPOSITIONAL_PATTERNS:
-            if re.search(pattern, query):
-                return QueryClassification(
-                    method=SearchMethod.HRR,
-                    confidence=0.90,
-                    reason="Compositional/conceptual query"
-                )
-        
-        # Check for exact keyword queries (FTS5)
-        for pattern in EXACT_KEYWORD_PATTERNS:
-            if re.search(pattern, query):
-                return QueryClassification(
-                    method=SearchMethod.FTS5,
-                    confidence=0.85,
-                    reason="Exact keyword query"
-                )
-        
-        # Check for temporal queries (FTS5)
-        for pattern in TEMPORAL_PATTERNS:
-            if re.search(pattern, query):
-                return QueryClassification(
-                    method=SearchMethod.FTS5,
-                    confidence=0.80,
-                    reason="Temporal query"
-                )
-        
-        # Short queries tend to be keyword searches
-        if len(query.split()) <= 3:
-            return QueryClassification(
-                method=SearchMethod.FTS5,
-                confidence=0.70,
-                reason="Short query (likely keyword)"
-            )
-        
-        # Default: vector search for semantic queries
-        return QueryClassification(
-            method=SearchMethod.VECTOR,
-            confidence=0.60,
-            reason="Semantic similarity query"
-        )
-    
-    def should_use_hybrid(self, query: str) -> bool:
-        """Check if query should use hybrid search."""
-        classification = self.classify(query)
-        
-        # Low confidence -> use hybrid
-        if classification.confidence < 0.70:
-            return True
-        
-        # Mixed signals -> use hybrid
-        has_compositional = any(re.search(p, query) for p in COMPOSITIONAL_PATTERNS)
-        has_keywords = any(re.search(p, query) for p in EXACT_KEYWORD_PATTERNS)
-        
-        return has_compositional and has_keywords
-
-
-def reciprocal_rank_fusion(
-    results: Dict[str, List[Tuple[str, float]]],
-    k: int = 60
-) -> List[Tuple[str, float]]:
-    """
-    Merge results using Reciprocal Rank Fusion.
-    
-    Args:
-        results: Dict of method -> [(item_id, score), ...]
-        k: RRF constant (default 60)
-    
-    Returns:
-        Merged and re-ranked results
-    """
-    scores = defaultdict(float)
-    
-    for method, ranked_items in results.items():
-        for rank, (item_id, _) in enumerate(ranked_items, 1):
-            scores[item_id] += 1.0 / (k + rank)
-    
-    return sorted(scores.items(), key=lambda x: x[1], reverse=True)
-
-
-def merge_with_hrr_priority(
-    hrr_results: List[Tuple[str, float]],
-    vector_results: List[Tuple[str, float]],
-    fts5_results: List[Tuple[str, float]],
-    query_type: str = "default"
-) -> List[Tuple[str, float]]:
-    """
-    Merge results with HRR priority for compositional queries.
-    """
-    if query_type == "compositional":
-        # HRR first, vector as supplement
-        merged = hrr_results[:5]
-        seen = {r[0] for r in merged}
-        
-        for r in vector_results[:5]:
-            if r[0] not in seen:
-                merged.append(r)
-        
-        return merged
-    
-    # Default: RRF merge
-    return reciprocal_rank_fusion({
-        "hrr": hrr_results,
-        "vector": vector_results,
-        "fts5": fts5_results
-    })
-
-
-# Module-level router
-_router = QueryRouter()
-
-
-def route_query(query: str) -> QueryClassification:
-    """Route a query to the best search method."""
-    return _router.classify(query)
-
-
-def should_use_hybrid(query: str) -> bool:
-    """Check if query should use hybrid search."""
-    return _router.should_use_hybrid(query)