From e8dd02f2b167c8df09c623bc588a21aaa9f6b644 Mon Sep 17 00:00:00 2001 From: Aymen Hammouda Date: Mon, 5 Oct 2026 16:18:16 +0200 Subject: [PATCH] agent: append canonical targets without replacing search results --- README.md | 14 ++ src/mcp_server_python_docs/services/search.py | 62 +++++-- tests/test_search_spare_slot.py | 167 ++++++++++++++++++ 3 files changed, 231 insertions(+), 12 deletions(-) create mode 100644 tests/test_search_spare_slot.py diff --git a/README.md b/README.md index 13b033c..cc28e94 100644 --- a/README.md +++ b/README.md @@ -98,6 +98,20 @@ pipx install python-docs-mcp-server If `uv` is installed but the `uv` command is not on your `PATH`, reopen your shell or use `python -m uv ...` as a fallback for local contributor commands. +## Search result behavior + +For `kind="auto"` with an explicit version, a nonempty search with spare result +slots may append **one** exact inventory target when the prompt contains one +known qualified identifier in that version. Original hits, snippets, scores, +order and notes are preserved. Full results, ambiguous identifiers, other kinds +and versionless queries are unchanged; the existing empty-result fallback is +unchanged. Canonical locations are deduplicated, including inventory aliases. + +The append is admitted only when the **complete proposed response** serialized +by Pydantic's compact `model_dump_json()` is at most **8000 UTF-8 bytes**. This is +not a global search cap: oversized original responses remain unchanged. It is +separate from `get_docs(max_chars=8000)`, which limits content **characters**. + ## First run Build the local documentation index: diff --git a/src/mcp_server_python_docs/services/search.py b/src/mcp_server_python_docs/services/search.py index 0a35297..37ffdfb 100644 --- a/src/mcp_server_python_docs/services/search.py +++ b/src/mcp_server_python_docs/services/search.py @@ -8,8 +8,9 @@ import re import sqlite3 +from urllib.parse import urlsplit -from mcp_server_python_docs.models import SearchDocsResult +from mcp_server_python_docs.models import SearchDocsResult, SymbolHit from mcp_server_python_docs.retrieval.query import ( build_match_expression, classify_query, @@ -58,6 +59,47 @@ def _symbol_exists(self, name: str) -> bool: ).fetchone() return row is not None + def _known_identifiers(self, query: str, version: str) -> set[str]: + """Count exact prompt identifiers only in the requested inventory.""" + return { + name + for name in set(_DOTTED_IDENTIFIER.findall(query)) + if self._db.execute( + "SELECT 1 FROM symbols s JOIN doc_sets d ON d.id = s.doc_set_id " + "WHERE s.qualified_name = ? AND d.version = ? LIMIT 1", + (name, version), + ).fetchone() is not None + } + + def _append_canonical( + self, original: SearchDocsResult, query: str, version: str | None, + kind: str, max_results: int, + ) -> SearchDocsResult: + """Fill at most one spare slot without altering semantic evidence.""" + if kind != "auto" or version is None or not original.hits: + return original + if len(original.hits) >= max_results: + return original + known = self._known_identifiers(query, version) + if len(known) != 1: + return original + name = known.pop() + candidates = lookup_symbols_exact(self._db, name, version, 1) + if not candidates or candidates[0].title != name: + return original + candidate = candidates[0] + + def location(hit: SymbolHit) -> tuple[str, str, str]: + uri = urlsplit(hit.uri) + return (hit.version, (hit.slug or uri.path).removesuffix(".html"), + hit.anchor or uri.fragment) + + if any(location(hit) == location(candidate) for hit in original.hits): + return original + proposed = original.model_copy(update={"hits": [*original.hits, candidate]}) + # Admission only: never cap, replace or truncate the original response. + return proposed if len(proposed.model_dump_json().encode("utf-8")) <= 8000 else original + @log_tool_call("search_docs") def search( self, @@ -87,7 +129,9 @@ def search( hits = lookup_symbols_exact(self._db, query, resolved_version, max_results) if hits: self._last_resolution = "exact" - return SearchDocsResult(hits=hits) + return self._append_canonical( + SearchDocsResult(hits=hits), query, resolved_version, kind, max_results, + ) # Fall through to FTS if symbol lookup found nothing and kind is auto if kind == "symbol": self._last_resolution = "exact" @@ -113,15 +157,7 @@ def search( self._last_resolution = "fts" if not hits and kind == "auto" and resolved_version is not None: # Prompt-only candidates; ambiguity is decided in the requested inventory. - known = { - name - for name in set(_DOTTED_IDENTIFIER.findall(query)) - if self._db.execute( - "SELECT 1 FROM symbols s JOIN doc_sets d ON d.id = s.doc_set_id " - "WHERE s.qualified_name = ? AND d.version = ? LIMIT 1", - (name, resolved_version), - ).fetchone() is not None - } + known = self._known_identifiers(query, resolved_version) if len(known) == 1: name = known.pop() hits = lookup_symbols_exact(self._db, name, resolved_version, max_results) @@ -131,4 +167,6 @@ def search( hits=hits, note=f"No full-query results; showing symbol matches for {name!r}.", ) - return SearchDocsResult(hits=hits) + return self._append_canonical( + SearchDocsResult(hits=hits), query, resolved_version, kind, max_results, + ) diff --git a/tests/test_search_spare_slot.py b/tests/test_search_spare_slot.py new file mode 100644 index 0000000..80ca8a0 --- /dev/null +++ b/tests/test_search_spare_slot.py @@ -0,0 +1,167 @@ +"""Additive canonical admission preserves semantic originals and byte budgets.""" +from __future__ import annotations + +import pytest + +from mcp_server_python_docs.errors import VersionNotFoundError +from mcp_server_python_docs.models import SearchDocsResult, SymbolHit +from mcp_server_python_docs.retrieval.ranker import lookup_symbols_exact +from mcp_server_python_docs.services.search import SearchService + + +@pytest.fixture +def semantic_service(stability_db): + doc = stability_db.execute( + "SELECT id FROM documents WHERE slug = 'library/json.html'" + ).fetchone()[0] + for ordinal, anchor, text in [ + (3, "overview", "json.dumps settings overview for encoding"), + (4, "release-notes", "Release changes to json.dumps settings compatibility"), + ]: + stability_db.execute( + "INSERT INTO sections (document_id, uri, anchor, heading, level, ordinal, " + "content_text, char_count) VALUES (?, ?, ?, ?, 2, ?, ?, ?)", + (doc, f"library/json.html#{anchor}", anchor, anchor, ordinal, text, len(text)), + ) + stability_db.execute("INSERT INTO sections_fts(sections_fts) VALUES('rebuild')") + stability_db.commit() + return SearchService(stability_db, {}) + + +def test_real_search_retains_semantic_prefix_and_appends_once(semantic_service): + svc = semantic_service + query = "json.dumps settings" + original = svc.search(query, "3.13", "section") + assert len(original.hits) == 2 + assert {h.anchor for h in original.hits} == {"overview", "release-notes"} + result = svc.search(query, "3.13") + assert result.hits[:-1] == original.hits + assert result.note == original.note + assert result.hits[-1] == lookup_symbols_exact(svc._db, "json.dumps", "3.13", 1)[0] + assert len(svc.search(query, "3.13", max_results=3).hits) == 3 + assert svc.search(query, "3.13", max_results=2) == original + assert svc.search(query, "3.13", max_results=1).hits == original.hits[:1] + assert svc._last_resolution == "fts" + + +@pytest.mark.parametrize("query", [ + "json.dump", "json.dumps.missing", "unknown.missing", "json.dumps json.loads", + "éjson.dumps", "json.dumpsé", "", " ", +]) +def test_unknown_partial_ambiguous_unicode_identifiers_do_not_append(search_service, query): + original = SearchDocsResult(hits=[SymbolHit(uri="other.html", title="Semantic", + kind="section", version="3.13")]) + assert search_service._append_canonical(original, query, "3.13", "auto", 5) is original + + +@pytest.mark.parametrize("kind,version,limit", [ + ("symbol", "3.13", 5), ("page", "3.13", 5), ("section", "3.13", 5), + ("example", "3.13", 5), ("auto", None, 5), ("auto", "3.13", 1), +]) +def test_gates_preserve_original_object(search_service, kind, version, limit): + original = SearchDocsResult(hits=[SymbolHit(uri="other.html", title="Semantic", + kind="section", version="3.13")], note="Keep note") + result = search_service._append_canonical(original, "json.dumps", version, kind, limit) + assert result is original + empty = SearchDocsResult(note="Empty") + assert search_service._append_canonical(empty, "json.dumps", "3.13", "auto", 5) is empty + + +@pytest.mark.parametrize("suffix", ["", " alongside unknown.missing", " json.dumps()"]) +def test_single_known_identifier_and_original_identity(search_service, suffix): + original = SearchDocsResult(hits=[SymbolHit(uri="other.html", title="Semantic", + kind="section", version="3.13", score=0.37)], + note='Keep "note" 😀') + result = search_service._append_canonical(original, "Explain json.dumps" + suffix, + "3.13", "auto", 5) + assert len(result.hits) == 2 + assert result.hits[0] is original.hits[0] + assert result.note == original.note + assert original.hits[0].score == 0.37 + assert len(original.hits) == 1 + + +def test_canonical_location_alias_and_extension_dedup(search_service): + candidate = lookup_symbols_exact(search_service._db, "json.dumps", "3.13", 1)[0] + # An alias/semantic title and extensionless get_docs slug share the same location. + alias = candidate.model_copy(update={"title": "Alias", "slug": "library/json", + "anchor": None}) + original = SearchDocsResult(hits=[alias]) + assert search_service._append_canonical(original, "json.dumps", "3.13", "auto", 5) is original + assert search_service.search("json.dumps", "3.13").hits == [candidate] + + +def test_requested_version_inventory_and_location(search_service): + db = search_service._db + db.execute("INSERT INTO doc_sets (source, version, language, label, base_url) " + "VALUES ('python-docs', '3.12', 'en', 'Python 3.12', 'https://docs.python.org/3.12/')") + db.execute("INSERT INTO symbols (doc_set_id, qualified_name, normalized_name, symbol_type, " + "uri, anchor, module) SELECT d.id, s.qualified_name, s.normalized_name, " + "s.symbol_type, s.uri, s.anchor, s.module FROM symbols s CROSS JOIN doc_sets d " + "WHERE s.qualified_name = 'json.dumps' AND d.version = '3.12'") + db.commit() + original = SearchDocsResult(hits=[SymbolHit(uri="other.html", title="Semantic", + kind="section", version="3.12")]) + result = search_service._append_canonical(original, "json.dumps json.loads", "3.12", "auto", 5) + assert len(result.hits) == 2 and all(h.version == "3.12" for h in result.hits) + assert search_service._append_canonical(original, "json.loads", "3.12", "auto", 5) is original + with pytest.raises(VersionNotFoundError): + search_service.search("json.dumps settings", "9.99") + + +@pytest.mark.parametrize("extra_bytes,admitted", [(0, True), (1, False), (9000, False)]) +def test_full_compact_utf8_budget_exact_fit_one_byte_short_and_oversized( + search_service, extra_bytes, admitted, +): + original = SearchDocsResult(hits=[SymbolHit( + uri="other.html", title='Semantic "quoted"', kind="section", version="3.13", + snippet='😀 café 漢字 "quote" \\escape\n\t', score=0.41, + )], note='Preserve "note" \\ 😀') + candidate = lookup_symbols_exact(search_service._db, "json.dumps", "3.13", 1)[0] + proposed = original.model_copy(update={"hits": [*original.hits, candidate]}) + padding = 8000 - len(proposed.model_dump_json().encode("utf-8")) + extra_bytes + original.hits[0].snippet += "x" * padding + proposed = original.model_copy(update={"hits": [*original.hits, candidate]}) + assert len(proposed.model_dump_json().encode("utf-8")) == 8000 + extra_bytes + assert len(proposed.model_dump_json()) < len(proposed.model_dump_json().encode("utf-8")) + before = original.model_dump_json() + result = search_service._append_canonical(original, "json.dumps", "3.13", "auto", 5) + assert original.model_dump_json() == before + if admitted: + assert result.model_dump_json() == proposed.model_dump_json() + assert result.hits[0] is original.hits[0] + else: + assert result is original + if extra_bytes == 9000: + assert len(original.model_dump_json().encode("utf-8")) > 8000 + + +def test_inventory_alias_resolves_to_existing_section(search_service): + db = search_service._db + section = db.execute( + "SELECT id, document_id FROM sections WHERE anchor = 'json.dumps'", + ).fetchone() + db.execute( + "INSERT INTO symbols (doc_set_id, qualified_name, normalized_name, symbol_type, " + "uri, anchor, module, document_id, section_id) " + "VALUES (1, 'json.encode_alias', 'json_encode_alias', 'function', " + "'library/alias.html#alias', 'alias', 'json', ?, ?)", + (section['document_id'], section['id']), + ) + db.commit() + existing = lookup_symbols_exact(db, "json.dumps", "3.13", 1)[0] + original = SearchDocsResult(hits=[existing]) + assert search_service._append_canonical( + original, "json.encode_alias", "3.13", "auto", 5, + ) is original + alias = lookup_symbols_exact(db, "json.encode_alias", "3.13", 1)[0] + assert (alias.slug, alias.anchor) == (existing.slug, existing.anchor) + + +def test_oversized_candidate_does_not_replace_original(search_service): + search_service._db.execute( + "UPDATE symbols SET uri = ? WHERE qualified_name = 'json.dumps'", ("x" * 9000,), + ) + original = SearchDocsResult(hits=[SymbolHit(uri="other.html", title="Semantic", + kind="section", version="3.13")]) + assert search_service._append_canonical(original, "json.dumps", "3.13", "auto", 5) is original