admin/mcp-nextcloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(chunk-context): address PR #767 review — drop dead PDF branch, redundant alias, add boundary tests

Browse Source 
- Remove unreachable doc_type=="file" branch (and pymupdf/pymupdf4llm
  imports) from _fetch_document_text in search/context.py — the file
  path is short-circuited in get_chunk_with_context before reaching it.
- Drop the redundant `username = request.user.display_name` alias in
  auth/viz_routes.py; both Qdrant scroll filters now reference user_id
  consistently with the rest of the handler.
- Add TestAdjacentChunkBoundary in tests/unit/test_chunk_context_offset_gate.py
  covering chunk_index=0 (before-fetch gate closed) and
  chunk_index=total_chunks-1 (after-fetch gate closed) — the two
  off-by-one boundaries previously untested.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-05-09 00:07:51 +02:00
co-authored by Claude Opus 4.7
parent 47b0b737b6
commit c780f96d2b
3 changed files with 97 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nextcloud_mcp_server/auth/viz_routes.py
+2 -3
View File
@@ -632,7 +632,6 @@ async def chunk_context_endpoint(request: Request) -> JSONResponse:
try:
settings = get_settings()
qdrant_client = await get_qdrant_client()
username = request.user.display_name
# Prefer chunk_index for the chunk-bbox lookup (always indexed);
# fall back to (chunk_start_offset, chunk_end_offset) when not provided.
@@ -646,7 +645,7 @@ async def chunk_context_endpoint(request: Request) -> JSONResponse:
key="doc_id", match=MatchValue(value=doc_id_int)
),
FieldCondition(
key="user_id", match=MatchValue(value=username)
key="user_id", match=MatchValue(value=user_id)
),
FieldCondition(
key="chunk_index",
@@ -674,7 +673,7 @@ async def chunk_context_endpoint(request: Request) -> JSONResponse:
key="doc_id", match=MatchValue(value=doc_id_int)
),
FieldCondition(
key="user_id", match=MatchValue(value=username)
key="user_id", match=MatchValue(value=user_id)
),
FieldCondition(
key="chunk_start_offset",
nextcloud_mcp_server/search/context.py
+6 -53
View File
@@ -7,8 +7,6 @@ position markers for better visualization and understanding of search results.
import logging
from dataclasses import dataclass
import pymupdf
import pymupdf4llm
from qdrant_client.models import FieldCondition, Filter, MatchValue
from nextcloud_mcp_server.client import NextcloudClient
@@ -473,10 +471,14 @@ async def _fetch_document_text(
) -> str | None:
"""Fetch full text content of a document.
Note: doc_type=="file" is short-circuited in get_chunk_with_context before
this function is called (re-parsing PDFs is too slow for the request
timeout), so no file branch exists here.
Args:
nc_client: Authenticated Nextcloud client
doc_id: Document ID (note ID or file path)
doc_type: Type of document ("note", "file", etc.)
doc_id: Document ID
doc_type: Type of document ("note", "news_item", "deck_card")
Returns:
Full document text, or None if document cannot be retrieved
@@ -490,55 +492,6 @@ async def _fetch_document_text(
title = note.get("title", "")
content = note.get("content", "")
return f"{title}\n\n{content}"
elif doc_type == "file":
# Fetch file content via WebDAV
try:
file_path = str(doc_id)
file_content, content_type = await nc_client.webdav.read_file(file_path)
# Check if it's a PDF (by content type or file extension)
is_pdf = (
content_type and "pdf" in content_type.lower()
) or file_path.lower().endswith(".pdf")
if is_pdf:
# Extract text from PDF using PyMuPDF
# IMPORTANT: Use pymupdf4llm.to_markdown() to match indexing extraction
# This ensures character offsets align between indexed chunks and retrieval
logger.debug(f"Extracting text from PDF: {file_path}")
pdf_doc = pymupdf.open(stream=file_content, filetype="pdf")
text_parts = []
page_count = pdf_doc.page_count
# Extract each page as markdown (same as indexing)
for page_num in range(page_count):
page_md = pymupdf4llm.to_markdown(
pdf_doc,
pages=[page_num],
write_images=False, # Don't need images for context
page_chunks=False,
)
text_parts.append(page_md)
pdf_doc.close()
# Join pages (no separator - matches indexing)
full_text = "".join(text_parts)
logger.debug(
f"Extracted {len(full_text)} characters from "
f"{page_count} pages in {file_path}"
)
return full_text
else:
# Assume it's a text file, decode to string
logger.debug(f"Decoding text file: {file_path}")
return file_content.decode("utf-8", errors="replace")
except Exception as e:
logger.error(
f"Error fetching file content for {doc_id}: {e}", exc_info=True
)
return None
elif doc_type == "news_item":
# Fetch news item by ID
item = await nc_client.news.get_item(int(doc_id))
tests/unit/test_chunk_context_offset_gate.py
+89
View File
@@ -269,6 +269,95 @@ class TestNullableChunkIndexPropagation:
assert "Chunk ?/10" in result.marked_text
class TestAdjacentChunkBoundary:
"""Boundary cases for the `chunk_index > 0` / `chunk_index < total_chunks - 1`
gates that decide whether to fetch the previous / next chunk via Qdrant.
See PR #767 review (🟡 missing boundary tests).
"""
async def test_first_chunk_skips_before_fetch_only(self, mock_nc_client):
"""At chunk_index=0 the before-fetch gate is closed (no previous
chunk exists) but the after-fetch still runs.
"""
with (
patch.object(
context_module,
"_get_chunk_by_index_from_qdrant",
new_callable=AsyncMock,
side_effect=[
"current chunk text", # primary lookup
"next chunk text", # adjacent after only
],
) as mock_indexed,
patch.object(
context_module,
"_get_chunk_from_qdrant",
new_callable=AsyncMock,
return_value=None,
),
):
result = await get_chunk_with_context(
nc_client=mock_nc_client,
user_id="alice",
doc_id=42,
doc_type="note",
chunk_start=0,
chunk_end=10,
chunk_index=0,
total_chunks=10,
)
assert result is not None
assert result.chunk_index == 0
assert result.has_before_truncation is False
assert result.has_after_truncation is False
assert mock_indexed.await_count == 2, (
"expected primary lookup + after-fetch only (no before-fetch at index 0)"
)
assert "Chunk 1 of 10" in result.marked_text
async def test_last_chunk_skips_after_fetch_only(self, mock_nc_client):
"""At chunk_index=total_chunks-1 the after-fetch gate is closed (no
next chunk exists) but the before-fetch still runs.
"""
with (
patch.object(
context_module,
"_get_chunk_by_index_from_qdrant",
new_callable=AsyncMock,
side_effect=[
"current chunk text", # primary lookup
"previous chunk text", # adjacent before only
],
) as mock_indexed,
patch.object(
context_module,
"_get_chunk_from_qdrant",
new_callable=AsyncMock,
return_value=None,
),
):
result = await get_chunk_with_context(
nc_client=mock_nc_client,
user_id="alice",
doc_id=42,
doc_type="note",
chunk_start=0,
chunk_end=10,
chunk_index=9,
total_chunks=10,
)
assert result is not None
assert result.chunk_index == 9
assert result.has_before_truncation is False
assert result.has_after_truncation is False
assert mock_indexed.await_count == 2, (
"expected primary lookup + before-fetch only (no after-fetch at last index)"
)
assert "Chunk 10 of 10" in result.marked_text
class TestPositionMarkers:
"""Direct tests for `_insert_position_markers` rendering when chunk_index
is None vs explicit.