From fd2020f57411211d8bff002933081de611bcc7ab Mon Sep 17 00:00:00 2001 From: Kris Wong Date: Fri, 18 Sep 2026 16:30:56 -0500 Subject: [PATCH] fix(code-intel): serve code tools from the last committed index during a reindex A held index-write lock made every code tool refuse to answer for the length of every reindex. A reindex commits in one transaction and the databases run in WAL mode, so readers see the last committed index whole until it commits. Answer from it and append a note that results may lag the newest edits; a first build with nothing committed still refuses. Co-Authored-By: lemoncrow <302591943+lemoncrow-agent[bot]@users.noreply.github.com> LemonCrow-Session: s1 --- src/lemoncrow/gateway/adapters/mcp_server.py | 20 +++ src/lemoncrow/infra/code_intel/freshness.py | 68 ++++++++-- .../gateway/test_code_tools_during_reindex.py | 87 +++++++++++++ tests/infra/code_intel/test_clones.py | 2 +- tests/infra/code_intel/test_freshness.py | 61 +++++++-- .../code_intel/test_reads_during_reindex.py | 120 ++++++++++++++++++ 6 files changed, 337 insertions(+), 21 deletions(-) create mode 100644 tests/gateway/test_code_tools_during_reindex.py create mode 100644 tests/infra/code_intel/test_reads_during_reindex.py diff --git a/src/lemoncrow/gateway/adapters/mcp_server.py b/src/lemoncrow/gateway/adapters/mcp_server.py index 6dec547b7..777ab6a70 100644 --- a/src/lemoncrow/gateway/adapters/mcp_server.py +++ b/src/lemoncrow/gateway/adapters/mcp_server.py @@ -174,9 +174,11 @@ ) from lemoncrow.infra.code_intel.freshness import ( # noqa: F401 (IndexRebuilding re-exported for handlers/tests) FRESHNESS_REBUILT, + FRESHNESS_REFRESHING, IndexRebuilding, VersionedEngineCache, reset_readiness_probes, + take_refreshing, ) from lemoncrow.infra.runtime.run_ledger import ( RunLedger, @@ -12459,6 +12461,14 @@ def _model_recommendation_state(led: RunLedger, args: dict[str, Any]) -> dict[st return session_state +# Appended to any response that read the code index while a reindex held the +# write lock: the answer is the last committed index, whole but possibly behind. +_INDEX_REFRESHING_NOTE = ( + "note: the code index is refreshing; results come from the last completed index " + "and may not reflect the newest edits" +) + + def _handle(request: dict[str, Any]) -> dict[str, Any] | _Deferred | None: rid = request.get("id") method = request.get("method") @@ -12649,6 +12659,10 @@ def _finalize_error_response(exc: Exception) -> dict[str, Any]: }, ) + # Set once the handler returns: whether this call read the code index while + # a reindex held its write lock. Read by _finalize_response. + index_refreshing = False + def _finalize_response(result: dict[str, Any] | Any) -> dict[str, Any]: # Post-handler finalization pipeline. Runs synchronously on the worker # for the non-deferred path, and on bash_exec's watcher thread for a @@ -12731,6 +12745,8 @@ def _finalize_response(result: dict[str, Any] | Any) -> dict[str, Any]: with contextlib.suppress(Exception): _write_statusline_sidecar() + if index_refreshing and isinstance(result, dict): + result.setdefault("index_state", FRESHNESS_REFRESHING) response_text: str if rendered_text: response_text = rendered_text @@ -12744,6 +12760,8 @@ def _finalize_response(result: dict[str, Any] | Any) -> dict[str, Any]: # string, or JSON). Soft signal -- never replaces the result. if _loop_note and _loop_note not in response_text: response_text = f"{response_text}\n{_loop_note}" + if index_refreshing and _INDEX_REFRESHING_NOTE not in response_text: + response_text = f"{response_text}\n{_INDEX_REFRESHING_NOTE}" # Only pay the full-payload UTF-8 encode when a telemetry sink will # consume the byte count; otherwise approximate with the O(1) char len. @@ -12992,6 +13010,7 @@ def _finalize_response(result: dict[str, Any] | Any) -> dict[str, Any]: _tool_call_tokens_saved.value = 0 # reset before handler so stale values can't bleed through _tool_call_counterfactual.value = None # reset before handler _tool_call_rendered_text.value = None # reset before handler + take_refreshing() # drop a mark left by work that never reached a response wrapper_model = ( str(route_payload.get("model") or "") if _route_enforcement_enabled() and route_payload.get("configured") is not False @@ -13003,6 +13022,7 @@ def _finalize_response(result: dict[str, Any] | Any) -> dict[str, Any]: _handler_start = time.perf_counter() with active_model_override(wrapper_model or None): result = handler(args) + index_refreshing = take_refreshing() _call_duration_ms = round((time.perf_counter() - _handler_start) * 1000) finally: # Runs in finally; a raise here would mask the handler's real diff --git a/src/lemoncrow/infra/code_intel/freshness.py b/src/lemoncrow/infra/code_intel/freshness.py index 9019fd620..37643e67d 100644 --- a/src/lemoncrow/infra/code_intel/freshness.py +++ b/src/lemoncrow/infra/code_intel/freshness.py @@ -27,6 +27,15 @@ matches" as "this symbol has no callers" and files a finding on it. Callers get an exception they can catch instead. +A reindex in progress is not, by itself, that case. Every reindex writes in one +transaction and the databases run in WAL mode, so a reader sees the last +committed index, whole, until the writer commits. Treating a held write lock as +"mid-write" made every code tool fail for the length of every reindex -- two +thirds of code_search calls on a large repo. Such a state is now ``ready`` and +``refreshing``: readable, possibly behind the files being reindexed, and +announced as such by :func:`take_refreshing`. The lock still fails loud where +there is nothing committed to read: a first build that has written no files. + Nothing here writes to the engine's databases; see :mod:`lemoncrow.infra.code_intel.store` for that boundary. """ @@ -49,6 +58,7 @@ "DEFAULT_RECHECK_SECONDS", "FRESHNESS_FRESH", "FRESHNESS_REBUILT", + "FRESHNESS_REFRESHING", "INDEX_LOCK_SUFFIX", "LOCK_FREE", "LOCK_HELD", @@ -62,6 +72,7 @@ "index_state", "require_ready", "reset_readiness_probes", + "take_refreshing", ] logger = logging.getLogger(__name__) @@ -80,6 +91,8 @@ FRESHNESS_FRESH = "fresh" FRESHNESS_REBUILT = "rebuilt" +#: Answered from the last committed index while a reindex holds the write lock. +FRESHNESS_REFRESHING = "refreshing" LOCK_FREE = "free" LOCK_HELD = "held" @@ -117,10 +130,13 @@ class IndexState: ``status`` is the field that gates behaviour: ``ready`` - The index can be read. + The index can be read -- also while a reindex holds the write lock, in + which case :attr:`refreshing` is true and answers come from the last + committed index. ``rebuilding`` - Mid-write. A query against it would return a torn or empty view, so - callers must raise rather than return what they find. + Torn, or a first build with nothing committed yet. A query against it + would return a torn or empty view, so callers must raise rather than + return what they find. ``absent`` Never indexed, or indexed to nothing. An answer, not a failure -- the engine creates the databases on first use. @@ -135,6 +151,31 @@ class IndexState: def rebuilding(self) -> bool: return self.status == STATUS_REBUILDING + @property + def refreshing(self) -> bool: + """Readable while a reindex holds the lock; may lag the files it is reindexing.""" + return self.status == STATUS_READY and self.lock == LOCK_HELD + + +_refreshing_seen = threading.local() + + +def _noted(state: IndexState) -> IndexState: + if state.refreshing: + _refreshing_seen.value = True + return state + + +def take_refreshing() -> bool: + """Whether a probe on this thread answered during a reindex since the last take. + + Clears the mark. The MCP dispatcher takes it once before a tool call and once + after, so a response is flagged exactly when that call read a refreshing index. + """ + seen = bool(getattr(_refreshing_seen, "value", False)) + _refreshing_seen.value = False + return seen + def index_lock_path(repo_root: Path | str = ".") -> Path: """Path of the engine's index-write lock for *repo_root*.""" @@ -193,8 +234,11 @@ def index_state(repo_root: Path | str = ".") -> IndexState: 1. the database file is missing -> ``absent`` 2. a required table is missing -> ``rebuilding`` (caught mid-DDL) 3. symbols without files -> ``rebuilding`` (a torn index) - 4. the index-write lock is held -> ``rebuilding`` - 5. no rows at all -> ``absent``; otherwise ``ready`` + 4. no files -> ``rebuilding`` while the index-write lock is held (a first + build has committed nothing to answer from), otherwise ``absent`` + 5. otherwise ``ready`` -- ``refreshing`` too if the lock is held, since a + reindex commits in one transaction and a WAL reader sees the last + committed index until it does Check 3 is deliberately one-directional. Symbols with no files cannot be a resting state -- every symbol row references a file row. Files with no @@ -231,11 +275,11 @@ def index_state(repo_root: Path | str = ".") -> IndexState: f"index partially populated ({files} files, {symbols} symbols)", lock, ) - if lock == LOCK_HELD: - return IndexState(version, STATUS_REBUILDING, "index-write lock is held", lock) if files == 0: + if lock == LOCK_HELD: + return IndexState(version, STATUS_REBUILDING, "first index build in progress", lock) return IndexState(version, STATUS_ABSENT, "index is empty", lock) - return IndexState(version, STATUS_READY, "", lock) + return _noted(IndexState(version, STATUS_READY, "", lock)) except sqlite3.Error as exc: # A torn database mid-rebuild reads as corruption. That is a rebuild in # progress, not a permanent failure, and it must not surface as empty. @@ -291,7 +335,7 @@ def state_for(self, repo_root: Path | str) -> IndexState: now = self._clock() probe = self._probes.get(key) if probe is not None and (now - probe.checked_at) < self.recheck_seconds: - return probe.state + return _noted(probe.state) state = index_state(repo_root) self._probes[key] = _Probe(state=state, checked_at=now) return state @@ -372,9 +416,9 @@ def require_ready(repo_root: Path | str = ".") -> IndexState: ``code_changes``, ``code_query``, ``code_coverage_check`` and the file-graph analytics open the engine's databases directly, so the engine cache's - rebuild check never ran for them: mid-reindex they read a torn index and - returned what was left, an empty answer delivered as a complete one. This - is that check, applied where they start. + rebuild check never ran for them: they could read a torn index and return + what was left, an empty answer delivered as a complete one. This is that + check, applied where they start. ``rebuilding`` raises :class:`IndexRebuilding`. ``absent`` raises :class:`~lemoncrow.infra.code_intel.store.CodeIntelUnavailable`: the probe diff --git a/tests/gateway/test_code_tools_during_reindex.py b/tests/gateway/test_code_tools_during_reindex.py new file mode 100644 index 000000000..fb9b00f7c --- /dev/null +++ b/tests/gateway/test_code_tools_during_reindex.py @@ -0,0 +1,87 @@ +"""Code tools answer while a reindex holds the index lock, and say so. + +On a large repo a reindex held the lock most of the time and every code tool +failed with "index is being rebuilt" for its whole length. A reindex commits in +one transaction, so the last committed index is whole and readable meanwhile; +the tools now serve it and append a note, so the model knows results may lag. +""" + +from __future__ import annotations + +import contextlib +import fcntl +from collections.abc import Iterator +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock + +import pytest + +from lemoncrow.gateway.adapters import mcp_server +from lemoncrow.infra.code_intel.freshness import INDEX_LOCK_SUFFIX, reset_readiness_probes +from lemoncrow.infra.code_intel.store import CODE_CONTEXT_DB, workspace_dir +from lemoncrow.pro.capabilities.code_context import CodeContextEngine +from tests.helpers import init_store_at + + +@pytest.fixture() +def indexed(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + store = tmp_path / ".lemoncrow" + init_store_at(str(store)) + monkeypatch.setenv("LEMONCROW_ROOT", str(store)) + monkeypatch.setenv("CLAUDE_WORKSPACE_ROOT", str(tmp_path)) + monkeypatch.chdir(tmp_path) + # setattr, not assignment: the calls below cache a ledger bound to this + # test's store, and a leaked one reroutes later tests' model recommendations. + monkeypatch.setattr(mcp_server._ledger, "_current_ledger", None) + monkeypatch.setattr(mcp_server._ledger, "_realtime_ctx", None) + remote = MagicMock() + remote.get_context.return_value = {"context": "", "run_ledger": []} + monkeypatch.setattr(mcp_server, "_remote_client", remote) + mcp_server._RECENT_CODE_SEARCH_QUERIES.clear() + (tmp_path / "billing.py").write_text("def reconcile_invoices():\n return 1\n", encoding="utf-8") + CodeContextEngine(tmp_path, autosync_enabled=False).index_repo(force=True) + return tmp_path + + +@contextlib.contextmanager +def _reindex_in_progress(root: Path) -> Iterator[None]: + lock_path = Path(str(workspace_dir(root) / CODE_CONTEXT_DB) + INDEX_LOCK_SUFFIX) + lock_path.touch() + with lock_path.open("r+") as handle: + fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + try: + yield + finally: + fcntl.flock(handle.fileno(), fcntl.LOCK_UN) + + +def _call(name: str, args: dict[str, Any]) -> tuple[bool, str]: + reset_readiness_probes() + mcp_server._code_engine_cache.clear() + resp = mcp_server._handle( + {"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": name, "arguments": args}} + ) + assert isinstance(resp, dict) and "result" in resp, resp + result = resp["result"] + return bool(result.get("isError")), str(result["content"][0]["text"]) + + +@pytest.mark.parametrize( + ("tool", "args"), + [ + ("code_search", {"query": "reconcile_invoices"}), # through the engine cache + ("code_query", {"select": "symbols"}), # through require_ready + ], +) +def test_a_code_tool_answers_during_a_reindex_and_notes_it(indexed: Path, tool: str, args: dict[str, Any]) -> None: + with _reindex_in_progress(indexed): + is_error, text = _call(tool, args) + + assert not is_error, text + assert "reconcile_invoices" in text + assert mcp_server._INDEX_REFRESHING_NOTE in text + + is_error, text = _call(tool, args) + assert not is_error, text + assert mcp_server._INDEX_REFRESHING_NOTE not in text, "the refreshing note leaked into a later call" diff --git a/tests/infra/code_intel/test_clones.py b/tests/infra/code_intel/test_clones.py index 628a271a7..c5b097cd3 100644 --- a/tests/infra/code_intel/test_clones.py +++ b/tests/infra/code_intel/test_clones.py @@ -858,7 +858,7 @@ def test_build_refuses_a_torn_index(clone_repo: Path, monkeypatch: pytest.Monkey monkeypatch.setattr( clones_mod, "index_state", - lambda root: IndexState(9, STATUS_REBUILDING, "index-write lock is held", "held"), + lambda root: IndexState(9, STATUS_REBUILDING, "first index build in progress", "held"), ) with pytest.raises(IndexRebuilding): build_clones(clone_repo) diff --git a/tests/infra/code_intel/test_freshness.py b/tests/infra/code_intel/test_freshness.py index 67c4d0b3d..5712b20f8 100644 --- a/tests/infra/code_intel/test_freshness.py +++ b/tests/infra/code_intel/test_freshness.py @@ -9,9 +9,10 @@ from __future__ import annotations +import contextlib import sqlite3 import threading -from collections.abc import Callable +from collections.abc import Callable, Iterator from pathlib import Path import pytest @@ -27,6 +28,7 @@ IndexRebuilding, VersionedEngineCache, index_state, + take_refreshing, ) from lemoncrow.infra.code_intel.store import CODE_CONTEXT_DB, workspace_dir @@ -125,25 +127,68 @@ def test_missing_table_reads_as_rebuilding(make_workspace: WorkspaceFactory) -> assert "imports" in state.detail -def test_held_index_lock_reads_as_rebuilding(make_workspace: WorkspaceFactory) -> None: +@contextlib.contextmanager +def _index_lock_held(root: Path) -> Iterator[None]: fcntl = pytest.importorskip("fcntl") - root = make_workspace(files=_FILES, symbols=_SYMBOLS, index_version=3) lock_path = Path(str(_code_db(root)) + INDEX_LOCK_SUFFIX) lock_path.touch() - - assert index_state(root).status == STATUS_READY # lock exists but is free - handle = lock_path.open("r+") try: fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) - state = index_state(root) + yield finally: fcntl.flock(handle.fileno(), fcntl.LOCK_UN) handle.close() - assert state.status == STATUS_REBUILDING + +def test_held_index_lock_on_a_populated_index_is_ready_and_refreshing(make_workspace: WorkspaceFactory) -> None: + """A reindex holds the lock but commits in one transaction; readers keep the last index.""" + root = make_workspace(files=_FILES, symbols=_SYMBOLS, index_version=3) + take_refreshing() + + free = index_state(root) + assert free.status == STATUS_READY and not free.refreshing + assert take_refreshing() is False + + with _index_lock_held(root): + state = index_state(root) + + assert state.status == STATUS_READY + assert state.refreshing assert state.lock == LOCK_HELD assert state.index_version == 3 + assert take_refreshing() is True + assert take_refreshing() is False, "the mark must clear once taken" + + +def test_held_index_lock_on_an_empty_index_is_a_first_build(make_workspace: WorkspaceFactory) -> None: + """Nothing is committed yet, so there is nothing to answer from: that still fails loud.""" + root = make_workspace(index_version=0) + + assert index_state(root).status == STATUS_ABSENT + with _index_lock_held(root): + state = index_state(root) + + assert state.status == STATUS_REBUILDING + assert state.detail == "first index build in progress" + + +def test_the_engine_cache_serves_a_refreshing_index_and_marks_every_probe(make_workspace: WorkspaceFactory) -> None: + root = make_workspace(files=_FILES, symbols=_SYMBOLS, index_version=3) + clock = _Clock() + cache = VersionedEngineCache("test", recheck_seconds=5.0, clock=clock) + first, _ = cache.get("k", root, object) + take_refreshing() + + with _index_lock_held(root): + clock.now = 10.0 + during, _ = cache.get("k", root, object) + assert take_refreshing() is True + clock.now = 11.0 # inside the throttle window: the cached probe is reused + cache.get("k", root, object) + assert take_refreshing() is True, "a throttled probe of a refreshing index went unmarked" + + assert during is first # --------------------------------------------------------------------------- # diff --git a/tests/infra/code_intel/test_reads_during_reindex.py b/tests/infra/code_intel/test_reads_during_reindex.py new file mode 100644 index 000000000..ba661edf6 --- /dev/null +++ b/tests/infra/code_intel/test_reads_during_reindex.py @@ -0,0 +1,120 @@ +"""Readers keep the whole previous index while a reindex runs in another process. + +The premise behind serving code tools during a reindex: the indexer writes in one +transaction, and the databases run in WAL mode, so until it commits a reader in +another process -- the MCP daemon, while autosync's `lc code index` subprocess +holds the lock -- sees the last committed index, not a torn one. These tests pause +a real indexer mid-write (its deletes executed, its inserts not yet) and look. +""" + +from __future__ import annotations + +import sqlite3 +import subprocess +import sys +import time +from pathlib import Path + +import pytest + +from lemoncrow.infra.code_intel.freshness import IndexRebuilding, index_state, require_ready, reset_readiness_probes +from lemoncrow.infra.code_intel.store import CODE_CONTEXT_DB, FTS_DB, workspace_dir +from lemoncrow.pro.capabilities.code_context import CodeContextEngine + +_WRITER = """ +import sys, time +from pathlib import Path +from lemoncrow.pro.capabilities.code_context import CodeContextEngine + +root, gate, force = Path(sys.argv[1]), Path(sys.argv[2]), sys.argv[3] == "1" +original = CodeContextEngine._parallel_extract + +def paused(self, *args, **kwargs): + (gate / "mid-write").touch() + deadline = time.monotonic() + 60 + while not (gate / "release").exists() and time.monotonic() < deadline: + time.sleep(0.02) + return original(self, *args, **kwargs) + +CodeContextEngine._parallel_extract = paused +CodeContextEngine(root, autosync_enabled=False).index_repo(force=force) +""" + + +def _repo(tmp_path: Path) -> Path: + root = (tmp_path / "repo").resolve() + (root / "pkg").mkdir(parents=True) + for i in range(4): + (root / "pkg" / f"mod{i}.py").write_text(f"def alpha_{i}():\n return {i}\n", encoding="utf-8") + subprocess.run(["git", "init", "-q"], cwd=root, check=True) + return root + + +def _start_writer(root: Path, gate: Path, *, force: bool) -> subprocess.Popen[bytes]: + gate.mkdir() + writer = subprocess.Popen([sys.executable, "-c", _WRITER, str(root), str(gate), "1" if force else "0"]) + deadline = time.monotonic() + 60 + while not (gate / "mid-write").exists(): + assert writer.poll() is None, "the writer exited before reaching its write" + assert time.monotonic() < deadline, "the writer never reached its write" + time.sleep(0.02) + return writer + + +def _finish(writer: subprocess.Popen[bytes], gate: Path) -> None: + (gate / "release").touch() + assert writer.wait(timeout=120) == 0 + + +def _committed(root: Path) -> tuple[set[str], int]: + """Symbol names and alpha-bearing lines, as a separate reader connection sees them.""" + ws = workspace_dir(root) + conn = sqlite3.connect(f"file:{ws / CODE_CONTEXT_DB}?mode=ro", uri=True) + try: + conn.execute("ATTACH DATABASE ? AS fts", (f"file:{ws / FTS_DB}?mode=ro",)) + names = {str(row[0]) for row in conn.execute("SELECT symbol_name FROM symbols")} + lines = int(conn.execute("SELECT COUNT(*) FROM fts.file_line_fts WHERE text LIKE '%alpha_%'").fetchone()[0]) + finally: + conn.close() + return names, lines + + +@pytest.mark.parametrize("force", [True, False], ids=["full-rebuild", "incremental"]) +def test_a_paused_reindex_leaves_readers_the_whole_previous_index(tmp_path: Path, force: bool) -> None: + root = _repo(tmp_path) + CodeContextEngine(root, autosync_enabled=False).index_repo(force=True) + before_names, before_lines = _committed(root) + assert {"alpha_0", "alpha_3"} <= before_names and before_lines == 4 + + (root / "pkg" / "mod0.py").write_text("def omega_0():\n return 0\n", encoding="utf-8") + gate = tmp_path / "gate" + writer = _start_writer(root, gate, force=force) + try: + reset_readiness_probes() + state = index_state(root) + assert state.refreshing, state + require_ready(root) # answers instead of raising IndexRebuilding + assert _committed(root) == (before_names, before_lines), "a reader saw the reindex's uncommitted deletes" + finally: + _finish(writer, gate) + + after_names, after_lines = _committed(root) + assert "omega_0" in after_names and "alpha_0" not in after_names + assert after_lines == 3 + + +def test_a_first_build_in_progress_still_refuses_to_answer(tmp_path: Path) -> None: + """No committed index yet: an empty answer would read as "no such code", so raise.""" + root = _repo(tmp_path) + gate = tmp_path / "gate" + writer = _start_writer(root, gate, force=True) + try: + reset_readiness_probes() + assert index_state(root).rebuilding + with pytest.raises(IndexRebuilding): + require_ready(root) + finally: + _finish(writer, gate) + + reset_readiness_probes() + assert index_state(root).status == "ready"