fix: return retrievable slugs for symbol hits

Author: ayhammouda

src/mcp_server_python_docs/retrieval/ranker.py

@@ -16,6 +16,89 @@
 logger = logging.getLogger(__name__)
 
 
+def _page_uri(uri: str) -> str:
+    """Return the page part of an objects.inv URI."""
+    return uri.split("#", 1)[0]
+
+
+def _document_candidates(uri: str) -> tuple[str, ...]:
+    """Return possible document slugs for a symbol URI.
+
+    ``objects.inv`` entries use HTML paths such as ``library/json.html``,
+    while Sphinx JSON content is ingested with extensionless slugs such as
+    ``library/json``. Prefer the exact URI first, then the extensionless form.
+    """
+    page_uri = _page_uri(uri)
+    if page_uri.endswith(".html"):
+        return (page_uri, page_uri[:-5])
+    return (page_uri,)
+
+
+def _resolve_symbol_location(
+    conn: sqlite3.Connection,
+    row: sqlite3.Row,
+) -> tuple[str, str]:
+    """Resolve a symbol row to a get_docs-compatible slug and anchor."""
+    uri = str(row["uri"])
+    fallback_slug = _page_uri(uri)
+    fallback_anchor = str(row["anchor"] or "")
+
+    section_id = row["section_id"]
+    if section_id is not None:
+        section_row = conn.execute(
+            """
+            SELECT d.slug, s.anchor
+            FROM sections s
+            JOIN documents d ON s.document_id = d.id
+            WHERE s.id = ?
+            LIMIT 1
+            """,
+            (section_id,),
+        ).fetchone()
+        if section_row is not None:
+            return section_row["slug"], section_row["anchor"] or ""
+
+    doc_row = None
+    document_id = row["document_id"]
+    if document_id is not None:
+        doc_row = conn.execute(
+            "SELECT id, slug FROM documents WHERE id = ? LIMIT 1",
+            (document_id,),
+        ).fetchone()
+
+    if doc_row is None:
+        for candidate in _document_candidates(uri):
+            doc_row = conn.execute(
+                """
+                SELECT id, slug
+                FROM documents
+                WHERE doc_set_id = ? AND (slug = ? OR uri = ?)
+                LIMIT 1
+                """,
+                (row["doc_set_id"], candidate, candidate),
+            ).fetchone()
+            if doc_row is not None:
+                break
+
+    if doc_row is None:
+        return fallback_slug, fallback_anchor
+
+    if fallback_anchor:
+        section_row = conn.execute(
+            """
+            SELECT anchor
+            FROM sections
+            WHERE document_id = ? AND anchor = ?
+            LIMIT 1
+            """,
+            (doc_row["id"], fallback_anchor),
+        ).fetchone()
+        if section_row is not None:
+            return doc_row["slug"], section_row["anchor"] or ""
+
+    return doc_row["slug"], ""
+
+
 def _normalize_scores(hits: list[SymbolHit]) -> list[SymbolHit]:
     """Normalize BM25 scores to [0.1, 1.0] range.
 
@@ -129,7 +212,8 @@ def search_symbols(
     try:
         cursor = conn.execute(
             """
-            SELECT sym.id, sym.qualified_name, sym.symbol_type, sym.uri,
+            SELECT sym.id, sym.doc_set_id, sym.document_id, sym.section_id,
+                   sym.qualified_name, sym.symbol_type, sym.uri,
                    sym.anchor, sym.module, d.version,
                    bm25(symbols_fts, 10.0, 1.0) as score,
                    snippet(symbols_fts, 0, '**', '**', '...', 32) as snippet_text
@@ -148,19 +232,19 @@ def search_symbols(
         logger.warning("FTS5 query failed for symbols: %r", match_expr)
         return []
 
-    hits = [
-        SymbolHit(
+    hits = []
+    for row in rows:
+        slug, anchor = _resolve_symbol_location(conn, row)
+        hits.append(SymbolHit(
             uri=row["uri"],
             title=row["qualified_name"],
             kind=row["symbol_type"] or "symbol",
             snippet=row["snippet_text"] or "",
             score=row["score"],
             version=row["version"],
-            slug=row["uri"].split("#")[0] if "#" in row["uri"] else row["uri"],
-            anchor=row["anchor"] or "",
-        )
-        for row in rows
-    ]
+            slug=slug,
+            anchor=anchor,
+        ))
 
     return _normalize_scores(hits)
 
@@ -251,7 +335,8 @@ def lookup_symbols_exact(
 
     cursor = conn.execute(
         """
-        SELECT s.qualified_name, s.symbol_type, s.uri, s.anchor,
+        SELECT s.doc_set_id, s.document_id, s.section_id,
+               s.qualified_name, s.symbol_type, s.uri, s.anchor,
                s.module, d.version
         FROM symbols s
         JOIN doc_sets d ON s.doc_set_id = d.id
@@ -264,16 +349,17 @@ def lookup_symbols_exact(
     )
     rows = cursor.fetchall()
 
-    return [
-        SymbolHit(
+    hits = []
+    for row in rows:
+        slug, anchor = _resolve_symbol_location(conn, row)
+        hits.append(SymbolHit(
             uri=row["uri"],
             title=row["qualified_name"],
             kind=row["symbol_type"] or "symbol",
             snippet="",
             score=1.0 if row["qualified_name"] == query else 0.8,
             version=row["version"],
-            slug=row["uri"].split("#")[0] if "#" in row["uri"] else row["uri"],
-            anchor=row["anchor"] or "",
-        )
-        for row in rows
-    ]
+            slug=slug,
+            anchor=anchor,
+        ))
+    return hits

tests/test_services.py

@@ -102,6 +102,41 @@ def test_search_symbol_fast_path(self, populated_with_content):
         assert len(result.hits) >= 1
         assert result.hits[0].title == "asyncio.TaskGroup"
 
+    def test_symbol_hit_slug_is_retrievable_when_content_slug_is_extensionless(
+        self,
+        populated_db,
+    ):
+        db = populated_db
+        doc_set_id = db.execute("SELECT id FROM doc_sets LIMIT 1").fetchone()[0]
+        db.execute(
+            "INSERT INTO documents (doc_set_id, uri, slug, title, content_text, char_count) "
+            "VALUES (?, 'library/json', 'library/json', 'json', "
+            "'json page content', 17)",
+            (doc_set_id,),
+        )
+        doc_id = db.execute("SELECT last_insert_rowid()").fetchone()[0]
+        db.execute(
+            "INSERT INTO sections (document_id, uri, anchor, heading, level, ordinal, "
+            "content_text, char_count) VALUES (?, 'library/json', '', 'Introduction', "
+            "1, 0, 'json.dumps guidance for command line tools.', 43)",
+            (doc_id,),
+        )
+        db.execute(
+            "INSERT INTO symbols (doc_set_id, qualified_name, normalized_name, "
+            "symbol_type, uri, anchor, module) VALUES (?, 'json.dumps', "
+            "'json.dumps', 'function', 'library/json.html#json.dumps', "
+            "'json.dumps', 'json')",
+            (doc_set_id,),
+        )
+        db.commit()
+
+        hit = SearchService(db, {}).search("json.dumps", version="3.13", kind="symbol").hits[0]
+
+        assert hit.slug == "library/json"
+        assert hit.anchor == ""
+        docs = ContentService(db).get_docs(hit.slug, hit.version, hit.anchor or None)
+        assert "json.dumps guidance" in docs.content
+
     def test_search_no_results(self, populated_with_content):
         svc = SearchService(populated_with_content, {})
         result = svc.search("nonexistent_xyz_symbol", kind="symbol")