diff --git a/nextcloud_mcp_server/acl_hash.py b/nextcloud_mcp_server/acl_hash.py new file mode 100644 index 00000000..761b78ba --- /dev/null +++ b/nextcloud_mcp_server/acl_hash.py @@ -0,0 +1,66 @@ +"""Canonical ACL hash (design §11) — a cross-implementation contract. + +The external document-processor and the local processor inside this server both +write the ``acl_hash`` Qdrant payload independently; the query path +(``search/verification.py``) builds the matching accessible set. All three must +agree byte-for-byte, so the canonicalization, hash, and accessible-set rules are +pinned here and exercised by an identical ``tests/fixtures/acl_hash_corpus.json`` +in both repos (§11.5). + +Key invariants: +- ``principal_id`` is NFC-normalized; **case is preserved, not folded** (§11.2). +- ``BLAKE2b-128`` (``digest_size=16`` → 32 hex chars), NOT the 64-byte default + (§11.3). +- Public-link shares are excluded by the caller — they are bearer secrets not + bound to an identity and never reachable via authenticated MCP queries (§11.1). +""" + +from __future__ import annotations + +import hashlib +import unicodedata +from collections.abc import Iterable + +from nextcloud_mcp_server.canonical import canonical_json + +PRINCIPAL_TYPES = frozenset({"user", "group", "public"}) + +# The world-readable principal, included in every requester's accessible set +# (§11.4) and written on any document Nextcloud marks world-readable. +PUBLIC_PRINCIPAL: tuple[str, str] = ("public", "public") + + +def compute_principal_hash(principal_type: str, principal_id: str) -> str: + """Hash one ``(principal_type, principal_id)`` tuple (§11.2–11.3).""" + if principal_type not in PRINCIPAL_TYPES: + raise ValueError( + f"principal_type must be one of {sorted(PRINCIPAL_TYPES)}; " + f"got {principal_type!r}" + ) + # principal_type is ASCII by construction; only the id is normalized. + normalized_id = unicodedata.normalize("NFC", principal_id) + canonical = canonical_json([principal_type, normalized_id]) + return hashlib.blake2b(canonical, digest_size=16).hexdigest() + + +def compute_acl_hash(share_set: Iterable[tuple[str, str]]) -> list[str]: + """Per-principal hash array for a document's share-set (§11.2). + + One element per share-set entry, in input order (Qdrant treats the array + with set-semantics, so order is irrelevant at query time). The caller must + have already dropped public-link shares (§11.1). + """ + return [compute_principal_hash(ptype, pid) for ptype, pid in share_set] + + +def accessible_hash_set(username: str, groups: Iterable[str] = ()) -> set[str]: + """Accessible hash set for an authenticated requester (§11.4). + + Derived entirely from OIDC claims: the requester's own ``(user, username)``, + each ``(group, )`` they hold, and unconditionally ``(public, public)``. + """ + hashes = {compute_principal_hash("user", username)} + for group in groups: + hashes.add(compute_principal_hash("group", group)) + hashes.add(compute_principal_hash(*PUBLIC_PRINCIPAL)) + return hashes diff --git a/nextcloud_mcp_server/admin/__init__.py b/nextcloud_mcp_server/admin/__init__.py new file mode 100644 index 00000000..dc267668 --- /dev/null +++ b/nextcloud_mcp_server/admin/__init__.py @@ -0,0 +1 @@ +"""Admin-only REST endpoints (``/api/v1/admin/*``).""" diff --git a/nextcloud_mcp_server/admin/payload_backfill.py b/nextcloud_mcp_server/admin/payload_backfill.py new file mode 100644 index 00000000..f851ee46 --- /dev/null +++ b/nextcloud_mcp_server/admin/payload_backfill.py @@ -0,0 +1,104 @@ +"""One-shot payload backfill admin endpoint (design §10.2). + +``POST /api/v1/admin/payload-backfill`` walks the collection and adds default +values for any *missing* decomposition payload keys (so existing corpora gain +them without a re-index), then upserts the collection-metadata sentinel. + +Scope: this backfills the cheap, deployment-level scalar keys +(``processor_version``, ``parsed_at``, ``pipeline_tier``, ``embedding_identity``) +only. It deliberately does NOT synthesize ``acl_hash`` — a correct value needs +per-document share enumeration (a separate job), and writing a placeholder +``acl_hash`` would be unsafe to pre-filter on. The query-side ACL pre-filter +therefore stays disabled until a real ACL backfill runs (see +``ACL_PREFILTER_ENABLED``). +""" + +from __future__ import annotations + +import logging + +from qdrant_client import models +from starlette.requests import Request +from starlette.responses import JSONResponse + +from nextcloud_mcp_server.api.management import ( + AdminScopeRequired, + require_admin_scope, +) +from nextcloud_mcp_server.config import get_settings +from nextcloud_mcp_server.embedding import get_embedding_service +from nextcloud_mcp_server.vector import payload_keys +from nextcloud_mcp_server.vector.collection_metadata import ( + env_default_metadata, + upsert_sentinel, +) +from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client + +logger = logging.getLogger(__name__) + + +async def handle_payload_backfill(request: Request) -> JSONResponse: + try: + await require_admin_scope(request) + except AdminScopeRequired: + return JSONResponse({"error": "admin scope required"}, status_code=403) + except Exception: + return JSONResponse({"error": "unauthorized"}, status_code=401) + + settings = get_settings() + if not settings.vector_sync_enabled: + return JSONResponse({"error": "vector sync disabled"}, status_code=404) + + client = await get_qdrant_client() + collection = settings.get_collection_name() + meta = env_default_metadata(settings) + + # Deployment-level scalar defaults (safe to set only where missing). + defaults = { + payload_keys.PROCESSOR_VERSION: "backfill", + payload_keys.PIPELINE_TIER: "fast", + payload_keys.EMBEDDING_IDENTITY: meta["embedding_identity"], + } + applied: dict[str, str] = {} + for key, value in defaults.items(): + try: + await client.set_payload( + collection_name=collection, + payload={key: value}, + # Only points missing this key (don't clobber existing values). + points=models.Filter( + must=[ + models.IsEmptyCondition(is_empty=models.PayloadField(key=key)) + ] + ), + wait=True, + ) + applied[key] = "set-where-missing" + except Exception as e: + logger.warning("payload backfill failed for key %s: %s", key, e) + applied[key] = f"error: {e}" + + # Upsert the collection-metadata sentinel so query-path metadata reads work + # for this collection even without a control plane. + sentinel_ok = True + try: + dimension = get_embedding_service().get_dimension() + await upsert_sentinel( + client, + collection, + embedding_identity=meta["embedding_identity"], + chunking_config=meta["chunking_config"], + dimension=dimension, + ) + except Exception as e: + sentinel_ok = False + logger.warning("sentinel upsert failed during backfill: %s", e) + + return JSONResponse( + { + "status": "ok", + "collection": collection, + "keys_applied": applied, + "sentinel_upserted": sentinel_ok, + } + ) diff --git a/nextcloud_mcp_server/api/management.py b/nextcloud_mcp_server/api/management.py index 12283408..9dde161f 100644 --- a/nextcloud_mcp_server/api/management.py +++ b/nextcloud_mcp_server/api/management.py @@ -122,6 +122,25 @@ async def validate_token_and_get_user( return user_id, validated +class AdminScopeRequired(Exception): + """Raised when an authenticated caller lacks the ``admin`` scope.""" + + +async def require_admin_scope(request: Request) -> str: + """Authenticate the caller and require the ``admin`` scope. + + This is the first ``/api/v1/admin/*`` guard; it establishes the pattern + (auth via :func:`validate_token_and_get_user`, then an explicit scope check). + Raises ``ValueError`` on auth failure and :class:`AdminScopeRequired` on a + valid token that lacks ``admin``; callers map these to 401 / 403. + """ + user_id, validated = await validate_token_and_get_user(request) + scopes = validated.get("scopes") or [] + if "admin" not in scopes: + raise AdminScopeRequired("admin scope required") + return user_id + + def _sanitize_error_for_client(error: Exception, context: str = "") -> str: """ Return a safe, generic error message for clients. @@ -274,6 +293,31 @@ async def get_vector_sync_status(request: Request) -> JSONResponse: ) try: + # Bus status backend (INGEST_MODE=external): there is no in-process + # queue; pending/terminal state comes from the NATS status subscriber's + # store. indexed_documents stays the mode-independent Qdrant count. + if settings.status_backend == "bus": + store = getattr(request.app.state, "status_store", None) + indexed_count = 0 + try: + qdrant_client = await get_qdrant_client() + count_result = await qdrant_client.count( + collection_name=settings.get_collection_name(), + count_filter=Filter(must=[get_placeholder_filter()]), + ) + indexed_count = count_result.count + except Exception as e: + logger.warning("Failed to query Qdrant for indexed count: %s", e) + return JSONResponse( + { + "status": "idle", + "indexed_documents": indexed_count, + "pending_documents": 0, + "status_backend": "bus", + "recent_states": store.counts() if store is not None else {}, + } + ) + # Get document receive stream from app state (set by starlette_lifespan in app.py) document_receive_stream = getattr( request.app.state, "document_receive_stream", None diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index f935fc56..6d175ad8 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -31,6 +31,7 @@ from starlette.staticfiles import StaticFiles from starlette.types import ASGIApp, Receive, Send from starlette.types import Scope as StarletteScope +from nextcloud_mcp_server.admin.payload_backfill import handle_payload_backfill from nextcloud_mcp_server.api import ( create_webhook, delete_app_password, @@ -132,7 +133,13 @@ from nextcloud_mcp_server.vector.oauth_sync import ( from nextcloud_mcp_server.vector.placeholder import sweep_orphan_placeholders from nextcloud_mcp_server.vector.processor import processor_task from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client -from nextcloud_mcp_server.vector.scanner import scanner_task +from nextcloud_mcp_server.vector.queue import ( + MemoryTaskProducer, + TaskProducer, + build_external_producer, +) +from nextcloud_mcp_server.vector.queue.status import NatsStatusSubscriber, StatusStore +from nextcloud_mcp_server.vector.scanner import DocumentTask, scanner_task from nextcloud_mcp_server.vector.webhook_receiver import handle_nextcloud_webhook logger = logging.getLogger(__name__) @@ -324,6 +331,13 @@ class VectorSyncState: document_send_stream: MemoryObjectSendStream | None = None document_receive_stream: MemoryObjectReceiveStream | None = None + # Ingest producer the scanner/webhook send to: the in-memory send stream + # (local mode) or the NATS bus producer (external mode). The webhook reads + # this; in local mode it is the same object as document_send_stream. + task_producer: "TaskProducer | None" = None + # Bus status store (STATUS_BACKEND=bus): populated by the NATS status + # subscriber, read by the vector-sync status endpoint. None in local mode. + status_store: "StatusStore | None" = None shutdown_event: anyio.Event | None = None scanner_wake_event: anyio.Event | None = None # Long-lived task group used for fire-and-forget background work spawned @@ -336,6 +350,28 @@ class VectorSyncState: _vector_sync_state = VectorSyncState() +async def _build_status_subscriber( + settings: "Settings", +) -> "tuple[StatusStore | None, NatsStatusSubscriber | None]": + """Build the bus status store + subscriber when STATUS_BACKEND=bus. + + Returns ``(None, None)`` for local status (the status endpoint reads the + in-memory stream buffer instead). __post_init__ guarantees that bus status + only pairs with external ingest, so ingest_bus_url/tenant_id are set. + """ + if not (settings.ingest_mode == "external" and settings.status_backend == "bus"): + return None, None + assert settings.ingest_bus_url is not None + assert settings.tenant_id is not None + store = StatusStore(max_size=settings.vector_sync_queue_max_size) + subscriber = await NatsStatusSubscriber.connect( + url=settings.ingest_bus_url, + tenant_id=settings.tenant_id, + store=store, + ) + return store, subscriber + + @dataclass class AppContext: """Application context for BasicAuth mode.""" @@ -344,6 +380,7 @@ class AppContext: storage: "RefreshTokenStorage | None" = None document_send_stream: MemoryObjectSendStream | None = None document_receive_stream: MemoryObjectReceiveStream | None = None + task_producer: "TaskProducer | None" = None shutdown_event: anyio.Event | None = None scanner_wake_event: anyio.Event | None = None @@ -369,6 +406,7 @@ class OAuthAppContext: server_client_id: str | None = None # MCP server's OAuth client ID (static or DCR) document_send_stream: MemoryObjectSendStream | None = None document_receive_stream: MemoryObjectReceiveStream | None = None + task_producer: "TaskProducer | None" = None shutdown_event: anyio.Event | None = None scanner_wake_event: anyio.Event | None = None @@ -1643,22 +1681,47 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = # Orphan-sweep before scanner starts — card #101. await _sweep_orphan_placeholders_if_enabled() - # Initialize shared state - send_stream, receive_stream = anyio.create_memory_object_stream( - max_buffer_size=settings.vector_sync_queue_max_size - ) + # Initialize shared state. INGEST_MODE selects the transport + # (design §10.1): local uses the in-memory anyio stream + the + # in-process processor pool; external publishes to NATS and runs no + # in-process consumer (the document-processor consumes). + external = settings.ingest_mode == "external" shutdown_event = anyio.Event() scanner_wake_event = anyio.Event() - # Store in app state for access from routes (ADR-007) + send_stream = None + receive_stream = None + task_producer: TaskProducer + if external: + task_producer = await build_external_producer(settings) + logger.info( + "Ingest mode external: publishing to %s", settings.ingest_bus_url + ) + else: + send_stream, receive_stream = anyio.create_memory_object_stream[ + DocumentTask + ](max_buffer_size=settings.vector_sync_queue_max_size) + task_producer = MemoryTaskProducer(send_stream) + + # Bus status backend: subscribe to mcp.document.* into a store the + # status endpoint reads (STATUS_BACKEND=bus; external mode only). + status_store, status_subscriber = await _build_status_subscriber(settings) + + # Store in app state for access from routes (ADR-007). In external + # mode there is no in-memory stream, so document_send/receive_stream + # stay None; task_producer is the bus producer. app.state.document_send_stream = send_stream app.state.document_receive_stream = receive_stream + app.state.task_producer = task_producer + app.state.status_store = status_store app.state.shutdown_event = shutdown_event app.state.scanner_wake_event = scanner_wake_event # Also store in module singleton for FastMCP session lifespans _vector_sync_state.document_send_stream = send_stream _vector_sync_state.document_receive_stream = receive_stream + _vector_sync_state.task_producer = task_producer + _vector_sync_state.status_store = status_store _vector_sync_state.shutdown_event = shutdown_event _vector_sync_state.scanner_wake_event = scanner_wake_event logger.info("Vector sync state stored in module singleton") @@ -1669,6 +1732,8 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = browser_app = cast(Starlette, route.app) browser_app.state.document_send_stream = send_stream browser_app.state.document_receive_stream = receive_stream + browser_app.state.task_producer = task_producer + browser_app.state.status_store = status_store browser_app.state.shutdown_event = shutdown_event browser_app.state.scanner_wake_event = scanner_wake_event logger.info("Vector sync state shared with browser_app for /app") @@ -1676,26 +1741,33 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = # Start background tasks using anyio TaskGroup async with anyio.create_task_group() as tg: - # Start scanner task + # Start scanner task (publishes to task_producer) await tg.start( scanner_task, - send_stream, + task_producer, shutdown_event, scanner_wake_event, client, username, ) - # Start processor pool (each gets a cloned receive stream) - for i in range(settings.vector_sync_processor_workers): - await tg.start( - processor_task, - i, - receive_stream.clone(), - shutdown_event, - client, - username, - ) + # The in-process processor pool runs only in local mode; in + # external mode the document-processor service is the consumer. + if not external: + assert receive_stream is not None + for i in range(settings.vector_sync_processor_workers): + await tg.start( + processor_task, + i, + receive_stream.clone(), + shutdown_event, + client, + username, + ) + + # Bus status subscriber (STATUS_BACKEND=bus). + if status_subscriber is not None: + await tg.start(status_subscriber.run, shutdown_event) # Expose this long-lived task group to request-path code that # wants to spawn background work (e.g. ADR-019 verify-on-read @@ -1704,8 +1776,9 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = _vector_sync_state.eviction_task_group = tg logger.info( - "Background sync tasks started: 1 scanner + %s processors", - settings.vector_sync_processor_workers, + "Background sync tasks started: 1 scanner + %s processors (ingest=%s)", + 0 if external else settings.vector_sync_processor_workers, + settings.ingest_mode, ) # Run MCP session manager and yield @@ -1718,6 +1791,12 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = shutdown_event.set() # Request path must not spawn into a cancelling group. _vector_sync_state.eviction_task_group = None + # Drain the shared bus connection (external mode only). + _drain = getattr(task_producer, "drain", None) + if external and _drain is not None: + await _drain() + if status_subscriber is not None: + await status_subscriber.aclose() await client.close() # TaskGroup automatically cancels all tasks on exit @@ -1825,25 +1904,51 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = except Exception as e: logger.warning("App password cleanup failed (non-fatal): %s", e) - # Initialize shared state - send_stream, receive_stream = anyio.create_memory_object_stream( - max_buffer_size=settings.vector_sync_queue_max_size - ) + # Initialize shared state. INGEST_MODE selects the transport + # (design §10.1): local uses the in-memory anyio stream + the + # in-process processor pool; external publishes to NATS and runs + # no in-process consumer (the document-processor consumes). + external = settings.ingest_mode == "external" shutdown_event = anyio.Event() scanner_wake_event = anyio.Event() # User state tracking for user manager user_states: dict = {} + send_stream = None + receive_stream = None + task_producer: TaskProducer + if external: + task_producer = await build_external_producer(settings) + logger.info( + "Ingest mode external: publishing to %s", + settings.ingest_bus_url, + ) + else: + send_stream, receive_stream = anyio.create_memory_object_stream[ + DocumentTask + ](max_buffer_size=settings.vector_sync_queue_max_size) + task_producer = MemoryTaskProducer(send_stream) + + # Bus status backend: subscribe to mcp.document.* into a store + # the status endpoint reads (STATUS_BACKEND=bus; external only). + status_store, status_subscriber = await _build_status_subscriber( + settings + ) + # Store in app state for access from routes (ADR-007) app.state.document_send_stream = send_stream app.state.document_receive_stream = receive_stream + app.state.task_producer = task_producer + app.state.status_store = status_store app.state.shutdown_event = shutdown_event app.state.scanner_wake_event = scanner_wake_event # Also store in module singleton for FastMCP session lifespans _vector_sync_state.document_send_stream = send_stream _vector_sync_state.document_receive_stream = receive_stream + _vector_sync_state.task_producer = task_producer + _vector_sync_state.status_store = status_store _vector_sync_state.shutdown_event = shutdown_event _vector_sync_state.scanner_wake_event = scanner_wake_event logger.info("Vector sync state stored in module singleton") @@ -1854,6 +1959,8 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = browser_app = cast(Starlette, route.app) browser_app.state.document_send_stream = send_stream browser_app.state.document_receive_stream = receive_stream + browser_app.state.task_producer = task_producer + browser_app.state.status_store = status_store browser_app.state.shutdown_event = shutdown_event browser_app.state.scanner_wake_event = scanner_wake_event logger.info( @@ -1870,10 +1977,12 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = # `token_broker` constructed above is still used by the # management API revoke endpoint (via app.state.oauth_context). async with anyio.create_task_group() as tg: - # Start user manager task (supervises per-user scanners) + # Start user manager task (supervises per-user scanners). + # Each per-user scanner clones task_producer; for the bus + # producer clone() returns the shared connection. await tg.start( user_manager_task, - send_stream, + task_producer, shutdown_event, scanner_wake_event, token_storage, @@ -1882,15 +1991,22 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = tg, ) - # Start processor pool (each gets a cloned receive stream) - for i in range(settings.vector_sync_processor_workers): - await tg.start( - oauth_processor_task, - i, - receive_stream.clone(), - shutdown_event, - nextcloud_host_for_sync, - ) + # In-process processor pool runs only in local mode; in + # external mode the document-processor service consumes. + if not external: + assert receive_stream is not None + for i in range(settings.vector_sync_processor_workers): + await tg.start( + oauth_processor_task, + i, + receive_stream.clone(), + shutdown_event, + nextcloud_host_for_sync, + ) + + # Bus status subscriber (STATUS_BACKEND=bus). + if status_subscriber is not None: + await tg.start(status_subscriber.run, shutdown_event) # Expose this long-lived task group to request-path code # that wants to spawn background work (e.g. ADR-019 @@ -1899,8 +2015,9 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = _vector_sync_state.eviction_task_group = tg logger.info( - "Background sync tasks started: 1 user manager + %s processors", - settings.vector_sync_processor_workers, + "Background sync tasks started: 1 user manager + %s processors (ingest=%s)", + 0 if external else settings.vector_sync_processor_workers, + settings.ingest_mode, ) # Run MCP session manager and yield @@ -1913,6 +2030,12 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = shutdown_event.set() # Request path must not spawn into a cancelling group. _vector_sync_state.eviction_task_group = None + # Drain the shared bus connection (external only). + _drain = getattr(task_producer, "drain", None) + if external and _drain is not None: + await _drain() + if status_subscriber is not None: + await status_subscriber.aclose() # Close token broker HTTP client if token_broker._http_client: await token_broker._http_client.aclose() @@ -2114,6 +2237,15 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = settings.enable_multi_user_basic_auth and settings.enable_offline_access ) if enable_authenticated_management_apis: + # Admin: one-shot payload backfill (design §10.2). Requires the `admin` + # scope (enforced inside the handler via require_admin_scope). + routes.append( + Route( + "/api/v1/admin/payload-backfill", + handle_payload_backfill, + methods=["POST"], + ) + ) routes.append( Route( "/api/v1/users/{user_id}/session", diff --git a/nextcloud_mcp_server/canonical.py b/nextcloud_mcp_server/canonical.py new file mode 100644 index 00000000..6872edfa --- /dev/null +++ b/nextcloud_mcp_server/canonical.py @@ -0,0 +1,26 @@ +"""Canonical JSON encoding shared across cross-implementation hashes. + +The Astrolabe Cloud decomposition (design §2.3) fixes a single canonical JSON +encoding so hashes computed here match those computed independently by the +external document-processor and embedding-gateway services. Any drift in +separators, key ordering, or unicode handling would break NATS dedup keys, +Qdrant point-ID idempotency, and ACL-hash compatibility. +""" + +from __future__ import annotations + +import json +from typing import Any + + +def canonical_json(obj: Any) -> bytes: + """Encode ``obj`` to canonical JSON bytes. + + Deterministic across implementations: sorted keys, no inter-token + whitespace, non-ASCII preserved (UTF-8). Consumers: the NATS + ``Nats-Msg-Id`` dedup header (vector/queue/nats.py), Qdrant point IDs + (vector/payload_keys.py), and ACL hashes (acl_hash.py). + """ + return json.dumps( + obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False + ).encode("utf-8") diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index b37c19d4..c42a6860 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -161,6 +161,38 @@ _DEFAULTS: dict[str, Any] = { # Nextcloud system tag names. Files/folders carrying any of these tags # are hidden from WebDAV MCP tools. Empty = feature off. "excluded_tags": "", + # MCP decomposition hook points (design §10). Every default reproduces + # the current monolithic behavior; self-hosters who set none are + # unaffected. See docs/architecture/mcp-decomposition.md (sibling repo). + "embedding_provider": "autodetect", # autodetect | gateway + "ingest_mode": "local", # local | external + "status_backend": "local", # local | bus + "collection_metadata_source": "qdrant", # qdrant | api + # CP base URL for COLLECTION_METADATA_SOURCE=api (e.g. http://control-plane). + # Required only when the source is api. + "collection_metadata_api_url": None, + "fact_event_emitter": "none", # none | nats | stdout + "ingest_bus_url": None, # required when ingest_mode=external + "embedding_gateway_url": None, # required when embedding_provider=gateway + # Logical model the gateway routes on (e.g. mistral-embed → Mistral for + # the MVP). Only consulted when embedding_provider=gateway. + "embedding_gateway_model": "mistral-embed", + # Gateway auth: the MCP server is an OIDC *client* in the gateway's own + # M2M realm (parallel to, and distinct from, the tenant realm it already + # serves). It obtains a client-credentials token and the gateway maps the + # client-id → the tenant's underlying provider API key. All four unset = + # call the gateway unauthenticated (matches today's not-yet-authed gateway). + "embedding_gateway_token_url": None, # M2M token endpoint + "embedding_gateway_client_id": None, + "embedding_gateway_client_secret": None, + "embedding_gateway_scope": None, # e.g. astrolabe-embedding-gateway/embed + "tenant_id": None, # NATS per-tenant subject token (UUID form) + "ingest_bus_num_replicas": 1, # JetStream stream replicas (prod: 3) + # Query-side ACL pre-filter (design §11). OFF by default: a Qdrant + # `match any` on `acl_hash` excludes points missing the key, so enabling + # this before a real ACL backfill would silently drop legacy results. + # verify-on-read remains the correctness backstop regardless. + "acl_prefilter_enabled": False, } @@ -653,6 +685,26 @@ class Settings: # are hidden from WebDAV MCP tools. excluded_tags: str = "" + # MCP decomposition hook points (design §10, opt-in). All defaults + # reproduce the current monolith; validated in __post_init__. + embedding_provider: str = "autodetect" # autodetect | gateway + ingest_mode: str = "local" # local | external + status_backend: str = "local" # local | bus + collection_metadata_source: str = "qdrant" # qdrant | api + collection_metadata_api_url: str | None = None # CP URL when source=api + fact_event_emitter: str = "none" # none | nats | stdout + ingest_bus_url: str | None = None # required when ingest_mode=external + embedding_gateway_url: str | None = None # required when provider=gateway + embedding_gateway_model: str = "mistral-embed" # logical model gateway routes + # Gateway M2M OIDC client creds (separate realm; see _DEFAULTS comment). + embedding_gateway_token_url: str | None = None + embedding_gateway_client_id: str | None = None + embedding_gateway_client_secret: str | None = None + embedding_gateway_scope: str | None = None + tenant_id: str | None = None # NATS per-tenant subject token (UUID form) + ingest_bus_num_replicas: int = 1 # JetStream stream replicas (prod: 3) + acl_prefilter_enabled: bool = False # query-side ACL pre-filter (§11); OFF + def __post_init__(self): """Validate configuration and set defaults.""" logger = logging.getLogger(__name__) @@ -733,6 +785,84 @@ class Settings: self.document_chunk_size, ) + # --- MCP decomposition hook points (design §10) --- + # Normalize + validate the opt-in enum settings. Defaults reproduce + # the monolith, so deployments that set none of these pass through. + _enum_fields = { + "embedding_provider": {"autodetect", "gateway"}, + "ingest_mode": {"local", "external"}, + "status_backend": {"local", "bus"}, + "collection_metadata_source": {"qdrant", "api"}, + "fact_event_emitter": {"none", "nats", "stdout"}, + } + for _field, _allowed in _enum_fields.items(): + _val = (getattr(self, _field) or "").strip().lower() + setattr(self, _field, _val) + if _val not in _allowed: + raise ValueError( + f"{_field.upper()} must be one of {sorted(_allowed)}; got {_val!r}" + ) + + # Fail-fast: external ingest sources its status from the bus. With the + # in-process state machine empty, STATUS_BACKEND=local would leave + # status streams silently empty — crash loudly instead (design §10.1). + if self.status_backend == "local" and self.ingest_mode == "external": + raise RuntimeError( + "STATUS_BACKEND=local is incompatible with INGEST_MODE=external; " + "set STATUS_BACKEND=bus" + ) + + # Conditional-required settings for the active hook points. + if self.ingest_mode == "external": + if not self.ingest_bus_url: + raise ValueError("INGEST_BUS_URL is required when INGEST_MODE=external") + if not self.tenant_id: + raise ValueError("TENANT_ID is required when INGEST_MODE=external") + if self.embedding_provider == "gateway" and not self.embedding_gateway_url: + raise ValueError( + "EMBEDDING_GATEWAY_URL is required when EMBEDDING_PROVIDER=gateway" + ) + if ( + self.collection_metadata_source == "api" + and not self.collection_metadata_api_url + ): + raise ValueError( + "COLLECTION_METADATA_API_URL is required when " + "COLLECTION_METADATA_SOURCE=api" + ) + + # Gateway M2M OIDC creds are all-or-nothing: a partial set (e.g. a + # client_id with no token endpoint) is a misconfiguration that would + # silently fall back to unauthenticated calls. scope is optional. + _gw_creds = ( + self.embedding_gateway_token_url, + self.embedding_gateway_client_id, + self.embedding_gateway_client_secret, + ) + if any(_gw_creds) and not all(_gw_creds): + raise ValueError( + "EMBEDDING_GATEWAY_TOKEN_URL, EMBEDDING_GATEWAY_CLIENT_ID, and " + "EMBEDDING_GATEWAY_CLIENT_SECRET must be set together (M2M OIDC " + "client-credentials) or all left unset (unauthenticated gateway)" + ) + + # TENANT_ID is a NATS subject token; '.', '*', '>', and whitespace are + # reserved/illegal there and would silently break subscriptions (§3.4). + if self.tenant_id and ( + any(c in self.tenant_id for c in ".*>") + or any(c.isspace() for c in self.tenant_id) + ): + raise ValueError( + "TENANT_ID must not contain '.', '*', '>', or whitespace " + "(it is used as a NATS subject token)" + ) + + if self.ingest_bus_num_replicas < 1: + raise ValueError( + f"INGEST_BUS_NUM_REPLICAS must be >= 1; " + f"got {self.ingest_bus_num_replicas}" + ) + # --- ADR-022 follow-up: deployment mode is the single source of truth --- # The ENABLE_MULTI_USER_BASIC_AUTH and ENABLE_LOGIN_FLOW env vars were # removed in favour of MCP_DEPLOYMENT_MODE. We do TWO things here: @@ -1128,6 +1258,23 @@ def get_settings() -> Settings: "log_level": "LOG_LEVEL", "log_include_trace_context": "LOG_INCLUDE_TRACE_CONTEXT", "excluded_tags": "EXCLUDED_TAGS", + # MCP decomposition hook points (design §10) + "embedding_provider": "EMBEDDING_PROVIDER", + "ingest_mode": "INGEST_MODE", + "status_backend": "STATUS_BACKEND", + "collection_metadata_source": "COLLECTION_METADATA_SOURCE", + "collection_metadata_api_url": "COLLECTION_METADATA_API_URL", + "fact_event_emitter": "FACT_EVENT_EMITTER", + "ingest_bus_url": "INGEST_BUS_URL", + "embedding_gateway_url": "EMBEDDING_GATEWAY_URL", + "embedding_gateway_model": "EMBEDDING_GATEWAY_MODEL", + "embedding_gateway_token_url": "EMBEDDING_GATEWAY_TOKEN_URL", + "embedding_gateway_client_id": "EMBEDDING_GATEWAY_CLIENT_ID", + "embedding_gateway_client_secret": "EMBEDDING_GATEWAY_CLIENT_SECRET", + "embedding_gateway_scope": "EMBEDDING_GATEWAY_SCOPE", + "tenant_id": "TENANT_ID", + "ingest_bus_num_replicas": "INGEST_BUS_NUM_REPLICAS", + "acl_prefilter_enabled": "ACL_PREFILTER_ENABLED", } # Only pass values that dynaconf actually has; omit unset keys so diff --git a/nextcloud_mcp_server/embedding/gateway_client.py b/nextcloud_mcp_server/embedding/gateway_client.py new file mode 100644 index 00000000..7220f4a5 --- /dev/null +++ b/nextcloud_mcp_server/embedding/gateway_client.py @@ -0,0 +1,137 @@ +"""OpenAI-compatible embedding provider targeting the Astrolabe Cloud embedding +gateway (design §10.2). + +Active only when ``EMBEDDING_PROVIDER=gateway``. Registered *manually* in +``providers/registry.py`` — never part of the autodetect chain — so self-hosters +who don't opt in are unaffected. + +**Auth model.** The MCP server is an OIDC *client* in the gateway's own +machine-to-machine realm — a realm *parallel to, and distinct from*, the tenant +realm the MCP server already serves as a client (Nextcloud user_oidc). It +obtains a ``client_credentials`` token and presents it as a Bearer; the gateway +maps the token's client-id → the tenant's underlying provider API key. This +mirrors the control-plane CLI's ``fetch_m2m_token`` pattern +(astrolabe-cloud-website ``services/control-plane/.../cli/_common.py``). When no +M2M creds are configured the client calls the gateway unauthenticated — matching +the gateway's current (not-yet-authenticated) state. + +The gateway speaks the OpenAI ``/v1/embeddings`` wire format and routes by model +name (e.g. ``mistral-embed`` → Mistral for the MVP). Embeddings-only: ``generate`` +is disabled (inherited ``NotImplementedError``). +""" + +from __future__ import annotations + +import logging +import time + +import httpx + +from ..providers.openai import OpenAIProvider + +logger = logging.getLogger(__name__) + +# Refresh the cached token this many seconds before its stated expiry, so a +# token never expires mid-flight (matches AstrolabeClient / CP CLI behavior). +_EARLY_REFRESH_SECONDS = 60 + + +class GatewayTokenProvider: + """Caches a gateway M2M access token via the ``client_credentials`` grant. + + HTTP Basic client auth + form-encoded grant, mirroring the website's + ``fetch_m2m_token``. Tokens are cached until ``_EARLY_REFRESH_SECONDS`` + before expiry. + """ + + def __init__( + self, + token_url: str, + client_id: str, + client_secret: str, + scope: str | None = None, + timeout: float = 10.0, + ): + self.token_url = token_url + self.client_id = client_id + self.client_secret = client_secret + self.scope = scope + self.timeout = timeout + self._cache: tuple[str, float] | None = None # (token, expires_at) + + async def get_token(self, *, force_refresh: bool = False) -> str: + if ( + self._cache is not None + and not force_refresh + and time.time() < self._cache[1] + ): + return self._cache[0] + + data = {"grant_type": "client_credentials"} + if self.scope: + data["scope"] = self.scope + + async with httpx.AsyncClient( + timeout=httpx.Timeout(self.timeout, connect=5.0) + ) as client: + resp = await client.post( + self.token_url, + data=data, + auth=(self.client_id, self.client_secret), + ) + resp.raise_for_status() + body = resp.json() + + expires_in = body.get("expires_in", 3600) + self._cache = ( + body["access_token"], + time.time() + expires_in - _EARLY_REFRESH_SECONDS, + ) + logger.info("Obtained embedding-gateway M2M token (expires in %ss)", expires_in) + return self._cache[0] + + +class GatewayProvider(OpenAIProvider): + """Embeddings-only OpenAI-compatible provider pointed at the gateway.""" + + def __init__( + self, + *, + base_url: str, + embedding_model: str, + token_provider: GatewayTokenProvider | None = None, + timeout: float = 120.0, + ): + # AsyncOpenAI rejects an empty key; use a non-secret placeholder when + # the gateway is unauthenticated. When a token provider is configured, + # the real Bearer is set on the client before each request. + super().__init__( + api_key="gateway-unauthenticated", + base_url=base_url, + embedding_model=embedding_model, + generation_model=None, # gateway never generates + timeout=timeout, + ) + self._token_provider = token_provider + logger.info( + "Initialized gateway embedding provider: base_url=%s, model=%s, auth=%s", + base_url, + embedding_model, + "oidc-m2m" if token_provider else "none", + ) + + async def _ensure_bearer(self) -> None: + """Refresh the OIDC M2M token onto the OpenAI client (no-op when + unauthenticated). AsyncOpenAI reads ``api_key`` per request to build + the Authorization header, so updating it here applies to the next call. + """ + if self._token_provider is not None: + self.client.api_key = await self._token_provider.get_token() + + async def embed(self, text: str) -> list[float]: + await self._ensure_bearer() + return await super().embed(text) + + async def embed_batch(self, texts: list[str]) -> list[list[float]]: + await self._ensure_bearer() + return await super().embed_batch(texts) diff --git a/nextcloud_mcp_server/providers/registry.py b/nextcloud_mcp_server/providers/registry.py index 4768f998..bea0e029 100644 --- a/nextcloud_mcp_server/providers/registry.py +++ b/nextcloud_mcp_server/providers/registry.py @@ -50,6 +50,44 @@ class ProviderRegistry: """ settings = get_settings() + # 0. Gateway (manual-only; never autodetected). When + # EMBEDDING_PROVIDER=gateway, bypass the autodetect chain entirely + # and route embeddings through the Astrolabe Cloud embedding gateway + # (design §10.1/§10.2). Lazy import to avoid an import cycle + # (embedding/gateway_client → providers/openai → ... → registry). + if settings.embedding_provider == "gateway": + from ..embedding.gateway_client import ( # noqa: PLC0415 + GatewayProvider, + GatewayTokenProvider, + ) + + # Settings.__post_init__ guarantees the URL is set when the + # provider is gateway; assert narrows the type for the checker. + assert settings.embedding_gateway_url is not None + # __post_init__ enforces all-or-nothing on the M2M creds, so + # checking one is enough to know the full triple is present. + token_provider = None + if settings.embedding_gateway_client_id: + assert settings.embedding_gateway_token_url is not None + assert settings.embedding_gateway_client_secret is not None + token_provider = GatewayTokenProvider( + token_url=settings.embedding_gateway_token_url, + client_id=settings.embedding_gateway_client_id, + client_secret=settings.embedding_gateway_client_secret, + scope=settings.embedding_gateway_scope, + ) + logger.info( + "Using embedding gateway provider: url=%s, model=%s, auth=%s", + settings.embedding_gateway_url, + settings.embedding_gateway_model, + "oidc-m2m" if token_provider else "none", + ) + return GatewayProvider( + base_url=settings.embedding_gateway_url, + embedding_model=settings.embedding_gateway_model, + token_provider=token_provider, + ) + # 1. Bedrock if ( settings.aws_region diff --git a/nextcloud_mcp_server/search/semantic.py b/nextcloud_mcp_server/search/semantic.py index 54465c03..321080db 100644 --- a/nextcloud_mcp_server/search/semantic.py +++ b/nextcloud_mcp_server/search/semantic.py @@ -3,8 +3,9 @@ import logging from typing import Any -from qdrant_client.models import FieldCondition, Filter, MatchValue +from qdrant_client.models import FieldCondition, Filter, MatchAny, MatchValue +from nextcloud_mcp_server.acl_hash import accessible_hash_set from nextcloud_mcp_server.config import get_settings from nextcloud_mcp_server.embedding import get_embedding_service from nextcloud_mcp_server.observability.metrics import record_qdrant_operation @@ -13,6 +14,7 @@ from nextcloud_mcp_server.search.algorithms import ( SearchResult, build_search_result_from_point, ) +from nextcloud_mcp_server.vector.payload_keys import ACL_HASH from nextcloud_mcp_server.vector.placeholder import get_placeholder_filter from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client @@ -112,6 +114,20 @@ class SemanticSearchAlgorithm(SearchAlgorithm): ) ) + # ACL pre-filter (design §11), opt-in via ACL_PREFILTER_ENABLED and OFF + # by default. Additive `must` condition — it can only narrow results, + # never broaden them, and verify-on-read remains the correctness + # backstop. Only enable after a real acl_hash backfill: a MatchAny on + # acl_hash excludes points missing the key (legacy docs), so enabling + # it on an un-backfilled collection would silently drop results. + if settings.acl_prefilter_enabled: + # Groups are not yet threaded into the search signature; user + + # public principals are covered. Group support is a follow-up. + accessible = accessible_hash_set(user_id) + filter_conditions.append( + FieldCondition(key=ACL_HASH, match=MatchAny(any=sorted(accessible))) + ) + # Search Qdrant qdrant_client = await get_qdrant_client() try: diff --git a/nextcloud_mcp_server/vector/collection_metadata.py b/nextcloud_mcp_server/vector/collection_metadata.py new file mode 100644 index 00000000..9d8f7f4c --- /dev/null +++ b/nextcloud_mcp_server/vector/collection_metadata.py @@ -0,0 +1,155 @@ +"""Per-collection metadata: embedding identity + chunking config (design §10.1). + +The query path needs to know which embedding produced a collection's vectors +(``embedding_identity``) and how it was chunked (``chunking_config``) so it can +request the matching embedding at lookup time. Two sources, selected by +``COLLECTION_METADATA_SOURCE``: + +- ``qdrant`` — a sentinel point (deterministic UUID, normalisable non-zero dense + vector) stored inside the collection. Works for any Qdrant deployment, so + self-hosters benefit even without a control plane. +- ``api`` — an HTTP GET against the control plane + (``/v1/qdrant-collections/{name}/metadata``). + +On a missing/unreadable sentinel the query path logs a warning and falls back to +the environment-configured defaults — matching today's monolith behavior, so +query availability is preserved (design §10.1). +""" + +from __future__ import annotations + +import logging +from typing import Any + +import httpx +from qdrant_client import AsyncQdrantClient, models + +from ..config import Settings, get_settings +from .payload_keys import EMBEDDING_IDENTITY + +logger = logging.getLogger(__name__) + +# Deterministic sentinel point id (design §10.1). Carries collection metadata +# and never matches a search (no user_id/doc_id/doc_type payload to match). +SENTINEL_POINT_ID = "00000000-0000-0000-0000-000000000000" + +# Sentinel payload keys. +CHUNKING_CONFIG = "chunking_config" +IS_SENTINEL = "is_sentinel" + + +def build_embedding_identity(settings: Settings | None = None) -> str: + """The embedding identity for locally-produced vectors: the model name. + + The gateway and query path route on this name; for the monolith it is the + active embedding model (matching the collection-name derivation). + """ + s = settings or get_settings() + return s.get_embedding_model_name() + + +def env_default_metadata(settings: Settings | None = None) -> dict[str, Any]: + """Metadata derived purely from environment config — the fallback when no + sentinel/API metadata is available.""" + s = settings or get_settings() + return { + "embedding_identity": build_embedding_identity(s), + "chunking_config": { + "chunk_size": s.document_chunk_size, + "chunk_overlap": s.document_chunk_overlap, + }, + } + + +def _sentinel_dense(dimension: int) -> list[float]: + # Cosine distance is undefined for the zero vector (and Qdrant Cloud strict + # mode rejects it), so use one tiny non-zero element — mirrors the doc-id + # backfill sentinel in qdrant_client.py. + return [1e-9] + [0.0] * (dimension - 1) + + +async def upsert_sentinel( + client: AsyncQdrantClient, + collection_name: str, + *, + embedding_identity: str, + chunking_config: dict[str, Any], + dimension: int, +) -> None: + """Idempotently write the metadata sentinel point for a collection.""" + point = models.PointStruct( + id=SENTINEL_POINT_ID, + vector={ + "dense": _sentinel_dense(dimension), + "sparse": models.SparseVector(indices=[], values=[]), + }, + payload={ + EMBEDDING_IDENTITY: embedding_identity, + CHUNKING_CONFIG: chunking_config, + IS_SENTINEL: True, + }, + ) + await client.upsert(collection_name=collection_name, points=[point], wait=True) + logger.debug("Upserted metadata sentinel on '%s'", collection_name) + + +async def _read_from_qdrant( + client: AsyncQdrantClient, collection_name: str +) -> dict[str, Any] | None: + points = await client.retrieve( + collection_name=collection_name, + ids=[SENTINEL_POINT_ID], + with_payload=True, + ) + if not points: + return None + payload = points[0].payload or {} + if EMBEDDING_IDENTITY not in payload: + return None + return { + "embedding_identity": payload.get(EMBEDDING_IDENTITY), + "chunking_config": payload.get(CHUNKING_CONFIG), + } + + +async def _read_from_api(api_url: str, collection_name: str) -> dict[str, Any] | None: + url = f"{api_url.rstrip('/')}/v1/qdrant-collections/{collection_name}/metadata" + async with httpx.AsyncClient(timeout=httpx.Timeout(10.0, connect=5.0)) as client: + resp = await client.get(url) + if resp.status_code == 404: + return None + resp.raise_for_status() + return resp.json() + + +async def read_collection_metadata( + client: AsyncQdrantClient, + collection_name: str, + settings: Settings | None = None, +) -> dict[str, Any]: + """Read collection metadata from the configured source, falling back to env + defaults on any miss/error (preserves query availability — §10.1).""" + s = settings or get_settings() + meta: dict[str, Any] | None = None + try: + if s.collection_metadata_source == "api": + assert s.collection_metadata_api_url is not None + meta = await _read_from_api(s.collection_metadata_api_url, collection_name) + else: + meta = await _read_from_qdrant(client, collection_name) + except Exception: + logger.warning( + "Collection metadata read failed for '%s' (source=%s); using env defaults", + collection_name, + s.collection_metadata_source, + exc_info=True, + ) + + if not meta or not meta.get("embedding_identity"): + logger.warning( + "Collection metadata missing for '%s' (source=%s); using env defaults", + collection_name, + s.collection_metadata_source, + ) + return env_default_metadata(s) + return meta diff --git a/nextcloud_mcp_server/vector/oauth_sync.py b/nextcloud_mcp_server/vector/oauth_sync.py index 6a44eebf..bc60cf60 100644 --- a/nextcloud_mcp_server/vector/oauth_sync.py +++ b/nextcloud_mcp_server/vector/oauth_sync.py @@ -24,16 +24,14 @@ from dataclasses import dataclass, field import anyio from anyio.abc import TaskGroup, TaskStatus -from anyio.streams.memory import ( - MemoryObjectReceiveStream, - MemoryObjectSendStream, -) +from anyio.streams.memory import MemoryObjectReceiveStream from httpx import BasicAuth, HTTPStatusError from nextcloud_mcp_server.auth.storage import RefreshTokenStorage from nextcloud_mcp_server.client import NextcloudClient from nextcloud_mcp_server.config import get_settings from nextcloud_mcp_server.vector.processor import process_document +from nextcloud_mcp_server.vector.queue.ports import TaskProducer from nextcloud_mcp_server.vector.scanner import DocumentTask, scan_user_documents logger = logging.getLogger(__name__) @@ -101,7 +99,7 @@ async def get_user_client_basic_auth( async def user_scanner_task( user_id: str, - send_stream: MemoryObjectSendStream[DocumentTask], + send_stream: TaskProducer, shutdown_event: anyio.Event, wake_event: anyio.Event, nextcloud_host: str, @@ -330,7 +328,7 @@ oauth_processor_task = multi_user_processor_task async def _run_user_scanner_with_scope( user_id: str, cancel_scope: anyio.CancelScope, - send_stream: MemoryObjectSendStream[DocumentTask], + send_stream: TaskProducer, shutdown_event: anyio.Event, wake_event: anyio.Event, nextcloud_host: str, @@ -358,7 +356,7 @@ async def _run_user_scanner_with_scope( async def user_manager_task( - send_stream: MemoryObjectSendStream[DocumentTask], + send_stream: TaskProducer, shutdown_event: anyio.Event, wake_event: anyio.Event, refresh_token_storage: "RefreshTokenStorage", diff --git a/nextcloud_mcp_server/vector/payload_keys.py b/nextcloud_mcp_server/vector/payload_keys.py new file mode 100644 index 00000000..1b4f5795 --- /dev/null +++ b/nextcloud_mcp_server/vector/payload_keys.py @@ -0,0 +1,45 @@ +"""Qdrant payload-key constants + the shared point-ID namespace (design §2.2). + +These names and the ``NAMESPACE`` UUID are a cross-implementation contract: the +external document-processor (astrolabe-cloud-website) computes identical chunk +point IDs and writes the same payload keys. Divergence in ``NAMESPACE`` would +break Qdrant upsert idempotency and duplicate chunks at the Phase 2 cutover, so +the literal is pinned by a fixture checked into both repos +(``tests/fixtures/namespace_uuid.txt``) and asserted equal in each repo's suite. + +Even in the default local mode the MCP server writes these keys on upsert, so a +later migration to the external processor is friction-free (design §10.2). +""" + +from __future__ import annotations + +import uuid + +from nextcloud_mcp_server.canonical import canonical_json + +# Payload keys introduced by the decomposition (design §10.2). +EMBEDDING_IDENTITY = "embedding_identity" +ACL_HASH = "acl_hash" +PROCESSOR_VERSION = "processor_version" +PARSED_AT = "parsed_at" +PIPELINE_TIER = "pipeline_tier" + +# Fixed platform namespace for deterministic chunk point IDs (design §2.2). +# Derived once from ``uuid5(NAMESPACE_DNS, "astrolabe.cloud/mcp/point-id/v1")`` +# and pinned here as a literal so neither repo recomputes it. DO NOT CHANGE — +# see the module docstring for the cross-implementation contract. +NAMESPACE = uuid.UUID("b050b8ac-c6aa-5566-9584-506b39c1c096") + + +def point_id(tenant_id: str, doc_id: str, chunk_index: int) -> str: + """Deterministic chunk point ID, identical across MCP server + processor. + + ``uuid5(NAMESPACE, canonical_json({...}))`` per design §2.2. The canonical + JSON name (sorted keys, no whitespace) must match the processor + byte-for-byte so re-indexing the same chunk upserts in place rather than + duplicating. + """ + name = canonical_json( + {"tenant_id": tenant_id, "doc_id": doc_id, "chunk_index": chunk_index} + ).decode("utf-8") + return str(uuid.uuid5(NAMESPACE, name)) diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index d77f1868..01fbcfdf 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -14,6 +14,7 @@ from anyio.streams.memory import MemoryObjectReceiveStream from httpx import HTTPStatusError from qdrant_client.models import FieldCondition, Filter, MatchValue, PointStruct +from nextcloud_mcp_server.acl_hash import compute_acl_hash from nextcloud_mcp_server.client import NextcloudClient from nextcloud_mcp_server.config import get_settings from nextcloud_mcp_server.document_processors import get_registry @@ -25,6 +26,7 @@ from nextcloud_mcp_server.observability.metrics import ( ) from nextcloud_mcp_server.observability.tracing import trace_operation from nextcloud_mcp_server.search.pdf_highlighter import PDFHighlighter +from nextcloud_mcp_server.vector import payload_keys from nextcloud_mcp_server.vector.document_chunker import DocumentChunker from nextcloud_mcp_server.vector.html_processor import html_to_markdown from nextcloud_mcp_server.vector.placeholder import delete_placeholder_point @@ -672,6 +674,15 @@ async def _index_document( indexed_at = int(time.time()) points = [] + # Decomposition payload keys (design §10.2) — written even in local mode so + # a future migration to the external processor is friction-free. Computed + # once per document (not per chunk). The local processor has no triage, so + # PIPELINE_TIER is "fast"; ACL hash records at least the owner principal + # (full share enumeration is a follow-up — a missing/partial acl_hash is + # safe because the query-side pre-filter only applies when present + enabled). + _embedding_identity = get_settings().get_embedding_model_name() + _acl_hash = compute_acl_hash([("user", 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 @@ -719,6 +730,12 @@ async def _index_document( "chunk_start_offset": chunk.start_offset, "chunk_end_offset": chunk.end_offset, "metadata_version": 2, # v2 includes position metadata + # Decomposition payload keys (design §10.2), additive. + payload_keys.PROCESSOR_VERSION: "monolith-v1", + payload_keys.PARSED_AT: indexed_at, + payload_keys.PIPELINE_TIER: "fast", + payload_keys.EMBEDDING_IDENTITY: _embedding_identity, + payload_keys.ACL_HASH: _acl_hash, # File-specific metadata (PDF, etc.) **( { diff --git a/nextcloud_mcp_server/vector/queue/__init__.py b/nextcloud_mcp_server/vector/queue/__init__.py new file mode 100644 index 00000000..dcc55e2b --- /dev/null +++ b/nextcloud_mcp_server/vector/queue/__init__.py @@ -0,0 +1,7 @@ +"""Ingest-path ports & adapters (design §10, hexagonal).""" + +from .factory import build_external_producer +from .memory import MemoryTaskProducer +from .ports import TaskProducer + +__all__ = ["MemoryTaskProducer", "TaskProducer", "build_external_producer"] diff --git a/nextcloud_mcp_server/vector/queue/factory.py b/nextcloud_mcp_server/vector/queue/factory.py new file mode 100644 index 00000000..8001e016 --- /dev/null +++ b/nextcloud_mcp_server/vector/queue/factory.py @@ -0,0 +1,46 @@ +"""Composition root for the ingest producer (design §10). + +``build_external_producer`` is called by the lifespan only when +``INGEST_MODE=external``; local mode uses the in-memory stream directly (it +already satisfies :class:`TaskProducer`). The transport under ``external`` is +selected from the ``INGEST_BUS_URL`` scheme (``nats://`` now, ``postgres://`` +later) so moving the external processor to Postgres needs no new INGEST_MODE. +""" + +from __future__ import annotations + +from urllib.parse import urlsplit + +from ...config import Settings +from .ports import TaskProducer + + +def _transport_for(url: str) -> str: + scheme = urlsplit(url).scheme.lower() + if scheme.startswith("postgres"): + return "postgres" + return "nats" + + +async def build_external_producer(settings: Settings) -> TaskProducer: + """Build the external-ingest producer for the configured transport. + + Precondition: ``settings.ingest_mode == "external"`` (so __post_init__ has + guaranteed ``ingest_bus_url`` and ``tenant_id`` are set). + """ + assert settings.ingest_bus_url is not None + assert settings.tenant_id is not None + + transport = _transport_for(settings.ingest_bus_url) + if transport == "postgres": + from .postgres import PostgresTaskProducer # noqa: PLC0415 + + return await PostgresTaskProducer.connect(settings) + + from .nats import NatsTaskProducer # noqa: PLC0415 + + return await NatsTaskProducer.connect( + url=settings.ingest_bus_url, + tenant_id=settings.tenant_id, + num_replicas=settings.ingest_bus_num_replicas, + ) diff --git a/nextcloud_mcp_server/vector/queue/memory.py b/nextcloud_mcp_server/vector/queue/memory.py new file mode 100644 index 00000000..f4a3a248 --- /dev/null +++ b/nextcloud_mcp_server/vector/queue/memory.py @@ -0,0 +1,45 @@ +"""In-memory ``TaskProducer`` — the default (local) ingest transport. + +A thin adapter over anyio's ``MemoryObjectSendStream`` so the local path is an +explicit :class:`TaskProducer` (rather than relying on structural typing of a +third-party class). Semantics are identical to using the stream directly: +``send`` enqueues, ``clone`` yields an independent per-user handle, ``async +with`` / ``aclose`` close the (cloned) send end so the processor pool's +receivers observe end-of-stream. +""" + +from __future__ import annotations + +from types import TracebackType +from typing import TYPE_CHECKING + +from anyio.streams.memory import MemoryObjectSendStream + +if TYPE_CHECKING: + from ..scanner import DocumentTask + + +class MemoryTaskProducer: + def __init__(self, stream: MemoryObjectSendStream[DocumentTask]): + self._stream = stream + + async def send(self, task: DocumentTask, /) -> None: + await self._stream.send(task) + + def clone(self) -> MemoryTaskProducer: + return MemoryTaskProducer(self._stream.clone()) + + async def __aenter__(self) -> MemoryTaskProducer: + await self._stream.__aenter__() + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + tb: TracebackType | None, + ) -> None: + await self._stream.__aexit__(exc_type, exc, tb) + + async def aclose(self) -> None: + await self._stream.aclose() diff --git a/nextcloud_mcp_server/vector/queue/nats.py b/nextcloud_mcp_server/vector/queue/nats.py new file mode 100644 index 00000000..4d25a9be --- /dev/null +++ b/nextcloud_mcp_server/vector/queue/nats.py @@ -0,0 +1,147 @@ +"""NATS JetStream ``TaskProducer`` — external ingest transport (design §3.4). + +Publishes ``mcp.ingest.requested.{tenant_id}`` for the external +document-processor to consume. Translates the in-process ``DocumentTask`` into +the wire ``IngestMessage`` schema (mirrored in astrolabe-cloud-website's +``bus/messages.py``), with the JetStream ``Nats-Msg-Id`` dedup header per §3.4. + +This server is only the *producer* on this transport; the document-processor +owns the consumer. ``nats-py`` is imported lazily so deployments that never +enable external ingest don't pay the import. +""" + +from __future__ import annotations + +import hashlib +import logging +from datetime import datetime, timezone +from types import TracebackType +from typing import TYPE_CHECKING, Any + +from ...canonical import canonical_json + +if TYPE_CHECKING: + from ..scanner import DocumentTask + +logger = logging.getLogger(__name__) + +STREAM_NAME = "mcp" +INGEST_SUBJECT_PREFIX = "mcp.ingest.requested" + + +def _modified_at_rfc3339(modified_at: int) -> str: + """DocumentTask.modified_at is an epoch int (0 for deletes).""" + return datetime.fromtimestamp(int(modified_at), tz=timezone.utc).isoformat() + + +def _content_hash(task: DocumentTask) -> str: + """etag is the change-detection token; fall back to modified_at when it is + absent (e.g. deletes, or sources whose etag we don't thread through).""" + return task.etag or str(task.modified_at) + + +def msg_id(tenant_id: str, doc_id: str, modified_at_rfc3339: str) -> str: + """JetStream dedup header per §3.4. SHA-256 over canonical JSON (NOT the + BLAKE2b helper) — it is an opaque external header, not a stored field.""" + return hashlib.sha256( + canonical_json( + { + "tenant_id": tenant_id, + "doc_id": doc_id, + "modified_at": modified_at_rfc3339, + } + ) + ).hexdigest() + + +class NatsTaskProducer: + """Publishes ingest requests to NATS JetStream.""" + + def __init__(self, nc: Any, js: Any, tenant_id: str): + self._nc = nc + self._js = js + self.tenant_id = tenant_id + + @classmethod + async def connect( + cls, *, url: str, tenant_id: str, num_replicas: int = 1 + ) -> NatsTaskProducer: + import nats # noqa: PLC0415 (lazy: optional dependency for external mode) + + nc = await nats.connect(url) + js = nc.jetstream() + await cls._ensure_stream(js, num_replicas) + logger.info("Connected NATS ingest producer: url=%s, tenant=%s", url, tenant_id) + return cls(nc, js, tenant_id) + + @staticmethod + async def _ensure_stream(js: Any, num_replicas: int) -> None: + # noqa: PLC0415 — nats.js types are only importable once nats-py is present. + from nats.js.api import RetentionPolicy, StreamConfig # noqa: PLC0415 + + config = StreamConfig( + name=STREAM_NAME, + subjects=["mcp.>"], + retention=RetentionPolicy.LIMITS, + num_replicas=num_replicas, + ) + try: + await js.add_stream(config=config) + logger.info("nats.stream_created stream=%s", STREAM_NAME) + except Exception as exc: + # add_stream is idempotent in spirit but errors when the stream + # already exists; treat as benign (mirrors the processor's + # ensure_stream). A genuinely broken broker surfaces on publish. + logger.info("nats.stream_exists_or_unavailable detail=%s", exc) + + def ingest_message(self, task: DocumentTask) -> dict[str, Any]: + """DocumentTask → wire IngestMessage dict (mirrors the sibling schema).""" + return { + "tenant_id": self.tenant_id, + "doc_id": task.doc_id, + "content_hash": _content_hash(task), + "modified_at": _modified_at_rfc3339(task.modified_at), + "doc_type": task.doc_type, + "operation": task.operation, + "user_id": task.user_id, + "file_path": task.file_path, + } + + async def send(self, task: DocumentTask) -> None: + message = self.ingest_message(task) + subject = f"{INGEST_SUBJECT_PREFIX}.{self.tenant_id}" + headers = { + "Nats-Msg-Id": msg_id(self.tenant_id, task.doc_id, message["modified_at"]) + } + await self._js.publish(subject, canonical_json(message), headers=headers) + + # The scanner/oauth_sync use the producer as a clone-able async context + # manager (memory-stream semantics). The bus connection is owned by the + # lifespan, so cloning shares it and __aexit__ is a no-op (close happens via + # aclose() on shutdown). + def clone(self) -> NatsTaskProducer: + return self + + async def __aenter__(self) -> NatsTaskProducer: + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + tb: TracebackType | None, + ) -> None: + return None + + async def aclose(self) -> None: + # Per-handle close (e.g. a per-user scanner clone exiting). The bus + # connection is shared and owned by the lifespan, so this is a no-op; + # the connection is torn down once via ``drain()`` on shutdown. + return None + + async def drain(self) -> None: + """Drain + close the shared NATS connection (lifespan shutdown only).""" + try: + await self._nc.drain() + except Exception: + logger.warning("NATS drain on shutdown failed", exc_info=True) diff --git a/nextcloud_mcp_server/vector/queue/ports.py b/nextcloud_mcp_server/vector/queue/ports.py new file mode 100644 index 00000000..978bad4c --- /dev/null +++ b/nextcloud_mcp_server/vector/queue/ports.py @@ -0,0 +1,59 @@ +"""Ingest-path ports (design §10, hexagonal). + +A ``TaskProducer`` is where the scanner + webhook receiver send a +``DocumentTask``. The transport behind it is swappable: + +- the in-process anyio ``MemoryObjectSendStream`` (local ingest — the default), +- ``NatsTaskProducer`` (external ingest → the document-processor), and +- a future Postgres-queue producer (seam only; the *external* processor owns the + consume side — see ``postgres.py``). + +The protocol is exactly the surface the scanner/oauth_sync already use on the +memory stream (``send`` + ``clone`` + ``async with``), so both adapters drop in +with only a type-annotation change at the call sites. There is intentionally NO +consumer port: the MCP server's only in-process consumer is the memory stream; +when ingest is external the document-processor is the consumer, not this server. +""" + +from __future__ import annotations + +from types import TracebackType +from typing import TYPE_CHECKING, Protocol, runtime_checkable + +if TYPE_CHECKING: + from ..scanner import DocumentTask + + +@runtime_checkable +class TaskProducer(Protocol): + """Sink for scanner/webhook ``DocumentTask``s (see module docstring).""" + + # Positional-only so anyio's MemoryObjectSendStream.send(item) structurally + # satisfies this protocol (its parameter is named "item", not "task"). + async def send(self, task: DocumentTask, /) -> None: ... + + def clone(self) -> TaskProducer: + """Return a producer handle for one user's scanner (multi-user mode). + + For the memory stream this is a real clone (each closed independently); + for the bus it returns ``self`` (one shared connection). + """ + ... + + async def __aenter__(self) -> TaskProducer: ... + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + tb: TracebackType | None, + ) -> None: ... + + async def aclose(self) -> None: + """Close *this* handle (e.g. a per-user clone when its scanner exits). + + For the memory stream this closes the clone; for the shared bus + connection it is a no-op (the connection is owned by the lifespan, + which drains it once on shutdown). + """ + ... diff --git a/nextcloud_mcp_server/vector/queue/postgres.py b/nextcloud_mcp_server/vector/queue/postgres.py new file mode 100644 index 00000000..526265b7 --- /dev/null +++ b/nextcloud_mcp_server/vector/queue/postgres.py @@ -0,0 +1,49 @@ +"""Postgres-queue ``TaskProducer`` — documented seam, not implemented. + +The external document-processor may later drain a Postgres-backed queue instead +of NATS to limit NATS operational overhead. Processing stays *external*; only +the transport changes — so on this server it would be a drop-in producer swap. +The consume side + the queue-table migration belong to that processor-side +refactor (cross-repo), NOT here. This stub exists so the transport value and the +``TaskProducer`` Protocol conformance are testable today. +""" + +from __future__ import annotations + +from types import TracebackType +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from ..scanner import DocumentTask + +_NOT_IMPLEMENTED = ( + "Postgres ingest transport is a documented seam. The external " + "document-processor owns the Postgres-drain refactor (transport swap only; " + "processing stays external). Use INGEST_BUS_URL=nats://… for now." +) + + +class PostgresTaskProducer: + @classmethod + async def connect(cls, settings: Any) -> PostgresTaskProducer: + raise NotImplementedError(_NOT_IMPLEMENTED) + + async def send(self, task: DocumentTask) -> None: # pragma: no cover + raise NotImplementedError(_NOT_IMPLEMENTED) + + def clone(self) -> PostgresTaskProducer: # pragma: no cover + return self + + async def __aenter__(self) -> PostgresTaskProducer: # pragma: no cover + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + tb: TracebackType | None, + ) -> None: # pragma: no cover + return None + + async def aclose(self) -> None: # pragma: no cover + return None diff --git a/nextcloud_mcp_server/vector/queue/status.py b/nextcloud_mcp_server/vector/queue/status.py new file mode 100644 index 00000000..010dc4de --- /dev/null +++ b/nextcloud_mcp_server/vector/queue/status.py @@ -0,0 +1,163 @@ +"""Status surface for ingest jobs (design §10.1, ``STATUS_BACKEND``). + +- ``local``: in-process job state — the memory-stream buffer (today's behavior, + read directly by the status endpoint). +- ``bus``: a background subscriber consumes + ``mcp.document.{ready,failed,reparsed}.{tenant_id}`` into a bounded in-process + :class:`StatusStore` that the status endpoint / ``nc_get_vector_sync_status`` + read. + +**Honest constraint (design §10.2 / decision):** MCP progress notifications +(``ctx.report_progress``) can only be emitted inside an *active tool-call +request*; a background subscriber has no ``ctx`` and the MCP SDK exposes no +out-of-band push. So "surface events as MCP progress notifications" is delivered +via this store (polled by the status endpoint / a tool), not an unsolicited +server push. True server-initiated progress / SSE is a follow-up — the +``on_event`` callback seam is left in place for it. +""" + +from __future__ import annotations + +import logging +from collections import OrderedDict +from typing import TYPE_CHECKING, Any, Callable + +if TYPE_CHECKING: + import anyio + from anyio.abc import TaskStatus + +logger = logging.getLogger(__name__) + +# Terminal/intermediate document states carried on mcp.document.* subjects. +_VALID_STATES = {"ready", "failed", "reparsed"} + + +class StatusStore: + """Bounded LRU of recent document states keyed by ``doc_id``.""" + + def __init__(self, max_size: int = 10_000): + self._entries: OrderedDict[str, dict[str, Any]] = OrderedDict() + self._max = max_size + + def record( + self, + doc_id: str, + state: str, + *, + content_hash: str | None = None, + transitioned_at: str | None = None, + ) -> None: + self._entries[doc_id] = { + "state": state, + "content_hash": content_hash, + "transitioned_at": transitioned_at, + } + self._entries.move_to_end(doc_id) + while len(self._entries) > self._max: + self._entries.popitem(last=False) + + def get(self, doc_id: str) -> dict[str, Any] | None: + return self._entries.get(doc_id) + + def counts(self) -> dict[str, int]: + out: dict[str, int] = {} + for entry in self._entries.values(): + out[entry["state"]] = out.get(entry["state"], 0) + 1 + return out + + def __len__(self) -> int: + return len(self._entries) + + +def state_from_subject(subject: str) -> str | None: + """``mcp.document..`` → ```` (or None if unknown).""" + parts = subject.split(".") + if len(parts) >= 4 and parts[0] == "mcp" and parts[1] == "document": + state = parts[2] + if state in _VALID_STATES: + return state + return None + + +class NatsStatusSubscriber: + """Consumes ``mcp.document.*.{tenant_id}`` into a :class:`StatusStore`.""" + + def __init__( + self, + nc: Any, + js: Any, + tenant_id: str, + store: StatusStore, + on_event: Callable[[str, str], None] | None = None, + ): + self._nc = nc + self._js = js + self.tenant_id = tenant_id + self.store = store + # on_event(doc_id, state) — seam for a future SSE / progress bridge. + self._on_event = on_event + + def handle_message(self, subject: str, data: bytes) -> None: + """Parse one status message into the store. Unit-testable without NATS.""" + import json # noqa: PLC0415 + + state = state_from_subject(subject) + if state is None: + logger.warning("status.unknown_subject subject=%s", subject) + return + try: + payload = json.loads(data) + doc_id = payload["doc_id"] + except Exception: + logger.warning("status.bad_message subject=%s", subject, exc_info=True) + return + self.store.record( + doc_id, + state, + content_hash=payload.get("content_hash"), + transitioned_at=payload.get("transitioned_at"), + ) + if self._on_event is not None: + self._on_event(doc_id, state) + + @classmethod + async def connect( + cls, *, url: str, tenant_id: str, store: StatusStore + ) -> NatsStatusSubscriber: + import nats # noqa: PLC0415 + + nc = await nats.connect(url) + js = nc.jetstream() + return cls(nc, js, tenant_id, store) + + async def run( + self, + shutdown_event: anyio.Event, + *, + task_status: TaskStatus = None, # type: ignore[assignment] + ) -> None: + """Durable pull-consumer loop. Requires a live broker (integration).""" + import anyio # noqa: PLC0415 + + subject = f"mcp.document.*.{self.tenant_id}" + sub = await self._js.pull_subscribe( + subject, durable=f"mcp-status-{self.tenant_id}" + ) + if task_status is not None: + task_status.started() + while not shutdown_event.is_set(): + try: + msgs = await sub.fetch(batch=16, timeout=5) + except Exception: + # fetch timeout when idle — loop and re-check shutdown. + await anyio.sleep(0) + continue + for msg in msgs: + self.handle_message(msg.subject, msg.data) + await msg.ack() + + async def aclose(self) -> None: + try: + await self._nc.drain() + except Exception: + logger.warning("NATS status subscriber drain failed", exc_info=True) diff --git a/nextcloud_mcp_server/vector/scanner.py b/nextcloud_mcp_server/vector/scanner.py index d14bbcec..eb00ebd8 100644 --- a/nextcloud_mcp_server/vector/scanner.py +++ b/nextcloud_mcp_server/vector/scanner.py @@ -13,7 +13,6 @@ from typing import cast import anyio from anyio.abc import TaskStatus -from anyio.streams.memory import MemoryObjectSendStream from qdrant_client import AsyncQdrantClient from qdrant_client.models import FieldCondition, Filter, MatchValue, Record @@ -31,6 +30,7 @@ from nextcloud_mcp_server.vector.placeholder import ( write_placeholder_point, ) from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client +from nextcloud_mcp_server.vector.queue.ports import TaskProducer logger = logging.getLogger(__name__) @@ -108,6 +108,11 @@ class DocumentTask: metadata: dict[str, int | str] | None = ( None # Additional metadata (e.g., board_id/stack_id for deck_card) ) + # Change-detection token (Nextcloud etag). Used as the external ingest + # content_hash when present; the bus producer falls back to modified_at + # when it is None (deletes, or sources whose etag isn't threaded). Harmless + # in local mode — the in-process processor reads its own etag. + etag: str | None = None # Track documents potentially deleted (grace period before actual deletion) @@ -179,7 +184,7 @@ async def get_last_indexed_timestamp(user_id: str) -> int | None: async def scanner_task( - send_stream: MemoryObjectSendStream[DocumentTask], + send_stream: TaskProducer, shutdown_event: anyio.Event, wake_event: anyio.Event, nc_client: NextcloudClient, @@ -233,7 +238,7 @@ async def scanner_task( async def scan_user_documents( user_id: str, - send_stream: MemoryObjectSendStream[DocumentTask], + send_stream: TaskProducer, nc_client: NextcloudClient, initial_sync: bool = False, ): @@ -339,6 +344,7 @@ async def scan_user_documents( doc_type="note", operation="index", modified_at=modified_at, + etag=note.get("etag", ""), ) ) queued += 1 @@ -405,6 +411,7 @@ async def scan_user_documents( doc_type="note", operation="index", modified_at=modified_at, + etag=note.get("etag", ""), ) ) queued += 1 @@ -740,7 +747,7 @@ async def scan_user_documents( async def scan_news_items( user_id: str, - send_stream: MemoryObjectSendStream[DocumentTask], + send_stream: TaskProducer, nc_client: NextcloudClient, initial_sync: bool, scan_id: int, @@ -928,7 +935,7 @@ async def scan_news_items( async def scan_deck_cards( user_id: str, - send_stream: MemoryObjectSendStream[DocumentTask], + send_stream: TaskProducer, nc_client: NextcloudClient, initial_sync: bool, scan_id: int, diff --git a/nextcloud_mcp_server/vector/webhook_receiver.py b/nextcloud_mcp_server/vector/webhook_receiver.py index d55d666c..c1cb50fa 100644 --- a/nextcloud_mcp_server/vector/webhook_receiver.py +++ b/nextcloud_mcp_server/vector/webhook_receiver.py @@ -41,8 +41,9 @@ def _warn_missing_secret_once() -> None: async def handle_nextcloud_webhook(request: Request) -> JSONResponse: """Receive a Nextcloud webhook and queue a DocumentTask for vector sync. - Returns quickly so NC's webhook worker is not blocked. The send-stream is - read from ``request.app.state.document_send_stream``; when vector sync + Returns quickly so NC's webhook worker is not blocked. The task producer is + read from ``request.app.state.task_producer`` (the in-memory send stream in + local mode, or the NATS bus producer in external mode); when vector sync isn't running we return 503 so NC retries delivery. When ``WEBHOOK_SECRET`` is set, the request must carry @@ -93,8 +94,8 @@ async def handle_nextcloud_webhook(request: Request) -> JSONResponse: status_code=200, ) - send_stream = getattr(request.app.state, "document_send_stream", None) - if send_stream is None: + producer = getattr(request.app.state, "task_producer", None) + if producer is None: logger.warning( "Webhook received but vector sync is not running; rejecting so NC retries" ) @@ -105,7 +106,7 @@ async def handle_nextcloud_webhook(request: Request) -> JSONResponse: try: with anyio.fail_after(1.0): - await send_stream.send(task) + await producer.send(task) except TimeoutError: # Queue is saturated (default 10 000 tasks). Returning 503 lets NC # retry rather than pinning this handler until its outbound timeout diff --git a/pyproject.toml b/pyproject.toml index a210ae60..0aabcac8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -46,6 +46,7 @@ dependencies = [ "dynaconf>=3.2.13,<4.0", "mistralai>=2.4.5", "sqlalchemy[asyncio]>=2.0", + "nats-py>=2.14.0", ] classifiers = [ "Development Status :: 4 - Beta", diff --git a/tests/fixtures/acl_hash_corpus.json b/tests/fixtures/acl_hash_corpus.json new file mode 100644 index 00000000..9369a596 --- /dev/null +++ b/tests/fixtures/acl_hash_corpus.json @@ -0,0 +1,285 @@ +{ + "_comment": "Cross-impl ACL hash corpus (design §11.5). Pinned BLAKE2b-128 values; both repos must reproduce. Do not edit by hand.", + "cases": [ + { + "name": "single_user_ascii", + "share_set": [ + [ + "user", + "alice" + ] + ], + "expected": [ + "c63a5fa2d6df5c20ee3700dd85e39e1d" + ] + }, + { + "name": "single_group", + "share_set": [ + [ + "group", + "admins" + ] + ], + "expected": [ + "fe57bf2fd0682f713ff967073808b2c1" + ] + }, + { + "name": "public_only", + "share_set": [ + [ + "public", + "public" + ] + ], + "expected": [ + "97990046391027e6ec24575a3ebdd136" + ] + }, + { + "name": "user_group_public", + "share_set": [ + [ + "user", + "alice" + ], + [ + "group", + "admins" + ], + [ + "public", + "public" + ] + ], + "expected": [ + "c63a5fa2d6df5c20ee3700dd85e39e1d", + "fe57bf2fd0682f713ff967073808b2c1", + "97990046391027e6ec24575a3ebdd136" + ] + }, + { + "name": "user_uppercase_distinct", + "share_set": [ + [ + "user", + "Alice" + ] + ], + "expected": [ + "245e3e55068fbb623b397a8ca21b0126" + ] + }, + { + "name": "unicode_precomposed", + "share_set": [ + [ + "user", + "café" + ] + ], + "expected": [ + "053451c238fa2c6118b95b2ab2d7964a" + ] + }, + { + "name": "unicode_decomposed_same_as_precomposed", + "share_set": [ + [ + "user", + "café" + ] + ], + "expected": [ + "053451c238fa2c6118b95b2ab2d7964a" + ] + }, + { + "name": "pipe_in_username", + "share_set": [ + [ + "user", + "a|b" + ] + ], + "expected": [ + "a5f7e378a36ed9dd4e8e02007bca3624" + ] + }, + { + "name": "colon_in_username", + "share_set": [ + [ + "user", + "a:b" + ] + ], + "expected": [ + "7049ec93c0bf28a5f02a8527434dd0a0" + ] + }, + { + "name": "newline_in_username", + "share_set": [ + [ + "user", + "a\nb" + ] + ], + "expected": [ + "8e62aee620abaf9852e471d603cbe88b" + ] + }, + { + "name": "backslash_in_username", + "share_set": [ + [ + "user", + "a\\b" + ] + ], + "expected": [ + "c16a30d33ef2f834ebb642496206467e" + ] + }, + { + "name": "quote_in_username", + "share_set": [ + [ + "user", + "a\"b" + ] + ], + "expected": [ + "61405bcb3db7afd0fbf13693d2157bbb" + ] + }, + { + "name": "space_in_group", + "share_set": [ + [ + "group", + "Dept 7" + ] + ], + "expected": [ + "e80d92f6227d1ae64a8c8231491335d8" + ] + }, + { + "name": "emoji_username", + "share_set": [ + [ + "user", + "🦏fox" + ] + ], + "expected": [ + "52156572976858eec6361800b2f5ad03" + ] + }, + { + "name": "cjk_group", + "share_set": [ + [ + "group", + "管理员" + ] + ], + "expected": [ + "a3e5a6e8b405d5c749a031595309491c" + ] + }, + { + "name": "email_style_user", + "share_set": [ + [ + "user", + "jane.doe@example.com" + ] + ], + "expected": [ + "3216fecd6203036e6c23ff9c055031d9" + ] + }, + { + "name": "numeric_group", + "share_set": [ + [ + "group", + "12345" + ] + ], + "expected": [ + "1ae023f071c8d70bacdb388230fc2cfb" + ] + }, + { + "name": "multi_group_large", + "share_set": [ + [ + "user", + "bob" + ], + [ + "group", + "g1" + ], + [ + "group", + "g2" + ], + [ + "group", + "g3" + ], + [ + "group", + "g4" + ], + [ + "group", + "g5" + ] + ], + "expected": [ + "830bb4f705bc0aa13929061db587b696", + "4eac98d1b59fec00e15e4fe28a32a3f9", + "2c76357c3f95284911f6be1a30d4850e", + "898271cc3d99563bf8dd396ef6c0f0b2", + "4704c80af47f56c566cf5b83f506e704", + "80bcc464007a51518f87876758c6ccf5" + ] + }, + { + "name": "group_and_public", + "share_set": [ + [ + "group", + "staff" + ], + [ + "public", + "public" + ] + ], + "expected": [ + "b2d6c7e2a6bc94f2b0310492ac6316fc", + "97990046391027e6ec24575a3ebdd136" + ] + }, + { + "name": "trailing_space_user_distinct", + "share_set": [ + [ + "user", + "alice " + ] + ], + "expected": [ + "d7f07f4624e3e06b7f0b77f483e9941e" + ] + } + ] +} diff --git a/tests/fixtures/ingest_message_example.json b/tests/fixtures/ingest_message_example.json new file mode 100644 index 00000000..c15c20cc --- /dev/null +++ b/tests/fixtures/ingest_message_example.json @@ -0,0 +1,10 @@ +{ + "tenant_id": "00000000-0000-0000-0000-000000000001", + "doc_id": "12345", + "content_hash": "etag-abc123", + "modified_at": "2026-05-27T00:00:00Z", + "doc_type": "file", + "operation": "index", + "user_id": "alice", + "file_path": "/Documents/report.pdf" +} diff --git a/tests/fixtures/namespace_uuid.txt b/tests/fixtures/namespace_uuid.txt new file mode 100644 index 00000000..b6ac3e36 --- /dev/null +++ b/tests/fixtures/namespace_uuid.txt @@ -0,0 +1 @@ +b050b8ac-c6aa-5566-9584-506b39c1c096 diff --git a/tests/unit/providers/test_gateway_provider.py b/tests/unit/providers/test_gateway_provider.py new file mode 100644 index 00000000..f0d43467 --- /dev/null +++ b/tests/unit/providers/test_gateway_provider.py @@ -0,0 +1,117 @@ +"""Gateway provider registration + M2M OIDC auth (design §10.2). + +The gateway is manual-only: selected by EMBEDDING_PROVIDER=gateway and never by +the autodetect chain. Auth is the gateway's own M2M OIDC realm (parallel to the +tenant realm); creds are all-or-nothing. +""" + +import time + +import httpx +import pytest + +from nextcloud_mcp_server.config import Settings +from nextcloud_mcp_server.embedding.gateway_client import ( + GatewayProvider, + GatewayTokenProvider, +) +from nextcloud_mcp_server.providers.registry import ProviderRegistry, reset_provider +from nextcloud_mcp_server.providers.simple import SimpleProvider + + +def _patch_settings(monkeypatch, settings): + monkeypatch.setattr( + "nextcloud_mcp_server.providers.registry.get_settings", lambda: settings + ) + reset_provider() + + +def test_gateway_selected_unauthenticated(monkeypatch): + settings = Settings( + embedding_provider="gateway", + embedding_gateway_url="http://gateway:8083", + embedding_gateway_model="mistral-embed", + ) + _patch_settings(monkeypatch, settings) + provider = ProviderRegistry.create_provider() + assert isinstance(provider, GatewayProvider) + assert provider.embedding_model == "mistral-embed" + assert provider.supports_embeddings is True + assert provider.supports_generation is False + assert provider._token_provider is None # unauthenticated + + +def test_gateway_selected_with_m2m_oidc(monkeypatch): + settings = Settings( + embedding_provider="gateway", + embedding_gateway_url="http://gateway:8083", + embedding_gateway_token_url="https://idp.example/oauth2/token", + embedding_gateway_client_id="mcp-server", + embedding_gateway_client_secret="shh", + embedding_gateway_scope="astrolabe-embedding-gateway/embed", + ) + _patch_settings(monkeypatch, settings) + provider = ProviderRegistry.create_provider() + assert isinstance(provider, GatewayProvider) + assert isinstance(provider._token_provider, GatewayTokenProvider) + + +def test_partial_m2m_creds_rejected(): + with pytest.raises(ValueError, match="must be set together"): + Settings( + embedding_provider="gateway", + embedding_gateway_url="http://gateway:8083", + embedding_gateway_client_id="mcp-server", # missing token_url/secret + ) + + +def test_autodetect_default_does_not_pick_gateway(monkeypatch): + settings = Settings() + _patch_settings(monkeypatch, settings) + assert isinstance(ProviderRegistry.create_provider(), SimpleProvider) + + +def test_openai_creds_do_not_trigger_gateway(monkeypatch): + settings = Settings(openai_api_key="sk-test") + _patch_settings(monkeypatch, settings) + assert not isinstance(ProviderRegistry.create_provider(), GatewayProvider) + + +async def test_token_provider_caches_and_refreshes(monkeypatch): + calls = {"n": 0} + + def handler(request: httpx.Request) -> httpx.Response: + calls["n"] += 1 + assert request.headers["Authorization"].startswith("Basic ") + body = dict(httpx.QueryParams(request.content.decode())) + assert body["grant_type"] == "client_credentials" + assert body["scope"] == "embed" + return httpx.Response( + 200, json={"access_token": f"tok{calls['n']}", "expires_in": 3600} + ) + + transport = httpx.MockTransport(handler) + orig_async_client = httpx.AsyncClient + + def _client(*args, **kwargs): + kwargs["transport"] = transport + return orig_async_client(*args, **kwargs) + + monkeypatch.setattr(httpx, "AsyncClient", _client) + + tp = GatewayTokenProvider( + token_url="https://idp.example/oauth2/token", + client_id="cid", + client_secret="sec", + scope="embed", + ) + t1 = await tp.get_token() + t2 = await tp.get_token() # cached → no new HTTP call + assert t1 == t2 == "tok1" + assert calls["n"] == 1 + + # Expire the cache → next call refreshes. + tp._cache = (tp._cache[0], time.time() - 1) + t3 = await tp.get_token() + assert t3 == "tok2" + assert calls["n"] == 2 diff --git a/tests/unit/test_acl_hash.py b/tests/unit/test_acl_hash.py new file mode 100644 index 00000000..c1d1ac35 --- /dev/null +++ b/tests/unit/test_acl_hash.py @@ -0,0 +1,71 @@ +"""ACL hash spec tests (design §11) + cross-impl corpus. + +The corpus fixture is checked into both repos; the processor runs a mirror of +``test_corpus`` against the same file. Divergence is caught in CI. +""" + +import json +from pathlib import Path + +import pytest + +from nextcloud_mcp_server.acl_hash import ( + PUBLIC_PRINCIPAL, + accessible_hash_set, + compute_acl_hash, + compute_principal_hash, +) + +CORPUS = json.loads( + (Path(__file__).parent.parent / "fixtures" / "acl_hash_corpus.json").read_text( + encoding="utf-8" + ) +) + + +@pytest.mark.parametrize( + "case", CORPUS["cases"], ids=[c["name"] for c in CORPUS["cases"]] +) +def test_corpus(case): + share_set = [tuple(p) for p in case["share_set"]] + assert compute_acl_hash(share_set) == case["expected"] + + +def test_blake2b_128_bit_length(): + assert len(compute_principal_hash("user", "alice")) == 32 + + +def test_case_is_preserved_not_folded(): + assert compute_principal_hash("user", "Alice") != compute_principal_hash( + "user", "alice" + ) + + +def test_nfc_normalization(): + # decomposed 'café' (e + U+0301) hashes the same as precomposed 'café'. + assert compute_principal_hash("user", "café") == compute_principal_hash( + "user", "café" + ) + + +def test_invalid_principal_type_rejected(): + with pytest.raises(ValueError, match="principal_type must be one of"): + compute_principal_hash("link", "sometoken") + + +def test_accessible_set_always_includes_public(): + s = accessible_hash_set("alice", []) + assert compute_principal_hash(*PUBLIC_PRINCIPAL) in s + + +def test_accessible_set_membership_matches_share(): + # A document shared with a group the requester holds is admitted. + doc = compute_acl_hash([("group", "engineering")]) + accessible = accessible_hash_set("bob", ["engineering", "all-staff"]) + assert any(h in accessible for h in doc) + + +def test_accessible_set_excludes_unheld_group(): + doc = compute_acl_hash([("group", "finance")]) + accessible = accessible_hash_set("bob", ["engineering"]) + assert not any(h in accessible for h in doc) diff --git a/tests/unit/test_decomposition_config.py b/tests/unit/test_decomposition_config.py new file mode 100644 index 00000000..d6c1e016 --- /dev/null +++ b/tests/unit/test_decomposition_config.py @@ -0,0 +1,133 @@ +"""Tests for the MCP decomposition hook-point settings (design §10). + +Every default must reproduce the monolith; the opt-in settings are validated +in ``Settings.__post_init__``. +""" + +import pytest + +from nextcloud_mcp_server.canonical import canonical_json +from nextcloud_mcp_server.config import Settings + + +class TestDecompositionDefaults: + """With nothing set, behavior matches today's monolith.""" + + def test_defaults_are_monolith(self): + s = Settings() + assert s.embedding_provider == "autodetect" + assert s.ingest_mode == "local" + assert s.status_backend == "local" + assert s.collection_metadata_source == "qdrant" + assert s.fact_event_emitter == "none" + assert s.ingest_bus_url is None + assert s.embedding_gateway_url is None + assert s.tenant_id is None + assert s.ingest_bus_num_replicas == 1 + + def test_enum_values_normalized(self): + # Mixed case / surrounding whitespace is normalized before validation. + s = Settings( + collection_metadata_source=" QDRANT ", + fact_event_emitter="NONE", + ) + assert s.collection_metadata_source == "qdrant" + assert s.fact_event_emitter == "none" + + +class TestEnumValidation: + @pytest.mark.parametrize( + "field,value", + [ + ("embedding_provider", "openai"), + ("ingest_mode", "remote"), + ("status_backend", "redis"), + ("collection_metadata_source", "postgres"), + ("fact_event_emitter", "kafka"), + ], + ) + def test_invalid_enum_rejected(self, field, value): + with pytest.raises(ValueError, match=field.upper()): + Settings(**{field: value}) + + +class TestFailFast: + def test_external_with_local_status_crashes(self): + with pytest.raises( + RuntimeError, + match="STATUS_BACKEND=local is incompatible with INGEST_MODE=external", + ): + Settings( + ingest_mode="external", + status_backend="local", + ingest_bus_url="nats://nats:4222", + tenant_id="tenant-uuid", + ) + + +class TestConditionalRequired: + def test_external_requires_bus_url(self): + with pytest.raises(ValueError, match="INGEST_BUS_URL is required"): + Settings(ingest_mode="external", status_backend="bus", tenant_id="t1") + + def test_external_requires_tenant_id(self): + with pytest.raises(ValueError, match="TENANT_ID is required"): + Settings( + ingest_mode="external", + status_backend="bus", + ingest_bus_url="nats://nats:4222", + ) + + def test_gateway_requires_gateway_url(self): + with pytest.raises(ValueError, match="EMBEDDING_GATEWAY_URL is required"): + Settings(embedding_provider="gateway") + + def test_external_happy_path(self): + s = Settings( + ingest_mode="external", + status_backend="bus", + ingest_bus_url="nats://nats:4222", + tenant_id="0a1b2c3d-0000-0000-0000-000000000000", + ) + assert s.ingest_mode == "external" + assert s.status_backend == "bus" + + def test_gateway_happy_path(self): + s = Settings( + embedding_provider="gateway", + embedding_gateway_url="http://gateway:8083", + ) + assert s.embedding_provider == "gateway" + + +class TestTenantIdSubjectToken: + @pytest.mark.parametrize( + "tenant_id", + ["a.b", "a*b", "a>b", "a b", "a\tb"], + ) + def test_illegal_subject_chars_rejected(self, tenant_id): + with pytest.raises(ValueError, match="TENANT_ID must not contain"): + Settings(tenant_id=tenant_id) + + def test_uuid_form_accepted(self): + s = Settings(tenant_id="0a1b2c3d-0000-0000-0000-000000000000") + assert s.tenant_id == "0a1b2c3d-0000-0000-0000-000000000000" + + +class TestReplicas: + def test_zero_replicas_rejected(self): + with pytest.raises(ValueError, match="INGEST_BUS_NUM_REPLICAS must be >= 1"): + Settings(ingest_bus_num_replicas=0) + + +class TestCanonicalJson: + def test_sorted_keys_no_whitespace(self): + assert canonical_json({"b": 1, "a": 2}) == b'{"a":2,"b":1}' + + def test_non_ascii_preserved(self): + # ensure_ascii=False keeps the literal UTF-8 bytes. + assert canonical_json({"k": "café"}) == '{"k":"café"}'.encode("utf-8") + + def test_stable_across_calls(self): + obj = {"tenant_id": "t", "doc_id": "d", "modified_at": "2026-01-01T00:00:00Z"} + assert canonical_json(obj) == canonical_json(dict(reversed(list(obj.items())))) diff --git a/tests/unit/test_payload_backfill.py b/tests/unit/test_payload_backfill.py new file mode 100644 index 00000000..eed2c05a --- /dev/null +++ b/tests/unit/test_payload_backfill.py @@ -0,0 +1,56 @@ +"""Admin payload-backfill endpoint (design §10.2).""" + +import json + +from nextcloud_mcp_server.api.management import AdminScopeRequired +from nextcloud_mcp_server.config import Settings + + +def _request(mocker): + return mocker.MagicMock() + + +async def test_backfill_requires_admin_scope(mocker): + from nextcloud_mcp_server.admin import payload_backfill as mod + + mocker.patch.object( + mod, "require_admin_scope", side_effect=AdminScopeRequired("nope") + ) + resp = await mod.handle_payload_backfill(_request(mocker)) + assert resp.status_code == 403 + + +async def test_backfill_unauthorized_on_auth_error(mocker): + from nextcloud_mcp_server.admin import payload_backfill as mod + + mocker.patch.object(mod, "require_admin_scope", side_effect=ValueError("no token")) + resp = await mod.handle_payload_backfill(_request(mocker)) + assert resp.status_code == 401 + + +async def test_backfill_happy_path_sets_keys_and_sentinel(mocker): + from nextcloud_mcp_server.admin import payload_backfill as mod + + mocker.patch.object(mod, "require_admin_scope", return_value="admin") + mocker.patch.object( + mod, "get_settings", return_value=Settings(vector_sync_enabled=True) + ) + qdrant = mocker.AsyncMock() + mocker.patch.object(mod, "get_qdrant_client", return_value=qdrant) + embed = mocker.MagicMock() + embed.get_dimension.return_value = 4 + mocker.patch( + "nextcloud_mcp_server.embedding.get_embedding_service", return_value=embed + ) + # upsert_sentinel uses the same qdrant client; it is an AsyncMock so .upsert + # is awaitable. Patch upsert_sentinel to assert it was invoked. + sentinel = mocker.patch.object(mod, "upsert_sentinel", new=mocker.AsyncMock()) + + resp = await mod.handle_payload_backfill(_request(mocker)) + assert resp.status_code == 200 + body = json.loads(resp.body) + assert body["status"] == "ok" + # processor_version, pipeline_tier, embedding_identity → 3 set_payload calls. + assert qdrant.set_payload.await_count == 3 + sentinel.assert_awaited_once() + assert body["sentinel_upserted"] is True diff --git a/tests/unit/test_webhook_endpoint.py b/tests/unit/test_webhook_endpoint.py index 800958bd..42b396e3 100644 --- a/tests/unit/test_webhook_endpoint.py +++ b/tests/unit/test_webhook_endpoint.py @@ -42,7 +42,10 @@ def _make_app(send_stream=None) -> Starlette: Route("/webhooks/nextcloud", handle_nextcloud_webhook, methods=["POST"]) ] ) + # The webhook reads app.state.task_producer; a raw MemoryObjectSendStream + # satisfies the TaskProducer.send contract directly. app.state.document_send_stream = send_stream + app.state.task_producer = send_stream return app diff --git a/tests/unit/vector/test_collection_metadata.py b/tests/unit/vector/test_collection_metadata.py new file mode 100644 index 00000000..bd55fb1c --- /dev/null +++ b/tests/unit/vector/test_collection_metadata.py @@ -0,0 +1,91 @@ +"""Collection metadata source + sentinel (design §10.1).""" + +from types import SimpleNamespace + +import httpx + +from nextcloud_mcp_server.config import Settings +from nextcloud_mcp_server.vector import collection_metadata as cm +from nextcloud_mcp_server.vector.payload_keys import EMBEDDING_IDENTITY + + +async def test_qdrant_read_hit(mocker): + client = mocker.AsyncMock() + client.retrieve.return_value = [ + SimpleNamespace( + payload={ + EMBEDDING_IDENTITY: "mistral-embed", + cm.CHUNKING_CONFIG: {"chunk_size": 1024, "chunk_overlap": 100}, + cm.IS_SENTINEL: True, + } + ) + ] + meta = await cm.read_collection_metadata(client, "col", Settings()) + assert meta["embedding_identity"] == "mistral-embed" + assert meta["chunking_config"]["chunk_size"] == 1024 + + +async def test_qdrant_miss_falls_back_to_env(mocker): + client = mocker.AsyncMock() + client.retrieve.return_value = [] # no sentinel + settings = Settings(document_chunk_size=2048, document_chunk_overlap=200) + meta = await cm.read_collection_metadata(client, "col", settings) + assert meta == cm.env_default_metadata(settings) + assert meta["chunking_config"]["chunk_size"] == 2048 + + +async def test_qdrant_error_falls_back_to_env(mocker): + client = mocker.AsyncMock() + client.retrieve.side_effect = RuntimeError("qdrant down") + settings = Settings() + meta = await cm.read_collection_metadata(client, "col", settings) + assert meta == cm.env_default_metadata(settings) + + +async def test_api_source(mocker): + settings = Settings( + collection_metadata_source="api", + collection_metadata_api_url="http://cp", + ) + + def handler(request: httpx.Request) -> httpx.Response: + assert request.url.path == "/v1/qdrant-collections/col/metadata" + return httpx.Response( + 200, + json={ + "embedding_identity": "amazon.titan-embed-text-v2:0", + "chunking_config": {"chunk_size": 512, "chunk_overlap": 50}, + }, + ) + + transport = httpx.MockTransport(handler) + orig = httpx.AsyncClient + mocker.patch.object( + httpx, + "AsyncClient", + lambda *a, **k: orig(*a, **{**k, "transport": transport}), + ) + + meta = await cm.read_collection_metadata(mocker.AsyncMock(), "col", settings) + assert meta["embedding_identity"] == "amazon.titan-embed-text-v2:0" + + +async def test_upsert_sentinel_builds_point(mocker): + client = mocker.AsyncMock() + await cm.upsert_sentinel( + client, + "col", + embedding_identity="mistral-embed", + chunking_config={"chunk_size": 2048, "chunk_overlap": 200}, + dimension=4, + ) + client.upsert.assert_awaited_once() + kwargs = client.upsert.await_args.kwargs + assert kwargs["collection_name"] == "col" + point = kwargs["points"][0] + assert str(point.id) == cm.SENTINEL_POINT_ID + # Non-zero dense (cosine-safe), empty sparse. + assert point.vector["dense"][0] != 0.0 + assert len(point.vector["dense"]) == 4 + assert point.payload[EMBEDDING_IDENTITY] == "mistral-embed" + assert point.payload[cm.IS_SENTINEL] is True diff --git a/tests/unit/vector/test_nats_producer.py b/tests/unit/vector/test_nats_producer.py new file mode 100644 index 00000000..2fa0edfd --- /dev/null +++ b/tests/unit/vector/test_nats_producer.py @@ -0,0 +1,136 @@ +"""NATS ingest producer: DocumentTask → IngestMessage + dedup header (§3.4).""" + +import hashlib +import json +from pathlib import Path + +import pytest + +from nextcloud_mcp_server.canonical import canonical_json +from nextcloud_mcp_server.vector.queue.factory import _transport_for +from nextcloud_mcp_server.vector.queue.nats import ( + NatsTaskProducer, + _modified_at_rfc3339, + msg_id, +) +from nextcloud_mcp_server.vector.queue.postgres import PostgresTaskProducer +from nextcloud_mcp_server.vector.scanner import DocumentTask + +FIXTURE = Path(__file__).parents[2] / "fixtures" / "ingest_message_example.json" +TENANT = "00000000-0000-0000-0000-000000000001" + + +def _producer(mocker, tenant_id=TENANT): + return NatsTaskProducer( + nc=mocker.MagicMock(), js=mocker.AsyncMock(), tenant_id=tenant_id + ) + + +def test_ingest_message_translation(mocker): + p = _producer(mocker) + task = DocumentTask( + user_id="alice", + doc_id="12345", + doc_type="file", + operation="index", + modified_at=1700000000, + file_path="/Documents/report.pdf", + etag="etag-abc123", + ) + msg = p.ingest_message(task) + assert msg["tenant_id"] == TENANT # from settings, not the task + assert msg["content_hash"] == "etag-abc123" # etag wins + assert msg["user_id"] == "alice" + assert msg["doc_type"] == "file" + assert msg["operation"] == "index" + assert msg["file_path"] == "/Documents/report.pdf" + + +def test_content_hash_falls_back_to_modified_at(mocker): + p = _producer(mocker) + task = DocumentTask( + user_id="u", doc_id="d", doc_type="note", operation="delete", modified_at=0 + ) + assert p.ingest_message(task)["content_hash"] == "0" + + +async def test_send_publishes_with_dedup_header(mocker): + p = _producer(mocker) + task = DocumentTask( + user_id="alice", + doc_id="12345", + doc_type="file", + operation="index", + modified_at=1700000000, + etag="e", + ) + await p.send(task) + p._js.publish.assert_awaited_once() + args = p._js.publish.await_args.args + kwargs = p._js.publish.await_args.kwargs + assert args[0] == f"mcp.ingest.requested.{TENANT}" + expected_mid = msg_id(TENANT, "12345", _modified_at_rfc3339(1700000000)) + assert kwargs["headers"]["Nats-Msg-Id"] == expected_mid + assert json.loads(args[1])["doc_id"] == "12345" + + +def test_msg_id_known_vector(): + mid = msg_id("t", "d", "2026-01-01T00:00:00+00:00") + expected = hashlib.sha256( + canonical_json( + { + "tenant_id": "t", + "doc_id": "d", + "modified_at": "2026-01-01T00:00:00+00:00", + } + ) + ).hexdigest() + assert mid == expected + + +def test_publisher_matches_shared_fixture(mocker): + # The same fixture is validated as an IngestMessage in the processor repo. + # Here we assert the publisher emits exactly the fixture's key set + stable + # field values (modified_at format is allowed to differ — epoch→ISO). + fixture = json.loads(FIXTURE.read_text(encoding="utf-8")) + p = _producer(mocker, tenant_id=fixture["tenant_id"]) + task = DocumentTask( + user_id=fixture["user_id"], + doc_id=fixture["doc_id"], + doc_type=fixture["doc_type"], + operation=fixture["operation"], + modified_at=1764201600, + file_path=fixture["file_path"], + etag=fixture["content_hash"], + ) + msg = p.ingest_message(task) + assert set(msg.keys()) == set(fixture.keys()) + for key in ( + "tenant_id", + "doc_id", + "content_hash", + "doc_type", + "operation", + "user_id", + "file_path", + ): + assert msg[key] == fixture[key] + assert msg["modified_at"] # non-empty ISO timestamp + + +@pytest.mark.parametrize( + "url,expected", + [ + ("nats://nats:4222", "nats"), + ("postgres://h/db", "postgres"), + ("postgresql://h/db", "postgres"), + ("http://elsewhere", "nats"), + ], +) +def test_transport_for(url, expected): + assert _transport_for(url) == expected + + +async def test_postgres_producer_is_a_seam(): + with pytest.raises(NotImplementedError, match="documented seam"): + await PostgresTaskProducer.connect(object()) diff --git a/tests/unit/vector/test_payload_keys.py b/tests/unit/vector/test_payload_keys.py new file mode 100644 index 00000000..ce7a415b --- /dev/null +++ b/tests/unit/vector/test_payload_keys.py @@ -0,0 +1,44 @@ +"""Drift guard for the cross-implementation point-ID namespace (design §2.2). + +The MCP server and the external document-processor must compute identical chunk +point IDs. This test pins the NAMESPACE against a fixture checked into both +repos; the processor repo runs a mirror of this test against the same fixture. +""" + +import uuid +from pathlib import Path + +from nextcloud_mcp_server.vector import payload_keys + +FIXTURE = Path(__file__).parents[2] / "fixtures" / "namespace_uuid.txt" + + +def test_namespace_matches_fixture(): + assert str(payload_keys.NAMESPACE) == FIXTURE.read_text().strip() + + +def test_point_id_deterministic(): + a = payload_keys.point_id("t", "d", 0) + assert a == payload_keys.point_id("t", "d", 0) + assert a != payload_keys.point_id("t", "d", 1) + assert a != payload_keys.point_id("t2", "d", 0) + + +def test_point_id_exact_value(): + # Pins the canonical-JSON name + uuid5 formula so a refactor that changes + # key ordering or encoding is caught (would silently break idempotency). + expected = str( + uuid.uuid5( + payload_keys.NAMESPACE, + '{"chunk_index":0,"doc_id":"d","tenant_id":"t"}', + ) + ) + assert payload_keys.point_id("t", "d", 0) == expected + + +def test_payload_key_constants(): + assert payload_keys.EMBEDDING_IDENTITY == "embedding_identity" + assert payload_keys.ACL_HASH == "acl_hash" + assert payload_keys.PROCESSOR_VERSION == "processor_version" + assert payload_keys.PARSED_AT == "parsed_at" + assert payload_keys.PIPELINE_TIER == "pipeline_tier" diff --git a/tests/unit/vector/test_status_store.py b/tests/unit/vector/test_status_store.py new file mode 100644 index 00000000..59c98588 --- /dev/null +++ b/tests/unit/vector/test_status_store.py @@ -0,0 +1,70 @@ +"""StatusStore + NATS status message handling (design §10.1, STATUS_BACKEND=bus).""" + +import json + +from nextcloud_mcp_server.vector.queue.status import ( + NatsStatusSubscriber, + StatusStore, + state_from_subject, +) + + +def test_store_records_and_counts(): + store = StatusStore() + store.record("d1", "ready", content_hash="h1") + store.record("d2", "failed") + store.record("d1", "ready", content_hash="h1") # idempotent overwrite + assert len(store) == 2 + assert store.counts() == {"ready": 1, "failed": 1} + assert store.get("d1")["content_hash"] == "h1" + + +def test_store_is_bounded_lru(): + store = StatusStore(max_size=2) + store.record("d1", "ready") + store.record("d2", "ready") + store.record("d3", "ready") # evicts d1 + assert len(store) == 2 + assert store.get("d1") is None + assert store.get("d3") is not None + + +def test_state_from_subject(): + assert state_from_subject("mcp.document.ready.tenant-1") == "ready" + assert state_from_subject("mcp.document.failed.tenant-1") == "failed" + assert state_from_subject("mcp.document.reparsed.tenant-1") == "reparsed" + assert state_from_subject("mcp.document.bogus.tenant-1") is None + assert state_from_subject("mcp.ingest.requested.tenant-1") is None + + +def test_handle_message_records_state(): + store = StatusStore() + events = [] + sub = NatsStatusSubscriber( + nc=None, + js=None, + tenant_id="t1", + store=store, + on_event=lambda d, s: events.append((d, s)), + ) + payload = json.dumps( + { + "tenant_id": "t1", + "doc_id": "doc-9", + "content_hash": "abc", + "transitioned_at": "2026-05-27T00:00:00Z", + } + ).encode() + sub.handle_message("mcp.document.ready.t1", payload) + entry = store.get("doc-9") + assert entry["state"] == "ready" + assert entry["content_hash"] == "abc" + assert events == [("doc-9", "ready")] + + +def test_handle_message_ignores_bad_payload_and_subject(): + store = StatusStore() + sub = NatsStatusSubscriber(nc=None, js=None, tenant_id="t1", store=store) + sub.handle_message("mcp.document.ready.t1", b"not json") + sub.handle_message("mcp.ingest.requested.t1", b'{"doc_id":"x"}') + assert len(store) == 0 diff --git a/uv.lock b/uv.lock index b097f068..af78eab8 100644 --- a/uv.lock +++ b/uv.lock @@ -2169,6 +2169,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/6c/28/dd72947e59a6a8c856448a5e74da6201cb5502ddff644fbc790e4bd40b9a/multiprocess-0.70.18-py39-none-any.whl", hash = "sha256:e78ca805a72b1b810c690b6b4cc32579eba34f403094bbbae962b7b5bf9dfcb8", size = 133478, upload-time = "2025-04-17T03:11:26.253Z" }, ] +[[package]] +name = "nats-py" +version = "2.14.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/c3/f8/b956c4621ba88748ed707c52e69f95b7a50c8914e750edca59a5bef84a76/nats_py-2.14.0.tar.gz", hash = "sha256:4ed02cb8e3b55c68074a063aa2687087115d805d1513297da90cb2068fb07bed", size = 120751, upload-time = "2026-02-23T22:44:58.988Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/f9/39/0e87753df1072254bac190b33ed34b264f28f6aa9bea0f01b7e818071756/nats_py-2.14.0-py3-none-any.whl", hash = "sha256:4116f5d2233ce16e63c3d5538fa40a5e207f75fcf42a741773929ddf1e29d19d", size = 82259, upload-time = "2026-02-23T22:45:00.152Z" }, +] + [[package]] name = "nextcloud-mcp-server" version = "0.89.0" @@ -2190,6 +2199,7 @@ dependencies = [ { name = "markdownify" }, { name = "mcp", extra = ["cli"] }, { name = "mistralai" }, + { name = "nats-py" }, { name = "openai" }, { name = "opentelemetry-api" }, { name = "opentelemetry-exporter-otlp-proto-grpc" }, @@ -2252,6 +2262,7 @@ requires-dist = [ { name = "markdownify", specifier = ">=0.14.1" }, { name = "mcp", extras = ["cli"], specifier = ">=1.27,<1.28" }, { name = "mistralai", specifier = ">=2.4.5" }, + { name = "nats-py", specifier = ">=2.14.0" }, { name = "openai", specifier = ">=2.8.1" }, { name = "opentelemetry-api", specifier = ">=1.28.2" }, { name = "opentelemetry-exporter-otlp-proto-grpc", specifier = ">=1.28.2" },