fix(mcp): stop a replaced code engine's autosync loop

src/lemoncrow/gateway/adapters/mcp_server.py

@@ -8879,6 +8879,7 @@ def _memory_summary(session_id: str) -> dict[str, Any]:
 # from the one before it.
 _code_index_freshness_for_current_call: threading.local = threading.local()
 
+
 # Process-level engine cache keyed by resolved repo path.
 # Reusing the same engine across tool calls avoids re-opening the SQLite DB
 # and restarting autosync threads on every invocation — critical for both
@@ -8889,7 +8890,17 @@ def _memory_summary(session_id: str) -> dict[str, Any]:
 # still serving it at version 23, returning empty results for every query with
 # no error of any kind. VersionedEngineCache stamps each entry and rebuilds on a
 # bump, and raises IndexRebuilding mid-reindex rather than answering with [].
-_code_engine_cache = VersionedEngineCache("code_engine")
+#
+# A rebuild must also stop the engine it replaces. Each engine runs its own
+# autosync thread, which keeps it alive, so a superseded engine kept polling
+# and spawning reindexes: one more loop per index bump, until the daemon ran a
+# reindex storm that held the index-write lock nearly all the time and
+# code_search answered "index is being rebuilt" for most calls.
+def _retire_code_engine(engine: Any) -> None:
+    engine.stop_autosync()
+
+
+_code_engine_cache = VersionedEngineCache("code_engine", on_evict=_retire_code_engine)
 
 # ``cache_key -> (capability, engine_it_was_built_from)``.
 #

src/lemoncrow/infra/code_intel/freshness.py

@@ -267,16 +267,25 @@ class VersionedEngineCache:
 
     Rebuilds happen under a lock with a double check, so N concurrent callers
     arriving at a version bump together produce one rebuild, not N.
+
+    Dropping an entry does not end whatever the value started. *on_evict* is
+    called with every value the cache lets go of -- superseded, discarded or
+    cleared -- so its owner can stop it. The code engine is why this exists: its
+    background autosync thread references the engine, so an evicted engine never
+    died. Every index bump left one more loop polling the tree and spawning its
+    own reindex, and those reindexes bumped the version again.
     """
 
     def __init__(
         self,
         name: str,
         recheck_seconds: float = DEFAULT_RECHECK_SECONDS,
         clock: Callable[[], float] = time.monotonic,
+        on_evict: Callable[[Any], None] | None = None,
     ) -> None:
         self.name = name
         self.recheck_seconds = float(recheck_seconds)
+        self.on_evict = on_evict
         self.evictions = 0
         self._clock = clock
         self._lock = threading.Lock()
@@ -332,7 +341,19 @@ def get(self, key: str, repo_root: Path | str, build: Callable[[], Any]) -> tupl
                 self.evictions += 1
             value = build()
             self._entries[key] = _Entry(value=value, index_version=state.index_version)
-            return value, FRESHNESS_FRESH if superseded is None else FRESHNESS_REBUILT
+        # Outside the lock: retiring may join threads, and every other caller of
+        # this cache would wait on it.
+        if entry is not None:
+            self._retire(entry.value)
+        return value, FRESHNESS_FRESH if superseded is None else FRESHNESS_REBUILT
+
+    def _retire(self, value: Any) -> None:
+        if self.on_evict is None:
+            return
+        try:
+            self.on_evict(value)
+        except Exception:
+            logger.warning("%s: retiring an evicted entry failed", self.name, exc_info=True)
 
     def peek(self, key: str) -> Any | None:
         """The cached value for *key* without probing or building."""
@@ -346,13 +367,18 @@ def version_of(self, key: str) -> int | None:
 
     def discard(self, key: str) -> None:
         with self._lock:
-            self._entries.pop(key, None)
+            entry = self._entries.pop(key, None)
+        if entry is not None:
+            self._retire(entry.value)
 
     def clear(self) -> None:
         with self._lock:
+            dropped = [entry.value for entry in self._entries.values()]
             self._entries.clear()
             self._probes.clear()
             self.evictions = 0
+        for value in dropped:
+            self._retire(value)
 
     def __len__(self) -> int:
         return len(self._entries)

src/lemoncrow/pro/capabilities/code_context/engine.py

@@ -14793,6 +14793,17 @@ def _stop_autosync_worker(self) -> None:
         self._autosync_stop.set()
         self._stop_file_watcher()
 
+    def stop_autosync(self) -> None:
+        """Stop this engine's background index refresh, permanently.
+
+        Queries keep working; this instance just stops polling the tree and
+        spawning reindexes. Call it when replacing the engine: the worker thread
+        references the engine, so dropping every other reference does not stop
+        it, and a replaced engine otherwise refreshes the index for the life of
+        the process.
+        """
+        self._stop_autosync_worker()
+
     # --- File watcher (event-driven via watchdog) ---
 
     def _start_file_watcher(self) -> None:

tests/gateway/test_code_engine_cache_invalidation.py

@@ -49,10 +49,14 @@ def __init__(self, root: Path) -> None:
         type(self).instances += 1
         self.root = root
         self.db_path = str(root)
+        self.stopped = False
 
     def index_ready(self) -> bool:
         return True
 
+    def stop_autosync(self) -> None:
+        self.stopped = True
+
 
 @pytest.fixture
 def indexed_repo(tmp_path: Path) -> Path:
@@ -110,7 +114,11 @@ def _isolated_caches(monkeypatch: pytest.MonkeyPatch) -> None:
 
     _FakeEngine.instances = 0
     monkeypatch.setattr(code_context, "CodeContextEngine", _FakeEngine)
-    monkeypatch.setattr(mcp_server, "_code_engine_cache", VersionedEngineCache("test", recheck_seconds=0.0))
+    monkeypatch.setattr(
+        mcp_server,
+        "_code_engine_cache",
+        VersionedEngineCache("test", recheck_seconds=0.0, on_evict=mcp_server._code_engine_cache.on_evict),
+    )
     monkeypatch.setattr(mcp_server, "_scoped_context_cache", {})
     mcp_server._code_index_freshness_for_current_call.value = None
     mcp_server._code_engine_for_current_call.value = None
@@ -140,6 +148,21 @@ def test_index_version_bump_rebuilds_the_engine(indexed_repo: Path) -> None:
     assert mcp_server._code_index_freshness_for_current_call.value == FRESHNESS_REBUILT
 
 
+def test_a_rebuilt_engine_stops_the_one_it_replaced(indexed_repo: Path) -> None:
+    """A superseded engine left the cache but kept running.
+
+    Its autosync thread held it alive, so each index bump added one more loop
+    polling the tree and spawning its own reindex -- a storm that kept the
+    index-write lock held and made code_search answer "being rebuilt".
+    """
+    first = mcp_server._code_context_engine(str(indexed_repo))
+    _bump(indexed_repo, 2)
+    second = mcp_server._code_context_engine(str(indexed_repo))
+
+    assert first.stopped, "the replaced engine was left running"
+    assert not second.stopped
+
+
 class _FakeScoped:
     """Stands in for the compiled ScopedContextCapability."""
 
@@ -281,11 +304,12 @@ def test_freshness_does_not_leak_into_the_next_call(indexed_repo: Path) -> None:
 
 
 def test_runtime_cache_reset_clears_the_engine_cache(indexed_repo: Path) -> None:
-    mcp_server._code_context_engine(str(indexed_repo))
+    first = mcp_server._code_context_engine(str(indexed_repo))
     assert len(mcp_server._code_engine_cache) == 1
 
     mcp_server._code_engine_cache.clear()
 
     assert len(mcp_server._code_engine_cache) == 0
+    assert first.stopped, "a cleared engine was left running"
     mcp_server._code_context_engine(str(indexed_repo))
     assert _FakeEngine.instances == 2

tests/infra/code_intel/test_code_engine_autosync_retirement.py

@@ -0,0 +1,70 @@
+"""A daemon runs one autosync loop per repo, however often the index moves.
+
+Measured on a live daemon: the symphony-alpha index-write lock was held for the
+whole of a sampled window, and code_search answered "index is being rebuilt" for
+most calls. Every index bump had left the replaced engine's autosync loop
+running, and each leaked loop spawned its own reindex per change (reproduced: six
+engines, one edited file, six `lc code index` processes). The fake-engine tests
+pin the wiring; this one drives the real engine through the production cache, so
+the live thread count is the evidence.
+"""
+
+from __future__ import annotations
+
+import sqlite3
+import subprocess
+import threading
+import time
+from pathlib import Path
+
+import pytest
+
+from lemoncrow.gateway.adapters import mcp_server
+from lemoncrow.infra.code_intel.freshness import VersionedEngineCache
+from lemoncrow.infra.code_intel.store import CODE_CONTEXT_DB, workspace_dir
+from lemoncrow.pro.capabilities.code_context import CodeContextEngine
+
+
+def _live_autosync_threads(name: str) -> list[threading.Thread]:
+    return [t for t in threading.enumerate() if t.name == name and t.is_alive()]
+
+
+def _bump(root: Path) -> None:
+    conn = sqlite3.connect(workspace_dir(root) / CODE_CONTEXT_DB)
+    try:
+        conn.execute("UPDATE engine_state SET value = CAST(value AS INTEGER) + 1 WHERE key = 'index_version'")
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def test_index_bumps_leave_exactly_one_autosync_loop(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    repo = tmp_path / "repo"
+    (repo / "pkg").mkdir(parents=True)
+    for i in range(3):
+        (repo / "pkg" / f"mod{i}.py").write_text(f"def f{i}():\n    return {i}\n")
+    subprocess.run(["git", "init", "-q"], cwd=repo, check=True)
+    CodeContextEngine(repo, autosync_enabled=False).index_repo(force=True)
+
+    # The suite disables autosync globally; this test is about autosync.
+    monkeypatch.setenv("LEMONCROW_CODE_AUTOSYNC", "1")
+    monkeypatch.setenv("LEMONCROW_CODE_FILE_WATCHER", "0")
+    cache = VersionedEngineCache("test", recheck_seconds=0.0, on_evict=mcp_server._code_engine_cache.on_evict)
+
+    engines: list[CodeContextEngine] = []
+    try:
+        for _ in range(4):
+            engine, _freshness = cache.get(str(repo), repo, lambda: CodeContextEngine(repo))
+            engines.append(engine)
+            _bump(repo)
+        assert len({id(e) for e in engines}) == 4, "the index bumps did not rebuild the engine"
+
+        name = f"lemoncrow-code-autosync-{engines[0].repo_id[:8]}"
+        deadline = time.monotonic() + 10
+        while len(_live_autosync_threads(name)) > 1 and time.monotonic() < deadline:
+            time.sleep(0.05)
+
+        assert len(_live_autosync_threads(name)) == 1, "replaced engines kept their autosync loops"
+    finally:
+        for engine in engines:
+            engine.stop_autosync()

tests/infra/code_intel/test_freshness.py

@@ -198,6 +198,53 @@ def test_version_bump_evicts_and_rebuilds(make_workspace: WorkspaceFactory) -> N
     assert cache.evictions == 1
 
 
+def test_a_superseded_entry_is_retired_once(make_workspace: WorkspaceFactory) -> None:
+    """Eviction has to end what the evicted value started, not just forget it."""
+    root = make_workspace(files=_FILES, symbols=_SYMBOLS, index_version=1)
+    retired: list[object] = []
+    cache = VersionedEngineCache("test", recheck_seconds=0.0, on_evict=retired.append)
+
+    first, _ = cache.get("k", root, object)
+    cache.get("k", root, object)
+    assert retired == [], "a stable index retired its own live entry"
+
+    _set_index_version(root, 2)
+    second, _ = cache.get("k", root, object)
+
+    assert retired == [first]
+    assert second is not first
+
+
+def test_discard_and_clear_retire_what_they_drop(make_workspace: WorkspaceFactory) -> None:
+    root = make_workspace(files=_FILES, symbols=_SYMBOLS, index_version=1)
+    retired: list[object] = []
+    cache = VersionedEngineCache("test", recheck_seconds=0.0, on_evict=retired.append)
+
+    discarded, _ = cache.get("a", root, object)
+    cache.discard("a")
+    cleared, _ = cache.get("b", root, object)
+    cache.clear()
+
+    assert retired == [discarded, cleared]
+
+
+def test_a_failing_retirement_does_not_fail_the_lookup(make_workspace: WorkspaceFactory) -> None:
+    """The rebuilt value is already cached; a stop hook that raises must not lose it."""
+    root = make_workspace(files=_FILES, symbols=_SYMBOLS, index_version=1)
+
+    def refuse(_value: object) -> None:
+        raise RuntimeError("stop failed")
+
+    cache = VersionedEngineCache("test", recheck_seconds=0.0, on_evict=refuse)
+    first, _ = cache.get("k", root, object)
+    _set_index_version(root, 2)
+    second, freshness = cache.get("k", root, object)
+
+    assert second is not first
+    assert freshness == FRESHNESS_REBUILT
+    assert cache.peek("k") is second
+
+
 def test_first_build_is_fresh_not_rebuilt(make_workspace: WorkspaceFactory) -> None:
     """Nothing was superseded on a cold cache; only an eviction is 'rebuilt'."""
     root = make_workspace(files=_FILES, symbols=_SYMBOLS)