Merge pull request #89 from GoodbyePlanet/fix/prune-orphaned-services

fix: prune orphaned service data on rename/removal (#75)

Author: GoodbyePlanet

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

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)

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)

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()

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 {

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"]

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

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

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)