Merge pull request #306 from cbcoutinho/rag-evaluation

feat: RAG evaluation framework with performance improvements

nextcloud_mcp_server/app.py

@@ -446,7 +446,7 @@ async def app_lifespan_basic(server: FastMCP) -> AsyncIterator[AppContext]:
         # Start background tasks using anyio TaskGroup
         async with anyio.create_task_group() as tg:
             # Start scanner task
-            tg.start_soon(
+            await tg.start(
                 scanner_task,
                 send_stream,
                 shutdown_event,
@@ -457,7 +457,7 @@ async def app_lifespan_basic(server: FastMCP) -> AsyncIterator[AppContext]:
 
             # Start processor pool (each gets a cloned receive stream)
             for i in range(settings.vector_sync_processor_workers):
-                tg.start_soon(
+                await tg.start(
                     processor_task,
                     i,
                     receive_stream.clone(),
@@ -1147,7 +1147,7 @@ def get_app(transport: str = "sse", enabled_apps: list[str] | None = None):
                 # Start background tasks using anyio TaskGroup
                 async with anyio_module.create_task_group() as tg:
                     # Start scanner task
-                    tg.start_soon(
+                    await tg.start(
                         scanner_task,
                         send_stream,
                         shutdown_event,
@@ -1158,7 +1158,7 @@ def get_app(transport: str = "sse", enabled_apps: list[str] | None = None):
 
                     # Start processor pool (each gets a cloned receive stream)
                     for i in range(settings.vector_sync_processor_workers):
-                        tg.start_soon(
+                        await tg.start(
                             processor_task,
                             i,
                             receive_stream.clone(),

nextcloud_mcp_server/auth/storage.py

@@ -1310,7 +1310,7 @@ async def generate_encryption_key() -> str:
 
 # Example usage
 if __name__ == "__main__":
-    import asyncio
+    import anyio
 
     async def main():
         # Generate a key for testing
@@ -1318,4 +1318,4 @@ if __name__ == "__main__":
         print(f"Generated encryption key: {key}")
         print(f"Set this in your environment: export TOKEN_ENCRYPTION_KEY='{key}'")
 
-    asyncio.run(main())
+    anyio.run(main)

nextcloud_mcp_server/auth/token_broker.py

@@ -14,11 +14,11 @@ The Token Broker provides:
 - Session vs background token separation (RFC 8693)
 """
 
-import asyncio
 import logging
 from datetime import datetime, timedelta, timezone
 from typing import Dict, Optional, Tuple
 
+import anyio
 import httpx
 import jwt
 from cryptography.fernet import Fernet
@@ -43,7 +43,7 @@ class TokenCache:
         self._cache: Dict[str, Tuple[str, datetime]] = {}
         self._ttl = timedelta(seconds=ttl_seconds)
         self._early_refresh = timedelta(seconds=early_refresh_seconds)
-        self._lock = asyncio.Lock()
+        self._lock = anyio.Lock()
 
     async def get(self, user_id: str) -> Optional[str]:
         """Get cached token if valid."""

nextcloud_mcp_server/auth/userinfo_routes.py

@@ -725,6 +725,11 @@ async def user_info_html(request: Request) -> HTMLResponse:
                     }},
 
                     renderPlot(coordinates, results) {{
+                        // Calculate score range for auto-scaling
+                        const scores = results.map(r => r.score);
+                        const minScore = Math.min(...scores);
+                        const maxScore = Math.max(...scores);
+
                         const trace = {{
                             x: coordinates.map(c => c[0]),
                             y: coordinates.map(c => c[1]),
@@ -732,11 +737,18 @@ async def user_info_html(request: Request) -> HTMLResponse:
                             type: 'scatter',
                             text: results.map(r => `${{r.title}}<br>Score: ${{r.score.toFixed(3)}}`),
                             marker: {{
-                                size: 8,
-                                color: results.map(r => r.score),
+                                // Multi-channel encoding: size + opacity + color for visual hierarchy
+                                // Power scaling (score^2) amplifies visual differences dramatically
+                                // score=0.0 → 6px, score=0.5 → 9.5px, score=1.0 → 20px
+                                size: results.map(r => 6 + (Math.pow(r.score, 2) * 14)),
+                                // Linear opacity scaling (0.2-1.0 range keeps all points visible)
+                                opacity: results.map(r => 0.2 + (r.score * 0.8)),
+                                // Color gradient shows score
+                                color: scores,
                                 colorscale: 'Viridis',
                                 showscale: true,
-                                colorbar: {{ title: 'Score' }},
+                                colorbar: {{ title: 'Relative Score' }},
+                                // Scores are normalized 0-1 within result set
                                 cmin: 0,
                                 cmax: 1
                             }}

nextcloud_mcp_server/auth/viz_routes.py

@@ -11,6 +11,7 @@ All processing happens server-side following ADR-012:
 """
 
 import logging
+import time
 
 import numpy as np
 from starlette.authentication import requires
@@ -381,56 +382,17 @@ async def vector_visualization_search(request: Request) -> JSONResponse:
     )
 
     try:
+        # Start total request timer
+        request_start = time.perf_counter()
         # Get authenticated HTTP client from session
         # In BasicAuth mode: uses username/password from session
         # In OAuth mode: uses access token from session
         from nextcloud_mcp_server.auth.userinfo_routes import (
             _get_authenticated_client_for_userinfo,
         )
-        from nextcloud_mcp_server.client.notes import NotesClient
 
-        async with await _get_authenticated_client_for_userinfo(request) as http_client:
-            # Create NotesClient directly with authenticated HTTP client
-            notes_client = NotesClient(http_client, username)
-
-            # Wrap in a minimal client object for search algorithms
-            # This conforms to NextcloudClientProtocol but only implements notes
-            class MinimalNextcloudClient:
-                def __init__(self, notes_client, username):
-                    self._notes = notes_client
-                    self.username = username
-
-                @property
-                def notes(self):
-                    return self._notes
-
-                @property
-                def webdav(self):
-                    return None
-
-                @property
-                def calendar(self):
-                    return None
-
-                @property
-                def contacts(self):
-                    return None
-
-                @property
-                def deck(self):
-                    return None
-
-                @property
-                def cookbook(self):
-                    return None
-
-                @property
-                def tables(self):
-                    return None
-
-            nextcloud_client = MinimalNextcloudClient(notes_client, username)
-
-            # Create search algorithm
+        async with await _get_authenticated_client_for_userinfo(request) as http_client:  # noqa: F841
+            # Create search algorithm (no client needed - verification removed)
             if algorithm == "semantic":
                 search_algo = SemanticSearchAlgorithm(score_threshold=score_threshold)
             elif algorithm == "keyword":
@@ -451,6 +413,7 @@ async def vector_visualization_search(request: Request) -> JSONResponse:
 
             # Execute search (supports cross-app when doc_types=None)
             # Get unverified results with buffer for filtering
+            search_start = time.perf_counter()
             all_results = []
             if doc_types is None or len(doc_types) == 0:
                 # Cross-app search - search all indexed types
@@ -476,13 +439,28 @@ async def vector_visualization_search(request: Request) -> JSONResponse:
                 # Sort by score before verification
                 all_results.sort(key=lambda r: r.score, reverse=True)
 
-            # Verify access for all results (deduplicates and filters)
-            from nextcloud_mcp_server.search.verification import verify_search_results
+            # No verification needed for visualization - we only need Qdrant metadata
+            # (title, excerpt, doc_type) which is already in search results.
+            # Verification is only needed for sampling (LLM needs full content).
+            search_results = all_results[:limit]
+            search_duration = time.perf_counter() - search_start
 
-            verified_results = await verify_search_results(
-                all_results, nextcloud_client
+        # Normalize scores relative to this result set for better visualization
+        # (best result = 1.0, worst result = 0.0 within THIS result set)
+        # This makes visual encoding meaningful regardless of RRF normalization
+        if search_results:
+            scores = [r.score for r in search_results]
+            min_score, max_score = min(scores), max(scores)
+            score_range = max_score - min_score if max_score > min_score else 1.0
+
+            logger.info(
+                f"Normalizing scores for viz: original range [{min_score:.3f}, {max_score:.3f}] "
+                f"→ [0.0, 1.0]"
             )
-            search_results = verified_results[:limit]
+
+            # Rescale each result's score to 0-1 within this result set
+            for r in search_results:
+                r.score = (r.score - min_score) / score_range
 
         if not search_results:
             return JSONResponse(
@@ -495,6 +473,7 @@ async def vector_visualization_search(request: Request) -> JSONResponse:
             )
 
         # Fetch vectors for matching results from Qdrant
+        vector_fetch_start = time.perf_counter()
         qdrant_client = await get_qdrant_client()
         doc_ids = [r.id for r in search_results]
 
@@ -534,6 +513,7 @@ async def vector_visualization_search(request: Request) -> JSONResponse:
 
         # Extract vectors
         vectors = np.array([p.vector for p in points if p.vector is not None])
+        vector_fetch_duration = time.perf_counter() - vector_fetch_start
 
         if len(vectors) < 2:
             # Not enough points for PCA
@@ -556,8 +536,10 @@ async def vector_visualization_search(request: Request) -> JSONResponse:
             )
 
         # Apply PCA dimensionality reduction (768-dim → 2D)
+        pca_start = time.perf_counter()
         pca = PCA(n_components=2)
         coords_2d = pca.fit_transform(vectors)
+        pca_duration = time.perf_counter() - pca_start
 
         # After fit, these attributes are guaranteed to be set
         assert pca.explained_variance_ratio_ is not None
@@ -590,6 +572,18 @@ async def vector_visualization_search(request: Request) -> JSONResponse:
             for r in search_results
         ]
 
+        # Calculate total request duration
+        total_duration = time.perf_counter() - request_start
+
+        # Log comprehensive timing metrics
+        logger.info(
+            f"Viz search timing: total={total_duration * 1000:.1f}ms, "
+            f"search={search_duration * 1000:.1f}ms ({search_duration / total_duration * 100:.1f}%), "
+            f"vector_fetch={vector_fetch_duration * 1000:.1f}ms ({vector_fetch_duration / total_duration * 100:.1f}%), "
+            f"pca={pca_duration * 1000:.1f}ms ({pca_duration / total_duration * 100:.1f}%), "
+            f"results={len(search_results)}, vectors={len(vectors)}"
+        )
+
         return JSONResponse(
             {
                 "success": True,
@@ -599,6 +593,14 @@ async def vector_visualization_search(request: Request) -> JSONResponse:
                     "pc1": float(pca.explained_variance_ratio_[0]),
                     "pc2": float(pca.explained_variance_ratio_[1]),
                 },
+                "timing": {
+                    "total_ms": round(total_duration * 1000, 2),
+                    "search_ms": round(search_duration * 1000, 2),
+                    "vector_fetch_ms": round(vector_fetch_duration * 1000, 2),
+                    "pca_ms": round(pca_duration * 1000, 2),
+                    "num_results": len(search_results),
+                    "num_vectors": len(vectors),
+                },
             }
         )
 

nextcloud_mcp_server/client/base.py

@@ -5,6 +5,7 @@ import time
 from abc import ABC
 from functools import wraps
 
+import anyio
 from httpx import AsyncClient, HTTPStatusError, RequestError, codes
 
 from nextcloud_mcp_server.observability.metrics import (
@@ -47,7 +48,7 @@ def retry_on_429(func):
                     # Record retry metric (extract app name from args if available)
                     if len(args) > 0 and hasattr(args[0], "app_name"):
                         record_nextcloud_api_retry(app=args[0].app_name, reason="429")
-                    time.sleep(5)
+                    await anyio.sleep(5)
                 elif e.response.status_code == 404:
                     # 404 errors are often expected (e.g., checking if attachments exist)
                     # Log as debug instead of warning

nextcloud_mcp_server/client/notes.py

@@ -40,7 +40,7 @@ class NotesClient(BaseNextcloudClient):
         seen_ids: set[int] = set()
 
         while True:
-            params: Dict[str, Any] = {"chunkSize": 10}
+            params: Dict[str, Any] = {"chunkSize": 100}
             if cursor:
                 params["chunkCursor"] = cursor
             if prune_before is not None:

nextcloud_mcp_server/observability/logging_config.py

@@ -39,7 +39,12 @@ class HealthCheckFilter(logging.Filter):
         message = record.getMessage()
         return not any(
             endpoint in message
-            for endpoint in ["/health/live", "/health/ready", "/metrics"]
+            for endpoint in [
+                "/health/live",
+                "/health/ready",
+                "/metrics",
+                "/app/vector-sync/status",
+            ]
         )
 
 

nextcloud_mcp_server/search/hybrid.py

@@ -1,10 +1,11 @@
 """Hybrid search algorithm using Reciprocal Rank Fusion (RRF)."""
 
-import asyncio
 import logging
 from collections import defaultdict
 from typing import Any
 
+import anyio
+
 from nextcloud_mcp_server.search.algorithms import SearchAlgorithm, SearchResult
 from nextcloud_mcp_server.search.fuzzy import FuzzySearchAlgorithm
 from nextcloud_mcp_server.search.keyword import KeywordSearchAlgorithm
@@ -105,30 +106,70 @@ class HybridSearchAlgorithm(SearchAlgorithm):
             f"fuzzy={self.fuzzy_weight})"
         )
 
-        # Run algorithms in parallel
-        tasks = []
-        algo_names = []
-
+        # Prepare algorithm configurations for parallel execution
+        algo_configs = []
         if self.semantic_weight > 0:
-            tasks.append(
-                self.semantic.search(query, user_id, limit * 2, doc_type, **kwargs)
+            algo_configs.append(
+                (
+                    "semantic",
+                    self.semantic.search,
+                    query,
+                    user_id,
+                    limit * 2,
+                    doc_type,
+                    kwargs,
+                )
             )
-            algo_names.append("semantic")
-
         if self.keyword_weight > 0:
-            tasks.append(
-                self.keyword.search(query, user_id, limit * 2, doc_type, **kwargs)
+            algo_configs.append(
+                (
+                    "keyword",
+                    self.keyword.search,
+                    query,
+                    user_id,
+                    limit * 2,
+                    doc_type,
+                    kwargs,
+                )
             )
-            algo_names.append("keyword")
-
         if self.fuzzy_weight > 0:
-            tasks.append(
-                self.fuzzy.search(query, user_id, limit * 2, doc_type, **kwargs)
+            algo_configs.append(
+                (
+                    "fuzzy",
+                    self.fuzzy.search,
+                    query,
+                    user_id,
+                    limit * 2,
+                    doc_type,
+                    kwargs,
+                )
             )
-            algo_names.append("fuzzy")
 
-        # Execute searches in parallel
-        results_list = await asyncio.gather(*tasks)
+        # Pre-allocate results list and extract algorithm names
+        results_list = [None] * len(algo_configs)
+        algo_names = [name for name, *_ in algo_configs]
+
+        async def search_one(
+            index: int,
+            search_func,
+            query_arg: str,
+            user_id_arg: str,
+            limit_arg: int,
+            doc_type_arg: str | None,
+            kwargs_arg: dict,
+        ):
+            """Execute one search algorithm and store result at index."""
+            result = await search_func(
+                query_arg, user_id_arg, limit_arg, doc_type_arg, **kwargs_arg
+            )
+            results_list[index] = result
+
+        # Execute searches in parallel using anyio task group
+        async with anyio.create_task_group() as tg:
+            for idx, (name, search_func, q, uid, lim, dt, kw) in enumerate(
+                algo_configs
+            ):
+                tg.start_soon(search_one, idx, search_func, q, uid, lim, dt, kw)
 
         # Build results dict
         algo_results = {}

nextcloud_mcp_server/search/verification.py

@@ -1,122 +0,0 @@
-"""Access verification for search results.
-
-This module provides centralized verification of Nextcloud access permissions
-for search results. Verification happens at the final output stage (MCP tool/viz endpoint)
-rather than within individual search algorithms, preventing redundant API calls.
-
-Key benefits:
-- Deduplication: Each document verified exactly once (even in hybrid mode)
-- Parallel execution: All verifications run concurrently via anyio task groups
-- Separation of concerns: Algorithms handle scoring, this module handles security
-"""
-
-import logging
-from dataclasses import replace
-from typing import Protocol
-
-import anyio
-
-from nextcloud_mcp_server.search.algorithms import SearchResult
-
-logger = logging.getLogger(__name__)
-
-
-class NextcloudClientProtocol(Protocol):
-    """Protocol for Nextcloud client with app-specific access."""
-
-    @property
-    def notes(self):
-        """Notes client for accessing notes API."""
-        ...
-
-
-async def verify_search_results(
-    results: list[SearchResult],
-    nextcloud_client: NextcloudClientProtocol,
-) -> list[SearchResult]:
-    """
-    Verify Nextcloud access for search results.
-
-    Deduplicates by (doc_id, doc_type), verifies in parallel using anyio task groups,
-    and filters out inaccessible documents. Maintains original result ordering.
-
-    Args:
-        results: Unverified search results from Qdrant
-        nextcloud_client: Nextcloud client for access checks
-
-    Returns:
-        Verified and accessible results (same order as input)
-
-    Example:
-        >>> unverified = await search_algo.search(query="test", limit=10)
-        >>> verified = await verify_search_results(unverified, client)
-        >>> # verified contains only documents user can access
-    """
-    # Deduplicate by (doc_id, doc_type) while preserving order
-    # This is critical for hybrid search where same doc may appear in multiple algorithm results
-    seen = set()
-    unique_results = []
-    for result in results:
-        key = (result.id, result.doc_type)
-        if key not in seen:
-            seen.add(key)
-            unique_results.append(result)
-
-    if not unique_results:
-        return []
-
-    logger.debug(
-        f"Verifying access for {len(unique_results)} unique documents "
-        f"(from {len(results)} total results)"
-    )
-
-    # Verify all unique documents in parallel using anyio task group
-    # Use list to maintain order (index-based storage)
-    verified_results = [None] * len(unique_results)
-
-    async def verify_one(index: int, result: SearchResult):
-        """
-        Verify a single document and store result at index.
-
-        Args:
-            index: Position in verified_results list
-            result: Search result to verify
-        """
-        try:
-            if result.doc_type == "note":
-                # Fetch note to verify access and get fresh metadata
-                note = await nextcloud_client.notes.get_note(result.id)
-                # Update metadata with fresh data from Nextcloud
-                updated_metadata = {**(result.metadata or {}), **note}
-                verified_results[index] = replace(result, metadata=updated_metadata)
-            # TODO: Add verification for other doc types (calendar, deck, file, etc.)
-            else:
-                # For now, assume other types are accessible
-                # In production, add proper verification for each type
-                logger.debug(
-                    f"No verification implemented for doc_type={result.doc_type}, "
-                    "assuming accessible"
-                )
-                verified_results[index] = result
-
-        except Exception as e:
-            # Document is inaccessible (403, 404, or other error)
-            # Log at debug level since this is expected for filtered results
-            logger.debug(f"Document {result.doc_type}/{result.id} not accessible: {e}")
-            verified_results[index] = None
-
-    # Run all verifications in parallel using anyio task group
-    # This provides structured concurrency with automatic cancellation on errors
-    async with anyio.create_task_group() as tg:
-        for idx, result in enumerate(unique_results):
-            tg.start_soon(verify_one, idx, result)
-
-    # Filter out None (inaccessible) and return verified results
-    accessible = [r for r in verified_results if r is not None]
-
-    logger.debug(
-        f"Verification complete: {len(accessible)} accessible, "
-        f"{len(unique_results) - len(accessible)} filtered out"
-    )
-
-    return accessible

nextcloud_mcp_server/server/semantic.py

@@ -3,6 +3,7 @@
 import logging
 from typing import Literal
 
+import anyio
 from httpx import RequestError
 from mcp.server.fastmcp import Context, FastMCP
 from mcp.shared.exceptions import McpError
@@ -153,11 +154,18 @@ def configure_semantic_tools(mcp: FastMCP):
                 # Sort combined results by score
                 all_results.sort(key=lambda r: r.score, reverse=True)
 
-            # Verify access for all results (deduplicates and filters)
-            from nextcloud_mcp_server.search.verification import verify_search_results
+            # Deduplicate results (hybrid search may return same doc from dense + sparse)
+            # Qdrant already filters by user_id for multi-tenant isolation
+            # Sampling tool will verify access when fetching full content
+            seen = set()
+            unique_results = []
+            for result in all_results:
+                key = (result.id, result.doc_type)
+                if key not in seen:
+                    seen.add(key)
+                    unique_results.append(result)
 
-            verified_results = await verify_search_results(all_results, client)
-            search_results = verified_results[:limit]  # Final limit after verification
+            search_results = unique_results[:limit]  # Final limit after deduplication
 
             # Convert SearchResult objects to SemanticSearchResult for response
             results = []
@@ -334,35 +342,55 @@ def configure_semantic_tools(mcp: FastMCP):
                 success=True,
             )
 
-        # 4. Fetch full content for notes to provide complete context to LLM
-        # Filter out inaccessible notes (deleted or permissions changed)
+        # 4. Fetch full content for notes in parallel (also verifies access)
+        # Use anyio task group for concurrent fetching with semaphore to prevent
+        # connection pool exhaustion
         client = await get_client(ctx)
-        accessible_results = []
-        full_contents = []  # Full content for accessible notes
+        accessible_results = [None] * len(search_response.results)
+        full_contents = [None] * len(search_response.results)
 
-        for result in search_response.results:
-            if result.doc_type == "note":
-                try:
-                    note = await client.notes.get_note(result.id)
-                    # Note is accessible, store full content
-                    accessible_results.append(result)
-                    full_contents.append(note.get("content", ""))
-                    logger.debug(
-                        f"Fetched full content for note {result.id} "
-                        f"(length: {len(full_contents[-1])} chars)"
-                    )
-                except Exception as e:
-                    # Note might have been deleted or permissions changed
-                    # Filter it out to avoid corrupting LLM with inaccessible data
-                    logger.warning(
-                        f"Failed to fetch full content for note {result.id}: {e}. "
-                        f"Excluding from results."
-                    )
-            else:
-                # Non-note document types (future: calendar, deck, files)
-                # For now, keep them with excerpts
-                accessible_results.append(result)
-                full_contents.append(None)
+        # Limit concurrent requests to prevent connection pool exhaustion
+        max_concurrent = 20
+        semaphore = anyio.Semaphore(max_concurrent)
+
+        async def fetch_content(index: int, result: SemanticSearchResult):
+            """Fetch full content for a single document (parallel with semaphore)."""
+            async with semaphore:
+                if result.doc_type == "note":
+                    try:
+                        note = await client.notes.get_note(result.id)
+                        # Note is accessible, store result and full content
+                        content = note.get("content", "")
+                        accessible_results[index] = result
+                        full_contents[index] = content
+                        logger.debug(
+                            f"Fetched full content for note {result.id} "
+                            f"(length: {len(content)} chars)"
+                        )
+                    except Exception as e:
+                        # Note might have been deleted or permissions changed
+                        # Leave as None to filter out later
+                        logger.debug(
+                            f"Note {result.id} not accessible: {e}. "
+                            f"Excluding from results."
+                        )
+                else:
+                    # Non-note document types (future: calendar, deck, files)
+                    # For now, keep them with excerpts
+                    accessible_results[index] = result
+                    # full_contents[index] remains None (will use excerpt)
+
+        # Run all fetches in parallel using anyio task group
+        async with anyio.create_task_group() as tg:
+            for idx, result in enumerate(search_response.results):
+                tg.start_soon(fetch_content, idx, result)
+
+        # Filter out None (inaccessible notes) while preserving order
+        final_pairs = [
+            (r, c) for r, c in zip(accessible_results, full_contents) if r is not None
+        ]
+        accessible_results = [r for r, c in final_pairs]
+        full_contents = [c for r, c in final_pairs]
 
         # Check if we filtered out all results
         if not accessible_results:
@@ -414,7 +442,6 @@ def configure_semantic_tools(mcp: FastMCP):
         )
 
         # 6. Request LLM completion via MCP sampling with timeout
-        import anyio
 
         try:
             with anyio.fail_after(30):

nextcloud_mcp_server/vector/processor.py

@@ -8,6 +8,7 @@ import time
 import uuid
 
 import anyio
+from anyio.abc import TaskStatus
 from anyio.streams.memory import MemoryObjectReceiveStream
 from httpx import HTTPStatusError
 from qdrant_client.models import FieldCondition, Filter, MatchValue, PointStruct
@@ -34,6 +35,8 @@ async def processor_task(
     shutdown_event: anyio.Event,
     nc_client: NextcloudClient,
     user_id: str,
+    *,
+    task_status: TaskStatus = anyio.TASK_STATUS_IGNORED,
 ):
     """
     Process documents from stream concurrently.
@@ -53,9 +56,13 @@ async def processor_task(
         shutdown_event: Event signaling shutdown
         nc_client: Authenticated Nextcloud client
         user_id: User being processed
+        task_status: Status object for signaling task readiness
     """
     logger.info(f"Processor {worker_id} started")
 
+    # Signal that the task has started and is ready
+    task_status.started()
+
     while not shutdown_event.is_set():
         try:
             # Get document with timeout (allows checking shutdown)

nextcloud_mcp_server/vector/scanner.py

@@ -8,6 +8,7 @@ import time
 from dataclasses import dataclass
 
 import anyio
+from anyio.abc import TaskStatus
 from anyio.streams.memory import MemoryObjectSendStream
 from qdrant_client.models import FieldCondition, Filter, MatchValue
 
@@ -93,6 +94,8 @@ async def scanner_task(
     wake_event: anyio.Event,
     nc_client: NextcloudClient,
     user_id: str,
+    *,
+    task_status: TaskStatus = anyio.TASK_STATUS_IGNORED,
 ):
     """
     Periodic scanner that detects changed documents for enabled user.
@@ -105,10 +108,14 @@ async def scanner_task(
         wake_event: Event to trigger immediate scan
         nc_client: Authenticated Nextcloud client
         user_id: User to scan
+        task_status: Status object for signaling task readiness
     """
     logger.info(f"Scanner task started for user: {user_id}")
     settings = get_settings()
 
+    # Signal that the task has started and is ready
+    task_status.started()
+
     async with send_stream:
         while not shutdown_event.is_set():
             try:
@@ -175,73 +182,43 @@ async def scan_user_documents(
                 f"[SCAN-{scan_id}] Using pruneBefore={prune_before} to optimize data transfer"
             )
 
-        # Fetch all notes from Nextcloud
-        notes = [
-            note
-            async for note in nc_client.notes.get_all_notes(prune_before=prune_before)
-        ]
-        logger.info(f"[SCAN-{scan_id}] Found {len(notes)} notes for {user_id}")
+        # Get indexed state from Qdrant first (for incremental sync)
+        indexed_docs = {}
+        if not initial_sync:
+            qdrant_client = await get_qdrant_client()
+            scroll_result = await qdrant_client.scroll(
+                collection_name=get_settings().get_collection_name(),
+                scroll_filter=Filter(
+                    must=[
+                        FieldCondition(key="user_id", match=MatchValue(value=user_id)),
+                        FieldCondition(key="doc_type", match=MatchValue(value="note")),
+                    ]
+                ),
+                with_payload=["doc_id", "indexed_at"],
+                with_vectors=False,
+                limit=10000,
+            )
 
-        # Record documents scanned
-        record_vector_sync_scan(len(notes))
+            indexed_docs = {
+                point.payload["doc_id"]: point.payload["indexed_at"]
+                for point in scroll_result[0]
+            }
 
-        if initial_sync:
-            # Send everything on first sync
-            for note in notes:
-                modified_at = note.get("modified", 0)
-                await send_stream.send(
-                    DocumentTask(
-                        user_id=user_id,
-                        doc_id=str(note["id"]),
-                        doc_type="note",
-                        operation="index",
-                        modified_at=modified_at,
-                    )
-                )
-            logger.info(f"Sent {len(notes)} documents for initial sync: {user_id}")
-            return
+            logger.debug(f"Found {len(indexed_docs)} indexed documents in Qdrant")
 
-        # Get indexed state from Qdrant
-        qdrant_client = await get_qdrant_client()
-        scroll_result = await qdrant_client.scroll(
-            collection_name=get_settings().get_collection_name(),
-            scroll_filter=Filter(
-                must=[
-                    FieldCondition(key="user_id", match=MatchValue(value=user_id)),
-                    FieldCondition(key="doc_type", match=MatchValue(value="note")),
-                ]
-            ),
-            with_payload=["doc_id", "indexed_at"],
-            with_vectors=False,
-            limit=10000,
-        )
-
-        indexed_docs = {
-            point.payload["doc_id"]: point.payload["indexed_at"]
-            for point in scroll_result[0]
-        }
-
-        logger.debug(f"Found {len(indexed_docs)} indexed documents in Qdrant")
-
-        # Compare and queue changes
+        # Stream notes from Nextcloud and process immediately
+        note_count = 0
         queued = 0
-        nextcloud_doc_ids = {str(note["id"]) for note in notes}
+        nextcloud_doc_ids = set()
 
-        for note in notes:
+        async for note in nc_client.notes.get_all_notes(prune_before=prune_before):
+            note_count += 1
             doc_id = str(note["id"])
-            indexed_at = indexed_docs.get(doc_id)
+            nextcloud_doc_ids.add(doc_id)
             modified_at = note.get("modified", 0)
 
-            # If document reappeared, remove from potentially_deleted
-            doc_key = (user_id, doc_id)
-            if doc_key in _potentially_deleted:
-                logger.debug(
-                    f"Document {doc_id} reappeared, removing from deletion grace period"
-                )
-                del _potentially_deleted[doc_key]
-
-            # Send if never indexed or modified since last index
-            if indexed_at is None or modified_at > indexed_at:
+            if initial_sync:
+                # Send everything on first sync
                 await send_stream.send(
                     DocumentTask(
                         user_id=user_id,
@@ -252,6 +229,38 @@ async def scan_user_documents(
                     )
                 )
                 queued += 1
+            else:
+                # Incremental sync: compare with indexed state
+                indexed_at = indexed_docs.get(doc_id)
+
+                # If document reappeared, remove from potentially_deleted
+                doc_key = (user_id, doc_id)
+                if doc_key in _potentially_deleted:
+                    logger.debug(
+                        f"Document {doc_id} reappeared, removing from deletion grace period"
+                    )
+                    del _potentially_deleted[doc_key]
+
+                # Send if never indexed or modified since last index
+                if indexed_at is None or modified_at > indexed_at:
+                    await send_stream.send(
+                        DocumentTask(
+                            user_id=user_id,
+                            doc_id=doc_id,
+                            doc_type="note",
+                            operation="index",
+                            modified_at=modified_at,
+                        )
+                    )
+                    queued += 1
+
+        # Log and record metrics after streaming
+        logger.info(f"[SCAN-{scan_id}] Found {note_count} notes for {user_id}")
+        record_vector_sync_scan(note_count)
+
+        if initial_sync:
+            logger.info(f"Sent {queued} documents for initial sync: {user_id}")
+            return
 
         # Check for deleted documents (in Qdrant but not in Nextcloud)
         # Use grace period: only delete after 2 consecutive scans confirm absence