refactor(search): address PR #750 round 4 review feedback

- Guard eviction_task_group.start_soon against shutdown race so a
  RuntimeError on a closed group never surfaces as a search error.
- Correct ADR-019 news_item row: there is no per-item REST endpoint;
  verification batches via get_items(batch_size=-1) and intersects.
- Modernize models/semantic.py typing to PEP 604 / lowercase generics
  per CLAUDE.md.

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

docs/ADR-019-verify-on-read-for-semantic-search.md

@@ -49,7 +49,7 @@ The vector pipeline (`vector/scanner.py`, `vector/processor.py`) currently index
 | doc_type | Cheapest authoritative check | Notes |
 |---|---|---|
 | `note` | `notes.get_note(id)` — single REST call, 404 on deletion | Per-user store; access is binary (yours or not). |
-| `news_item` | `news.get_item(id)` — single REST call | Per-user feeds; clean 404 semantics. |
+| `news_item` | `get_items(batch_size=-1)` once per search + intersect | No per-item REST endpoint (`get_item()` is itself a fetch-all + filter); batching once is cheaper than N fetch-all-and-filter calls. |
 | `file` | WebDAV `PROPFIND` with `Depth: 0` on `file_path` (already stored in Qdrant payload, see `server/semantic.py:161`) | `read_file()` works but downloads the body — too heavy for a verification check. PROPFIND is the WebDAV equivalent of HEAD. Catches both deletes and unshares. |
 | `deck_card` | `deck.get_card(board_id, stack_id, card_id)` using metadata cached in Qdrant (`search/context.py::_get_deck_metadata_from_qdrant`) | Fallback iteration through all boards/stacks (used by context expansion) is O(boards × stacks) and far too expensive to run on every query. |
 

nextcloud_mcp_server/models/semantic.py

@@ -1,7 +1,5 @@
 """Pydantic models for semantic search responses."""
 
-from typing import List, Optional
-
 from pydantic import BaseModel, Field
 
 from .base import BaseResponse
@@ -37,36 +35,36 @@ class SemanticSearchResult(BaseModel):
     )
     chunk_index: int = Field(description="Index of matching chunk in document")
     total_chunks: int = Field(description="Total number of chunks in document")
-    chunk_start_offset: Optional[int] = Field(
+    chunk_start_offset: int | None = Field(
         default=None, description="Character position where chunk starts in document"
     )
-    chunk_end_offset: Optional[int] = Field(
+    chunk_end_offset: int | None = Field(
         default=None, description="Character position where chunk ends in document"
     )
-    page_number: Optional[int] = Field(
+    page_number: int | None = Field(
         default=None, description="Page number for PDF documents"
     )
-    page_count: Optional[int] = Field(
+    page_count: int | None = Field(
         default=None, description="Total number of pages in PDF document"
     )
     # Context expansion fields (optional, populated when include_context=True)
     has_context_expansion: bool = Field(
         default=False, description="Whether context expansion was performed"
     )
-    marked_text: Optional[str] = Field(
+    marked_text: str | None = Field(
         default=None,
         description="Full text with position markers around matched chunk",
     )
-    before_context: Optional[str] = Field(
+    before_context: str | None = Field(
         default=None, description="Text before the matched chunk"
     )
-    after_context: Optional[str] = Field(
+    after_context: str | None = Field(
         default=None, description="Text after the matched chunk"
     )
-    has_before_truncation: Optional[bool] = Field(
+    has_before_truncation: bool | None = Field(
         default=None, description="Whether before_context was truncated"
     )
-    has_after_truncation: Optional[bool] = Field(
+    has_after_truncation: bool | None = Field(
         default=None, description="Whether after_context was truncated"
     )
 
@@ -74,7 +72,7 @@ class SemanticSearchResult(BaseModel):
 class SemanticSearchResponse(BaseResponse):
     """Response model for semantic search across all indexed Nextcloud apps."""
 
-    results: List[SemanticSearchResult] = Field(
+    results: list[SemanticSearchResult] = Field(
         description="Semantic search results with similarity scores"
     )
     query: str = Field(description="The search query used")
@@ -106,7 +104,7 @@ class SamplingSearchResponse(BaseResponse):
     generated_answer: str = Field(
         ..., description="LLM-generated answer based on retrieved documents"
     )
-    sources: List[SemanticSearchResult] = Field(
+    sources: list[SemanticSearchResult] = Field(
         default_factory=list,
         description="Source documents with excerpts and relevance scores",
     )
@@ -114,10 +112,10 @@ class SamplingSearchResponse(BaseResponse):
     search_method: str = Field(
         default="semantic_sampling", description="Search method used"
     )
-    model_used: Optional[str] = Field(
+    model_used: str | None = Field(
         default=None, description="Model that generated the answer"
     )
-    stop_reason: Optional[str] = Field(
+    stop_reason: str | None = Field(
         default=None, description="Reason generation stopped"
     )
 

nextcloud_mcp_server/search/verification.py

@@ -462,7 +462,15 @@ async def verify_search_results(
 
         if eviction_task_group is not None:
             for doc_id, doc_type in inaccessible:
-                eviction_task_group.start_soon(evict, doc_id, doc_type)
+                # Guard against the lifespan task group having exited between
+                # the getattr() capture in server/semantic.py and this call —
+                # start_soon raises RuntimeError on a closed group, which
+                # would otherwise surface as a search error. Eviction is
+                # best-effort: the next query re-verifies and re-attempts.
+                try:
+                    eviction_task_group.start_soon(evict, doc_id, doc_type)
+                except Exception:
+                    logger.debug("Eviction task group closed; will retry on next query")
         else:
             async with anyio.create_task_group() as tg:
                 for doc_id, doc_type in inaccessible: