Use SeekDB as the primary RAG retrieval backend

[LRriver]

Summary

• make native pyseekdb the authoritative store for chunk vectors, vector search, full-text search, and hybrid RAG recall
• keep SQLite focused on business metadata while adding recoverable synchronization, migration, compensation, and interrupted-ingestion recovery
• guard embedding-profile changes, expose accurate storage health, and fail clearly when strict SeekDB requirements are not met
• add real native SeekDB verification to CI and document strict embedding configuration

Verification

• python -m ruff check backend tests
• PYTHONPATH=. pytest -q (180 passed)
• python -m compileall -q backend
• python scripts/verify_seekdb_native.py
• frontend lint, 32 unit tests, typecheck, and production build
• live smoke test with doc/L9.md: 24 chunks, real embedding generation, native SeekDB hybrid recall, real model answer, and 5 citations

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

[gemini-code-assist]

Code Review

This pull request implements a native SeekDB vector and hybrid retrieval backend for knowledge chunks, delegating vector operations from SQLite to a new SeekDBChunkIndex while retaining SQLite for metadata and optional fallback. The review feedback focuses on critical performance and robustness enhancements, including offloading blocking synchronous database initialization to a background thread, optimizing the backfill process to avoid eager deserialization of all chunk payloads, safely handling potential None values in model profiles, adding bounds checking to prevent IndexError during search result deserialization, wrapping health check stats retrieval in a try...except block to avoid unhandled 500 errors, and enabling SQLite WAL mode to prevent database locking under concurrent access.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

The initialize_storage method is defined as async def but performs heavy, synchronous, and blocking I/O operations (such as _recover_pending_sync_operations and _backfill_native_chunks_sync which loads and deserializes all chunks for all sources). Since there are no await points inside this method, it will block the FastAPI event loop, preventing other concurrent requests from being processed.

Consider offloading these heavy synchronous operations to a background thread using asyncio.to_thread to keep the event loop responsive.

    async def initialize_storage(self) -> int:
        """Recover interrupted writes and migrate legacy SQLite vectors into SeekDB."""
        if self._storage_initialized:
            return 0
        import asyncio
        return await asyncio.to_thread(self._initialize_storage_sync)

    def _initialize_storage_sync(self) -> int:
        if self.native_chunk_index is None:
            if self.allow_sqlite_vector_fallback:
                self._reconcile_processing_sources()
                self._storage_initialized = True
            return 0
        probe_native = getattr(self.native_chunk_index, "probe", None)
        if callable(probe_native):
            probe_native()
        self._recover_pending_sync_operations()
        written = 0
        if (
            not self.allow_sqlite_vector_fallback
            and self._get_repository_meta("native_chunk_migration_v2") != "complete"
        ):
            written = self._backfill_native_chunks_sync()
            if not self._migration_has_unresolved_sources:
                self._set_repository_meta("native_chunk_migration_v2", "complete")
        self._reconcile_processing_sources()
        self._storage_initialized = True
        return written

[gemini-code-assist]

In _backfill_native_chunks_sync, the payloads of all chunks for every source are loaded from the database and deserialized into KnowledgeChunk objects (lines 470-474) before checking if the source is already fully reconciled (lines 487-491).

If there are many sources or large documents, this will cause a massive performance bottleneck on startup or during health checks due to unnecessary database I/O and CPU-heavy JSON deserialization.

We can optimize this by checking the reconciliation status using a lightweight SQL query first, and only loading/deserializing the chunk payloads if migration/backfill is actually required for that source.

        reconciled_states = {"seekdb"}
        if self.allow_sqlite_vector_fallback:
            reconciled_states.add("sqlite_fallback_reconciled")
        placeholders = ",".join("?" for _ in reconciled_states)

        for source in sources:
            # Check if the source is already fully reconciled using a lightweight query
            state_row = self._conn.execute(
                "SELECT source_id FROM chunk_index_state WHERE source_id = ?",
                (source.id,),
            ).fetchone()
            if state_row:
                unreconciled_count = self._conn.execute(
                    f"SELECT COUNT(*) FROM chunks WHERE source_id = ? AND vector_state NOT IN ({placeholders})",
                    (source.id, *reconciled_states),
                ).fetchone()[0]
                if unreconciled_count == 0:
                    continue

            sqlite_rows = self._conn.execute(
                "SELECT payload, vector_state FROM chunks WHERE source_id = ? ORDER BY chunk_index ASC",
                (source.id,),
            ).fetchall()
            chunks = [KnowledgeChunk.model_validate_json(row["payload"]) for row in sqlite_rows]
            if not chunks:
                with self._conn:
                    self._upsert_chunk_index_state(
                        source.id,
                        uuid4().hex,
                        [],
                        storage_mode="seekdb",
                    )
                continue

[gemini-code-assist]

If any of the fields profile.adapter, profile.base_url, or profile.model are None (which can happen if they are configured as null in config.yaml or passed as null in API updates), calling .strip() or .rstrip() on them will raise an AttributeError: 'NoneType' object has no attribute 'strip'.

To prevent potential crashes, we should safely handle None values by falling back to empty strings before stripping.

Suggested change

	payload = {
	"adapter": profile.adapter.strip(),
	"base_url": profile.base_url.strip().rstrip("/"),
	"model": profile.model.strip(),
	}
	payload = {
	"adapter": (profile.adapter or "").strip(),
	"base_url": (profile.base_url or "").strip().rstrip("/"),
	"model": (profile.model or "").strip(),
	}

[gemini-code-assist]

In _rows_to_results, accessing metadatas[index] directly without checking if index < len(metadatas) is unsafe. If metadatas is shorter than ids (e.g., due to an incomplete or corrupted database response), this will raise an IndexError and crash the search query.

Additionally, we should ensure metadata is not None and contains the "payload" key before attempting to deserialize it.

Suggested change

	for index, _chunk_id in enumerate(ids):
	try:
	metadata = metadatas[index]
	chunk = KnowledgeChunk.model_validate_json(metadata["payload"])
	except Exception as exc:
	raise ValueError("Malformed SeekDB chunk result payload") from exc
	for index, _chunk_id in enumerate(ids):
	if index >= len(metadatas) or not metadatas[index] or "payload" not in metadatas[index]:
	continue
	try:
	metadata = metadatas[index]
	chunk = KnowledgeChunk.model_validate_json(metadata["payload"])
	except Exception as exc:
	raise ValueError("Malformed SeekDB chunk result payload") from exc

[gemini-code-assist]

The health check endpoint calls vector_store.get_stats() directly without error handling. If the database is locked, SeekDB is unavailable, or any other exception is raised during stats retrieval, the health check will crash with a 500 Internal Server Error instead of returning a structured 503 Service Unavailable response.

To make the health check resilient, we should wrap the vector store calls in a try...except block and report the unhealthy status gracefully.

    vector_store = get_vector_store()
    try:
        stats = await vector_store.get_stats()
        storage = stats.get("storage", {})
        native_available = storage.get("native_available", False)
    except Exception as exc:
        stats = {}
        storage = {}
        native_available = False

[gemini-code-assist]

With the introduction of complex background synchronization, migration, and compensation operations, SQLite is written to and read from much more frequently. By default, SQLite operates in rollback journal mode, which locks the entire database during writes and blocks concurrent reads.

To prevent database is locked errors and improve concurrent performance in FastAPI, consider enabling Write-Ahead Logging (WAL) mode in _init_schema.

        self._conn.execute("PRAGMA journal_mode=WAL;")
        chunk_columns = {
            row["name"] for row in self._conn.execute("PRAGMA table_info(chunks)").fetchall()
        }