fix: persist symbol-only build mode so validate-corpus respects it

publish_index now records build_mode=symbol_only in ingestion_runs.notes
when require_content=False. validate-corpus reads this flag to auto-detect
symbol-only indexes instead of always enforcing full-content checks.

Also fixes read-only SQLite regression (PRAGMA WAL on RO connections),
adds symbol-only smoke test support with require_content parameter,
and consolidates connection helpers via get_readonly_connection.

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

Author: ayhammouda

src/mcp_server_python_docs/__main__.py

@@ -81,7 +81,7 @@ def serve() -> None:
 @click.option(
     "--skip-content",
     is_flag=True,
-    help="Skip Sphinx JSON build, only ingest objects.inv symbols",
+    help="Skip Sphinx JSON build and publish a symbol-only index (search_docs only).",
 )
 def build_index(versions: str, skip_content: bool) -> None:
     """Build the documentation index from objects.inv and Sphinx JSON."""
@@ -239,7 +239,8 @@ def build_index(versions: str, skip_content: bool) -> None:
                         )
                         logger.warning(
                             "Version %s has SYMBOLS ONLY (sphinx-build failed). "
-                            "search_docs will work but get_docs will return empty pages.",
+                            "search_docs will work but get_docs will fail until content "
+                            "ingestion succeeds.",
                             version,
                         )
                         any_version_succeeded = True
@@ -301,7 +302,7 @@ def build_index(versions: str, skip_content: bool) -> None:
 
     # Publish: smoke test + atomic swap (PUBL-01 through PUBL-05)
     versions_str = ",".join(version_list)
-    success = publish_index(build_db_path, versions_str)
+    success = publish_index(build_db_path, versions_str, require_content=not skip_content)
     if not success:
         logger.error("Publishing failed — smoke tests did not pass")
         raise SystemExit(1)
@@ -323,7 +324,7 @@ def validate_corpus(db_path: str | None) -> None:
     from pathlib import Path
 
     from mcp_server_python_docs.ingestion.publish import run_smoke_tests
-    from mcp_server_python_docs.storage.db import get_index_path
+    from mcp_server_python_docs.storage.db import get_index_path, get_readonly_connection
 
     if db_path is not None:
         target = Path(db_path)
@@ -337,7 +338,22 @@ def validate_corpus(db_path: str | None) -> None:
 
     logger.info("Validating corpus at %s", target)
 
-    passed, messages = run_smoke_tests(target)
+    # Auto-detect symbol-only builds from the last published ingestion run
+    require_content = True
+    try:
+        ro_conn = get_readonly_connection(target)
+        row = ro_conn.execute(
+            "SELECT notes FROM ingestion_runs "
+            "WHERE status = 'published' ORDER BY id DESC LIMIT 1"
+        ).fetchone()
+        ro_conn.close()
+        if row and row[0] and "build_mode=symbol_only" in row[0]:
+            require_content = False
+            logger.info("Detected symbol-only build — skipping content checks")
+    except Exception:
+        pass  # If we can't read the metadata, default to full validation
+
+    passed, messages = run_smoke_tests(target, require_content=require_content)
 
     for msg in messages:
         if msg.startswith("OK:"):

src/mcp_server_python_docs/ingestion/publish.py

@@ -90,19 +90,23 @@ def record_ingestion_run(
     return cursor.lastrowid  # type: ignore[return-value]
 
 
-def run_smoke_tests(db_path: Path) -> tuple[bool, list[str]]:
+def run_smoke_tests(
+    db_path: Path,
+    *,
+    require_content: bool = True,
+) -> tuple[bool, list[str]]:
     """Run smoke tests against a newly built database (PUBL-03).
 
     Validates that the index has sufficient data to be useful:
     - doc_sets table has at least 1 row
-    - documents table has at least 10 rows
-    - sections table has at least 50 rows
     - symbols table has at least 1000 rows
-    - Spot-check: an asyncio-related document exists
-    - FTS5 check: sections_fts is searchable
+    - For content builds: documents/sections are populated and sections_fts is searchable
+    - For symbol-only builds: symbols_fts is searchable
 
     Args:
         db_path: Path to the database to test.
+        require_content: When True, enforce document/section checks suitable for
+            full-content builds. When False, validate a symbol-only build.
 
     Returns:
         Tuple of (passed, messages). ``passed`` is True only if ALL
@@ -125,22 +129,6 @@ def run_smoke_tests(db_path: Path) -> tuple[bool, list[str]]:
             messages.append(f"FAIL: doc_sets: {count} rows (need >= 1)")
             passed = False
 
-        # Check documents
-        count = conn.execute("SELECT COUNT(*) FROM documents").fetchone()[0]
-        if count >= 10:
-            messages.append(f"OK: documents: {count} rows")
-        else:
-            messages.append(f"FAIL: documents: {count} rows (need >= 10)")
-            passed = False
-
-        # Check sections
-        count = conn.execute("SELECT COUNT(*) FROM sections").fetchone()[0]
-        if count >= 50:
-            messages.append(f"OK: sections: {count} rows")
-        else:
-            messages.append(f"FAIL: sections: {count} rows (need >= 50)")
-            passed = False
-
         # Check symbols
         count = conn.execute("SELECT COUNT(*) FROM symbols").fetchone()[0]
         if count >= 1000:
@@ -149,31 +137,64 @@ def run_smoke_tests(db_path: Path) -> tuple[bool, list[str]]:
             messages.append(f"FAIL: symbols: {count} rows (need >= 1000)")
             passed = False
 
-        # Spot-check: asyncio document exists
-        row = conn.execute(
-            "SELECT 1 FROM documents WHERE slug LIKE '%asyncio%' LIMIT 1"
-        ).fetchone()
-        if row:
-            messages.append("OK: spot-check: asyncio document found")
-        else:
-            messages.append("FAIL: spot-check: no asyncio document found")
-            passed = False
+        if require_content:
+            # Check documents
+            count = conn.execute("SELECT COUNT(*) FROM documents").fetchone()[0]
+            if count >= 10:
+                messages.append(f"OK: documents: {count} rows")
+            else:
+                messages.append(f"FAIL: documents: {count} rows (need >= 10)")
+                passed = False
 
-        # FTS5 check: sections_fts is searchable
-        try:
+            # Check sections
+            count = conn.execute("SELECT COUNT(*) FROM sections").fetchone()[0]
+            if count >= 50:
+                messages.append(f"OK: sections: {count} rows")
+            else:
+                messages.append(f"FAIL: sections: {count} rows (need >= 50)")
+                passed = False
+
+            # Spot-check: asyncio document exists
             row = conn.execute(
-                "SELECT 1 FROM sections_fts WHERE sections_fts MATCH '\"asyncio\"' LIMIT 1"
+                "SELECT 1 FROM documents WHERE slug LIKE '%asyncio%' LIMIT 1"
             ).fetchone()
             if row:
-                messages.append("OK: fts5: sections_fts searchable")
+                messages.append("OK: spot-check: asyncio document found")
             else:
-                messages.append(
-                    "WARN: fts5: sections_fts has no asyncio matches"
-                    " (may be OK for partial builds)"
-                )
-        except sqlite3.OperationalError as e:
-            messages.append(f"FAIL: fts5: sections_fts query failed: {e}")
-            passed = False
+                messages.append("FAIL: spot-check: no asyncio document found")
+                passed = False
+
+            # FTS5 check: sections_fts is searchable
+            try:
+                row = conn.execute(
+                    'SELECT 1 FROM sections_fts WHERE sections_fts MATCH \'"asyncio"\' LIMIT 1'
+                ).fetchone()
+                if row:
+                    messages.append("OK: fts5: sections_fts searchable")
+                else:
+                    messages.append(
+                        "WARN: fts5: sections_fts has no asyncio matches"
+                        " (may be OK for partial builds)"
+                    )
+            except sqlite3.OperationalError as e:
+                messages.append(f"FAIL: fts5: sections_fts query failed: {e}")
+                passed = False
+        else:
+            messages.append("OK: content checks skipped for symbol-only build")
+            try:
+                row = conn.execute(
+                    'SELECT 1 FROM symbols_fts WHERE symbols_fts MATCH \'"asyncio"\' LIMIT 1'
+                ).fetchone()
+                if row:
+                    messages.append("OK: fts5: symbols_fts searchable")
+                else:
+                    messages.append(
+                        "WARN: fts5: symbols_fts has no asyncio matches"
+                        " (unexpected for stdlib builds)"
+                    )
+            except sqlite3.OperationalError as e:
+                messages.append(f"FAIL: fts5: symbols_fts query failed: {e}")
+                passed = False
 
     except Exception as e:
         messages.append(f"FAIL: Unexpected error during smoke tests: {e}")
@@ -259,7 +280,12 @@ def print_restart_message() -> None:
     )
 
 
-def publish_index(build_db_path: Path, version: str) -> bool:
+def publish_index(
+    build_db_path: Path,
+    version: str,
+    *,
+    require_content: bool = True,
+) -> bool:
     """Orchestrate the full publish pipeline.
 
     1. Compute SHA256 of the build artifact
@@ -271,6 +297,7 @@ def publish_index(build_db_path: Path, version: str) -> bool:
     Args:
         build_db_path: Path to the build artifact database.
         version: Version string for the ingestion run record.
+        require_content: Whether publish validation should require content tables.
 
     Returns:
         True if publishing succeeded, False if smoke tests failed.
@@ -282,6 +309,7 @@ def publish_index(build_db_path: Path, version: str) -> bool:
     # Record ingestion run
     from mcp_server_python_docs.storage.db import get_readwrite_connection
 
+    build_notes = "build_mode=symbol_only" if not require_content else None
     conn = get_readwrite_connection(build_db_path)
     try:
         run_id = record_ingestion_run(
@@ -290,12 +318,13 @@ def publish_index(build_db_path: Path, version: str) -> bool:
             version=version,
             status="smoke_testing",
             artifact_hash=artifact_hash,
+            notes=build_notes,
         )
     finally:
         conn.close()
 
     # Run smoke tests (PUBL-03)
-    passed, messages = run_smoke_tests(build_db_path)
+    passed, messages = run_smoke_tests(build_db_path, require_content=require_content)
     for msg in messages:
         logger.info("Smoke test: %s", msg)
 
@@ -314,13 +343,13 @@ def publish_index(build_db_path: Path, version: str) -> bool:
         logger.error("Smoke tests failed — not publishing")
         return False
 
-    # Update run status to published
+    # Update run status to published (preserve build_mode note)
     conn = get_readwrite_connection(build_db_path)
     try:
         conn.execute(
-            "UPDATE ingestion_runs SET status = ?, "
+            "UPDATE ingestion_runs SET status = ?, notes = ?, "
             "finished_at = CURRENT_TIMESTAMP WHERE id = ?",
-            ("published", run_id),
+            ("published", build_notes, run_id),
         )
         conn.commit()
     finally:

src/mcp_server_python_docs/server.py

@@ -27,6 +27,7 @@
 from mcp_server_python_docs.services.content import ContentService
 from mcp_server_python_docs.services.search import SearchService
 from mcp_server_python_docs.services.version import VersionService
+from mcp_server_python_docs.storage.db import get_readonly_connection
 
 logger = logging.getLogger(__name__)
 
@@ -72,11 +73,7 @@ async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
     logger.info("Loaded %d synonym entries", len(synonyms))
 
     # Open read-only connection (STOR-06, STOR-07)
-    db = sqlite3.connect(f"file:{index_path}?mode=ro", uri=True, check_same_thread=False)
-    db.execute("PRAGMA journal_mode = WAL")
-    db.execute("PRAGMA synchronous = NORMAL")
-    db.execute("PRAGMA foreign_keys = ON")
-    db.row_factory = sqlite3.Row
+    db = get_readonly_connection(index_path)
 
     # Check FTS5 (STOR-08)
     _assert_fts5(db)

src/mcp_server_python_docs/storage/db.py

@@ -28,21 +28,31 @@ def get_index_path() -> Path:
     return get_cache_dir() / "index.db"
 
 
-def _set_pragmas(conn: sqlite3.Connection) -> None:
-    """Set required PRAGMAs on a connection (STOR-07)."""
-    conn.execute("PRAGMA journal_mode = WAL")
+def _set_common_pragmas(conn: sqlite3.Connection) -> None:
+    """Set PRAGMAs that are safe for both read-only and read-write connections."""
     conn.execute("PRAGMA synchronous = NORMAL")
     conn.execute("PRAGMA foreign_keys = ON")
 
 
+def _set_readonly_pragmas(conn: sqlite3.Connection) -> None:
+    """Set PRAGMAs for read-only serving connections (STOR-07)."""
+    _set_common_pragmas(conn)
+
+
+def _set_readwrite_pragmas(conn: sqlite3.Connection) -> None:
+    """Set PRAGMAs for read-write ingestion connections (STOR-07)."""
+    conn.execute("PRAGMA journal_mode = WAL")
+    _set_common_pragmas(conn)
+
+
 def get_readonly_connection(path: str | Path) -> sqlite3.Connection:
     """Open a read-only connection for serving (STOR-06).
 
     Uses SQLite URI mode with ?mode=ro to prevent accidental writes.
     """
     path = Path(path)
     conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True, check_same_thread=False)
-    _set_pragmas(conn)
+    _set_readonly_pragmas(conn)
     conn.row_factory = sqlite3.Row
     return conn
 
@@ -56,7 +66,7 @@ def get_readwrite_connection(path: str | Path) -> sqlite3.Connection:
     path = Path(path)
     path.parent.mkdir(parents=True, exist_ok=True)
     conn = sqlite3.connect(str(path))
-    _set_pragmas(conn)
+    _set_readwrite_pragmas(conn)
     conn.row_factory = sqlite3.Row
     return conn