diff --git a/src/lemoncrow/core/settings_registry.py b/src/lemoncrow/core/settings_registry.py index e067d4971..1094d50f0 100644 --- a/src/lemoncrow/core/settings_registry.py +++ b/src/lemoncrow/core/settings_registry.py @@ -718,6 +718,15 @@ class SettingSpec: "code_context", "Enable the on-disk tag cache for repo-map generation.", ), + SettingSpec( + "code_context.worktree_engine_idle_s", + "LEMONCROW_WORKTREE_ENGINE_IDLE_S", + "float", + 1800.0, + "code_context", + "Seconds a git worktree's code engine may go without a request before the daemon unloads it " + "(its index stays on disk; the next request reopens it). The main checkout's engine never unloads.", + ), SettingSpec( "code_context.zoekt_gate", "LEMONCROW_ZOEKT_GATE", diff --git a/src/lemoncrow/gateway/adapters/mcp_server.py b/src/lemoncrow/gateway/adapters/mcp_server.py index 117994603..d8530973b 100644 --- a/src/lemoncrow/gateway/adapters/mcp_server.py +++ b/src/lemoncrow/gateway/adapters/mcp_server.py @@ -165,6 +165,7 @@ tool_web_fetch, ) from lemoncrow.gateway.adapters.mcp_branding import icon_metadata +from lemoncrow.infra.code_intel import worktree_seed from lemoncrow.infra.code_intel.completeness import ( CODE_OP_MATCH_KINDS, CODE_OP_OBJECTIVES, @@ -3778,38 +3779,8 @@ def _record_session_cwd(name: str, args: Any) -> None: def _linked_worktree_root(workspace_root: Path, candidate_dir: Path) -> Path | None: - """The linked worktree of ``workspace_root`` that contains ``candidate_dir``, else None. - - Detected without spawning git: a linked worktree's ``.git`` is a *file* - holding ``gitdir: ``, and for a worktree of THIS repo that path lives - under ``/.git/worktrees/``. A normal checkout has ``.git`` - as a directory, which ends the walk immediately. - - Returns None for every uncertain case -- a plain directory, a worktree - belonging to a different repo, the workspace root itself. - """ - try: - candidate = candidate_dir.expanduser().resolve() - if not candidate.is_dir(): - return None - root = workspace_root.resolve() - worktrees_dir = (root / ".git" / "worktrees").resolve() - for directory in (candidate, *candidate.parents): - marker = directory / ".git" - if marker.is_dir(): - return None # a normal checkout, not a linked worktree - if not marker.is_file(): - continue - gitdir = marker.read_text(encoding="utf-8").strip() - if not gitdir.startswith("gitdir:"): - return None - target = Path(gitdir.split(":", 1)[1].strip()).resolve() - if not target.is_relative_to(worktrees_dir): - return None # a worktree, but of some other repo - return None if directory == root else directory - except OSError: - return None - return None + """The linked worktree of ``workspace_root`` that contains ``candidate_dir``, else None.""" + return worktree_seed.linked_worktree_of(workspace_root, candidate_dir) def _session_worktree_root(workspace_root: Path) -> Path | None: @@ -8929,9 +8900,15 @@ def _memory_summary(session_id: str) -> dict[str, Any]: # code_search answered "index is being rebuilt" for most calls. def _retire_code_engine(engine: Any) -> None: engine.stop_autosync() + # A scoped capability holds its engine; left cached it would pin a retired one. + with _scoped_context_cache_lock: + for key in [key for key, entry in _scoped_context_cache.items() if entry[1] is engine]: + del _scoped_context_cache[key] -_code_engine_cache = VersionedEngineCache("code_engine", on_evict=_retire_code_engine) +_code_engine_cache = VersionedEngineCache( + "code_engine", on_evict=_retire_code_engine, retire=worktree_seed.retire_worktree_engine +) # ``cache_key -> (capability, engine_it_was_built_from)``. # @@ -9031,7 +9008,15 @@ def _code_context_engine(repo_root: str = ".") -> Any: root = Path(repo_root) resolved = (root if root.is_absolute() else Path(workspace) / root).resolve() cache_key = str(resolved) + # A linked worktree's index starts as a clone of its main checkout's. + seed = worktree_seed.ensure_seeded( + resolved, + cached=cache_key in _code_engine_cache, + before_swap=lambda: _code_engine_cache.discard(cache_key), + ) engine, freshness = _code_engine_cache.get(cache_key, resolved, lambda: CodeContextEngine(resolved)) + if seed is not None and seed.seeded: + worktree_seed.start_first_refresh(engine) _code_index_freshness_for_current_call.value = freshness return engine diff --git a/src/lemoncrow/gateway/cli/commands/code.py b/src/lemoncrow/gateway/cli/commands/code.py index a199e97ae..6f7f2e391 100644 --- a/src/lemoncrow/gateway/cli/commands/code.py +++ b/src/lemoncrow/gateway/cli/commands/code.py @@ -700,7 +700,11 @@ def code_host_remove_cmd(ctx: click.Context, engine: str, yes: bool) -> None: ) @click.option("--include", "include_globs", multiple=True) @click.option("--exclude", "exclude_globs", multiple=True) -@click.option("--reindex", is_flag=True, help="Full rebuild from scratch (default: incremental).") +@click.option( + "--reindex", + is_flag=True, + help="Full rebuild from scratch (default: incremental); in a linked git worktree, re-seed from the main checkout's index when it can.", +) @click.option( "--force", "steal_lock", @@ -725,7 +729,10 @@ def code_index_cmd( """Index a repository into the SQLite FTS5 symbol store. Incremental by default (only re-indexes changed files). Use --reindex - for a full rebuild from scratch. + for a full rebuild from scratch. In a linked git worktree whose main + checkout has a current index, --reindex re-seeds from that index instead, + then re-indexes only the files that differ; a seeded worktree index is + never rebuilt. """ if repo_root is None: # Resolve the same way the MCP code_search / read tools do @@ -734,8 +741,23 @@ def code_index_cmd( from lemoncrow.core.foundation.paths import resolve_workspace_root repo_root = str(resolve_workspace_root()) - engine = _code_context_engine(repo_root, db_path=Path(db_path) if db_path else None) + from lemoncrow.infra.code_intel import worktree_seed + force = reindex + seeded_worktree = False + if db_path is None: + # A linked worktree's index is seeded from its main checkout's, and --reindex + # re-seeds it: a full build would un-share every page of the clone. + from lemoncrow.infra.code_intel.freshness import IndexRebuilding + + try: + seed = worktree_seed.ensure_seeded(Path(repo_root).resolve(), reseed=reindex) + except IndexRebuilding as exc: + raise click.ClickException(str(exc)) from exc + if seed is not None and seed.seeded: + force = False + seeded_worktree = worktree_seed.seeded_main_root(Path(repo_root).resolve()) is not None + engine = _code_context_engine(repo_root, db_path=Path(db_path) if db_path else None) if as_json: payload = engine.index_repo( force=force, @@ -744,14 +766,16 @@ def code_index_cmd( include_globs=list(include_globs) or None, exclude_globs=list(exclude_globs) or None, ).model_dump(mode="json") + worktree_seed.checkpoint_index(Path(engine.db_path).parent) # so a worktree seed finds a small WAL try: engine._deleted_history_adapter()._ensure_history_ready() except Exception: logging.exception("Failed to prepare background indexes") - try: - _trigger_zoekt_with_progress(Path(repo_root).resolve(), quiet=True) - except Exception: - logging.exception("Failed to prewarm Zoekt index") + if not seeded_worktree: # a seeded worktree searches its main checkout's Zoekt index + try: + _trigger_zoekt_with_progress(Path(repo_root).resolve(), quiet=True) + except Exception: + logging.exception("Failed to prewarm Zoekt index") _emit(payload, as_json=True) return @@ -765,9 +789,11 @@ def code_index_cmd( success_description="Indexed code", frame_prefix=frame_prefix, ) + worktree_seed.checkpoint_index(Path(engine.db_path).parent) # so a worktree seed finds a small WAL git_summary = _index_git_history_with_progress(engine, frame_prefix=frame_prefix) - _trigger_zoekt_with_progress(Path(repo_root).resolve(), frame_prefix=frame_prefix) + if not seeded_worktree: + _trigger_zoekt_with_progress(Path(repo_root).resolve(), frame_prefix=frame_prefix) stats_line = ( f"{click.style('✓', fg='green')} Indexed {payload['files_indexed']} files, {payload['symbols_indexed']} " diff --git a/src/lemoncrow/infra/code_intel/freshness.py b/src/lemoncrow/infra/code_intel/freshness.py index a1057ff3e..a009170a0 100644 --- a/src/lemoncrow/infra/code_intel/freshness.py +++ b/src/lemoncrow/infra/code_intel/freshness.py @@ -318,6 +318,12 @@ class VersionedEngineCache: 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. + + *retire* decides which entries to drop on its own: it is called with each + key and the seconds since that key was last requested, and True discards the + entry (through *on_evict*). The check runs on every :meth:`get` and on a + background timer every *sweep_seconds*, so an entry nobody requests again + still goes. """ def __init__( @@ -326,15 +332,22 @@ def __init__( recheck_seconds: float = DEFAULT_RECHECK_SECONDS, clock: Callable[[], float] = time.monotonic, on_evict: Callable[[Any], None] | None = None, + retire: Callable[[str, float], bool] | None = None, + sweep_seconds: float = 60.0, ) -> None: self.name = name self.recheck_seconds = float(recheck_seconds) self.on_evict = on_evict + self.retire = retire + self.sweep_seconds = float(sweep_seconds) self.evictions = 0 self._clock = clock self._lock = threading.Lock() self._entries: dict[str, _Entry] = {} self._probes: dict[str, _Probe] = {} + self._last_access: dict[str, float] = {} + self._sweeper: threading.Thread | None = None + self._sweeper_stop = threading.Event() # -- probing ----------------------------------------------------------- @@ -363,6 +376,10 @@ def get(self, key: str, repo_root: Path | str, build: Callable[[], Any]) -> tupl results from a torn index is the failure mode this whole module exists to remove. """ + self._last_access[key] = self._clock() + if self.retire is not None: + self._ensure_sweeper() + self.sweep(skip=key) state = self.state_for(repo_root) if state.rebuilding: raise IndexRebuilding(repo_root, state.detail) @@ -412,14 +429,58 @@ def version_of(self, key: str) -> int | None: def discard(self, key: str) -> None: with self._lock: entry = self._entries.pop(key, None) + self._probes.pop(key, None) + self._last_access.pop(key, None) if entry is not None: self._retire(entry.value) + def sweep(self, *, skip: str | None = None) -> list[str]: + """Discard every entry *retire* says to; returns the keys discarded.""" + retire = self.retire + if retire is None: + return [] + now = self._clock() + with self._lock: + idle = {key: now - self._last_access.get(key, now) for key in self._entries if key != skip} + retired: list[str] = [] + for key, idle_seconds in idle.items(): + try: + if not retire(key, idle_seconds): + continue + except Exception: + logger.warning("%s: retire check for %s failed", self.name, key, exc_info=True) + continue + logger.info("%s: retiring %s after %.0fs idle", self.name, key, idle_seconds) + self.discard(key) + retired.append(key) + return retired + + def _ensure_sweeper(self) -> None: + if self._sweeper is not None: + return + with self._lock: + if self._sweeper is not None: + return + self._sweeper = threading.Thread(target=self._sweep_loop, name=f"{self.name}-sweeper", daemon=True) + self._sweeper.start() + + def _sweep_loop(self) -> None: + while not self._sweeper_stop.wait(self.sweep_seconds): + try: + self.sweep() + except Exception: + logger.warning("%s: sweep failed", self.name, exc_info=True) + + def stop_sweeper(self) -> None: + """End the background sweep; :meth:`get` still sweeps.""" + self._sweeper_stop.set() + def clear(self) -> None: with self._lock: dropped = [entry.value for entry in self._entries.values()] self._entries.clear() self._probes.clear() + self._last_access.clear() self.evictions = 0 for value in dropped: self._retire(value) diff --git a/src/lemoncrow/infra/code_intel/store.py b/src/lemoncrow/infra/code_intel/store.py index 5d4245cda..a5901b256 100644 --- a/src/lemoncrow/infra/code_intel/store.py +++ b/src/lemoncrow/infra/code_intel/store.py @@ -24,6 +24,7 @@ __all__ = [ "CODE_CONTEXT_DB", + "CODE_INDEXER_SEMANTICS_VERSION", "FTS_DB", "INTEL_DB", "REPO_MAP_TAGS_DB", @@ -47,6 +48,13 @@ VECTORS_DB = "vectors.sqlite" REPO_MAP_TAGS_DB = "repo_map_tags.sqlite" +#: The engine's indexer format, stamped into ``engine_state`` as +#: ``indexer_semantics_version``. Bump when source selection or symbol/text +#: extraction semantics change in a way an incremental mtime/hash check cannot +#: see for unchanged files. +#: 3: FTS5 rows carry explicit rowids derived from files.rowid / symbols.rowid. +CODE_INDEXER_SEMANTICS_VERSION = 3 + # The engine holds WAL writers during a reindex; wait rather than fail fast. _BUSY_TIMEOUT_MS = 5_000 diff --git a/src/lemoncrow/infra/code_intel/worktree_seed.py b/src/lemoncrow/infra/code_intel/worktree_seed.py new file mode 100644 index 000000000..6e01d7dde --- /dev/null +++ b/src/lemoncrow/infra/code_intel/worktree_seed.py @@ -0,0 +1,741 @@ +"""Seed a linked git worktree's code index from its main checkout's index. + +A worktree index built from scratch costs a full build: minutes and gigabytes for +a large repo. The worktree's files are almost all identical to the main +checkout's, so its index starts as a copy-on-write clone of main's index +instead (``clonefile`` on APFS, ``--reflink`` on Linux, a plain copy +elsewhere). One incremental pass then re-extracts only the files whose content +differs. + +The clone keeps main's ``repo_id``. Every row is keyed by it, and rewriting the +key would un-share every page of the clone, so the seed records an alias instead: +``engine_state['repo_id_alias:'] =
``. The engine adopts +the alias when it opens the database (:func:`resolve_repo_id`). The alias is +keyed by the worktree's own id, so another repo sharing the same database file +keeps its own id and rows. + +The seed also writes ``engine_state['seeded_from'] =
@
``. A seeded index is never fully rebuilt or vacuumed: either +would rewrite every page of the clone. When its format goes stale, the factory +re-seeds it from main's rebuilt index. + +This module also owns the worktree engines' lifecycle in the daemon: which cached +roots are worktree engines, when they retire, and where their Zoekt candidates +come from. +""" + +from __future__ import annotations + +import contextlib +import hashlib +import logging +import os +import shutil +import sqlite3 +import subprocess +import sys +import tempfile +import threading +import time +from collections.abc import Callable, Iterator +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from lemoncrow.core.foundation.paths import resolve_workspace_store_dir, workspace_store_dir +from lemoncrow.infra.code_intel.freshness import INDEX_LOCK_SUFFIX, IndexRebuilding +from lemoncrow.infra.code_intel.store import ( + CODE_CONTEXT_DB, + CODE_INDEXER_SEMANTICS_VERSION, + FTS_DB, + INTEL_DB, + VECTORS_DB, +) + +try: + import fcntl +except ImportError: # pragma: no cover - non-POSIX platforms + fcntl = None # type: ignore[assignment] + +__all__ = [ + "ALIAS_KEY_PREFIX", + "BUSY", + "INDEX_DBS", + "SEEDED", + "SEEDED_FROM_KEY", + "UNAVAILABLE", + "WORKTREE_ENGINE_IDLE_ENV", + "SeedResult", + "checkpoint_index", + "ensure_seeded", + "forget_worktree", + "is_seeded_index", + "linked_worktree_of", + "main_root_of", + "path_repo_id", + "resolve_repo_id", + "retire_worktree_engine", + "seed_worktree_index", + "seeded_main_root", + "start_first_refresh", + "worktree_engine_idle_s", +] + +logger = logging.getLogger(__name__) + +#: The databases an index consists of. Everything else in the store (session +#: state, blocks, rubrics, loop discipline, Zoekt shards) is never cloned. +INDEX_DBS: tuple[str, ...] = (FTS_DB, INTEL_DB, VECTORS_DB, CODE_CONTEXT_DB) + +ALIAS_KEY_PREFIX = "repo_id_alias:" +SEEDED_FROM_KEY = "seeded_from" + +SEEDED = "seeded" +BUSY = "busy" +UNAVAILABLE = "unavailable" + +#: How long a seed waits for main's index-write lock before reporting busy. +SEED_LOCK_WAIT_S = 2.0 +#: How long a seed waits for SQLite's own write lock on each of main's databases. +#: Writers outside the index flock (the retrieval cache, the centrality map) hold +#: it for one short transaction. +_WRITE_LOCK_WAIT_S = 2.0 +#: How often the factory re-checks a worktree it already has an engine for, so a +#: worktree whose index went stale (main rebuilt to a new format) gets re-seeded. +SEED_RECHECK_S = 60.0 + +WORKTREE_ENGINE_IDLE_ENV = "LEMONCROW_WORKTREE_ENGINE_IDLE_S" +#: Seconds a worktree engine may go without a request before the daemon unloads +#: it (``code_context.worktree_engine_idle_s``). Measured on a symphony-alpha clone +#: (17,870 indexed files): five seeded worktree engines added 190 MiB RSS to a +#: process holding main's -- 140 MiB of it with the first, ~13 MiB for each one +#: after -- and reopening one takes about a second. Half an hour keeps an engine +#: through a pause in a session at that price. +DEFAULT_WORKTREE_ENGINE_IDLE_S = 1800.0 + +_STAGING_PREFIX = ".seed-" + +_lock = threading.Lock() +#: Linked worktree root -> its main checkout, for every worktree engine the +#: factory has opened. The retire policy reads it: only these roots idle-retire. +_worktrees: dict[str, Path] = {} +#: Monotonic time a worktree's seed need was last checked. +_checked_at: dict[str, float] = {} +#: Serialises ensure_seeded's check-and-seed within the process. Two threads +#: opening one unseeded worktree would otherwise both seed it: the loser's +#: non-blocking flock on the worktree's own index fails and surfaces as +#: IndexRebuilding, or it seeds again and swaps files under the engine the first +#: one just handed out. +#: lc-debt: one lock for every worktree, so checking one waits out another's seed +#: (seconds, once per worktree per index format); upgrade path: per-root locks +#: dropped by forget_worktree. +_seed_lock = threading.Lock() + + +@dataclass(frozen=True) +class SeedResult: + """How a seed attempt ended, with the timings the measurements report.""" + + status: str + detail: str = "" + checkpoint_s: float = 0.0 + clone_s: float = 0.0 + wal_cloned: tuple[str, ...] = () + + @property + def seeded(self) -> bool: + return self.status == SEEDED + + +@dataclass(frozen=True) +class _IndexFacts: + files: int + semantics: int | None + index_version: int + seeded_from: str | None + + +def path_repo_id(root: Path) -> str: + """The engine's ``repo_id`` for *root* before any alias: a hash of its resolved path.""" + return hashlib.sha256(str(root.resolve()).encode("utf-8")).hexdigest()[:16] + + +def _open_ro(db: Path) -> sqlite3.Connection: + conn = sqlite3.connect(f"file:{db}?mode=ro", uri=True, timeout=5.0) + conn.execute("PRAGMA busy_timeout = 5000") + return conn + + +def resolve_repo_id(db_path: Path, computed_id: str) -> str: + """The ``repo_id`` an engine opening *db_path* as *computed_id* should use. + + A seeded worktree index carries an alias to its main checkout's id; every + other database, including one shared by several repos, maps an id to itself. + """ + if not db_path.is_file(): + return computed_id + try: + conn = _open_ro(db_path) + except sqlite3.Error: + return computed_id + try: + row = conn.execute("SELECT value FROM engine_state WHERE key = ?", (ALIAS_KEY_PREFIX + computed_id,)).fetchone() + except sqlite3.Error: + return computed_id + finally: + conn.close() + return str(row[0]) if row is not None and row[0] else computed_id + + +def effective_repo_id(root: Path) -> str: + """*root*'s ``repo_id`` in its own default index, after alias resolution.""" + return resolve_repo_id(workspace_store_dir(root) / CODE_CONTEXT_DB, path_repo_id(root)) + + +def is_seeded_index(conn: sqlite3.Connection) -> bool: + """Whether the index behind *conn* was seeded from a main checkout's.""" + try: + return conn.execute("SELECT 1 FROM engine_state WHERE key = ?", (SEEDED_FROM_KEY,)).fetchone() is not None + except sqlite3.Error: + return False + + +def linked_worktree_of(workspace_root: Path, candidate_dir: Path) -> Path | None: + """The linked worktree of ``workspace_root`` that contains ``candidate_dir``, else None. + + Detected without spawning git: a linked worktree's ``.git`` is a *file* + holding ``gitdir: ``, and for a worktree of THIS repo that path lives + under ``/.git/worktrees/``. A normal checkout has ``.git`` + as a directory, which ends the walk immediately. + + Returns None for every uncertain case -- a plain directory, a worktree + belonging to a different repo, the workspace root itself. + """ + try: + candidate = candidate_dir.expanduser().resolve() + if not candidate.is_dir(): + return None + root = workspace_root.resolve() + worktrees_dir = (root / ".git" / "worktrees").resolve() + for directory in (candidate, *candidate.parents): + marker = directory / ".git" + if marker.is_dir(): + return None # a normal checkout, not a linked worktree + if not marker.is_file(): + continue + gitdir = marker.read_text(encoding="utf-8").strip() + if not gitdir.startswith("gitdir:"): + return None + target = Path(gitdir.split(":", 1)[1].strip()).resolve() + if not target.is_relative_to(worktrees_dir): + return None # a worktree, but of some other repo + return None if directory == root else directory + except OSError: + return None + return None + + +def main_root_of(worktree: Path) -> Path | None: + """The main checkout *worktree* is a linked worktree of, else None. + + None for a normal checkout, a submodule, a bare repository's worktree, or + anything unreadable. + """ + marker = worktree / ".git" + try: + if not marker.is_file(): + return None + text = marker.read_text(encoding="utf-8").strip() + if not text.startswith("gitdir:"): + return None + gitdir = Path(text.split(":", 1)[1].strip()) + if not gitdir.is_absolute(): + gitdir = worktree / gitdir + gitdir = gitdir.resolve() + if gitdir.parent.name != "worktrees": + return None # e.g. a submodule's .git/modules/ + commondir_file = gitdir / "commondir" + common = gitdir.parent.parent + if commondir_file.is_file(): + raw = Path(commondir_file.read_text(encoding="utf-8").strip()) + common = (raw if raw.is_absolute() else gitdir / raw).resolve() + if common.name != ".git" or not common.is_dir(): + return None # a bare repository has no main checkout + main = common.parent + return main if main != worktree.resolve() else None + except OSError: + return None + + +def _read_facts(db: Path, repo_id: str | None) -> _IndexFacts | None: + """The index state a seed decision needs; None when *db* is missing or unreadable. + + *repo_id* None counts every file row, which is what a worktree's own index + holds whatever key its rows were written under. + """ + if not db.is_file(): + return None + try: + conn = _open_ro(db) + except sqlite3.Error: + return None + try: + state = { + str(key): str(value) + for key, value in conn.execute( + "SELECT key, value FROM engine_state WHERE key IN (?, ?, ?)", + ("indexer_semantics_version", "index_version", SEEDED_FROM_KEY), + ) + } + if repo_id is None: + files = int(conn.execute("SELECT COUNT(*) FROM files").fetchone()[0]) + else: + files = int(conn.execute("SELECT COUNT(*) FROM files WHERE repo_id = ?", (repo_id,)).fetchone()[0]) + except sqlite3.Error: + return None + finally: + conn.close() + semantics: int | None + try: + semantics = int(state["indexer_semantics_version"]) + except (KeyError, ValueError): + semantics = None + try: + index_version = int(state.get("index_version", "0")) + except ValueError: + index_version = 0 + return _IndexFacts( + files=files, semantics=semantics, index_version=index_version, seeded_from=state.get(SEEDED_FROM_KEY) + ) + + +def _main_unavailable(facts: _IndexFacts | None, current: int) -> str | None: + """Why main's index cannot seed a worktree, or None when it can.""" + if facts is None or facts.files == 0: + return "the main checkout has no index" + if facts.semantics != current: + return f"the main checkout's index is at semantics version {facts.semantics}, not {current}" + return None + + +def _seed_reason(worktree_facts: _IndexFacts | None, main_facts: _IndexFacts | None, current: int) -> str | None: + """Why a worktree index should be replaced by a seed, or None when it should stay.""" + if worktree_facts is None or worktree_facts.files == 0: + return "missing" + if worktree_facts.semantics != current: + return "stale" + if worktree_facts.seeded_from is None and main_facts is not None and worktree_facts.files < main_facts.files: + return "partial" + return None + + +@contextlib.contextmanager +def _flock(lock_path: Path, *, wait_s: float) -> Iterator[bool]: + """The engine's cross-process index-write lock; yields whether it was acquired.""" + if fcntl is None: # pragma: no cover - non-POSIX platforms + yield True + return + lock_path.parent.mkdir(parents=True, exist_ok=True) + fd = os.open(str(lock_path), os.O_RDWR | os.O_CREAT, 0o644) + acquired = False + try: + deadline = time.monotonic() + max(0.0, wait_s) + while True: + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + acquired = True + break + except OSError: + if time.monotonic() >= deadline: + break + time.sleep(0.05) + yield acquired + finally: + if acquired: + with contextlib.suppress(OSError): + fcntl.flock(fd, fcntl.LOCK_UN) + os.close(fd) + + +def _lock_held(lock_path: Path) -> bool: + with _flock(lock_path, wait_s=0.0) as acquired: + return not acquired + + +def _clone_file(src: Path, dst: Path) -> None: + """Copy *src* to *dst*, sharing its blocks where the filesystem can.""" + if sys.platform == "darwin": + command = ["cp", "-c", str(src), str(dst)] + elif sys.platform.startswith("linux"): + command = ["cp", "--reflink=auto", str(src), str(dst)] + else: + shutil.copyfile(src, dst) + return + result = subprocess.run(command, capture_output=True, check=False) + if result.returncode != 0: + # clonefile needs both paths on one APFS volume; fall back to a real copy. + shutil.copyfile(src, dst) + + +def _checkpoint(db: Path) -> bool: + """Run a PASSIVE checkpoint on *db*; True when its WAL is now fully in the database file.""" + conn = sqlite3.connect(db, timeout=_WRITE_LOCK_WAIT_S) + try: + _busy, log_frames, checkpointed = conn.execute("PRAGMA wal_checkpoint(PASSIVE)").fetchone() + finally: + conn.close() + # (0, -1, -1) is a database that is not in WAL mode: there is no WAL to carry. + return int(log_frames) == int(checkpointed) + + +def checkpoint_index(store: Path, *, attempts: int = 3, pause_s: float = 0.5) -> bool: + """Fold the WALs of the index databases in *store* into their files; True when all are empty. + + Run after a reindex, so a seed usually finds a small WAL: a clone carries the + un-checkpointed WAL along, and checkpointing it later rewrites -- un-shares -- + every page it touches. SQLite's own auto-checkpoint stops short while a reader + holds an older snapshot and does not run again until the next commit, which + left symphony-alpha with 195k frames (808 MB) in fts.sqlite's WAL: a 5.4 s + checkpoint on the seed's path. + """ + pending = [store / name for name in INDEX_DBS if (store / name).is_file()] + for attempt in range(max(1, attempts)): + if attempt: + time.sleep(pause_s) + remaining: list[Path] = [] + for db in pending: + try: + done = _checkpoint(db) + except sqlite3.Error: + done = False + if not done: + remaining.append(db) + pending = remaining + if not pending: + return True + return False + + +def _snapshot(main_store: Path, staging: Path) -> tuple[float, float, tuple[str, ...]]: + """Clone main's index databases into *staging*; returns (checkpoint_s, clone_s, wal_cloned). + + Callers hold main's index flock, which stops every indexer. It does not stop + the query-time writers (the retrieval cache in ``code_context.sqlite`` and the + centrality map in ``intel.sqlite``), so each database is also held under + SQLite's own write lock while it is cloned: with no commit in flight, the + database file and its WAL are one consistent state. The bulk checkpoint runs + before those locks are taken, so query-time writers wait only for the clone. + """ + names = [name for name in INDEX_DBS if (main_store / name).is_file()] + started = time.monotonic() + for name in names: + _checkpoint(main_store / name) + checkpoint_s = time.monotonic() - started + holders: list[sqlite3.Connection] = [] + wal_cloned: list[str] = [] + try: + for name in names: + holder = sqlite3.connect(main_store / name, timeout=_WRITE_LOCK_WAIT_S, isolation_level=None) + holders.append(holder) + holder.execute("BEGIN IMMEDIATE") + cloning = time.monotonic() + for name in names: + source = main_store / name + backfilled = _checkpoint(source) + _clone_file(source, staging / name) + wal = source.with_name(name + "-wal") + if not backfilled and wal.is_file() and wal.stat().st_size > 0: + _clone_file(wal, staging / wal.name) + wal_cloned.append(name) + clone_s = time.monotonic() - cloning + finally: + for holder in holders: + with contextlib.suppress(sqlite3.Error): + holder.execute("ROLLBACK") + holder.close() + return checkpoint_s, clone_s, tuple(wal_cloned) + + +def _mark(staging: Path, *, worktree_id: str, main_id: str, seeded_from: str, previous_version: int) -> None: + """Stamp the staged clone as a seed and fold every cloned WAL into its database. + + The swap then moves database files only, so a WAL can never end up beside a + database it does not belong to. + + The index_version is main's unless that equals the version the worktree index + being replaced had: the daemon's engine cache rebuilds an engine only when the + version it was built at moves, and an engine left on the replaced files would + keep serving them. + """ + code_db = staging / CODE_CONTEXT_DB + conn = sqlite3.connect(code_db, timeout=5.0) + try: + upsert = ( + "INSERT INTO engine_state(key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value" + ) + conn.execute(upsert, (ALIAS_KEY_PREFIX + worktree_id, main_id)) + conn.execute(upsert, (SEEDED_FROM_KEY, seeded_from)) + row = conn.execute("SELECT value FROM engine_state WHERE key = 'index_version'").fetchone() + version = int(row[0]) if row is not None else 0 + if version == previous_version: + conn.execute(upsert, ("index_version", str(version + 1))) + # Keyed by (repo_id, index_version), both of which the clone shares with + # main: a cached payload would answer for the worktree with main's paths. + with contextlib.suppress(sqlite3.OperationalError): + conn.execute("DELETE FROM retrieval_cache") + conn.commit() + finally: + conn.close() + for name in INDEX_DBS: + db = staging / name + if db.is_file() and db.with_name(name + "-wal").exists(): + folder = sqlite3.connect(db, timeout=5.0) + try: + folder.execute("PRAGMA wal_checkpoint(TRUNCATE)") + finally: + folder.close() + + +def _sweep_staging(parent: Path) -> None: + """Remove staging directories a crashed seed left behind.""" + with contextlib.suppress(OSError): + for entry in parent.iterdir(): + if entry.name.startswith(_STAGING_PREFIX) and entry.is_dir(): + shutil.rmtree(entry, ignore_errors=True) + + +def _swap(staging: Path, store: Path, parent: Path) -> None: + """Replace the worktree's index databases with the staged seed. + + Each database moves by one atomic rename over the old one, so its path is + never missing and never half-copied. Its old WAL and shared-memory files are + moved aside first: SQLite pairs a database with whatever ``-wal`` sits beside + it, and the old one's frames would be applied to the seed. + + lc-debt: a reader already holding the replaced files open (another thread's + in-flight query, another process) keeps reading them and can pair with the new + ones until it reconnects; the daemon retires its own engine before the swap. + Upgrade path: a versioned store directory switched by one rename. + """ + aside = Path(tempfile.mkdtemp(prefix=_STAGING_PREFIX + "old-", dir=parent)) + try: + for name in INDEX_DBS: + for suffix in ("-wal", "-shm"): + side = store / (name + suffix) + if side.exists(): + os.replace(side, aside / side.name) + staged = staging / name + if staged.is_file(): + os.replace(staged, store / name) + elif (store / name).exists(): + os.replace(store / name, aside / name) + finally: + shutil.rmtree(aside, ignore_errors=True) + + +def seed_worktree_index( + worktree_root: Path, + main_root: Path, + *, + lock_wait_s: float | None = None, + before_swap: Callable[[], None] | None = None, +) -> SeedResult: + """Replace *worktree_root*'s index with a clone of *main_root*'s. + + ``unavailable`` when main's index is empty or at an older format -- the caller + serves the worktree as before. ``busy`` when main's index-write lock stays + held past *lock_wait_s* (default :data:`SEED_LOCK_WAIT_S`), or the worktree's + own index is being written. + *before_swap* runs after the clone is staged and before it replaces anything, + so an owner can retire an engine still reading the old files. + """ + worktree = worktree_root.resolve() + main = main_root.resolve() + main_store = workspace_store_dir(main) + main_db = main_store / CODE_CONTEXT_DB + main_lock = main_store / (CODE_CONTEXT_DB + INDEX_LOCK_SUFFIX) + main_id = effective_repo_id(main) + current = CODE_INDEXER_SEMANTICS_VERSION + unavailable = _main_unavailable(_read_facts(main_db, main_id), current) + if unavailable is not None: + if main_lock.exists() and _lock_held(main_lock): + return SeedResult(BUSY, f"the main checkout's index is being rebuilt ({unavailable})") + return SeedResult(UNAVAILABLE, unavailable) + + store = resolve_workspace_store_dir(workspace_root=worktree) + store.mkdir(parents=True, exist_ok=True) + parent = store.parent + previous = _read_facts(store / CODE_CONTEXT_DB, None) + with _flock(store / (CODE_CONTEXT_DB + INDEX_LOCK_SUFFIX), wait_s=0.0) as own: + if not own: + return SeedResult(BUSY, "the worktree's own index is being written") + _sweep_staging(parent) + staging = Path(tempfile.mkdtemp(prefix=_STAGING_PREFIX, dir=parent)) + try: + with _flock(main_lock, wait_s=SEED_LOCK_WAIT_S if lock_wait_s is None else lock_wait_s) as held: + if not held: + return SeedResult(BUSY, "the main checkout's index is being written") + facts = _read_facts(main_db, main_id) + unavailable = _main_unavailable(facts, current) + if facts is None or unavailable is not None: + return SeedResult(UNAVAILABLE, unavailable or "the main checkout has no index") + try: + checkpoint_s, clone_s, wal_cloned = _snapshot(main_store, staging) + except sqlite3.OperationalError as exc: + return SeedResult(BUSY, f"the main checkout's index is locked: {exc}") + _mark( + staging, + worktree_id=path_repo_id(worktree), + main_id=main_id, + seeded_from=f"{main}@{facts.index_version}", + previous_version=previous.index_version if previous is not None else -1, + ) + if before_swap is not None: + before_swap() + _swap(staging, store, parent) + finally: + shutil.rmtree(staging, ignore_errors=True) + logger.info( + "worktree_seed: seeded %s from %s (checkpoint %.2fs, clone %.2fs, WAL cloned for %s)", + worktree, + main, + checkpoint_s, + clone_s, + ", ".join(wal_cloned) or "none", + ) + return SeedResult(SEEDED, "", checkpoint_s, clone_s, wal_cloned) + + +def _route_zoekt(worktree: Path, main: Path, *, seeded: bool) -> None: + """Serve *worktree*'s Zoekt searches from *main*'s server when *seeded*, else from its own.""" + try: + from lemoncrow.infra.code_intel.zoekt.adapter import clear_zoekt_root_override, set_zoekt_root_override + except ImportError: # pragma: no cover - zoekt adapter is part of the package + return + if seeded: + set_zoekt_root_override(worktree, main) + else: + clear_zoekt_root_override(worktree) + + +def ensure_seeded( + root: Path, + *, + cached: bool = False, + reseed: bool = False, + lock_wait_s: float | None = None, + before_swap: Callable[[], None] | None = None, +) -> SeedResult | None: + """Seed *root*'s index from its main checkout's when *root* is a linked worktree that needs one. + + None when *root* is not a linked worktree, or its index is complete and + current. A missing index, a partial one (unseeded, fewer files than main's) + and one at an older format are replaced; *reseed* replaces any. *cached* says + the caller already holds an engine for *root*, which limits the check to one + per :data:`SEED_RECHECK_S`. + + Raises :class:`IndexRebuilding` while main's index is being written: the + worktree has nothing complete to answer from until the seed lands, and the + next call retries. + """ + key = str(root) + now = time.monotonic() + with _lock: + main = _worktrees.get(key) + checked = _checked_at.get(key) + if main is not None and cached and not reseed and checked is not None and now - checked < SEED_RECHECK_S: + return None + if main is None: + main = main_root_of(root) + if main is None: + return None + with _lock: + _worktrees[key] = main + _checked_at[key] = now + with _seed_lock: + worktree_facts = _read_facts(workspace_store_dir(root) / CODE_CONTEXT_DB, None) + main_facts = _read_facts(workspace_store_dir(main) / CODE_CONTEXT_DB, effective_repo_id(main)) + reason = "reseed" if reseed else _seed_reason(worktree_facts, main_facts, CODE_INDEXER_SEMANTICS_VERSION) + if reason is None: + _route_zoekt(root, main, seeded=worktree_facts is not None and worktree_facts.seeded_from is not None) + return None + result = seed_worktree_index(root, main, lock_wait_s=lock_wait_s, before_swap=before_swap) + if result.status == BUSY: + with _lock: + _checked_at.pop(key, None) + raise IndexRebuilding(root, f"seeding the worktree index from {main}: {result.detail}") + if result.seeded: + logger.info("worktree_seed: %s index was %s; seeded from %s", root, reason, main) + seeded = result.seeded or (worktree_facts is not None and worktree_facts.seeded_from is not None) + _route_zoekt(root, main, seeded=seeded) + return result + + +def seeded_main_root(root: Path) -> Path | None: + """The main checkout *root*'s index was seeded from, when *root* is a seeded linked worktree.""" + main = main_root_of(root) + if main is None: + return None + facts = _read_facts(workspace_store_dir(root) / CODE_CONTEXT_DB, None) + return main if facts is not None and facts.seeded_from is not None else None + + +def _first_refresh(engine: Any) -> None: + try: + with engine._autosync_lock: + engine._maybe_autosync_reindex_locked(known_change="seeded") + except Exception: + logger.exception("worktree_seed: first refresh after the seed failed") + + +def start_first_refresh(engine: Any) -> None: + """Bring a just-seeded engine's index to the worktree's files now, off the request path. + + Queries answer from the seed meanwhile, marked refreshing. Without this the + worktree's own edits would wait for autosync's next full-tree poll. + """ + if not getattr(engine, "_autosync_enabled", False): + return + threading.Thread(target=_first_refresh, args=(engine,), name="lemoncrow-worktree-seed-refresh", daemon=True).start() + + +def worktree_engine_idle_s() -> float: + """``code_context.worktree_engine_idle_s``: idle seconds before a worktree engine unloads.""" + raw = os.environ.get(WORKTREE_ENGINE_IDLE_ENV, "").strip() + if not raw: + return DEFAULT_WORKTREE_ENGINE_IDLE_S + try: + return max(0.0, float(raw)) + except ValueError: + return DEFAULT_WORKTREE_ENGINE_IDLE_S + + +def forget_worktree(root: Path | str) -> None: + """Drop everything this module remembers about the worktree engine at *root*.""" + key = str(root) + with _lock: + main = _worktrees.pop(key, None) + _checked_at.pop(key, None) + if main is not None: + _route_zoekt(Path(key), main, seeded=False) + + +def retire_worktree_engine(key: str, idle_s: float) -> bool: + """Retire policy for the daemon's engine cache: True retires the entry at *key*. + + Only worktree engines retire: when the worktree is gone, or when they have had + no request for :func:`worktree_engine_idle_s`. A main checkout's engine never + does. A retiring worktree is forgotten here, Zoekt routing included. + + Gone means its ``.git`` file is gone, not its directory: a reindex still + running when ``git worktree remove`` deleted the checkout writes its store + back, and that recreates the directory. + """ + with _lock: + tracked = key in _worktrees + if not tracked: + return False + if (Path(key) / ".git").is_file() and idle_s < worktree_engine_idle_s(): + return False + forget_worktree(key) + return True diff --git a/src/lemoncrow/infra/code_intel/zoekt/adapter.py b/src/lemoncrow/infra/code_intel/zoekt/adapter.py index fae82f0b6..ab3f17f91 100644 --- a/src/lemoncrow/infra/code_intel/zoekt/adapter.py +++ b/src/lemoncrow/infra/code_intel/zoekt/adapter.py @@ -41,6 +41,10 @@ _SOURCE_PATH_PARTS = ("src", "lemoncrow") _SUPERVISORS: dict[str, ZoektSupervisor] = {} _SUPERVISORS_LOCK = threading.Lock() +#: Checkout root -> the root whose Zoekt index serves it. A seeded worktree's +#: files are its main checkout's at the same relative paths, so main's server +#: supplies its candidates instead of a build and a server per worktree. +_ROOT_OVERRIDES: dict[str, Path] = {} @dataclass(frozen=True) @@ -55,8 +59,11 @@ class ZoektBackendHealth: class ZoektSupervisor: """Session-scoped lifecycle owner for the search backend.""" - def __init__(self, repo_root: str | Path) -> None: + def __init__(self, repo_root: str | Path, *, checkout_root: str | Path | None = None) -> None: + # repo_root is the checkout whose index and server answer; checkout_root + # is the one results are read from and must exist in (a seeded worktree). self.repo_root = Path(repo_root).resolve() + self.checkout_root = Path(checkout_root).resolve() if checkout_root is not None else self.repo_root self._binary_resolution: ZoektBinaryResolution | None = None self._client: ZoektClient | None = None self._indexer = ZoektIndexer(self.repo_root) @@ -90,10 +97,19 @@ def should_route(self, search_path: str | Path) -> bool: cached = self._route_cache.get(cache_key) if cached is not None: return cached - should_route = self._indexer.line_count(search_path) >= threshold + should_route = self._indexer.line_count(self._served_path(Path(search_path))) >= threshold self._route_cache[cache_key] = should_route return should_route + def _served_path(self, path: Path) -> Path: + """*path* under the checkout, re-rooted under the served one.""" + if self.checkout_root == self.repo_root: + return path + try: + return self.repo_root / path.resolve().relative_to(self.checkout_root) + except ValueError: + return path + def _resolution(self) -> ZoektBinaryResolution: if self._binary_resolution is None: self._binary_resolution = discover_zoekt_binary(self.repo_root) @@ -180,6 +196,9 @@ def refresh_index_if_head_changed(self) -> bool: """ if zoekt_mode() == "off": return False + if self.checkout_root != self.repo_root: + # The served checkout's own supervisor owns its index and its build lock. + return get_zoekt_supervisor(self.repo_root).refresh_index_if_head_changed() if not (self.repo_root / ".git").exists(): return False if not self._build_lock.acquire(blocking=False): @@ -242,7 +261,7 @@ def search( return SearchReadResult( matches=[], total_tokens=0, tokens_saved_vs_naive=0, cache_hit=False, backend="zoekt" ) - rel_glob = _path_to_glob(self.repo_root, Path(search_path).resolve()) + rel_glob = _path_to_glob(self.checkout_root, Path(search_path).resolve()) raw_limit = max(max_files * 4, max_files, 20) raw_matches = client.search(query, num_matches=raw_limit, file_glob=rel_glob) reranked = _rank_zoekt_file_results( @@ -272,7 +291,9 @@ def search( naive_tokens = 0 for _score, file_match in selected: rel_path = _normalize_zoekt_path(file_match.path) - abs_path = self.repo_root / rel_path + abs_path = self.checkout_root / rel_path + if self.checkout_root != self.repo_root and not abs_path.is_file(): + continue # in the served checkout only; the worktree deleted or never had it lang = _detect_lang(rel_path) raw_line_text = "\n".join(match.line_text for match in file_match.matches if match.line_text) naive_tokens += _count_tokens(rel_path) + _count_tokens(raw_line_text) @@ -416,14 +437,35 @@ def get_zoekt_supervisor(repo_root: str | Path) -> ZoektSupervisor: with _SUPERVISORS_LOCK: supervisor = _SUPERVISORS.get(root) if supervisor is None: - supervisor = ZoektSupervisor(root) + served = _ROOT_OVERRIDES.get(root) + supervisor = ZoektSupervisor(root) if served is None else ZoektSupervisor(served, checkout_root=root) _SUPERVISORS[root] = supervisor return supervisor +def set_zoekt_root_override(checkout_root: str | Path, served_root: str | Path) -> None: + """Serve *checkout_root*'s Zoekt searches from *served_root*'s index and server.""" + checkout = str(Path(checkout_root).resolve()) + served = Path(served_root).resolve() + with _SUPERVISORS_LOCK: + if _ROOT_OVERRIDES.get(checkout) == served: + return + _ROOT_OVERRIDES[checkout] = served + _SUPERVISORS.pop(checkout, None) + + +def clear_zoekt_root_override(checkout_root: str | Path) -> None: + """Undo :func:`set_zoekt_root_override` for *checkout_root*.""" + checkout = str(Path(checkout_root).resolve()) + with _SUPERVISORS_LOCK: + if _ROOT_OVERRIDES.pop(checkout, None) is not None: + _SUPERVISORS.pop(checkout, None) + + def reset_zoekt_supervisors() -> None: with _SUPERVISORS_LOCK: _SUPERVISORS.clear() + _ROOT_OVERRIDES.clear() reset_zoekt_servers() @@ -440,6 +482,8 @@ def _path_to_glob(repo_root: Path, search_path: Path) -> str | None: __all__ = [ "ZoektBackendHealth", "ZoektSupervisor", + "clear_zoekt_root_override", "get_zoekt_supervisor", "reset_zoekt_supervisors", + "set_zoekt_root_override", ] diff --git a/src/lemoncrow/infra/code_intel/zoekt/server.py b/src/lemoncrow/infra/code_intel/zoekt/server.py index e35596959..f2be59296 100644 --- a/src/lemoncrow/infra/code_intel/zoekt/server.py +++ b/src/lemoncrow/infra/code_intel/zoekt/server.py @@ -713,6 +713,34 @@ def indexed_git_head(self) -> str | None: return None +def _git_dirs(repo_root: Path) -> tuple[Path, Path] | None: + """``(gitdir, commondir)`` for a checkout: both ``.git`` in a normal one. + + A linked worktree's ``.git`` is a file naming its own admin directory, which + holds its HEAD; branch refs live in the shared directory ``commondir`` names. + """ + dot_git = repo_root / ".git" + if dot_git.is_dir(): + return dot_git, dot_git + try: + text = dot_git.read_text(encoding="utf-8").strip() + except OSError: + return None + if not text.startswith("gitdir:"): + return None + gitdir = Path(text.split(":", 1)[1].strip()) + if not gitdir.is_absolute(): + gitdir = repo_root / gitdir + common = gitdir + try: + raw = (gitdir / "commondir").read_text(encoding="utf-8").strip() + except OSError: + raw = "" + if raw: + common = Path(raw) if Path(raw).is_absolute() else gitdir / raw + return gitdir, common + + def _read_git_head(repo_root: Path) -> str | None: """Resolve a repo's git HEAD to a commit sha via cheap file reads. @@ -720,17 +748,22 @@ def _read_git_head(repo_root: Path) -> str | None: the symbolic ref string when the ref is packed/unborn -- still a stable change-detection key. None when the path is not a git repo. """ - head_file = repo_root / ".git" / "HEAD" + dirs = _git_dirs(repo_root) + if dirs is None: + return None + gitdir, common = dirs try: - ref = head_file.read_text(encoding="utf-8").strip() + ref = (gitdir / "HEAD").read_text(encoding="utf-8").strip() except OSError: return None if ref.startswith("ref: "): - ref_path = repo_root / ".git" / ref[5:] - try: - return ref_path.read_text(encoding="utf-8").strip() - except OSError: - return ref + name = ref[5:] + for base in (gitdir, common): + try: + return (base / name).read_text(encoding="utf-8").strip() + except OSError: + continue + return ref return ref diff --git a/src/lemoncrow/pro/capabilities/code_context/engine.py b/src/lemoncrow/pro/capabilities/code_context/engine.py index f72401c75..5379d9bcd 100644 --- a/src/lemoncrow/pro/capabilities/code_context/engine.py +++ b/src/lemoncrow/pro/capabilities/code_context/engine.py @@ -45,6 +45,8 @@ PatternRewriteResult, PatternSearchResult, ) +from lemoncrow.infra.code_intel.store import CODE_INDEXER_SEMANTICS_VERSION as _CODE_INDEXER_SEMANTICS_VERSION +from lemoncrow.infra.code_intel.worktree_seed import is_seeded_index, resolve_repo_id from lemoncrow.infra.tree_sitter.tags import Tag, detect_language, extract_tags from lemoncrow.pro.capabilities.code_context.ann_symbol_index import ( SymbolAnnIndex, @@ -404,10 +406,6 @@ def close(self) -> None: ] _SEARCH_COMPACT_DEFAULT_KEYS = set([*_SEARCH_ESSENTIAL_KEYS, "score", "commit_sha"]) _LINEAGE_INDEX_VERSION = 2 -# Bump when source selection or symbol/text extraction semantics change in a way -# an incremental mtime/hash check cannot see for unchanged files. -# 3: FTS5 rows carry explicit rowids derived from files.rowid / symbols.rowid. -_CODE_INDEXER_SEMANTICS_VERSION = 3 _LINEAGE_DEFAULT_SCORE_PENALTY = 0.1 # --- File-watcher constants --- @@ -3715,8 +3713,9 @@ def __init__( autosync_enabled: bool | None = None, ) -> None: self.repo_root = Path(repo_root).resolve() - self.repo_id = _repo_id(self.repo_root) self.db_path = Path(db_path).resolve() if db_path is not None else _default_db_path(self.repo_root) + # A seeded worktree index answers under its main checkout's id (worktree_seed). + self.repo_id = resolve_repo_id(self.db_path, _repo_id(self.repo_root)) self._db_lock = _shared_db_lock(self.db_path) self._schema_ready = False self._cache = RetrievalCache(self.db_path) @@ -4388,6 +4387,16 @@ def _index_repo_unsafe( self._init_schema(conn) if not force and not self._rowid_scheme_trustworthy(conn): force = True + seeded = is_seeded_index(conn) + if force and seeded: + # A rebuild would rewrite every page this clone shares with the main + # checkout's index. The engine factory re-seeds it instead. + logger.warning( + "context_engine: %s holds a seeded worktree index; not rebuilding it -- it needs a re-seed " + "from the main checkout's index", + self.db_path, + ) + return self._current_index_stats() if force: # --- Full rebuild: wipe everything, then parallel-extract + batch-write --- @@ -4580,7 +4589,7 @@ def _index_repo_unsafe( symbols_indexed = sum(len(r.symbols) for r in results) imports_indexed = sum(len(r.imports) for r in results) - if force: + if force and not seeded: # Compact all DBs after a full rebuild — DELETE + re-insert leaves free pages # that inflate file size until VACUUMed (e.g. 87 MB → 31 MB for pylint). for _vac_db in (self.db_path, self.intel_db_path, self.vectors_db_path, self.fts_db_path): @@ -15168,6 +15177,11 @@ def _autosync_tick(self, now_ms: int) -> None: watcher is alive: it already reindexes the files any change touches, a checkout included, so checking HEAD too would reindex a checkout twice. """ + if not self.repo_root.is_dir(): + # A removed git worktree. Stop before anything connects: opening the + # index would recreate its store inside the deleted checkout. + self._stop_autosync_worker() + return if not self.index_ready(): # Still empty (e.g. the initial build lost an index-lock race # with a concurrent prewarm). Keep retrying until it exists. diff --git a/tests/infra/code_intel/test_worktree_seed.py b/tests/infra/code_intel/test_worktree_seed.py new file mode 100644 index 000000000..2a3a11158 --- /dev/null +++ b/tests/infra/code_intel/test_worktree_seed.py @@ -0,0 +1,431 @@ +"""A linked worktree's code index is seeded from its main checkout's (ISS-10812, PLN-2070 PR 1). + +Each test builds a real git repository with a real linked worktree, indexes the +main checkout with the real engine, and opens the worktree through the daemon's +engine factory -- the path every code tool takes. +""" + +from __future__ import annotations + +import os +import shutil +import sqlite3 +import subprocess +import threading +from collections.abc import Iterator +from pathlib import Path +from typing import Any + +import pytest + +from lemoncrow.gateway.adapters import mcp_server +from lemoncrow.infra.code_intel import worktree_seed +from lemoncrow.infra.code_intel.freshness import IndexRebuilding, VersionedEngineCache +from lemoncrow.infra.code_intel.store import CODE_CONTEXT_DB, workspace_dir +from lemoncrow.infra.code_intel.zoekt import adapter as zoekt_adapter +from lemoncrow.infra.code_intel.zoekt.server import ZoektServer, _read_git_head +from lemoncrow.pro.capabilities.code_context import CodeContextEngine + +try: + import fcntl +except ImportError: # pragma: no cover - non-POSIX platforms + fcntl = None # type: ignore[assignment] + +_GIT = ["git", "-c", "user.email=seed@test", "-c", "user.name=seed", "-c", "commit.gpgsign=false"] + + +class _Clock: + def __init__(self) -> None: + self.now = 1000.0 + + def __call__(self) -> float: + return self.now + + +@pytest.fixture +def clock() -> _Clock: + return _Clock() + + +@pytest.fixture(autouse=True) +def _isolated(monkeypatch: pytest.MonkeyPatch, clock: _Clock) -> Iterator[None]: + monkeypatch.setenv("LEMONCROW_CODE_AUTOSYNC", "0") + monkeypatch.setenv("LEMONCROW_CODE_FILE_WATCHER", "0") + monkeypatch.delenv(worktree_seed.WORKTREE_ENGINE_IDLE_ENV, raising=False) + monkeypatch.setattr(worktree_seed, "_worktrees", {}) + monkeypatch.setattr(worktree_seed, "_checked_at", {}) + monkeypatch.setattr(zoekt_adapter, "_ROOT_OVERRIDES", {}) + monkeypatch.setattr(zoekt_adapter, "_SUPERVISORS", {}) + cache = VersionedEngineCache( + "test", + recheck_seconds=0.0, + clock=clock, + on_evict=mcp_server._retire_code_engine, + retire=worktree_seed.retire_worktree_engine, + ) + monkeypatch.setattr(mcp_server, "_code_engine_cache", cache) + monkeypatch.setattr(mcp_server, "_scoped_context_cache", {}) + yield + cache.stop_sweeper() + + +def _git(cwd: Path, *args: str) -> str: + return subprocess.run([*_GIT, *args], cwd=cwd, check=True, capture_output=True, text=True).stdout.strip() + + +@pytest.fixture +def repos(tmp_path: Path) -> tuple[Path, Path]: + """(main, worktree): main indexed at the current format, worktree freshly added.""" + main = (tmp_path / "main").resolve() + (main / "pkg").mkdir(parents=True) + for i in range(6): + (main / "pkg" / f"mod{i}.py").write_text(f"def alpha_{i}():\n return {i}\n", encoding="utf-8") + (main / ".gitignore").write_text(".lemoncrow/\n", encoding="utf-8") + _git(main, "init", "-q", "-b", "main") + _git(main, "add", "-A") + _git(main, "commit", "-q", "-m", "init") + worktree = (tmp_path / "wt").resolve() + _git(main, "worktree", "add", "-q", "-b", "feature", str(worktree)) + CodeContextEngine(main, autosync_enabled=False).index_repo(force=True) + return main, worktree + + +def _db(root: Path) -> Path: + return workspace_dir(root) / CODE_CONTEXT_DB + + +def _state(root: Path) -> dict[str, str]: + conn = sqlite3.connect(_db(root)) + try: + return {str(k): str(v) for k, v in conn.execute("SELECT key, value FROM engine_state")} + finally: + conn.close() + + +def _file_count(root: Path, repo_id: str | None = None) -> int: + conn = sqlite3.connect(_db(root)) + try: + if repo_id is None: + return int(conn.execute("SELECT COUNT(*) FROM files").fetchone()[0]) + return int(conn.execute("SELECT COUNT(*) FROM files WHERE repo_id = ?", (repo_id,)).fetchone()[0]) + finally: + conn.close() + + +def _names(engine: Any, query: str) -> set[str]: + return {s.symbol_name for s in engine.search_symbols(query, limit=50, auto_index=False)} + + +def _open(root: Path) -> Any: + return mcp_server._code_context_engine(str(root)) + + +@pytest.fixture +def extractions(monkeypatch: pytest.MonkeyPatch) -> list[str]: + """Relative paths the engine re-extracts, in order.""" + seen: list[str] = [] + original = CodeContextEngine._parallel_extract + + def spy(self: Any, paths: list[Path], *args: Any, **kwargs: Any) -> Any: + seen.extend(os.path.relpath(p, self.repo_root) for p in paths) + return original(self, paths, *args, **kwargs) + + monkeypatch.setattr(CodeContextEngine, "_parallel_extract", spy) + return seen + + +def test_opening_a_worktree_engine_seeds_it_by_clone_under_mains_lock( + repos: tuple[Path, Path], monkeypatch: pytest.MonkeyPatch, extractions: list[str] +) -> None: + main, worktree = repos + main_id = worktree_seed.path_repo_id(main) + main_store = workspace_dir(main) + lock = main_store / (CODE_CONTEXT_DB + ".indexlock") + clones: list[tuple[str, bool]] = [] + real_clone = worktree_seed._clone_file + + def spy_clone(src: Path, dst: Path) -> None: + clones.append((src.name, worktree_seed._lock_held(lock))) + real_clone(src, dst) + + monkeypatch.setattr(worktree_seed, "_clone_file", spy_clone) + # A reader pinned on an old snapshot keeps the next commit's frames out of the + # database file, so the seed has to carry main's WAL to include it. + reader = sqlite3.connect(main_store / CODE_CONTEXT_DB, isolation_level=None) + reader.execute("BEGIN") + reader.execute("SELECT COUNT(*) FROM files").fetchone() + writer = sqlite3.connect(main_store / CODE_CONTEXT_DB) + writer.execute("INSERT INTO engine_state(key, value) VALUES ('probe', 'after-the-reader')") + writer.commit() + writer.close() + try: + assert (main_store / (CODE_CONTEXT_DB + "-shm")).exists() + engine = _open(worktree) + finally: + reader.execute("ROLLBACK") + reader.close() + + assert extractions == [], "opening a worktree ran an index build" + state = _state(worktree) + assert state[f"repo_id_alias:{worktree_seed.path_repo_id(worktree)}"] == main_id + assert state["seeded_from"].startswith(f"{main}@") + assert state["probe"] == "after-the-reader", "the seed lost a commit still in main's WAL" + assert engine.repo_id == main_id + assert _file_count(worktree) == _file_count(main, main_id) == 6 + assert "alpha_3" in _names(engine, "alpha_3") + assert clones and all(held for _name, held in clones), clones + assert (CODE_CONTEXT_DB + "-wal") in {name for name, _held in clones} + assert not [name for name, _held in clones if name.endswith("-shm")] + + +def test_the_first_refresh_reextracts_only_what_differs_from_main( + repos: tuple[Path, Path], extractions: list[str] +) -> None: + main, worktree = repos + engine = _open(worktree) + assert "omega_wt" not in _names(engine, "omega_wt") + (worktree / "pkg" / "mod1.py").write_text("def beta_1():\n return 1\n", encoding="utf-8") + (worktree / "pkg" / "only_here.py").write_text("def omega_wt():\n return 9\n", encoding="utf-8") + + engine.index_repo(force=False) + + assert sorted(extractions) == ["pkg/mod1.py", "pkg/only_here.py"] + assert "omega_wt" in _names(engine, "omega_wt") + assert "alpha_1" not in _names(engine, "alpha_1") + conn = sqlite3.connect(_db(worktree)) + try: + stored = conn.execute("SELECT mtime_ns FROM files WHERE file_path = 'pkg/mod0.py'").fetchone()[0] + finally: + conn.close() + assert stored == (worktree / "pkg" / "mod0.py").stat().st_mtime_ns, "an identical file kept main's mtime" + assert "omega_wt" not in _names(_open(main), "omega_wt"), "the worktree's refresh wrote main's index" + + +@pytest.mark.parametrize("stamped", [False, True], ids=["edit-reindex", "format-stamped"]) +def test_a_partial_worktree_index_is_replaced_by_a_seed(repos: tuple[Path, Path], stamped: bool) -> None: + """Today's edit-triggered indexes carry no format stamp; one built by a current indexer does.""" + _main, worktree = repos + partial = CodeContextEngine(worktree, autosync_enabled=False) + if stamped: + partial.index_repo(force=True, include_globs=["pkg/mod0.py"]) + assert "indexer_semantics_version" in _state(worktree) + else: + partial._reindex_files([str(worktree / "pkg" / "mod0.py")]) + assert _file_count(worktree) == 1 and "seeded_from" not in _state(worktree) + + engine = _open(worktree) + + assert "seeded_from" in _state(worktree) + assert _file_count(worktree) == 6 + assert "alpha_5" in _names(engine, "alpha_5") + + +def test_a_stale_seeded_index_is_reseeded_and_never_rebuilt(repos: tuple[Path, Path], extractions: list[str]) -> None: + main, worktree = repos + _open(worktree) + conn = sqlite3.connect(_db(worktree)) + conn.execute("UPDATE engine_state SET value = '2' WHERE key = 'indexer_semantics_version'") + conn.execute("INSERT INTO engine_state(key, value) VALUES ('stale-marker', 'x')") + conn.commit() + conn.close() + version_before = int(_state(worktree)["index_version"]) + + stale = CodeContextEngine(worktree, autosync_enabled=False) + stale.index_repo(force=False) # the stale format forces a rebuild -- refused for a seed + stale.index_repo(force=True) + assert extractions == [], "a seeded index was rebuilt" + assert int(_state(worktree)["index_version"]) == version_before + + mcp_server._code_engine_cache.clear() + engine = _open(worktree) + + state = _state(worktree) + assert "stale-marker" not in state, "the stale index was not replaced" + assert state["indexer_semantics_version"] == _state(main)["indexer_semantics_version"] + assert extractions == [] + assert "alpha_2" in _names(engine, "alpha_2") + + +@pytest.mark.skipif(fcntl is None, reason="flock is POSIX-only") +def test_mains_lock_held_reports_seeding_then_the_next_call_seeds( + repos: tuple[Path, Path], monkeypatch: pytest.MonkeyPatch +) -> None: + main, worktree = repos + monkeypatch.setattr(worktree_seed, "SEED_LOCK_WAIT_S", 0.1) + fd = os.open(workspace_dir(main) / (CODE_CONTEXT_DB + ".indexlock"), os.O_RDWR | os.O_CREAT, 0o644) + try: + fcntl.flock(fd, fcntl.LOCK_EX) + with pytest.raises(IndexRebuilding, match="seeding the worktree index"): + _open(worktree) + finally: + fcntl.flock(fd, fcntl.LOCK_UN) + os.close(fd) + + engine = _open(worktree) + assert "seeded_from" in _state(worktree) + assert "alpha_0" in _names(engine, "alpha_0") + + +def test_concurrent_first_opens_of_a_worktree_seed_it_once( + repos: tuple[Path, Path], monkeypatch: pytest.MonkeyPatch +) -> None: + """The daemon's warm-up and a first request, or two parallel tool calls, open one worktree at once.""" + _main, worktree = repos + real_seed = worktree_seed.seed_worktree_index + calls: list[Path] = [] + errors: list[Exception] = [] + overlapped = threading.Event() + + def open_worktree() -> None: + try: + worktree_seed.ensure_seeded(worktree) + except Exception as exc: + errors.append(exc) + + rival = threading.Thread(target=open_worktree) + + def spy_seed(root: Path, main: Path, **kwargs: Any) -> worktree_seed.SeedResult: + calls.append(root) + if len(calls) == 1: + rival.start() + overlapped.wait(timeout=1.0) # set only when the rival reaches a seed of its own + else: + overlapped.set() + return real_seed(root, main, **kwargs) + + monkeypatch.setattr(worktree_seed, "seed_worktree_index", spy_seed) + open_worktree() + rival.join(timeout=60) + + assert calls == [worktree], "both openers seeded the worktree" + assert errors == [] + assert "seeded_from" in _state(worktree) + + +def test_an_alias_in_a_shared_database_leaves_the_other_repo_alone(repos: tuple[Path, Path], tmp_path: Path) -> None: + main, worktree = repos + other = (tmp_path / "other").resolve() + other.mkdir() + (other / "lib.py").write_text("def gamma():\n return 0\n", encoding="utf-8") + _git(other, "init", "-q") + shared = _db(main) + other_id = worktree_seed.path_repo_id(other) + CodeContextEngine(other, db_path=shared, autosync_enabled=False).index_repo(force=False) + other_rows = _file_count(main, other_id) + assert other_rows == 1 + + engine = _open(worktree) + + assert engine.repo_id == worktree_seed.path_repo_id(main) + assert CodeContextEngine(other, db_path=shared, autosync_enabled=False).repo_id == other_id + assert CodeContextEngine(other, db_path=_db(worktree), autosync_enabled=False).repo_id == other_id + assert _file_count(main, other_id) == other_rows + assert _file_count(worktree, other_id) == other_rows + + +def test_a_removed_worktrees_engine_retires_within_one_tick(repos: tuple[Path, Path]) -> None: + main, worktree = repos + main_engine = _open(main) + engine = _open(worktree) + mcp_server._scoped_context_capability(str(worktree)) + assert str(worktree) in mcp_server._scoped_context_cache + shutil.rmtree(worktree) + + engine._autosync_tick(0) + assert engine._autosync_stop.is_set(), "the engine's own tick kept running on a removed worktree" + assert not (worktree / ".lemoncrow").exists(), "the tick recreated the removed worktree's store" + # A reindex that was mid-write when the worktree went writes its store back. + (worktree / ".lemoncrow" / "workspace").mkdir(parents=True) + + retired = mcp_server._code_engine_cache.sweep() + assert retired == [str(worktree)] + assert str(worktree) not in mcp_server._code_engine_cache + assert str(worktree) not in mcp_server._scoped_context_cache, "a scoped capability pinned the retired engine" + assert str(main) in mcp_server._code_engine_cache + assert not main_engine._autosync_stop.is_set() + assert zoekt_adapter._ROOT_OVERRIDES == {} + + +def test_an_idle_worktree_engine_retires_and_main_never_does( + repos: tuple[Path, Path], monkeypatch: pytest.MonkeyPatch, clock: _Clock +) -> None: + main, worktree = repos + monkeypatch.setenv(worktree_seed.WORKTREE_ENGINE_IDLE_ENV, "600") + _open(main) + engine = _open(worktree) + + clock.now += 599 + assert mcp_server._code_engine_cache.sweep() == [] + clock.now += 2 + assert mcp_server._code_engine_cache.sweep() == [str(worktree)] + assert engine._autosync_stop.is_set() + clock.now += 10_000 + assert mcp_server._code_engine_cache.sweep() == [] + assert str(main) in mcp_server._code_engine_cache + + +def test_a_seeded_engine_takes_zoekt_from_main_and_reads_its_own_head(repos: tuple[Path, Path]) -> None: + main, worktree = repos + _open(worktree) + + supervisor = zoekt_adapter.get_zoekt_supervisor(worktree) + assert supervisor.repo_root == main + assert supervisor.checkout_root == worktree + assert supervisor._served_path(worktree / "pkg") == main / "pkg" + + (worktree / "pkg" / "new.py").write_text("x = 1\n", encoding="utf-8") + _git(worktree, "add", "-A") + _git(worktree, "commit", "-q", "-m", "wt") + head = _git(worktree, "rev-parse", "HEAD") + assert head != _git(main, "rev-parse", "HEAD") + assert _read_git_head(worktree) == head + assert ZoektServer(worktree).current_git_head() == head + + +def _db_file_alone_has_probe(db: Path, tmp: Path) -> bool: + """Whether the database file, without its WAL, already holds the probe row.""" + tmp.mkdir(exist_ok=True) + shutil.copyfile(db, tmp / db.name) + conn = sqlite3.connect(tmp / db.name) + try: + return conn.execute("SELECT 1 FROM engine_state WHERE key = 'probe'").fetchone() is not None + except sqlite3.DatabaseError: + return False # a partly checkpointed file is no snapshot at all without its WAL + finally: + conn.close() + shutil.rmtree(tmp) + + +def test_checkpoint_index_folds_the_wal_a_reader_held_back(repos: tuple[Path, Path], tmp_path: Path) -> None: + """After a reindex the CLI folds main's WAL in, so a later seed finds it small.""" + main, _worktree = repos + db = _db(main) + keeper = sqlite3.connect(db, isolation_level=None) # keeps the WAL from being deleted on close + keeper.execute("BEGIN") + keeper.execute("SELECT COUNT(*) FROM files").fetchone() + writer = sqlite3.connect(db) + writer.execute("INSERT INTO engine_state(key, value) VALUES ('probe', '1')") + writer.commit() + writer.close() + try: + assert worktree_seed.checkpoint_index(workspace_dir(main), attempts=1) is False + assert not _db_file_alone_has_probe(db, tmp_path / "before") + keeper.execute("ROLLBACK") + + assert worktree_seed.checkpoint_index(workspace_dir(main)) is True + assert _db_file_alone_has_probe(db, tmp_path / "after"), "the WAL was left for the seed to carry" + finally: + keeper.close() + + +def test_the_main_checkout_engine_is_unchanged(repos: tuple[Path, Path]) -> None: + main, _worktree = repos + before = _state(main) + + engine = _open(main) + + assert engine.repo_id == worktree_seed.path_repo_id(main) + assert _state(main) == before + assert not [key for key in before if key.startswith("repo_id_alias:") or key == "seeded_from"] + assert zoekt_adapter.get_zoekt_supervisor(main).checkout_root == main + assert not worktree_seed.retire_worktree_engine(str(main), float("inf")), "main was tracked as a worktree"