diff --git a/nextcloud_mcp_server/api/__init__.py b/nextcloud_mcp_server/api/__init__.py
index 6b79a44e..05d7bb39 100644
--- a/nextcloud_mcp_server/api/__init__.py
+++ b/nextcloud_mcp_server/api/__init__.py
@@ -36,6 +36,9 @@ from nextcloud_mcp_server.api.passwords import (
     get_app_password_status,
     provision_app_password,
 )
+from nextcloud_mcp_server.api.vector_sync import (
+    purge_doc_types_route,
+)
 from nextcloud_mcp_server.api.visualization import (
     get_chunk_context,
     get_pdf_preview,
@@ -78,6 +81,8 @@ __all__ = [
     "list_webhooks",
     "create_webhook",
     "delete_webhook",
+    # Vector-sync admin endpoints (from vector_sync.py)
+    "purge_doc_types_route",
     # Visualization endpoints (from visualization.py)
     "unified_search",
     "vector_search",
diff --git a/nextcloud_mcp_server/api/vector_sync.py b/nextcloud_mcp_server/api/vector_sync.py
new file mode 100644
index 00000000..4d49ed77
--- /dev/null
+++ b/nextcloud_mcp_server/api/vector_sync.py
@@ -0,0 +1,152 @@
+"""Vector-sync admin API endpoints.
+
+Provides the purge endpoint Astrolabe calls when an admin disables a content
+source for semantic search. Consent is binding on data-at-rest, so the
+already-indexed content for the disabled source's doc type(s) is deleted
+globally (every owner) — see :mod:`nextcloud_mcp_server.vector.purge`.
+
+Auth: the OAuth bearer identifies the caller (``validate_token_and_get_user``);
+because the purge deletes every owner's content for a doc type, it is further
+restricted to Nextcloud administrators (verified via the ``admin`` group using
+the caller's app password). This is stricter than the per-user webhook routes
+in :mod:`nextcloud_mcp_server.api.webhooks` precisely because the blast radius
+is global.
+"""
+
+import logging
+
+import httpx
+from starlette.requests import Request
+from starlette.responses import JSONResponse
+
+from nextcloud_mcp_server.api._auth import get_basic_auth_for_user
+from nextcloud_mcp_server.api.management import (
+    _sanitize_error_for_client,
+    validate_token_and_get_user,
+)
+from nextcloud_mcp_server.auth.scope_authorization import ProvisioningRequiredError
+from nextcloud_mcp_server.client.users import UsersClient
+from nextcloud_mcp_server.vector.purge import purge_doc_types
+
+from ..http import nextcloud_httpx_client
+
+logger = logging.getLogger(__name__)
+
+# Upper bound on doc_types per purge request. There are only a handful of real
+# indexed types; this caps a hostile/buggy caller's fan-out of count+delete
+# calls without constraining legitimate use.
+_MAX_PURGE_DOC_TYPES = 64
+
+
+def _bad_request(message: str) -> JSONResponse:
+    return JSONResponse({"error": "Bad request", "message": message}, status_code=400)
+
+
+async def purge_doc_types_route(request: Request) -> JSONResponse:
+    """POST /api/v1/vector-sync/purge — delete indexed vectors by doc type.
+
+    Request body::
+
+        {"doc_types": ["file", "note"]}
+
+    Returns ``{"purged": {doc_type: deleted_count}}``. Admin-only.
+
+    Requires OAuth bearer token for authentication.
+    """
+    try:
+        user_id, _ = await validate_token_and_get_user(request)
+    except Exception as e:
+        logger.warning("Unauthorized access to /api/v1/vector-sync/purge: %s", e)
+        return JSONResponse(
+            {
+                "error": "Unauthorized",
+                "message": _sanitize_error_for_client(e, "purge_doc_types"),
+            },
+            status_code=401,
+        )
+
+    try:
+        body = await request.json()
+    except Exception as e:
+        logger.warning("Purge payload was not valid JSON: %s", e)
+        return _bad_request("invalid JSON")
+
+    if not isinstance(body, dict):
+        return _bad_request("body must be a JSON object")
+
+    raw = body.get("doc_types")
+    if raw is None:
+        return _bad_request("doc_types is required")
+    if not isinstance(raw, list) or not all(isinstance(d, str) for d in raw):
+        return _bad_request("doc_types must be a list of strings")
+    doc_types = [d for d in raw if d]
+    # No whitelist against INDEXED_DOC_TYPES on purpose: an unknown type yields a
+    # zero-match Qdrant filter (harmless no-op), and the canonical set lives with
+    # the indexer — the route shouldn't need a server update to purge a new type.
+    # Bound the batch: there are only a handful of real indexed types, so a huge
+    # list is abuse — cap it rather than fan out unbounded count+delete calls.
+    if len(doc_types) > _MAX_PURGE_DOC_TYPES:
+        return _bad_request(f"doc_types exceeds the maximum of {_MAX_PURGE_DOC_TYPES}")
+
+    try:
+        username, app_password = await get_basic_auth_for_user(user_id)
+
+        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")
+
+        # Verify admin via the caller's own app password before any deletion —
+        # enforced even for an empty (no-op) request, since this is a
+        # destructive admin route.
+        async with nextcloud_httpx_client(
+            base_url=nextcloud_host,
+            auth=httpx.BasicAuth(username, app_password),
+            timeout=30.0,
+        ) as client:
+            users_client = UsersClient(client, username)
+            user_groups = await users_client.get_user_groups(username)
+            if "admin" not in user_groups:
+                logger.warning("Non-admin user %s attempted vector-sync purge", user_id)
+                return JSONResponse(
+                    {
+                        "error": "Forbidden",
+                        "message": "Administrator privileges required",
+                    },
+                    status_code=403,
+                )
+
+        if not doc_types:
+            return JSONResponse({"purged": {}})
+
+        purged = await purge_doc_types(doc_types)
+        # Surface a partial-failure signal so Astrolabe knows which types were
+        # NOT purged (consent not yet enforced for them) — the scanner backstop
+        # still catches these, but the caller shouldn't assume full success.
+        failed = [dt for dt in dict.fromkeys(doc_types) if dt not in purged]
+        resp: dict = {"purged": purged}
+        if failed:
+            resp["failed"] = failed
+        logger.info(
+            "Vector-sync purge by admin %s: purged=%s failed=%s",
+            user_id,
+            purged,
+            failed,
+        )
+        return JSONResponse(resp)
+
+    except ProvisioningRequiredError as e:
+        logger.info("Provisioning required for user %s: %s", user_id, e)
+        return JSONResponse(
+            {"error": "Provisioning required", "message": str(e)},
+            status_code=428,
+        )
+    except Exception as e:
+        logger.exception("Error purging doc types for user %s", user_id)
+        return JSONResponse(
+            {
+                "error": "Internal error",
+                "message": _sanitize_error_for_client(e, "purge_doc_types"),
+            },
+            status_code=500,
+        )
diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py
index 6168fbc2..39c383b8 100644
--- a/nextcloud_mcp_server/app.py
+++ b/nextcloud_mcp_server/app.py
@@ -47,6 +47,7 @@ from nextcloud_mcp_server.api import (
     list_supported_scopes,
     list_webhooks,
     provision_app_password,
+    purge_doc_types_route,
     revoke_user_access,
     unified_search,
     update_user_scopes,
@@ -2422,6 +2423,19 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
         routes.append(
             Route("/api/v1/webhooks/{webhook_id}", delete_webhook, methods=["DELETE"])
         )
+        # Vector-sync admin: purge indexed vectors by doc type (admin consent —
+        # called by Astrolabe when a source is disabled for semantic search).
+        # Gated on vector_sync_enabled: without it there is no Qdrant client, so
+        # the purge would 500 rather than no-op.
+        if settings.vector_sync_enabled:
+            routes.append(
+                Route(
+                    "/api/v1/vector-sync/purge",
+                    purge_doc_types_route,
+                    methods=["POST"],
+                )
+            )
+            logger.info("Vector-sync admin endpoint enabled: /api/v1/vector-sync/purge")
         # Access and scope management endpoints (ADR-022)
         routes.append(
             Route(
diff --git a/nextcloud_mcp_server/capabilities.py b/nextcloud_mcp_server/capabilities.py
new file mode 100644
index 00000000..40fc75c7
--- /dev/null
+++ b/nextcloud_mcp_server/capabilities.py
@@ -0,0 +1,123 @@
+"""Reads admin-approved searchable sources from the Astrolabe capability.
+
+The Astrolabe Nextcloud app advertises, per user, which content sources an
+admin has approved for semantic search, under
+``capabilities.astrolabe.semantic_search.enabled_doc_types`` on the OCS
+capabilities endpoint (``/ocs/v2.php/cloud/capabilities``). This is the single
+source of truth for admin consent: the search layer filters results to these
+doc types, and the indexing layer (scanner + webhook ingest) skips everything
+else (the hard data-at-rest guarantee is the eager purge Astrolabe triggers on
+disable; see ``vector.purge``).
+
+Fail-open for *availability*: if the capability block is absent (an older
+Astrolabe that predates this feature) or the OCS call fails, ``allowed_doc_types``
+returns ``None`` meaning "no restriction", so search keeps working. ``None`` is
+distinct from an empty set, which means "the admin disabled every source".
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from collections import OrderedDict
+from typing import Any, Protocol
+
+logger = logging.getLogger(__name__)
+
+# Short-lived per-user cache for the OCS capabilities lookup. Admin consent
+# changes rarely, but search/scan paths consult it frequently, so trade a little
+# staleness for keeping the OCS round-trip off the hot path. Mirrors the
+# list_accessible_owners cache in search/access_filter.py.
+#
+# Keyed by user_id even though enabled_doc_types is an admin-wide value: the OCS
+# call is authenticated per-user (and ``installed`` resolves per-user on the
+# Astrolabe side), so we cache per-user for correctness. The redundancy is
+# bounded by _CACHE_MAXSIZE; on an admin change all entries reconverge within
+# one TTL window.
+_CACHE_TTL_SECONDS = 30.0
+_CACHE_MAXSIZE = 1024
+# user_id -> (monotonic_ts, frozenset[doc_type] | None). None = no restriction.
+_cache: OrderedDict[str, tuple[float, frozenset[str] | None]] = OrderedDict()
+
+
+class _CapabilitiesClientProtocol(Protocol):
+    async def capabilities(self) -> Any: ...
+
+
+def _parse_enabled_doc_types(payload: Any) -> frozenset[str] | None:
+    """Extract ``enabled_doc_types`` from an OCS capabilities payload.
+
+    Returns ``None`` when the ``astrolabe.semantic_search`` block is absent or
+    malformed (treated as "no restriction"). Returns a frozenset (possibly
+    empty) when the block is present and well-formed; an empty set means the
+    admin disabled every source.
+    """
+    if not isinstance(payload, dict):
+        return None
+    try:
+        caps = payload["ocs"]["data"]["capabilities"]
+    except (KeyError, TypeError):
+        return None
+    if not isinstance(caps, dict):
+        return None
+    block = caps.get("astrolabe")
+    if not isinstance(block, dict):
+        return None
+    semantic = block.get("semantic_search")
+    if not isinstance(semantic, dict):
+        return None
+    raw = semantic.get("enabled_doc_types")
+    if not isinstance(raw, list):
+        return None
+    return frozenset(dt for dt in raw if isinstance(dt, str) and dt)
+
+
+async def allowed_doc_types(
+    client: _CapabilitiesClientProtocol, user_id: str
+) -> frozenset[str] | None:
+    """Admin-approved doc types for ``user_id``, or ``None`` for "no restriction".
+
+    Cached per user with a short TTL (+ LRU eviction). Failures are not cached so
+    a transient OCS hiccup retries on the next call. Fail-open: a missing
+    capability block or an error yields ``None`` so search remains available.
+    """
+    now = time.monotonic()
+    cached = _cache.get(user_id)
+    if cached is not None and now - cached[0] < _CACHE_TTL_SECONDS:
+        _cache.move_to_end(user_id)  # mark recently used (LRU)
+        return cached[1]
+
+    try:
+        payload = await client.capabilities()
+    except Exception as exc:  # noqa: BLE001 — degrade gracefully (fail-open)
+        logger.warning(
+            "Astrolabe capabilities unavailable for user %s (%s); "
+            "not restricting doc types this cycle",
+            user_id,
+            exc,
+        )
+        return None  # don't cache failures — retry next call
+
+    result = _parse_enabled_doc_types(payload)
+    _cache[user_id] = (now, result)
+    # Needed only for an existing (expired) key: __setitem__ updates it in place,
+    # keeping its old position, so move it to the end to preserve LRU order. For
+    # a brand-new key __setitem__ already appends, so this is a harmless no-op.
+    _cache.move_to_end(user_id)
+    while len(_cache) > _CACHE_MAXSIZE:
+        _cache.popitem(last=False)  # evict least-recently-used
+    return result
+
+
+def is_doc_type_allowed(doc_type: str, allowed: frozenset[str] | None) -> bool:
+    """Whether ``doc_type`` may be indexed/searched given an allow-set.
+
+    ``allowed=None`` means "no restriction" (fail-open / older Astrolabe), so
+    everything is permitted.
+    """
+    return allowed is None or doc_type in allowed
+
+
+def clear_cache() -> None:
+    """Test hook: drop all cached entries."""
+    _cache.clear()
diff --git a/nextcloud_mcp_server/server/semantic.py b/nextcloud_mcp_server/server/semantic.py
index 0696fbcc..27f84102 100644
--- a/nextcloud_mcp_server/server/semantic.py
+++ b/nextcloud_mcp_server/server/semantic.py
@@ -20,6 +20,7 @@ from mcp.types import (
 from pydantic import Field
 
 from nextcloud_mcp_server.auth import require_scopes
+from nextcloud_mcp_server.capabilities import allowed_doc_types
 from nextcloud_mcp_server.config import get_settings
 from nextcloud_mcp_server.context import get_client
 from nextcloud_mcp_server.models.semantic import (
@@ -56,6 +57,25 @@ logger = logging.getLogger(__name__)
 _USAGE_METADATA_MAX_DOC_TYPES = 16
 
 
+def _consent_narrowed_doc_types(
+    doc_types: list[str] | None, allowed: frozenset[str]
+) -> list[str]:
+    """Apply the admin allow-set to a requested ``doc_types`` filter.
+
+    Caller has already established ``allowed is not None`` (a concrete allow-set;
+    ``None`` means "no restriction" and is handled by skipping this call). When
+    no explicit ``doc_types`` are requested, restrict to the full allow-set
+    (returned ``sorted`` for determinism only — order is a filter, not a ranking
+    hint); otherwise intersect (preserving the caller's order). An empty result
+    means nothing the caller asked for is admin-approved — the caller
+    short-circuits to an empty response rather than falling through to an
+    all-types search.
+    """
+    if doc_types is None:
+        return sorted(allowed)
+    return [dt for dt in doc_types if dt in allowed]
+
+
 async def record_search_usage(
     *,
     enabled: bool,
@@ -300,6 +320,37 @@ def configure_semantic_tools(mcp: FastMCP):
         # files under their own user_id.
         accessible_owners = await list_accessible_owners(client.sharing, username)
 
+        # Admin consent gate: restrict to source types the Astrolabe admin has
+        # approved (and that are installed for this user). This mirrors
+        # Astrolabe's own server-side enforcement but is independent because
+        # this tool queries Qdrant directly. ``None`` = no restriction
+        # (fail-open / Astrolabe predating this feature). An empty allow-set
+        # means the admin disabled every source.
+        #
+        # Perf trade-off (accepted): when Astrolabe is present and the caller
+        # passed no doc_types, narrowing turns ``None`` into a concrete list, so
+        # the search takes the per-type query branch (N queries) instead of the
+        # single cross-type query. N is the count of admin-approved types
+        # (typically 1-4), so the overhead is small; left as-is rather than
+        # adding a "search all approved in one query" fast path.
+        allowed = await allowed_doc_types(client, username)
+        if allowed is not None:
+            doc_types = _consent_narrowed_doc_types(doc_types, allowed)
+            if not doc_types:
+                logger.info(
+                    "Semantic search short-circuited for user %s: no requested "
+                    "doc_type is admin-approved for semantic search",
+                    username,
+                )
+                return SemanticSearchResponse(
+                    results=[],
+                    query=query,
+                    total_found=0,
+                    search_method=f"bm25_hybrid_{fusion}",
+                    verified_chunk_count=0,
+                    dropped_document_count=0,
+                )
+
         try:
             # The nc_semantic_search tool deliberately uses BM25-hybrid (dense +
             # sparse with RRF/DBSF fusion) as the single tool-layer algorithm.
diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py
index 2ed2ccce..8d517862 100644
--- a/nextcloud_mcp_server/vector/processor.py
+++ b/nextcloud_mcp_server/vector/processor.py
@@ -21,6 +21,7 @@ if TYPE_CHECKING:
     from nextcloud_mcp_server.document_processors.registry import ProcessorRegistry
 
 from nextcloud_mcp_server.acl_hash import compute_acl_hash
+from nextcloud_mcp_server.capabilities import allowed_doc_types, is_doc_type_allowed
 from nextcloud_mcp_server.client import NextcloudClient
 from nextcloud_mcp_server.config import get_settings
 from nextcloud_mcp_server.embedding import get_bm25_service, get_embedding_service
@@ -590,6 +591,27 @@ async def process_document(
             ):
                 await _reconcile_tag_event(doc_task, nc_client)
 
+            # Admin consent gate (Astrolabe): never index a source the admin has
+            # disabled for semantic search — this catches near-real-time webhook
+            # events that bypass the scanner's discovery gate. Deletes always
+            # proceed (removing data honours consent). ``None`` from the reader
+            # means no restriction (fail-open / older Astrolabe), so a transient
+            # capabilities failure never silently drops indexing.
+            if doc_task.operation == "index":
+                allowed = await allowed_doc_types(nc_client, doc_task.user_id)
+                if not is_doc_type_allowed(doc_task.doc_type, allowed):
+                    logger.info(
+                        "Skipping index of %s_%s for %s: doc_type disabled by admin",
+                        doc_task.doc_type,
+                        doc_task.doc_id,
+                        doc_task.user_id,
+                    )
+                    # Alertable counter so a flood of webhook events for a
+                    # disabled source is observable (not silently swallowed).
+                    record_ingest_dropped("admin_disabled")
+                    record_vector_sync_processing(time.time() - start_time, "skipped")
+                    return
+
             # Handle deletion
             if doc_task.operation == "delete":
                 # Release this user rather than blind-delete: a file shared across
diff --git a/nextcloud_mcp_server/vector/purge.py b/nextcloud_mcp_server/vector/purge.py
new file mode 100644
index 00000000..7e363d3a
--- /dev/null
+++ b/nextcloud_mcp_server/vector/purge.py
@@ -0,0 +1,77 @@
+"""Global purge of indexed vectors by doc type (admin consent enforcement).
+
+When an admin disables a content source for semantic search in Astrolabe,
+consent is binding on data-at-rest: the already-indexed content for that
+source's doc type(s) must be deleted, not merely hidden. Astrolabe calls the
+``/api/v1/vector-sync/purge`` route on disable, which delegates here.
+
+The purge is global (every owner) because the admin disable is a global
+decision. It is safe to call for a doc type with no indexed points — Qdrant
+deletes zero points and reports a count of 0.
+"""
+
+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.qdrant_client import get_qdrant_client
+
+logger = logging.getLogger(__name__)
+
+
+def _doc_type_filter(doc_type: str) -> Filter:
+    return Filter(
+        must=[FieldCondition(key="doc_type", match=MatchValue(value=doc_type))]
+    )
+
+
+async def purge_doc_types(doc_types: list[str]) -> dict[str, int]:
+    """Delete every indexed point whose ``doc_type`` is in ``doc_types``.
+
+    Returns a mapping of doc_type -> number of points deleted (counted before
+    deletion). Each doc type is purged independently so a failure on one does
+    not abort the rest; failures re-raise after the loop only if every doc type
+    failed, otherwise partial progress is returned.
+
+    The count is taken just before the delete (two separate Qdrant calls), so
+    it is approximate — a point indexed in the gap is deleted but not counted.
+    This is acceptable: indexing of a disabled source is already gated upstream,
+    so the window is effectively empty in practice.
+    """
+    qdrant_client = await get_qdrant_client()
+    collection = get_settings().get_collection_name()
+
+    purged: dict[str, int] = {}
+    last_error: Exception | None = None
+    for doc_type in dict.fromkeys(doc_types):  # de-dupe, preserve order
+        flt = _doc_type_filter(doc_type)
+        try:
+            count_result = await qdrant_client.count(
+                collection_name=collection,
+                count_filter=flt,
+                exact=True,
+            )
+            await qdrant_client.delete(
+                collection_name=collection,
+                points_selector=flt,
+            )
+            purged[doc_type] = int(count_result.count)
+            logger.info(
+                "Purged %d indexed point(s) for disabled doc_type=%s",
+                purged[doc_type],
+                doc_type,
+            )
+        except Exception as exc:  # noqa: BLE001 — record and continue
+            last_error = exc
+            logger.exception(
+                "Failed to purge indexed points for doc_type=%s",
+                doc_type,
+            )
+
+    if not purged and last_error is not None:
+        # Nothing succeeded — surface the failure to the caller (HTTP 500).
+        raise last_error
+    return purged
diff --git a/nextcloud_mcp_server/vector/scanner.py b/nextcloud_mcp_server/vector/scanner.py
index b1085aa0..b00d4fe1 100644
--- a/nextcloud_mcp_server/vector/scanner.py
+++ b/nextcloud_mcp_server/vector/scanner.py
@@ -16,6 +16,7 @@ from httpx import HTTPStatusError
 from qdrant_client import AsyncQdrantClient
 from qdrant_client.models import FieldCondition, Filter, MatchValue, Record
 
+from nextcloud_mcp_server.capabilities import allowed_doc_types, is_doc_type_allowed
 from nextcloud_mcp_server.client import NextcloudClient
 from nextcloud_mcp_server.client.news import NewsItemType
 from nextcloud_mcp_server.config import get_settings
@@ -279,6 +280,150 @@ def _app_enabled(app_id: str, enabled_apps: set[str] | None) -> bool:
     return enabled_apps is None or app_id in enabled_apps
 
 
+def _should_scan(
+    app_id: str,
+    doc_type: str,
+    enabled_apps: set[str] | None,
+    allowed: frozenset[str] | None,
+) -> bool:
+    """Whether to scan ``app_id``: installed for the user AND admin-approved."""
+    return _app_enabled(app_id, enabled_apps) and is_doc_type_allowed(doc_type, allowed)
+
+
+# Text doc types whose deletion-tracking lives *inside* their scan_* function,
+# so skipping that function (when admin-disabled) leaves indexed points with no
+# grace-period backstop. Derived from INDEXED_DOC_TYPES so a newly-indexed type
+# automatically gets the backstop. ``file`` is excluded: its scan path empties
+# discovery and lets the existing reconcile loop purge on disable.
+_TEXT_BACKSTOP_DOC_TYPES: tuple[str, ...] = tuple(sorted(INDEXED_DOC_TYPES - {"file"}))
+
+# Per-process record of (user_id, doc_type) whose consent backstop deletes have
+# already been enqueued, so a *standing* admin-disable doesn't re-flood the
+# processor with idempotent deletes on every scan tick. An entry is cleared once
+# the type is allowed again, so a later re-disable re-triggers the backstop.
+# A dict (not a set) so it stays insertion-ordered for oldest-first eviction.
+_consent_backstop_done: dict[tuple[str, str], None] = {}
+
+# Safety bound on the tracking dict so a long-running multi-tenant process with
+# heavy user churn (deprovisioned users leave stale entries) can't grow it
+# without limit. At <= len(INDEXED_DOC_TYPES) entries per user this is generous;
+# on overflow we evict the *oldest* entries down to half capacity (not a full
+# clear) so the backstop re-fires for only those, avoiding a fleet-wide burst.
+_CONSENT_BACKSTOP_MAX = 50_000
+
+
+def _mark_backstop_done(key: tuple[str, str]) -> None:
+    """Record a one-shot backstop marker, evicting oldest entries on overflow.
+
+    Evicts oldest-first down to half capacity (insertion-ordered dict) rather
+    than clearing wholesale, so a bound hit re-fires the backstop only for the
+    oldest markers, not the whole fleet. A re-fire is idempotent regardless.
+    """
+    if len(_consent_backstop_done) >= _CONSENT_BACKSTOP_MAX:
+        overage = len(_consent_backstop_done) - _CONSENT_BACKSTOP_MAX // 2
+        logger.info(
+            "consent backstop tracking hit %d entries; evicting %d oldest",
+            _CONSENT_BACKSTOP_MAX,
+            overage,
+        )
+        for stale_key in list(_consent_backstop_done)[:overage]:
+            del _consent_backstop_done[stale_key]
+    _consent_backstop_done[key] = None
+
+
+async def _backstop_delete_doc_type(
+    user_id: str,
+    send_stream: TaskProducer,
+    doc_type: str,
+    qdrant_client: AsyncQdrantClient,
+    collection: str,
+    scan_id: int,
+) -> int:
+    """Enqueue delete tasks for every indexed point of one disabled doc_type.
+
+    Returns the number of delete tasks enqueued.
+    """
+    points = await _scroll_all_points(
+        qdrant_client,
+        collection_name=collection,
+        scroll_filter=Filter(
+            must=[
+                FieldCondition(key="user_id", match=MatchValue(value=user_id)),
+                FieldCondition(key="doc_type", match=MatchValue(value=doc_type)),
+            ]
+        ),
+        payload_fields=["doc_id"],
+    )
+    doc_ids = {
+        str(p.payload["doc_id"])
+        for p in points
+        if p.payload is not None and "doc_id" in p.payload
+    }
+    if doc_ids:
+        logger.info(
+            "[SCAN-%s] %s disabled by admin for %s; enqueueing %d delete(s) (backstop)",
+            scan_id,
+            doc_type,
+            user_id,
+            len(doc_ids),
+        )
+    for doc_id in doc_ids:
+        await send_stream.send(
+            DocumentTask(
+                user_id=user_id,
+                doc_id=doc_id,
+                doc_type=doc_type,
+                operation="delete",
+                modified_at=0,
+            )
+        )
+    return len(doc_ids)
+
+
+async def _enqueue_deletes_for_disabled_types(
+    user_id: str,
+    send_stream: TaskProducer,
+    allowed: frozenset[str] | None,
+    scan_id: int,
+) -> int:
+    """Enqueue delete tasks for indexed text-source points the admin disabled.
+
+    Backstop for a failed eager purge: scrolls this user's indexed points for
+    each admin-disallowed text doc_type and queues a delete. No-op when
+    ``allowed`` is ``None`` (fail-open — never delete on a transient capability
+    read failure). Returns the number of delete tasks enqueued.
+    """
+    if allowed is None:
+        return 0
+
+    # Re-enabled types: clear their one-shot marker so a later re-disable
+    # re-triggers the backstop.
+    for doc_type in _TEXT_BACKSTOP_DOC_TYPES:
+        if doc_type in allowed:
+            _consent_backstop_done.pop((user_id, doc_type), None)
+
+    # Disabled types not yet backstopped this episode.
+    disabled = [
+        dt
+        for dt in _TEXT_BACKSTOP_DOC_TYPES
+        if dt not in allowed and (user_id, dt) not in _consent_backstop_done
+    ]
+    if not disabled:
+        return 0
+
+    qdrant_client = await get_qdrant_client()
+    collection = get_settings().get_collection_name()
+    queued = 0
+    for doc_type in disabled:
+        queued += await _backstop_delete_doc_type(
+            user_id, send_stream, doc_type, qdrant_client, collection, scan_id
+        )
+        # Mark backstopped (even when nothing was found) so subsequent scans
+        # don't re-scroll/re-enqueue for this disable episode.
+        _mark_backstop_done((user_id, doc_type))
+    return queued
+
+
 async def scan_user_documents(
     user_id: str,
     send_stream: TaskProducer,
@@ -365,6 +510,13 @@ async def scan_user_documents(
         # detection failed: fall back to scanning every app (prior behaviour).
         enabled_apps = await _get_enabled_apps_or_none(nc_client, user_id, scan_id)
 
+        # Admin consent gate (Astrolabe): only index sources the admin has
+        # approved for semantic search. ``None`` = no restriction (fail-open /
+        # older Astrolabe), so a transient capabilities failure never silently
+        # halts (or worse, mass-deletes) indexing. This is independent of
+        # ``enabled_apps``, which reflects only what the user has installed.
+        allowed = await allowed_doc_types(nc_client, user_id)
+
         # Notes (isolated so an uninstalled or disabled Notes app — whose API
         # returns 404 — cannot abort scanning of the other apps; this mirrors the
         # per-app try/except guards already wrapping files/news/deck below).
@@ -373,7 +525,17 @@ async def scan_user_documents(
         current_time = time.time()
         queued = 0
 
-        if _app_enabled("notes", enabled_apps):
+        # Backstop purge for admin-disabled text sources. Their deletion-
+        # tracking lives inside the scan_* function we skip below, so (unlike
+        # files, whose discovery-empties-then-reconcile path purges on disable)
+        # they'd linger if Astrolabe's eager purge failed. Enqueue deletes for
+        # any indexed points of a now-disallowed type. Gated on a concrete
+        # allow-set, so a fail-open None never triggers deletion.
+        queued += await _enqueue_deletes_for_disabled_types(
+            user_id, send_stream, allowed, scan_id
+        )
+
+        if _should_scan("notes", "note", enabled_apps, allowed):
             try:
                 queued += await scan_notes(
                     user_id=user_id,
@@ -454,9 +616,24 @@ async def scan_user_documents(
             # folder applies to every PDF beneath it.
             settings = get_settings()
             tag_name = settings.vector_sync_pdf_tag
-            tagged_files = await nc_client.find_files_by_tag(
-                tag_name, mime_type_filter="application/pdf"
-            )
+            if is_doc_type_allowed("file", allowed):
+                tagged_files = await nc_client.find_files_by_tag(
+                    tag_name, mime_type_filter="application/pdf"
+                )
+            else:
+                # Files disabled by admin: discover nothing so no new file is
+                # indexed. The deletion-reconcile below then sees every indexed
+                # file as "missing" and purges it after the grace period — the
+                # backstop for the eager purge Astrolabe runs on disable.
+                # Asymmetry (intentional): files purge up to 1.5x scan_interval
+                # later than text types, which get immediate one-shot backstop
+                # deletes via _enqueue_deletes_for_disabled_types.
+                logger.debug(
+                    "[SCAN-%s] Files disabled by admin for %s; skipping tagged-file discovery",
+                    scan_id,
+                    user_id,
+                )
+                tagged_files = []
 
             # Apply EXCLUDED_TAGS as defense-in-depth: a folder marked
             # off-limits via the exclusion tag must not be indexed even if
@@ -710,7 +887,7 @@ async def scan_user_documents(
 
         # Scan News items (starred + unread)
         news_queued = 0
-        if _app_enabled("news", enabled_apps):
+        if _should_scan("news", "news_item", enabled_apps, allowed):
             try:
                 news_queued = await scan_news_items(
                     user_id=user_id,
@@ -731,7 +908,7 @@ async def scan_user_documents(
 
         # Scan Deck cards
         deck_queued = 0
-        if _app_enabled("deck", enabled_apps):
+        if _should_scan("deck", "deck_card", enabled_apps, allowed):
             try:
                 deck_queued = await scan_deck_cards(
                     user_id=user_id,
diff --git a/tests/unit/server/test_semantic_consent_narrowing.py b/tests/unit/server/test_semantic_consent_narrowing.py
new file mode 100644
index 00000000..6b4ee3cb
--- /dev/null
+++ b/tests/unit/server/test_semantic_consent_narrowing.py
@@ -0,0 +1,41 @@
+"""Unit tests for the search-side consent narrowing in nc_semantic_search.
+
+The narrowing logic (intersect requested doc_types with the admin allow-set,
+or restrict to the allow-set when none requested) is extracted into
+``_consent_narrowed_doc_types`` so it can be tested without exercising the full
+search path. ``allowed is None`` (no restriction / fail-open) is handled by the
+caller skipping this helper entirely.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from nextcloud_mcp_server.server.semantic import _consent_narrowed_doc_types
+
+pytestmark = pytest.mark.unit
+
+
+def test_none_request_restricts_to_allow_set():
+    # No explicit doc_types -> search exactly the allowed set (sorted).
+    assert _consent_narrowed_doc_types(None, frozenset({"file", "note"})) == [
+        "file",
+        "note",
+    ]
+
+
+def test_request_intersected_with_allow_set_preserves_order():
+    result = _consent_narrowed_doc_types(
+        ["deck_card", "note", "file"], frozenset({"note", "file"})
+    )
+    assert result == ["note", "file"]
+
+
+def test_disjoint_request_yields_empty():
+    # Caller short-circuits to an empty response on [].
+    assert _consent_narrowed_doc_types(["deck_card"], frozenset({"note"})) == []
+
+
+def test_empty_allow_set_blocks_all():
+    assert _consent_narrowed_doc_types(None, frozenset()) == []
+    assert _consent_narrowed_doc_types(["note"], frozenset()) == []
diff --git a/tests/unit/test_capabilities.py b/tests/unit/test_capabilities.py
new file mode 100644
index 00000000..1bbbec64
--- /dev/null
+++ b/tests/unit/test_capabilities.py
@@ -0,0 +1,162 @@
+"""Unit tests for the Astrolabe searchable-sources capability reader."""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock
+
+import pytest
+
+import nextcloud_mcp_server.capabilities as cap
+from nextcloud_mcp_server.capabilities import (
+    _parse_enabled_doc_types,
+    allowed_doc_types,
+    clear_cache,
+    is_doc_type_allowed,
+)
+
+pytestmark = pytest.mark.unit
+
+
+def _payload(enabled_doc_types) -> dict:
+    """Build an OCS capabilities envelope carrying the astrolabe block.
+
+    ``enabled_doc_types=...`` (Ellipsis) omits the key entirely.
+    """
+    semantic: dict = {}
+    if enabled_doc_types is not ...:
+        semantic["enabled_doc_types"] = enabled_doc_types
+    return {
+        "ocs": {
+            "meta": {"status": "ok"},
+            "data": {"capabilities": {"astrolabe": {"semantic_search": semantic}}},
+        }
+    }
+
+
+# ---------------------------------------------------------------------------
+# _parse_enabled_doc_types
+# ---------------------------------------------------------------------------
+
+
+def test_parse_present_list_returns_set():
+    assert _parse_enabled_doc_types(_payload(["note", "file"])) == {"note", "file"}
+
+
+def test_parse_empty_list_returns_empty_set():
+    # Admin disabled every source — distinct from "no restriction".
+    assert _parse_enabled_doc_types(_payload([])) == set()
+
+
+def test_parse_missing_astrolabe_block_returns_none():
+    payload = {"ocs": {"data": {"capabilities": {}}}}
+    assert _parse_enabled_doc_types(payload) is None
+
+
+def test_parse_missing_enabled_key_returns_none():
+    assert _parse_enabled_doc_types(_payload(...)) is None
+
+
+def test_parse_malformed_payload_returns_none():
+    assert _parse_enabled_doc_types(None) is None
+    assert _parse_enabled_doc_types({"ocs": "nope"}) is None
+    assert _parse_enabled_doc_types(_payload("not-a-list")) is None
+
+
+def test_parse_drops_non_string_entries():
+    assert _parse_enabled_doc_types(_payload(["note", 5, None])) == {"note"}
+
+
+# ---------------------------------------------------------------------------
+# is_doc_type_allowed
+# ---------------------------------------------------------------------------
+
+
+def test_is_doc_type_allowed_none_means_no_restriction():
+    assert is_doc_type_allowed("anything", None) is True
+
+
+def test_is_doc_type_allowed_respects_set():
+    allowed = frozenset({"note"})
+    assert is_doc_type_allowed("note", allowed) is True
+    assert is_doc_type_allowed("file", allowed) is False
+
+
+def test_is_doc_type_allowed_empty_set_blocks_all():
+    assert is_doc_type_allowed("note", frozenset()) is False
+
+
+# ---------------------------------------------------------------------------
+# allowed_doc_types (cache + fail-open)
+# ---------------------------------------------------------------------------
+
+
+def _client(payload=None, raises: Exception | None = None) -> AsyncMock:
+    """An object with an async ``capabilities()`` method (AsyncMock-backed)."""
+    m = AsyncMock()
+    if raises is not None:
+        m.capabilities.side_effect = raises
+    else:
+        m.capabilities.return_value = payload
+    return m
+
+
+async def test_allowed_doc_types_parses_and_caches():
+    clear_cache()
+    client = _client(_payload(["note", "file"]))
+
+    first = await allowed_doc_types(client, "alice")
+    second = await allowed_doc_types(client, "alice")
+
+    assert first == frozenset({"note", "file"})
+    assert second == frozenset({"note", "file"})
+    # Second call served from the cache — only one OCS round-trip.
+    assert client.capabilities.await_count == 1
+
+
+async def test_allowed_doc_types_missing_block_returns_none():
+    clear_cache()
+    client = _client({"ocs": {"data": {"capabilities": {}}}})
+    assert await allowed_doc_types(client, "bob") is None
+
+
+async def test_allowed_doc_types_fail_open_not_cached():
+    clear_cache()
+    client = _client(raises=RuntimeError("ocs down"))
+
+    assert await allowed_doc_types(client, "carol") is None
+    # Failures are not cached — the next call retries the OCS lookup.
+    assert await allowed_doc_types(client, "carol") is None
+    assert client.capabilities.await_count == 2
+
+
+async def test_allowed_doc_types_cache_is_per_user():
+    clear_cache()
+    alice = _client(_payload(["note"]))
+    bob = _client(_payload(["file"]))
+
+    assert await allowed_doc_types(alice, "alice") == frozenset({"note"})
+    assert await allowed_doc_types(bob, "bob") == frozenset({"file"})
+
+
+async def test_allowed_doc_types_refetches_after_ttl(monkeypatch):
+    clear_cache()
+    client = _client(_payload(["note"]))
+
+    # Drive the module clock so the second call lands past the TTL window.
+    clock = {"now": 1000.0}
+    monkeypatch.setattr(cap.time, "monotonic", lambda: clock["now"])
+
+    await allowed_doc_types(client, "erin")
+    clock["now"] += cap._CACHE_TTL_SECONDS + 1
+    await allowed_doc_types(client, "erin")
+
+    assert client.capabilities.await_count == 2
+
+
+async def test_clear_cache_forces_refetch():
+    clear_cache()
+    client = _client(_payload(["note"]))
+    await allowed_doc_types(client, "dave")
+    cap.clear_cache()
+    await allowed_doc_types(client, "dave")
+    assert client.capabilities.await_count == 2
diff --git a/tests/unit/test_processor_drop_reason.py b/tests/unit/test_processor_drop_reason.py
index ee1e05c9..45a0cedf 100644
--- a/tests/unit/test_processor_drop_reason.py
+++ b/tests/unit/test_processor_drop_reason.py
@@ -72,7 +72,7 @@ def test_nested_exception_group_descends_to_leaf():
 def test_qdrant_namespace_classified():
     from qdrant_client.http.exceptions import UnexpectedResponse
 
-    exc = UnexpectedResponse(500, "err", b"", headers=None)
+    exc = UnexpectedResponse(500, "err", b"", headers=httpx.Headers())
     assert processor._drop_reason(exc) == "qdrant"
 
 
@@ -109,3 +109,71 @@ async def test_process_document_records_drop_on_exhausted_retries(mocker):
         await processor.process_document(doc_task, mocker.MagicMock(), max_retries=1)
 
     rec.assert_called_once_with("connection")
+
+
+@pytest.mark.unit
+async def test_process_document_drops_admin_disabled_index_task(mocker):
+    """A near-real-time index task for an admin-disabled doc_type is dropped
+    before indexing, and recorded under the ``admin_disabled`` reason."""
+    from nextcloud_mcp_server.vector.scanner import DocumentTask
+
+    doc_task = DocumentTask(
+        user_id="alice",
+        doc_id="42",
+        doc_type="note",
+        operation="index",
+        modified_at=0,
+        file_path="/x.md",  # set so the tag-reconcile branch is skipped
+    )
+
+    mocker.patch.object(
+        processor,
+        "get_qdrant_client",
+        mocker.AsyncMock(return_value=mocker.MagicMock()),
+    )
+    # Admin disabled everything → note is not allowed.
+    mocker.patch.object(
+        processor, "allowed_doc_types", mocker.AsyncMock(return_value=frozenset())
+    )
+    index = mocker.patch.object(processor, "_index_document")
+    rec = mocker.patch.object(processor, "record_ingest_dropped")
+
+    await processor.process_document(doc_task, mocker.MagicMock(), max_retries=1)
+
+    index.assert_not_called()
+    rec.assert_called_once_with("admin_disabled")
+
+
+@pytest.mark.unit
+async def test_process_document_allows_when_doc_type_approved(mocker):
+    """The consent gate does not drop an index task for an allowed doc_type."""
+    from nextcloud_mcp_server.vector.scanner import DocumentTask
+
+    doc_task = DocumentTask(
+        user_id="alice",
+        doc_id="42",
+        doc_type="note",
+        operation="index",
+        modified_at=0,
+        file_path="/x.md",
+    )
+
+    mocker.patch.object(
+        processor,
+        "get_qdrant_client",
+        mocker.AsyncMock(return_value=mocker.MagicMock()),
+    )
+    mocker.patch.object(
+        processor,
+        "allowed_doc_types",
+        mocker.AsyncMock(return_value=frozenset({"note"})),
+    )
+    index = mocker.patch.object(
+        processor, "_index_document", mocker.AsyncMock(return_value=1)
+    )
+    rec = mocker.patch.object(processor, "record_ingest_dropped")
+
+    await processor.process_document(doc_task, mocker.MagicMock(), max_retries=1)
+
+    index.assert_awaited()  # indexing proceeded
+    rec.assert_not_called()
diff --git a/tests/unit/test_vector_sync_purge_route.py b/tests/unit/test_vector_sync_purge_route.py
new file mode 100644
index 00000000..37d5cdbb
--- /dev/null
+++ b/tests/unit/test_vector_sync_purge_route.py
@@ -0,0 +1,258 @@
+"""Unit tests for the /api/v1/vector-sync/purge admin route.
+
+The purge is global and destructive (deletes every owner's content for a doc
+type), so the route must: authenticate the bearer, restrict to Nextcloud
+admins, validate the body, and only then delegate to the global purge.
+"""
+
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from starlette.applications import Starlette
+from starlette.routing import Route
+from starlette.testclient import TestClient
+
+from nextcloud_mcp_server.api.vector_sync import purge_doc_types_route
+from nextcloud_mcp_server.auth.scope_authorization import ProvisioningRequiredError
+
+pytestmark = pytest.mark.unit
+
+
+def _build_app() -> Starlette:
+    app = Starlette(
+        routes=[
+            Route(
+                "/api/v1/vector-sync/purge",
+                purge_doc_types_route,
+                methods=["POST"],
+            )
+        ]
+    )
+    app.state.oauth_context = {"config": {"nextcloud_host": "http://nc.test"}}
+    return app
+
+
+def _patch_token(mocker, user_id="admin"):
+    mocker.patch(
+        "nextcloud_mcp_server.api.vector_sync.validate_token_and_get_user",
+        new=AsyncMock(return_value=(user_id, {"sub": user_id})),
+    )
+
+
+def _patch_basic_auth(mocker, username="admin"):
+    mocker.patch(
+        "nextcloud_mcp_server.api.vector_sync.get_basic_auth_for_user",
+        new=AsyncMock(return_value=(username, "app-pwd")),
+    )
+
+
+def _patch_outbound_client(mocker):
+    client = AsyncMock()
+    client.__aenter__ = AsyncMock(return_value=client)
+    client.__aexit__ = AsyncMock(return_value=False)
+    mocker.patch(
+        "nextcloud_mcp_server.api.vector_sync.nextcloud_httpx_client",
+        MagicMock(return_value=client),
+    )
+    return client
+
+
+def _patch_groups(mocker, groups):
+    instance = MagicMock()
+    instance.get_user_groups = AsyncMock(return_value=groups)
+    mocker.patch(
+        "nextcloud_mcp_server.api.vector_sync.UsersClient",
+        MagicMock(return_value=instance),
+    )
+
+
+def _patch_purge(mocker, result=None):
+    return mocker.patch(
+        "nextcloud_mcp_server.api.vector_sync.purge_doc_types",
+        new=AsyncMock(return_value=result or {}),
+    )
+
+
+def test_unauthorized_when_token_invalid(mocker):
+    mocker.patch(
+        "nextcloud_mcp_server.api.vector_sync.validate_token_and_get_user",
+        new=AsyncMock(side_effect=ValueError("bad token")),
+    )
+    purge = _patch_purge(mocker)
+
+    client = TestClient(_build_app())
+    resp = client.post("/api/v1/vector-sync/purge", json={"doc_types": ["file"]})
+
+    assert resp.status_code == 401
+    purge.assert_not_called()
+
+
+def test_bad_request_when_doc_types_not_list(mocker):
+    _patch_token(mocker)
+    purge = _patch_purge(mocker)
+
+    client = TestClient(_build_app())
+    resp = client.post("/api/v1/vector-sync/purge", json={"doc_types": "file"})
+
+    assert resp.status_code == 400
+    purge.assert_not_called()
+
+
+def test_bad_request_when_doc_types_has_non_string(mocker):
+    # Covers the all(isinstance(d, str)) branch (a list with non-string items).
+    _patch_token(mocker)
+    purge = _patch_purge(mocker)
+
+    client = TestClient(_build_app())
+    resp = client.post("/api/v1/vector-sync/purge", json={"doc_types": [1, 2]})
+
+    assert resp.status_code == 400
+    purge.assert_not_called()
+
+
+def test_total_failure_returns_500(mocker):
+    # purge_doc_types raising (total failure) hits the route's except -> 500.
+    _patch_token(mocker, "admin")
+    _patch_basic_auth(mocker, "admin")
+    _patch_outbound_client(mocker)
+    _patch_groups(mocker, ["admin"])
+    mocker.patch(
+        "nextcloud_mcp_server.api.vector_sync.purge_doc_types",
+        new=AsyncMock(side_effect=RuntimeError("qdrant down")),
+    )
+
+    client = TestClient(_build_app())
+    resp = client.post("/api/v1/vector-sync/purge", json={"doc_types": ["file"]})
+
+    assert resp.status_code == 500
+
+
+def test_forbidden_when_not_admin(mocker):
+    _patch_token(mocker, "bob")
+    _patch_basic_auth(mocker, "bob")
+    _patch_outbound_client(mocker)
+    _patch_groups(mocker, ["users"])  # not an admin
+    purge = _patch_purge(mocker)
+
+    client = TestClient(_build_app())
+    resp = client.post("/api/v1/vector-sync/purge", json={"doc_types": ["file"]})
+
+    assert resp.status_code == 403
+    purge.assert_not_called()
+
+
+def test_missing_doc_types_key_returns_400(mocker):
+    _patch_token(mocker)
+    purge = _patch_purge(mocker)
+
+    client = TestClient(_build_app())
+    resp = client.post("/api/v1/vector-sync/purge", json={})
+
+    assert resp.status_code == 400
+    purge.assert_not_called()
+
+
+def test_empty_doc_types_is_admin_gated_noop(mocker):
+    # An empty (no-op) request still requires admin — this is a destructive route.
+    _patch_token(mocker, "admin")
+    _patch_basic_auth(mocker, "admin")
+    _patch_outbound_client(mocker)
+    _patch_groups(mocker, ["admin"])
+    purge = _patch_purge(mocker)
+
+    client = TestClient(_build_app())
+    resp = client.post("/api/v1/vector-sync/purge", json={"doc_types": []})
+
+    assert resp.status_code == 200
+    assert resp.json() == {"purged": {}}
+    purge.assert_not_called()
+
+
+def test_empty_doc_types_forbidden_for_non_admin(mocker):
+    _patch_token(mocker, "bob")
+    _patch_basic_auth(mocker, "bob")
+    _patch_outbound_client(mocker)
+    _patch_groups(mocker, ["users"])
+    purge = _patch_purge(mocker)
+
+    client = TestClient(_build_app())
+    resp = client.post("/api/v1/vector-sync/purge", json={"doc_types": []})
+
+    assert resp.status_code == 403
+    purge.assert_not_called()
+
+
+def test_admin_purge_happy_path(mocker):
+    _patch_token(mocker, "admin")
+    _patch_basic_auth(mocker, "admin")
+    _patch_outbound_client(mocker)
+    _patch_groups(mocker, ["admin"])
+    purge = _patch_purge(mocker, {"file": 12})
+
+    client = TestClient(_build_app())
+    resp = client.post("/api/v1/vector-sync/purge", json={"doc_types": ["file"]})
+
+    assert resp.status_code == 200
+    assert resp.json() == {"purged": {"file": 12}}
+    purge.assert_awaited_once_with(["file"])
+
+
+def test_partial_failure_reports_failed_types(mocker):
+    # purge_doc_types returns only the succeeded types; the route must tell the
+    # caller which requested types were NOT purged.
+    _patch_token(mocker, "admin")
+    _patch_basic_auth(mocker, "admin")
+    _patch_outbound_client(mocker)
+    _patch_groups(mocker, ["admin"])
+    _patch_purge(mocker, {"file": 3})  # "note" failed
+
+    client = TestClient(_build_app())
+    resp = client.post(
+        "/api/v1/vector-sync/purge", json={"doc_types": ["file", "note"]}
+    )
+
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["purged"] == {"file": 3}
+    assert body["failed"] == ["note"]
+
+
+def test_bad_request_when_body_not_object(mocker):
+    # A valid JSON non-object (e.g. a list) must 400, not 500.
+    _patch_token(mocker)
+    purge = _patch_purge(mocker)
+
+    client = TestClient(_build_app())
+    resp = client.post("/api/v1/vector-sync/purge", json=[1, 2, 3])
+
+    assert resp.status_code == 400
+    purge.assert_not_called()
+
+
+def test_bad_request_when_too_many_doc_types(mocker):
+    _patch_token(mocker)
+    purge = _patch_purge(mocker)
+
+    client = TestClient(_build_app())
+    resp = client.post(
+        "/api/v1/vector-sync/purge",
+        json={"doc_types": [f"t{i}" for i in range(65)]},
+    )
+
+    assert resp.status_code == 400
+    purge.assert_not_called()
+
+
+def test_provisioning_required_returns_428(mocker):
+    _patch_token(mocker, "admin")
+    mocker.patch(
+        "nextcloud_mcp_server.api.vector_sync.get_basic_auth_for_user",
+        new=AsyncMock(side_effect=ProvisioningRequiredError("not provisioned")),
+    )
+    purge = _patch_purge(mocker)
+
+    client = TestClient(_build_app())
+    resp = client.post("/api/v1/vector-sync/purge", json={"doc_types": ["file"]})
+
+    assert resp.status_code == 428
+    purge.assert_not_called()
diff --git a/tests/unit/vector/test_purge.py b/tests/unit/vector/test_purge.py
new file mode 100644
index 00000000..abd3c039
--- /dev/null
+++ b/tests/unit/vector/test_purge.py
@@ -0,0 +1,99 @@
+"""Unit tests for global purge-by-doc-type (admin consent enforcement)."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+
+import nextcloud_mcp_server.vector.purge as purge_module
+from nextcloud_mcp_server.vector.purge import purge_doc_types
+
+pytestmark = pytest.mark.unit
+
+
+def _patch_qdrant(monkeypatch, *, counts: dict[str, int], delete_raises=None):
+    """Wire a fake Qdrant client whose ``count`` reflects ``counts`` per
+    doc_type (read off the filter's MatchValue) and whose ``delete`` optionally
+    raises for given doc_types."""
+    client = AsyncMock()
+
+    def _doc_type_of(flt):
+        return flt.must[0].match.value
+
+    # Sync side_effects: AsyncMock awaits the call and returns the value, so the
+    # helpers don't need to be coroutines themselves.
+    def fake_count(*, collection_name, count_filter, exact):
+        return SimpleNamespace(count=counts.get(_doc_type_of(count_filter), 0))
+
+    def fake_delete(*, collection_name, points_selector):
+        dt = _doc_type_of(points_selector)
+        if delete_raises and dt in delete_raises:
+            raise RuntimeError(f"delete failed for {dt}")
+
+    client.count.side_effect = fake_count
+    client.delete.side_effect = fake_delete
+
+    monkeypatch.setattr(
+        purge_module, "get_qdrant_client", AsyncMock(return_value=client)
+    )
+    monkeypatch.setattr(
+        purge_module,
+        "get_settings",
+        lambda: SimpleNamespace(get_collection_name=lambda: "test_collection"),
+    )
+    return client
+
+
+async def test_purges_each_doc_type_and_reports_counts(monkeypatch):
+    client = _patch_qdrant(monkeypatch, counts={"file": 7, "note": 3})
+
+    result = await purge_doc_types(["file", "note"])
+
+    assert result == {"file": 7, "note": 3}
+    assert client.delete.await_count == 2
+
+
+async def test_purge_is_owner_agnostic_global(monkeypatch):
+    # The admin disable is global, so the delete filter must match by doc_type
+    # ONLY — no owner_id/user_id condition that would scope it to one user.
+    client = _patch_qdrant(monkeypatch, counts={"file": 1})
+
+    await purge_doc_types(["file"])
+
+    flt = client.delete.await_args.kwargs["points_selector"]
+    keys = [c.key for c in flt.must]
+    assert keys == ["doc_type"]
+    assert flt.must[0].match.value == "file"
+
+
+async def test_dedupes_doc_types(monkeypatch):
+    client = _patch_qdrant(monkeypatch, counts={"file": 2})
+
+    result = await purge_doc_types(["file", "file"])
+
+    assert result == {"file": 2}
+    assert client.delete.await_count == 1
+
+
+async def test_zero_points_is_safe(monkeypatch):
+    _patch_qdrant(monkeypatch, counts={})
+    assert await purge_doc_types(["deck_card"]) == {"deck_card": 0}
+
+
+async def test_partial_failure_returns_partial(monkeypatch):
+    _patch_qdrant(
+        monkeypatch,
+        counts={"file": 5, "note": 4},
+        delete_raises={"note"},
+    )
+    # "note" delete fails, "file" succeeds — partial progress is returned.
+    result = await purge_doc_types(["file", "note"])
+    assert result == {"file": 5}
+
+
+async def test_total_failure_raises(monkeypatch):
+    _patch_qdrant(monkeypatch, counts={"file": 5}, delete_raises={"file"})
+    with pytest.raises(RuntimeError):
+        await purge_doc_types(["file"])
diff --git a/tests/unit/vector/test_scanner_consent_backstop.py b/tests/unit/vector/test_scanner_consent_backstop.py
new file mode 100644
index 00000000..e87ebed9
--- /dev/null
+++ b/tests/unit/vector/test_scanner_consent_backstop.py
@@ -0,0 +1,151 @@
+"""Unit tests for the scanner's admin-consent backstop deletion.
+
+When an admin disables a text source (note/news_item/deck_card), the scanner
+skips its scan_* function, so the in-function deletion-tracking never runs. The
+backstop enqueues deletes for any indexed points of the disabled type, mirroring
+the files path — but only on a concrete allow-set (never on fail-open None).
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from typing import cast
+from unittest.mock import AsyncMock
+
+import pytest
+
+from nextcloud_mcp_server.vector import scanner as scanner_module
+from nextcloud_mcp_server.vector.queue.ports import TaskProducer
+from nextcloud_mcp_server.vector.scanner import _enqueue_deletes_for_disabled_types
+
+pytestmark = pytest.mark.unit
+
+
+@pytest.fixture(autouse=True)
+def _clear_backstop_state():
+    """The one-shot guard is module-level; reset it between tests."""
+    scanner_module._consent_backstop_done.clear()
+    yield
+    scanner_module._consent_backstop_done.clear()
+
+
+def _producer(send: AsyncMock) -> TaskProducer:
+    """A minimal stand-in for the TaskProducer protocol (only ``send`` is used)."""
+    return cast(TaskProducer, SimpleNamespace(send=send))
+
+
+def _patch_qdrant(monkeypatch, points_by_type: dict[str, list[str]]):
+    client = AsyncMock()
+
+    def fake_scroll(
+        *, collection_name, scroll_filter, with_payload, with_vectors, limit, offset
+    ):
+        # must=[user_id, doc_type] — doc_type is the second condition.
+        doc_type = scroll_filter.must[1].match.value
+        points = [
+            SimpleNamespace(payload={"doc_id": doc_id})
+            for doc_id in points_by_type.get(doc_type, [])
+        ]
+        return (points, None)
+
+    client.scroll.side_effect = fake_scroll
+    monkeypatch.setattr(
+        scanner_module, "get_qdrant_client", AsyncMock(return_value=client)
+    )
+    monkeypatch.setattr(
+        scanner_module,
+        "get_settings",
+        lambda: SimpleNamespace(get_collection_name=lambda: "c"),
+    )
+
+
+async def test_enqueues_deletes_for_disabled_text_type(monkeypatch):
+    _patch_qdrant(monkeypatch, {"note": ["n1", "n2"], "deck_card": ["d1"]})
+    sent: list = []
+    stream = _producer(AsyncMock(side_effect=lambda t: sent.append(t)))
+
+    # note disabled; news_item + deck_card still allowed.
+    allowed = frozenset({"file", "news_item", "deck_card"})
+    queued = await _enqueue_deletes_for_disabled_types("alice", stream, allowed, 1)
+
+    assert queued == 2
+    assert {t.doc_id for t in sent} == {"n1", "n2"}
+    assert all(t.operation == "delete" and t.doc_type == "note" for t in sent)
+
+
+async def test_all_text_types_disabled_enqueues_all(monkeypatch):
+    # Admin disabled everything at once (empty allow-set): every text type's
+    # indexed points are enqueued for deletion in a single call.
+    _patch_qdrant(
+        monkeypatch, {"note": ["n1"], "news_item": ["ni1"], "deck_card": ["d1"]}
+    )
+    sent: list = []
+    stream = _producer(AsyncMock(side_effect=lambda t: sent.append(t)))
+
+    queued = await _enqueue_deletes_for_disabled_types("alice", stream, frozenset(), 1)
+
+    assert queued == 3
+    assert {(t.doc_type, t.doc_id) for t in sent} == {
+        ("note", "n1"),
+        ("news_item", "ni1"),
+        ("deck_card", "d1"),
+    }
+
+
+async def test_noop_when_allowed_is_none(monkeypatch):
+    # Fail-open: a transient capability read must never trigger deletion.
+    send = AsyncMock()
+    queued = await _enqueue_deletes_for_disabled_types(
+        "alice", _producer(send), None, 1
+    )
+    assert queued == 0
+    send.assert_not_called()
+
+
+async def test_noop_when_all_text_types_allowed(monkeypatch):
+    send = AsyncMock()
+    allowed = frozenset({"note", "news_item", "deck_card", "file"})
+    queued = await _enqueue_deletes_for_disabled_types(
+        "alice", _producer(send), allowed, 1
+    )
+    assert queued == 0
+    send.assert_not_called()
+
+
+async def test_one_shot_does_not_reflood_on_subsequent_scans(monkeypatch):
+    _patch_qdrant(monkeypatch, {"note": ["n1", "n2"]})
+    send = AsyncMock()
+    allowed = frozenset({"file"})  # note disabled
+
+    first = await _enqueue_deletes_for_disabled_types(
+        "alice", _producer(send), allowed, 1
+    )
+    second = await _enqueue_deletes_for_disabled_types(
+        "alice", _producer(send), allowed, 2
+    )
+
+    assert first == 2
+    # Standing disable: the next scan must not re-enqueue the same deletes.
+    assert second == 0
+
+
+async def test_re_enable_then_disable_retriggers_backstop(monkeypatch):
+    _patch_qdrant(monkeypatch, {"note": ["n1"]})
+    send = AsyncMock()
+    disabled = frozenset({"file"})
+    enabled = frozenset({"file", "note"})
+
+    assert (
+        await _enqueue_deletes_for_disabled_types("alice", _producer(send), disabled, 1)
+        == 1
+    )
+    # Re-enabled: clears the one-shot marker.
+    assert (
+        await _enqueue_deletes_for_disabled_types("alice", _producer(send), enabled, 2)
+        == 0
+    )
+    # Disabled again: backstop fires once more.
+    assert (
+        await _enqueue_deletes_for_disabled_types("alice", _producer(send), disabled, 3)
+        == 1
+    )