feat: dedup shared-file parsing/embedding across users in vector sync

A file shared across many users — directly, or via a group folder shared
to a group — was parsed and embedded once per user. Chunk point IDs are
user-agnostic (uuid5(tenant_id, doc_id=fileid, chunk_index)), but the
per-user freshness gate filtered Qdrant by user_id, so two readers
ping-ponged: each overwrote the other's points and each kept seeing "not
indexed for me", reprocessing every scan. Production telemetry (note
386945, finding #5) measured identical docs re-processed every few hours
at 7-13s each, with PDF parse ~62% of per-doc cost.

Layer 1 — tenant-wide dedup:
- Thread the scanner's tag-REPORT etag into the file DocumentTask and the
  chunk payload; index `etag` as a KEYWORD field.
- vector/sharing_state.find_indexed_content scrolls tenant-wide (no
  user_id filter) for a non-placeholder point matching
  (doc_id, doc_type, etag), gated on embedding_identity in Python so a
  model switch correctly forces a re-embed.
- Scanner skips enqueue and the processor skips fetch/parse/embed when a
  match exists (cross-worker race-guard before WebDAV read). Dedup is
  fail-safe: a Qdrant error degrades to "process normally".

Layer 2 — observed-access ACL (no admin / GroupFolders API needed):
- Each point carries `acl_principals` = the set of user:<uid> whose
  scanner has observed (hence can read) the file. The per-user tag REPORT
  is the access oracle; group membership/GroupFolders enumeration is
  admin-only and unavailable in multi-user modes.
- build_ownership_filter ORs MatchAny(acl_principals, ["user:<me>"]) so a
  deduplicated shared/group-folder point surfaces to every reader;
  verify-on-read (_verify_files) remains the precise ACL gate.
- Deletion/eviction become "release one user": drop the principal and
  delete the points only when the set empties, so one user untagging a
  shared file doesn't evict it for the others. Legacy points without the
  field keep the original per-user delete.

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

nextcloud_mcp_server/search/access_filter.py

@@ -162,10 +162,14 @@ def build_ownership_filter(
     """Build the Qdrant ``Filter`` constraining a search to readable points.
 
     Matches points whose ``owner_id`` is in ``accessible_owners`` (excluding
-    self) OR whose ``user_id`` equals ``user_id``. The ``user_id`` branch covers
-    *all* of the caller's own content — both new points (where
-    ``owner_id == user_id``) and legacy points indexed before ``owner_id``
-    existed — so self is intentionally NOT repeated in the ``owner_id`` branch.
+    self) OR whose ``user_id`` equals ``user_id`` OR whose ``acl_principals``
+    contains ``user:<user_id>``. The ``user_id`` branch covers *all* of the
+    caller's own content — both new points (where ``owner_id == user_id``) and
+    legacy points indexed before ``owner_id`` existed — so self is intentionally
+    NOT repeated in the ``owner_id`` branch. The ``acl_principals`` branch covers
+    files that were indexed once and deduplicated across users (user-agnostic
+    point IDs): such a point's ``user_id``/``owner_id`` are the first indexer's,
+    so only the observed-access principal set surfaces it to other readers.
 
     Args:
         user_id: Querying user (matched by the ``user_id`` branch, which is the
@@ -188,6 +192,13 @@ def build_ownership_filter(
     other_owners = [owner for owner in owners if owner != user_id]
     conditions: list[Condition] = [
         FieldCondition(key="user_id", match=MatchValue(value=user_id)),
+        # Observed-access branch: a file deduplicated across users carries one
+        # user-agnostic point set whose ``user_id``/``owner_id`` are the first
+        # indexer's. ``acl_principals`` lists every user whose scanner has seen
+        # (hence can read) the file, so this branch surfaces a shared/group-folder
+        # file to every reader even when they were not the indexer. Verify-on-read
+        # (_verify_files) is the precise ACL gate on the returned candidates.
+        FieldCondition(key="acl_principals", match=MatchAny(any=[f"user:{user_id}"])),
     ]
     if other_owners:
         conditions.insert(

nextcloud_mcp_server/vector/eviction.py

@@ -9,10 +9,7 @@ simply re-verify and re-attempt.
 
 import logging
 
-from qdrant_client.models import FieldCondition, Filter, MatchValue
-
-from nextcloud_mcp_server.config import get_settings
-from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client
+from nextcloud_mcp_server.vector.sharing_state import release_document_for_user
 
 logger = logging.getLogger(__name__)
 
@@ -22,39 +19,29 @@ async def delete_document_points(
     doc_type: str,
     user_id: str,
 ) -> None:
-    """Remove all Qdrant points for a single document.
+    """Revoke one user's access to a document's points (verify-on-read eviction).
 
-    Deletes both real chunk points and any leftover placeholder points for the
-    given (user_id, doc_id, doc_type) tuple. Safe to call when the document is
+    A document can be indexed once and shared across users (user-agnostic point
+    IDs), so eviction must *release* this user rather than blindly delete: it
+    drops ``user:<user_id>`` from the point's ``acl_principals`` and removes the
+    points only when no reader remains. Legacy points without a principal set
+    fall back to the original per-user delete. Safe to call when the document is
     not present — Qdrant returns successfully with zero points affected.
 
     Args:
         doc_id: Document ID (str — keyword-indexed in Qdrant payload)
         doc_type: Document type (note, file, deck_card, news_item)
-        user_id: Owner of the points being evicted
+        user_id: User whose access is being revoked
 
     Raises:
         Exception: If the underlying Qdrant client raises. Callers in the
             search hot path should catch and log; eviction failures must not
             block search responses.
     """
-    qdrant_client = await get_qdrant_client()
-    settings = get_settings()
-
-    await qdrant_client.delete(
-        collection_name=settings.get_collection_name(),
-        points_selector=Filter(
-            must=[
-                FieldCondition(key="user_id", match=MatchValue(value=user_id)),
-                FieldCondition(key="doc_id", match=MatchValue(value=doc_id)),
-                FieldCondition(key="doc_type", match=MatchValue(value=doc_type)),
-            ]
-        ),
-    )
+    await release_document_for_user(doc_id, doc_type, user_id)
 
     logger.info(
-        "Evicted Qdrant points for %s_%s (user=%s); "
-        "document was inaccessible at verification time",
+        "Released %s_%s for user=%s; document was inaccessible at verification time",
         doc_type,
         doc_id,
         user_id,

nextcloud_mcp_server/vector/processor.py

@@ -11,7 +11,7 @@ from typing import Any, cast
 import anyio
 from anyio.abc import TaskStatus
 from anyio.streams.memory import MemoryObjectReceiveStream
-from qdrant_client.models import FieldCondition, Filter, MatchValue, PointStruct
+from qdrant_client.models import PointStruct
 
 from nextcloud_mcp_server.acl_hash import compute_acl_hash
 from nextcloud_mcp_server.client import NextcloudClient
@@ -33,6 +33,11 @@ from nextcloud_mcp_server.vector.html_processor import html_to_markdown
 from nextcloud_mcp_server.vector.placeholder import delete_placeholder_point
 from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client
 from nextcloud_mcp_server.vector.scanner import DocumentTask
+from nextcloud_mcp_server.vector.sharing_state import (
+    claim_existing_index,
+    existing_principals,
+    release_document_for_user,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -192,28 +197,15 @@ async def process_document(
     ):
         try:
             qdrant_client = await get_qdrant_client()
-            settings = get_settings()
 
             # Handle deletion
             if doc_task.operation == "delete":
-                await qdrant_client.delete(
-                    collection_name=settings.get_collection_name(),
-                    points_selector=Filter(
-                        must=[
-                            FieldCondition(
-                                key="user_id",
-                                match=MatchValue(value=doc_task.user_id),
-                            ),
-                            FieldCondition(
-                                key="doc_id",
-                                match=MatchValue(value=doc_task.doc_id),
-                            ),
-                            FieldCondition(
-                                key="doc_type",
-                                match=MatchValue(value=doc_task.doc_type),
-                            ),
-                        ]
-                    ),
+                # Release this user rather than blind-delete: a file shared across
+                # users has one user-agnostic point set referenced by multiple
+                # principals, so the points are removed only once the last reader
+                # is gone (see vector/sharing_state.release_document_for_user).
+                await release_document_for_user(
+                    doc_task.doc_id, doc_task.doc_type, doc_task.user_id
                 )
                 logger.info(
                     "Deleted %s_%s for %s",
@@ -479,6 +471,28 @@ async def _index_document(
                 )
             file_path = doc_task.file_path
 
+            # Cross-worker dedup race-guard: two users' tasks for the same shared
+            # file can be enqueued before either finishes. If another worker has
+            # already indexed this exact content (fileid + etag + embedding model)
+            # in the tenant, claim it for this user (observed-access ACL) and skip
+            # the expensive fetch/parse/embed entirely.
+            if doc_task.etag and await claim_existing_index(
+                doc_task.doc_id, "file", doc_task.etag, doc_task.user_id
+            ):
+                await delete_placeholder_point(
+                    doc_id=doc_task.doc_id,
+                    doc_type="file",
+                    user_id=doc_task.user_id,
+                )
+                logger.info(
+                    "Dedup hit for file %s (etag=%s); claimed for user %s "
+                    "without reprocessing",
+                    doc_task.doc_id,
+                    doc_task.etag,
+                    doc_task.user_id,
+                )
+                return
+
             # Read file content via WebDAV
             content_bytes, content_type = await nc_client.webdav.read_file(file_path)
         else:
@@ -510,7 +524,10 @@ async def _index_document(
                 content = result.text
                 file_metadata = result.metadata
                 title = file_metadata.get("title") or file_path.split("/")[-1]
-                etag = ""  # WebDAV read_file doesn't return etag
+                # etag comes from the scanner's tag REPORT (threaded via the
+                # DocumentTask); read_file itself returns no etag. It is the
+                # tenant-wide content-dedup key, so it must be persisted.
+                etag = doc_task.etag or ""
 
                 # Diagnostic: Log page boundary information if available
                 if "page_boundaries" in file_metadata:
@@ -763,6 +780,19 @@ async def _index_document(
     _embedding_identity = settings.get_embedding_model_name()
     _acl_hash = compute_acl_hash([("user", doc_task.user_id)])
 
+    # Observed-access ACL principals (computed once per document, not per chunk).
+    # Seed with the indexer (and owner, if distinct) unioned with any principals
+    # already recorded — so re-indexing after a content change preserves
+    # visibility for readers who had previously claimed the file rather than
+    # resetting it to just the indexer.
+    _acl_principals = sorted(
+        set(await existing_principals(doc_task.doc_id, doc_task.doc_type))
+        | {
+            f"user:{doc_task.user_id}",
+            f"user:{doc_task.owner_id or doc_task.user_id}",
+        }
+    )
+
     # Surface deck card data quality issues at indexing time rather than
     # only at verification time (where _verify_deck_cards falls through to
     # legacy-data pass-through when board_id/stack_id are missing). This is
@@ -807,6 +837,13 @@ async def _index_document(
                     # crawl shared-with-them content can set owner_id to the
                     # true owner without losing the "who indexed this" trail.
                     "owner_id": doc_task.owner_id or doc_task.user_id,
+                    # Observed-access ACL set: every user whose scanner has seen
+                    # (hence can read) this document. Seeded with the indexer (and
+                    # owner, if distinct); grown lazily as other readers' scanners
+                    # hit the tenant-wide dedup path. Search ORs a
+                    # MatchAny(acl_principals, ["user:<me>"]) branch so a
+                    # deduplicated shared file stays findable by every reader.
+                    "acl_principals": _acl_principals,
                     "doc_id": doc_task.doc_id,
                     "doc_type": doc_task.doc_type,
                     "is_placeholder": False,  # Real indexed document (not placeholder)

nextcloud_mcp_server/vector/qdrant_client.py

@@ -70,6 +70,19 @@ _PAYLOAD_INDEX_FIELDS: dict[str, PayloadSchemaType] = {
     # migration like modified_at. Local/embedded qdrant-client matches by
     # substring without an index, so dev stacks work without it too.
     "file_path": PayloadSchemaType.TEXT,
+    # etag is the tenant-wide content-dedup key: the scanner/processor scroll for
+    # a non-placeholder point matching (doc_id, doc_type, etag) to decide whether
+    # a file's content is already indexed and reprocessing can be skipped (see
+    # vector/sharing_state.find_indexed_content). KEYWORD for exact match; the
+    # value is the Nextcloud etag already written to every point. Idempotent
+    # startup migration like the fields above.
+    "etag": PayloadSchemaType.KEYWORD,
+    # acl_principals is the observed-access ACL set ("user:<uid>" entries). Search
+    # ORs MatchAny(key="acl_principals", any=["user:<me>"]) so a deduplicated
+    # shared point reaches every reader's candidate set (see
+    # search/access_filter.build_ownership_filter); KEYWORD indexes match array
+    # membership element-wise. Idempotent startup migration.
+    "acl_principals": PayloadSchemaType.KEYWORD,
 }
 
 # Sentinel point that records "this collection has been backfilled to str

nextcloud_mcp_server/vector/scanner.py

@@ -31,6 +31,7 @@ from nextcloud_mcp_server.vector.placeholder import (
 )
 from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client
 from nextcloud_mcp_server.vector.queue.ports import TaskProducer
+from nextcloud_mcp_server.vector.sharing_state import claim_existing_index
 
 logger = logging.getLogger(__name__)
 
@@ -448,6 +449,24 @@ async def scan_user_documents(
                     except (ValueError, KeyError):
                         pass
 
+                # Tenant-wide content dedup (Layer 1 / observed-access ACL): if
+                # this exact file content (fileid + etag) is already indexed under
+                # the current embedding model by ANY user in the tenant, skip
+                # re-parsing/re-embedding and just record that this user can read
+                # it. Eliminates the per-user reprocessing ping-pong that arises
+                # because chunk point IDs are user-agnostic (note 386945 #5).
+                etag = str(file_info.get("etag") or "")
+                if etag and await claim_existing_index(file_id, "file", etag, user_id):
+                    _potentially_deleted.pop((user_id, file_id), None)
+                    logger.debug(
+                        "Dedup: file %s (ID: %s) already indexed in tenant; "
+                        "granted access to %s without reprocessing",
+                        file_path,
+                        file_id,
+                        user_id,
+                    )
+                    continue
+
                 if initial_sync:
                     # Send everything on first sync - write placeholder first
                     await write_placeholder_point(
@@ -455,6 +474,7 @@ async def scan_user_documents(
                         doc_type="file",
                         user_id=user_id,
                         modified_at=modified_at,
+                        etag=etag,
                         file_path=file_path,
                     )
                     await send_stream.send(
@@ -465,6 +485,7 @@ async def scan_user_documents(
                             operation="index",
                             modified_at=modified_at,
                             file_path=file_path,
+                            etag=etag,
                         )
                     )
                     file_queued += 1
@@ -525,6 +546,7 @@ async def scan_user_documents(
                             doc_type="file",
                             user_id=user_id,
                             modified_at=modified_at,
+                            etag=etag,
                             file_path=file_path,
                         )
                         await send_stream.send(
@@ -535,6 +557,7 @@ async def scan_user_documents(
                                 operation="index",
                                 modified_at=modified_at,
                                 file_path=file_path,
+                                etag=etag,
                             )
                         )
                         file_queued += 1

nextcloud_mcp_server/vector/sharing_state.py

@@ -0,0 +1,286 @@
+"""Tenant-wide content dedup + observed-access ACL state for the vector index.
+
+A file shared across users (directly, or via a group folder shared to a group)
+has one Nextcloud ``fileid`` and one ``etag`` for everyone, and chunk point IDs
+are user-agnostic (``uuid5(tenant_id, doc_id, chunk_index)`` — see
+``vector/payload_keys.py``). So two users indexing the same file produce the
+*same* points. The per-user freshness gate (filtered by ``user_id``) nonetheless
+made them re-parse + re-embed the identical content on every scan (note 386945,
+finding #5). This module lets the pipeline detect "already indexed by someone in
+this tenant" and skip the expensive work.
+
+Visibility is handled by an *observed-access* model rather than push-enumeration
+of share/group-folder grants (which the server cannot read without admin creds —
+group membership and the GroupFolders API are admin-only, and WebDAV PROPFIND
+carries no ACL). The per-user scanner crawl is itself the access oracle: a tagged
+file appears in a user's ``find_files_by_tag`` REPORT **iff** that user can read
+it. So each point carries ``acl_principals`` — the set of ``user:<uid>`` whose
+scanner has observed (hence can access) the file. The search filter ORs a
+``MatchAny(acl_principals, ["user:<me>"])`` branch, and ``_verify_files`` (the
+verify-on-read gate) re-checks each result against the user's tagged REPORT, so
+an over-broad principal match can never leak content.
+
+All point IDs are user-agnostic, so deletion must *release one user* (drop their
+principal) and only remove the points when the principal set empties — otherwise
+one user untagging a shared file would evict it for everyone still reading it.
+"""
+
+from __future__ import annotations
+
+import logging
+
+from qdrant_client.models import FieldCondition, Filter, MatchValue
+
+from nextcloud_mcp_server.config import get_settings
+from nextcloud_mcp_server.vector import payload_keys
+from nextcloud_mcp_server.vector.placeholder import get_placeholder_filter
+from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client
+
+logger = logging.getLogger(__name__)
+
+ACL_PRINCIPALS_KEY = "acl_principals"
+
+
+def user_principal(user_id: str) -> str:
+    """The ``acl_principals`` entry representing a single user's read access."""
+    return f"user:{user_id}"
+
+
+def _document_filter(doc_id: str, doc_type: str, *, real_only: bool) -> Filter:
+    """Match every chunk of one document; optionally exclude placeholder points."""
+    must: list = [
+        FieldCondition(key="doc_id", match=MatchValue(value=doc_id)),
+        FieldCondition(key="doc_type", match=MatchValue(value=doc_type)),
+    ]
+    if real_only:
+        must.append(get_placeholder_filter())
+    return Filter(must=must)
+
+
+async def find_indexed_content(
+    doc_id: str,
+    doc_type: str,
+    etag: str,
+    embedding_identity: str,
+) -> dict | None:
+    """Return a real point's payload if this exact content is already indexed.
+
+    Looks tenant-wide (no ``user_id`` filter) for a non-placeholder point with
+    the given ``doc_id``/``doc_type``/``etag``. The match is gated on
+    ``embedding_identity`` in Python (not the Qdrant filter, to avoid requiring an
+    index on that field): since point IDs are model-agnostic, a model switch
+    overwrites the same points, so all live points for a doc share one identity —
+    a mismatch means the existing vectors were produced by a different model and
+    must be re-embedded, so we report "not indexed".
+
+    Returns the payload dict (including ``acl_principals``) on a hit, else None.
+    """
+    if not etag:
+        return None
+    qdrant_client = await get_qdrant_client()
+    settings = get_settings()
+    points, _ = await qdrant_client.scroll(
+        collection_name=settings.get_collection_name(),
+        scroll_filter=Filter(
+            must=[
+                FieldCondition(key="doc_id", match=MatchValue(value=doc_id)),
+                FieldCondition(key="doc_type", match=MatchValue(value=doc_type)),
+                FieldCondition(key="etag", match=MatchValue(value=etag)),
+                get_placeholder_filter(),
+            ]
+        ),
+        limit=1,
+        with_payload=True,
+        with_vectors=False,
+    )
+    if not points:
+        return None
+    payload = dict(points[0].payload or {})
+    if payload.get(payload_keys.EMBEDDING_IDENTITY) != embedding_identity:
+        # Existing vectors were produced by a different embedding model — a
+        # re-embed is required, so this content is not reusable as-is.
+        return None
+    return payload
+
+
+async def existing_principals(doc_id: str, doc_type: str) -> list[str]:
+    """Return the ``acl_principals`` already recorded for a document (or []).
+
+    Used when re-indexing after a content change (etag differs, so the dedup
+    race-guard misses and the points are overwritten): seeding the new points
+    with the prior principal set preserves visibility for readers who had
+    already claimed the file, instead of resetting it to just the indexer.
+    """
+    qdrant_client = await get_qdrant_client()
+    settings = get_settings()
+    points, _ = await qdrant_client.scroll(
+        collection_name=settings.get_collection_name(),
+        scroll_filter=_document_filter(doc_id, doc_type, real_only=True),
+        limit=1,
+        with_payload=True,
+        with_vectors=False,
+    )
+    if not points:
+        return []
+    return list(dict(points[0].payload or {}).get(ACL_PRINCIPALS_KEY) or [])
+
+
+async def add_principal(
+    doc_id: str,
+    doc_type: str,
+    user_id: str,
+    current_principals: list[str] | None,
+) -> bool:
+    """Record that ``user_id`` can read this document (observed-access ACL).
+
+    No-op (returns False) when the user's principal is already present — so the
+    steady state of a repeat scan writes nothing. Otherwise unions the principal
+    onto every real chunk of the document via a single ``set_payload`` and
+    returns True. Concurrent adds race to a last-writer-wins union; a dropped
+    add is re-applied on the losing user's next scan, and verify-on-read gates
+    correctness in the meantime.
+    """
+    principal = user_principal(user_id)
+    existing = current_principals or []
+    if principal in existing:
+        return False
+    new_principals = sorted(set(existing) | {principal})
+    qdrant_client = await get_qdrant_client()
+    settings = get_settings()
+    await qdrant_client.set_payload(
+        collection_name=settings.get_collection_name(),
+        payload={ACL_PRINCIPALS_KEY: new_principals},
+        points=_document_filter(doc_id, doc_type, real_only=True),
+        wait=True,
+    )
+    logger.debug(
+        "Granted read principal %s on %s_%s (now %d principal(s))",
+        principal,
+        doc_type,
+        doc_id,
+        len(new_principals),
+    )
+    return True
+
+
+async def claim_existing_index(
+    doc_id: str,
+    doc_type: str,
+    etag: str,
+    user_id: str,
+) -> bool:
+    """Tenant-wide dedup claim: skip reprocessing if content is already indexed.
+
+    Returns True when a non-placeholder point for this exact content (fileid +
+    etag + current embedding model) already exists for some user in the tenant —
+    in which case ``user_id`` is added to ``acl_principals`` (so the file remains
+    searchable for them) and the caller should skip fetch/parse/embed. Returns
+    False when nothing reusable exists and the document must be processed.
+
+    Fail-safe: a Qdrant error during the lookup degrades to False (process the
+    document normally) rather than aborting the scan — the dedup is an
+    optimisation, never a correctness gate. A failure to record the principal
+    after a confirmed hit is non-fatal (logged, not raised): verify-on-read still
+    gates access and the user's next scan re-claims it.
+    """
+    embedding_identity = get_settings().get_embedding_model_name()
+    try:
+        existing = await find_indexed_content(
+            doc_id, doc_type, etag, embedding_identity
+        )
+    except Exception as exc:  # noqa: BLE001 — degrade to "process normally"
+        logger.warning(
+            "Dedup lookup failed for %s_%s (%s); processing without dedup",
+            doc_type,
+            doc_id,
+            exc,
+        )
+        return False
+    if existing is None:
+        return False
+    try:
+        await add_principal(doc_id, doc_type, user_id, existing.get(ACL_PRINCIPALS_KEY))
+    except Exception as exc:  # noqa: BLE001 — non-fatal; recovered on next scan
+        logger.warning(
+            "Failed to grant read principal user:%s on %s_%s (%s); "
+            "verify-on-read and the next scan will reconcile",
+            user_id,
+            doc_type,
+            doc_id,
+            exc,
+        )
+    return True
+
+
+async def release_document_for_user(
+    doc_id: str,
+    doc_type: str,
+    user_id: str,
+) -> None:
+    """Drop ``user_id``'s access to a document; delete points only when orphaned.
+
+    Replaces a blind per-document delete. Because point IDs are user-agnostic, a
+    shared document has one point set referenced by multiple principals; removing
+    one user must not evict it for the others. Removes the user's principal and
+    deletes the points only once no principal remains.
+
+    Legacy points written before ``acl_principals`` existed have no principal
+    set; for those we preserve the original behaviour (delete by
+    ``user_id``/``doc_id``/``doc_type``) so a single-owner delete still works.
+    """
+    qdrant_client = await get_qdrant_client()
+    settings = get_settings()
+    collection = settings.get_collection_name()
+
+    points, _ = await qdrant_client.scroll(
+        collection_name=collection,
+        scroll_filter=_document_filter(doc_id, doc_type, real_only=True),
+        limit=1,
+        with_payload=True,
+        with_vectors=False,
+    )
+    principals = (
+        (dict(points[0].payload or {}).get(ACL_PRINCIPALS_KEY)) if points else None
+    )
+
+    if not principals:
+        # No real points, or legacy points without a principal set: fall back to
+        # the original per-user delete (also clears this user's placeholder).
+        await qdrant_client.delete(
+            collection_name=collection,
+            points_selector=Filter(
+                must=[
+                    FieldCondition(key="user_id", match=MatchValue(value=user_id)),
+                    FieldCondition(key="doc_id", match=MatchValue(value=doc_id)),
+                    FieldCondition(key="doc_type", match=MatchValue(value=doc_type)),
+                ]
+            ),
+        )
+        return
+
+    remaining = sorted(p for p in principals if p != user_principal(user_id))
+    if not remaining:
+        # Last reader released — remove every point (real + placeholder).
+        await qdrant_client.delete(
+            collection_name=collection,
+            points_selector=_document_filter(doc_id, doc_type, real_only=False),
+        )
+        logger.info(
+            "Released last principal for %s_%s — document removed from index",
+            doc_type,
+            doc_id,
+        )
+    else:
+        await qdrant_client.set_payload(
+            collection_name=collection,
+            payload={ACL_PRINCIPALS_KEY: remaining},
+            points=_document_filter(doc_id, doc_type, real_only=True),
+            wait=True,
+        )
+        logger.info(
+            "Released principal %s for %s_%s — %d reader(s) remain",
+            user_principal(user_id),
+            doc_type,
+            doc_id,
+            len(remaining),
+        )

tests/unit/search/test_access_filter.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from typing import Any
 from unittest.mock import AsyncMock
 
 import pytest
@@ -157,26 +158,32 @@ class TestOwnersCacheBehavior:
 
 
 class TestBuildOwnershipFilter:
+    @staticmethod
+    def _by_key(flt: Filter) -> dict[str, Any]:
+        assert flt.should is not None
+        return {cond.key: cond for cond in flt.should}
+
     def test_defaults_to_self_only_when_owners_omitted(self) -> None:
         flt = build_ownership_filter("alice")
 
-        # Self-only: just the user_id branch. Self is NOT duplicated into an
-        # owner_id branch (the user_id branch already covers self-owned content).
-        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"
+        # Self-only: the user_id branch plus the observed-access acl_principals
+        # branch (so a deduplicated shared file the user has claimed is still
+        # findable). No owner_id branch — self is covered by user_id.
+        branches = self._by_key(flt)
+        assert set(branches) == {"user_id", "acl_principals"}
+        assert branches["user_id"].match.value == "alice"
+        assert branches["acl_principals"].match.any == ["user:alice"]
 
     def test_expands_owner_branch_with_accessible_owners(self) -> None:
         flt = build_ownership_filter("alice", ["alice", "bob", "carol"])
 
-        owner_branch, user_branch = flt.should
+        branches = self._by_key(flt)
+        assert set(branches) == {"owner_id", "user_id", "acl_principals"}
         # Owner branch holds only the OTHER owners — self ("alice") is excluded
         # because the user_id branch already matches self-owned content.
-        assert set(owner_branch.match.any) == {"bob", "carol"}
-        assert user_branch.key == "user_id"
-        assert user_branch.match.value == "alice"
+        assert set(branches["owner_id"].match.any) == {"bob", "carol"}
+        assert branches["user_id"].match.value == "alice"
+        assert branches["acl_principals"].match.any == ["user:alice"]
 
     def test_explicit_empty_list_omits_owner_branch_keeps_legacy(self) -> None:
         # Edge case: caller passed an explicit empty list. The owner_id branch
@@ -185,11 +192,9 @@ class TestBuildOwnershipFilter:
         # user still finds their own content from before the migration.
         flt = build_ownership_filter("alice", [])
 
-        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"
+        branches = self._by_key(flt)
+        assert set(branches) == {"user_id", "acl_principals"}
+        assert branches["user_id"].match.value == "alice"
 
 
 class TestBuildBaseFilterConditions:

tests/unit/test_document_parse_metrics.py

@@ -312,8 +312,12 @@ class TestProcessDocumentMetricCounting:
         )
 
         qmock = MagicMock()
-        qmock.delete = AsyncMock()
-        with patch.object(proc, "get_qdrant_client", new=AsyncMock(return_value=qmock)):
+        # Deletion now delegates to release_document_for_user (release-one-user
+        # semantics); stub it so the test exercises only the metric accounting.
+        with (
+            patch.object(proc, "get_qdrant_client", new=AsyncMock(return_value=qmock)),
+            patch.object(proc, "release_document_for_user", new=AsyncMock()),
+        ):
             await proc.process_document(task, MagicMock())
 
         assert metric_sample(
@@ -339,8 +343,16 @@ class TestProcessDocumentMetricCounting:
         )
 
         qmock = MagicMock()
-        qmock.delete = AsyncMock(side_effect=RuntimeError("boom"))
-        with patch.object(proc, "get_qdrant_client", new=AsyncMock(return_value=qmock)):
+        # A failed release still counts as a processed (error) delete and must
+        # not touch the indexed counter.
+        with (
+            patch.object(proc, "get_qdrant_client", new=AsyncMock(return_value=qmock)),
+            patch.object(
+                proc,
+                "release_document_for_user",
+                new=AsyncMock(side_effect=RuntimeError("boom")),
+            ),
+        ):
             with pytest.raises(RuntimeError):
                 await proc.process_document(task, MagicMock())
 

tests/unit/vector/test_sharing_state.py

@@ -0,0 +1,229 @@
+"""Unit tests for tenant-wide content dedup + observed-access ACL state.
+
+Covers vector/sharing_state.py: the tenant-wide content lookup that lets a
+shared/group-folder file be parsed+embedded once per tenant instead of once per
+user, and the ``acl_principals`` maintenance (grant/release) that keeps a
+deduplicated point findable by every reader without re-indexing.
+
+All functions reach Qdrant via ``get_qdrant_client`` and resolve the collection
+via ``get_settings``; both are monkeypatched here so the logic is exercised
+without a live Qdrant.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+
+from nextcloud_mcp_server.vector import payload_keys
+from nextcloud_mcp_server.vector import sharing_state as ss
+
+pytestmark = pytest.mark.unit
+
+_COLLECTION = "test_collection"
+_MODEL = "model-x"
+
+
+class _Settings:
+    def get_collection_name(self) -> str:
+        return _COLLECTION
+
+    def get_embedding_model_name(self) -> str:
+        return _MODEL
+
+
+def _point(payload: dict) -> SimpleNamespace:
+    """Stand-in for a qdrant_client Record (only id/payload are read)."""
+    return SimpleNamespace(id="pt", payload=payload)
+
+
+@pytest.fixture
+def client(monkeypatch) -> AsyncMock:
+    """An AsyncMock Qdrant client wired into sharing_state, with a stub Settings.
+
+    ``scroll`` defaults to "no points"; individual tests override
+    ``client.scroll.return_value``/``side_effect``.
+    """
+    qc = AsyncMock()
+    qc.scroll.return_value = ([], None)
+    monkeypatch.setattr(ss, "get_qdrant_client", AsyncMock(return_value=qc))
+    monkeypatch.setattr(ss, "get_settings", lambda: _Settings())
+    return qc
+
+
+def _must_keys(flt) -> list[str | None]:
+    """Collect the FieldCondition keys in a Filter's ``must`` clause."""
+    return [getattr(c, "key", None) for c in (flt.must or [])]
+
+
+class TestFindIndexedContent:
+    async def test_returns_payload_on_etag_and_model_match(self, client) -> None:
+        payload = {
+            "doc_id": "42",
+            "etag": "abc",
+            payload_keys.EMBEDDING_IDENTITY: _MODEL,
+            ss.ACL_PRINCIPALS_KEY: ["user:alice"],
+        }
+        client.scroll.return_value = ([_point(payload)], None)
+
+        result = await ss.find_indexed_content("42", "file", "abc", _MODEL)
+        assert result == payload
+
+    async def test_none_when_no_points(self, client) -> None:
+        client.scroll.return_value = ([], None)
+        assert await ss.find_indexed_content("42", "file", "abc", _MODEL) is None
+
+    async def test_none_on_embedding_model_mismatch(self, client) -> None:
+        # A model switch overwrites the same point IDs; existing vectors made by
+        # a different model must be re-embedded, so this reports "not indexed".
+        client.scroll.return_value = (
+            [_point({payload_keys.EMBEDDING_IDENTITY: "other-model"})],
+            None,
+        )
+        assert await ss.find_indexed_content("42", "file", "abc", _MODEL) is None
+
+    async def test_empty_etag_short_circuits_without_query(self, client) -> None:
+        assert await ss.find_indexed_content("42", "file", "", _MODEL) is None
+        client.scroll.assert_not_called()
+
+
+class TestAddPrincipal:
+    async def test_noop_when_principal_already_present(self, client) -> None:
+        added = await ss.add_principal("42", "file", "alice", ["user:alice"])
+        assert added is False
+        client.set_payload.assert_not_called()
+
+    async def test_unions_principal_when_absent(self, client) -> None:
+        added = await ss.add_principal("42", "file", "bob", ["user:alice"])
+        assert added is True
+        client.set_payload.assert_awaited_once()
+        kwargs = client.set_payload.await_args.kwargs
+        assert kwargs["payload"][ss.ACL_PRINCIPALS_KEY] == ["user:alice", "user:bob"]
+        # Updates only real (non-placeholder) chunks of this document.
+        assert _must_keys(kwargs["points"]) == ["doc_id", "doc_type", "is_placeholder"]
+
+    async def test_handles_none_current_principals(self, client) -> None:
+        added = await ss.add_principal("42", "file", "alice", None)
+        assert added is True
+        kwargs = client.set_payload.await_args.kwargs
+        assert kwargs["payload"][ss.ACL_PRINCIPALS_KEY] == ["user:alice"]
+
+
+class TestClaimExistingIndex:
+    async def test_true_and_grants_principal_on_hit(self, client) -> None:
+        client.scroll.return_value = (
+            [
+                _point(
+                    {
+                        payload_keys.EMBEDDING_IDENTITY: _MODEL,
+                        ss.ACL_PRINCIPALS_KEY: ["user:alice"],
+                    }
+                )
+            ],
+            None,
+        )
+        claimed = await ss.claim_existing_index("42", "file", "abc", "bob")
+        assert claimed is True
+        # bob was added to the existing point's principals.
+        client.set_payload.assert_awaited_once()
+        assert client.set_payload.await_args.kwargs["payload"][
+            ss.ACL_PRINCIPALS_KEY
+        ] == ["user:alice", "user:bob"]
+
+    async def test_false_when_not_indexed(self, client) -> None:
+        client.scroll.return_value = ([], None)
+        assert await ss.claim_existing_index("42", "file", "abc", "bob") is False
+        client.set_payload.assert_not_called()
+
+    async def test_hit_for_already_listed_user_writes_nothing(self, client) -> None:
+        client.scroll.return_value = (
+            [
+                _point(
+                    {
+                        payload_keys.EMBEDDING_IDENTITY: _MODEL,
+                        ss.ACL_PRINCIPALS_KEY: ["user:alice"],
+                    }
+                )
+            ],
+            None,
+        )
+        # alice already present -> claim still True (skip reprocess) but no write.
+        assert await ss.claim_existing_index("42", "file", "abc", "alice") is True
+        client.set_payload.assert_not_called()
+
+    async def test_lookup_error_degrades_to_process_normally(self, client) -> None:
+        # A Qdrant hiccup during dedup must not abort the scan — fall back to
+        # processing the document (return False), not raise.
+        client.scroll.side_effect = RuntimeError("qdrant down")
+        assert await ss.claim_existing_index("42", "file", "abc", "bob") is False
+
+    async def test_principal_grant_failure_after_hit_is_non_fatal(self, client) -> None:
+        # The content IS indexed (skip reprocess), so a failure to record the
+        # principal still returns True; verify-on-read + next scan reconcile.
+        client.scroll.return_value = (
+            [_point({payload_keys.EMBEDDING_IDENTITY: _MODEL})],
+            None,
+        )
+        client.set_payload.side_effect = RuntimeError("set_payload failed")
+        assert await ss.claim_existing_index("42", "file", "abc", "bob") is True
+
+
+class TestExistingPrincipals:
+    async def test_returns_recorded_principals(self, client) -> None:
+        client.scroll.return_value = (
+            [_point({ss.ACL_PRINCIPALS_KEY: ["user:alice", "user:bob"]})],
+            None,
+        )
+        assert await ss.existing_principals("42", "file") == ["user:alice", "user:bob"]
+
+    async def test_empty_when_no_points(self, client) -> None:
+        client.scroll.return_value = ([], None)
+        assert await ss.existing_principals("42", "file") == []
+
+
+class TestReleaseDocumentForUser:
+    async def test_keeps_points_and_trims_principals_when_readers_remain(
+        self, client
+    ) -> None:
+        client.scroll.return_value = (
+            [_point({ss.ACL_PRINCIPALS_KEY: ["user:alice", "user:bob"]})],
+            None,
+        )
+        await ss.release_document_for_user("42", "file", "alice")
+
+        client.delete.assert_not_called()
+        kwargs = client.set_payload.await_args.kwargs
+        assert kwargs["payload"][ss.ACL_PRINCIPALS_KEY] == ["user:bob"]
+
+    async def test_deletes_all_points_when_last_reader_released(self, client) -> None:
+        client.scroll.return_value = (
+            [_point({ss.ACL_PRINCIPALS_KEY: ["user:alice"]})],
+            None,
+        )
+        await ss.release_document_for_user("42", "file", "alice")
+
+        client.set_payload.assert_not_called()
+        client.delete.assert_awaited_once()
+        selector = client.delete.await_args.kwargs["points_selector"]
+        # Whole document removed: doc_id + doc_type, no user_id, no placeholder gate.
+        assert _must_keys(selector) == ["doc_id", "doc_type"]
+
+    async def test_legacy_points_without_principals_delete_by_user(
+        self, client
+    ) -> None:
+        # Pre-acl_principals points: preserve the original per-user delete.
+        client.scroll.return_value = ([_point({"doc_id": "42"})], None)
+        await ss.release_document_for_user("42", "file", "alice")
+
+        client.set_payload.assert_not_called()
+        selector = client.delete.await_args.kwargs["points_selector"]
+        assert _must_keys(selector) == ["user_id", "doc_id", "doc_type"]
+
+    async def test_no_points_falls_back_to_per_user_delete(self, client) -> None:
+        client.scroll.return_value = ([], None)
+        await ss.release_document_for_user("42", "file", "alice")
+
+        selector = client.delete.await_args.kwargs["points_selector"]
+        assert _must_keys(selector) == ["user_id", "doc_id", "doc_type"]