feat: add test data cleanup endpoint

Add POST /maintenance/cleanup/test-data endpoint to purge LLM test data
from wiki, graph, and vectors. Security-restricted to test user namespace
only (users/llm-tester/*, users/llm_tester/*).

- Supports dry_run=true (default) to preview before deleting
- Cleans vectors, graph nodes, and wiki pages
- Scheduler task configured for weekly cleanup (Sunday 3:00 AM)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

CHANGELOG.md

@@ -5,6 +5,46 @@ All notable changes to Library Desk will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [1.4.4] - 2025-12-24
+
+### Added
+
+- **Test Data Cleanup Endpoint** - `POST /maintenance/cleanup/test-data`
+  - Purges LLM test data from wiki, graph, and vectors
+  - Security-restricted to test user namespace only (`users/llm-tester/*`, `users/llm_tester/*`)
+  - Supports `dry_run=true` (default) to preview before deleting
+  - Scheduler task configured for weekly cleanup (Sunday 3:00 AM)
+
+## [1.4.3] - 2025-12-24
+
+### Changed
+
+- **Volatile Cache System Refactored to Vector Storage**
+  - Backend migrated from Redis to Qdrant for semantic search capability
+  - Data converted to natural language for embedding and semantic retrieval
+  - Collection naming: `volatile_{user}` for per-user isolation
+  - TTL implemented via `ttl_expiry` timestamp in vector payload
+  - Simplified endpoints:
+    - `GET /volatile/search?q=...` - Semantic search across volatile data
+    - `POST /volatile/store?namespace=...&key=...` - Store with query params
+    - `GET /volatile/{namespace}/{key}` - Get specific record
+    - `DELETE /volatile/{namespace}/{key}` - Delete record
+  - Removed namespace-specific URL patterns (simpler API for LLM tool use)
+
+### Added
+
+- **HybridRAG Volatile Integration** - Volatile cache now included in multi-source search
+  - Volatile results get priority boost in RRF fusion (current data ranks higher)
+  - New config options: `enable_volatile`, `volatile_limit` (default 1), `volatile_threshold`
+  - Timing breakdown includes `volatile_ms`
+- **Volatile Cleanup Endpoint** - `POST /maintenance/cleanup/volatile`
+  - Purges expired records across all `volatile_*` collections
+  - Scheduler task for every 10 minutes recommended
+  - Returns per-collection cleanup counts
+- **Natural Language Conversion** - Structured data converted for embedding
+  - Template-based conversion for each namespace (weather, news, financial, etc.)
+  - Fallback for custom namespaces
+
 ## [1.4.2] - 2025-12-24
 
 ### Added

docs/MEMORY_SYSTEM_PLAN.md

@@ -6,15 +6,24 @@ A three-tier memory architecture for Library Desk with intelligent orchestration
 
 | Tier | Storage | Purpose | TTL |
 |------|---------|---------|-----|
-| **Volatile** | Redis | Weather, news, financial, ephemeral context | 5min - 2hr |
+| **Volatile** | Qdrant (vectors) | Weather, news, financial, ephemeral context | 5min - 2hr |
 | **Documents** | TBD (research) | Git mirrors, PDFs, video, images | Permanent |
 | **Knowledge** | Wiki + Neo4j | Personal dossiers, research, summaries | Permanent |
 
-**Implementation Priority**: Cleanup → Volatile → Documents
+**Implementation Priority**: Cleanup → Volatile → Documents → Test Data Cleanup
+
+### Phase Status
+
+| Phase | Status | Version |
+|-------|--------|---------|
+| Phase 1: Cleanup System | ✅ Complete | v1.4.0 |
+| Phase 2: Volatile Memory | ✅ Complete | v1.4.3 |
+| Phase 3: Document Storage | ⏳ Pending | - |
+| Phase 4: Test Data Cleanup | ✅ Complete | v1.4.4 |
 
 ---
 
-## Phase 1: Cleanup System Completion
+## Phase 1: Cleanup System Completion ✅
 
 ### Current State
 - **COMPLETE** - All Phase 1 tasks implemented
@@ -57,9 +66,9 @@ A three-tier memory architecture for Library Desk with intelligent orchestration
 
 ---
 
-## Phase 2: Volatile Memory System
+## Phase 2: Volatile Memory System ✅
 
-### Architecture
+### Architecture (Final Implementation)
 
 ```
 ┌─────────────────┐    ┌──────────────┐    ┌─────────────────┐
@@ -71,88 +80,46 @@ A three-tier memory architecture for Library Desk with intelligent orchestration
          │
          ▼
 ┌─────────────────┐
-│     Redis       │
-│  (DB 4, TTL)    │
+│     Qdrant      │
+│ (volatile_{user})│
 └─────────────────┘
 ```
 
-### Data Model
+**Key design decisions:**
+- Vector storage in Qdrant (not Redis) for semantic search
+- Collection per user: `volatile_{user}`
+- TTL via `ttl_expiry` timestamp in payload
+- Natural language conversion for embedding structured data
+- Integrated into HybridRAG with priority boost
 
-```python
-class VolatileRecord(BaseModel):
-    key: str                           # e.g., "weather:rotterdam"
-    namespace: str                     # e.g., "weather", "news", "financial"
-    data: dict                         # Actual content
-    source: Optional[str]              # Origin API/service
-    created_at: datetime
-    updated_at: datetime
-    ttl: int                           # Seconds until expiration
-    refresh_schedule: Optional[str]    # Cron expression, if repeating
-    user: str                          # Multi-tenant isolation
-```
-
-**Key pattern**: `{user}:volatile:{namespace}:{key_hash}`
-
-### Implementation Order: Integration-First
-
-1. **Start with Consolidation Hook** - Understand data flow through existing system
-2. **Build Service Layer** - VolatileCacheService with Redis operations
-3. **Add API Endpoints** - REST interface for volatile data
-4. **Biographer Integration** - Query user preferences for relevance
-
-### Tasks
-
-#### 2.1 Integrate with Consolidation (FIRST)
-**New file**: `src/services/volatile_service.py`
-
-```python
-class VolatileCacheService:
-    async def get(user, namespace, key) -> Optional[VolatileRecord]
-    async def set(user, namespace, key, data, ttl, refresh_schedule=None)
-    async def delete(user, namespace, key)
-    async def list_namespace(user, namespace) -> List[str]
-    async def get_scheduled(user) -> List[VolatileRecord]  # For scheduler
-```
-
-#### 2.2 Create Volatile API Router
-**New file**: `src/routers/volatile.py`
+### Endpoints (Implemented)
 
 | Endpoint | Method | Purpose |
 |----------|--------|---------|
-| `/volatile/{namespace}/{key}` | GET | Retrieve record |
-| `/volatile/{namespace}/{key}` | POST | Store/update record |
+| `/volatile/search?q=...` | GET | Semantic search across volatile data |
+| `/volatile/store?namespace=...&key=...` | POST | Store/update record |
+| `/volatile/{namespace}/{key}` | GET | Retrieve specific record |
 | `/volatile/{namespace}/{key}` | DELETE | Remove record |
-| `/volatile/{namespace}` | GET | List keys in namespace |
-| `/volatile/scheduled` | GET | List records needing refresh |
 | `/volatile/stats` | GET | Cache statistics |
+| `/volatile/scheduled` | GET | Records needing refresh |
+| `/volatile/namespaces` | GET | List available namespaces |
+| `/maintenance/cleanup/volatile` | POST | Purge expired records |
 
-#### 2.3 Integrate with Consolidation
-**File**: `src/services/consolidation_service.py`
+### Namespaces
 
-Add relevance trigger detection:
-1. During consolidation, analyze search results for location/interest patterns
-2. Query tatlock's Biographer collection for user preferences
-3. If match found, create/update volatile refresh schedule
-
-#### 2.4 Biographer Integration
-**File**: `src/core/dependencies.py`
-
-```python
-def get_biographer_qdrant() -> QdrantClientWrapper:
-    """Direct access to tatlock's Biographer collection."""
-    # Configure to connect to tatlock's Qdrant
-```
-
-#### 2.5 Scheduler-Side Configuration
-Document required scheduler tasks:
-```json
-{
-  "task_name": "volatile_refresh",
-  "schedule": "*/15 * * * *",
-  "endpoint": "GET /volatile/scheduled",
-  "follow_up": "For each record, call refresh endpoint with record.refresh_schedule"
-}
-```
+| Namespace | Default TTL | Use Case |
+|-----------|-------------|----------|
+| weather | 30 min | Current conditions, forecasts |
+| news | 1 hour | Headlines, breaking news |
+| financial | 5 min | Stock prices, exchange rates |
+| transit | 5 min | Train/bus schedules, delays |
+| traffic | 10 min | Commute times, road conditions |
+| air_quality | 1 hour | Pollution, pollen counts |
+| sports | 1 min | Live scores, matches |
+| social | 10 min | Social notifications |
+| system | 1 min | Service health status |
+| context | 1 hour | Session state |
+| custom | 1 hour | User-defined data |
 
 ---
 
@@ -219,34 +186,116 @@ Add LLM-powered category descriptor generation:
 
 ---
 
-## Files to Modify/Create
+## Phase 4: LLM Tester Data Cleanup ✅
 
-### Phase 1 (Cleanup)
-- `src/routers/maintenance.py` - Add timestamp tracking
+### Problem
+
+LLM testing creates accumulated cruft across the system:
+- Wiki.js pages under `llm-tester/` and `llm_tester/` paths
+- Graph nodes (Document, Entity) linked to test pages
+- Vector chunks in Qdrant for test content
+
+This data accumulates over time and clutters Wiki.js visually (no separate tenant scope for tests).
+
+### Solution
+
+Add a maintenance endpoint to purge all LLM tester artifacts across wiki, graph, and vectors.
+
+### Tasks
+
+#### 4.1 Identify Test Data Patterns ✅
+**Patterns matched** (security-restricted to test user namespace):
+- `users/llm-tester/*`
+- `users/llm_tester/*`
+
+#### 4.2 Add Cleanup Endpoint ✅
+**File**: `src/routers/maintenance.py`
+
+```python
+@router.post("/cleanup/test-data")
+async def cleanup_test_data(
+    dry_run: bool = Query(default=True),
+    wiki: WikiJSDep = None,
+    vector_service: VectorServiceDep = None,
+    graph_service: GraphServiceDep = None,
+    api_key: str = Depends(verify_api_key)
+):
+    """
+    Purge LLM tester data from wiki, graph, and vectors.
+
+    **Security**: Only deletes pages in the test user namespace:
+    - users/llm-tester/*
+    - users/llm_tester/*
+
+    Use dry_run=true to preview what would be deleted.
+    """
+```
+
+#### 4.3 Implementation Steps ✅
+
+1. **Wiki cleanup**: Delete pages via GraphQL mutation
+2. **Graph cleanup**: Delete Document nodes using `delete_page()` method
+3. **Vector cleanup**: Delete chunks using `delete_page_chunks()` method
+
+#### 4.4 Scheduler Integration ✅
+**Recommended schedule**: Weekly (Sunday 3:00 AM)
+
+```json
+{
+  "task_name": "test_data_cleanup",
+  "schedule": "0 3 * * 0",
+  "endpoint": "POST /maintenance/cleanup/test-data?dry_run=false",
+  "description": "Weekly cleanup of LLM test data"
+}
+```
+
+### Files to Modify
+
+- `src/routers/maintenance.py` - Add cleanup endpoint
+- `src/services/wiki_service.py` - Add bulk delete by path pattern (if needed)
+- `src/services/graph_service.py` - May need pattern-based node deletion
+- `src/services/vector_service.py` - Add pattern-based chunk deletion
+
+---
+
+## Files Modified/Created
+
+### Phase 1 (Cleanup) ✅
+- `src/routers/maintenance.py` - Timestamp tracking, cleanup endpoints
 - `src/services/graph_service.py` - Bidirectional validation
 - `src/services/vector_service.py` - Cross-reference checks
 - `LIBRARIAN_INTEGRATION.md` - Scheduler config docs
 
-### Phase 2 (Volatile)
-- `src/services/volatile_service.py` - **NEW**
-- `src/routers/volatile.py` - **NEW**
-- `src/models/volatile.py` - **NEW**
-- `src/core/dependencies.py` - Add Biographer client
-- `src/services/consolidation_service.py` - Relevance triggers
-- `tests/test_volatile.py` - **NEW**
+### Phase 2 (Volatile) ✅
+- `src/services/volatile_service.py` - Qdrant-based volatile cache
+- `src/routers/volatile.py` - Simplified endpoints
+- `src/models/volatile.py` - Namespaces and models
+- `src/models/hybrid_rag.py` - Volatile config options
+- `src/services/hybrid_rag_service.py` - Volatile integration
+- `src/clients/qdrant_client.py` - Expiry filter methods
+- `tests/test_volatile.py` - 37 tests
 
 ### Phase 3 (Documents)
 - `docs/DOCUMENT_STORAGE_RESEARCH.md` - **NEW**
 - `src/services/document_store_service.py` - **NEW** (post-research)
 - `src/routers/documents.py` - **NEW** (post-research)
 
+### Phase 4 (Test Data Cleanup)
+- `src/routers/maintenance.py` - Add cleanup endpoint
+- `src/services/wiki_service.py` - Bulk delete by path pattern
+- `src/services/graph_service.py` - Pattern-based node deletion
+- `src/services/vector_service.py` - Pattern-based chunk deletion
+
 ---
 
 ## Resolved Design Decisions
 
-1. **Biographer Qdrant**: Same Qdrant instance, different collection. Library-Desk queries directly.
-2. **Scheduler API**: Has REST API for task registration. Library-Desk can programmatically create refresh schedules.
-3. **External API calls**: Library-Desk routes through SearXNG for web search. Consider dedicated API integrations for high-value volatiles (weather, financial) for consistent quality.
+1. **Volatile Storage**: Qdrant vectors (not Redis) for semantic search capability
+2. **Collection Naming**: `volatile_{user}` for per-user isolation
+3. **TTL Mechanism**: `ttl_expiry` timestamp in payload, background cleanup job
+4. **HybridRAG Integration**: Volatile as third source with RRF priority boost
+5. **Biographer Qdrant**: Same Qdrant instance, different collection
+6. **Scheduler API**: Has REST API for task registration
 
 ---
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "library-desk"
-version = "1.4.2"
+version = "1.4.4"
 description = "Coordination service for The Library system - HybridRAG queries, document ingestion, entity extraction, and knowledge consolidation"
 readme = "README.md"
 requires-python = ">=3.12"

src/clients/qdrant_client.py

@@ -11,7 +11,7 @@ Provides async vector operations with:
 from qdrant_client import QdrantClient
 from qdrant_client.models import (
     Distance, VectorParams, PointStruct,
-    Filter, FieldCondition, MatchValue
+    Filter, FieldCondition, MatchValue, Range
 )
 from typing import List, Dict, Any, Optional
 import uuid
@@ -676,4 +676,134 @@ class QdrantClientWrapper:
 
         except Exception as e:
             logger.error(f"Failed to list collections: {e}", exc_info=True)
+            return []
+
+    # ========== Volatile Data Methods ==========
+
+    async def search_with_expiry_filter(
+        self,
+        collection_name: str,
+        query_vector: List[float],
+        current_timestamp: int,
+        limit: int = 10,
+        score_threshold: float = 0.7
+    ) -> List[Dict[str, Any]]:
+        """
+        Search vectors filtering out expired records.
+
+        Args:
+            collection_name: Collection name
+            query_vector: Query embedding vector
+            current_timestamp: Current time in milliseconds
+            limit: Maximum results
+            score_threshold: Minimum similarity score
+
+        Returns:
+            List of non-expired search results
+        """
+        # Filter: ttl_expiry > current_timestamp (not expired)
+        expiry_filter = Filter(
+            must=[
+                FieldCondition(
+                    key="ttl_expiry",
+                    range=Range(gt=current_timestamp)
+                )
+            ]
+        )
+
+        try:
+            response = self.client.query_points(
+                collection_name=collection_name,
+                query=query_vector,
+                limit=limit,
+                score_threshold=score_threshold,
+                query_filter=expiry_filter,
+                with_payload=True
+            )
+
+            return [
+                {
+                    "id": str(point.id),
+                    "score": point.score,
+                    "payload": dict(point.payload)
+                }
+                for point in response.points
+            ]
+        except Exception as e:
+            logger.error(f"Volatile search failed: {e}", exc_info=True)
+            return []
+
+    async def delete_expired_vectors(
+        self,
+        collection_name: str,
+        current_timestamp: int
+    ) -> int:
+        """
+        Delete all vectors where ttl_expiry < current_timestamp.
+
+        Args:
+            collection_name: Collection name
+            current_timestamp: Current time in milliseconds
+
+        Returns:
+            Number of points deleted (approximate)
+        """
+        # Filter: ttl_expiry < current_timestamp (expired)
+        expiry_filter = Filter(
+            must=[
+                FieldCondition(
+                    key="ttl_expiry",
+                    range=Range(lt=current_timestamp)
+                )
+            ]
+        )
+
+        try:
+            # First count how many will be deleted (scroll to count)
+            count = 0
+            offset = None
+            while True:
+                points, next_offset = self.client.scroll(
+                    collection_name=collection_name,
+                    scroll_filter=expiry_filter,
+                    limit=100,
+                    offset=offset,
+                    with_payload=False
+                )
+                count += len(points)
+                if next_offset is None:
+                    break
+                offset = next_offset
+
+            if count == 0:
+                return 0
+
+            # Delete expired points
+            self.client.delete(
+                collection_name=collection_name,
+                points_selector=expiry_filter
+            )
+
+            logger.info(f"Deleted {count} expired vectors from {collection_name}")
+            return count
+
+        except Exception as e:
+            logger.error(f"Failed to delete expired vectors: {e}", exc_info=True)
+            return 0
+
+    async def get_volatile_collections(self) -> List[str]:
+        """
+        Get all volatile collections (prefixed with 'volatile_').
+
+        Returns:
+            List of volatile collection names
+        """
+        try:
+            collections = self.client.get_collections()
+            return [
+                c.name for c in collections.collections
+                if c.name.startswith("volatile_")
+            ]
+        except Exception as e:
+            logger.error(f"Failed to list volatile collections: {e}", exc_info=True)
             return []

src/models/hybrid_rag.py

@@ -14,13 +14,16 @@ class HybridRAGConfig(BaseModel):
     vector_limit: int = Field(default=10, ge=1, le=50, description="Max vector results")
     graph_limit: int = Field(default=10, ge=1, le=50, description="Max graph results")
     web_limit: int = Field(default=5, ge=1, le=20, description="Max web results")
+    volatile_limit: int = Field(default=1, ge=1, le=5, description="Max volatile results (typically 1)")
     enable_vector: bool = Field(default=True, description="Enable vector search")
     enable_graph: bool = Field(default=True, description="Enable graph search")
     enable_web: bool = Field(default=True, description="Enable web search")
+    enable_volatile: bool = Field(default=True, description="Enable volatile cache search")
     enable_reranking: bool = Field(default=True, description="Enable LLM re-ranking")
     enable_enrichment: bool = Field(default=True, description="Enable graph enrichment")
     final_result_count: int = Field(default=10, ge=1, le=50, description="Final results to return")
     rrf_k: int = Field(default=60, ge=1, le=100, description="RRF constant")
+    volatile_threshold: float = Field(default=0.8, ge=0.5, le=1.0, description="Volatile similarity threshold")
 
 
 class RelatedDossier(BaseModel):
@@ -34,7 +37,7 @@ class RelatedDossier(BaseModel):
 
 class HybridRAGResult(BaseModel):
     """Single result from HybridRAG query."""
-    source_type: str = Field(..., description="Source: 'vector', 'graph', 'web'")
+    source_type: str = Field(..., description="Source: 'wiki', 'web', 'volatile'")
     title: str
     content: str
     url: Optional[str] = Field(None, description="URL for web results")
@@ -53,6 +56,7 @@ class TimingBreakdown(BaseModel):
     vector_ms: float = Field(..., description="Phase 1: Vector search")
     graph_ms: float = Field(..., description="Phase 1: Graph search")
     web_ms: float = Field(..., description="Phase 1: Web search")
+    volatile_ms: float = Field(default=0, description="Phase 1: Volatile cache search")
     fusion_ms: float = Field(..., description="Phase 2: RRF fusion")
     enrichment_ms: float = Field(..., description="Phase 3: Graph enrichment")
     reranking_ms: float = Field(..., description="Phase 4: LLM re-ranking")

src/routers/hybrid_rag.py

@@ -1,7 +1,7 @@
 """
 HybridRAG router for multi-source search API.
 
-Provides endpoint for combining vector, graph, and web search
+Provides endpoint for combining vector, graph, volatile cache, and web search
 with RRF fusion and LLM re-ranking.
 """
 
@@ -34,10 +34,12 @@ def get_hybrid_rag_service(
     """Get HybridRAG service instance with all dependencies."""
     from src.services.vector_service import VectorService
     from src.services.graph_service import GraphService
+    from src.services.volatile_service import VolatileCacheService
 
     # Create component services
     vector_service = VectorService(qdrant_client, wiki_client, ollama_client)
     graph_service = GraphService(neo4j_client, wiki_client)
+    volatile_service = VolatileCacheService(qdrant_client, ollama_client, settings)
 
     # Create HybridRAG service
     return HybridRAGService(
@@ -46,7 +48,8 @@ def get_hybrid_rag_service(
         searxng_client=searxng_client,
         ollama_client=ollama_client,
         content_extractor=content_extractor,
-        settings=settings
+        settings=settings,
+        volatile_service=volatile_service
     )
 
 
@@ -58,26 +61,28 @@ async def hybrid_search(
     api_key: str = Depends(verify_api_key)
 ):
     """
-    Execute HybridRAG query combining vector, graph, and web search.
+    Execute HybridRAG query combining vector, graph, volatile cache, and web search.
 
     **6-Phase Pipeline:**
     1. **Query Enhancement**: Extract keywords/synonyms with LLM
-    2. **Parallel Retrieval**: Search vector (Qdrant), graph (Neo4j), web (SearXNG)
-    3. **RRF Fusion**: Merge results with Reciprocal Rank Fusion
+    2. **Parallel Retrieval**: Search vector (Qdrant), graph (Neo4j), volatile cache, web (SearXNG)
+    3. **RRF Fusion**: Merge results with Reciprocal Rank Fusion (volatile gets priority boost)
     4. **Enrichment**: Add related documents via shared entities
-    5. **LLM Re-ranking**: Re-rank with mistral-nemo for relevance
+    5. **LLM Re-ranking**: Re-rank with configured model for relevance
     6. **Context Formatting**: Format for LLM consumption
     7. **Persistence**: Store for Librarian knowledge consolidation
 
     **Example Request:**
     ```json
     {
-        "query": "How does Docker orchestration work with Kubernetes?",
+        "query": "What's the weather in Rotterdam?",
         "user": "jpmschweitzer",
         "config": {
             "vector_limit": 10,
             "graph_limit": 10,
             "web_limit": 5,
+            "volatile_limit": 5,
+            "enable_volatile": true,
             "enable_reranking": true,
             "final_result_count": 10
         }
@@ -85,7 +90,7 @@ async def hybrid_search(
     ```
 
     **Returns:**
-    - Ranked results from all sources
+    - Ranked results from all sources (wiki, volatile, web)
     - Extracted keywords/synonyms
     - Related dossiers (via graph)
     - Formatted context for LLM

src/routers/maintenance.py

@@ -16,10 +16,13 @@ import time
 
 from src.services.vector_service import VectorService
 from src.services.graph_service import GraphService
+from src.services.volatile_service import VolatileCacheService
 from src.core.dependencies import (
     VectorServiceDep, GraphServiceDep, WikiJSDep, RedisDep,
-    verify_api_key
+    QdrantDep, OllamaDep, verify_api_key
 )
+from src.config import get_settings
+from src.core.multi_tenancy import DEFAULT_USER
 from datetime import datetime, timezone
 
 logger = logging.getLogger(__name__)
@@ -209,6 +212,44 @@ class ReconcileIndexResponse(BaseModel):
     total_duration_ms: float
 
 
+class VolatileCleanupResponse(BaseModel):
+    """Response from volatile cache cleanup operation."""
+    success: bool
+    collections_processed: int
+    total_expired_purged: int
+    by_collection: Dict[str, int] = Field(default_factory=dict)
+    duration_ms: float
+
+
+class TestDataCleanupResponse(BaseModel):
+    """Response from test data cleanup operation."""
+    success: bool
+    dry_run: bool
+    wiki_pages_deleted: int
+    graph_nodes_deleted: int
+    vector_chunks_deleted: int
+    pages_found: List[Dict[str, Any]] = Field(default_factory=list)
+    duration_ms: float
+
+
+# Test data path patterns - restricted to test user namespace only
+# These are the only paths that can be cleaned up for safety
+TEST_USER_PATH_PREFIXES = [
+    "users/llm-tester/",
+    "users/llm_tester/",
+]
+
+
+def _matches_test_user_path(path: str) -> bool:
+    """Check if a path is in the test user namespace.
+
+    Only matches paths that START with test user prefixes for safety.
+    This prevents accidental deletion of non-test data.
+    """
+    path_lower = path.lower()
+    return any(path_lower.startswith(prefix) for prefix in TEST_USER_PATH_PREFIXES)
+
+
 # ========== Endpoints ==========
 
 @router.post("/cleanup/vectors", response_model=VectorCleanupResponse)
@@ -490,6 +531,150 @@ async def cleanup_all(
         raise HTTPException(status_code=500, detail=str(e))
 
 
+@router.post("/cleanup/volatile", response_model=VolatileCleanupResponse)
+async def cleanup_volatile(
+    qdrant: QdrantDep = None,
+    ollama: OllamaDep = None,
+    api_key: str = Depends(verify_api_key)
+):
+    """
+    Purge expired volatile cache records across all users.
+
+    Loops through all volatile_* collections and removes records where
+    ttl_expiry < current_timestamp.
+
+    **Scheduler Task** - Recommended to run every 10 minutes.
+
+    **Scheduler Integration:**
+    ```json
+    {
+      "task_name": "volatile_cleanup",
+      "schedule": "*/10 * * * *",
+      "endpoint": "POST /maintenance/cleanup/volatile",
+      "description": "Purge expired volatile cache records"
+    }
+    ```
+    """
+    start_time = time.time()
+
+    try:
+        settings = get_settings()
+        service = VolatileCacheService(
+            qdrant_client=qdrant,
+            ollama_client=ollama,
+            settings=settings
+        )
+
+        # Purge expired from all volatile collections
+        results = await service.purge_all_expired()
+
+        total_purged = sum(results.values())
+        duration_ms = (time.time() - start_time) * 1000
+
+        logger.info(f"Volatile cleanup complete: {total_purged} expired records purged from {len(results)} collections")
+
+        return VolatileCleanupResponse(
+            success=True,
+            collections_processed=len(results),
+            total_expired_purged=total_purged,
+            by_collection=results,
+            duration_ms=duration_ms
+        )
+
+    except Exception as e:
+        logger.error(f"Volatile cleanup failed: {e}", exc_info=True)
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@router.post("/cleanup/test-data", response_model=TestDataCleanupResponse)
+async def cleanup_test_data(
+    dry_run: bool = Query(default=True, description="Preview only, don't delete"),
+    wiki: WikiJSDep = None,
+    vector_service: VectorServiceDep = None,
+    graph_service: GraphServiceDep = None,
+    api_key: str = Depends(verify_api_key)
+):
+    """
+    Purge LLM tester data from wiki, graph, and vectors.
+
+    **Security**: Only deletes pages in the test user namespace:
+    - users/llm-tester/*
+    - users/llm_tester/*
+
+    This endpoint cannot delete data outside these paths.
+
+    **Use dry_run=true (default) to preview what would be deleted.**
+
+    **Scheduler Integration:**
+    ```json
+    {
+      "task_name": "test_data_cleanup",
+      "schedule": "0 3 * * 0",
+      "endpoint": "POST /maintenance/cleanup/test-data?dry_run=false",
+      "description": "Weekly cleanup of LLM test data"
+    }
+    ```
+    """
+    start_time = time.time()
+
+    try:
+        # List all wiki pages
+        all_pages = await wiki.list_all_pages(batch_size=500)
+
+        # Filter for test user paths only (security: restricted to test namespace)
+        test_pages = [
+            {"id": p["id"], "path": p["path"], "title": p.get("title", "")}
+            for p in all_pages
+            if _matches_test_user_path(p.get("path", ""))
+        ]
+
+        logger.info(f"Found {len(test_pages)} test pages matching patterns: {TEST_USER_PATH_PREFIXES}")
+
+        wiki_deleted = 0
+        graph_deleted = 0
+        vector_deleted = 0
+
+        if not dry_run and test_pages:
+            for page in test_pages:
+                page_id = page["id"]
+                page_path = page["path"]
+
+                try:
+                    # Delete vector chunks for this page (using DEFAULT_USER collection)
+                    chunks_removed = await vector_service.delete_page_chunks(page_id, DEFAULT_USER)
+                    vector_deleted += chunks_removed
+
+                    # Delete graph node for this page (returns count, may be 0 if no node)
+                    graph_removed = await graph_service.delete_page(page_id, DEFAULT_USER)
+                    graph_deleted += graph_removed
+
+                    # Delete wiki page (raises exception on failure, returns None on success)
+                    await wiki.delete_page(page_id)
+                    wiki_deleted += 1
+
+                    logger.info(f"Deleted test page: {page_path} (id={page_id})")
+
+                except Exception as e:
+                    logger.error(f"Failed to delete page {page_path}: {e}")
+                    continue
+
+        duration_ms = (time.time() - start_time) * 1000
+
+        return TestDataCleanupResponse(
+            success=True,
+            dry_run=dry_run,
+            wiki_pages_deleted=wiki_deleted,
+            graph_nodes_deleted=graph_deleted,
+            vector_chunks_deleted=vector_deleted,
+            pages_found=test_pages,
+            duration_ms=duration_ms
+        )
+
+    except Exception as e:
+        logger.error(f"Test data cleanup failed: {e}", exc_info=True)
+        raise HTTPException(status_code=500, detail=str(e))
+
+
 @router.get("/health", response_model=HealthCheckResponse)
 async def maintenance_health(
     user: str = Query(..., description="User identifier"),

src/routers/volatile.py

@@ -2,6 +2,7 @@
 Volatile cache router for Library Desk API.
 
 Endpoints for ephemeral cached data with TTL - weather, news, financial, etc.
+Data is stored as vectors in Qdrant for semantic search retrieval.
 """
 
 from fastapi import APIRouter, HTTPException, Depends, Query
@@ -14,11 +15,11 @@ from src.models.volatile import (
     VolatileScheduledResponse,
     VolatileStatsResponse,
     VolatileDeleteResponse,
-    VolatileBulkDeleteResponse,
     VolatileNamespace,
+    NAMESPACE_DEFAULT_TTL,
 )
 from src.services.volatile_service import VolatileCacheService
-from src.core.dependencies import verify_api_key, RedisDep
+from src.core.dependencies import verify_api_key, QdrantDep, OllamaDep
 from src.core.multi_tenancy import DEFAULT_USER
 from src.config import get_settings
 
@@ -27,16 +28,21 @@ logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/volatile", tags=["Volatile Cache"])
 
 
-def get_volatile_service(redis: RedisDep) -> VolatileCacheService:
+def get_volatile_service(qdrant: QdrantDep, ollama: OllamaDep) -> VolatileCacheService:
     """Get volatile cache service instance."""
     settings = get_settings()
-    return VolatileCacheService(redis_client=redis, settings=settings)
+    return VolatileCacheService(
+        qdrant_client=qdrant,
+        ollama_client=ollama,
+        settings=settings
+    )
 
 
 @router.get("/stats", response_model=VolatileStatsResponse)
 async def get_stats(
     user: str = Query(default=DEFAULT_USER, description="User identifier"),
-    redis: RedisDep = None,
+    qdrant: QdrantDep = None,
+    ollama: OllamaDep = None,
     api_key: str = Depends(verify_api_key)
 ):
     """
@@ -44,14 +50,14 @@ async def get_stats(
 
     Returns counts of records by namespace and scheduled refresh info.
     """
-    service = get_volatile_service(redis)
+    service = get_volatile_service(qdrant, ollama)
     stats = await service.get_stats(user)
 
     return VolatileStatsResponse(
         total_records=stats["total_records"],
         by_namespace=stats["by_namespace"],
         scheduled_count=stats["scheduled_count"],
-        total_memory_bytes=stats.get("total_memory_bytes"),
+        total_memory_bytes=None,
         user=user,
     )
 
@@ -59,7 +65,8 @@ async def get_stats(
 @router.get("/scheduled", response_model=VolatileScheduledResponse)
 async def get_scheduled(
     user: str = Query(default=DEFAULT_USER, description="User identifier"),
-    redis: RedisDep = None,
+    qdrant: QdrantDep = None,
+    ollama: OllamaDep = None,
     api_key: str = Depends(verify_api_key)
 ):
     """
@@ -68,7 +75,7 @@ async def get_scheduled(
     Used by scheduler to determine what volatile data needs refreshing.
     Returns all records that have a refresh_schedule cron expression set.
     """
-    service = get_volatile_service(redis)
+    service = get_volatile_service(qdrant, ollama)
     records = await service.get_scheduled(user)
 
     return VolatileScheduledResponse(
@@ -86,10 +93,7 @@ async def list_namespaces(
     List available namespaces and their default TTLs.
 
     Returns predefined namespaces with their default TTL values.
-    Custom namespaces can also be used with the default TTL.
     """
-    from src.models.volatile import NAMESPACE_DEFAULT_TTL
-
     return {
         "namespaces": [
             {
@@ -120,103 +124,62 @@ def _get_namespace_description(ns: VolatileNamespace) -> str:
     return descriptions.get(ns, "Custom namespace")
 
 
-@router.get("/{namespace}", response_model=VolatileListResponse)
-async def list_keys(
-    namespace: str,
+@router.get("/search")
+async def search_volatile(
+    q: str = Query(..., min_length=1, description="Search query"),
     user: str = Query(default=DEFAULT_USER, description="User identifier"),
-    redis: RedisDep = None,
+    limit: int = Query(default=5, ge=1, le=20, description="Maximum results"),
+    threshold: float = Query(default=0.75, ge=0.5, le=1.0, description="Minimum similarity score"),
+    qdrant: QdrantDep = None,
+    ollama: OllamaDep = None,
     api_key: str = Depends(verify_api_key)
 ):
     """
-    List all keys in a namespace.
+    Semantic search across volatile data.
 
-    Returns the list of keys stored in the specified namespace.
-    """
-    service = get_volatile_service(redis)
-    keys = await service.list_namespace(user, namespace)
-
-    return VolatileListResponse(
-        namespace=namespace,
-        keys=keys,
-        count=len(keys),
-        user=user,
-    )
-
-
-@router.delete("/{namespace}", response_model=VolatileBulkDeleteResponse)
-async def delete_namespace(
-    namespace: str,
-    user: str = Query(default=DEFAULT_USER, description="User identifier"),
-    redis: RedisDep = None,
-    api_key: str = Depends(verify_api_key)
-):
-    """
-    Delete all records in a namespace.
-
-    Removes all volatile data for the specified namespace.
-    """
-    service = get_volatile_service(redis)
-    deleted = await service.delete_namespace(user, namespace)
-
-    return VolatileBulkDeleteResponse(
-        namespace=namespace,
-        deleted_count=deleted,
-        user=user,
-    )
-
-
-@router.get("/{namespace}/{key}", response_model=VolatileRecordResponse)
-async def get_record(
-    namespace: str,
-    key: str,
-    user: str = Query(default=DEFAULT_USER, description="User identifier"),
-    redis: RedisDep = None,
-    api_key: str = Depends(verify_api_key)
-):
-    """
-    Get a volatile record.
-
-    Returns the record if it exists and has not expired.
+    Searches all volatile data for semantically similar content.
+    Higher threshold = stricter matching.
 
     **Example:**
     ```
-    GET /volatile/weather/rotterdam?user=jpmschweitzer
+    GET /volatile/search?q=weather%20rotterdam&user=jpmschweitzer
     ```
     """
-    service = get_volatile_service(redis)
-    record = await service.get(user, namespace, key)
+    service = get_volatile_service(qdrant, ollama)
+    results = await service.search(user, q, limit=limit, score_threshold=threshold)
 
-    if not record:
-        raise HTTPException(
-            status_code=404,
-            detail=f"Record '{key}' not found in namespace '{namespace}'"
-        )
-
-    return record
+    return {
+        "query": q,
+        "results": results,
+        "count": len(results),
+        "user": user,
+    }
 
 
-@router.post("/{namespace}/{key}", response_model=VolatileRecordResponse)
-async def set_record(
-    namespace: str,
-    key: str,
-    request: VolatileRecordCreate,
+@router.post("/store", response_model=VolatileRecordResponse)
+async def store_volatile(
+    namespace: str = Query(..., description="Data namespace (weather, news, etc.)"),
+    key: str = Query(..., description="Record key (e.g., 'rotterdam', 'nos-headlines')"),
+    request: VolatileRecordCreate = None,
     user: str = Query(default=DEFAULT_USER, description="User identifier"),
-    redis: RedisDep = None,
+    qdrant: QdrantDep = None,
+    ollama: OllamaDep = None,
     api_key: str = Depends(verify_api_key)
 ):
     """
-    Store or update a volatile record.
+    Store volatile data.
 
-    Creates or updates a record with the specified TTL.
-    If TTL is not provided, the namespace default is used.
+    Data is converted to natural language and embedded for semantic search.
+    If the same namespace+key already exists, it will be updated.
 
     **Example Request:**
     ```json
+    POST /volatile/store?namespace=weather&key=rotterdam
     {
         "data": {
-            "temperature": 18,
-            "conditions": "Partly cloudy",
-            "humidity": 65
+            "temperature": 8,
+            "conditions": "Cloudy",
+            "humidity": 85
         },
         "source": "openweathermap",
         "ttl": 1800,
@@ -228,10 +191,21 @@ async def set_record(
     Optional cron expression for automatic refresh. The scheduler
     will query `/volatile/scheduled` and trigger refreshes.
     """
-    service = get_volatile_service(redis)
+    # Validate namespace if not custom
+    if namespace != VolatileNamespace.CUSTOM:
+        try:
+            VolatileNamespace(namespace)
+        except ValueError:
+            valid = [ns.value for ns in VolatileNamespace]
+            raise HTTPException(
+                status_code=400,
+                detail=f"Invalid namespace '{namespace}'. Valid: {valid}"
+            )
+
+    service = get_volatile_service(qdrant, ollama)
 
     try:
-        record = await service.set(
+        record = await service.store(
             user=user,
             namespace=namespace,
             key=key,
@@ -244,7 +218,36 @@ async def set_record(
 
     except Exception as e:
         logger.error(f"Failed to store volatile record: {e}")
-        raise HTTPException(status_code=500, detail="Failed to store record")
+        raise HTTPException(status_code=500, detail=f"Failed to store record: {str(e)}")
+
+
+@router.get("/{namespace}/{key}", response_model=VolatileRecordResponse)
+async def get_record(
+    namespace: str,
+    key: str,
+    user: str = Query(default=DEFAULT_USER, description="User identifier"),
+    qdrant: QdrantDep = None,
+    ollama: OllamaDep = None,
+    api_key: str = Depends(verify_api_key)
+):
+    """
+    Get a specific volatile record by namespace and key.
+
+    **Example:**
+    ```
+    GET /volatile/weather/rotterdam?user=jpmschweitzer
+    ```
+    """
+    service = get_volatile_service(qdrant, ollama)
+    record = await service.get(user, namespace, key)
+
+    if not record:
+        raise HTTPException(
+            status_code=404,
+            detail=f"Record '{key}' not found in namespace '{namespace}'"
+        )
+
+    return record
 
 
 @router.delete("/{namespace}/{key}", response_model=VolatileDeleteResponse)
@@ -252,15 +255,14 @@ async def delete_record(
     namespace: str,
     key: str,
     user: str = Query(default=DEFAULT_USER, description="User identifier"),
-    redis: RedisDep = None,
+    qdrant: QdrantDep = None,
+    ollama: OllamaDep = None,
     api_key: str = Depends(verify_api_key)
 ):
     """
-    Delete a volatile record.
-
-    Removes the record from the cache.
+    Delete a specific volatile record.
     """
-    service = get_volatile_service(redis)
+    service = get_volatile_service(qdrant, ollama)
     deleted = await service.delete(user, namespace, key)
 
     return VolatileDeleteResponse(

src/services/hybrid_rag_service.py

@@ -20,6 +20,7 @@ import logging
 
 from src.services.vector_service import VectorService
 from src.services.graph_service import GraphService
+from src.services.volatile_service import VolatileCacheService
 from src.clients.searxng_client import SearXNGClient
 from src.clients.ollama_client import OllamaClient
 from src.clients.content_extractor import ContentExtractor
@@ -46,7 +47,8 @@ class HybridRAGService:
         searxng_client: SearXNGClient,
         ollama_client: OllamaClient,
         content_extractor: ContentExtractor,
-        settings: Settings
+        settings: Settings,
+        volatile_service: Optional[VolatileCacheService] = None
     ):
         """
         Initialize HybridRAG service.
@@ -58,6 +60,7 @@ class HybridRAGService:
             ollama_client: Client for LLM (keyword extraction, re-ranking)
             content_extractor: Client for extracting full content from URLs
             settings: Application settings
+            volatile_service: Service for volatile cache search (optional)
         """
         self.vector = vector_service
         self.graph = graph_service
@@ -65,6 +68,7 @@ class HybridRAGService:
         self.ollama = ollama_client
         self.content_extractor = content_extractor
         self.settings = settings
+        self.volatile = volatile_service
         self.reranker_model = settings.ollama_model
 
     async def search(
@@ -104,8 +108,9 @@ class HybridRAGService:
         timing["vector_ms"] = raw_results.get("timing", {}).get("vector_ms", 0)
         timing["graph_ms"] = raw_results.get("timing", {}).get("graph_ms", 0)
         timing["web_ms"] = raw_results.get("timing", {}).get("web_ms", 0)
+        timing["volatile_ms"] = raw_results.get("timing", {}).get("volatile_ms", 0)
 
-        # Phase 2: Two-Stage RRF Fusion
+        # Phase 2: Three-Source RRF Fusion
         phase2_start = time.time()
 
         # Stage 1: Merge wiki sources (vector + graph) into single ranking
@@ -115,10 +120,12 @@ class HybridRAGService:
             k=config.rrf_k
         )
 
-        # Stage 2: Final RRF between wiki and web (equal footing)
+        # Stage 2: Final RRF between wiki, volatile, and web
+        # Volatile gets priority boost (smaller k = higher contribution per rank)
         fused_results = self._reciprocal_rank_fusion(
             wiki_results=wiki_merged,
             web_results=raw_results.get("web", []),
+            volatile_results=raw_results.get("volatile", []),
             k=config.rrf_k
         )
         timing["fusion_ms"] = (time.time() - phase2_start) * 1000
@@ -389,6 +396,37 @@ JSON:"""
 
             tasks["web"] = web_search()
 
+        # Volatile cache search
+        if config.enable_volatile and self.volatile:
+            async def volatile_search():
+                start = time.time()
+                try:
+                    results = await self.volatile.search(
+                        user=user,
+                        query=query,
+                        limit=config.volatile_limit,
+                        score_threshold=config.volatile_threshold
+                    )
+                    formatted = [
+                        {
+                            "key": r.key,
+                            "namespace": r.namespace,
+                            "title": f"{r.namespace}: {r.key}",
+                            "content": r.data.get("text", "") if isinstance(r.data, dict) else str(r.data),
+                            "raw_data": r.data,
+                            "source_api": r.source,
+                            "ttl_remaining": r.ttl_remaining,
+                            "source": "volatile"
+                        }
+                        for r in results
+                    ]
+                    return formatted, (time.time() - start) * 1000
+                except Exception as e:
+                    logger.error(f"Volatile search failed: {e}", exc_info=True)
+                    return [], (time.time() - start) * 1000
+
+            tasks["volatile"] = volatile_search()
+
         # Execute all searches in parallel
         results_dict = await asyncio.gather(*tasks.values())
 
@@ -401,7 +439,8 @@ JSON:"""
 
         logger.info(
             f"Parallel retrieval: vector={len(output.get('vector', []))}, "
-            f"graph={len(output.get('graph', []))}, web={len(output.get('web', []))}"
+            f"graph={len(output.get('graph', []))}, web={len(output.get('web', []))}, "
+            f"volatile={len(output.get('volatile', []))}"
         )
 
         return output
@@ -491,23 +530,42 @@ JSON:"""
         self,
         wiki_results: List[Dict],
         web_results: List[Dict],
+        volatile_results: Optional[List[Dict]] = None,
         k: int = 60
     ) -> List[Dict[str, Any]]:
         """
-        Stage 2: Final RRF between wiki (single source) and web.
+        Stage 2: Final RRF between wiki, volatile, and web.
 
-        Wiki results are pre-merged from vector+graph, so wiki and web
-        now compete on equal footing.
+        Wiki results are pre-merged from vector+graph. Volatile results
+        get a priority boost (smaller effective k) since they represent
+        current, time-sensitive information.
 
         Args:
             wiki_results: Pre-merged wiki results from _merge_wiki_sources()
             web_results: Results from web search
+            volatile_results: Results from volatile cache (fresh data)
             k: RRF constant (default 60)
 
         Returns:
             Final merged and sorted results
         """
         rrf_scores = {}
+        volatile_results = volatile_results or []
+
+        # Volatile results get priority boost (k/2 = stronger score per rank)
+        volatile_k = k // 2
+        for rank, result in enumerate(volatile_results, start=1):
+            key = result.get("key")
+            namespace = result.get("namespace", "unknown")
+            if not key:
+                continue
+            result_id = f"volatile_{namespace}_{key}"
+            rrf_scores[result_id] = {
+                "result": result,
+                "rrf_score": 1 / (volatile_k + rank),  # Priority boost
+                "sources": ["volatile"],
+                "source_type": "volatile"
+            }
 
         # Wiki results (single source, already merged)
         for rank, result in enumerate(wiki_results, start=1):
@@ -542,7 +600,8 @@ JSON:"""
             reverse=True
         )
 
-        logger.info(f"Final RRF: {len(sorted_results)} results (wiki + web)")
+        volatile_count = len([r for r in sorted_results if r["source_type"] == "volatile"])
+        logger.info(f"Final RRF: {len(sorted_results)} results (wiki + volatile[{volatile_count}] + web)")
 
         return sorted_results
 

src/services/volatile_service.py

@@ -1,23 +1,24 @@
 """
 Volatile Cache service for Library Desk.
 
-Provides ephemeral data storage with TTL for time-sensitive information:
+Provides ephemeral data storage with TTL using Qdrant vectors:
 - Weather, news, financial data
 - Transit schedules, traffic conditions
 - System status, social notifications
+
+Data is stored as embedded vectors for semantic search retrieval.
 """
 
-import json
-import logging
 import hashlib
+import logging
+import time
 from datetime import datetime
 from typing import List, Optional, Dict, Any
 
-import redis.asyncio as aioredis
-
+from src.clients.qdrant_client import QdrantClientWrapper
+from src.clients.ollama_client import OllamaClient
 from src.config import Settings
 from src.models.volatile import (
-    VolatileRecord,
     VolatileRecordResponse,
     VolatileNamespace,
     NAMESPACE_DEFAULT_TTL,
@@ -28,47 +29,46 @@ logger = logging.getLogger(__name__)
 
 class VolatileCacheService:
     """
-    Service for volatile data with TTL.
+    Service for volatile data with TTL stored in Qdrant.
 
-    Stores ephemeral data in Redis with automatic expiration.
-    Supports multiple namespaces with configurable TTLs.
+    Stores ephemeral data as vectors for semantic search retrieval.
+    Each user has an isolated volatile collection.
     """
 
-    # Redis key prefix for volatile data
-    KEY_PREFIX = "volatile"
+    COLLECTION_PREFIX = "volatile_"
 
     def __init__(
         self,
-        redis_client: aioredis.Redis,
+        qdrant_client: QdrantClientWrapper,
+        ollama_client: OllamaClient,
         settings: Settings
     ):
         """
         Initialize volatile cache service.
 
         Args:
-            redis_client: Async Redis client
+            qdrant_client: Qdrant client for vector storage
+            ollama_client: Ollama client for embeddings
             settings: Application settings
         """
-        self.redis = redis_client
+        self.qdrant = qdrant_client
+        self.ollama = ollama_client
         self.settings = settings
 
-        logger.info("Initialized VolatileCacheService")
+        logger.info("Initialized VolatileCacheService (Qdrant backend)")
 
-    def _build_key(self, user: str, namespace: str, key: str) -> str:
+    def _collection_name(self, user: str) -> str:
+        """Get volatile collection name for user."""
+        return f"{self.COLLECTION_PREFIX}{user}"
+
+    def _make_vector_id(self, namespace: str, key: str) -> str:
         """
-        Build Redis key for volatile record.
+        Generate deterministic vector ID for namespace/key.
 
-        Pattern: {user}:volatile:{namespace}:{key_hash}
-        Uses hash to ensure safe key characters and consistent length.
+        Same namespace+key always produces same ID for upsert behavior.
         """
-        key_hash = hashlib.md5(key.encode()).hexdigest()[:12]
-        return f"{user}:{self.KEY_PREFIX}:{namespace}:{key_hash}"
-
-    def _build_pattern(self, user: str, namespace: Optional[str] = None) -> str:
-        """Build pattern for key scanning."""
-        if namespace:
-            return f"{user}:{self.KEY_PREFIX}:{namespace}:*"
-        return f"{user}:{self.KEY_PREFIX}:*"
+        combined = f"{namespace}:{key}"
+        return hashlib.md5(combined.encode()).hexdigest()
 
     def _get_default_ttl(self, namespace: str) -> int:
         """Get default TTL for a namespace."""
@@ -78,84 +78,116 @@ class VolatileCacheService:
         except ValueError:
             return self.settings.volatile_default_ttl
 
-    def _serialize_record(self, record: VolatileRecord) -> str:
-        """Serialize record to JSON for storage."""
-        return json.dumps({
-            "key": record.key,
-            "namespace": record.namespace,
-            "data": record.data,
-            "source": record.source,
-            "created_at": record.created_at.isoformat(),
-            "updated_at": record.updated_at.isoformat(),
-            "ttl": record.ttl,
-            "refresh_schedule": record.refresh_schedule,
-            "user": record.user,
-        })
+    def _current_timestamp_ms(self) -> int:
+        """Get current timestamp in milliseconds."""
+        return int(time.time() * 1000)
 
-    def _deserialize_record(self, data: str) -> VolatileRecord:
-        """Deserialize record from JSON."""
-        obj = json.loads(data)
-        return VolatileRecord(
-            key=obj["key"],
-            namespace=obj["namespace"],
-            data=obj["data"],
-            source=obj.get("source"),
-            created_at=datetime.fromisoformat(obj["created_at"]),
-            updated_at=datetime.fromisoformat(obj["updated_at"]),
-            ttl=obj["ttl"],
-            refresh_schedule=obj.get("refresh_schedule"),
-            user=obj["user"],
-        )
-
-    async def get(
+    def _to_natural_language(
         self,
-        user: str,
         namespace: str,
-        key: str
-    ) -> Optional[VolatileRecordResponse]:
+        key: str,
+        data: Dict[str, Any]
+    ) -> str:
         """
-        Get a volatile record.
+        Convert structured data to natural language for embedding.
 
-        Args:
-            user: User identifier
-            namespace: Data namespace
-            key: Record key
-
-        Returns:
-            Record if found and not expired, None otherwise
+        This creates a text representation that embeds well semantically.
         """
-        redis_key = self._build_key(user, namespace, key)
+        # Template-based conversion for known namespaces
+        if namespace == VolatileNamespace.WEATHER:
+            temp = data.get("temperature", data.get("temp", "unknown"))
+            conditions = data.get("conditions", data.get("weather", ""))
+            humidity = data.get("humidity", "")
+            text = f"Current weather in {key}: {temp}°C"
+            if conditions:
+                text += f", {conditions}"
+            if humidity:
+                text += f", humidity {humidity}%"
+            return text
 
-        try:
-            data = await self.redis.get(redis_key)
-            if not data:
-                return None
+        elif namespace == VolatileNamespace.NEWS:
+            title = data.get("title", data.get("headline", ""))
+            summary = data.get("summary", data.get("description", ""))
+            source = data.get("source", "")
+            text = f"News: {title}"
+            if summary:
+                text += f". {summary}"
+            if source:
+                text += f" (Source: {source})"
+            return text
 
-            record = self._deserialize_record(data)
+        elif namespace == VolatileNamespace.FINANCIAL:
+            symbol = data.get("symbol", key)
+            price = data.get("price", "")
+            change = data.get("change", data.get("change_percent", ""))
+            text = f"Financial data for {symbol}"
+            if price:
+                text += f": price {price}"
+            if change:
+                text += f", change {change}%"
+            return text
 
-            # Get TTL remaining
-            ttl_remaining = await self.redis.ttl(redis_key)
-            if ttl_remaining < 0:
-                return None
+        elif namespace == VolatileNamespace.TRANSIT:
+            route = data.get("route", data.get("line", key))
+            status = data.get("status", "")
+            delay = data.get("delay", data.get("delay_minutes", ""))
+            text = f"Transit {route}"
+            if status:
+                text += f": {status}"
+            if delay:
+                text += f", delay {delay} minutes"
+            return text
 
-            return VolatileRecordResponse(
-                key=record.key,
-                namespace=record.namespace,
-                data=record.data,
-                source=record.source,
-                created_at=record.created_at,
-                updated_at=record.updated_at,
-                ttl=record.ttl,
-                ttl_remaining=max(0, ttl_remaining),
-                refresh_schedule=record.refresh_schedule,
-                user=record.user,
-            )
+        elif namespace == VolatileNamespace.TRAFFIC:
+            location = data.get("location", key)
+            duration = data.get("duration", data.get("travel_time", ""))
+            congestion = data.get("congestion", "")
+            text = f"Traffic for {location}"
+            if duration:
+                text += f": {duration} minutes"
+            if congestion:
+                text += f", congestion level {congestion}"
+            return text
 
-        except Exception as e:
-            logger.error(f"Failed to get volatile record {redis_key}: {e}")
-            return None
+        elif namespace == VolatileNamespace.AIR_QUALITY:
+            location = data.get("location", key)
+            aqi = data.get("aqi", data.get("index", ""))
+            quality = data.get("quality", "")
+            text = f"Air quality in {location}"
+            if aqi:
+                text += f": AQI {aqi}"
+            if quality:
+                text += f" ({quality})"
+            return text
 
-    async def set(
+        elif namespace == VolatileNamespace.SPORTS:
+            event = data.get("event", data.get("match", key))
+            score = data.get("score", "")
+            status = data.get("status", "")
+            text = f"Sports: {event}"
+            if score:
+                text += f" - Score: {score}"
+            if status:
+                text += f" ({status})"
+            return text
+
+        elif namespace == VolatileNamespace.SYSTEM:
+            service = data.get("service", key)
+            status = data.get("status", "unknown")
+            message = data.get("message", "")
+            text = f"System status for {service}: {status}"
+            if message:
+                text += f". {message}"
+            return text
+
+        # Fallback: serialize key fields
+        text_parts = [f"{namespace} data for {key}:"]
+        for k, v in data.items():
+            if isinstance(v, (str, int, float, bool)):
+                text_parts.append(f"{k}: {v}")
+        return " ".join(text_parts)
+
+    async def store(
         self,
         user: str,
         namespace: str,
@@ -166,13 +198,13 @@ class VolatileCacheService:
         refresh_schedule: Optional[str] = None
     ) -> VolatileRecordResponse:
         """
-        Store or update a volatile record.
+        Store volatile data as an embedded vector.
 
         Args:
             user: User identifier
-            namespace: Data namespace
-            key: Record key
-            data: Content to store
+            namespace: Data namespace (from controlled list)
+            key: Record key (normalized slug)
+            data: Structured data to store
             source: Origin API/service
             ttl: TTL in seconds (uses namespace default if not set)
             refresh_schedule: Optional cron expression for refresh
@@ -180,49 +212,158 @@ class VolatileCacheService:
         Returns:
             The stored record
         """
-        redis_key = self._build_key(user, namespace, key)
+        collection = self._collection_name(user)
 
-        # Use provided TTL or namespace default
+        # Ensure collection exists
+        await self.qdrant.ensure_collection(collection)
+
+        # Calculate TTL and expiry
         effective_ttl = ttl if ttl is not None else self._get_default_ttl(namespace)
+        now_ms = self._current_timestamp_ms()
+        expiry_ms = now_ms + (effective_ttl * 1000)
 
-        # Check if record exists (for created_at)
-        existing = await self.get(user, namespace, key)
+        # Convert to natural language for embedding
+        text = self._to_natural_language(namespace, key, data)
+
+        # Generate embedding
+        embedding = await self.ollama.embed(text)
+        if not embedding:
+            raise ValueError("Failed to generate embedding for volatile data")
+
+        # Build payload
         now = datetime.utcnow()
+        payload = {
+            "doc_type": "volatile",
+            "namespace": namespace,
+            "key": key,
+            "text": text,
+            "raw_data": data,
+            "source": source,
+            "created_at": now.isoformat(),
+            "updated_at": now.isoformat(),
+            "ttl": effective_ttl,
+            "ttl_expiry": expiry_ms,
+            "refresh_schedule": refresh_schedule,
+            "user": user,
+        }
 
-        record = VolatileRecord(
+        # Upsert vector (same namespace+key = same ID = update)
+        vector_id = self._make_vector_id(namespace, key)
+        success = await self.qdrant.upsert_vector(
+            collection_name=collection,
+            vector_id=vector_id,
+            vector=embedding,
+            payload=payload
+        )
+
+        if not success:
+            raise ValueError("Failed to store volatile vector")
+
+        logger.debug(f"Stored volatile {namespace}:{key} with TTL {effective_ttl}s")
+
+        return VolatileRecordResponse(
             key=key,
             namespace=namespace,
             data=data,
             source=source,
-            created_at=existing.created_at if existing else now,
+            created_at=now,
             updated_at=now,
             ttl=effective_ttl,
+            ttl_remaining=effective_ttl,
             refresh_schedule=refresh_schedule,
             user=user,
         )
 
-        try:
-            serialized = self._serialize_record(record)
-            await self.redis.setex(redis_key, effective_ttl, serialized)
+    async def search(
+        self,
+        user: str,
+        query: str,
+        limit: int = 5,
+        score_threshold: float = 0.75
+    ) -> List[VolatileRecordResponse]:
+        """
+        Semantic search across volatile data.
 
-            logger.debug(f"Stored volatile record {redis_key} with TTL {effective_ttl}s")
+        Args:
+            user: User identifier
+            query: Search query
+            limit: Maximum results
+            score_threshold: Minimum similarity score (higher = stricter)
 
-            return VolatileRecordResponse(
-                key=record.key,
-                namespace=record.namespace,
-                data=record.data,
-                source=record.source,
-                created_at=record.created_at,
-                updated_at=record.updated_at,
-                ttl=record.ttl,
-                ttl_remaining=effective_ttl,
-                refresh_schedule=record.refresh_schedule,
-                user=record.user,
-            )
+        Returns:
+            List of matching volatile records
+        """
+        collection = self._collection_name(user)
 
-        except Exception as e:
-            logger.error(f"Failed to store volatile record {redis_key}: {e}")
-            raise
+        # Check if collection exists
+        if not await self.qdrant.collection_exists(collection):
+            return []
+
+        # Generate query embedding
+        query_embedding = await self.ollama.embed(query)
+        if not query_embedding:
+            logger.error("Failed to embed query for volatile search")
+            return []
+
+        # Search with expiry filter
+        now_ms = self._current_timestamp_ms()
+        results = await self.qdrant.search_with_expiry_filter(
+            collection_name=collection,
+            query_vector=query_embedding,
+            current_timestamp=now_ms,
+            limit=limit,
+            score_threshold=score_threshold
+        )
+
+        # Convert to response models
+        responses = []
+        for result in results:
+            payload = result["payload"]
+            ttl_expiry = payload.get("ttl_expiry", 0)
+            ttl_remaining = max(0, (ttl_expiry - now_ms) // 1000)
+
+            responses.append(VolatileRecordResponse(
+                key=payload["key"],
+                namespace=payload["namespace"],
+                data=payload.get("raw_data", {}),
+                source=payload.get("source"),
+                created_at=datetime.fromisoformat(payload["created_at"]),
+                updated_at=datetime.fromisoformat(payload["updated_at"]),
+                ttl=payload.get("ttl", 0),
+                ttl_remaining=ttl_remaining,
+                refresh_schedule=payload.get("refresh_schedule"),
+                user=payload["user"],
+            ))
+
+        return responses
+
+    async def get(
+        self,
+        user: str,
+        namespace: str,
+        key: str
+    ) -> Optional[VolatileRecordResponse]:
+        """
+        Get a specific volatile record by namespace and key.
+
+        Args:
+            user: User identifier
+            namespace: Data namespace
+            key: Record key
+
+        Returns:
+            Record if found and not expired, None otherwise
+        """
+        # Use search with high threshold to find exact match
+        query = self._to_natural_language(namespace, key, {"key": key})
+        results = await self.search(user, query, limit=10, score_threshold=0.5)
+
+        # Find exact namespace+key match
+        for result in results:
+            if result.namespace == namespace and result.key == key:
+                return result
+
+        return None
 
     async def delete(
         self,
@@ -231,7 +372,7 @@ class VolatileCacheService:
         key: str
     ) -> bool:
         """
-        Delete a volatile record.
+        Delete a specific volatile record.
 
         Args:
             user: User identifier
@@ -239,51 +380,24 @@ class VolatileCacheService:
             key: Record key
 
         Returns:
-            True if record was deleted, False if not found
+            True if deleted, False if not found
         """
-        redis_key = self._build_key(user, namespace, key)
+        collection = self._collection_name(user)
 
-        try:
-            deleted = await self.redis.delete(redis_key)
-            if deleted:
-                logger.debug(f"Deleted volatile record {redis_key}")
-            return deleted > 0
-
-        except Exception as e:
-            logger.error(f"Failed to delete volatile record {redis_key}: {e}")
+        if not await self.qdrant.collection_exists(collection):
             return False
 
-    async def list_namespace(
-        self,
-        user: str,
-        namespace: str
-    ) -> List[str]:
-        """
-        List all keys in a namespace.
-
-        Args:
-            user: User identifier
-            namespace: Data namespace
-
-        Returns:
-            List of keys (original keys, not Redis keys)
-        """
-        pattern = self._build_pattern(user, namespace)
+        vector_id = self._make_vector_id(namespace, key)
 
         try:
-            keys = []
-            async for redis_key in self.redis.scan_iter(match=pattern):
-                # Get the record to retrieve original key
-                data = await self.redis.get(redis_key)
-                if data:
-                    record = self._deserialize_record(data)
-                    keys.append(record.key)
-
-            return keys
-
+            deleted = await self.qdrant.delete_by_ids(
+                collection_name=collection,
+                point_ids=[vector_id]
+            )
+            return deleted > 0
         except Exception as e:
-            logger.error(f"Failed to list namespace {namespace}: {e}")
-            return []
+            logger.error(f"Failed to delete volatile {namespace}:{key}: {e}")
+            return False
 
     async def get_scheduled(
         self,
@@ -300,33 +414,49 @@ class VolatileCacheService:
         Returns:
             List of records with refresh_schedule set
         """
-        pattern = self._build_pattern(user)
+        collection = self._collection_name(user)
 
+        if not await self.qdrant.collection_exists(collection):
+            return []
+
+        now_ms = self._current_timestamp_ms()
+        scheduled = []
+
+        # Scroll through all non-expired records
         try:
-            scheduled = []
-            async for redis_key in self.redis.scan_iter(match=pattern):
-                data = await self.redis.get(redis_key)
-                if data:
-                    record = self._deserialize_record(data)
-                    if record.refresh_schedule:
-                        ttl_remaining = await self.redis.ttl(redis_key)
-                        scheduled.append(VolatileRecordResponse(
-                            key=record.key,
-                            namespace=record.namespace,
-                            data=record.data,
-                            source=record.source,
-                            created_at=record.created_at,
-                            updated_at=record.updated_at,
-                            ttl=record.ttl,
-                            ttl_remaining=max(0, ttl_remaining),
-                            refresh_schedule=record.refresh_schedule,
-                            user=record.user,
-                        ))
+            all_points = await self.qdrant.scroll_all_points(
+                collection_name=collection,
+                with_payload=True
+            )
+
+            for point in all_points:
+                payload = point.get("payload", {})
+                ttl_expiry = payload.get("ttl_expiry", 0)
+
+                # Skip expired
+                if ttl_expiry <= now_ms:
+                    continue
+
+                # Only include if has refresh schedule
+                if payload.get("refresh_schedule"):
+                    ttl_remaining = max(0, (ttl_expiry - now_ms) // 1000)
+                    scheduled.append(VolatileRecordResponse(
+                        key=payload["key"],
+                        namespace=payload["namespace"],
+                        data=payload.get("raw_data", {}),
+                        source=payload.get("source"),
+                        created_at=datetime.fromisoformat(payload["created_at"]),
+                        updated_at=datetime.fromisoformat(payload["updated_at"]),
+                        ttl=payload.get("ttl", 0),
+                        ttl_remaining=ttl_remaining,
+                        refresh_schedule=payload["refresh_schedule"],
+                        user=payload["user"],
+                    ))
 
             return scheduled
 
         except Exception as e:
-            logger.error(f"Failed to get scheduled records: {e}")
+            logger.error(f"Failed to get scheduled volatile records: {e}")
             return []
 
     async def get_stats(
@@ -342,92 +472,93 @@ class VolatileCacheService:
         Returns:
             Statistics dict
         """
-        pattern = self._build_pattern(user)
+        collection = self._collection_name(user)
+
+        if not await self.qdrant.collection_exists(collection):
+            return {
+                "total_records": 0,
+                "by_namespace": {},
+                "scheduled_count": 0,
+                "expired_count": 0,
+            }
+
+        now_ms = self._current_timestamp_ms()
+        by_namespace: Dict[str, int] = {}
+        total = 0
+        scheduled = 0
+        expired = 0
 
         try:
-            by_namespace: Dict[str, int] = {}
-            total = 0
-            scheduled = 0
+            all_points = await self.qdrant.scroll_all_points(
+                collection_name=collection,
+                with_payload=True
+            )
 
-            async for redis_key in self.redis.scan_iter(match=pattern):
-                data = await self.redis.get(redis_key)
-                if data:
-                    record = self._deserialize_record(data)
+            for point in all_points:
+                payload = point.get("payload", {})
+                namespace = payload.get("namespace", "unknown")
+                ttl_expiry = payload.get("ttl_expiry", 0)
+
+                if ttl_expiry <= now_ms:
+                    expired += 1
+                else:
                     total += 1
-                    by_namespace[record.namespace] = by_namespace.get(record.namespace, 0) + 1
-                    if record.refresh_schedule:
+                    by_namespace[namespace] = by_namespace.get(namespace, 0) + 1
+                    if payload.get("refresh_schedule"):
                         scheduled += 1
 
             return {
                 "total_records": total,
                 "by_namespace": by_namespace,
                 "scheduled_count": scheduled,
-                "total_memory_bytes": None,  # Could implement with DEBUG MEMORY
+                "expired_count": expired,
             }
 
         except Exception as e:
-            logger.error(f"Failed to get stats: {e}")
+            logger.error(f"Failed to get volatile stats: {e}")
             return {
                 "total_records": 0,
                 "by_namespace": {},
                 "scheduled_count": 0,
-                "total_memory_bytes": None,
+                "expired_count": 0,
             }
 
-    async def delete_namespace(
-        self,
-        user: str,
-        namespace: str
-    ) -> int:
-        """
-        Delete all records in a namespace.
-
-        Args:
-            user: User identifier
-            namespace: Data namespace
-
-        Returns:
-            Number of records deleted
-        """
-        pattern = self._build_pattern(user, namespace)
-
-        try:
-            deleted = 0
-            async for redis_key in self.redis.scan_iter(match=pattern):
-                await self.redis.delete(redis_key)
-                deleted += 1
-
-            logger.info(f"Deleted {deleted} records from namespace {namespace}")
-            return deleted
-
-        except Exception as e:
-            logger.error(f"Failed to delete namespace {namespace}: {e}")
-            return 0
-
-    async def delete_all(
+    async def purge_expired(
         self,
         user: str
     ) -> int:
         """
-        Delete all volatile records for user.
+        Purge all expired volatile records for user.
 
         Args:
             user: User identifier
 
         Returns:
-            Number of records deleted
+            Number of records purged
         """
-        pattern = self._build_pattern(user)
+        collection = self._collection_name(user)
 
-        try:
-            deleted = 0
-            async for redis_key in self.redis.scan_iter(match=pattern):
-                await self.redis.delete(redis_key)
-                deleted += 1
-
-            logger.info(f"Deleted all {deleted} volatile records for user {user}")
-            return deleted
-
-        except Exception as e:
-            logger.error(f"Failed to delete all records: {e}")
+        if not await self.qdrant.collection_exists(collection):
             return 0
+
+        now_ms = self._current_timestamp_ms()
+        return await self.qdrant.delete_expired_vectors(collection, now_ms)
+
+    async def purge_all_expired(self) -> Dict[str, int]:
+        """
+        Purge expired records from all volatile collections.
+
+        Returns:
+            Dict of collection -> purged count
+        """
+        collections = await self.qdrant.get_volatile_collections()
+        results = {}
+        now_ms = self._current_timestamp_ms()
+
+        for collection in collections:
+            purged = await self.qdrant.delete_expired_vectors(collection, now_ms)
+            if purged > 0:
+                results[collection] = purged
+                logger.info(f"Purged {purged} expired from {collection}")
+
+        return results

tests/test_volatile.py

@@ -1,11 +1,13 @@
 """
-Tests for volatile cache router and service.
+Tests for volatile cache router and service (Qdrant backend).
 
 Tests:
 - Volatile record CRUD operations
 - Namespace listing and management
 - Scheduled record retrieval
-- TTL behavior
+- TTL behavior and expiry filtering
+- Semantic search
+- Natural language conversion
 """
 
 import pytest
@@ -106,6 +108,10 @@ class TestVolatileNamespaces:
         """Test sports namespace default TTL (fast updates)."""
         assert NAMESPACE_DEFAULT_TTL[VolatileNamespace.SPORTS] == 60  # 1 min
 
+    def test_namespace_count(self):
+        """Test we have the expected number of namespaces."""
+        assert len(VolatileNamespace) == 11
+
 
 class TestVolatileListResponse:
     """Test list response models."""
@@ -210,18 +216,28 @@ class TestVolatileDeleteResponses:
 
 
 class TestVolatileService:
-    """Test VolatileCacheService functionality."""
+    """Test VolatileCacheService functionality (Qdrant backend)."""
 
     @pytest.fixture
-    def mock_redis(self):
-        """Create mock Redis client."""
-        redis = AsyncMock()
-        redis.get = AsyncMock(return_value=None)
-        redis.setex = AsyncMock()
-        redis.delete = AsyncMock(return_value=1)
-        redis.ttl = AsyncMock(return_value=1500)
-        redis.scan_iter = MagicMock(return_value=iter([]))
-        return redis
+    def mock_qdrant(self):
+        """Create mock Qdrant client."""
+        qdrant = AsyncMock()
+        qdrant.ensure_collection = AsyncMock()
+        qdrant.collection_exists = AsyncMock(return_value=True)
+        qdrant.upsert_vector = AsyncMock(return_value=True)
+        qdrant.delete_by_ids = AsyncMock(return_value=1)
+        qdrant.search_with_expiry_filter = AsyncMock(return_value=[])
+        qdrant.scroll_all_points = AsyncMock(return_value=[])
+        qdrant.delete_expired_vectors = AsyncMock(return_value=0)
+        qdrant.get_volatile_collections = AsyncMock(return_value=[])
+        return qdrant
+
+    @pytest.fixture
+    def mock_ollama(self):
+        """Create mock Ollama client."""
+        ollama = AsyncMock()
+        ollama.embed = AsyncMock(return_value=[0.1] * 768)  # Return 768-dim embedding
+        return ollama
 
     @pytest.fixture
     def mock_settings(self):
@@ -231,29 +247,29 @@ class TestVolatileService:
         return settings
 
     @pytest.fixture
-    def volatile_service(self, mock_redis, mock_settings):
+    def volatile_service(self, mock_qdrant, mock_ollama, mock_settings):
         """Create VolatileCacheService with mocks."""
         from src.services.volatile_service import VolatileCacheService
         return VolatileCacheService(
-            redis_client=mock_redis,
+            qdrant_client=mock_qdrant,
+            ollama_client=mock_ollama,
             settings=mock_settings
         )
 
-    def test_build_key(self, volatile_service):
-        """Test Redis key building."""
-        key = volatile_service._build_key("jpmschweitzer", "weather", "rotterdam")
-        assert key.startswith("jpmschweitzer:volatile:weather:")
-        assert len(key) > 30  # Has hash suffix
+    def test_collection_name(self, volatile_service):
+        """Test collection naming pattern."""
+        name = volatile_service._collection_name("jpmschweitzer")
+        assert name == "volatile_jpmschweitzer"
 
-    def test_build_pattern(self, volatile_service):
-        """Test pattern building."""
-        pattern = volatile_service._build_pattern("jpmschweitzer", "weather")
-        assert pattern == "jpmschweitzer:volatile:weather:*"
+    def test_make_vector_id(self, volatile_service):
+        """Test deterministic vector ID generation."""
+        id1 = volatile_service._make_vector_id("weather", "rotterdam")
+        id2 = volatile_service._make_vector_id("weather", "rotterdam")
+        id3 = volatile_service._make_vector_id("weather", "amsterdam")
 
-    def test_build_pattern_all(self, volatile_service):
-        """Test pattern building for all namespaces."""
-        pattern = volatile_service._build_pattern("jpmschweitzer")
-        assert pattern == "jpmschweitzer:volatile:*"
+        assert id1 == id2  # Same namespace+key = same ID
+        assert id1 != id3  # Different key = different ID
+        assert len(id1) == 32  # MD5 hex length
 
     def test_get_default_ttl_known_namespace(self, volatile_service):
         """Test default TTL for known namespace."""
@@ -265,23 +281,287 @@ class TestVolatileService:
         ttl = volatile_service._get_default_ttl("unknown_namespace")
         assert ttl == 3600  # Falls back to settings default
 
-    @pytest.mark.asyncio
-    async def test_get_not_found(self, volatile_service, mock_redis):
-        """Test get when record not found."""
-        mock_redis.get.return_value = None
-        result = await volatile_service.get("jpmschweitzer", "weather", "rotterdam")
-        assert result is None
+    def test_to_natural_language_weather(self, volatile_service):
+        """Test natural language conversion for weather data."""
+        text = volatile_service._to_natural_language(
+            namespace="weather",
+            key="rotterdam",
+            data={"temperature": 18, "conditions": "Cloudy", "humidity": 75}
+        )
+        assert "rotterdam" in text.lower()
+        assert "18" in text
+        assert "Cloudy" in text
+        assert "75" in text
+
+    def test_to_natural_language_news(self, volatile_service):
+        """Test natural language conversion for news data."""
+        text = volatile_service._to_natural_language(
+            namespace="news",
+            key="nos-headlines",
+            data={"title": "Breaking News", "summary": "Something happened", "source": "NOS"}
+        )
+        assert "Breaking News" in text
+        assert "Something happened" in text
+        assert "NOS" in text
+
+    def test_to_natural_language_financial(self, volatile_service):
+        """Test natural language conversion for financial data."""
+        text = volatile_service._to_natural_language(
+            namespace="financial",
+            key="AAPL",
+            data={"symbol": "AAPL", "price": 150.50, "change": 2.3}
+        )
+        assert "AAPL" in text
+        assert "price" in text.lower()
+        assert "change" in text.lower()
+
+    def test_to_natural_language_transit(self, volatile_service):
+        """Test natural language conversion for transit data."""
+        text = volatile_service._to_natural_language(
+            namespace="transit",
+            key="ns-intercity",
+            data={"route": "Amsterdam-Rotterdam", "status": "On time", "delay": 0}
+        )
+        assert "Amsterdam-Rotterdam" in text or "ns-intercity" in text.lower()
+        assert "On time" in text
+
+    def test_to_natural_language_fallback(self, volatile_service):
+        """Test natural language fallback for unknown namespace."""
+        text = volatile_service._to_natural_language(
+            namespace="custom",
+            key="test-key",
+            data={"foo": "bar", "count": 42}
+        )
+        assert "custom" in text.lower()
+        assert "foo" in text or "bar" in text
 
     @pytest.mark.asyncio
-    async def test_delete_success(self, volatile_service, mock_redis):
+    async def test_store_success(self, volatile_service, mock_qdrant, mock_ollama):
+        """Test successful store operation."""
+        result = await volatile_service.store(
+            user="jpmschweitzer",
+            namespace="weather",
+            key="rotterdam",
+            data={"temperature": 18, "conditions": "Sunny"},
+            source="openweathermap",
+            ttl=1800
+        )
+
+        assert result.key == "rotterdam"
+        assert result.namespace == "weather"
+        assert result.ttl == 1800
+        mock_qdrant.ensure_collection.assert_called_once()
+        mock_ollama.embed.assert_called_once()
+        mock_qdrant.upsert_vector.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_store_uses_namespace_default_ttl(self, volatile_service, mock_qdrant, mock_ollama):
+        """Test store uses namespace default TTL when not specified."""
+        result = await volatile_service.store(
+            user="jpmschweitzer",
+            namespace="weather",
+            key="amsterdam",
+            data={"temperature": 16},
+            source="openweathermap",
+            ttl=None  # Not specified
+        )
+
+        assert result.ttl == 1800  # Weather default
+
+    @pytest.mark.asyncio
+    async def test_search_empty_collection(self, volatile_service, mock_qdrant, mock_ollama):
+        """Test search when collection doesn't exist."""
+        mock_qdrant.collection_exists.return_value = False
+
+        results = await volatile_service.search(
+            user="jpmschweitzer",
+            query="weather rotterdam"
+        )
+
+        assert results == []
+        mock_ollama.embed.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_search_with_results(self, volatile_service, mock_qdrant, mock_ollama):
+        """Test search returns results."""
+        import time
+        now_ms = int(time.time() * 1000)
+
+        mock_qdrant.search_with_expiry_filter.return_value = [
+            {
+                "score": 0.95,
+                "payload": {
+                    "key": "rotterdam",
+                    "namespace": "weather",
+                    "raw_data": {"temperature": 18},
+                    "source": "openweathermap",
+                    "created_at": datetime.utcnow().isoformat(),
+                    "updated_at": datetime.utcnow().isoformat(),
+                    "ttl": 1800,
+                    "ttl_expiry": now_ms + 900000,  # 15 min remaining
+                    "refresh_schedule": None,
+                    "user": "jpmschweitzer"
+                }
+            }
+        ]
+
+        results = await volatile_service.search(
+            user="jpmschweitzer",
+            query="weather rotterdam"
+        )
+
+        assert len(results) == 1
+        assert results[0].key == "rotterdam"
+        assert results[0].namespace == "weather"
+
+    @pytest.mark.asyncio
+    async def test_delete_success(self, volatile_service, mock_qdrant):
         """Test successful delete."""
-        mock_redis.delete.return_value = 1
+        mock_qdrant.delete_by_ids.return_value = 1
+
         result = await volatile_service.delete("jpmschweitzer", "weather", "rotterdam")
+
         assert result is True
+        mock_qdrant.delete_by_ids.assert_called_once()
 
     @pytest.mark.asyncio
-    async def test_delete_not_found(self, volatile_service, mock_redis):
+    async def test_delete_not_found(self, volatile_service, mock_qdrant):
         """Test delete when record not found."""
-        mock_redis.delete.return_value = 0
+        mock_qdrant.delete_by_ids.return_value = 0
+
         result = await volatile_service.delete("jpmschweitzer", "weather", "nonexistent")
+
         assert result is False
+
+    @pytest.mark.asyncio
+    async def test_get_stats_empty(self, volatile_service, mock_qdrant):
+        """Test stats with no records."""
+        mock_qdrant.collection_exists.return_value = False
+
+        stats = await volatile_service.get_stats("jpmschweitzer")
+
+        assert stats["total_records"] == 0
+        assert stats["by_namespace"] == {}
+        assert stats["scheduled_count"] == 0
+
+    @pytest.mark.asyncio
+    async def test_get_stats_with_records(self, volatile_service, mock_qdrant):
+        """Test stats with records."""
+        import time
+        now_ms = int(time.time() * 1000)
+
+        mock_qdrant.scroll_all_points.return_value = [
+            {"payload": {"namespace": "weather", "ttl_expiry": now_ms + 100000}},
+            {"payload": {"namespace": "weather", "ttl_expiry": now_ms + 100000, "refresh_schedule": "0 * * * *"}},
+            {"payload": {"namespace": "news", "ttl_expiry": now_ms + 100000}},
+            {"payload": {"namespace": "weather", "ttl_expiry": now_ms - 100000}},  # Expired
+        ]
+
+        stats = await volatile_service.get_stats("jpmschweitzer")
+
+        assert stats["total_records"] == 3  # Excludes expired
+        assert stats["by_namespace"]["weather"] == 2
+        assert stats["by_namespace"]["news"] == 1
+        assert stats["scheduled_count"] == 1
+        assert stats["expired_count"] == 1
+
+    @pytest.mark.asyncio
+    async def test_purge_expired(self, volatile_service, mock_qdrant):
+        """Test purging expired records."""
+        mock_qdrant.delete_expired_vectors.return_value = 5
+
+        result = await volatile_service.purge_expired("jpmschweitzer")
+
+        assert result == 5
+        mock_qdrant.delete_expired_vectors.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_purge_all_expired(self, volatile_service, mock_qdrant):
+        """Test purging expired from all collections."""
+        mock_qdrant.get_volatile_collections.return_value = [
+            "volatile_user1",
+            "volatile_user2"
+        ]
+        mock_qdrant.delete_expired_vectors.side_effect = [3, 2]
+
+        results = await volatile_service.purge_all_expired()
+
+        assert results["volatile_user1"] == 3
+        assert results["volatile_user2"] == 2
+
+    @pytest.mark.asyncio
+    async def test_get_scheduled(self, volatile_service, mock_qdrant):
+        """Test getting scheduled records."""
+        import time
+        now_ms = int(time.time() * 1000)
+
+        mock_qdrant.scroll_all_points.return_value = [
+            {
+                "payload": {
+                    "key": "nos-headlines",
+                    "namespace": "news",
+                    "raw_data": {"headlines": []},
+                    "source": "nos.nl",
+                    "created_at": datetime.utcnow().isoformat(),
+                    "updated_at": datetime.utcnow().isoformat(),
+                    "ttl": 3600,
+                    "ttl_expiry": now_ms + 1800000,
+                    "refresh_schedule": "0 */6 * * *",
+                    "user": "jpmschweitzer"
+                }
+            },
+            {
+                "payload": {
+                    "key": "rotterdam",
+                    "namespace": "weather",
+                    "raw_data": {"temperature": 18},
+                    "source": "openweathermap",
+                    "created_at": datetime.utcnow().isoformat(),
+                    "updated_at": datetime.utcnow().isoformat(),
+                    "ttl": 1800,
+                    "ttl_expiry": now_ms + 900000,
+                    "refresh_schedule": None,  # Not scheduled
+                    "user": "jpmschweitzer"
+                }
+            }
+        ]
+
+        scheduled = await volatile_service.get_scheduled("jpmschweitzer")
+
+        assert len(scheduled) == 1
+        assert scheduled[0].key == "nos-headlines"
+        assert scheduled[0].refresh_schedule == "0 */6 * * *"
+
+
+class TestVolatileCleanupEndpoint:
+    """Test volatile cleanup in maintenance router."""
+
+    @pytest.mark.asyncio
+    async def test_cleanup_volatile(self):
+        """Test volatile cleanup endpoint."""
+        from src.routers.maintenance import cleanup_volatile, VolatileCleanupResponse
+
+        mock_qdrant = AsyncMock()
+        mock_qdrant.get_volatile_collections = AsyncMock(return_value=[
+            "volatile_user1",
+            "volatile_user2"
+        ])
+        mock_qdrant.delete_expired_vectors = AsyncMock(side_effect=[3, 2])
+
+        mock_ollama = AsyncMock()
+
+        mock_settings = MagicMock()
+        mock_settings.volatile_default_ttl = 3600
+
+        with patch('src.routers.maintenance.get_settings', return_value=mock_settings):
+            result = await cleanup_volatile(
+                qdrant=mock_qdrant,
+                ollama=mock_ollama,
+                api_key="test"
+            )
+
+        assert result.success is True
+        assert result.collections_processed == 2
+        assert result.total_expired_purged == 5
+        assert result.by_collection["volatile_user1"] == 3
+        assert result.by_collection["volatile_user2"] == 2