Merge remote-tracking branch 'origin/master' into fix/309-embed-resilience

# Conflicts:
#	nextcloud_mcp_server/vector/processor.py

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]

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:

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,