feat: refactor volatile cache to vector storage with HybridRAG integration

- Migrate volatile backend from Redis to Qdrant for semantic search
- Add natural language conversion for structured data embedding
- Simplify API: /volatile/search, /volatile/store, /{namespace}/{key}
- Integrate volatile into HybridRAG with priority boost in RRF fusion
- Add POST /maintenance/cleanup/volatile for expiry purging
- Update tests for new Qdrant-based architecture (37/37 pass)

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

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

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