fix: PR #813 review — shared-file context in MCP tool path + viz over-fetch cap

🟡 Important: nc_semantic_search's include_context branch did not forward
accessible_owners to get_chunk_with_context, so context expansion for shared
files stayed self-only, found nothing in Qdrant, and silently fell back to the
plain excerpt. Forward accessible_owners (the per-file file_accessible_by_id
gate still enforces access).

🟡 Performance: auth/viz_routes.py's multi-doc_type branch sorted but did not
cap the candidate pool before verify-on-read, so N doc_types × limit*2 went
into verification (N× the Nextcloud round-trips). Cap to limit*2 after the
sort, matching server/semantic.py and the cross-app branch.

Also clear the SonarCloud gate (new_duplicated_lines_density 5.1% > 3%) the
ACL wiring introduced: extract the duplicated /api/v1 client-resolution +
owner-expansion + verify-on-read block from unified_search/vector_search into a
shared _search_with_acl helper, define a constant for the repeated
"Nextcloud host not configured" literal (S1192), and reword the access_filter
move_to_end comment so it isn't misread as commented-out code (S125) while
adding the other-owner count to its debug log (review nits).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

nextcloud_mcp_server/api/visualization.py

@@ -11,6 +11,7 @@ All endpoints require OAuth bearer token authentication via UnifiedTokenVerifier
 
 import base64
 import logging
+from collections.abc import Awaitable, Callable
 from typing import Any
 
 import pymupdf
@@ -45,6 +46,70 @@ from nextcloud_mcp_server.vector.visualization import compute_pca_coordinates
 
 logger = logging.getLogger(__name__)
 
+_NEXTCLOUD_HOST_NOT_CONFIGURED = "Nextcloud host not configured"
+
+
+async def _search_with_acl(
+    request: Request,
+    user_id: str,
+    execute: Callable[[list[str] | None], Awaitable[list]],
+) -> list:
+    """Resolve the caller's Nextcloud client, run ``execute(accessible_owners)``,
+    and verify-on-read — shared by the /api/v1 search endpoints.
+
+    The OAuth bearer only authenticates Astrolabe → MCP Server; MCP Server →
+    Nextcloud uses the provisioned app password. When the caller never
+    provisioned background sync there is no client to expand shares or verify
+    with, so we fall back to self-only, unverified search (the pre-ACL
+    behaviour) rather than 401 — keeping search working for users who haven't
+    opted into background indexing.
+
+    Args:
+        request: The Starlette request (carries ``app.state.oauth_context``).
+        user_id: The authenticated caller.
+        execute: Coroutine that runs the search for a given owner scope
+            (``None`` ⇒ self-only).
+
+    Returns:
+        The result list (verified for provisioned callers).
+
+    Raises:
+        ValueError: If the Nextcloud host is not configured.
+    """
+    oauth_ctx = request.app.state.oauth_context
+    nextcloud_host = oauth_ctx.get("config", {}).get("nextcloud_host", "")
+    if not nextcloud_host:
+        raise ValueError(_NEXTCLOUD_HOST_NOT_CONFIGURED)
+
+    try:
+        nc_client = await get_user_client_basic_auth(user_id, nextcloud_host)
+    except NotProvisionedError:
+        logger.debug("User %s not provisioned; self-only unverified search", user_id)
+        results = await execute(None)
+    else:
+        async with nc_client:
+            # Expand to owners who shared content with the caller (same as the
+            # MCP tool path) so shared documents are searchable.
+            accessible_owners = await list_accessible_owners(nc_client.sharing, user_id)
+            results = await execute(accessible_owners)
+            # Verify-on-read (ADR-019): drop documents the caller can no longer
+            # access (e.g. a revoked share). Eviction runs inline — this
+            # Starlette route has no FastMCP lifespan task group.
+            results, _dropped = await verify_search_results(nc_client, results)
+
+    # Safe to log titles now: provisioned callers passed verify-on-read;
+    # non-provisioned ran self-only (unverified titles are never logged — see
+    # the search algorithms).
+    if results:
+        logger.debug(
+            "Top verified results: %s",
+            ", ".join(
+                f"{r.doc_type}_{r.id} (score={r.score:.3f}, title='{r.title}')"
+                for r in results[:5]
+            ),
+        )
+    return results
+
 
 async def unified_search(request: Request) -> JSONResponse:
     """POST /api/v1/search - Search endpoint for Nextcloud Unified Search.
@@ -192,50 +257,7 @@ async def unified_search(request: Request) -> JSONResponse:
                 )
             return results
 
-        # Resolve a Nextcloud client so search is ACL-aware and verify-on-read
+        all_results = await _search_with_acl(request, user_id, _execute)
-        # can confirm access. The OAuth bearer only authenticates Astrolabe →
-        # MCP Server; MCP Server → Nextcloud uses the provisioned app password.
-        # If the caller never provisioned background sync there is no client to
-        # expand shares or verify with — fall back to self-only, unverified
-        # search (the pre-ACL behaviour) rather than 401, so unified search keeps
-        # working for users who haven't opted into background indexing.
-        oauth_ctx = request.app.state.oauth_context
-        nextcloud_host = oauth_ctx.get("config", {}).get("nextcloud_host", "")
-        if not nextcloud_host:
-            raise ValueError("Nextcloud host not configured")
-        try:
-            nc_client = await get_user_client_basic_auth(user_id, nextcloud_host)
-        except NotProvisionedError:
-            logger.debug(
-                "User %s not provisioned; self-only unverified search", user_id
-            )
-            all_results = await _execute(None)
-        else:
-            async with nc_client:
-                # Expand to owners who shared content with the caller (same as
-                # the MCP tool path) so shared documents are searchable.
-                accessible_owners = await list_accessible_owners(
-                    nc_client.sharing, user_id
-                )
-                all_results = await _execute(accessible_owners)
-                # Verify-on-read (ADR-019): drop documents the caller can no
-                # longer access (e.g. a revoked share) before formatting.
-                # Eviction runs inline — this Starlette route has no FastMCP
-                # lifespan task group.
-                all_results, _dropped = await verify_search_results(
-                    nc_client, all_results
-                )
-
-        # Safe to log titles now: provisioned callers passed verify-on-read;
-        # non-provisioned ran self-only (unverified titles are never logged).
-        if all_results:
-            logger.debug(
-                "Top verified results: %s",
-                ", ".join(
-                    f"{r.doc_type}_{r.id} (score={r.score:.3f}, title='{r.title}')"
-                    for r in all_results[:5]
-                ),
-            )
 
         # Sort results by score (no deduplication - show all chunks)
         sorted_results = sorted(all_results, key=lambda r: r.score, reverse=True)
@@ -440,47 +462,7 @@ async def vector_search(request: Request) -> JSONResponse:
                 )
             return results
 
-        # Resolve a Nextcloud client so search is ACL-aware and verify-on-read
+        all_results = await _search_with_acl(request, user_id, _execute)
-        # can confirm access (same pattern as /api/v1/search). If the caller
-        # never provisioned background sync there is no client to expand shares
-        # or verify with — fall back to self-only, unverified search (pre-ACL
-        # behaviour) rather than 401.
-        oauth_ctx = request.app.state.oauth_context
-        nextcloud_host = oauth_ctx.get("config", {}).get("nextcloud_host", "")
-        if not nextcloud_host:
-            raise ValueError("Nextcloud host not configured")
-        try:
-            nc_client = await get_user_client_basic_auth(user_id, nextcloud_host)
-        except NotProvisionedError:
-            logger.debug(
-                "User %s not provisioned; self-only unverified search", user_id
-            )
-            all_results = await _execute(None)
-        else:
-            async with nc_client:
-                # Expand to owners who shared content with the caller (same as
-                # the MCP tool path) so shared documents are searchable.
-                accessible_owners = await list_accessible_owners(
-                    nc_client.sharing, user_id
-                )
-                all_results = await _execute(accessible_owners)
-                # Verify-on-read (ADR-019): drop now-inaccessible docs before
-                # formatting (inline eviction — no FastMCP lifespan task group
-                # on this Starlette route).
-                all_results, _dropped = await verify_search_results(
-                    nc_client, all_results
-                )
-
-        # Safe to log titles now: provisioned callers passed verify-on-read;
-        # non-provisioned ran self-only (unverified titles are never logged).
-        if all_results:
-            logger.debug(
-                "Top verified results: %s",
-                ", ".join(
-                    f"{r.doc_type}_{r.id} (score={r.score:.3f}, title='{r.title}')"
-                    for r in all_results[:5]
-                ),
-            )
 
         # Format results for PHP client
         formatted_results = []
@@ -655,7 +637,7 @@ async def get_chunk_context(request: Request) -> JSONResponse:
         nextcloud_host = oauth_ctx.get("config", {}).get("nextcloud_host", "")
 
         if not nextcloud_host:
-            raise ValueError("Nextcloud host not configured")
+            raise ValueError(_NEXTCLOUD_HOST_NOT_CONFIGURED)
 
         # Use the user's stored app password for Nextcloud calls.
         # The OAuth bearer is only used to authenticate Astrolabe → MCP Server;
@@ -821,7 +803,7 @@ async def get_pdf_preview(request: Request) -> JSONResponse:
         nextcloud_host = oauth_ctx.get("config", {}).get("nextcloud_host", "")
 
         if not nextcloud_host:
-            raise ValueError("Nextcloud host not configured")
+            raise ValueError(_NEXTCLOUD_HOST_NOT_CONFIGURED)
 
         # Use the user's stored app password for Nextcloud calls.
         # The OAuth bearer is only used to authenticate Astrolabe → MCP Server;

nextcloud_mcp_server/auth/viz_routes.py

@@ -224,8 +224,13 @@ async def vector_visualization_search(request: Request) -> JSONResponse:
                             accessible_owners=accessible_owners,
                         )
                     all_results.extend(unverified_results)
-                # Sort by score before verification
+                # Sort by score, then cap to the same limit*2 over-fetch budget
+                # as the cross-app branch and the nc_semantic_search tool path
+                # (server/semantic.py). Without this, N doc_types each fetched
+                # at limit*2 would send N*limit*2 candidates into verify-on-read,
+                # multiplying the Nextcloud round-trip cost (and latency) by N.
                 all_results.sort(key=lambda r: r.score, reverse=True)
+                all_results = all_results[: limit * 2]
 
             # Verify-on-read (ADR-019). Now that accessible_owners is expanded
             # via OCS shares, the result set can include OTHER users' shared

nextcloud_mcp_server/search/access_filter.py

@@ -124,10 +124,18 @@ async def list_accessible_owners(
 
     result = list(owners)
     _owners_cache[user_id] = (now, result)
-    _owners_cache.move_to_end(user_id)  # newest = most-recently-used
+    # Promote to the most-recently-used end. This is a no-op for a brand-new
+    # key (dict insertion already appends) but is needed when re-inserting an
+    # existing key after its TTL expired.
+    _owners_cache.move_to_end(user_id)
     while len(_owners_cache) > _OWNERS_CACHE_MAXSIZE:
-        _owners_cache.popitem(last=False)  # evict least-recently-used
+        _owners_cache.popitem(last=False)  # evict the least-recently-used entry
-    logger.debug("Accessible owners for user %s: %d entries", user_id, len(result))
+    logger.debug(
+        "Accessible owners for user %s: %d entries (%d other owner(s))",
+        user_id,
+        len(result),
+        len(result) - 1,
+    )
     return list(result)
 
 

nextcloud_mcp_server/server/semantic.py

@@ -339,6 +339,12 @@ def configure_semantic_tools(mcp: FastMCP):
                                 chunk_index=result.chunk_index,
                                 total_chunks=result.total_chunks,
                                 context_chars=context_chars,
+                                # Forward the share-expanded owner set so context
+                                # expansion works for shared files (the per-file
+                                # file_accessible_by_id gate inside still enforces
+                                # access). Without this the lookup stays self-only
+                                # and silently falls back to the plain excerpt.
+                                accessible_owners=accessible_owners,
                             )
 
                             if chunk_context: