Files
mcp-nextcloud/nextcloud_mcp_server/server/semantic.py
T
Chris CoutinhoandClaude Opus 4.7 aa4b9498a1 refactor(search): address PR #750 round 3 review feedback
- _verify_deck_cards: hoist int(board_id|stack_id|doc_id) out of the generic
  except Exception into an explicit try/except (TypeError, ValueError) before
  the network call, mirroring _verify_news_items. Malformed payloads now log
  a specific warning instead of "unexpected error".
- _verify_news_items: add TODO(perf) above the get_items(batch_size=-1) call
  to mark the known fetch-all cost as a future profiling target.
- SemanticSearchResult.id: revert from int|str back to int. The internal
  SearchResult.id stays int|str for forward-compat; the MCP response model
  narrows at the boundary. server/semantic.py casts r.id to int when
  constructing the response so future string-id types fail loudly here
  instead of silently widening the public API.
- nc_semantic_search: replace the terse "extra for access filtering" comment
  with an ADR-019 NOTE block explaining the 2x over-fetch trade-off and the
  ghost-density under-delivery case (self-heals via lazy eviction).
- tests/integration/test_verify_on_read.py: extend the module docstring to
  call out that only the note verifier is exercised against real Nextcloud,
  while file/deck_card/news_item are unit-only — documenting the suite split
  for future contributors.
- ADR-019: rewrite "Module shape", "Verifier registry", example verifier,
  and "Deduplication" sections to match the shipped BatchVerifier interface
  (was per-id Verifier in the original draft). Add a "Why batch?" paragraph
  explaining the design choice. Update implementation checklist — every
  item is now [x] with corrected verifier names (plural) and the eviction
  module path (vector/eviction.py).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-01 19:17:35 +02:00

761 lines
33 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Semantic search MCP tools using vector database."""
import logging
import anyio
from httpx import RequestError
from mcp.server.fastmcp import Context, FastMCP
from mcp.shared.exceptions import McpError
from mcp.types import (
ClientCapabilities,
ErrorData,
ModelHint,
ModelPreferences,
SamplingCapability,
SamplingMessage,
TextContent,
ToolAnnotations,
)
from qdrant_client.models import Filter
from nextcloud_mcp_server.auth import require_scopes
from nextcloud_mcp_server.config import get_settings
from nextcloud_mcp_server.context import get_client
from nextcloud_mcp_server.models.semantic import (
SamplingSearchResponse,
SemanticSearchResponse,
SemanticSearchResult,
VectorSyncStatusResponse,
)
from nextcloud_mcp_server.observability.metrics import (
instrument_tool,
)
from nextcloud_mcp_server.search.bm25_hybrid import BM25HybridSearchAlgorithm
from nextcloud_mcp_server.search.context import get_chunk_with_context
from nextcloud_mcp_server.search.verification import verify_search_results
from nextcloud_mcp_server.vector.placeholder import get_placeholder_filter
from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client
logger = logging.getLogger(__name__)
def configure_semantic_tools(mcp: FastMCP):
"""Configure semantic search tools for MCP server."""
@mcp.tool(
title="Semantic Search",
annotations=ToolAnnotations(
readOnlyHint=True, # Search doesn't modify data
openWorldHint=True, # Queries external Nextcloud service
),
)
@require_scopes("semantic.read")
@instrument_tool
async def nc_semantic_search(
query: str,
ctx: Context,
limit: int = 10,
doc_types: list[str] | None = None,
score_threshold: float = 0.0,
fusion: str = "rrf",
include_context: bool = False,
context_chars: int = 300,
) -> SemanticSearchResponse:
"""
Search Nextcloud content using BM25 hybrid search with cross-app support.
Uses Qdrant's native hybrid search combining:
- Dense semantic vectors: For conceptual similarity and natural language queries
- BM25 sparse vectors: For precise keyword matching, acronyms, and specific terms
Results are automatically fused using the selected fusion algorithm in the
database for optimal relevance. This provides the best of both semantic
understanding and keyword precision.
Requires VECTOR_SYNC_ENABLED=true. Supports indexing of notes, files,
news items, and deck cards.
Args:
query: Natural language or keyword search query
limit: Maximum number of results to return (default: 10)
doc_types: Document types to search (e.g., ["note", "file", "deck_card", "news_item"]). None = search all indexed types (default)
score_threshold: Minimum fusion score (0-1, default: 0.0)
fusion: Fusion algorithm: "rrf" (Reciprocal Rank Fusion, default) or "dbsf" (Distribution-Based Score Fusion)
RRF: Good general-purpose fusion using reciprocal ranks
DBSF: Uses distribution-based normalization, may better balance different score ranges
include_context: Whether to expand results with surrounding context (default: False)
context_chars: Number of characters to include before/after matched chunk (default: 300)
Returns:
SemanticSearchResponse with matching documents ranked by fusion scores
"""
settings = get_settings()
client = await get_client(ctx)
username = client.username
logger.info(
f"BM25 hybrid search: query='{query}', user={username}, "
f"limit={limit}, score_threshold={score_threshold}, fusion={fusion}"
)
# Check that vector sync is enabled
if not settings.vector_sync_enabled:
raise McpError(
ErrorData(
code=-1,
message="BM25 hybrid search requires VECTOR_SYNC_ENABLED=true",
)
)
try:
# Create BM25 hybrid search algorithm with specified fusion
search_algo = BM25HybridSearchAlgorithm(
score_threshold=score_threshold, fusion=fusion
)
# Execute search across requested document types
# If doc_types is None, search all indexed types (cross-app search)
# If doc_types is a list, search only those types
all_results = []
if doc_types is None:
# Cross-app search: search all indexed types
# Get unverified results from Qdrant.
#
# NOTE (ADR-019): Over-fetch by 2× to absorb ghost-record drops
# during verify-on-read. When ghost density is high (e.g. a
# large board share was just revoked) this budget can still
# under-deliver against the requested ``limit``; the index
# self-heals via lazy eviction so subsequent searches recover.
# The 2× factor is a deliberate v1 trade-off — raising it
# costs Nextcloud round-trips on every search. Trim to
# ``limit`` happens AFTER verification.
unverified_results = await search_algo.search(
query=query,
user_id=username,
limit=limit * 2,
doc_type=None, # Signal to search all types
score_threshold=score_threshold,
)
all_results.extend(unverified_results)
else:
# Search specific document types
# For each requested type, execute search and combine results
# under the same 2× over-fetch budget (see NOTE above).
for dtype in doc_types:
unverified_results = await search_algo.search(
query=query,
user_id=username,
limit=limit * 2,
doc_type=dtype,
score_threshold=score_threshold,
)
all_results.extend(unverified_results)
# Sort combined results by score, then cap to `limit * 2` to
# match the cross-app branch's over-fetch budget. Without this
# cap, N requested doc_types × `limit * 2` results would all
# flow into verification, multiplying the Nextcloud round-trip
# cost by N.
all_results.sort(key=lambda r: r.score, reverse=True)
all_results = all_results[: limit * 2]
# ADR-019: Verify-on-read. The vector index is a recall layer;
# Nextcloud is the source of truth for access. Filter out ghost
# records (deleted/unshared docs not yet reconciled by webhooks)
# BEFORE trimming to `limit`, so we don't lose accessible results
# to the limit slot that ghosts would otherwise occupy. We also
# run this BEFORE context expansion to avoid re-fetching docs that
# are about to be dropped. Pass the lifespan-owned task group so
# eviction of dropped points is fire-and-forget (does not block
# the response).
eviction_task_group = getattr(
ctx.request_context.lifespan_context, "eviction_task_group", None
)
verified_results = await verify_search_results(
client,
all_results,
eviction_task_group=eviction_task_group,
)
search_results = verified_results[:limit]
# Convert SearchResult objects to SemanticSearchResult for response.
# SearchResult.id is typed `int | str` for forward-compat with future
# doc_types, but every currently indexed type uses numeric ids and
# the MCP response model narrows to `int`. Casting here makes the
# narrowing explicit and surfaces any future string-id type as a
# loud failure at the boundary instead of silently widening the
# public API.
results = []
for r in search_results:
results.append(
SemanticSearchResult(
id=int(r.id),
doc_type=r.doc_type,
title=r.title,
category=r.metadata.get("category", "") if r.metadata else "",
excerpt=r.excerpt,
score=r.score,
chunk_index=r.metadata.get("chunk_index", 0)
if r.metadata
else 0,
total_chunks=r.metadata.get("total_chunks", 1)
if r.metadata
else 1,
chunk_start_offset=r.chunk_start_offset,
chunk_end_offset=r.chunk_end_offset,
page_number=r.page_number,
)
)
# Expand results with surrounding context if requested
if include_context and results:
logger.info(
f"Expanding {len(results)} results with context "
f"(context_chars={context_chars})"
)
# Fetch context for all results in parallel
# Limit concurrent requests to prevent connection pool exhaustion
max_concurrent = 20
semaphore = anyio.Semaphore(max_concurrent)
expanded_results = [None] * len(results)
async def fetch_context(index: int, result: SemanticSearchResult):
"""Fetch context for a single result (parallel with semaphore)."""
async with semaphore:
# Only expand if we have valid chunk offsets
if (
result.chunk_start_offset is None
or result.chunk_end_offset is None
):
# Keep result as-is without context expansion
expanded_results[index] = result
return
try:
chunk_context = await get_chunk_with_context(
nc_client=client,
user_id=username,
doc_id=result.id,
doc_type=result.doc_type,
chunk_start=result.chunk_start_offset,
chunk_end=result.chunk_end_offset,
page_number=result.page_number,
chunk_index=result.chunk_index,
total_chunks=result.total_chunks,
context_chars=context_chars,
)
if chunk_context:
# Create new result with context fields populated
expanded_results[index] = SemanticSearchResult(
id=result.id,
doc_type=result.doc_type,
title=result.title,
category=result.category,
excerpt=result.excerpt,
score=result.score,
chunk_index=result.chunk_index,
total_chunks=result.total_chunks,
chunk_start_offset=result.chunk_start_offset,
chunk_end_offset=result.chunk_end_offset,
page_number=result.page_number,
# Context expansion fields
has_context_expansion=True,
marked_text=chunk_context.marked_text,
before_context=chunk_context.before_context,
after_context=chunk_context.after_context,
has_before_truncation=chunk_context.has_before_truncation,
has_after_truncation=chunk_context.has_after_truncation,
)
logger.debug(
f"Expanded context for {result.doc_type} {result.id}"
)
else:
# Context expansion failed, keep original result
expanded_results[index] = result
logger.debug(
f"Failed to expand context for {result.doc_type} {result.id}, "
"keeping original result"
)
except Exception as e:
# Context expansion failed, keep original result
expanded_results[index] = result
logger.warning(
f"Error expanding context for {result.doc_type} {result.id}: {e}"
)
# Run all context fetches in parallel using anyio task group
async with anyio.create_task_group() as tg:
for idx, result in enumerate(results):
tg.start_soon(fetch_context, idx, result)
# Replace results with expanded versions
results = [r for r in expanded_results if r is not None]
logger.info(
f"Context expansion completed: {len(results)} results with context"
)
logger.info(f"Returning {len(results)} results from BM25 hybrid search")
return SemanticSearchResponse(
results=results,
query=query,
total_found=len(results),
search_method=f"bm25_hybrid_{fusion}",
)
except ValueError as e:
error_msg = str(e)
if "No embedding provider configured" in error_msg:
raise McpError(
ErrorData(
code=-1,
message="Embedding service not configured. Set OLLAMA_BASE_URL environment variable.",
)
)
raise McpError(
ErrorData(code=-1, message=f"Configuration error: {error_msg}")
)
except RequestError as e:
raise McpError(
ErrorData(code=-1, message=f"Network error during search: {str(e)}")
)
except Exception as e:
logger.error(f"Search error: {e}", exc_info=True)
raise McpError(ErrorData(code=-1, message=f"Search failed: {str(e)}"))
@mcp.tool(
title="Search with AI-Generated Answer",
annotations=ToolAnnotations(
readOnlyHint=True, # Search doesn't modify data
openWorldHint=True, # Calls into Nextcloud via nc_semantic_search
),
)
@require_scopes("semantic.read")
@instrument_tool
async def nc_semantic_search_answer(
query: str,
ctx: Context,
limit: int = 5,
score_threshold: float = 0.7,
max_answer_tokens: int = 500,
fusion: str = "rrf",
include_context: bool = False,
context_chars: int = 300,
) -> SamplingSearchResponse:
"""
Semantic search with LLM-generated answer using MCP sampling.
Retrieves relevant documents from indexed Nextcloud apps (notes, calendar, deck,
files, contacts) using vector similarity search, then uses MCP sampling to request
the client's LLM to generate a natural language answer based on the retrieved context.
This tool combines the power of semantic search (finding relevant content across
all your Nextcloud apps) with LLM generation (synthesizing that content into
coherent answers). The generated answer includes citations to specific documents
with their types, allowing users to verify claims and explore sources.
The LLM generation happens client-side via MCP sampling. The MCP client
controls which model is used, who pays for it, and whether to prompt the
user for approval. This keeps the server simple (no LLM API keys needed)
while giving users full control over their LLM interactions.
Args:
query: Natural language question to answer (e.g., "What are my Q1 objectives?" or "When is my next dentist appointment?")
ctx: MCP context for session access
limit: Maximum number of documents to retrieve (default: 5)
score_threshold: Minimum similarity score 0-1 (default: 0.7)
max_answer_tokens: Maximum tokens for generated answer (default: 500)
fusion: Fusion algorithm: "rrf" (Reciprocal Rank Fusion, default) or "dbsf" (Distribution-Based Score Fusion)
include_context: Whether to expand results with surrounding context (default: False)
context_chars: Number of characters to include before/after matched chunk (default: 300)
Returns:
SamplingSearchResponse containing:
- generated_answer: Natural language answer with citations
- sources: List of documents with excerpts and relevance scores
- model_used: Which model generated the answer
- stop_reason: Why generation stopped
Note: Requires MCP client to support sampling. If sampling is unavailable,
the tool gracefully degrades to returning documents with an explanation.
The client may prompt the user to approve the sampling request.
"""
# 1. Retrieve relevant documents via existing semantic search
search_response = await nc_semantic_search(
query=query,
ctx=ctx,
limit=limit,
score_threshold=score_threshold,
fusion=fusion,
include_context=include_context,
context_chars=context_chars,
)
# 2. Handle no results case - don't waste a sampling call
if not search_response.results:
logger.debug(f"No documents found for query: {query}")
return SamplingSearchResponse(
query=query,
generated_answer="No relevant documents found in your Nextcloud content for this query.",
sources=[],
total_found=0,
search_method="semantic_sampling",
success=True,
)
# 3. Check if client supports sampling
client_has_sampling = ctx.session.check_client_capability(
ClientCapabilities(sampling=SamplingCapability())
)
# Log capability check result for debugging
logger.info(
f"Sampling capability check: client_has_sampling={client_has_sampling}, "
f"query='{query}'"
)
if hasattr(ctx.session, "_client_params") and ctx.session._client_params:
client_caps = ctx.session._client_params.capabilities
logger.debug(
f"Client advertised capabilities: "
f"roots={client_caps.roots is not None}, "
f"sampling={client_caps.sampling is not None}, "
f"experimental={client_caps.experimental is not None}"
)
if not client_has_sampling:
logger.info(
f"Client does not support sampling (query: '{query}'), "
f"returning {len(search_response.results)} documents"
)
return SamplingSearchResponse(
query=query,
generated_answer=(
f"[Sampling not supported by client]\n\n"
f"Your MCP client doesn't support answer generation. "
f"Found {search_response.total_found} relevant documents. "
f"Please review the sources below."
),
sources=search_response.results,
total_found=search_response.total_found,
search_method="semantic_sampling_unsupported",
success=True,
)
# 4. Fetch full content for notes in parallel.
# Access verification has already happened upstream in
# nc_semantic_search via verify_search_results (ADR-019), so any
# exception here is a sub-second race (doc deleted between
# verification and this fetch) — drop the result in that case.
client = await get_client(ctx)
accessible_results = [None] * len(search_response.results)
full_contents = [None] * len(search_response.results)
# Limit concurrent requests to prevent connection pool exhaustion
max_concurrent = 20
semaphore = anyio.Semaphore(max_concurrent)
async def fetch_content(index: int, result: SemanticSearchResult):
"""Fetch full content for a single document (parallel with semaphore)."""
async with semaphore:
if result.doc_type == "note":
try:
note = await client.notes.get_note(int(result.id))
content = note.get("content", "")
accessible_results[index] = result
full_contents[index] = content
logger.debug(
f"Fetched full content for note {result.id} "
f"(length: {len(content)} chars)"
)
except Exception as e:
# Race window after verify_search_results — drop result.
logger.debug(
"Note %s disappeared between verification and "
"content fetch: %s. Excluding from results.",
result.id,
e,
)
else:
# Non-note types (file, news_item, deck_card) keep the
# excerpt — already access-verified upstream.
accessible_results[index] = result
# full_contents[index] remains None (will use excerpt)
# Run all fetches in parallel using anyio task group
async with anyio.create_task_group() as tg:
for idx, result in enumerate(search_response.results):
tg.start_soon(fetch_content, idx, result)
# Filter out None (inaccessible notes) while preserving order
final_pairs = [
(r, c) for r, c in zip(accessible_results, full_contents) if r is not None
]
accessible_results = [r for r, c in final_pairs]
full_contents = [c for r, c in final_pairs]
# Check if we filtered out all results
if not accessible_results:
logger.warning(f"All search results became inaccessible for query: {query}")
return SamplingSearchResponse(
query=query,
generated_answer="All matching documents are no longer accessible.",
sources=[],
total_found=0,
search_method="semantic_sampling",
success=True,
)
# 5. Construct context from accessible documents with full content
context_parts = []
for idx, (result, content) in enumerate(
zip(accessible_results, full_contents), 1
):
# Use full content if available (notes), otherwise use excerpt
if content is not None:
content_field = f"Content: {content}"
else:
content_field = f"Excerpt: {result.excerpt}"
context_parts.append(
f"[Document {idx}]\n"
f"Type: {result.doc_type}\n"
f"Title: {result.title}\n"
f"Category: {result.category}\n"
f"{content_field}\n"
f"Relevance Score: {result.score:.2f}\n"
)
context = "\n".join(context_parts)
# 6. Construct prompt - reuse user's query, add context and instructions
prompt = (
f"{query}\n\n"
f"Here are relevant documents from Nextcloud (notes, calendar events, deck cards, files, contacts):\n\n"
f"{context}\n\n"
f"Based on the documents above, please provide a comprehensive answer. "
f"Cite the document numbers when referencing specific information."
)
logger.info(
f"Initiating sampling request: query_length={len(query)}, "
f"documents={len(search_response.results)}, "
f"prompt_length={len(prompt)}, max_tokens={max_answer_tokens}"
)
# 6. Request LLM completion via MCP sampling with timeout
# Note: 5 minute timeout to accommodate slower local LLMs (e.g., Ollama)
sampling_timeout_seconds = 300
try:
with anyio.fail_after(sampling_timeout_seconds):
sampling_result = await ctx.session.create_message(
messages=[
SamplingMessage(
role="user",
content=TextContent(type="text", text=prompt),
)
],
max_tokens=max_answer_tokens,
temperature=0.7,
model_preferences=ModelPreferences(
hints=[ModelHint(name="claude-3-5-sonnet")],
intelligencePriority=0.8,
speedPriority=0.5,
),
include_context="thisServer",
)
# 7. Extract answer from sampling response
if sampling_result.content.type == "text":
generated_answer = sampling_result.content.text
else:
# Handle non-text responses (shouldn't happen for text prompts)
generated_answer = f"Received non-text response of type: {sampling_result.content.type}"
logger.warning(
f"Unexpected content type from sampling: {sampling_result.content.type}"
)
logger.info(
f"Sampling successful: model={sampling_result.model}, "
f"stop_reason={sampling_result.stopReason}, "
f"answer_length={len(generated_answer)}"
)
return SamplingSearchResponse(
query=query,
generated_answer=generated_answer,
sources=accessible_results,
total_found=len(accessible_results),
search_method="semantic_sampling",
model_used=sampling_result.model,
stop_reason=sampling_result.stopReason,
success=True,
)
except TimeoutError:
logger.warning(
f"Sampling request timed out after {sampling_timeout_seconds} seconds for query: '{query}', "
f"returning search results only"
)
return SamplingSearchResponse(
query=query,
generated_answer=(
f"[Sampling request timed out]\n\n"
f"The answer generation took too long (>{sampling_timeout_seconds}s). "
f"Found {len(accessible_results)} relevant documents. "
f"Please review the sources below or try a simpler query."
),
sources=accessible_results,
total_found=len(accessible_results),
search_method="semantic_sampling_timeout",
success=True,
)
except McpError as e:
# Expected MCP protocol errors (user rejection, unsupported, etc.)
error_msg = str(e)
if "rejected" in error_msg.lower() or "denied" in error_msg.lower():
# User explicitly declined - this is normal, not an error
logger.info(f"User declined sampling request for query: '{query}'")
search_method = "semantic_sampling_user_declined"
user_message = "User declined to generate an answer"
elif "not supported" in error_msg.lower():
# Client doesn't support sampling - also normal
logger.info(f"Sampling not supported by client for query: '{query}'")
search_method = "semantic_sampling_unsupported"
user_message = "Sampling not supported by this client"
else:
# Other MCP protocol errors
logger.warning(
f"MCP error during sampling for query '{query}': {error_msg}"
)
search_method = "semantic_sampling_mcp_error"
user_message = f"Sampling unavailable: {error_msg}"
return SamplingSearchResponse(
query=query,
generated_answer=(
f"[{user_message}]\n\n"
f"Found {len(accessible_results)} relevant documents. "
f"Please review the sources below."
),
sources=accessible_results,
total_found=len(accessible_results),
search_method=search_method,
success=True,
)
except Exception as e:
# Truly unexpected errors - these SHOULD have tracebacks
logger.error(
f"Unexpected error during sampling for query '{query}': "
f"{type(e).__name__}: {e}",
exc_info=True,
)
return SamplingSearchResponse(
query=query,
generated_answer=(
f"[Unexpected error during sampling]\n\n"
f"Found {len(accessible_results)} relevant documents. "
f"Please review the sources below."
),
sources=accessible_results,
total_found=len(accessible_results),
search_method="semantic_sampling_error",
success=True,
)
@mcp.tool(
title="Check Indexing Status",
annotations=ToolAnnotations(
readOnlyHint=True, # Only checks status
openWorldHint=True,
),
)
@require_scopes("semantic.read")
@instrument_tool
async def nc_get_vector_sync_status(ctx: Context) -> VectorSyncStatusResponse:
"""Get the current vector sync status.
Returns information about the vector sync process, including:
- Number of documents indexed in the vector database
- Number of documents pending processing
- Current sync status (idle, syncing, or disabled)
This is useful for determining when vector indexing is complete
after creating or updating content across all indexed apps.
"""
# Check if vector sync is enabled (supports both old and new env var names)
settings = get_settings()
if not settings.vector_sync_enabled:
return VectorSyncStatusResponse(
indexed_count=0,
pending_count=0,
status="disabled",
enabled=False,
)
try:
# Get document receive stream from lifespan context
lifespan_ctx = ctx.request_context.lifespan_context
document_receive_stream = getattr(
lifespan_ctx, "document_receive_stream", None
)
if document_receive_stream is None:
logger.debug(
"document_receive_stream not available in lifespan context"
)
return VectorSyncStatusResponse(
indexed_count=0,
pending_count=0,
status="unknown",
enabled=True,
)
# Get pending count from stream statistics
stream_stats = document_receive_stream.statistics()
pending_count = stream_stats.current_buffer_used
# Get Qdrant client and query indexed count
indexed_count = 0
try:
qdrant_client = await get_qdrant_client()
# Count documents in collection, excluding placeholders
# Placeholders are zero-vector points used to track processing state
count_result = await qdrant_client.count(
collection_name=settings.get_collection_name(),
count_filter=Filter(must=[get_placeholder_filter()]),
)
indexed_count = count_result.count
except Exception as e:
logger.warning(f"Failed to query Qdrant for indexed count: {e}")
# Continue with indexed_count = 0
# Determine status
status = "syncing" if pending_count > 0 else "idle"
return VectorSyncStatusResponse(
indexed_count=indexed_count,
pending_count=pending_count,
status=status,
enabled=True,
)
except Exception as e:
logger.error(f"Error getting vector sync status: {e}")
raise McpError(
ErrorData(
code=-1,
message=f"Failed to retrieve vector sync status: {str(e)}",
)
)