fix(ocr): round-2 review — lazy store lock, mode enum normalization, type hints

Round 2 review (PR #910):
- BLOCKING: BatchOcrJobStore._shared_lock is now lazy-init (anyio.Lock | None,
  created on first shared() call) instead of at class-definition time — matches
  the CLAUDE.md "no anyio primitives at import time" rule and OcrProcessor's
  pattern. The None-check->assign has no await between, so it's race-free.
- document_ocr_mode now normalizes via _enum_fields (case-insensitive, like
  document_ocr_provider) instead of a strict dynaconf is_in Validator, so
  DOCUMENT_OCR_MODE=Batch normalizes to "batch" rather than erroring. Tests for
  case-normalization + invalid-value rejection.
- TYPE_CHECKING-gated GatewayBatchOcrClient import so build_gateway_batch_client
  / _get_batch_client are typed `GatewayBatchOcrClient | None` instead of Any
  (runtime import stays lazy to avoid the import cycle).
- Rename ocr_options -> doc_identity_options (it's threaded to all tiers; only
  OCR reads it) + clarify the comment.
- Drop the redundant forward-ref quotes on _shared_instance.
- Add direct _batch_identity unit tests (partial/empty options branches).

Left as follow-up: reusing one httpx.AsyncClient across submit/poll (same
per-call pattern as the existing sync _GatewayOcrBackend; no clean aclose hook
on the cached client today).

1653 unit tests pass; ruff + ty green.

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

nextcloud_mcp_server/config.py

@@ -366,7 +366,10 @@ _dynaconf = Dynaconf(
         Validator("DOCUMENT_CHUNK_SIZE", gte=1),
         Validator("DOCUMENT_PARSE_TIMEOUT_SECONDS", gte=1),
         Validator("DOCUMENT_OCR_TIMEOUT_SECONDS", gte=1),
-        Validator("DOCUMENT_OCR_MODE", is_in=("sync", "batch")),
+        # DOCUMENT_OCR_MODE is normalised + membership-checked in
+        # Settings.__post_init__ via _enum_fields (case-insensitive, like
+        # DOCUMENT_OCR_PROVIDER) — no strict dynaconf Validator here, so
+        # "Batch"/"SYNC" normalise instead of erroring.
         # Poll cadence well above a few seconds (each poll re-runs the tier);
         # deadline at least one poll interval.
         Validator("DOCUMENT_OCR_BATCH_POLL_SECONDS", gte=5),
@@ -1029,6 +1032,7 @@ class Settings:
             "collection_metadata_source": {"qdrant", "api"},
             "document_tier1_engine": {"pypdfium2", "pymupdf"},
             "document_ocr_provider": {"auto", "gateway", "mistral", "none"},
+            "document_ocr_mode": {"sync", "batch"},
         }
         for _field, _allowed in _enum_fields.items():
             _val = (getattr(self, _field) or "").strip().lower()

nextcloud_mcp_server/document_processors/ocr.py

@@ -21,7 +21,7 @@ import logging
 import time
 from abc import ABC, abstractmethod
 from collections.abc import Awaitable, Callable
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import anyio
 import httpx
@@ -30,6 +30,12 @@ from nextcloud_mcp_server.config import Settings, get_settings
 
 from .base import DocumentProcessor, ProcessingResult
 
+if TYPE_CHECKING:
+    # Annotation-only import (the runtime import is lazy, inside
+    # build_gateway_batch_client, to avoid a document_processors -> embedding
+    # cycle at load).
+    from ..embedding.gateway_batch_client import GatewayBatchOcrClient
+
 logger = logging.getLogger(__name__)
 
 # Connect timeout for the OCR backend request. The overall (read) timeout is
@@ -197,7 +203,7 @@ def _build_gateway_token_provider(settings: Settings) -> Any:
     )
 
 
-def build_gateway_batch_client(settings: Settings) -> Any:
+def build_gateway_batch_client(settings: Settings) -> "GatewayBatchOcrClient | None":
     """Build a ``GatewayBatchOcrClient`` when the gateway is the OCR backend, else
     ``None`` (so batch mode falls back to sync for provider=mistral / no gateway).
     Batch OCR is gateway-only — Mistral's Batch API is reached *through* the
@@ -373,7 +379,7 @@ class OcrProcessor(DocumentProcessor):
             processor=self.name,
         )
 
-    async def _get_batch_client(self) -> Any:
+    async def _get_batch_client(self) -> "GatewayBatchOcrClient | None":
         """Cached gateway batch client (or ``None`` when batch isn't applicable —
         provider=mistral / no gateway). Resolved once under the backend lock so the
         token provider's M2M cache survives across documents."""

nextcloud_mcp_server/vector/batch_ocr_store.py

@@ -38,18 +38,24 @@ class BatchOcrJob:
 class BatchOcrJobStore:
     """CRUD for the ``batch_ocr_jobs`` table (one row per in-flight job)."""
 
-    _shared_instance: "BatchOcrJobStore | None" = None
+    _shared_instance: BatchOcrJobStore | None = None
-    _shared_lock: anyio.Lock = anyio.Lock()
+    # Lazy-init: anyio primitives must not be created at import time (CLAUDE.md;
+    # mirrors OcrProcessor._backend_lock). Created on first shared() call.
+    _shared_lock: anyio.Lock | None = None
 
     def __init__(self, storage: RefreshTokenStorage) -> None:
         self._storage = storage
 
     @classmethod
-    async def shared(cls) -> "BatchOcrJobStore":
+    async def shared(cls) -> BatchOcrJobStore:
-        """Process-wide store backed by the storage singleton (mirrors
+        """Process-wide store backed by the storage singleton. Tests should
-        ``UsageEventStore.shared``). Tests should construct
+        construct ``BatchOcrJobStore(storage)`` directly — the cache is a process
-        ``BatchOcrJobStore(storage)`` directly — the cache is a process global
+        global with no teardown hook."""
-        with no teardown hook."""
+        # No await between the None-check and the assignment, so this is atomic
+        # within the single event loop (anyio is cooperative) — two cold-start
+        # callers can't both create a lock.
+        if cls._shared_lock is None:
+            cls._shared_lock = anyio.Lock()
         async with cls._shared_lock:
             if cls._shared_instance is None:
                 cls._shared_instance = cls(await get_shared_storage())

nextcloud_mcp_server/vector/processor.py

@@ -984,10 +984,11 @@ async def _index_document(
                 # and the in-process/memory pool (tier is None) -- runs the inline
                 # tiered pipeline (fast -> OCR escalation in one call).
                 if tier is not None and _is_pdf(content_type):
-                    # Thread per-document identity to the OCR tier so batch mode
+                    # Per-document identity, forwarded to every tier's processor.
-                    # (Deck #332) can key its job-tracking table; other tiers
+                    # Only the OCR tier reads it (batch mode keys its job-tracking
-                    # ignore it.
+                    # table on it, Deck #332); fast/structured ignore it, so it's
-                    ocr_options = {
+                    # safe to pass on all tiers.
+                    doc_identity_options = {
                         "user_id": doc_task.user_id,
                         "doc_id": doc_task.doc_id,
                         "doc_type": doc_task.doc_type,
@@ -1000,7 +1001,7 @@ async def _index_document(
                         file_path,
                         tier,
                         settings,
-                        options=ocr_options,
+                        options=doc_identity_options,
                     )
                 else:
                     result = await registry.process(

tests/unit/test_config.py

@@ -137,6 +137,19 @@ class TestGetSettings:
         assert settings.document_ocr_batch_poll_seconds == 45
         assert settings.document_ocr_batch_max_wait_seconds == 3600
 
+    @patch.dict(os.environ, {"DOCUMENT_OCR_MODE": "Batch"}, clear=True)
+    def test_document_ocr_mode_case_normalised(self):
+        """DOCUMENT_OCR_MODE is case-insensitive (normalised in __post_init__ via
+        _enum_fields, like DOCUMENT_OCR_PROVIDER) — "Batch" -> "batch"."""
+        _reload_config()
+        assert get_settings().document_ocr_mode == "batch"
+
+    @patch.dict(os.environ, {"DOCUMENT_OCR_MODE": "bogus"}, clear=True)
+    def test_document_ocr_mode_invalid_rejected(self):
+        _reload_config()
+        with pytest.raises(ValueError, match="DOCUMENT_OCR_MODE"):
+            get_settings()
+
     @patch.dict(
         os.environ,
         {"QDRANT_LOCATION": "/app/data/qdrant"},

tests/unit/test_ocr_processor.py

@@ -225,6 +225,35 @@ async def test_mistral_backend_applies_timeout(mocker, monkeypatch):
 
 # --- batch mode (Deck #332) --------------------------------------------------
 
+
+@pytest.mark.parametrize(
+    "options",
+    [
+        None,
+        {},
+        {"doc_id": "d", "doc_type": "file"},  # missing user_id
+        {"user_id": "u", "doc_type": "file"},  # missing doc_id
+        {"user_id": "u", "doc_id": "d"},  # missing doc_type
+        {"user_id": "u", "doc_id": "d", "doc_type": ""},  # empty doc_type
+    ],
+)
+def test_batch_identity_returns_none_without_full_identity(options):
+    assert ocr._batch_identity(options) is None
+
+
+def test_batch_identity_extracts_tuple_and_defaults_etag():
+    assert ocr._batch_identity(
+        {"user_id": "u", "doc_id": "d", "doc_type": "file", "etag": "v1"}
+    ) == ("u", "d", "file", "v1")
+    # etag may be absent/empty -> normalised to "".
+    assert ocr._batch_identity({"user_id": "u", "doc_id": "d", "doc_type": "file"}) == (
+        "u",
+        "d",
+        "file",
+        "",
+    )
+
+
 from nextcloud_mcp_server.embedding.gateway_batch_client import (  # noqa: E402
     BatchPollResult,
 )