feat: FTS5 lazy migration + CRUD sync + nuclear rebuild swap - #167
Merged
Conversation
Task 05 - makes FTS5 fast-path actually work in production. - Lazy migration: background thread rebuilds FTS5 index from ChromaDB on first access if fts5_index.db is empty; first query does not block and falls back to hybrid gracefully while rebuild runs. - CRUD sync: _index_document / remove_document_by_path / update_document_content now update FTS5 incrementally, guarded by config.fts5_enabled. - nuclear_rebuild swap: FTS5 file swap atomic with ChromaDB collection swap (integrates with v4.8.0 zero-downtime staging pattern). - Standalone script scripts/build_fts5_index.py for manual migration. - Ops runbook docs/runbooks/fts5_migration.md. - 18 new tests in tests/test_fts5_migration.py + 24 lines in test_e2e_fts5.py. Test count baseline 308 -> 314 (+6). LEI 1 preserved - internal helpers only, no MCP tool signature change. Fase 4 of FTS5 Lexical Fast-Path plan.
📝 WalkthroughWalkthroughFTS5 now supports resumable background migration, progress metrics, synchronized document CRUD, rebuild workflows, and readiness handling. The server integrates migration with ChromaDB operations, and tests cover lifecycle, failure, resume, synchronization, and forced-search behavior. ChangesFTS5 migration lifecycle
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant KnowledgeOrchestrator
participant ChromaDB
participant Fts5LexicalIndex
participant PrometheusGauges
KnowledgeOrchestrator->>ChromaDB: read document count and sorted chunk rows
KnowledgeOrchestrator->>Fts5LexicalIndex: start migration with checkpoint
Fts5LexicalIndex->>ChromaDB: iterate chunk rows
Fts5LexicalIndex->>Fts5LexicalIndex: insert batches and persist checkpoints
Fts5LexicalIndex->>PrometheusGauges: publish migration progress
Fts5LexicalIndex-->>KnowledgeOrchestrator: mark migration complete and ready
Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
Comment |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
Comment on lines
+404
to
+413
| def _populate_batch(self, rows: Sequence[ChunkRow]) -> None: | ||
| """Insert a batch under ``_fts5_lock``. Raises on SQL failure.""" | ||
| if self._conn is None: | ||
| raise Fts5MigrationError("FTS5 connection is closed during migration") | ||
| with self._fts5_lock: | ||
| self._conn.executemany( | ||
| "INSERT INTO fts5_documents (chunk_id, content, filename, category) VALUES (?, ?, ?, ?)", | ||
| rows, | ||
| ) | ||
| self._conn.commit() |
There was a problem hiding this comment.
If a document is updated or removed after migration materializes its Chroma snapshot, the later unconditional batch insert restores the stale FTS5 row after CRUD synchronization, causing obsolete terms, duplicate rankings, or empty results for deleted chunk IDs.
Prompt To Fix With AI
This is a comment left during a code review.
Path: mcp_server/fts5_index.py
Line: 404-413
Comment:
**Stale snapshots overwrite CRUD state**
If a document is updated or removed after migration materializes its Chroma snapshot, the later unconditional batch insert restores the stale FTS5 row after CRUD synchronization, causing obsolete terms, duplicate rankings, or empty results for deleted chunk IDs.
---
For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 14
🧹 Nitpick comments (1)
mcp_server/server.py (1)
2394-2430: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider a public readiness API instead of three private accesses.
This method reaches into
_write_state,_fts5_lock, and_readywith threenoqa: SLF001suppressions.scripts/build_fts5_index.pyperforms the same private access. A small public method onFts5LexicalIndex, for examplemark_empty_corpus_complete(), would encapsulate the marker write and the readiness flip in one place and remove the duplicated invariant from two callers.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@mcp_server/server.py` around lines 2394 - 2430, add a public Fts5LexicalIndex method such as mark_empty_corpus_complete() that atomically writes the complete empty-corpus state and marks the index ready. Update _maybe_start_fts5_migration and scripts/build_fts5_index.py to call this method instead of accessing _write_state, _fts5_lock, or _ready directly, removing the SLF001 suppressions.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@docs/runbooks/fts5_migration.md`:
- Around line 55-60: The failed-marker recovery description in
docs/runbooks/fts5_migration.md must state that daemon startup begins a fresh
background migration when status is "failed"; retain manual rebuild only as an
optional recovery path for corruption or operator-controlled rebuilds.
README.md:1417-1417 requires no direct change because its release note already
matches runtime behavior.
In `@mcp_server/fts5_index.py`:
- Around line 259-300: Move the self._conn None checks inside _fts5_lock in
add_document, remove_document, and update_document, then bind the validated
connection to a local variable and use it for every SQL statement and commit.
Apply the same locking and local-connection pattern in mcp_server/fts5_index.py
lines 404-413 within _populate_batch, preserving Fts5MigrationError there.
- Around line 359-391: Make _run_migration_batches replay-safe by removing
existing FTS5 rows for each row’s chunk_id before _populate_batch inserts the
batch, or otherwise enforce equivalent idempotent insertion. Preserve progress
accounting and callbacks, and leave _maybe_start_fts5_migration unchanged; the
earlier migration-start path at mcp_server/fts5_index.py lines 328-357 requires
no direct change once replayed inserts are idempotent.
- Around line 310-326: Update start_migration_background to accurately describe
the non-daemon migration thread and retain the created thread on the owning
object instead of discarding it. Add shutdown handling that cancels migration
when supported and joins the stored fts5-migration thread with a bounded wait,
ensuring interpreter shutdown is not blocked indefinitely.
In `@mcp_server/server.py`:
- Around line 2408-2421: Update the empty-corpus branch in the FTS5 migration
flow to write both marker timestamps using timezone-aware UTC datetimes,
matching Fts5LexicalIndex._migration_worker and _run_migration_sync. Import
timezone alongside datetime and replace the naive datetime.now() calls passed to
_write_state, preserving the existing completion behavior.
- Around line 2505-2526: Update the FTS5 migration orchestration around
_maybe_start_fts5_migration to retain the worker thread when it is started, then
have _fts5_reset_and_rebuild stop starting over until any prior migration worker
has been joined. Ensure the old worker finishes before deleting or recreating
the shared state file and assigning the new Fts5LexicalIndex, while preserving
the existing reset and migration-start behavior.
- Around line 3202-3210: Update _fts5_sync_add_from_doc to derive FTS5 ids,
content, and metadata from the deduplicated chunks returned or used by
_index_document, rather than iterating over every doc.chunks entry. Reuse the
ids and associated metadata that were actually written to ChromaDB, preserving
the existing FTS5 guard and no-op behavior when no chunks survive deduplication.
- Around line 2439-2464: Update _iter_chroma_chunks_for_fts5 to avoid loading
the entire corpus at once: fetch and sort IDs first, then retrieve documents and
metadata using bounded batches of IDs and yield each batch in sorted-ID order.
Preserve the existing tuple fields, empty-collection behavior, and count-error
handling while removing any corpus-sized buffering.
In `@scripts/build_fts5_index.py`:
- Around line 145-153: Update the rebuild flow around _drop_existing,
_open_collection, and the empty-corpus branch so every invocation clears the
existing target index before population, regardless of args.force. When
_iter_chroma_chunks returns no rows, persist the index’s complete zero-row state
before returning, including when --force is used.
- Around line 73-94: Update _iter_chroma_chunks to avoid calling ids.index for
each chunk. Build aligned (id, document, metadata) tuples and sort those tuples
once by ID, then construct rows from the sorted tuples while preserving
document/metadata alignment and existing fallback handling.
- Around line 65-70: Update the cleanup loop in the build/rebuild flow to
terminate with a non-zero error when path.unlink() raises OSError. Preserve the
existing diagnostic output, then raise or otherwise propagate the failure so no
indexing or insertion proceeds after cleanup fails.
- Around line 148-157: Serialize manual FTS5 rebuilds with live CRUD: in
scripts/build_fts5_index.py lines 148-157, acquire the daemon’s exclusive
CRUD/rebuild lock before _open_collection snapshot extraction and hold it
through _run_migration_sync, or otherwise enforce an atomic index build-and-swap
while CRUD is paused; in docs/runbooks/fts5_migration.md lines 62-98, document
that the daemon must be stopped or CRUD exclusively locked before manual
rebuilds, particularly when using --force.
- Around line 105-111: Update _open_collection to retrieve the configured
collection with get_collection(name=config.collection_name) instead of
get_or_create_collection, ensuring a missing collection raises an explicit error
rather than creating an empty one.
In `@tests/test_fts5_migration.py`:
- Around line 443-486: Update _build_sync_orch to accept pytest’s request
fixture and register a finalizer with request.addfinalizer that stops the config
patcher and closes real_index. Remove the __teardown__ attribute and any
ineffective autouse scanning, then update every caller to pass request; remove
the pytest import only if no other fixture uses it.
---
Nitpick comments:
In `@mcp_server/server.py`:
- Around line 2394-2430: add a public Fts5LexicalIndex method such as
mark_empty_corpus_complete() that atomically writes the complete empty-corpus
state and marks the index ready. Update _maybe_start_fts5_migration and
scripts/build_fts5_index.py to call this method instead of accessing
_write_state, _fts5_lock, or _ready directly, removing the SLF001 suppressions.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 4113d2fd-b74d-4d0e-968e-0451696fc378
📒 Files selected for processing (10)
.github/api-surface-baseline.json.github/test-count-baseline.txtREADME.mddocs/runbooks/fts5_migration.mdmcp_server/fts5_index.pymcp_server/metrics.pymcp_server/server.pyscripts/build_fts5_index.pytests/test_e2e_fts5.pytests/test_fts5_migration.py
Comment on lines
+55
to
+60
| - `status: "complete"` → fast-path is live, queries dispatch to FTS5. | ||
| - `status: "in_progress"` → migration still running (or was interrupted). | ||
| The daemon resumes from `docs_indexed` on the next restart — it never | ||
| rebuilds from zero. | ||
| - `status: "failed"` → see `error` field for the exception class + message. | ||
| Queries fall back permanently until you rebuild manually. |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
ast-grep outline mcp_server/server.py --items all --match 'migration|fts5'
rg -n -C 8 'maybe_start_fts5_migration|start_migration_background|failed|in_progress|docs_indexed' mcp_server/server.pyRepository: lyonzin/knowledge-rag
Length of output: 30947
🏁 Script executed:
#!/bin/bash
set -euo pipefail
fd -i 'fts5_index|fts5_migration' .
printf '\n--- migration startup and worker implementation ---\n'
sed -n '2368,2432p' mcp_server/server.py
printf '\n--- FTS5 migration state handling ---\n'
rg -n -C 10 'class .*State|state|start_migration_background|status|docs_indexed|failed|complete|in_progress' mcp_server/fts5_index.py
printf '\n--- runbook recovery section ---\n'
sed -n '45,105p' docs/runbooks/fts5_migration.mdRepository: lyonzin/knowledge-rag
Length of output: 16525
Align failed-marker recovery documentation
A failed marker starts a fresh background migration at daemon startup. Update the runbook to state this behavior. Keep manual rebuild as an optional recovery path for corruption or operator-controlled rebuilds. The README release note matches the runtime behavior and needs no change.
📍 Affects 2 files
docs/runbooks/fts5_migration.md#L55-L60(this comment)README.md#L1417-L1417
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@docs/runbooks/fts5_migration.md` around lines 55 - 60, The failed-marker
recovery description in docs/runbooks/fts5_migration.md must state that daemon
startup begins a fresh background migration when status is "failed"; retain
manual rebuild only as an optional recovery path for corruption or
operator-controlled rebuilds. README.md:1417-1417 requires no direct change
because its release note already matches runtime behavior.
Comment on lines
+259
to
+300
|
|
||
| # ----------------------------------------------------------------- | ||
| # CRUD sync (Task 05, ADR-008). SQL nativo incremental — diverge do | ||
| # BM25 full-rebuild pattern porque FTS5 tem INSERT/DELETE O(1) e o | ||
| # full rebuild custaria segundos por mutation em corpus 3865 docs. | ||
| # Todos os writes acquire ``_fts5_lock`` (RLock, Q2 do TechSpec) e | ||
| # sao serializados pelo WAL SQLite (ADR-001). | ||
| # ----------------------------------------------------------------- | ||
|
|
||
| def add_document(self, chunk_id: str, content: str, filename: str, category: str) -> None: | ||
| """Insert one chunk row via ``INSERT`` (ADR-008).""" | ||
| if self._conn is None: | ||
| raise Fts5CorruptError("FTS5 connection is closed") | ||
| with self._fts5_lock: | ||
| self._conn.execute( | ||
| "INSERT INTO fts5_documents (chunk_id, content, filename, category) VALUES (?, ?, ?, ?)", | ||
| (chunk_id, content, filename, category), | ||
| ) | ||
| self._conn.commit() | ||
|
|
||
| def remove_document(self, chunk_id: str) -> None: | ||
| """Delete every row matching ``chunk_id`` (ADR-008).""" | ||
| if self._conn is None: | ||
| raise Fts5CorruptError("FTS5 connection is closed") | ||
| with self._fts5_lock: | ||
| self._conn.execute( | ||
| "DELETE FROM fts5_documents WHERE chunk_id = ?", | ||
| (chunk_id,), | ||
| ) | ||
| self._conn.commit() | ||
|
|
||
| def update_document(self, chunk_id: str, content: str, filename: str, category: str) -> None: | ||
| """DELETE + INSERT atomico — FTS5 nao tem UPDATE efficient em virtual table.""" | ||
| if self._conn is None: | ||
| raise Fts5CorruptError("FTS5 connection is closed") | ||
| with self._fts5_lock: | ||
| self._conn.execute("DELETE FROM fts5_documents WHERE chunk_id = ?", (chunk_id,)) | ||
| self._conn.execute( | ||
| "INSERT INTO fts5_documents (chunk_id, content, filename, category) VALUES (?, ?, ?, ?)", | ||
| (chunk_id, content, filename, category), | ||
| ) | ||
| self._conn.commit() |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Unlocked self._conn check in every FTS5 write path. All four write methods test self._conn is None before they acquire _fts5_lock, then dereference self._conn inside the lock. close() holds the same lock and sets self._conn = None, so a concurrent close between the check and the acquisition raises AttributeError on None instead of the documented Fts5CorruptError or Fts5MigrationError. _fts5_reset_and_rebuild in mcp_server/server.py calls close() while the migration worker can still be active, so the window is reachable.
mcp_server/fts5_index.py#L259-L300: move theself._conn is Nonecheck insidewith self._fts5_lockinadd_document,remove_document, andupdate_document, and bind the connection to a local variable for all statements.mcp_server/fts5_index.py#L404-L413: apply the same pattern in_populate_batch, keepingFts5MigrationErroras the raised type.
📍 Affects 1 file
mcp_server/fts5_index.py#L259-L300(this comment)mcp_server/fts5_index.py#L404-L413
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@mcp_server/fts5_index.py` around lines 259 - 300, Move the self._conn None
checks inside _fts5_lock in add_document, remove_document, and update_document,
then bind the validated connection to a local variable and use it for every SQL
statement and commit. Apply the same locking and local-connection pattern in
mcp_server/fts5_index.py lines 404-413 within _populate_batch, preserving
Fts5MigrationError there.
Comment on lines
+310
to
+326
| def start_migration_background( | ||
| self, | ||
| chunk_iter_factory: ChunkIterFactory, | ||
| docs_total: int, | ||
| *, | ||
| resume_from: int = 0, | ||
| on_progress: Optional[ProgressCallback] = None, | ||
| ) -> threading.Thread: | ||
| """Launch the migration daemon thread. Returns the started thread.""" | ||
| thread = threading.Thread( | ||
| target=self._migration_worker, | ||
| args=(chunk_iter_factory, docs_total, resume_from, on_progress), | ||
| name="fts5-migration", | ||
| daemon=False, | ||
| ) | ||
| thread.start() | ||
| return thread |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Find shutdown/atexit handling and any join of the fts5-migration thread.
rg -nP -C4 '(atexit|signal\.signal|SIGTERM|SIGINT|\.join\(|fts5-migration|_migration_thread)' --type=py -g '!tests/**'Repository: lyonzin/knowledge-rag
Length of output: 159
🏁 Script executed:
#!/bin/bash
set -eu
printf '%s\n' '--- migration implementation and call sites ---'
rg -n -C6 'start_migration_background|_migration_worker|fts5-migration|daemon\s*=|shutdown|close\(|stop\(' --glob '*.py' .
printf '%s\n' '--- target file ---'
sed -n '270,350p' mcp_server/fts5_index.py
printf '%s\n' '--- project entrypoints and lifecycle references ---'
rg -n -C4 'FastMCP|mcp\.run|asyncio|atexit|signal|SIGTERM|SIGINT|join\(' --glob '*.py' . --glob '!tests/**'Repository: lyonzin/knowledge-rag
Length of output: 50377
🏁 Script executed:
#!/bin/bash
set -eu
printf '%s\n' '--- migration startup context ---'
sed -n '2380,2450p' mcp_server/server.py
printf '%s\n' '--- server lifecycle and shutdown code ---'
rg -n -C8 'def (close|shutdown|run|main)|with .*instance|instance_lock|fts5_index\.close|observer\.stop|observer\.join|mcp\.run|uvicorn|serve' mcp_server/server.py mcp_server/instance_lock.py
printf '%s\n' '--- all production references to the returned migration thread ---'
rg -n 'start_migration_background|fts5-migration|migration_thread|_migration_thread|Thread\(' mcp_server scripts --glob '*.py'Repository: lyonzin/knowledge-rag
Length of output: 41160
Add migration shutdown handling and correct the docstring.
The server discards the non-daemon fts5-migration thread and has no shutdown join or cancellation path. A normal interpreter shutdown can therefore wait for the migration to finish. Store the thread and join it, or add cancellation with a bounded shutdown wait. Change the docstring from “daemon thread” to match daemon=False.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@mcp_server/fts5_index.py` around lines 310 - 326, Update
start_migration_background to accurately describe the non-daemon migration
thread and retain the created thread on the owning object instead of discarding
it. Add shutdown handling that cancels migration when supported and joins the
stored fts5-migration thread with a bounded wait, ensuring interpreter shutdown
is not blocked indefinitely.
Comment on lines
+359
to
+391
| def _run_migration_batches( | ||
| self, | ||
| chunk_iter_factory: ChunkIterFactory, | ||
| docs_total: int, | ||
| docs_indexed: int, | ||
| started_at: str, | ||
| on_progress: Optional[ProgressCallback], | ||
| ) -> int: | ||
| """Consume the iterator batch-by-batch. Returns the final ``docs_indexed``.""" | ||
| resume_from = docs_indexed | ||
| seen = 0 | ||
| batch: List[ChunkRow] = [] | ||
| last_percent_logged = -10 | ||
| for row in chunk_iter_factory(): | ||
| if seen < resume_from: | ||
| seen += 1 | ||
| continue | ||
| seen += 1 | ||
| batch.append(row) | ||
| if len(batch) >= 100: | ||
| self._populate_batch(batch) | ||
| docs_indexed += len(batch) | ||
| batch = [] | ||
| self._write_state("in_progress", docs_total, docs_indexed, started_at, None, None) | ||
| if on_progress is not None: | ||
| on_progress(docs_indexed, docs_total) | ||
| last_percent_logged = self._maybe_log_progress(docs_indexed, docs_total, last_percent_logged) | ||
| if batch: | ||
| self._populate_batch(batch) | ||
| docs_indexed += len(batch) | ||
| if on_progress is not None: | ||
| on_progress(docs_indexed, docs_total) | ||
| return docs_indexed |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Migration inserts are not idempotent, so any rerun duplicates rows. _populate_batch issues a plain INSERT, and the FTS5 schema declares chunk_id UNINDEXED with no uniqueness constraint. Every path that replays rows already written therefore creates duplicate entries for the same chunk_id, which inflates BM25 scores and returns duplicate hits from search.
mcp_server/fts5_index.py#L359-L391: make replay safe. Either delete bychunk_idbefore each batch insert, or resume from a persistedchunk_idwatermark instead of the positionalresume_fromcount, which shifts when the corpus changes between runs.mcp_server/fts5_index.py#L328-L357: no change is needed here once inserts are idempotent. Note that afailedmarker makes_maybe_start_fts5_migrationrestart atresume_from=0, so the rows already written by the failed run are re-inserted today.
📍 Affects 1 file
mcp_server/fts5_index.py#L359-L391(this comment)mcp_server/fts5_index.py#L328-L357
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@mcp_server/fts5_index.py` around lines 359 - 391, Make _run_migration_batches
replay-safe by removing existing FTS5 rows for each row’s chunk_id before
_populate_batch inserts the batch, or otherwise enforce equivalent idempotent
insertion. Preserve progress accounting and callbacks, and leave
_maybe_start_fts5_migration unchanged; the earlier migration-start path at
mcp_server/fts5_index.py lines 328-357 requires no direct change once replayed
inserts are idempotent.
Comment on lines
+73
to
+94
| def _iter_chroma_chunks(collection) -> list[tuple[str, str, str, str]]: | ||
| count = collection.count() | ||
| if count == 0: | ||
| return [] | ||
| fetched = collection.get(include=["documents", "metadatas"], limit=count) | ||
| ids = fetched.get("ids") or [] | ||
| docs = fetched.get("documents") or [] | ||
| metas = fetched.get("metadatas") or [] | ||
| rows: list[tuple[str, str, str, str]] = [] | ||
| for chunk_id, content, meta in zip(sorted(ids), docs, metas): | ||
| # Preserve alignment with sorted ids | ||
| idx = ids.index(chunk_id) | ||
| meta_i = metas[idx] or {} | ||
| rows.append( | ||
| ( | ||
| str(chunk_id), | ||
| str(docs[idx] or ""), | ||
| str(meta_i.get("filename", "")), | ||
| str(meta_i.get("category", "")), | ||
| ) | ||
| ) | ||
| return rows |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟠 Major | ⚡ Quick win
Remove the quadratic ID lookup.
ids.index(chunk_id) scans ids for every row. A 100,000-row rebuild becomes O(n²) before indexing starts. Sort aligned rows once instead.
Proposed fix
- for chunk_id, content, meta in zip(sorted(ids), docs, metas):
- # Preserve alignment with sorted ids
- idx = ids.index(chunk_id)
- meta_i = metas[idx] or {}
+ for chunk_id, content, meta in sorted(zip(ids, docs, metas), key=lambda row: str(row[0])):
+ meta_i = meta or {}
rows.append(
(
str(chunk_id),
- str(docs[idx] or ""),
+ str(content or ""),
str(meta_i.get("filename", "")),
str(meta_i.get("category", "")),
)
)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| def _iter_chroma_chunks(collection) -> list[tuple[str, str, str, str]]: | |
| count = collection.count() | |
| if count == 0: | |
| return [] | |
| fetched = collection.get(include=["documents", "metadatas"], limit=count) | |
| ids = fetched.get("ids") or [] | |
| docs = fetched.get("documents") or [] | |
| metas = fetched.get("metadatas") or [] | |
| rows: list[tuple[str, str, str, str]] = [] | |
| for chunk_id, content, meta in zip(sorted(ids), docs, metas): | |
| # Preserve alignment with sorted ids | |
| idx = ids.index(chunk_id) | |
| meta_i = metas[idx] or {} | |
| rows.append( | |
| ( | |
| str(chunk_id), | |
| str(docs[idx] or ""), | |
| str(meta_i.get("filename", "")), | |
| str(meta_i.get("category", "")), | |
| ) | |
| ) | |
| return rows | |
| def _iter_chroma_chunks(collection) -> list[tuple[str, str, str, str]]: | |
| count = collection.count() | |
| if count == 0: | |
| return [] | |
| fetched = collection.get(include=["documents", "metadatas"], limit=count) | |
| ids = fetched.get("ids") or [] | |
| docs = fetched.get("documents") or [] | |
| metas = fetched.get("metadatas") or [] | |
| rows: list[tuple[str, str, str, str]] = [] | |
| for chunk_id, content, meta in sorted(zip(ids, docs, metas), key=lambda row: str(row[0])): | |
| meta_i = meta or {} | |
| rows.append( | |
| ( | |
| str(chunk_id), | |
| str(content or ""), | |
| str(meta_i.get("filename", "")), | |
| str(meta_i.get("category", "")), | |
| ) | |
| ) | |
| return rows |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@scripts/build_fts5_index.py` around lines 73 - 94, Update _iter_chroma_chunks
to avoid calling ids.index for each chunk. Build aligned (id, document,
metadata) tuples and sort those tuples once by ID, then construct rows from the
sorted tuples while preserving document/metadata alignment and existing fallback
handling.
Comment on lines
+105
to
+111
| def _open_collection(data_dir: Path): | ||
| import chromadb | ||
|
|
||
| from mcp_server.config import config | ||
|
|
||
| client = chromadb.PersistentClient(path=str(config.chroma_dir)) | ||
| return client.get_or_create_collection(name=config.collection_name) |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🌐 Web query:
For ChromaDB version 1.4.0, does PersistentClient.get_or_create_collection(name=...) create a new empty collection when no collection exists? Cite the official ChromaDB documentation.
💡 Result:
Yes, in ChromaDB version 1.4.0, the PersistentClient.get_or_create_collection(name=...) method creates a new, empty collection if a collection with the specified name does not already exist [1][2]. According to the official ChromaDB documentation, the method functions as follows: - If the collection does not exist, it is created with the provided parameters (such as metadata) [1][3]. - If the collection already exists, the method simply returns the existing collection, and any additional parameters provided (like metadata) are ignored [1][4][3]. This behavior ensures that the requested collection is available for use regardless of its prior existence [2].
Citations:
- 1: https://docs.trychroma.com/reference/python/client
- 2: https://docs.trychroma.com/docs/collections/manage-collections
- 3: https://github.com/chroma-core/chroma/blob/main/chromadb/api/client.py
- 4: https://www.mintlify.com/chroma-core/chroma/api/python/client
🏁 Script executed:
#!/bin/bash
set -eu
printf '%s\n' '--- target file ---'
sed -n '1,180p' scripts/build_fts5_index.py
printf '%s\n' '--- ChromaDB version/config references ---'
rg -n --glob '!node_modules' --glob '!dist' --glob '!build' 'chromadb|collection_name|chroma_dir' .Repository: lyonzin/knowledge-rag
Length of output: 16585
Use get_collection() for rebuilds
If the configured collection does not exist, get_or_create_collection() creates an empty collection and the script reports a successful empty rebuild. Use get_collection(name=config.collection_name) so configuration errors fail explicitly.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@scripts/build_fts5_index.py` around lines 105 - 111, Update _open_collection
to retrieve the configured collection with
get_collection(name=config.collection_name) instead of get_or_create_collection,
ensuring a missing collection raises an explicit error rather than creating an
empty one.
Comment on lines
+145
to
+153
| if args.force: | ||
| _drop_existing(data_dir) | ||
|
|
||
| start = time.time() | ||
| collection = _open_collection(data_dir) | ||
| rows = _iter_chroma_chunks(collection) | ||
| if not rows: | ||
| print("[BUILD-FTS5] corpus is empty — nothing to index") | ||
| return 0 |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Make every rebuild replace the target index state.
Without --force, the script preserves existing FTS5 rows and _populate_batch() appends duplicate chunk IDs. When the corpus is empty, this early return preserves stale rows without --force and leaves no complete zero-row marker with --force. Require a clean target before population, and write a complete zero-row state for an empty corpus.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@scripts/build_fts5_index.py` around lines 145 - 153, Update the rebuild flow
around _drop_existing, _open_collection, and the empty-corpus branch so every
invocation clears the existing target index before population, regardless of
args.force. When _iter_chroma_chunks returns no rows, persist the index’s
complete zero-row state before returning, including when --force is used.
Comment on lines
+148
to
+157
| start = time.time() | ||
| collection = _open_collection(data_dir) | ||
| rows = _iter_chroma_chunks(collection) | ||
| if not rows: | ||
| print("[BUILD-FTS5] corpus is empty — nothing to index") | ||
| return 0 | ||
|
|
||
| index = _open_index(data_dir) | ||
| try: | ||
| _run_migration_sync(index, rows, args.verbose) |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Serialize manual rebuilds with FTS5 CRUD.
The script reads a ChromaDB snapshot before it writes FTS5 rows. A daemon-side update_document or remove_document after that snapshot can complete first, then the rebuild inserts the old row and restores stale content. Require exclusive rebuild access or build a separate index and swap it while CRUD is paused.
scripts/build_fts5_index.py#L148-L157: prevent live CRUD during snapshot extraction and FTS5 population.docs/runbooks/fts5_migration.md#L62-L98: require the daemon to be stopped or CRUD to be exclusively locked before a manual rebuild, especially with--force.
📍 Affects 2 files
scripts/build_fts5_index.py#L148-L157(this comment)docs/runbooks/fts5_migration.md#L62-L98
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@scripts/build_fts5_index.py` around lines 148 - 157, Serialize manual FTS5
rebuilds with live CRUD: in scripts/build_fts5_index.py lines 148-157, acquire
the daemon’s exclusive CRUD/rebuild lock before _open_collection snapshot
extraction and hold it through _run_migration_sync, or otherwise enforce an
atomic index build-and-swap while CRUD is paused; in
docs/runbooks/fts5_migration.md lines 62-98, document that the daemon must be
stopped or CRUD exclusively locked before manual rebuilds, particularly when
using --force.
Comment on lines
+443
to
+486
| def _build_sync_orch(tmp_path, *, fts5_enabled: bool = True): | ||
| """Stub orchestrator with the real CRUD-sync helpers bound to a capture index.""" | ||
| import mcp_server.server as srv | ||
| from mcp_server.server import KnowledgeOrchestrator | ||
|
|
||
| Fts5MigrationState(tmp_path / "fts5_migration.state").write( | ||
| { | ||
| "status": "complete", | ||
| "docs_total": 0, | ||
| "docs_indexed": 0, | ||
| "started_at": "2026-08-07T12:00:00Z", | ||
| "completed_at": "2026-08-07T12:00:00Z", | ||
| "error": None, | ||
| } | ||
| ) | ||
| real_index = Fts5LexicalIndex( | ||
| db_path=tmp_path / "fts5_index.db", | ||
| state_path=tmp_path / "fts5_migration.state", | ||
| ) | ||
| capture = _CaptureIndex(real_index) | ||
|
|
||
| orch = object.__new__(KnowledgeOrchestrator) | ||
| orch.fts5_index = capture # helpers accept the duck-typed wrapper | ||
| orch.collection = types.SimpleNamespace(get=lambda where, include: {"ids": []}) | ||
|
|
||
| # Bind the real hooks so behaviour under test is production-shaped. | ||
| orch._fts5_sync_add = types.MethodType( # noqa: SLF001 | ||
| KnowledgeOrchestrator._fts5_sync_add, orch | ||
| ) | ||
| orch._fts5_sync_remove_by_doc_id = types.MethodType( # noqa: SLF001 | ||
| KnowledgeOrchestrator._fts5_sync_remove_by_doc_id, orch | ||
| ) | ||
|
|
||
| # Patch config toggles for the duration of the test. | ||
| patcher = patch.object(srv.config, "fts5_enabled", fts5_enabled) | ||
| patcher.start() | ||
|
|
||
| def _teardown(): | ||
| patcher.stop() | ||
| real_index.close() | ||
|
|
||
| # Register teardown via a finalizer bound to the orchestrator stub. | ||
| orch.__teardown__ = _teardown # noqa: SLF001 | ||
| return orch, capture |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔴 Critical | ⚡ Quick win
The teardown fixture never runs, so config.fts5_enabled leaks across tests.
_build_sync_orch attaches _teardown to the orchestrator stub and returns that stub. Each test binds it to a local variable. The autouse fixture then scans request.node.__dict__ for an attribute that holds an object with __teardown__. The stub is never assigned to request.node, so the scan finds nothing.
Two effects follow:
patcher.stop()never runs.test_it_crud_004_feature_off_skips_fts5_callspatchessrv.config.fts5_enabledtoFalseand leaves it there for the rest of the session. Any later test that depends on the FTS5 path then silently takes the disabled branch, and the failure appears in an unrelated test file depending on collection order.real_index.close()never runs. The SQLite connection stays open. The module docstring states that Windows CI is sensitive to SQLite locking, andtmp_pathcleanup can fail for that reason.
Convert the helper into a fixture-driven builder that registers cleanup through request.addfinalizer.
💚 Proposed fix
-def _build_sync_orch(tmp_path, *, fts5_enabled: bool = True):
+def _build_sync_orch(tmp_path, request, *, fts5_enabled: bool = True):
"""Stub orchestrator with the real CRUD-sync helpers bound to a capture index."""
@@
patcher = patch.object(srv.config, "fts5_enabled", fts5_enabled)
patcher.start()
- def _teardown():
- patcher.stop()
- real_index.close()
-
- # Register teardown via a finalizer bound to the orchestrator stub.
- orch.__teardown__ = _teardown # noqa: SLF001
+ request.addfinalizer(patcher.stop)
+ request.addfinalizer(real_index.close)
return orch, capture
-
-
-@pytest.fixture(autouse=True)
-def _cleanup_sync_orch(request):
- """Run any ``__teardown__`` recorded by ``_build_sync_orch``."""
- yield
- for name, obj in list(request.node.__dict__.items()):
- if hasattr(obj, "__teardown__"):
- try:
- obj.__teardown__()
- except Exception: # noqa: BLE001
- passEach caller then passes the built-in request fixture, for example:
def test_it_crud_004_feature_off_skips_fts5_calls(self, tmp_path, request):
orch, capture = _build_sync_orch(tmp_path, request, fts5_enabled=False)Removing the pytest import becomes possible only if no other fixture remains in the file.
Also applies to: 489-498
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/test_fts5_migration.py` around lines 443 - 486, Update _build_sync_orch
to accept pytest’s request fixture and register a finalizer with
request.addfinalizer that stops the config patcher and closes real_index. Remove
the __teardown__ attribute and any ineffective autouse scanning, then update
every caller to pass request; remove the pytest import only if no other fixture
uses it.
CI Pillar 7 (mypy strict) flagged 4 functions in the standalone migration script without type annotations. Standardize with the rest of the codebase (Rules Python DevStyle requires type hints on all function parameters and returns). - Added typing.TYPE_CHECKING + Any imports - _iter_chroma_chunks: collection: Any - _open_index: -> Fts5LexicalIndex (forward ref via TYPE_CHECKING) - _open_collection: -> Any (chromadb Collection, avoid top-level import) - _run_migration_sync: index: Fts5LexicalIndex
Merged
4 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Task 05 wires the FTS5 fast-path to real production data - before this PR, the module (task 01) and dispatch (task 03) existed but the index was always empty.
Fase 4 of FTS5 Lexical Fast-Path plan (workflow artifacts local-only, gitignored).
What ships
KnowledgeOrchestrator._ensure_fts5_indexspawns a background thread on first access iffts5_index.dbis empty. First query does not block - falls back to hybrid gracefully while rebuild runs._index_document,remove_document_by_path,update_document_contentnow also update FTS5 (guarded byconfig.fts5_enabled, idempotent when FTS5 empty).scripts/build_fts5_index.pyfor manual migration when ops needs to force a rebuild without daemon restart.docs/runbooks/fts5_migration.md.tests/test_fts5_migration.py+ 24 lines intest_e2e_fts5.py.What does NOT ship
search_knowledgesignature intact.LEI 1 preserved
Zero change to the 13 frozen MCP tools. Internal helpers
_index_document,remove_document_by_path,update_document_contentgain FTS5 sync but their signatures are internal - the MCP toolsadd_document,remove_document,update_documentthat call them are unchanged.Local validation notes
pytest tests/test_ingestion.py= 38 PASS (tests that dont import server.py)mcp_server.server(test_dedup,test_fts5_migration,test_search,test_swap_zero_downtime) blocked locally by pre-existingMCPServerimport failure in Python 3.14 - same story as PRs feat: add FTS5 lexical index module (opt-in, unused) #158/feat: add query lexical/semantic classifier (opt-in, unused) #159/feat: wire FTS5 fast-path (default OFF) #160/feat: add optional cross-encoder rerank in FTS5 fast-path #166. CI env with pinned deps validates.ruff check+ruff format= clean (2 files reformatted then verified)python scripts/check_api_surface.py --check= OK (no breaking changes)Rollback
git revert <sha>- feature is opt-in default-off, CRUD hooks are guarded byconfig.fts5_enabled, revert is safe.Summary by CodeRabbit
Greptile Summary
The PR connects the existing FTS5 fast path to production data through lazy migration, CRUD synchronization, rebuild integration, operational tooling, and migration metrics.
Confidence Score: 2/5
The PR is not yet safe to merge because committed migration batches can be duplicated after interruption, concurrent CRUD can be overwritten by stale migration data, and an obsolete worker can corrupt replacement migration state.
Three previously reported lifecycle defects remain in the current code: checkpoint persistence is not atomic with batch commits, migration snapshots are not coordinated with CRUD synchronization, and rebuild reset does not stop the active migration worker before replacing its index and marker.
Files Needing Attention: mcp_server/fts5_index.py and mcp_server/server.py
Important Files Changed
Sequence Diagram
sequenceDiagram participant O as KnowledgeOrchestrator participant C as ChromaDB participant M as FTS5 migration worker participant F as FTS5 index O->>C: Count and read corpus O->>M: Start background migration loop Batches M->>F: Insert chunk rows M->>F: Persist progress marker end M->>F: Mark migration complete O->>F: Synchronize later CRUD mutations O->>F: Reset and rebuild after nuclear swapReviews (2): Last reviewed commit: "fix(scripts): add missing type hints to ..." | Re-trigger Greptile