fix: address PR #813 latest review (ACL-aware doc-type discovery, robustness)

- get_indexed_doc_types: add optional accessible_owners param and reuse
  build_ownership_filter so cross-user doc-type discovery matches the real
  search scope (was self-only / ACL-blind); docstring documents the self-only
  default. Covered by test_get_indexed_doc_types_is_acl_aware.
- access_filter: build_ownership_filter now omits the owner_id branch entirely
  for an empty owner set instead of relying on undocumented MatchAny(any=[])
  semantics; updated the empty-list unit test accordingly.
- access_filter: make the uid_owner/owner share-owner extraction explicit
  ("absent, not empty") to avoid skipping on a falsy-but-present field.
- access_filter: add an operator note that pre-owner_id points need a re-index
  to surface to share recipients (ACL search is a no-op for legacy data).
- verification/webdav: lock the file_accessible_by_id(scope="") contract with a
  targeted multi-user test (owner + recipient True, non-recipient False).
- viz_routes: comment that verify-on-read eviction runs inline by design (no
  lifespan task group available on the Starlette route).

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

nextcloud_mcp_server/auth/viz_routes.py

@@ -234,6 +234,12 @@ async def vector_visualization_search(request: Request) -> JSONResponse:
             # exactly as the nc_semantic_search tool path does. Skipping this
             # would let the viz surface stale titles/excerpts from another
             # user's index after a share is revoked.
+            # Eviction of dropped (e.g. revoked-share) points runs INLINE here
+            # by design: this is a Starlette route with no access to the
+            # FastMCP lifespan-owned ``eviction_task_group`` that the
+            # nc_semantic_search tool path passes for fire-and-forget eviction.
+            # The visualization is an interactive, low-QPS endpoint, so blocking
+            # briefly on the Qdrant delete is acceptable.
             with trace_operation("vector_viz.verify_on_read"):
                 verified_results, _dropped = await verify_search_results(
                     nc_client, all_results

nextcloud_mcp_server/search/access_filter.py

@@ -12,6 +12,14 @@ This module turns "who can user X read?" into a Qdrant filter:
 A second OR-branch matches the legacy ``user_id`` field so points indexed
 before this change (which carry only ``user_id``) continue to be findable
 by their original indexer. New points carry both fields.
+
+Operator note (existing data): a Qdrant ``owner_id`` field condition matches
+nothing on points that lack the field, so documents indexed *before* this
+change never surface to share recipients — only to their original indexer via
+the legacy ``user_id`` branch. ACL-aware search is therefore effectively a
+no-op for pre-existing data until each owner's scanner re-indexes it. Trigger a
+re-index after deploying this feature if it should apply to already-indexed
+content immediately.
 """
 
 from __future__ import annotations
@@ -21,7 +29,7 @@ import time
 from collections import OrderedDict
 from typing import Any, Protocol
 
-from qdrant_client.models import FieldCondition, Filter, MatchAny, MatchValue
+from qdrant_client.models import Condition, FieldCondition, Filter, MatchAny, MatchValue
 
 logger = logging.getLogger(__name__)
 
@@ -106,10 +114,13 @@ async def list_accessible_owners(
     for share in shares:
         # OCS returns the share owner under `uid_owner` (the file owner,
         # not the share recipient). Some Nextcloud versions also surface
-        # `owner` as a fallback display field — we tolerate both.
-        owner = share.get("uid_owner") or share.get("owner")
-        if isinstance(owner, str) and owner:
-            owners.add(owner)
+        # `owner` as a fallback display field — we tolerate both. The intent is
+        # "absent, not empty": a missing/blank `uid_owner` falls through to
+        # `owner`, and a non-string or empty result skips the (malformed) share.
+        owner = share.get("uid_owner") or share.get("owner") or None
+        if not isinstance(owner, str) or not owner:
+            continue
+        owners.add(owner)
 
     result = list(owners)
     _owners_cache[user_id] = (now, result)
@@ -141,14 +152,15 @@ def build_ownership_filter(
         A Qdrant ``Filter`` ready to be nested under a parent ``must`` clause.
     """
     owners = accessible_owners if accessible_owners is not None else [user_id]
-    # Edge case: an explicit empty ``accessible_owners`` yields
-    # ``MatchAny(any=[])``, which Qdrant treats as matching nothing — so the
-    # owner_id branch contributes no results and access comes solely from the
-    # legacy ``user_id`` branch below (self-owned content). Callers that want
-    # share expansion must pass a non-empty list.
-    return Filter(
-        should=[
-            FieldCondition(key="owner_id", match=MatchAny(any=owners)),
-            FieldCondition(key="user_id", match=MatchValue(value=user_id)),
-        ]
-    )
+    # The legacy ``user_id`` branch is always present (self-owned content,
+    # incl. pre-migration points). The ``owner_id`` branch is appended only for
+    # a non-empty owner set: an empty list is handled explicitly here rather
+    # than relying on ``MatchAny(any=[])`` matching nothing, which is not a
+    # documented Qdrant guarantee and could change across versions. Self always
+    # matches via the ``user_id`` branch, so an empty owner set is safe.
+    conditions: list[Condition] = [
+        FieldCondition(key="user_id", match=MatchValue(value=user_id)),
+    ]
+    if owners:
+        conditions.insert(0, FieldCondition(key="owner_id", match=MatchAny(any=owners)))
+    return Filter(should=conditions)

nextcloud_mcp_server/search/algorithms.py

@@ -5,9 +5,10 @@ from abc import ABC, abstractmethod
 from dataclasses import dataclass
 from typing import Any, Protocol, runtime_checkable
 
-from qdrant_client.models import FieldCondition, Filter, MatchValue, ScoredPoint
+from qdrant_client.models import Filter, ScoredPoint
 
 from nextcloud_mcp_server.config import get_settings
+from nextcloud_mcp_server.search.access_filter import build_ownership_filter
 from nextcloud_mcp_server.vector.placeholder import get_placeholder_filter
 from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client
 
@@ -75,14 +76,24 @@ class NextcloudClientProtocol(Protocol):
         ...
 
 
-async def get_indexed_doc_types(user_id: str) -> set[str]:
+async def get_indexed_doc_types(
+    user_id: str, accessible_owners: list[str] | None = None
+) -> set[str]:
     """Query Qdrant to get actually-indexed document types for a user.
 
     This enables search algorithms to check which document types are available
     before attempting to search/verify them, allowing graceful cross-app search.
 
     Args:
-        user_id: User ID to filter by
+        user_id: User ID to filter by.
+        accessible_owners: Owner UIDs the user may read (self + share senders),
+            as computed by ``access_filter.list_accessible_owners``. When
+            provided, doc-type discovery is ACL-aware and matches the same
+            ownership scope as the actual search (so a share recipient discovers
+            cross-user doc_types). When ``None`` (the default), discovery is
+            **self-only** — a recipient won't see doc_types that exist only in
+            another owner's shared content. Pass the expanded set for cross-user
+            discovery.
 
     Returns:
         Set of document type strings (e.g., {"note", "file", "calendar"})
@@ -106,7 +117,9 @@ async def get_indexed_doc_types(user_id: str) -> set[str]:
             scroll_filter=Filter(
                 must=[
                     get_placeholder_filter(),  # Exclude placeholders from doc_type discovery
-                    FieldCondition(key="user_id", match=MatchValue(value=user_id)),
+                    # ACL-aware ownership scope (owner_id IN owners OR legacy
+                    # user_id == user_id), matching the real search filter.
+                    build_ownership_filter(user_id, accessible_owners),
                 ]
             ),
             limit=1000,  # Sample size to discover types

tests/integration/test_acl_owner_filter.py

@@ -23,6 +23,7 @@ from qdrant_client.models import Distance, PointStruct, VectorParams
 
 from nextcloud_mcp_server.config import get_settings
 from nextcloud_mcp_server.embedding import SimpleEmbeddingProvider
+from nextcloud_mcp_server.search.algorithms import get_indexed_doc_types
 from nextcloud_mcp_server.search.semantic import SemanticSearchAlgorithm
 
 pytestmark = pytest.mark.integration
@@ -92,6 +93,11 @@ async def seeded_collection(monkeypatch):
         "nextcloud_mcp_server.search.semantic.get_embedding_service",
         lambda: provider,
     )
+    # get_indexed_doc_types reads the client from the algorithms module.
+    monkeypatch.setattr(
+        "nextcloud_mcp_server.search.algorithms.get_qdrant_client",
+        AsyncMock(return_value=client),
+    )
 
     yield provider
 
@@ -170,3 +176,16 @@ async def test_owner_sees_own_new_style_point(seeded_collection):
     assert "101" in found
     assert "102" not in found
     assert "103" not in found
+
+
+async def test_get_indexed_doc_types_is_acl_aware(seeded_collection):
+    """get_indexed_doc_types respects the ownership scope: with the expanded
+    accessible_owners Bob discovers the shared "file" type, but self-only Bob
+    (who owns nothing here) discovers nothing — proving it is no longer
+    ACL-blind."""
+    # ACL-aware: Bob can read Alice's shared file → discovers "file".
+    assert await get_indexed_doc_types("bob", accessible_owners=["bob", "alice"]) == {
+        "file"
+    }
+    # Self-only (default): Bob owns nothing here → discovers nothing.
+    assert await get_indexed_doc_types("bob") == set()

tests/integration/test_acl_shared_search.py

@@ -195,3 +195,23 @@ async def test_non_recipient_does_not_find_file(acl_users, seeded_semantic):
     kept = await _search_as(acl_users["diana"], seeded_semantic)
 
     assert kept == [], "diana (no share) must not find alice's file"
+
+
+async def test_file_accessible_by_id_resolves_shares(acl_users, shared_file):
+    """Lock the verify-on-read contract directly on ``file_accessible_by_id``.
+
+    The WebDAV SEARCH-by-fileid with ``scope=""`` must resolve a file that the
+    caller does NOT own but which is shared with them. This is the exact check
+    verify-on-read depends on for shared, nested files; a Nextcloud change to
+    how ``scope=""`` is interpreted would otherwise silently break ACL-aware
+    verification. The file lives in a subfolder, so a path-based check would
+    404 for the recipient — only the by-id SEARCH gets it right.
+    """
+    file_id, _path = shared_file
+    fid = int(file_id)
+
+    # Owner and share recipient can both reach it...
+    assert await acl_users["alice"].webdav.file_accessible_by_id(fid) is True
+    assert await acl_users["bob"].webdav.file_accessible_by_id(fid) is True
+    # ...the non-recipient cannot.
+    assert await acl_users["diana"].webdav.file_accessible_by_id(fid) is False

tests/unit/search/test_access_filter.py

@@ -174,13 +174,15 @@ class TestBuildOwnershipFilter:
         # legacy match path, so it must NOT widen to other owners.
         assert user_branch.match.value == "alice"
 
-    def test_explicit_empty_list_still_keeps_legacy_branch(self) -> None:
-        # Edge case: caller passed an explicit empty list. We shouldn't
-        # silently re-default to [user_id] in the owner branch, but the
-        # legacy branch is still the safety net so the user can find their
-        # own content from before the migration.
+    def test_explicit_empty_list_omits_owner_branch_keeps_legacy(self) -> None:
+        # Edge case: caller passed an explicit empty list. The owner_id branch
+        # is omitted entirely (rather than relying on MatchAny(any=[]) matching
+        # nothing); the legacy user_id branch remains as the safety net so the
+        # user still finds their own content from before the migration.
         flt = build_ownership_filter("alice", [])
 
-        owner_branch, user_branch = flt.should
-        assert owner_branch.match.any == []
+        assert flt.should is not None
+        assert len(flt.should) == 1
+        (user_branch,) = flt.should
+        assert user_branch.key == "user_id"
         assert user_branch.match.value == "alice"