diff --git a/CHANGELOG.md b/CHANGELOG.md index b9b52c4b..b89ba129 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,34 @@ All notable changes to the Nextcloud MCP Server will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [PEP 440](https://peps.python.org/pep-0440/). +## v0.113.0 (2026-06-11) + +### Feat + +- **document**: configurable OCR timeout and fail-fast PDF size guard + +### Fix + +- **document**: catch httpx timeout from gateway OCR backend (#892 r3) +- **document**: timeout reason bucket + Sonar https hotspot (#892 round 2) +- **document**: apply OCR timeout to Mistral backend + review/Sonar fixes (#892) +- **vector**: guard unbound doc_task + address review nits (#891) +- **vector**: URL-encode DAV paths and unwrap TaskGroup exceptions + +## v0.112.0 (2026-06-11) + +### Feat + +- **worker**: structured logs + metrics + traces for ingest worker + +### Fix + +- **worker**: clear Sonar S5332 hotspot + address review nits + +### Refactor + +- **worker**: trim observability helper docstring; clarify test fake + ## v0.111.0 (2026-06-10) ### Feat diff --git a/docs/configuration.md b/docs/configuration.md index 18223d71..cc928444 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -536,6 +536,23 @@ DOCUMENT_CHUNK_OVERLAP=200 # Overlapping characters between chunks (d > **Note:** The `VECTOR_SYNC_*` tuning parameters keep their names as they're implementation details. Only the user-facing feature flag was renamed to `ENABLE_SEMANTIC_SEARCH`. +#### Document parsing robustness (PDF) + +These guard the parse/OCR tiers against pathological PDFs. Defaults are safe; +tune per tenant when a corpus has very large scans or a gateway with its own +shorter OCR ceiling: + +```dotenv +DOCUMENT_PARSE_TIMEOUT_SECONDS=120 # Wall-clock cap per isolated parse (default: 120) +DOCUMENT_OCR_TIMEOUT_SECONDS=180 # OCR backend request timeout (default: 180) +DOCUMENT_MAX_PDF_SIZE_MB=50 # Pre-parse size cap; 0 disables (default: 50) +``` + +A PDF larger than `DOCUMENT_MAX_PDF_SIZE_MB` fails fast with reason `oversize` +(exported on `astrolabe_document_parse_failed_total{reason="oversize"}`) instead +of being handed to the tiers, where a 40+ MB scan would otherwise burn the full +OCR timeout for zero recovered text. + ### Embedding Service Configuration The server picks an embedding provider via auto-detection. Priority order diff --git a/nextcloud_mcp_server/cli.py b/nextcloud_mcp_server/cli.py index aa714930..9f1d2283 100644 --- a/nextcloud_mcp_server/cli.py +++ b/nextcloud_mcp_server/cli.py @@ -6,6 +6,7 @@ import click import uvicorn from nextcloud_mcp_server.config import ( + Settings, get_database_url, get_settings, is_ephemeral_token_db, @@ -17,7 +18,12 @@ from nextcloud_mcp_server.migrations import ( show_migration_history, upgrade_database, ) -from nextcloud_mcp_server.observability import get_uvicorn_logging_config +from nextcloud_mcp_server.observability import ( + get_uvicorn_logging_config, + setup_logging, + setup_metrics, + setup_tracing, +) from nextcloud_mcp_server.server import AVAILABLE_APPS from .app import get_app @@ -284,6 +290,41 @@ def run( ) +def _init_worker_observability(settings: Settings) -> None: + """Configure logging, metrics, and tracing for the standalone ingest worker.""" + # Mirrors app.py's lifespan bootstrap; without it the worker's astrolabe_* + # metrics and document_processor.parse spans are invisible in external mode. + # Structured logging first, so every subsequent startup line is JSON like + # the API's — the worker entrypoint never went through uvicorn's log_config. + setup_logging( + log_format=settings.log_format, + log_level=settings.log_level, + include_trace_context=settings.log_include_trace_context, + ) + + if settings.metrics_enabled: + setup_metrics(port=settings.metrics_port) + logger.info( + "Prometheus metrics enabled on dedicated port %s", settings.metrics_port + ) + + if settings.otel_exporter_otlp_endpoint: + setup_tracing( + service_name=settings.otel_service_name, + otlp_endpoint=settings.otel_exporter_otlp_endpoint, + otlp_verify_ssl=settings.otel_exporter_verify_ssl, + sampling_rate=settings.otel_traces_sampler_arg, + ) + logger.info( + "OpenTelemetry tracing enabled (endpoint: %s)", + settings.otel_exporter_otlp_endpoint, + ) + else: + logger.info( + "OpenTelemetry tracing disabled (set OTEL_EXPORTER_OTLP_ENDPOINT to enable)" + ) + + @click.command() @click.option( "--concurrency", @@ -319,6 +360,11 @@ def worker(concurrency: int | None): f"resolved INGEST_QUEUE={settings.ingest_queue!r}" ) + # Initialize observability here, not in a lifespan — the worker never runs + # uvicorn, so it skips app.py's bootstrap (the WHY lives in the helper's + # docstring). Done after the queue check so a misconfig fails fast. + _init_worker_observability(settings) + from nextcloud_mcp_server.vector.queue.procrastinate import ( # noqa: PLC0415 INGEST_QUEUE_NAME, apply_ingest_queue_schema, diff --git a/nextcloud_mcp_server/client/webdav.py b/nextcloud_mcp_server/client/webdav.py index bceafacb..19b654d4 100644 --- a/nextcloud_mcp_server/client/webdav.py +++ b/nextcloud_mcp_server/client/webdav.py @@ -5,7 +5,7 @@ import mimetypes import xml.etree.ElementTree as ET from email.utils import parsedate_to_datetime from typing import Any, Dict, List, Optional, Tuple -from urllib.parse import unquote +from urllib.parse import quote, unquote from xml.sax.saxutils import escape as xml_escape from httpx import HTTPStatusError @@ -26,11 +26,42 @@ WEBDAV_SEARCH_PAGE_SIZE = 500 WEBDAV_SEARCH_MAX_RESULTS = 50000 +def _encode_dav_path(path: str) -> str: + """Percent-encode a *decoded* DAV path for use in a request URL/header. + + Paths flow through this client already URL-decoded (e.g. ``unquote`` on the + ```` of a PROPFIND/REPORT response, or raw user-supplied paths from + MCP tools), so characters like ``#``, ``,`` and spaces reach httpx verbatim. + An unencoded ``#`` is parsed as a URL fragment and silently truncates the + request path → spurious 404 on otherwise-valid files (issue: OHR-Bench + ingest, card 309). ``quote`` with ``safe="/"`` encodes the unsafe characters + while preserving the path separators; ASCII-only paths are unchanged. + + Encode exactly once: the input is decoded, so a literal ``%`` becomes + ``%25`` (correct) rather than being mistaken for an existing escape. + """ + return quote(path, safe="/") + + class WebDAVClient(BaseNextcloudClient): """Client for Nextcloud WebDAV operations.""" app_name = "webdav" + def _webdav_path(self, path: str) -> str: + """Build the request path for ``path`` under the user's DAV root. + + Percent-encodes the caller-supplied portion (see ``_encode_dav_path``) + so names with ``#``, commas, or spaces don't truncate/404; the base + ``/remote.php/dav/files/`` segment is left as-is. + + Precondition: ``path`` is a **decoded** path (the convention everywhere + in this client — PROPFIND/REPORT hrefs are ``unquote``d before storage, + and MCP-tool inputs are raw). It is encoded exactly once, so passing an + already-encoded path would double-encode it (``%20`` → ``%2520``). + """ + return f"{self._get_webdav_base_path()}/{_encode_dav_path(path.lstrip('/'))}" + async def delete_resource(self, path: str) -> Dict[str, Any]: """Delete a resource (file or directory) via WebDAV DELETE.""" # Ensure path ends with a slash if it's a directory @@ -39,7 +70,7 @@ class WebDAVClient(BaseNextcloudClient): else: path_with_slash = path - webdav_path = f"{self._get_webdav_base_path()}/{path_with_slash.lstrip('/')}" + webdav_path = self._webdav_path(path_with_slash) logger.debug("Deleting WebDAV resource: %s", webdav_path) headers = {"OCS-APIRequest": "true"} @@ -124,15 +155,15 @@ class WebDAVClient(BaseNextcloudClient): mime_type: Optional[str] = None, ) -> Dict[str, Any]: """Add/Update an attachment to a note via WebDAV PUT.""" - # Construct paths based on provided category - webdav_base = self._get_webdav_base_path() + # Construct paths based on provided category. Encode via _webdav_path so + # categories/filenames with '#', commas or spaces don't truncate/404. category_path_part = f"{category}/" if category else "" attachment_dir_segment = f".attachments.{note_id}" parent_dir_webdav_rel_path = ( f"Notes/{category_path_part}{attachment_dir_segment}" ) - parent_dir_path = f"{webdav_base}/{parent_dir_webdav_rel_path}" - attachment_path = f"{parent_dir_path}/{filename}" + parent_dir_path = self._webdav_path(parent_dir_webdav_rel_path) + attachment_path = self._webdav_path(f"{parent_dir_webdav_rel_path}/{filename}") logger.debug("Uploading attachment '%s' for note %s", filename, note_id) @@ -144,7 +175,7 @@ class WebDAVClient(BaseNextcloudClient): headers = {"Content-Type": mime_type, "OCS-APIRequest": "true"} try: # First check if we can access WebDAV at all - notes_dir_path = f"{webdav_base}/Notes" + notes_dir_path = self._webdav_path("Notes") propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"} notes_dir_response = await self._make_request( "PROPFIND", notes_dir_path, headers=propfind_headers @@ -209,10 +240,11 @@ class WebDAVClient(BaseNextcloudClient): self, note_id: int, filename: str, category: Optional[str] = None ) -> Tuple[bytes, str]: """Fetch a specific attachment from a note via WebDAV GET.""" - webdav_base = self._get_webdav_base_path() category_path_part = f"{category}/" if category else "" attachment_dir_segment = f".attachments.{note_id}" - attachment_path = f"{webdav_base}/Notes/{category_path_part}{attachment_dir_segment}/{filename}" + attachment_path = self._webdav_path( + f"Notes/{category_path_part}{attachment_dir_segment}/{filename}" + ) logger.debug("Fetching attachment '%s' for note %s", filename, note_id) @@ -252,7 +284,7 @@ class WebDAVClient(BaseNextcloudClient): async def list_directory(self, path: str = "") -> List[Dict[str, Any]]: """List files and directories in the specified path via WebDAV PROPFIND.""" - webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}" + webdav_path = self._webdav_path(path) if not webdav_path.endswith("/"): webdav_path += "/" @@ -352,7 +384,7 @@ class WebDAVClient(BaseNextcloudClient): async def read_file(self, path: str) -> Tuple[bytes, str]: """Read a file's content via WebDAV GET.""" - webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}" + webdav_path = self._webdav_path(path) logger.debug("Reading file: %s", path) @@ -379,7 +411,7 @@ class WebDAVClient(BaseNextcloudClient): self, path: str, content: bytes, content_type: Optional[str] = None ) -> Dict[str, Any]: """Write content to a file via WebDAV PUT.""" - webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}" + webdav_path = self._webdav_path(path) logger.debug("Writing file: %s", path) @@ -410,7 +442,7 @@ class WebDAVClient(BaseNextcloudClient): self, path: str, recursive: bool = False ) -> Dict[str, Any]: """Create a directory via WebDAV MKCOL.""" - webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}" + webdav_path = self._webdav_path(path) if not webdav_path.endswith("/"): webdav_path += "/" @@ -468,10 +500,8 @@ class WebDAVClient(BaseNextcloudClient): Returns: Dict with status_code and optional message """ - source_webdav_path = f"{self._get_webdav_base_path()}/{source_path.lstrip('/')}" - destination_webdav_path = ( - f"{self._get_webdav_base_path()}/{destination_path.lstrip('/')}" - ) + source_webdav_path = self._webdav_path(source_path) + destination_webdav_path = self._webdav_path(destination_path) # Ensure paths have consistent trailing slashes for directories if source_path.endswith("/") and not destination_path.endswith("/"): @@ -552,10 +582,8 @@ class WebDAVClient(BaseNextcloudClient): Returns: Dict with status_code and optional message """ - source_webdav_path = f"{self._get_webdav_base_path()}/{source_path.lstrip('/')}" - destination_webdav_path = ( - f"{self._get_webdav_base_path()}/{destination_path.lstrip('/')}" - ) + source_webdav_path = self._webdav_path(source_path) + destination_webdav_path = self._webdav_path(destination_path) # Ensure paths have consistent trailing slashes for directories if source_path.endswith("/") and not destination_path.endswith("/"): @@ -1678,7 +1706,7 @@ class WebDAVClient(BaseNextcloudClient): distinguish a definitive absence (HTTP 404) from a brittle response (None). """ - webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}" + webdav_path = self._webdav_path(path) propfind_body = """ diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index 43330fdf..3afe22ce 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -146,6 +146,10 @@ _DEFAULTS: dict[str, Any] = { "document_pdf_graphics_limit": 1000, "document_parse_timeout_seconds": 120.0, "document_parse_mem_limit_mb": 1536, + # Pre-parse size cap (MB): PDFs larger than this fail fast with reason + # "oversize" instead of burning the OCR timeout to 0 chars on a pathological + # file. 0 disables the guard. + "document_max_pdf_size_mb": 50.0, # Tier-0 classifier (records classification metrics on the tiered path) "document_classify_enabled": True, # Tiered PDF pipeline: pypdfium2 is the default/only hot-path extractor; @@ -168,6 +172,10 @@ _DEFAULTS: dict[str, Any] = { "document_ocr_page_fraction": 0.5, "document_ocr_min_page_chars": 16, "document_ocr_detect_scanned": True, + # OCR backend request timeout (seconds). Slow scanned newspapers can take + # 20-60s; raise/lower per tenant. Configurable so a tenant isn't stuck with + # the 180s default when its gateway has its own shorter ceiling. + "document_ocr_timeout_seconds": 180.0, # Observability "metrics_enabled": True, "metrics_port": 9090, @@ -319,7 +327,10 @@ _dynaconf = Dynaconf( Validator("VERIFICATION_CONCURRENCY", gte=1), Validator("DOCUMENT_CHUNK_SIZE", gte=1), Validator("DOCUMENT_PARSE_TIMEOUT_SECONDS", gte=1), + Validator("DOCUMENT_OCR_TIMEOUT_SECONDS", gte=1), Validator("DOCUMENT_PARSE_MEM_LIMIT_MB", gte=128), + # 0 disables the pre-parse PDF size cap; otherwise it must be positive. + Validator("DOCUMENT_MAX_PDF_SIZE_MB", gte=0), # >=1: pymupdf4llm treats graphics_limit=0 as "no cap", which would # re-expose the OOM this guards against. Validator("DOCUMENT_PDF_GRAPHICS_LIMIT", gte=1), @@ -782,6 +793,11 @@ class Settings: # float so a fractional DOCUMENT_PARSE_TIMEOUT_SECONDS is honoured, matching # anyio.move_on_after's float seconds. document_parse_timeout_seconds: float = 120.0 + # Pre-parse PDF size cap (MB). A PDF larger than this fails fast with + # parse_failed_reason="oversize" (placeholder marked "failed") rather than + # being handed to the fast/OCR tiers, where a pathological large file burns + # the OCR timeout for 0 chars. 0 disables the guard. + document_max_pdf_size_mb: float = 50.0 # RLIMIT_AS in the parse subprocess (below the pod limit). Applied once per # worker for its lifetime, so changing it needs a pod restart. document_parse_mem_limit_mb: int = 1536 @@ -801,6 +817,10 @@ class Settings: # gateway routes on the "/" prefix; the direct mistral backend # strips it. document_ocr_model: str = "mistral/mistral-ocr-latest" + # OCR backend HTTP request timeout (seconds). float for parity with the + # parse timeout / httpx.Timeout; per-tenant tunable so a gateway with a + # shorter ceiling isn't masked by the 180s default. + document_ocr_timeout_seconds: float = 180.0 # OCR escalation triggers (tier-0), per-tenant tunable. A page is OCR-worthy # if near-empty (< min_page_chars) OR low text-quality (< min_text_quality) # OR (when detect_scanned, image-analysis only runs when OCR is enabled) @@ -1431,12 +1451,14 @@ def get_settings() -> Settings: "document_chunk_page_aware": "DOCUMENT_CHUNK_PAGE_AWARE", "document_pdf_graphics_limit": "DOCUMENT_PDF_GRAPHICS_LIMIT", "document_parse_timeout_seconds": "DOCUMENT_PARSE_TIMEOUT_SECONDS", + "document_max_pdf_size_mb": "DOCUMENT_MAX_PDF_SIZE_MB", "document_parse_mem_limit_mb": "DOCUMENT_PARSE_MEM_LIMIT_MB", "document_classify_enabled": "DOCUMENT_CLASSIFY_ENABLED", "document_tier1_engine": "DOCUMENT_TIER1_ENGINE", "document_ocr_enabled": "DOCUMENT_OCR_ENABLED", "document_ocr_provider": "DOCUMENT_OCR_PROVIDER", "document_ocr_model": "DOCUMENT_OCR_MODEL", + "document_ocr_timeout_seconds": "DOCUMENT_OCR_TIMEOUT_SECONDS", "document_ocr_min_text_quality": "DOCUMENT_OCR_MIN_TEXT_QUALITY", "document_ocr_page_fraction": "DOCUMENT_OCR_PAGE_FRACTION", "document_ocr_min_page_chars": "DOCUMENT_OCR_MIN_PAGE_CHARS", diff --git a/nextcloud_mcp_server/document_processors/ocr.py b/nextcloud_mcp_server/document_processors/ocr.py index 6178ab6c..19ad4673 100644 --- a/nextcloud_mcp_server/document_processors/ocr.py +++ b/nextcloud_mcp_server/document_processors/ocr.py @@ -31,7 +31,9 @@ from .base import DocumentProcessor, ProcessingResult logger = logging.getLogger(__name__) -_OCR_TIMEOUT_SECONDS = 180.0 +# Connect timeout for the OCR backend request. The overall (read) timeout is +# configurable via DOCUMENT_OCR_TIMEOUT_SECONDS and resolved per call. +_OCR_CONNECT_TIMEOUT_SECONDS = 10.0 def _pages_to_text( @@ -93,8 +95,12 @@ class _GatewayOcrBackend(_OcrBackend): "document_b64": base64.b64encode(content).decode("ascii"), "mime_type": mime_type, } + # Resolved per call (get_settings builds fresh) so test monkeypatching is + # honoured; a live tenant change still needs a restart because the backend + # instance itself is cached for the pod's lifetime. + ocr_timeout = get_settings().document_ocr_timeout_seconds async with httpx.AsyncClient( - timeout=httpx.Timeout(_OCR_TIMEOUT_SECONDS, connect=10.0) + timeout=httpx.Timeout(ocr_timeout, connect=_OCR_CONNECT_TIMEOUT_SECONDS) ) as client: resp = await client.post(self._url, json=payload, headers=headers) resp.raise_for_status() @@ -120,10 +126,17 @@ class _MistralOcrBackend(_OcrBackend): data_url = ( f"data:{mime_type};base64,{base64.b64encode(content).decode('ascii')}" ) - resp = await self._client.ocr.process_async( - model=self._model, - document={"type": "document_url", "document_url": data_url}, - ) + # Apply DOCUMENT_OCR_TIMEOUT_SECONDS uniformly with the gateway backend. + # The Mistral SDK manages its own httpx client, so wrap the call in an + # anyio cancel-scope timeout rather than threading a per-request timeout + # through the SDK; on expiry this raises TimeoutError, which the + # OcrProcessor turns into a clean parse failure. + ocr_timeout = get_settings().document_ocr_timeout_seconds + with anyio.fail_after(ocr_timeout): + resp = await self._client.ocr.process_async( + model=self._model, + document={"type": "document_url", "document_url": data_url}, + ) pages = [(p.index, p.markdown or "") for p in (resp.pages or [])] return _pages_to_text(pages) @@ -251,6 +264,24 @@ class OcrProcessor(DocumentProcessor): text, boundaries = await backend.ocr( content, content_type.split(";")[0].strip().lower() ) + except (TimeoutError, httpx.TimeoutException): + # Two timeout shapes reach here: the Mistral backend's + # anyio.fail_after raises the builtin TimeoutError, while the gateway + # backend's httpx.Timeout raises httpx.ReadTimeout (a + # httpx.TimeoutException, NOT a TimeoutError). Catch both so a + # too-low DOCUMENT_OCR_TIMEOUT_SECONDS lands in its own reason bucket + # rather than being conflated with provider errors. + timeout = settings.document_ocr_timeout_seconds + logger.warning( + "OCR timed out for %s after %.1fs", filename or "", timeout + ) + return ProcessingResult( + text="", + metadata={"parse_failed_reason": "timeout"}, + processor=self.name, + success=False, + error=f"OCR timed out after {timeout:.1f}s", + ) except Exception as e: logger.warning("OCR failed for %s: %s", filename or "", e) return ProcessingResult( diff --git a/nextcloud_mcp_server/document_processors/registry.py b/nextcloud_mcp_server/document_processors/registry.py index d152d2a2..b6c181a5 100644 --- a/nextcloud_mcp_server/document_processors/registry.py +++ b/nextcloud_mcp_server/document_processors/registry.py @@ -202,6 +202,33 @@ class ProcessorRegistry: """ settings = get_settings() + # Pre-parse size guard: a pathologically large PDF (e.g. a 42 MB scanned + # DUDE) burns the OCR timeout for 0 chars. Fail fast with an explicit + # reason so the caller marks the placeholder "failed" instead of + # retrying. 0 disables the cap. This lives on the auto-tiered path only: + # an explicit processor_name="ocr" override (registry.process) bypasses + # _process_pdf entirely and is intentionally not size-gated (power-user + # escape hatch). Returning here also skips _run_processor, so the + # rejection is counted on astrolabe_document_parse_failed_total{oversize} + # (via vector/processor.py) but deliberately not on the parse-duration + # histogram -- there is no parse to time. + max_pdf_mb = settings.document_max_pdf_size_mb + if max_pdf_mb > 0 and len(content) > max_pdf_mb * 1024 * 1024: + size_mb = len(content) / (1024 * 1024) + logger.warning( + "PDF %s is %.1f MB (> %.1f MB cap); failing fast as oversize", + filename or "", + size_mb, + max_pdf_mb, + ) + return ProcessingResult( + text="", + metadata={"parse_failed_reason": "oversize"}, + processor="size_guard", + success=False, + error=(f"PDF exceeds size cap: {size_mb:.1f} MB > {max_pdf_mb:.1f} MB"), + ) + if settings.document_tier1_engine == "pymupdf": processor = self._pdf_processor_for_tier("structured") if processor is None: diff --git a/nextcloud_mcp_server/vector/_errors.py b/nextcloud_mcp_server/vector/_errors.py new file mode 100644 index 00000000..f2cddc79 --- /dev/null +++ b/nextcloud_mcp_server/vector/_errors.py @@ -0,0 +1,35 @@ +"""Error-formatting helpers for the vector-sync pipeline. + +Vector-sync work runs inside anyio task groups, so a failure in a child task +surfaces as a ``BaseExceptionGroup`` whose default ``str()`` is the useless +``"unhandled errors in a TaskGroup (N sub-exception)"`` -- it hides the real +``ConnectError`` / ``APIConnectionError`` that operators need to triage embed +drops (card 309). ``format_exception_group`` flattens the group to the leaf +exceptions so log lines name the actual cause; pair it with ``exc_info=True`` to +keep the full traceback. +""" + + +def format_exception_group(exc: BaseException) -> str: + """Return a concise, leaf-naming string for ``exc``. + + For a (possibly nested) ``BaseExceptionGroup`` this joins the ``repr`` of + each leaf exception; for an ordinary exception it returns its ``repr``. The + result is meant for the human-readable portion of a log message, not for + parsing. + """ + if not isinstance(exc, BaseExceptionGroup): + return repr(exc) + leaves = _flatten(exc) + noun = "sub-exception" if len(leaves) == 1 else "sub-exceptions" + return f"{len(leaves)} {noun}: " + "; ".join(repr(e) for e in leaves) + + +def _flatten(exc: BaseException) -> list[BaseException]: + """Depth-first list of the leaf exceptions within ``exc``.""" + if isinstance(exc, BaseExceptionGroup): + leaves: list[BaseException] = [] + for sub in exc.exceptions: + leaves.extend(_flatten(sub)) + return leaves + return [exc] diff --git a/nextcloud_mcp_server/vector/oauth_sync.py b/nextcloud_mcp_server/vector/oauth_sync.py index eb4f1bb0..6cf44564 100644 --- a/nextcloud_mcp_server/vector/oauth_sync.py +++ b/nextcloud_mcp_server/vector/oauth_sync.py @@ -30,6 +30,7 @@ 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._errors import format_exception_group 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 @@ -257,7 +258,7 @@ async def user_scanner_task( logger.error( "[BasicAuth] Scanner error for %s: %s (%s/%s)", user_id, - e, + format_exception_group(e), consecutive_errors, max_consecutive_errors, exc_info=True, @@ -331,7 +332,7 @@ async def multi_user_processor_task( break except NotProvisionedError: - if doc_task: + if doc_task is not None: logger.warning( "[BasicAuth] User %s not provisioned, skipping %s_%s", doc_task.user_id, @@ -341,18 +342,21 @@ async def multi_user_processor_task( continue except Exception as e: - if doc_task: + if doc_task is not None: logger.error( "[BasicAuth] Processor %s error processing %s_%s: %s", worker_id, doc_task.doc_type, doc_task.doc_id, - e, + format_exception_group(e), exc_info=True, ) else: logger.error( - "[BasicAuth] Processor %s error: %s", worker_id, e, exc_info=True + "[BasicAuth] Processor %s error: %s", + worker_id, + format_exception_group(e), + exc_info=True, ) finally: @@ -481,7 +485,11 @@ async def user_manager_task( logger.info("[BasicAuth] Stopped %s scanner(s)", len(revoked_users)) except Exception as e: - logger.error("[BasicAuth] User manager error: %s", e, exc_info=True) + logger.error( + "[BasicAuth] User manager error: %s", + format_exception_group(e), + exc_info=True, + ) # Sleep until next poll try: diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index dae0acdf..ddc5da73 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -33,6 +33,7 @@ from nextcloud_mcp_server.observability.tracing import trace_operation from nextcloud_mcp_server.search.pdf_highlighter import PDFHighlighter from nextcloud_mcp_server.usage import UsageEventStore from nextcloud_mcp_server.vector import payload_keys +from nextcloud_mcp_server.vector._errors import format_exception_group from nextcloud_mcp_server.vector.document_chunker import ( DocumentChunker, PageAwareChunker, @@ -284,6 +285,11 @@ async def processor_task( # Signal that the task has started and is ready task_status.started() + # Initialised before the loop so the broad except handler below can't hit an + # unbound name if receive() itself raises a non-TimeoutError/EndOfStream + # exception on the very first iteration (mirrors multi_user_processor_task). + doc_task: DocumentTask | None = None + while not shutdown_event.is_set(): try: # Get document with timeout (allows checking shutdown) @@ -313,14 +319,22 @@ async def processor_task( break except Exception as e: - logger.error( - "Processor %s error processing %s_%s: %s", - worker_id, - doc_task.doc_type, - doc_task.doc_id, - e, - exc_info=True, - ) + if doc_task is not None: + logger.error( + "Processor %s error processing %s_%s: %s", + worker_id, + doc_task.doc_type, + doc_task.doc_id, + format_exception_group(e), + exc_info=True, + ) + else: + logger.error( + "Processor %s error: %s", + worker_id, + format_exception_group(e), + exc_info=True, + ) # Continue to next document (no task_done() needed with streams) logger.info("Processor %s stopped", worker_id) @@ -499,7 +513,7 @@ async def process_document( max_retries, doc_task.doc_type, doc_task.doc_id, - e, + format_exception_group(e), extra={ "doc_id": doc_task.doc_id, "doc_type": doc_task.doc_type, @@ -518,7 +532,7 @@ async def process_document( doc_task.doc_id, max_retries, reason, - e, + format_exception_group(e), extra={ "doc_id": doc_task.doc_id, "doc_type": doc_task.doc_type, diff --git a/pyproject.toml b/pyproject.toml index 107cf47d..cbae66db 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "nextcloud-mcp-server" -version = "0.111.0" +version = "0.113.0" description = "Model Context Protocol (MCP) server for Nextcloud integration - enables AI assistants to interact with Nextcloud data" authors = [ {name = "Chris Coutinho", email = "chris@coutinho.io"} diff --git a/tests/test_cli.py b/tests/test_cli.py index 9d25dae9..87c006b6 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1,11 +1,12 @@ """Tests for CLI options using Click's testing utilities.""" import os +from types import SimpleNamespace import pytest from click.testing import CliRunner -from nextcloud_mcp_server.cli import run +from nextcloud_mcp_server.cli import _init_worker_observability, run, worker @pytest.fixture @@ -324,3 +325,146 @@ def test_stdio_calls_get_stdio_mcp(runner, clean_env, monkeypatch): assert result.exit_code == 0, result.output assert called_with.get("transport") == "stdio" assert called_with.get("enabled_apps") is None + + +# --------------------------------------------------------------------------- +# Ingest worker observability bootstrap (Deck #310 / #175) +# --------------------------------------------------------------------------- + + +def _fake_settings(**overrides): + """A lightweight settings stand-in for the worker observability helper. + + The helper only reads attributes, so a SimpleNamespace avoids running the + real Settings.__post_init__ validation/derivation. + """ + base = dict( + ingest_queue="postgres", # for realism / worker() gating; unused by the helper + log_format="json", + log_level="INFO", + log_include_trace_context=True, + metrics_enabled=True, + metrics_port=9090, + otel_exporter_otlp_endpoint=None, + otel_service_name="nextcloud-mcp-server", + otel_exporter_verify_ssl=False, + otel_traces_sampler_arg=1.0, + ) + base.update(overrides) + return SimpleNamespace(**base) + + +@pytest.fixture +def patched_observability(monkeypatch): + """Patch the worker's observability entrypoints and record their kwargs.""" + calls: dict[str, dict] = {} + monkeypatch.setattr( + "nextcloud_mcp_server.cli.setup_logging", + lambda **kw: calls.__setitem__("logging", kw), + ) + monkeypatch.setattr( + "nextcloud_mcp_server.cli.setup_metrics", + lambda **kw: calls.__setitem__("metrics", kw), + ) + monkeypatch.setattr( + "nextcloud_mcp_server.cli.setup_tracing", + lambda **kw: calls.__setitem__("tracing", kw), + ) + return calls + + +def test_init_worker_observability_configures_logging(patched_observability): + """Worker initializes structured logging from settings (AC: JSON logs).""" + _init_worker_observability(_fake_settings()) + + assert patched_observability["logging"] == { + "log_format": "json", + "log_level": "INFO", + "include_trace_context": True, + } + + +def test_init_worker_observability_starts_metrics_when_enabled(patched_observability): + """Worker starts the Prometheus server on the configured port (AC: /metrics).""" + _init_worker_observability(_fake_settings(metrics_port=9123)) + + assert patched_observability["metrics"] == {"port": 9123} + + +def test_init_worker_observability_skips_metrics_when_disabled(patched_observability): + """METRICS_ENABLED=false leaves the worker without a metrics server.""" + _init_worker_observability(_fake_settings(metrics_enabled=False)) + + assert "metrics" not in patched_observability + # Logging is still configured regardless of the metrics toggle. + assert "logging" in patched_observability + + +def test_init_worker_observability_sets_up_tracing_when_endpoint( + patched_observability, +): + """An OTLP endpoint enables tracing so worker spans (parse/embed) export.""" + _init_worker_observability( + _fake_settings( + otel_exporter_otlp_endpoint="https://otel:4317", + otel_traces_sampler_arg=0.5, + ) + ) + + assert patched_observability["tracing"] == { + "service_name": "nextcloud-mcp-server", + "otlp_endpoint": "https://otel:4317", + "otlp_verify_ssl": False, + "sampling_rate": 0.5, + } + + +def test_init_worker_observability_skips_tracing_without_endpoint( + patched_observability, +): + """No OTLP endpoint → tracing stays disabled (matches API pod behavior).""" + _init_worker_observability(_fake_settings(otel_exporter_otlp_endpoint=None)) + + assert "tracing" not in patched_observability + + +def test_worker_initializes_observability_on_postgres_queue(runner, monkeypatch): + """The worker command wires up observability once config is runnable.""" + monkeypatch.setattr( + "nextcloud_mcp_server.cli.get_settings", + lambda: _fake_settings(ingest_queue="postgres"), + ) + + called = {} + + def fake_init(settings): + called["settings"] = settings + # Stop before the procrastinate/worker machinery. + raise SystemExit(0) + + monkeypatch.setattr( + "nextcloud_mcp_server.cli._init_worker_observability", fake_init + ) + + result = runner.invoke(worker, []) + assert result.exit_code == 0, result.output + assert called.get("settings") is not None + + +def test_worker_rejects_non_postgres_queue_before_observability(runner, monkeypatch): + """A non-postgres queue fails fast, before any metrics server is started.""" + monkeypatch.setattr( + "nextcloud_mcp_server.cli.get_settings", + lambda: _fake_settings(ingest_queue="memory"), + ) + + called = {} + monkeypatch.setattr( + "nextcloud_mcp_server.cli._init_worker_observability", + lambda settings: called.setdefault("init", True), + ) + + result = runner.invoke(worker, []) + assert result.exit_code != 0 + assert "INGEST_QUEUE=postgres" in result.output + assert "init" not in called diff --git a/tests/unit/client/test_webdav.py b/tests/unit/client/test_webdav.py index c02144e0..ffa7c6f2 100644 --- a/tests/unit/client/test_webdav.py +++ b/tests/unit/client/test_webdav.py @@ -510,3 +510,133 @@ def test_parse_search_response_decodes_non_ascii_paths(mocker): assert results[0]["href"] == "/remote.php/dav/files/testuser/学生邮箱/report.pdf" # name comes from , which is not URL-encoded; sanity-check it. assert results[0]["name"] == "report.pdf" + + +def _request_url(mock_http_client) -> str: + """Positional URL passed to the underlying httpx ``request`` call.""" + return mock_http_client.request.call_args[0][1] + + +@pytest.mark.unit +@pytest.mark.parametrize( + "path, expected", + [ + ("", "/remote.php/dav/files/testuser/"), + ("/Documents/notes.txt", "/remote.php/dav/files/testuser/Documents/notes.txt"), + ("Documents/notes.txt", "/remote.php/dav/files/testuser/Documents/notes.txt"), + ("a/b #1.pdf", "/remote.php/dav/files/testuser/a/b%20%231.pdf"), + ("law/x, y z.pdf", "/remote.php/dav/files/testuser/law/x%2C%20y%20%20z.pdf"), + ( + "学生邮箱/r.pdf", + "/remote.php/dav/files/testuser/%E5%AD%A6%E7%94%9F%E9%82%AE%E7%AE%B1/r.pdf", + ), + ], +) +def test_webdav_path_encoding(path, expected): + """_webdav_path encodes the decoded caller path once, preserving '/', and + strips a leading slash. Every caller-path builder routes through this, so + it is the single source of truth for their encoding.""" + client = WebDAVClient(AsyncMock(), "testuser") + assert client._webdav_path(path) == expected + + +@pytest.mark.unit +def test_encode_dav_path_encodes_exactly_once(): + """Pins the decoded-input precondition: a literal '%' becomes '%25', so an + already-encoded path passed in error would double-encode (caught here).""" + from nextcloud_mcp_server.client.webdav import _encode_dav_path + + assert _encode_dav_path("already%20encoded.pdf") == "already%2520encoded.pdf" + + +@pytest.mark.unit +async def test_read_file_encodes_special_chars(mocker): + """read_file must percent-encode '#', commas, and spaces in the path (card 309). + + Paths arrive already URL-decoded from PROPFIND/REPORT, so an unencoded '#' + reaches httpx as a URL fragment and silently truncates the request → 404 on + valid files (e.g. OHR-Bench law filenames). The outgoing request path must be + percent-encoded. + """ + mock_http_client = AsyncMock() + client = WebDAVClient(mock_http_client, "testuser") + + mock_response = AsyncMock() + mock_response.content = b"%PDF-1.4 data" + mock_response.headers = {"content-type": "application/pdf"} + mock_response.raise_for_status = mocker.Mock() + mock_http_client.request = AsyncMock(return_value=mock_response) + + # Name with a '#', a comma, a double space and a trailing space before ".pdf". + await client.read_file("law/ADMA BioManufacturing, LLC - Amendment #2 .pdf") + + url = _request_url(mock_http_client) + assert url.startswith("/remote.php/dav/files/testuser/") + # The hazardous characters are encoded; path separators are preserved. + assert "%23" in url # '#' + assert "%2C" in url # ',' + assert "%20" in url # space + assert "#" not in url + assert ", " not in url + assert "/law/" in url + + +@pytest.mark.unit +async def test_read_file_ascii_path_unchanged(mocker): + """A plain ASCII path must pass through unchanged (no spurious encoding).""" + mock_http_client = AsyncMock() + client = WebDAVClient(mock_http_client, "testuser") + + mock_response = AsyncMock() + mock_response.content = b"data" + mock_response.headers = {"content-type": "text/plain"} + mock_response.raise_for_status = mocker.Mock() + mock_http_client.request = AsyncMock(return_value=mock_response) + + await client.read_file("Documents/notes.txt") + + assert ( + _request_url(mock_http_client) + == "/remote.php/dav/files/testuser/Documents/notes.txt" + ) + + +@pytest.mark.unit +async def test_move_resource_encodes_destination_header(mocker): + """The MOVE Destination header must be percent-encoded too (card 309).""" + mock_http_client = AsyncMock() + client = WebDAVClient(mock_http_client, "testuser") + + mock_response = AsyncMock() + mock_response.status_code = 201 + mock_response.raise_for_status = mocker.Mock() + mock_http_client.request = AsyncMock(return_value=mock_response) + + await client.move_resource("a/old.pdf", "b/new #1.pdf") + + call = mock_http_client.request.call_args + # Source is the request path; destination is the header. + assert call[0][1] == "/remote.php/dav/files/testuser/a/old.pdf" + destination = call.kwargs["headers"]["Destination"] + assert "%23" in destination + assert "#" not in destination + + +@pytest.mark.unit +async def test_copy_resource_encodes_destination_header(mocker): + """The COPY Destination header must be percent-encoded too (card 309).""" + mock_http_client = AsyncMock() + client = WebDAVClient(mock_http_client, "testuser") + + mock_response = AsyncMock() + mock_response.status_code = 201 + mock_response.raise_for_status = mocker.Mock() + mock_http_client.request = AsyncMock(return_value=mock_response) + + await client.copy_resource("a/old.pdf", "b/new #1.pdf") + + call = mock_http_client.request.call_args + assert call[0][1] == "/remote.php/dav/files/testuser/a/old.pdf" + destination = call.kwargs["headers"]["Destination"] + assert "%23" in destination + assert "#" not in destination diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index 2e6ba0e2..2f7cc371 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -213,6 +213,24 @@ class TestChunkConfigValidation: _reload_config() assert get_settings().document_chunk_page_aware is False + def test_ocr_timeout_default_and_env_override(self): + """document_ocr_timeout_seconds defaults to 180 and reads its env var. + + Guards the _DEFAULTS-key-must-match-env-var footgun: a mismatch would + leave the override silently ignored. + """ + assert Settings().document_ocr_timeout_seconds == pytest.approx(180.0) + with patch.dict(os.environ, {"DOCUMENT_OCR_TIMEOUT_SECONDS": "45"}, clear=True): + _reload_config() + assert get_settings().document_ocr_timeout_seconds == pytest.approx(45.0) + + def test_max_pdf_size_default_and_env_override(self): + """document_max_pdf_size_mb defaults to 50 and reads its env var.""" + assert Settings().document_max_pdf_size_mb == pytest.approx(50.0) + with patch.dict(os.environ, {"DOCUMENT_MAX_PDF_SIZE_MB": "12.5"}, clear=True): + _reload_config() + assert get_settings().document_max_pdf_size_mb == pytest.approx(12.5) + def test_valid_chunk_settings(self): """Test valid chunk size and overlap configuration.""" settings = Settings( @@ -491,6 +509,22 @@ class TestDynaconfValidators: with pytest.raises(ValidationError, match="DOCUMENT_CHUNK_SIZE"): _reload_config() + @patch.dict(os.environ, {"DOCUMENT_OCR_TIMEOUT_SECONDS": "0"}, clear=True) + def test_ocr_timeout_zero_rejected(self): + """DOCUMENT_OCR_TIMEOUT_SECONDS=0 fails the gte=1 validator.""" + from dynaconf import ValidationError + + with pytest.raises(ValidationError, match="DOCUMENT_OCR_TIMEOUT_SECONDS"): + _reload_config() + + @patch.dict(os.environ, {"DOCUMENT_MAX_PDF_SIZE_MB": "-1"}, clear=True) + def test_max_pdf_size_negative_rejected(self): + """DOCUMENT_MAX_PDF_SIZE_MB=-1 fails the gte=0 validator (0 = disabled).""" + from dynaconf import ValidationError + + with pytest.raises(ValidationError, match="DOCUMENT_MAX_PDF_SIZE_MB"): + _reload_config() + @patch.dict(os.environ, {"METRICS_PORT": "8080"}, clear=True) def test_valid_metrics_port(self): """Test valid METRICS_PORT passes validation.""" diff --git a/tests/unit/test_ocr_processor.py b/tests/unit/test_ocr_processor.py index 690f0c38..51e00906 100644 --- a/tests/unit/test_ocr_processor.py +++ b/tests/unit/test_ocr_processor.py @@ -3,6 +3,7 @@ from types import SimpleNamespace from typing import Any +import anyio import pytest from nextcloud_mcp_server.document_processors import ocr @@ -14,6 +15,7 @@ def _settings(**kw) -> Any: # a Settings stand-in (only the read fields matter) base = dict( document_ocr_provider="auto", document_ocr_model="mistral/mistral-ocr-latest", + document_ocr_timeout_seconds=180.0, embedding_gateway_url=None, embedding_gateway_client_id=None, embedding_gateway_client_secret=None, @@ -128,3 +130,91 @@ async def test_processor_backend_error_returns_success_false(monkeypatch): r = await ocr.OcrProcessor().process(b"%PDF-1.7", "application/pdf") assert r.success is False assert r.metadata["parse_failed_reason"] == "error" + + +async def test_processor_timeout_returns_timeout_reason(monkeypatch): + """A backend TimeoutError gets its own reason bucket (not 'error').""" + + class _TimeoutBackend: + async def ocr(self, content, mime_type): + raise TimeoutError + + monkeypatch.setattr( + ocr, "get_settings", lambda: _settings(document_ocr_timeout_seconds=5.0) + ) + monkeypatch.setattr(ocr, "build_ocr_backend", lambda s: _TimeoutBackend()) + r = await ocr.OcrProcessor().process(b"%PDF-1.7", "application/pdf") + assert r.success is False + assert r.metadata["parse_failed_reason"] == "timeout" + assert "timed out" in r.error + + +async def test_gateway_httpx_timeout_maps_to_timeout_reason(monkeypatch): + """A gateway httpx.ReadTimeout (not a builtin TimeoutError) must still map to + parse_failed_reason='timeout', not 'error'.""" + import httpx + + class _HttpxTimeoutBackend: + async def ocr(self, content, mime_type): + raise httpx.ReadTimeout("read timed out") + + monkeypatch.setattr( + ocr, "get_settings", lambda: _settings(document_ocr_timeout_seconds=5.0) + ) + monkeypatch.setattr(ocr, "build_ocr_backend", lambda s: _HttpxTimeoutBackend()) + r = await ocr.OcrProcessor().process(b"%PDF-1.7", "application/pdf") + assert r.success is False + assert r.metadata["parse_failed_reason"] == "timeout" + assert "timed out" in r.error + + +async def test_gateway_backend_uses_configured_timeout(mocker, monkeypatch): + """The gateway OCR call must use DOCUMENT_OCR_TIMEOUT_SECONDS (resolved per + call), not the old hardcoded 180s constant.""" + resp = mocker.Mock() + resp.raise_for_status = mocker.Mock() + resp.json = mocker.Mock(return_value={"pages": [{"index": 0, "markdown": "ok"}]}) + + client = mocker.MagicMock() + client.__aenter__ = mocker.AsyncMock(return_value=client) + client.__aexit__ = mocker.AsyncMock(return_value=False) + client.post = mocker.AsyncMock(return_value=resp) + + captured: dict[str, Any] = {} + + def _make_client(*args, **kwargs): + captured["timeout"] = kwargs.get("timeout") + return client + + monkeypatch.setattr(ocr.httpx, "AsyncClient", _make_client) + monkeypatch.setattr( + ocr, "get_settings", lambda: _settings(document_ocr_timeout_seconds=42.0) + ) + + backend = ocr._GatewayOcrBackend("https://gw", "mistral/mistral-ocr-latest") + await backend.ocr(b"%PDF-1.7", "application/pdf") + + # httpx.Timeout(42.0, connect=10.0): the read/overall budget is the setting. + assert captured["timeout"].read == pytest.approx(42.0) + assert captured["timeout"].connect == pytest.approx(10.0) + + +async def test_mistral_backend_applies_timeout(mocker, monkeypatch): + """The Mistral backend wraps process_async in DOCUMENT_OCR_TIMEOUT_SECONDS, + so a slow OCR call fails fast instead of hanging on the SDK default.""" + monkeypatch.setattr( + ocr, "get_settings", lambda: _settings(document_ocr_timeout_seconds=0.01) + ) + + # Bypass the SDK constructor; only the two attributes ocr() reads matter. + backend = ocr._MistralOcrBackend.__new__(ocr._MistralOcrBackend) + backend._model = "mistral-ocr-latest" + + async def _slow(*args, **kwargs): + await anyio.sleep(1.0) + + backend._client = mocker.MagicMock() + backend._client.ocr.process_async = _slow + + with pytest.raises(TimeoutError): + await backend.ocr(b"%PDF-1.7", "application/pdf") diff --git a/tests/unit/test_registry_tiering.py b/tests/unit/test_registry_tiering.py index 5637f4da..c111763d 100644 --- a/tests/unit/test_registry_tiering.py +++ b/tests/unit/test_registry_tiering.py @@ -74,6 +74,9 @@ class _Settings: page_fraction=0.5, min_page_chars=16, detect_scanned=False, + # Guard off by default so existing tiering tests are unaffected; tests + # that exercise the size guard pass an explicit cap. + max_pdf_size_mb=0.0, ): self.document_tier1_engine = engine self.document_classify_enabled = classify @@ -82,6 +85,7 @@ class _Settings: self.document_ocr_page_fraction = page_fraction self.document_ocr_min_page_chars = min_page_chars self.document_ocr_detect_scanned = detect_scanned + self.document_max_pdf_size_mb = max_pdf_size_mb def _registry(*procs: tuple[DocumentProcessor, int]) -> ProcessorRegistry: @@ -98,6 +102,43 @@ async def test_pdf_routes_to_fast_tier(monkeypatch): assert res.processor == "fast" +async def test_oversize_pdf_fails_fast_without_parsing(monkeypatch): + """A PDF over the size cap must fail fast as 'oversize' before any tier runs.""" + monkeypatch.setattr( + reg_mod, "get_settings", lambda: _Settings(max_pdf_size_mb=0.001) + ) + fast = _Fake("fast", "fast") + ran = False + orig = fast.process + + async def _tracking(*a, **k): + nonlocal ran + ran = True + return await orig(*a, **k) + + fast.process = _tracking # type: ignore[method-assign] + r = _registry((fast, 20)) + + # ~2 KB > 0.001 MB (~1 KB) cap. + res = await r.process(b"%PDF-1.7" + b"0" * 2048, "application/pdf", "big.pdf") + + assert res.success is False + assert res.metadata["parse_failed_reason"] == "oversize" + assert res.processor == "size_guard" + assert ran is False, "size guard must short-circuit before the fast tier runs" + + +async def test_under_cap_pdf_still_parses(monkeypatch): + """A PDF under the cap is unaffected by the guard.""" + monkeypatch.setattr( + reg_mod, "get_settings", lambda: _Settings(max_pdf_size_mb=10.0) + ) + r = _registry((_Fake("fast", "fast"), 20)) + res = await r.process(b"%PDF-1.7", "application/pdf") + assert res.success is True + assert res.processor == "fast" + + async def test_engine_rollback_uses_structured(monkeypatch): monkeypatch.setattr(reg_mod, "get_settings", lambda: _Settings(engine="pymupdf")) r = _registry((_Fake("fast", "fast"), 20), (_Fake("structured", "structured"), 10)) diff --git a/tests/unit/vector/test_errors.py b/tests/unit/vector/test_errors.py new file mode 100644 index 00000000..8c5dfe22 --- /dev/null +++ b/tests/unit/vector/test_errors.py @@ -0,0 +1,44 @@ +"""Unit tests for vector-sync error formatting (card 309).""" + +import httpx +import pytest + +from nextcloud_mcp_server.vector._errors import format_exception_group + + +@pytest.mark.unit +def test_format_plain_exception_returns_repr(): + exc = httpx.ConnectError("Connection error") + assert format_exception_group(exc) == repr(exc) + + +@pytest.mark.unit +def test_format_exception_group_names_leaf_cause(): + """A single-child group must surface the real ConnectError, not the group's + useless 'unhandled errors in a TaskGroup' default message.""" + leaf = httpx.ConnectError("Connection error") + group = BaseExceptionGroup("unhandled errors in a TaskGroup", [leaf]) + + formatted = format_exception_group(group) + + assert "ConnectError" in formatted + # Assert the full leaf repr survives, not just the type name -- guards a + # future format change that kept the type but dropped the message. + assert repr(leaf) in formatted + assert "unhandled errors in a TaskGroup" not in formatted + assert "1 sub-exception" in formatted + + +@pytest.mark.unit +def test_format_nested_exception_group_flattens_all_leaves(): + inner = BaseExceptionGroup( + "inner", [ValueError("bad value"), httpx.ConnectError("conn")] + ) + outer = BaseExceptionGroup("outer", [inner, RuntimeError("boom")]) + + formatted = format_exception_group(outer) + + assert "ValueError" in formatted + assert "ConnectError" in formatted + assert "RuntimeError" in formatted + assert "3 sub-exceptions" in formatted diff --git a/tests/unit/vector/test_processor_task.py b/tests/unit/vector/test_processor_task.py new file mode 100644 index 00000000..8276a073 --- /dev/null +++ b/tests/unit/vector/test_processor_task.py @@ -0,0 +1,50 @@ +"""Regression test for processor_task's exception handler (card 309 / PR #891). + +If ``receive_stream.receive()`` raises something other than +``TimeoutError``/``EndOfStream`` before any document is bound, the broad +``except`` handler must not crash on an unbound ``doc_task`` name. +""" + +from unittest.mock import MagicMock + +import anyio +import pytest + +from nextcloud_mcp_server.vector.processor import processor_task + + +class _ReceiveBoomThenEnd: + """First receive() raises a non-Timeout error (no doc_task bound yet); the + second ends the stream so the loop exits.""" + + def __init__(self, shutdown: anyio.Event): + self._calls = 0 + self._shutdown = shutdown + + async def receive(self): + self._calls += 1 + if self._calls == 1: + raise RuntimeError("transport blew up before any document") + self._shutdown.set() + raise anyio.EndOfStream + + def statistics(self): # pragma: no cover - not reached on the error path + return MagicMock(current_buffer_used=0) + + +@pytest.mark.unit +async def test_processor_task_receive_error_does_not_raise_unbound(caplog): + shutdown = anyio.Event() + stream = _ReceiveBoomThenEnd(shutdown) + + # Must complete without a NameError leaking out of the except handler. + with caplog.at_level("ERROR", logger="nextcloud_mcp_server.vector.processor"): + await processor_task( + worker_id=0, + receive_stream=stream, # type: ignore[arg-type] + shutdown_event=shutdown, + nc_client=MagicMock(), + user_id="alice", + ) + + assert any("RuntimeError" in rec.message for rec in caplog.records) diff --git a/uv.lock b/uv.lock index 97fe6302..da1f5df8 100644 --- a/uv.lock +++ b/uv.lock @@ -2183,7 +2183,7 @@ wheels = [ [[package]] name = "nextcloud-mcp-server" -version = "0.111.0" +version = "0.113.0" source = { editable = "." } dependencies = [ { name = "aiosqlite" },