fix: update Paperless webhook payload to match include_document format

- Change model field from document_id to id (Paperless sends id)
- Add content, created, modified, added, original_file_name, owner fields
- Add extra="ignore" config to handle additional Paperless fields
- Update sync service to use content from webhook payload
- Skip Paperless API call when content already provided

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

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

.env.example

@@ -7,6 +7,7 @@ QDRANT_PORT=6333
 OLLAMA_URL=http://192.168.86.149:11434
 SEARXNG_URL=http://192.168.86.149:8080
 REDIS_HOST=192.168.86.149
+PAPERLESS_URL=http://192.168.86.149:8000
 
 OLLAMA_MODEL=mistral-nemo-large:latest
 OLLAMA_EMBEDDING_MODEL=nomic-embed-text
@@ -20,4 +21,5 @@ WIKI_GRAPHQL_API=your_jwt_token_here
 LIBRARY_API_KEY=key_here
 NEO4J_PASSWORD=key_here
 WIKIJS_DB_PASSWORD=key_here
-SCHEDULER_API_KEY=key_here
+SCHEDULER_API_KEY=key_here
+PAPERLESS_TOKEN=key_here

CHANGELOG.md

@@ -5,6 +5,84 @@ 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.6] - 2025-12-25
+
+### Fixed
+
+- **Paperless Webhook Payload Format** - Updated model to match Paperless `include_document=true` format
+  - Paperless sends `id` instead of `document_id`
+  - Paperless sends full document data including `content`, `title`, `tags`, etc.
+  - Webhook now uses content from payload, skipping extra Paperless API call
+  - Added `extra = "ignore"` to handle additional Paperless fields
+
+## [1.4.5] - 2025-12-25
+
+### Added
+
+- **Document Storage Integration** - Paperless-ngx integration for PDFs, images, and documents
+  - Event-driven architecture via Paperless webhooks
+  - `POST /documents/webhook` - Receive document events from Paperless workflows
+  - `POST /documents/upload` - Upload files directly to Paperless
+  - `POST /documents/upload-url` - Download and upload documents from URL
+  - `POST /documents/search` - Semantic search across indexed documents
+  - `GET /documents/health` - Paperless connectivity health check
+- **DocumentSyncService** - Indexes Paperless documents into vectors and graph
+  - Fetches document content via Paperless API
+  - Chunks text and generates embeddings for Qdrant
+  - Creates Document nodes in Neo4j knowledge graph
+  - Supports multi-tenancy via user parameter in webhook URL
+- **PaperlessClient** - REST API client for Paperless-ngx
+  - Document retrieval, upload, and update operations
+  - Health check support
+- **Paperless Workflow Configuration**
+  - Production workflow: Document Added (NOT tagged llm-test) → webhook to Library Desk
+  - Test workflow: Document Added (tagged llm-test) → webhook with test user
+
+### Changed
+
+- Updated `src/config.py` with Paperless configuration settings
+- Added `PaperlessDep` dependency injection for document endpoints
+
+## [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/DOCUMENT_STORAGE_PLAN.md

@@ -0,0 +1,509 @@
+# Phase 3: Document Storage System - Implementation Plan
+
+## Overview
+
+Document storage tier for Library Desk - storing and indexing PDFs, images, videos, and git documentation mirrors.
+
+**User Decisions:**
+- Paperless-ngx container for OCR
+- Ebooks deferred to future phase
+- Video.js player deferred to after core implementation
+
+| Phase | Status | Version |
+|-------|--------|---------|
+| Phase 1: Cleanup System | Complete | v1.4.0 |
+| Phase 2: Volatile Memory | Complete | v1.4.3 |
+| Phase 3: Document Storage | Planning | - |
+| Phase 4: Test Data Cleanup | Complete | v1.4.4 |
+
+---
+
+## Architecture
+
+**Paperless-ngx as primary document store** (no SeaweedFS needed):
+
+```
+┌─────────────────────────────────────────────────────────────────┐
+│                     External Sources                             │
+│  ┌─────────┐  ┌────────────┐  ┌──────────────┐                  │
+│  │ GitHub  │  │ Direct     │  │ Email/Folder │                  │
+│  │ Docs    │  │ Upload     │  │ Ingestion    │                  │
+│  └────┬────┘  └─────┬──────┘  └──────┬───────┘                  │
+└───────┼─────────────┼────────────────┼──────────────────────────┘
+        │             │                │
+        ▼             ▼                ▼
+┌─────────────────────────────────────────────────────────────────┐
+│                      Paperless-ngx                               │
+│  ┌───────────────────────────────────────────────────────────┐  │
+│  │  - Document storage (PDFs, images, videos)                │  │
+│  │  - OCR via Tesseract (PDFs, images)                       │  │
+│  │  - Web UI for browsing/tagging                            │  │
+│  │  - REST API for integration                               │  │
+│  └─────────────────────────┬─────────────────────────────────┘  │
+└────────────────────────────┼────────────────────────────────────┘
+                             │ REST API (sync)
+                             ▼
+┌─────────────────────────────────────────────────────────────────┐
+│                        Library Desk                              │
+│  ┌───────────────────────────────────────────────────────────┐  │
+│  │              DocumentSyncService                          │  │
+│  │  - Polls Paperless for new/updated docs                   │  │
+│  │  - Extracts text + metadata via API                       │  │
+│  │  - Sends to vector/graph pipelines                        │  │
+│  └─────────────────────────┬─────────────────────────────────┘  │
+│                            │                                     │
+│           ┌────────────────┼────────────────┐                   │
+│           ▼                ▼                ▼                   │
+│      ┌──────────┐    ┌──────────┐    ┌──────────┐              │
+│      │  Qdrant  │    │  Neo4j   │    │ Wiki.js  │              │
+│      │ (vectors)│    │ (graph)  │    │ (catalog)│              │
+│      └──────────┘    └──────────┘    └──────────┘              │
+└─────────────────────────────────────────────────────────────────┘
+```
+
+**File handling by type:**
+
+| File Type | Paperless | Library Desk |
+|-----------|-----------|--------------|
+| PDFs | OCR → text | Index text → vectors/graph |
+| Images | OCR → text | Index text → vectors/graph |
+| Videos | Storage only | Index metadata → vectors/graph |
+
+---
+
+## Technology Stack
+
+| Component | Purpose | Rationale |
+|-----------|---------|-----------|
+| **Paperless-ngx** | Document storage + OCR | All-in-one: storage, OCR, web UI, REST API |
+| **ClamAV** | Virus scanning | Host OS install, pyclamd integration, better isolation |
+| **PDF.js** | PDF viewer | Embeddable in Wiki.js (deferred) |
+
+**Why Paperless-ngx as primary store:**
+- Eliminates need for separate blob storage (SeaweedFS/MinIO)
+- Built-in web UI for browsing and tagging
+- Tesseract OCR with 100+ language support
+- REST API for Library Desk integration
+- Handles videos as raw files (no OCR, but stored)
+- Email and folder watching for automatic ingestion
+- Active community, well-maintained
+
+---
+
+## Paperless-ngx API Deep Dive
+
+### Authentication
+```
+POST /api/token/
+Body: {"username": "...", "password": "..."}
+Response: {"token": "..."}
+
+Header: Authorization: Token <token>
+```
+
+### Document Upload (for HybridRAG → Paperless)
+```
+POST /api/documents/post_document/
+Content-Type: multipart/form-data
+
+Fields:
+- document (file, required)
+- title (string)
+- created (datetime)
+- correspondent (ID)
+- document_type (ID)
+- storage_path (ID)
+- tags (repeatable IDs)
+- custom_fields (JSON array)
+
+Response: {"task_id": "uuid"}
+```
+
+Track consumption: `GET /api/tasks/?task_id={uuid}` → returns document ID when complete
+
+### Document Search
+```
+GET /api/documents/?query=search+terms     # Full-text search
+GET /api/documents/?more_like_id=123       # Similarity search
+
+Response includes __search_hit__:
+{
+  "score": 0.95,
+  "highlights": "<span>matched</span> text",
+  "rank": 0
+}
+```
+
+### Custom Field Filtering
+```
+GET /api/documents/?custom_field_query=field_name__operation=value
+
+Operations:
+- exact, in, isnull, exists (all types)
+- icontains, istartswith, iendswith (text)
+- gt, gte, lt, lte, range (numeric/date)
+- contains (document links)
+```
+
+### Bulk Operations
+```
+POST /api/documents/bulk_edit/
+{
+  "documents": [1, 2, 3],
+  "method": "add_tag|remove_tag|set_correspondent|set_document_type|merge|split|...",
+  "parameters": {...}
+}
+```
+
+### Webhooks (Push to Library Desk!)
+Paperless workflows can trigger webhooks on document events:
+
+| Trigger | When | Available Data |
+|---------|------|----------------|
+| Consumption Started | Before OCR | file_path, source, filename |
+| Document Added | After OCR | content, tags, doc_type, correspondent, `{doc_url}` |
+| Document Updated | On change | Same as Added |
+| Scheduled | Time-based | Date offsets from document dates |
+
+**Webhook Action**: POST to Library Desk endpoint with document data
+
+### Organization Features
+
+| Feature | Purpose | API Endpoint |
+|---------|---------|--------------|
+| Tags | Nested labels (5 levels deep) | `/api/tags/` |
+| Correspondents | Source/destination | `/api/correspondents/` |
+| Document Types | Classification | `/api/document_types/` |
+| Storage Paths | File organization | `/api/storage_paths/` |
+| Custom Fields | Extensible metadata | `/api/custom_fields/` |
+
+### Custom Fields We Should Create
+| Field Name | Type | Purpose |
+|------------|------|---------|
+| `source_url` | URL | Original download URL (for HybridRAG uploads) |
+| `library_indexed` | Boolean | Sync status with Library Desk |
+| `library_doc_id` | Text | Library Desk document reference |
+| `collection` | Text | Logical grouping (e.g., "fastapi-docs") |
+
+### External LLM Add-ons (Optional)
+Community tools exist for Ollama integration:
+- **[paperless-ai](https://github.com/clusterzx/paperless-ai)** - Auto-tagging, RAG chat
+- **[paperless-gpt](https://github.com/icereed/paperless-gpt)** - LLM-enhanced OCR, auto-titling
+
+**Recommendation:** Skip these - Library Desk already has Ollama integration for:
+- Embedding (nomic-embed-text)
+- LLM analysis (mistral-nemo)
+- Entity extraction
+- HybridRAG
+
+We'll do our own classification/tagging via Library Desk after sync.
+
+---
+
+## Virus Scanning Integration
+
+**ClamAV daemon + pyclamd** (no third-party REST wrappers):
+
+```
+┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
+│  File Upload    │────►│  Library Desk   │────►│  ClamAV Daemon  │
+│  (URL or file)  │     │  (pyclamd)      │     │  (clamd:3310)   │
+└─────────────────┘     └────────┬────────┘     └─────────────────┘
+                                 │
+                    ┌────────────┴────────────┐
+                    ▼                         ▼
+              ┌──────────┐              ┌──────────┐
+              │  Clean   │              │ Infected │
+              │  ✓       │              │  ✗       │
+              └────┬─────┘              └────┬─────┘
+                   │                         │
+                   ▼                         ▼
+          Upload to Paperless         Reject + Log
+```
+
+### ClamAV Deployment (Host OS)
+
+ClamAV runs on the host OS (not containerized) for better security isolation:
+
+```bash
+# Installed via apt on Ubuntu/Debian
+# Config: /etc/clamav/clamd.conf
+# TCPSocket 3310
+# TCPAddr 0.0.0.0
+```
+
+Benefits: scans outside container isolation, single virus DB, survives container restarts.
+
+### Library Desk Integration
+```python
+# src/clients/clamav_client.py
+import pyclamd
+
+class ClamAVClient:
+    def __init__(self, host: str, port: int = 3310):
+        self.cd = pyclamd.ClamdNetworkSocket(host, port)
+
+    async def scan_bytes(self, data: bytes) -> ScanResult:
+        """Scan file bytes, return clean/infected status."""
+        result = self.cd.scan_stream(data)
+        if result is None:
+            return ScanResult(clean=True)
+        return ScanResult(clean=False, virus_name=result['stream'][1])
+
+    def ping(self) -> bool:
+        """Health check."""
+        return self.cd.ping()
+```
+
+### Scan Points
+| Location | When | Action on Infected |
+|----------|------|-------------------|
+| `/documents/upload` | Before Paperless upload | Reject with 400, log threat |
+| HybridRAG web fetch | Before saving PDF | Skip file, log threat |
+| `/documents/webhook` | Optional re-scan | Quarantine in Paperless |
+
+### Config Settings
+```python
+# src/config.py
+CLAMAV_HOST: str = "192.168.86.149"  # Host OS IP (not container)
+CLAMAV_PORT: int = 3310
+CLAMAV_ENABLED: bool = True  # Bypass for testing
+CLAMAV_TIMEOUT: int = 30  # seconds
+```
+
+---
+
+## Integration Strategy
+
+### Option A: Webhook Push (Preferred)
+```
+Paperless Workflow → POST webhook → Library Desk /documents/webhook
+```
+- Real-time indexing when documents added/updated
+- Configure in Paperless: Workflow → Document Added → Webhook Action
+- Library Desk receives document ID, fetches content via API
+
+### Option B: Polling Pull (Fallback)
+```
+Scheduler → POST /documents/sync → Library Desk polls Paperless
+```
+- Periodic sync for missed webhooks or initial bulk import
+- Track `library_indexed` custom field to skip already-processed docs
+
+### Option C: HybridRAG Upload (New!)
+```
+HybridRAG web search → finds PDF → POST to Paperless → webhook → indexed
+```
+- When HybridRAG finds a relevant PDF/document in web results
+- Download and upload to Paperless with `source_url` custom field
+- Paperless OCRs it, triggers webhook, Library Desk indexes
+
+---
+
+## Library Desk API Design
+
+### Documents Router (`/documents`)
+
+| Endpoint | Method | Purpose |
+|----------|--------|---------|
+| `/documents/webhook` | POST | Receive Paperless webhook (Document Added/Updated) |
+| `/documents/sync` | POST | Pull new/updated docs from Paperless → index |
+| `/documents/upload` | POST | Upload file to Paperless (for HybridRAG) |
+| `/documents/sync-from-git` | POST | Pull docs from Gitea → upload to Paperless → index |
+| `/documents/{document_id}` | GET | Get document metadata |
+| `/documents/{document_id}/text` | GET | Get extracted text |
+| `/documents/search` | POST | Semantic search across documents |
+| `/documents/collection/{name}` | GET | List documents in collection |
+| `/documents/collection/{name}/catalog` | POST | Generate wiki catalog page |
+
+**Upload flow (HybridRAG → Paperless):**
+1. HybridRAG finds PDF in web results
+2. POST `/documents/upload` with URL or file
+3. Library Desk downloads, uploads to Paperless with metadata
+4. Returns task_id for async tracking
+5. Paperless webhook triggers indexing when OCR complete
+
+### Viewers Router (`/viewers`) - Deferred
+
+| Endpoint | Method | Purpose |
+|----------|--------|---------|
+| `/viewers/pdf/{document_id}` | GET | Serve PDF.js viewer |
+| `/viewers/image/{document_id}` | GET | Serve image lightbox |
+| `/viewers/video/{document_id}` | GET | Serve Video.js player |
+
+---
+
+## Data Flow: Document Processing Pipeline
+
+```
+1. INTAKE (Paperless-ngx handles this)
+   └─ Upload via Paperless UI, email, or folder watch
+   └─ Paperless assigns document ID and stores file
+
+2. OCR EXTRACTION (Paperless-ngx handles this)
+   ├─ PDFs → Tesseract → Plain text
+   ├─ Images → Tesseract → Plain text
+   └─ Videos → Metadata only (no OCR)
+
+3. SYNC TO LIBRARY DESK (scheduled or manual)
+   └─ Poll Paperless API for new/updated documents
+   └─ Fetch text content + metadata
+
+4. TEXT CHUNKING
+   └─ VectorService._chunk_text() (existing)
+
+5. EMBEDDING
+   └─ OllamaClient.embed() (existing)
+
+6. VECTOR STORAGE (Qdrant)
+   └─ Payload: {doc_type: "document", paperless_id, ...}
+
+7. GRAPH STORAGE (Neo4j)
+   └─ Document node + MENTIONS relationships
+
+8. WIKI CATALOG (optional)
+   └─ Auto-generate catalog page via ConsolidationService
+```
+
+---
+
+## Git Docs Integration
+
+Extends existing `scheduler/src/executors/doc_sync_executor.py`:
+
+1. **Scheduler** syncs docs from GitHub → Gitea (existing)
+2. **Post-sync hook** calls `POST /documents/sync-from-git`
+3. **Library Desk** indexes docs into vectors/graph
+4. **Auto-generate** wiki catalog page for collection
+
+---
+
+## Wiki.js Viewer Integration
+
+Since Wiki.js v2 requires disabled HTML sanitization for iframes:
+
+```markdown
+<!-- In wiki catalog page -->
+## Document Preview
+
+<iframe
+  src="http://library-desk:8089/viewers/pdf/abc123"
+  width="100%" height="600px">
+</iframe>
+```
+
+**Wiki.js Settings Required:**
+- `Administration > Security > Allowed HTML Elements: iframe`
+- `Content Security Policy: frame-src http://library-desk:8089`
+
+---
+
+## Implementation Phases
+
+### Phase 3.1: Infrastructure Setup
+- [ ] Deploy Paperless-ngx container (Docker Compose)
+- [x] ClamAV installed on host OS (port 3310)
+- [ ] Configure Paperless: storage path, OCR settings, API token
+- [ ] Create custom fields in Paperless: `source_url`, `library_indexed`, `library_doc_id`, `collection`
+- [ ] Create `src/clients/paperless_client.py`
+- [ ] Create `src/clients/clamav_client.py` (pyclamd wrapper)
+- [ ] Create `src/models/document.py`
+- [ ] Add config settings to `src/config.py` (PAPERLESS_*, CLAMAV_*)
+
+### Phase 3.2: Webhook Integration (Push)
+- [ ] Create `src/routers/documents.py`
+- [ ] Implement `/documents/webhook` endpoint (receives Paperless events)
+- [ ] Configure Paperless Workflow: Document Added → Webhook → Library Desk
+- [ ] Create `src/services/document_sync_service.py`
+- [ ] Implement document indexing pipeline (fetch text → chunk → embed → graph)
+
+### Phase 3.3: Polling Sync (Pull Fallback)
+- [ ] Implement `/documents/sync` endpoint
+- [ ] Poll Paperless for docs where `library_indexed=false`
+- [ ] Track sync state (last_sync timestamp in Redis)
+- [ ] Update `library_indexed` after successful indexing
+
+### Phase 3.4: HybridRAG Upload Integration
+- [ ] Implement `/documents/upload` endpoint
+- [ ] Download file from URL
+- [ ] **Virus scan before upload** (reject if infected, log threat)
+- [ ] Upload clean files to Paperless with metadata
+- [ ] Set `source_url` custom field
+- [ ] Extend HybridRAG service to detect and upload relevant PDFs
+- [ ] Add `save_to_documents` option to HybridRAG config
+
+### Phase 3.5: Indexing Pipeline
+- [ ] Extend VectorService for `doc_type: "document"`
+- [ ] Extend GraphService for Document nodes (link to Paperless ID)
+- [ ] Implement `/documents/search` endpoint
+- [ ] Add dependency injection
+
+### Phase 3.6: Git Docs Integration
+- [ ] Create `src/clients/gitea_client.py`
+- [ ] Implement `/documents/sync-from-git` → bulk upload to Paperless
+- [ ] Create collection auto-cataloging (wiki pages)
+- [ ] Add scheduler task for periodic git sync
+
+### Phase 3.7: Viewers (Deferred)
+*After core implementation is working*
+- [ ] Create `static/pdf-viewer.html` (PDF.js)
+- [ ] Create `static/image-viewer.html`
+- [ ] Create `static/video-player.html` (Video.js)
+- [ ] Create `src/routers/viewers.py`
+
+### Phase 3.8: Maintenance & Testing
+- [ ] Extend cleanup for document orphans
+- [ ] Add document orphan detection (Paperless deleted but still in Qdrant/Neo4j)
+- [ ] Create `tests/test_document_sync.py`
+- [ ] Create `tests/test_paperless_client.py`
+
+---
+
+## Files to Create
+
+| Path | Purpose |
+|------|---------|
+| `src/clients/paperless_client.py` | Paperless-ngx REST API client |
+| `src/clients/clamav_client.py` | ClamAV scanner (pyclamd wrapper) |
+| `src/clients/gitea_client.py` | Gitea repo access |
+| `src/models/document.py` | Document/Collection/ScanResult models |
+| `src/services/document_sync_service.py` | Sync orchestrator |
+| `src/routers/documents.py` | Document endpoints (webhook, sync, upload, search) |
+| `tests/test_document_sync.py` | Sync service tests |
+| `tests/test_paperless_client.py` | API client tests |
+| `tests/test_clamav_client.py` | Virus scanner tests |
+| `docker/docker-compose.documents.yml` | Paperless + ClamAV deployment |
+
+**Deferred files (Phase 3.7):**
+
+| Path | Purpose |
+|------|---------|
+| `src/routers/viewers.py` | Viewer endpoints |
+| `static/pdf-viewer.html` | PDF.js viewer |
+| `static/image-viewer.html` | Image lightbox |
+| `static/video-player.html` | Video.js player |
+
+## Files to Modify
+
+| Path | Changes |
+|------|---------|
+| `src/config.py` | `PAPERLESS_*`, `CLAMAV_*` settings |
+| `src/core/dependencies.py` | DocumentSyncService, PaperlessClient, ClamAVClient DI |
+| `src/main.py` | Register documents router |
+| `src/services/vector_service.py` | `doc_type: "document"` handling |
+| `src/services/graph_service.py` | Document node with Paperless ID |
+| `src/services/hybrid_rag_service.py` | Add `save_to_documents` option + virus scan |
+| `src/models/hybrid_rag.py` | Add `save_to_documents` config |
+| `src/routers/maintenance.py` | Document orphan cleanup, ClamAV health check |
+| `requirements.txt` | Add `pyclamd` |
+
+## Paperless Custom Fields Setup
+
+Create these in Paperless UI (Administration → Custom Fields):
+
+| Field | Type | Purpose |
+|-------|------|---------|
+| `source_url` | URL | Original download URL |
+| `library_indexed` | Boolean | Sync status |
+| `library_doc_id` | Text | Library Desk reference |
+| `collection` | Text | Logical grouping |

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 |
+| **Documents** | Paperless-ngx + ClamAV (host) | 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 | ✅ Planned | See [DOCUMENT_STORAGE_PLAN.md](DOCUMENT_STORAGE_PLAN.md) |
+| 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       │
+│     Qdrant      │
-│  (DB 4, TTL)    │
+│ (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
+### Endpoints (Implemented)
-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`
 
 | Endpoint | Method | Purpose |
 |----------|--------|---------|
-| `/volatile/{namespace}/{key}` | GET | Retrieve record |
+| `/volatile/search?q=...` | GET | Semantic search across volatile data |
-| `/volatile/{namespace}/{key}` | POST | Store/update record |
+| `/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
+### Namespaces
-**File**: `src/services/consolidation_service.py`
 
-Add relevance trigger detection:
+| Namespace | Default TTL | Use Case |
-1. During consolidation, analyze search results for location/interest patterns
+|-----------|-------------|----------|
-2. Query tatlock's Biographer collection for user preferences
+| weather | 30 min | Current conditions, forecasts |
-3. If match found, create/update volatile refresh schedule
+| news | 1 hour | Headlines, breaking news |
-
+| financial | 5 min | Stock prices, exchange rates |
-#### 2.4 Biographer Integration
+| transit | 5 min | Train/bus schedules, delays |
-**File**: `src/core/dependencies.py`
+| traffic | 10 min | Commute times, road conditions |
-
+| air_quality | 1 hour | Pollution, pollen counts |
-```python
+| sports | 1 min | Live scores, matches |
-def get_biographer_qdrant() -> QdrantClientWrapper:
+| social | 10 min | Social notifications |
-    """Direct access to tatlock's Biographer collection."""
+| system | 1 min | Service health status |
-    # Configure to connect to tatlock's Qdrant
+| context | 1 hour | Session state |
-```
+| custom | 1 hour | User-defined data |
-
-#### 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"
-}
-```
 
 ---
 
@@ -219,34 +186,116 @@ Add LLM-powered category descriptor generation:
 
 ---
 
-## Files to Modify/Create
+## Phase 4: LLM Tester Data Cleanup ✅
 
-### Phase 1 (Cleanup)
+### Problem
-- `src/routers/maintenance.py` - Add timestamp tracking
+
+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)
+### Phase 2 (Volatile) ✅
-- `src/services/volatile_service.py` - **NEW**
+- `src/services/volatile_service.py` - Qdrant-based volatile cache
-- `src/routers/volatile.py` - **NEW**
+- `src/routers/volatile.py` - Simplified endpoints
-- `src/models/volatile.py` - **NEW**
+- `src/models/volatile.py` - Namespaces and models
-- `src/core/dependencies.py` - Add Biographer client
+- `src/models/hybrid_rag.py` - Volatile config options
-- `src/services/consolidation_service.py` - Relevance triggers
+- `src/services/hybrid_rag_service.py` - Volatile integration
-- `tests/test_volatile.py` - **NEW**
+- `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.
+1. **Volatile Storage**: Qdrant vectors (not Redis) for semantic search capability
-2. **Scheduler API**: Has REST API for task registration. Library-Desk can programmatically create refresh schedules.
+2. **Collection Naming**: `volatile_{user}` for per-user isolation
-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.
+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.6"
 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/paperless_client.py

@@ -0,0 +1,488 @@
+"""
+Paperless-ngx API client for Library Desk.
+
+Provides async document management via Paperless-ngx:
+- Document upload and retrieval
+- Search and filtering
+- Custom field management
+- Task status tracking
+"""
+
+import httpx
+from typing import Optional, List, Dict, Any
+from dataclasses import dataclass
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class PaperlessDocument:
+    """Represents a document from Paperless-ngx."""
+    id: int
+    title: str
+    content: str
+    created: Optional[str] = None
+    modified: Optional[str] = None
+    added: Optional[str] = None
+    correspondent: Optional[int] = None
+    document_type: Optional[int] = None
+    storage_path: Optional[int] = None
+    tags: List[int] = None
+    archive_serial_number: Optional[int] = None
+    original_file_name: Optional[str] = None
+    archived_file_name: Optional[str] = None
+    custom_fields: List[Dict[str, Any]] = None
+
+    def __post_init__(self):
+        if self.tags is None:
+            self.tags = []
+        if self.custom_fields is None:
+            self.custom_fields = []
+
+
+@dataclass
+class SearchHit:
+    """Search result with relevance info."""
+    document: PaperlessDocument
+    score: float
+    rank: int
+    highlights: Optional[str] = None
+
+
+class PaperlessClient:
+    """
+    Paperless-ngx REST API client.
+
+    Documentation: https://docs.paperless-ngx.com/api/
+    """
+
+    def __init__(self, base_url: str, token: str, timeout: int = 30):
+        """
+        Initialize Paperless-ngx client.
+
+        Args:
+            base_url: Paperless-ngx base URL (e.g., "http://paperless:8000")
+            token: API token for authentication
+            timeout: Request timeout in seconds
+        """
+        self.base_url = base_url.rstrip("/")
+        self.api_url = f"{self.base_url}/api"
+        self.headers = {
+            "Authorization": f"Token {token}",
+            "Accept": "application/json",
+        }
+        self.client = httpx.AsyncClient(timeout=float(timeout), headers=self.headers)
+        logger.info(f"Initialized Paperless client: {base_url}")
+
+    async def close(self):
+        """Close HTTP client."""
+        await self.client.aclose()
+
+    # =========================================================================
+    # Document Operations
+    # =========================================================================
+
+    async def get_document(self, document_id: int) -> Optional[PaperlessDocument]:
+        """
+        Get a document by ID.
+
+        Args:
+            document_id: Paperless document ID
+
+        Returns:
+            PaperlessDocument or None if not found
+        """
+        try:
+            response = await self.client.get(f"{self.api_url}/documents/{document_id}/")
+            response.raise_for_status()
+            data = response.json()
+            return self._parse_document(data)
+        except httpx.HTTPStatusError as e:
+            if e.response.status_code == 404:
+                return None
+            logger.error(f"Failed to get document {document_id}: {e}")
+            raise
+        except Exception as e:
+            logger.error(f"Failed to get document {document_id}: {e}")
+            raise
+
+    async def get_document_content(self, document_id: int) -> Optional[str]:
+        """
+        Get extracted text content of a document.
+
+        Args:
+            document_id: Paperless document ID
+
+        Returns:
+            Text content or None if not found
+        """
+        doc = await self.get_document(document_id)
+        return doc.content if doc else None
+
+    async def list_documents(
+        self,
+        page: int = 1,
+        page_size: int = 25,
+        ordering: str = "-added",
+        correspondent: Optional[int] = None,
+        document_type: Optional[int] = None,
+        tags: Optional[List[int]] = None,
+    ) -> Dict[str, Any]:
+        """
+        List documents with pagination and filtering.
+
+        Args:
+            page: Page number (starts at 1)
+            page_size: Results per page
+            ordering: Sort order (prefix with - for descending)
+            correspondent: Filter by correspondent ID
+            document_type: Filter by document type ID
+            tags: Filter by tag IDs
+
+        Returns:
+            Paginated response with count, next, previous, results
+        """
+        params = {
+            "page": page,
+            "page_size": page_size,
+            "ordering": ordering,
+        }
+        if correspondent:
+            params["correspondent__id"] = correspondent
+        if document_type:
+            params["document_type__id"] = document_type
+        if tags:
+            params["tags__id__in"] = ",".join(str(t) for t in tags)
+
+        try:
+            response = await self.client.get(f"{self.api_url}/documents/", params=params)
+            response.raise_for_status()
+            data = response.json()
+            return {
+                "count": data.get("count", 0),
+                "next": data.get("next"),
+                "previous": data.get("previous"),
+                "results": [self._parse_document(d) for d in data.get("results", [])],
+            }
+        except Exception as e:
+            logger.error(f"Failed to list documents: {e}")
+            raise
+
+    async def search_documents(
+        self,
+        query: str,
+        page: int = 1,
+        page_size: int = 25,
+    ) -> List[SearchHit]:
+        """
+        Full-text search documents.
+
+        Args:
+            query: Search query string
+            page: Page number
+            page_size: Results per page
+
+        Returns:
+            List of SearchHit with document and relevance info
+        """
+        params = {
+            "query": query,
+            "page": page,
+            "page_size": page_size,
+        }
+
+        try:
+            response = await self.client.get(f"{self.api_url}/documents/", params=params)
+            response.raise_for_status()
+            data = response.json()
+
+            results = []
+            for item in data.get("results", []):
+                doc = self._parse_document(item)
+                hit_info = item.get("__search_hit__", {})
+                results.append(SearchHit(
+                    document=doc,
+                    score=hit_info.get("score", 0.0),
+                    rank=hit_info.get("rank", 0),
+                    highlights=hit_info.get("highlights"),
+                ))
+            return results
+        except Exception as e:
+            logger.error(f"Search failed for '{query}': {e}")
+            raise
+
+    async def upload_document(
+        self,
+        file_content: bytes,
+        filename: str,
+        title: Optional[str] = None,
+        correspondent: Optional[int] = None,
+        document_type: Optional[int] = None,
+        tags: Optional[List[int]] = None,
+        custom_fields: Optional[List[Dict[str, Any]]] = None,
+    ) -> str:
+        """
+        Upload a document to Paperless-ngx.
+
+        Args:
+            file_content: File bytes
+            filename: Original filename
+            title: Document title (optional, derived from filename if not set)
+            correspondent: Correspondent ID
+            document_type: Document type ID
+            tags: List of tag IDs
+            custom_fields: List of custom field values
+
+        Returns:
+            Task UUID for tracking consumption status
+        """
+        files = {"document": (filename, file_content)}
+        data = {}
+
+        if title:
+            data["title"] = title
+        if correspondent:
+            data["correspondent"] = correspondent
+        if document_type:
+            data["document_type"] = document_type
+        if tags:
+            # Tags need to be sent multiple times for multiple values
+            data["tags"] = tags
+        if custom_fields:
+            data["custom_fields"] = custom_fields
+
+        try:
+            response = await self.client.post(
+                f"{self.api_url}/documents/post_document/",
+                files=files,
+                data=data,
+            )
+            response.raise_for_status()
+            result = response.json()
+            task_id = result.get("task_id", "")
+            logger.info(f"Uploaded document '{filename}', task_id: {task_id}")
+            return task_id
+        except Exception as e:
+            logger.error(f"Failed to upload document '{filename}': {e}")
+            raise
+
+    async def get_task_status(self, task_id: str) -> Dict[str, Any]:
+        """
+        Get status of a consumption task.
+
+        Args:
+            task_id: Task UUID from upload
+
+        Returns:
+            Task status with state, result, etc.
+        """
+        try:
+            response = await self.client.get(
+                f"{self.api_url}/tasks/",
+                params={"task_id": task_id},
+            )
+            response.raise_for_status()
+            data = response.json()
+            results = data.get("results", [])
+            if results:
+                return results[0]
+            return {"status": "NOT_FOUND"}
+        except Exception as e:
+            logger.error(f"Failed to get task status {task_id}: {e}")
+            raise
+
+    async def update_document(
+        self,
+        document_id: int,
+        title: Optional[str] = None,
+        correspondent: Optional[int] = None,
+        document_type: Optional[int] = None,
+        tags: Optional[List[int]] = None,
+        custom_fields: Optional[List[Dict[str, Any]]] = None,
+    ) -> PaperlessDocument:
+        """
+        Update a document's metadata.
+
+        Args:
+            document_id: Document ID to update
+            title: New title
+            correspondent: New correspondent ID
+            document_type: New document type ID
+            tags: New tag IDs (replaces existing)
+            custom_fields: New custom field values
+
+        Returns:
+            Updated document
+        """
+        data = {}
+        if title is not None:
+            data["title"] = title
+        if correspondent is not None:
+            data["correspondent"] = correspondent
+        if document_type is not None:
+            data["document_type"] = document_type
+        if tags is not None:
+            data["tags"] = tags
+        if custom_fields is not None:
+            data["custom_fields"] = custom_fields
+
+        try:
+            response = await self.client.patch(
+                f"{self.api_url}/documents/{document_id}/",
+                json=data,
+            )
+            response.raise_for_status()
+            return self._parse_document(response.json())
+        except Exception as e:
+            logger.error(f"Failed to update document {document_id}: {e}")
+            raise
+
+    # =========================================================================
+    # Custom Fields
+    # =========================================================================
+
+    async def list_custom_fields(self) -> List[Dict[str, Any]]:
+        """
+        List all custom fields.
+
+        Returns:
+            List of custom field definitions
+        """
+        try:
+            response = await self.client.get(f"{self.api_url}/custom_fields/")
+            response.raise_for_status()
+            return response.json().get("results", [])
+        except Exception as e:
+            logger.error(f"Failed to list custom fields: {e}")
+            raise
+
+    async def get_custom_field_by_name(self, name: str) -> Optional[Dict[str, Any]]:
+        """
+        Get a custom field by name.
+
+        Args:
+            name: Custom field name
+
+        Returns:
+            Custom field definition or None
+        """
+        fields = await self.list_custom_fields()
+        for field in fields:
+            if field.get("name") == name:
+                return field
+        return None
+
+    # =========================================================================
+    # Tags, Correspondents, Document Types
+    # =========================================================================
+
+    async def list_tags(self) -> List[Dict[str, Any]]:
+        """List all tags."""
+        try:
+            response = await self.client.get(f"{self.api_url}/tags/")
+            response.raise_for_status()
+            return response.json().get("results", [])
+        except Exception as e:
+            logger.error(f"Failed to list tags: {e}")
+            raise
+
+    async def list_correspondents(self) -> List[Dict[str, Any]]:
+        """List all correspondents."""
+        try:
+            response = await self.client.get(f"{self.api_url}/correspondents/")
+            response.raise_for_status()
+            return response.json().get("results", [])
+        except Exception as e:
+            logger.error(f"Failed to list correspondents: {e}")
+            raise
+
+    async def list_document_types(self) -> List[Dict[str, Any]]:
+        """List all document types."""
+        try:
+            response = await self.client.get(f"{self.api_url}/document_types/")
+            response.raise_for_status()
+            return response.json().get("results", [])
+        except Exception as e:
+            logger.error(f"Failed to list document types: {e}")
+            raise
+
+    # =========================================================================
+    # Bulk Operations
+    # =========================================================================
+
+    async def bulk_edit(
+        self,
+        document_ids: List[int],
+        method: str,
+        parameters: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        """
+        Bulk edit documents.
+
+        Args:
+            document_ids: List of document IDs
+            method: Operation (add_tag, remove_tag, set_correspondent, etc.)
+            parameters: Operation parameters
+
+        Returns:
+            Operation result
+        """
+        data = {
+            "documents": document_ids,
+            "method": method,
+        }
+        if parameters:
+            data["parameters"] = parameters
+
+        try:
+            response = await self.client.post(
+                f"{self.api_url}/documents/bulk_edit/",
+                json=data,
+            )
+            response.raise_for_status()
+            return response.json()
+        except Exception as e:
+            logger.error(f"Bulk edit failed: {e}")
+            raise
+
+    # =========================================================================
+    # Health Check
+    # =========================================================================
+
+    async def health_check(self) -> bool:
+        """
+        Check if Paperless-ngx is responding.
+
+        Returns:
+            True if service is healthy
+        """
+        try:
+            response = await self.client.get(f"{self.api_url}/", timeout=5.0)
+            return response.status_code < 400
+        except Exception as e:
+            logger.error(f"Paperless health check failed: {e}")
+            return False
+
+    # =========================================================================
+    # Helpers
+    # =========================================================================
+
+    def _parse_document(self, data: Dict[str, Any]) -> PaperlessDocument:
+        """Parse API response into PaperlessDocument."""
+        return PaperlessDocument(
+            id=data.get("id", 0),
+            title=data.get("title", ""),
+            content=data.get("content", ""),
+            created=data.get("created"),
+            modified=data.get("modified"),
+            added=data.get("added"),
+            correspondent=data.get("correspondent"),
+            document_type=data.get("document_type"),
+            storage_path=data.get("storage_path"),
+            tags=data.get("tags", []),
+            archive_serial_number=data.get("archive_serial_number"),
+            original_file_name=data.get("original_file_name"),
+            archived_file_name=data.get("archived_file_name"),
+            custom_fields=data.get("custom_fields", []),
+        )

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/config.py

@@ -101,6 +101,11 @@ class Settings(BaseSettings):
     content_extraction_timeout: int = Field(default=5, ge=1, le=30, description="Trafilatura per-URL timeout in seconds")
     content_max_length: int = Field(default=2000, ge=500, le=10000, description="Max extracted content length per result")
 
+    # Paperless-ngx Configuration
+    paperless_url: str = Field(default="http://paperless:8000", description="Paperless-ngx URL")
+    paperless_token: str = Field(default="", description="Paperless-ngx API token")
+    paperless_timeout: int = Field(default=30, ge=5, le=120, description="Paperless API timeout in seconds")
+
     # Document Store Configuration
     document_store_enabled: bool = Field(default=True, description="Enable document store feature")
     document_catalog_path_prefix: str = Field(default="docs", description="Wiki path prefix for catalog pages")

src/core/dependencies.py

@@ -22,6 +22,7 @@ from src.clients.wikijs_client import WikiJSClient
 from src.clients.searxng_client import SearXNGClient
 from src.clients.ollama_client import OllamaClient
 from src.clients.content_extractor import ContentExtractor
+from src.clients.paperless_client import PaperlessClient
 
 logger = logging.getLogger(__name__)
 
@@ -155,6 +156,28 @@ def get_content_extractor() -> ContentExtractor:
     return extractor
 
 
+@lru_cache
+def get_paperless_client() -> PaperlessClient:
+    """
+    Get Paperless-ngx client singleton.
+
+    Returns:
+        Initialized Paperless-ngx REST API client
+
+    Note: Returns None-like client if paperless_token is not configured
+    """
+    settings = get_settings()
+    if not settings.paperless_token:
+        logger.warning("Paperless token not configured - document storage disabled")
+    client = PaperlessClient(
+        base_url=settings.paperless_url,
+        token=settings.paperless_token,
+        timeout=settings.paperless_timeout
+    )
+    logger.debug(f"Created Paperless client: {settings.paperless_url}")
+    return client
+
+
 # Type aliases for FastAPI endpoint dependencies
 # Usage: def my_endpoint(neo4j: Neo4jDep):
 Neo4jDep = Annotated[Neo4jClient, Depends(get_neo4j_client)]
@@ -164,6 +187,7 @@ SearXNGDep = Annotated[SearXNGClient, Depends(get_searxng_client)]
 OllamaDep = Annotated[OllamaClient, Depends(get_ollama_client)]
 RedisDep = Annotated[aioredis.Redis, Depends(get_redis_client)]
 ContentExtractorDep = Annotated[ContentExtractor, Depends(get_content_extractor)]
+PaperlessDep = Annotated[PaperlessClient, Depends(get_paperless_client)]
 
 
 # Lifecycle management functions
@@ -202,6 +226,21 @@ async def startup_clients():
         logger.error(f"✗ Ollama health check failed: {e}")
         pass
 
+    # Check Paperless availability
+    settings = get_settings()
+    if settings.paperless_token:
+        try:
+            paperless = get_paperless_client()
+            is_healthy = await paperless.health_check()
+            if is_healthy:
+                logger.info(f"✓ Paperless-ngx ready: {settings.paperless_url}")
+            else:
+                logger.warning("✗ Paperless-ngx not responding")
+        except Exception as e:
+            logger.error(f"✗ Paperless health check failed: {e}")
+    else:
+        logger.info("○ Paperless-ngx not configured (document storage disabled)")
+
     # Qdrant, Wiki.js, SearXNG are lazy-initialized
     logger.info("Service clients startup complete")
 
@@ -230,7 +269,8 @@ async def shutdown_clients():
     clients_to_close = [
         ("Wiki.js", get_wikijs_client()),
         ("SearXNG", get_searxng_client()),
-        ("Ollama", get_ollama_client())
+        ("Ollama", get_ollama_client()),
+        ("Paperless", get_paperless_client()),
     ]
 
     for name, client in clients_to_close:
@@ -312,6 +352,18 @@ async def check_service_health() -> dict:
         logger.error(f"Ollama health check failed: {e}")
         health["ollama"] = False
 
+    # Paperless-ngx
+    settings = get_settings()
+    if settings.paperless_token:
+        try:
+            paperless = get_paperless_client()
+            health["paperless"] = await paperless.health_check()
+        except Exception as e:
+            logger.error(f"Paperless health check failed: {e}")
+            health["paperless"] = False
+    else:
+        health["paperless"] = None  # Not configured
+
     return health
 
 

src/main.py

@@ -51,7 +51,7 @@ app.add_middleware(
 from src.routers import (
     wiki, tools, graph, vector, hybrid_rag, consolidation,
     ingestion, entity_linking, webhooks, rag_search, content,
-    maintenance, volatile
+    maintenance, volatile, documents
 )
 
 app.include_router(wiki.router)
@@ -67,6 +67,7 @@ app.include_router(rag_search.router)
 app.include_router(content.router)
 app.include_router(maintenance.router)
 app.include_router(volatile.router)
+app.include_router(documents.router)
 
 # Mount static files directory for Wiki.js integration scripts
 static_dir = Path(__file__).parent.parent / "static"

src/models/document.py

@@ -0,0 +1,202 @@
+"""
+Document storage models for Library Desk.
+
+Models for Paperless-ngx document management, virus scanning,
+and document sync operations.
+"""
+
+from pydantic import BaseModel, Field
+from typing import Dict, Any, Optional, List
+from datetime import datetime
+from enum import Enum
+
+
+class DocumentType(str, Enum):
+    """Types of documents supported in the document store."""
+    PDF = "pdf"
+    IMAGE = "image"
+    VIDEO = "video"
+    TEXT = "text"
+    ARCHIVE = "archive"
+    OTHER = "other"
+
+
+class SyncStatus(str, Enum):
+    """Status of document sync with Library Desk."""
+    PENDING = "pending"
+    INDEXED = "indexed"
+    FAILED = "failed"
+    SKIPPED = "skipped"
+
+
+# =============================================================================
+# Document Models
+# =============================================================================
+
+
+class DocumentMetadata(BaseModel):
+    """Metadata for a document in Paperless-ngx."""
+    paperless_id: int = Field(..., description="Paperless-ngx document ID")
+    title: str = Field(..., description="Document title")
+    filename: Optional[str] = Field(None, description="Original filename")
+    content: Optional[str] = Field(None, description="Extracted text content")
+    created: Optional[datetime] = Field(None, description="Document creation date")
+    modified: Optional[datetime] = Field(None, description="Last modification date")
+    added: Optional[datetime] = Field(None, description="Date added to Paperless")
+    correspondent: Optional[str] = Field(None, description="Correspondent name")
+    document_type: Optional[str] = Field(None, description="Document type name")
+    tags: List[str] = Field(default_factory=list, description="Tag names")
+    custom_fields: Dict[str, Any] = Field(default_factory=dict, description="Custom field values")
+
+
+class DocumentRecord(BaseModel):
+    """A document record with sync status."""
+    metadata: DocumentMetadata = Field(..., description="Document metadata from Paperless")
+    sync_status: SyncStatus = Field(default=SyncStatus.PENDING, description="Library Desk sync status")
+    indexed_at: Optional[datetime] = Field(None, description="When indexed in Library Desk")
+    collection: Optional[str] = Field(None, description="Collection name (e.g., 'fastapi-docs')")
+    source_url: Optional[str] = Field(None, description="Original source URL if uploaded via HybridRAG")
+
+
+# =============================================================================
+# Upload Request/Response Models
+# =============================================================================
+
+
+class DocumentUploadRequest(BaseModel):
+    """Request to upload a document to Paperless-ngx."""
+    url: Optional[str] = Field(None, description="URL to download document from")
+    title: Optional[str] = Field(None, description="Document title (derived from filename if not set)")
+    collection: Optional[str] = Field(None, description="Collection to add document to")
+    tags: List[str] = Field(default_factory=list, description="Tags to apply")
+    correspondent: Optional[str] = Field(None, description="Correspondent name")
+    document_type: Optional[str] = Field(None, description="Document type name")
+
+
+class DocumentUploadResponse(BaseModel):
+    """Response from document upload."""
+    task_id: str = Field(..., description="Paperless task ID for tracking")
+    filename: str = Field(..., description="Uploaded filename")
+    message: str = Field(..., description="Status message")
+
+
+# =============================================================================
+# Webhook Models
+# =============================================================================
+
+
+class PaperlessWebhookPayload(BaseModel):
+    """Payload from Paperless-ngx webhook (include_document=true format)."""
+    id: int = Field(..., description="Paperless document ID")
+    title: Optional[str] = Field(None, description="Document title")
+    content: Optional[str] = Field(None, description="Document text content")
+    correspondent: Optional[int] = Field(None, description="Correspondent ID")
+    document_type: Optional[int] = Field(None, description="Document type ID")
+    tags: List[int] = Field(default_factory=list, description="Tag IDs")
+    created: Optional[str] = Field(None, description="Created date")
+    modified: Optional[str] = Field(None, description="Modified datetime")
+    added: Optional[str] = Field(None, description="Added datetime")
+    original_file_name: Optional[str] = Field(None, description="Original filename")
+    owner: Optional[int] = Field(None, description="Owner user ID")
+    custom_fields: List[Dict[str, Any]] = Field(default_factory=list, description="Custom fields")
+
+    class Config:
+        extra = "ignore"  # Ignore extra fields from Paperless
+
+
+class WebhookResponse(BaseModel):
+    """Response to webhook processing."""
+    document_id: int = Field(..., description="Processed document ID")
+    status: str = Field(..., description="Processing status")
+    indexed: bool = Field(..., description="Whether document was indexed")
+    message: Optional[str] = Field(None, description="Additional details")
+
+
+# =============================================================================
+# Sync Models
+# =============================================================================
+
+
+class SyncRequest(BaseModel):
+    """Request to sync documents from Paperless-ngx."""
+    since: Optional[datetime] = Field(None, description="Only sync documents modified after this time")
+    collection: Optional[str] = Field(None, description="Only sync documents in this collection")
+    limit: int = Field(default=100, ge=1, le=1000, description="Maximum documents to sync")
+    force_reindex: bool = Field(default=False, description="Re-index already indexed documents")
+
+
+class SyncResult(BaseModel):
+    """Result of a sync operation."""
+    documents_found: int = Field(..., description="Total documents matching criteria")
+    documents_indexed: int = Field(..., description="Successfully indexed")
+    documents_skipped: int = Field(..., description="Skipped (already indexed)")
+    documents_failed: int = Field(..., description="Failed to index")
+    errors: List[str] = Field(default_factory=list, description="Error messages")
+    duration_seconds: float = Field(..., description="Sync duration")
+
+
+# =============================================================================
+# Collection Models
+# =============================================================================
+
+
+class Collection(BaseModel):
+    """A logical grouping of documents."""
+    name: str = Field(..., description="Collection name (e.g., 'fastapi-docs')")
+    description: Optional[str] = Field(None, description="Collection description")
+    document_count: int = Field(default=0, description="Number of documents")
+    source: Optional[str] = Field(None, description="Source (e.g., 'github.com/tiangolo/fastapi')")
+    last_sync: Optional[datetime] = Field(None, description="Last sync timestamp")
+    wiki_page: Optional[str] = Field(None, description="Wiki catalog page path")
+
+
+class CollectionListResponse(BaseModel):
+    """Response listing all collections."""
+    collections: List[Collection] = Field(..., description="List of collections")
+    total_documents: int = Field(..., description="Total documents across all collections")
+
+
+# =============================================================================
+# Search Models
+# =============================================================================
+
+
+class DocumentSearchRequest(BaseModel):
+    """Request to search documents."""
+    query: str = Field(..., min_length=1, description="Search query")
+    collection: Optional[str] = Field(None, description="Limit to collection")
+    document_type: Optional[DocumentType] = Field(None, description="Filter by type")
+    limit: int = Field(default=10, ge=1, le=50, description="Maximum results")
+    include_content: bool = Field(default=False, description="Include full text content")
+
+
+class DocumentSearchHit(BaseModel):
+    """A document search result."""
+    paperless_id: int = Field(..., description="Paperless document ID")
+    title: str = Field(..., description="Document title")
+    score: float = Field(..., description="Relevance score")
+    highlights: Optional[str] = Field(None, description="Highlighted matching text")
+    collection: Optional[str] = Field(None, description="Collection name")
+    document_type: Optional[str] = Field(None, description="Document type")
+    content_preview: Optional[str] = Field(None, description="Content preview if requested")
+
+
+class DocumentSearchResponse(BaseModel):
+    """Response from document search."""
+    query: str = Field(..., description="Original query")
+    hits: List[DocumentSearchHit] = Field(..., description="Search results")
+    total: int = Field(..., description="Total matching documents")
+    duration_ms: int = Field(..., description="Search duration in milliseconds")
+
+
+# =============================================================================
+# Health Check Models
+# =============================================================================
+
+
+class DocumentStoreHealth(BaseModel):
+    """Health status of document storage components."""
+    paperless_healthy: bool = Field(..., description="Paperless-ngx responding")
+    paperless_version: Optional[str] = Field(None, description="Paperless version")
+    total_documents: Optional[int] = Field(None, description="Total documents in Paperless")
+    indexed_documents: Optional[int] = Field(None, description="Documents indexed in Library Desk")

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/documents.py

@@ -0,0 +1,300 @@
+"""
+Document storage router for Library Desk API.
+
+Event-driven integration with Paperless-ngx:
+- Webhook receiver triggers indexing after Paperless virus scan passes
+- Upload endpoint sends files to Paperless for processing
+- Search across indexed documents
+"""
+
+from fastapi import APIRouter, HTTPException, Depends, Query, UploadFile, File
+from typing import Optional
+import logging
+import time
+
+from src.models.document import (
+    PaperlessWebhookPayload,
+    WebhookResponse,
+    DocumentUploadRequest,
+    DocumentUploadResponse,
+    DocumentSearchRequest,
+    DocumentSearchResponse,
+    DocumentStoreHealth,
+)
+from src.core.dependencies import (
+    verify_api_key,
+    PaperlessDep,
+    QdrantDep,
+    OllamaDep,
+    Neo4jDep,
+    WikiJSDep,
+)
+from src.core.multi_tenancy import DEFAULT_USER
+from src.config import get_settings
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/documents", tags=["Documents"])
+
+
+# =============================================================================
+# Webhook Endpoint (primary integration - event-driven)
+# =============================================================================
+
+
+@router.post("/webhook", response_model=WebhookResponse)
+async def receive_webhook(
+    payload: PaperlessWebhookPayload,
+    paperless: PaperlessDep,
+    qdrant: QdrantDep,
+    ollama: OllamaDep,
+    neo4j: Neo4jDep,
+    wiki: WikiJSDep,
+    user: str = Query(default=DEFAULT_USER, description="User identifier"),
+):
+    """
+    Receive webhook events from Paperless-ngx.
+
+    This is the primary integration point. Configure Paperless workflow:
+    1. Trigger: Document Added (after consumption completes)
+    2. Condition: Document passed virus scan (ClamAV in Paperless)
+    3. Action: Webhook POST to this endpoint
+
+    Library Desk indexes the document into vectors and graph.
+    """
+    from src.services.document_sync_service import DocumentSyncService
+
+    logger.info(f"Webhook received: document_id={payload.id}, title={payload.title}")
+
+    settings = get_settings()
+    if not settings.document_store_enabled:
+        return WebhookResponse(
+            document_id=payload.id,
+            status="skipped",
+            indexed=False,
+            message="Document store is disabled"
+        )
+
+    try:
+        sync_service = DocumentSyncService(
+            paperless_client=paperless,
+            qdrant_client=qdrant,
+            ollama_client=ollama,
+            neo4j_client=neo4j,
+            wiki_client=wiki,
+            settings=settings
+        )
+
+        result = await sync_service.index_document(
+            document_id=payload.id,
+            user=user,
+            content=payload.content,  # Use content from webhook payload
+            title=payload.title,
+        )
+
+        return WebhookResponse(
+            document_id=payload.id,
+            status="indexed" if result.success else "failed",
+            indexed=result.success,
+            message=result.error if not result.success else f"Indexed: {result.title}"
+        )
+
+    except Exception as e:
+        logger.error(f"Webhook processing failed for document {payload.id}: {e}", exc_info=True)
+        return WebhookResponse(
+            document_id=payload.id,
+            status="error",
+            indexed=False,
+            message=str(e)
+        )
+
+
+# =============================================================================
+# Upload Endpoints
+# =============================================================================
+
+
+@router.post("/upload", response_model=DocumentUploadResponse)
+async def upload_document(
+    file: UploadFile = File(...),
+    title: Optional[str] = Query(None, description="Document title"),
+    collection: Optional[str] = Query(None, description="Collection name"),
+    paperless: PaperlessDep = None,
+    api_key: str = Depends(verify_api_key),
+):
+    """
+    Upload a document to Paperless-ngx.
+
+    Paperless handles virus scanning. If clean, Paperless webhook
+    triggers indexing back to Library Desk.
+    """
+    settings = get_settings()
+    if not settings.document_store_enabled:
+        raise HTTPException(status_code=503, detail="Document store is disabled")
+
+    content = await file.read()
+    filename = file.filename or "document"
+
+    custom_fields = []
+    if collection:
+        custom_fields.append({"field": "collection", "value": collection})
+
+    try:
+        task_id = await paperless.upload_document(
+            file_content=content,
+            filename=filename,
+            title=title,
+            custom_fields=custom_fields if custom_fields else None,
+        )
+
+        return DocumentUploadResponse(
+            task_id=task_id,
+            filename=filename,
+            message=f"Uploaded to Paperless, task {task_id}. Indexing via webhook after scan."
+        )
+
+    except Exception as e:
+        logger.error(f"Upload failed for '{filename}': {e}")
+        raise HTTPException(status_code=500, detail=f"Upload failed: {e}")
+
+
+@router.post("/upload-url", response_model=DocumentUploadResponse)
+async def upload_from_url(
+    request: DocumentUploadRequest,
+    paperless: PaperlessDep = None,
+    api_key: str = Depends(verify_api_key),
+):
+    """
+    Download document from URL and upload to Paperless-ngx.
+
+    Used by HybridRAG to save discovered PDFs. Paperless scans and
+    webhooks back for indexing.
+    """
+    import httpx
+
+    settings = get_settings()
+    if not settings.document_store_enabled:
+        raise HTTPException(status_code=503, detail="Document store is disabled")
+
+    if not request.url:
+        raise HTTPException(status_code=400, detail="URL is required")
+
+    try:
+        async with httpx.AsyncClient(timeout=60.0) as client:
+            response = await client.get(request.url, follow_redirects=True)
+            response.raise_for_status()
+            content = response.content
+            filename = request.url.split("/")[-1].split("?")[0] or "document"
+
+    except Exception as e:
+        logger.error(f"Download failed from {request.url}: {e}")
+        raise HTTPException(status_code=400, detail=f"Download failed: {e}")
+
+    try:
+        custom_fields = [{"field": "source_url", "value": request.url}]
+        if request.collection:
+            custom_fields.append({"field": "collection", "value": request.collection})
+
+        task_id = await paperless.upload_document(
+            file_content=content,
+            filename=filename,
+            title=request.title,
+            custom_fields=custom_fields,
+        )
+
+        return DocumentUploadResponse(
+            task_id=task_id,
+            filename=filename,
+            message=f"Uploaded from URL, task {task_id}. Indexing via webhook after scan."
+        )
+
+    except Exception as e:
+        logger.error(f"Upload failed for URL '{request.url}': {e}")
+        raise HTTPException(status_code=500, detail=f"Upload failed: {e}")
+
+
+# =============================================================================
+# Search
+# =============================================================================
+
+
+@router.post("/search", response_model=DocumentSearchResponse)
+async def search_documents(
+    request: DocumentSearchRequest,
+    qdrant: QdrantDep,
+    ollama: OllamaDep,
+    user: str = Query(default=DEFAULT_USER, description="User identifier"),
+    api_key: str = Depends(verify_api_key),
+):
+    """
+    Semantic search across indexed documents.
+    """
+    from src.services.vector_service import VectorService
+    from src.core.dependencies import get_wikijs_client
+    from src.models.document import DocumentSearchHit
+
+    settings = get_settings()
+    if not settings.document_store_enabled:
+        raise HTTPException(status_code=503, detail="Document store is disabled")
+
+    start_time = time.time()
+
+    try:
+        wiki = get_wikijs_client()
+        vector_service = VectorService(qdrant, wiki, ollama)
+
+        results = await vector_service.search(
+            query=request.query,
+            user=user,
+            limit=request.limit,
+            score_threshold=0.5,
+            doc_type="document"
+        )
+
+        hits = []
+        for result in results.get("results", []):
+            hits.append(DocumentSearchHit(
+                paperless_id=result.get("metadata", {}).get("paperless_id", 0),
+                title=result.get("title", ""),
+                score=result.get("score", 0.0),
+                highlights=result.get("chunk_text", "")[:200] if request.include_content else None,
+                collection=result.get("metadata", {}).get("collection"),
+                document_type=result.get("metadata", {}).get("document_type"),
+                content_preview=result.get("chunk_text", "")[:500] if request.include_content else None,
+            ))
+
+        return DocumentSearchResponse(
+            query=request.query,
+            hits=hits,
+            total=len(hits),
+            duration_ms=int((time.time() - start_time) * 1000)
+        )
+
+    except Exception as e:
+        logger.error(f"Document search failed: {e}", exc_info=True)
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+# =============================================================================
+# Health
+# =============================================================================
+
+
+@router.get("/health", response_model=DocumentStoreHealth)
+async def document_store_health(paperless: PaperlessDep):
+    """Check Paperless-ngx connectivity."""
+    settings = get_settings()
+
+    paperless_healthy = False
+    if settings.paperless_token:
+        try:
+            paperless_healthy = await paperless.health_check()
+        except Exception as e:
+            logger.error(f"Paperless health check failed: {e}")
+
+    return DocumentStoreHealth(
+        paperless_healthy=paperless_healthy,
+        paperless_version="connected" if paperless_healthy else None,
+        total_documents=None,
+        indexed_documents=None
+    )

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)
+    2. **Parallel Retrieval**: Search vector (Qdrant), graph (Neo4j), volatile cache, web (SearXNG)
-    3. **RRF Fusion**: Merge results with Reciprocal Rank Fusion
+    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)
+@router.get("/search")
-async def list_keys(
+async def search_volatile(
-    namespace: str,
+    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.
+    Searches all volatile data for semantically similar content.
-    """
+    Higher threshold = stricter matching.
-    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.
 
     **Example:**
     ```
-    GET /volatile/weather/rotterdam?user=jpmschweitzer
+    GET /volatile/search?q=weather%20rotterdam&user=jpmschweitzer
     ```
     """
-    service = get_volatile_service(redis)
+    service = get_volatile_service(qdrant, ollama)
-    record = await service.get(user, namespace, key)
+    results = await service.search(user, q, limit=limit, score_threshold=threshold)
 
-    if not record:
+    return {
-        raise HTTPException(
+        "query": q,
-            status_code=404,
+        "results": results,
-            detail=f"Record '{key}' not found in namespace '{namespace}'"
+        "count": len(results),
-        )
+        "user": user,
-
+    }
-    return record
 
 
-@router.post("/{namespace}/{key}", response_model=VolatileRecordResponse)
+@router.post("/store", response_model=VolatileRecordResponse)
-async def set_record(
+async def store_volatile(
-    namespace: str,
+    namespace: str = Query(..., description="Data namespace (weather, news, etc.)"),
-    key: str,
+    key: str = Query(..., description="Record key (e.g., 'rotterdam', 'nos-headlines')"),
-    request: VolatileRecordCreate,
+    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.
+    Data is converted to natural language and embedded for semantic search.
-    If TTL is not provided, the namespace default is used.
+    If the same namespace+key already exists, it will be updated.
 
     **Example Request:**
     ```json
+    POST /volatile/store?namespace=weather&key=rotterdam
     {
         "data": {
-            "temperature": 18,
+            "temperature": 8,
-            "conditions": "Partly cloudy",
+            "conditions": "Cloudy",
-            "humidity": 65
+            "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.
+    Delete a specific volatile record.
-
-    Removes the record from the cache.
     """
-    service = get_volatile_service(redis)
+    service = get_volatile_service(qdrant, ollama)
     deleted = await service.delete(user, namespace, key)
 
     return VolatileDeleteResponse(

src/services/document_sync_service.py

@@ -0,0 +1,290 @@
+"""
+Document sync service for Library Desk.
+
+Handles indexing of Paperless-ngx documents into vectors and graph.
+Called by webhook when Paperless completes document processing.
+"""
+
+import logging
+import re
+import hashlib
+import uuid
+from typing import Optional, List
+from dataclasses import dataclass
+
+from src.clients.paperless_client import PaperlessClient
+from src.clients.qdrant_client import QdrantClientWrapper
+from src.clients.ollama_client import OllamaClient
+from src.clients.neo4j_client import Neo4jClient
+from src.clients.wikijs_client import WikiJSClient
+from src.core.multi_tenancy import get_qdrant_collection_name
+from src.config import Settings
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class IndexResult:
+    """Result of indexing a single document."""
+    success: bool
+    document_id: int
+    title: str = ""
+    chunks_created: int = 0
+    error: Optional[str] = None
+
+
+class DocumentSyncService:
+    """
+    Service for syncing Paperless documents to Library Desk indexes.
+
+    Handles:
+    - Fetching document content from Paperless API
+    - Chunking and embedding into Qdrant
+    - Creating graph nodes in Neo4j
+    """
+
+    def __init__(
+        self,
+        paperless_client: PaperlessClient,
+        qdrant_client: QdrantClientWrapper,
+        ollama_client: OllamaClient,
+        neo4j_client: Neo4jClient,
+        wiki_client: WikiJSClient,
+        settings: Settings,
+        chunk_size: int = 500,
+        chunk_overlap: int = 50
+    ):
+        self.paperless = paperless_client
+        self.qdrant = qdrant_client
+        self.ollama = ollama_client
+        self.neo4j = neo4j_client
+        self.wiki = wiki_client
+        self.settings = settings
+        self.chunk_size = chunk_size
+        self.chunk_overlap = chunk_overlap
+
+    def _chunk_text(self, text: str) -> List[str]:
+        """Chunk text into overlapping segments."""
+        text = re.sub(r'\s+', ' ', text).strip()
+        words = text.split()
+
+        if len(words) <= self.chunk_size:
+            return [text] if text else []
+
+        chunks = []
+        start = 0
+
+        while start < len(words):
+            end = start + self.chunk_size
+            chunk_words = words[start:end]
+            chunks.append(' '.join(chunk_words))
+            start = end - self.chunk_overlap
+
+        return chunks
+
+    async def index_document(
+        self,
+        document_id: int,
+        user: str,
+        content: Optional[str] = None,
+        title: Optional[str] = None,
+    ) -> IndexResult:
+        """
+        Index a single document from Paperless into vectors and graph.
+
+        Args:
+            document_id: Paperless document ID
+            user: User identifier for multi-tenancy
+            content: Optional document content (if provided, skip Paperless API call)
+            title: Optional document title (if provided, skip Paperless API call)
+
+        Returns:
+            IndexResult with success status and details
+        """
+        logger.info(f"Indexing document {document_id} for user {user}")
+
+        try:
+            # If content and title provided (from webhook), skip API call
+            if content is not None and title is not None:
+                doc_title = title
+                doc_content = content
+                original_filename = None
+                correspondent = None
+                document_type = None
+                tags = []
+            else:
+                # Fetch document from Paperless
+                doc = await self.paperless.get_document(document_id)
+                if not doc:
+                    return IndexResult(
+                        success=False,
+                        document_id=document_id,
+                        error="Document not found in Paperless"
+                    )
+                doc_title = doc.title
+                doc_content = doc.content or ""
+                original_filename = doc.original_file_name
+                correspondent = doc.correspondent
+                document_type = doc.document_type
+                tags = doc.tags
+
+            if not doc_content.strip():
+                logger.warning(f"Document {document_id} has no text content")
+                return IndexResult(
+                    success=True,
+                    document_id=document_id,
+                    title=doc_title,
+                    chunks_created=0,
+                    error="No text content (possibly image/video only)"
+                )
+
+            # Index vectors
+            chunks_created = await self._index_vectors(
+                document_id=document_id,
+                title=doc_title,
+                content=doc_content,
+                user=user,
+                metadata={
+                    "paperless_id": document_id,
+                    "original_filename": original_filename,
+                    "correspondent": correspondent,
+                    "document_type": document_type,
+                    "tags": tags,
+                }
+            )
+
+            # Index graph node
+            await self._index_graph(
+                document_id=document_id,
+                title=doc_title,
+                content=doc_content,
+                user=user,
+            )
+
+            # Mark as indexed in Paperless (optional - if custom field exists)
+            try:
+                await self._mark_indexed(document_id)
+            except Exception as e:
+                logger.debug(f"Could not mark document as indexed: {e}")
+
+            logger.info(f"Successfully indexed document {document_id}: {chunks_created} chunks")
+
+            return IndexResult(
+                success=True,
+                document_id=document_id,
+                title=doc_title,
+                chunks_created=chunks_created
+            )
+
+        except Exception as e:
+            logger.error(f"Failed to index document {document_id}: {e}", exc_info=True)
+            return IndexResult(
+                success=False,
+                document_id=document_id,
+                error=str(e)
+            )
+
+    async def _index_vectors(
+        self,
+        document_id: int,
+        title: str,
+        content: str,
+        user: str,
+        metadata: dict,
+    ) -> int:
+        """Create vector embeddings for document content."""
+        collection = get_qdrant_collection_name(user)
+        self.qdrant.ensure_collection(collection)
+
+        # Delete existing chunks for this document
+        try:
+            self.qdrant.client.delete(
+                collection_name=collection,
+                points_selector={
+                    "filter": {
+                        "must": [
+                            {"key": "doc_type", "match": {"value": "document"}},
+                            {"key": "paperless_id", "match": {"value": document_id}},
+                        ]
+                    }
+                }
+            )
+        except Exception as e:
+            logger.debug(f"No existing chunks to delete: {e}")
+
+        # Chunk content
+        chunks = self._chunk_text(content)
+        if not chunks:
+            return 0
+
+        # Generate embeddings
+        embeddings = await self.ollama.embed_batch(chunks)
+
+        # Build points
+        points = []
+        for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)):
+            point_id = str(uuid.uuid4())
+            content_hash = hashlib.md5(chunk.encode()).hexdigest()
+
+            points.append({
+                "id": point_id,
+                "vector": embedding,
+                "payload": {
+                    "doc_type": "document",
+                    "paperless_id": document_id,
+                    "title": title,
+                    "chunk_text": chunk,
+                    "chunk_index": i,
+                    "content_hash": content_hash,
+                    **metadata
+                }
+            })
+
+        # Upsert to Qdrant
+        if points:
+            self.qdrant.client.upsert(
+                collection_name=collection,
+                points=points
+            )
+
+        return len(points)
+
+    async def _index_graph(
+        self,
+        document_id: int,
+        title: str,
+        content: str,
+        user: str,
+    ):
+        """Create graph node for document."""
+        # Create Document node in Neo4j
+        query = """
+        MERGE (d:Document {paperless_id: $paperless_id, user: $user})
+        SET d.title = $title,
+            d.doc_type = 'document',
+            d.updated_at = datetime()
+        RETURN d
+        """
+        await self.neo4j.execute_query(
+            query,
+            {
+                "paperless_id": document_id,
+                "user": user,
+                "title": title,
+            }
+        )
+
+        # TODO: Extract entities from content and create relationships
+        # This could use the same entity extraction as wiki pages
+
+    async def _mark_indexed(self, document_id: int):
+        """Mark document as indexed in Paperless custom field."""
+        # Try to update library_indexed custom field if it exists
+        try:
+            await self.paperless.update_document(
+                document_id=document_id,
+                custom_fields=[{"field": "library_indexed", "value": True}]
+            )
+        except Exception:
+            # Field might not exist, that's OK
+            pass

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
+        Wiki results are pre-merged from vector+graph. Volatile results
-        now compete on equal footing.
+        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.
+    Stores ephemeral data as vectors for semantic search retrieval.
-    Supports multiple namespaces with configurable TTLs.
+    Each user has an isolated volatile collection.
     """
 
-    # Redis key prefix for volatile data
+    COLLECTION_PREFIX = "volatile_"
-    KEY_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}
+        Same namespace+key always produces same ID for upsert behavior.
-        Uses hash to ensure safe key characters and consistent length.
         """
-        key_hash = hashlib.md5(key.encode()).hexdigest()[:12]
+        combined = f"{namespace}:{key}"
-        return f"{user}:{self.KEY_PREFIX}:{namespace}:{key_hash}"
+        return hashlib.md5(combined.encode()).hexdigest()
-
-    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}:*"
 
     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:
+    def _current_timestamp_ms(self) -> int:
-        """Serialize record to JSON for storage."""
+        """Get current timestamp in milliseconds."""
-        return json.dumps({
+        return int(time.time() * 1000)
-            "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 _deserialize_record(self, data: str) -> VolatileRecord:
+    def _to_natural_language(
-        """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(
         self,
-        user: str,
         namespace: str,
-        key: str
+        key: str,
-    ) -> Optional[VolatileRecordResponse]:
+        data: Dict[str, Any]
+    ) -> str:
         """
-        Get a volatile record.
+        Convert structured data to natural language for embedding.
 
-        Args:
+        This creates a text representation that embeds well semantically.
-            user: User identifier
-            namespace: Data namespace
-            key: Record key
-
-        Returns:
-            Record if found and not expired, None otherwise
         """
-        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:
+        elif namespace == VolatileNamespace.NEWS:
-            data = await self.redis.get(redis_key)
+            title = data.get("title", data.get("headline", ""))
-            if not data:
+            summary = data.get("summary", data.get("description", ""))
-                return None
+            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
+        elif namespace == VolatileNamespace.TRANSIT:
-            ttl_remaining = await self.redis.ttl(redis_key)
+            route = data.get("route", data.get("line", key))
-            if ttl_remaining < 0:
+            status = data.get("status", "")
-                return None
+            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(
+        elif namespace == VolatileNamespace.TRAFFIC:
-                key=record.key,
+            location = data.get("location", key)
-                namespace=record.namespace,
+            duration = data.get("duration", data.get("travel_time", ""))
-                data=record.data,
+            congestion = data.get("congestion", "")
-                source=record.source,
+            text = f"Traffic for {location}"
-                created_at=record.created_at,
+            if duration:
-                updated_at=record.updated_at,
+                text += f": {duration} minutes"
-                ttl=record.ttl,
+            if congestion:
-                ttl_remaining=max(0, ttl_remaining),
+                text += f", congestion level {congestion}"
-                refresh_schedule=record.refresh_schedule,
+            return text
-                user=record.user,
-            )
 
-        except Exception as e:
+        elif namespace == VolatileNamespace.AIR_QUALITY:
-            logger.error(f"Failed to get volatile record {redis_key}: {e}")
+            location = data.get("location", key)
-            return None
+            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
+            namespace: Data namespace (from controlled list)
-            key: Record key
+            key: Record key (normalized slug)
-            data: Content to store
+            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)
+        # Convert to natural language for embedding
-        existing = await self.get(user, namespace, key)
+        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:
+    async def search(
-            serialized = self._serialize_record(record)
+        self,
-            await self.redis.setex(redis_key, effective_ttl, serialized)
+        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(
+        Returns:
-                key=record.key,
+            List of matching volatile records
-                namespace=record.namespace,
+        """
-                data=record.data,
+        collection = self._collection_name(user)
-                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,
-            )
 
-        except Exception as e:
+        # Check if collection exists
-            logger.error(f"Failed to store volatile record {redis_key}: {e}")
+        if not await self.qdrant.collection_exists(collection):
-            raise
+            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:
+        if not await self.qdrant.collection_exists(collection):
-            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}")
             return False
 
-    async def list_namespace(
+        vector_id = self._make_vector_id(namespace, key)
-        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)
 
         try:
-            keys = []
+            deleted = await self.qdrant.delete_by_ids(
-            async for redis_key in self.redis.scan_iter(match=pattern):
+                collection_name=collection,
-                # Get the record to retrieve original key
+                point_ids=[vector_id]
-                data = await self.redis.get(redis_key)
+            )
-                if data:
+            return deleted > 0
-                    record = self._deserialize_record(data)
-                    keys.append(record.key)
-
-            return keys
-
         except Exception as e:
-            logger.error(f"Failed to list namespace {namespace}: {e}")
+            logger.error(f"Failed to delete volatile {namespace}:{key}: {e}")
-            return []
+            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 = []
+            all_points = await self.qdrant.scroll_all_points(
-            async for redis_key in self.redis.scan_iter(match=pattern):
+                collection_name=collection,
-                data = await self.redis.get(redis_key)
+                with_payload=True
-                if data:
+            )
-                    record = self._deserialize_record(data)
+
-                    if record.refresh_schedule:
+            for point in all_points:
-                        ttl_remaining = await self.redis.ttl(redis_key)
+                payload = point.get("payload", {})
-                        scheduled.append(VolatileRecordResponse(
+                ttl_expiry = payload.get("ttl_expiry", 0)
-                            key=record.key,
+
-                            namespace=record.namespace,
+                # Skip expired
-                            data=record.data,
+                if ttl_expiry <= now_ms:
-                            source=record.source,
+                    continue
-                            created_at=record.created_at,
+
-                            updated_at=record.updated_at,
+                # Only include if has refresh schedule
-                            ttl=record.ttl,
+                if payload.get("refresh_schedule"):
-                            ttl_remaining=max(0, ttl_remaining),
+                    ttl_remaining = max(0, (ttl_expiry - now_ms) // 1000)
-                            refresh_schedule=record.refresh_schedule,
+                    scheduled.append(VolatileRecordResponse(
-                            user=record.user,
+                        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] = {}
+            all_points = await self.qdrant.scroll_all_points(
-            total = 0
+                collection_name=collection,
-            scheduled = 0
+                with_payload=True
+            )
 
-            async for redis_key in self.redis.scan_iter(match=pattern):
+            for point in all_points:
-                data = await self.redis.get(redis_key)
+                payload = point.get("payload", {})
-                if data:
+                namespace = payload.get("namespace", "unknown")
-                    record = self._deserialize_record(data)
+                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
+                    by_namespace[namespace] = by_namespace.get(namespace, 0) + 1
-                    if record.refresh_schedule:
+                    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(
+    async def purge_expired(
-        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(
         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:
+        if not await self.qdrant.collection_exists(collection):
-            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}")
             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):
+    def mock_qdrant(self):
-        """Create mock Redis client."""
+        """Create mock Qdrant client."""
-        redis = AsyncMock()
+        qdrant = AsyncMock()
-        redis.get = AsyncMock(return_value=None)
+        qdrant.ensure_collection = AsyncMock()
-        redis.setex = AsyncMock()
+        qdrant.collection_exists = AsyncMock(return_value=True)
-        redis.delete = AsyncMock(return_value=1)
+        qdrant.upsert_vector = AsyncMock(return_value=True)
-        redis.ttl = AsyncMock(return_value=1500)
+        qdrant.delete_by_ids = AsyncMock(return_value=1)
-        redis.scan_iter = MagicMock(return_value=iter([]))
+        qdrant.search_with_expiry_filter = AsyncMock(return_value=[])
-        return redis
+        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):
+    def test_collection_name(self, volatile_service):
-        """Test Redis key building."""
+        """Test collection naming pattern."""
-        key = volatile_service._build_key("jpmschweitzer", "weather", "rotterdam")
+        name = volatile_service._collection_name("jpmschweitzer")
-        assert key.startswith("jpmschweitzer:volatile:weather:")
+        assert name == "volatile_jpmschweitzer"
-        assert len(key) > 30  # Has hash suffix
 
-    def test_build_pattern(self, volatile_service):
+    def test_make_vector_id(self, volatile_service):
-        """Test pattern building."""
+        """Test deterministic vector ID generation."""
-        pattern = volatile_service._build_pattern("jpmschweitzer", "weather")
+        id1 = volatile_service._make_vector_id("weather", "rotterdam")
-        assert pattern == "jpmschweitzer:volatile:weather:*"
+        id2 = volatile_service._make_vector_id("weather", "rotterdam")
+        id3 = volatile_service._make_vector_id("weather", "amsterdam")
 
-    def test_build_pattern_all(self, volatile_service):
+        assert id1 == id2  # Same namespace+key = same ID
-        """Test pattern building for all namespaces."""
+        assert id1 != id3  # Different key = different ID
-        pattern = volatile_service._build_pattern("jpmschweitzer")
+        assert len(id1) == 32  # MD5 hex length
-        assert pattern == "jpmschweitzer:volatile:*"
 
     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
+    def test_to_natural_language_weather(self, volatile_service):
-    async def test_get_not_found(self, volatile_service, mock_redis):
+        """Test natural language conversion for weather data."""
-        """Test get when record not found."""
+        text = volatile_service._to_natural_language(
-        mock_redis.get.return_value = None
+            namespace="weather",
-        result = await volatile_service.get("jpmschweitzer", "weather", "rotterdam")
+            key="rotterdam",
-        assert result is None
+            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