test: cron model/provider config preservation (#222)

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

hermes_cli/cron.py

@@ -72,6 +72,12 @@ def cron_list(show_all: bool = False):
             deliver = [deliver]
         deliver_str = ", ".join(deliver)
 
+        model = job.get("model")
+        provider = job.get("provider")
+        model_str = ""
+        if model:
+            model_str = f" @ {provider}/{model}" if provider else f" @ {model}"
+
         skills = job.get("skills") or ([job["skill"]] if job.get("skill") else [])
         if state == "paused":
             status = color("[paused]", Colors.YELLOW)
@@ -168,6 +174,8 @@ def cron_create(args):
         skill=getattr(args, "skill", None),
         skills=_normalize_skills(getattr(args, "skill", None), getattr(args, "skills", None)),
         script=getattr(args, "script", None),
+        model=getattr(args, "model", None),
+        provider=getattr(args, "provider", None),
     )
     if not result.get("success"):
         print(color(f"Failed to create job: {result.get('error', 'unknown error')}", Colors.RED))
@@ -180,6 +188,10 @@ def cron_create(args):
     job_data = result.get("job", {})
     if job_data.get("script"):
         print(f"  Script: {job_data['script']}")
+    if job_data.get("model"):
+        provider = job_data.get("provider", "")
+        model_str = f"{provider}/{job_data['model']}" if provider else job_data["model"]
+        print(f"  Model: {model_str}")
     print(f"  Next run: {result['next_run_at']}")
     return 0
 
@@ -217,6 +229,8 @@ def cron_edit(args):
         deliver=getattr(args, "deliver", None),
         repeat=getattr(args, "repeat", None),
         skills=final_skills,
+        model=getattr(args, "model", None),
+        provider=getattr(args, "provider", None),
         script=getattr(args, "script", None),
     )
     if not result.get("success"):

tests/test_cron_model_preservation.py

@@ -0,0 +1,73 @@
+"""Tests for cron model/provider config preservation (#222)."""
+
+import json
+import pytest
+from unittest.mock import patch, MagicMock
+
+
+def test_create_job_preserves_model_and_provider():
+    """create_job should store model and provider in the job dict."""
+    from cron.jobs import create_job, load_jobs, save_jobs
+    import tempfile, os
+
+    with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
+        json.dump([], f)
+        tmp_path = f.name
+
+    try:
+        with patch("cron.jobs._JOBS_FILE", tmp_path):
+            job = create_job(
+                schedule="0 * * * *",
+                prompt="test prompt",
+                model="xiaomi/mimo-v2-pro",
+                provider="nous",
+            )
+            assert job["model"] == "xiaomi/mimo-v2-pro"
+            assert job["provider"] == "nous"
+
+            # Verify persisted
+            jobs = load_jobs()
+            assert jobs[0]["model"] == "xiaomi/mimo-v2-pro"
+            assert jobs[0]["provider"] == "nous"
+    finally:
+        os.unlink(tmp_path)
+
+
+def test_update_job_preserves_model():
+    """update_job should preserve model/provider when updating other fields."""
+    from cron.jobs import create_job, update_job
+
+    with patch("cron.jobs._JOBS_FILE", "/tmp/test_cron_jobs.json"):
+        import os
+        if os.path.exists("/tmp/test_cron_jobs.json"):
+            os.unlink("/tmp/test_cron_jobs.json")
+
+        job = create_job(
+            schedule="0 * * * *",
+            prompt="test",
+            model="xiaomi/mimo-v2-pro",
+            provider="nous",
+        )
+        # Update prompt — model should be preserved
+        updated = update_job(job["id"], {"prompt": "new prompt"})
+        assert updated["model"] == "xiaomi/mimo-v2-pro"
+        assert updated["provider"] == "nous"
+        assert updated["prompt"] == "new prompt"
+
+        os.unlink("/tmp/test_cron_jobs.json")
+
+
+def test_create_job_without_model_is_none():
+    """create_job without model/provider should store None."""
+    from cron.jobs import create_job
+
+    with patch("cron.jobs._JOBS_FILE", "/tmp/test_cron_none.json"):
+        import os
+        if os.path.exists("/tmp/test_cron_none.json"):
+            os.unlink("/tmp/test_cron_none.json")
+
+        job = create_job(schedule="0 * * * *", prompt="test")
+        assert job["model"] is None
+        assert job["provider"] is None
+
+        os.unlink("/tmp/test_cron_none.json")

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)