feat: add opt-in MCP decomposition hook points (design §10)

Adds the seven §10.2 hook-point modules + five env vars so Astrolabe Cloud can
offload document processing to the external document-processor / embedding
gateway. Purely additive: with every setting unset the server behaves exactly
as today, so self-hosters are unaffected (Deck #92).

Hook points (all default to current monolith behavior):
- config: EMBEDDING_PROVIDER, INGEST_MODE, STATUS_BACKEND,
  COLLECTION_METADATA_SOURCE, FACT_EVENT_EMITTER (+ supporting settings),
  validated in Settings.__post_init__ (fail-fast STATUS_BACKEND=local with
  INGEST_MODE=external); shared canonical.py.
- vector/payload_keys.py + acl_hash.py: cross-impl NAMESPACE/point_id (§2.2)
  and BLAKE2b-128 ACL hash (§11), pinned by fixtures shared with the
  document-processor repo.
- embedding/gateway_client.py: OpenAI-compatible GatewayProvider authenticating
  via M2M OIDC client-credentials (separate realm); manual-only registry entry.
- vector/collection_metadata.py: sentinel-point / API metadata source with env
  fallback.
- vector/queue/: hexagonal ingest producer ports + memory/NATS adapters
  (Postgres seam); INGEST_MODE=external publishes mcp.ingest.requested.{tenant}
  instead of the in-memory stream and skips the in-process processor pool. The
  lifespan becomes a composition root across both deployment branches.
- vector/queue/status.py: STATUS_BACKEND=bus subscriber feeding a StatusStore
  the vector-sync status endpoint reads.
- admin/payload_backfill.py: POST /api/v1/admin/payload-backfill (admin scope);
  processor writes the new payload keys; query-side ACL pre-filter gated behind
  ACL_PREFILTER_ENABLED (default off).

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

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

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",

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))

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.)
                     **(
                         {

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"]

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,
+    )

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()

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)

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).
+        """
+        ...

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

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.<state>.<tenant_id>`` → ``<state>`` (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)

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,

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