diff --git a/server/indexer/cleanup.py b/server/indexer/cleanup.py new file mode 100644 index 0000000..de89c8d --- /dev/null +++ b/server/indexer/cleanup.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +import logging +from typing import Protocol + +logger = logging.getLogger(__name__) + + +class PrunableStore(Protocol): + """A store that can list indexed service names and delete all data for one.""" + + async def get_indexed_services(self) -> list[str]: ... + + async def delete_by_service(self, service: str) -> None: ... + + +async def prune_orphaned_services( + store: PrunableStore, + configured_names: set[str], + label: str = "data", +) -> list[str]: + """Delete all stored data for services that exist in the store but not in + the configured set. Returns the list of orphaned service names that were pruned. + + Refuses to prune when `configured_names` is empty — an empty configured set + is far more likely to be a load error or operator mistake than a genuine + intent to wipe every indexed service, and deletion is not reversible. + """ + if not configured_names: + logger.warning( + "No configured services; skipping prune of %s to avoid wiping the store.", + label, + ) + return [] + + indexed_names = await store.get_indexed_services() + orphaned = sorted(set(indexed_names) - configured_names) + + for name in orphaned: + logger.warning("Pruning orphaned service %r from %s", name, label) + await store.delete_by_service(name) + + if orphaned: + logger.info( + "Pruned %d orphaned service(s) from %s: %s", len(orphaned), label, orphaned + ) + + return orphaned diff --git a/server/indexer/git_history.py b/server/indexer/git_history.py index 0355268..33eebc9 100644 --- a/server/indexer/git_history.py +++ b/server/indexer/git_history.py @@ -15,6 +15,7 @@ fetch_commits_with_diffs, list_commits, ) +from server.indexer.cleanup import prune_orphaned_services from server.indexer.pipeline import ProgressEvent from server.store.commit_store import CommitStore @@ -197,6 +198,11 @@ async def index_all( progress_callback: Callable[[ProgressEvent], Awaitable[None]] | None = None, ) -> dict[str, Any]: services = settings.load_services() + await self._store.ensure_collection() + await prune_orphaned_services( + self._store, {s.name for s in services}, label="git commits" + ) + results: dict[str, Any] = {} for svc in services: logger.info("Indexing git history for: %s", svc.name) diff --git a/server/indexer/pipeline.py b/server/indexer/pipeline.py index 0dd923b..9bd8aa5 100644 --- a/server/indexer/pipeline.py +++ b/server/indexer/pipeline.py @@ -14,6 +14,7 @@ from server.embeddings.base import EmbeddingProvider from server.embeddings.bm25 import BM25SparseProvider, get_sparse_embedding_provider from server.embeddings import get_embedding_provider +from server.indexer.cleanup import prune_orphaned_services from server.indexer.github_source import fetch_blob_content, list_github_files from server.parser.base import CodeSymbol, ParseError from server.parser.registry import parse_file @@ -281,6 +282,11 @@ async def index_all( progress_callback: Callable[[ProgressEvent], Awaitable[None]] | None = None, ) -> dict[str, Any]: services = settings.load_services() + await self._store.ensure_collection() + await prune_orphaned_services( + self._store, {s.name for s in services}, label="code symbols" + ) + results: dict[str, Any] = {} for svc in services: logger.info("Indexing service: %s", svc.name) diff --git a/server/store/commit_store.py b/server/store/commit_store.py index fde71b4..8bc2de7 100644 --- a/server/store/commit_store.py +++ b/server/store/commit_store.py @@ -227,5 +227,33 @@ async def _update_one(p: dict[str, Any]) -> None: await asyncio.gather(*(_update_one(p) for p in payloads)) + async def get_indexed_services(self) -> list[str]: + """Return distinct service names that have indexed commits.""" + services: set[str] = set() + offset = None + while True: + results, offset = await self._client.scroll( + collection_name=self._collection, + limit=1000, + offset=offset, + with_payload=["service"], + with_vectors=False, + ) + for point in results: + svc = point.payload.get("service") + if svc: + services.add(svc) + if offset is None: + break + return sorted(services) + + async def delete_by_service(self, service: str) -> None: + await self._client.delete( + collection_name=self._collection, + points_selector=Filter( + must=[FieldCondition(key="service", match=MatchValue(value=service))] + ), + ) + async def close(self) -> None: await self._client.close() diff --git a/server/store/qdrant.py b/server/store/qdrant.py index 8ab2169..1cf1e1b 100644 --- a/server/store/qdrant.py +++ b/server/store/qdrant.py @@ -309,6 +309,13 @@ async def get_service_stats(self) -> list[dict[str, Any]]: ) return result + async def get_indexed_services(self) -> list[str]: + """Return distinct service names that have indexed code symbols.""" + stats = await self.get_service_stats() + return sorted( + s["service"] for s in stats if s["service"] and s["service"] != "unknown" + ) + async def collection_info(self) -> dict[str, Any]: info = await self._client.get_collection(self._collection) return { diff --git a/tests/test_cleanup.py b/tests/test_cleanup.py new file mode 100644 index 0000000..24c3d0c --- /dev/null +++ b/tests/test_cleanup.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +from server.indexer.cleanup import prune_orphaned_services + + +class _FakeStore: + def __init__(self, indexed_names: list[str]) -> None: + self.indexed_names = indexed_names + self.deleted: list[str] = [] + + async def get_indexed_services(self) -> list[str]: + return self.indexed_names + + async def delete_by_service(self, service: str) -> None: + self.deleted.append(service) + + +async def test_service_absent_from_config_is_deleted() -> None: + store = _FakeStore(["kept", "renamed-away"]) + + orphaned = await prune_orphaned_services(store, {"kept"}) + + assert orphaned == ["renamed-away"] + assert store.deleted == ["renamed-away"] + + +async def test_configured_service_is_never_deleted() -> None: + store = _FakeStore(["kept"]) + + orphaned = await prune_orphaned_services(store, {"kept"}) + + assert orphaned == [] + assert store.deleted == [] + + +async def test_empty_configured_names_skips_prune_entirely() -> None: + store = _FakeStore(["a", "b", "c"]) + + orphaned = await prune_orphaned_services(store, set()) + + assert orphaned == [] + assert store.deleted == [] + + +async def test_no_indexed_services_is_a_noop() -> None: + store = _FakeStore([]) + + orphaned = await prune_orphaned_services(store, {"kept"}) + + assert orphaned == [] + assert store.deleted == [] + + +async def test_multiple_orphans_are_returned_sorted() -> None: + store = _FakeStore(["zeta", "alpha", "kept", "beta"]) + + orphaned = await prune_orphaned_services(store, {"kept"}) + + assert orphaned == ["alpha", "beta", "zeta"] + assert store.deleted == ["alpha", "beta", "zeta"] diff --git a/tests/test_git_history.py b/tests/test_git_history.py index 52b7051..6a64835 100644 --- a/tests/test_git_history.py +++ b/tests/test_git_history.py @@ -1,7 +1,12 @@ from __future__ import annotations +from unittest.mock import AsyncMock, patch + +import server.indexer.git_history as git_history_module +from server.config import ServiceConfig from server.indexer.git_history import ( _MAX_PATCH_CHARS, + GitHistoryPipeline, _build_embedding_text, _commit_to_payload, ) @@ -27,6 +32,42 @@ def _file(filename: str = "src/Foo.java", patch: str | None = None) -> CommitFil ) +async def test_index_all_prunes_orphaned_services_before_indexing() -> None: + store = AsyncMock() + store.ensure_collection = AsyncMock() + store.get_indexed_services = AsyncMock(return_value=["kept", "renamed-away"]) + store.delete_by_service = AsyncMock() + pipeline = GitHistoryPipeline(store) + pipeline.index_service = AsyncMock(return_value={"commits": 0}) + + with patch.object( + type(git_history_module.settings), + "load_services", + return_value=[ServiceConfig(name="kept", github_repo="org/kept", exclude=[])], + ): + await pipeline.index_all() + + store.delete_by_service.assert_awaited_once_with("renamed-away") + pipeline.index_service.assert_awaited_once_with( + "kept", force=False, progress_callback=None + ) + + +async def test_index_all_skips_prune_when_no_services_configured() -> None: + store = AsyncMock() + store.ensure_collection = AsyncMock() + store.get_indexed_services = AsyncMock(return_value=["kept"]) + store.delete_by_service = AsyncMock() + pipeline = GitHistoryPipeline(store) + + with patch.object( + type(git_history_module.settings), "load_services", return_value=[] + ): + await pipeline.index_all() + + store.delete_by_service.assert_not_awaited() + + def test_embedding_text_contains_service_and_author() -> None: text = _build_embedding_text(_commit(), "auth-server") assert "auth-server" in text diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index e61d82a..e20c6cf 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -57,6 +57,42 @@ def _sym(docstring: str) -> CodeSymbol: ) +async def test_index_all_prunes_orphaned_services_before_indexing() -> None: + store = AsyncMock() + store.ensure_collection = AsyncMock() + store.get_indexed_services = AsyncMock(return_value=["kept", "renamed-away"]) + store.delete_by_service = AsyncMock() + pipeline = _make_pipeline(store) + pipeline.index_service = AsyncMock( + return_value={"files": 1, "chunks": 1, "skipped": 0} + ) + + with patch.object( + type(pipeline_module.settings), + "load_services", + return_value=[ServiceConfig(name="kept", github_repo="org/kept", exclude=[])], + ): + await pipeline.index_all() + + store.delete_by_service.assert_awaited_once_with("renamed-away") + pipeline.index_service.assert_awaited_once_with( + "kept", force=False, progress_callback=None + ) + + +async def test_index_all_skips_prune_when_no_services_configured() -> None: + store = AsyncMock() + store.ensure_collection = AsyncMock() + store.get_indexed_services = AsyncMock(return_value=["kept"]) + store.delete_by_service = AsyncMock() + pipeline = _make_pipeline(store) + + with patch.object(type(pipeline_module.settings), "load_services", return_value=[]): + await pipeline.index_all() + + store.delete_by_service.assert_not_awaited() + + def test_python_triple_double_quote_stripped() -> None: text = _build_embedding_text(_sym('"""Hello world"""'), "svc") assert "Hello world" in text diff --git a/tests/test_store.py b/tests/test_store.py index a488e0f..82feed1 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -18,6 +18,22 @@ def test_close_is_coroutine() -> None: assert asyncio.iscoroutinefunction(QdrantStore.close) +async def test_get_indexed_services_excludes_unknown_placeholder() -> None: + """The 'unknown' placeholder from unlabeled payloads must never look like a + real service name that pruning could delete.""" + store = QdrantStore.__new__(QdrantStore) + store.get_service_stats = AsyncMock( + return_value=[ + {"service": "billing", "symbols": 3}, + {"service": "unknown", "symbols": 1}, + ] + ) + + services = await store.get_indexed_services() + + assert services == ["billing"] + + async def test_find_by_name_fuzzy_scans_all_pages() -> None: """Non-exact search must paginate instead of relying on the first 20 results.""" store = QdrantStore.__new__(QdrantStore)