From 0fd68e7479d4bb460d9dc05380df5a31529e45c9 Mon Sep 17 00:00:00 2001 From: ec4t3rina Date: Fri, 3 Apr 2026 02:25:18 +0300 Subject: [PATCH] feat: complete V1 dynamic memory mesh with IPC and health monitor --- .claude/worktrees/gallant-bardeen | 1 + .claude/worktrees/keen-curran | 1 + QUICKSTART.md | 129 ++++- libucks/_cli.py | 83 +++ libucks/diff/diff_extractor.py | 49 +- libucks/embeddings/embedding_service.py | 4 +- libucks/git_hook_receiver.py | 109 ++++ libucks/health_monitor.py | 136 +++++ libucks/init_orchestrator.py | 27 +- libucks/librarian.py | 46 ++ libucks/mcp_bridge.py | 163 +++++- libucks/merging_service.py | 214 +++++++ libucks/models/bucket.py | 8 +- libucks/models/chunk.py | 3 + libucks/query_orchestrator.py | 25 +- libucks/stale_checker.py | 254 +++++++++ libucks/startup_recovery.py | 206 +++++++ libucks/storage/bucket_registry.py | 55 +- libucks/storage/bucket_store.py | 39 +- main.py | 35 +- pyproject.toml | 2 +- scripts/check_nervous_system.py | 296 ++++++++++ tests/integration/test_health_and_merge.py | 262 +++++++++ tests/unit/test_bucket_registry.py | 95 ++++ tests/unit/test_git_hook_receiver.py | 240 ++++++++ tests/unit/test_models.py | 53 +- tests/unit/test_stale_checker.py | 629 +++++++++++++++++++++ tests/unit/test_startup_recovery.py | 478 ++++++++++++++++ 28 files changed, 3544 insertions(+), 98 deletions(-) create mode 160000 .claude/worktrees/gallant-bardeen create mode 160000 .claude/worktrees/keen-curran create mode 100644 libucks/_cli.py create mode 100644 libucks/git_hook_receiver.py create mode 100644 libucks/health_monitor.py create mode 100644 libucks/merging_service.py create mode 100644 libucks/stale_checker.py create mode 100644 libucks/startup_recovery.py create mode 100644 scripts/check_nervous_system.py create mode 100644 tests/integration/test_health_and_merge.py create mode 100644 tests/unit/test_git_hook_receiver.py create mode 100644 tests/unit/test_stale_checker.py create mode 100644 tests/unit/test_startup_recovery.py diff --git a/.claude/worktrees/gallant-bardeen b/.claude/worktrees/gallant-bardeen new file mode 160000 index 0000000..2c48827 --- /dev/null +++ b/.claude/worktrees/gallant-bardeen @@ -0,0 +1 @@ +Subproject commit 2c48827d66c4a6e14b0090b2b5fef2c899c64694 diff --git a/.claude/worktrees/keen-curran b/.claude/worktrees/keen-curran new file mode 160000 index 0000000..2c48827 --- /dev/null +++ b/.claude/worktrees/keen-curran @@ -0,0 +1 @@ +Subproject commit 2c48827d66c4a6e14b0090b2b5fef2c899c64694 diff --git a/QUICKSTART.md b/QUICKSTART.md index 6e07e43..53f7610 100644 --- a/QUICKSTART.md +++ b/QUICKSTART.md @@ -1,58 +1,149 @@ -# libucks — Quickstart +# libucks — Quickstart (macOS, /Users/ecaterina/Developer/libucks) -## 1. Install +## 1. Create & activate the virtual environment ```bash -pip install -e ".[dev]" +cd /Users/ecaterina/Developer/libucks +python3 -m venv .venv +source .venv/bin/activate ``` -## 2. Required Environment Variables +## 2. Install the package (editable, with dev deps) ```bash -export ANTHROPIC_API_KEY="sk-ant-..." +pip install -e ".[dev]" ``` -That is the only secret required. `TextStrategy.from_env()` reads it via the `anthropic` SDK default. +This installs the `libucks` console script and makes `from libucks.xxx import yyy` +importable from anywhere on this Python interpreter. -## 3. Init the current repo +## 3. Set your API key ```bash -libucks init --local "$(pwd)" +export ANTHROPIC_API_KEY="sk-ant-..." ``` -## 4. Serve the MCP bridge +`TextStrategy.from_env()` reads it via the `anthropic` SDK default. You only need +this in your shell for manual runs; for Claude Desktop see §5 below. + +## 4. Index a repository ```bash -libucks serve +libucks init --local /Users/ecaterina/Developer/libucks ``` -Starts the MCP server on **stdio**. Register it in your MCP client (e.g. Claude Code `~/.claude/settings.json`): +This walks the repo, embeds every source file, clusters the chunks, and writes +`.libucks/buckets/` and `.libucks/registry.json` into the target directory. + +## 5. Claude Desktop config + +Write the following to +`/Users/ecaterina/Library/Application Support/Claude/claude_desktop_config.json` +(create the file if it does not exist — Claude Desktop will not overwrite it on +launch if it is already present): ```json { "mcpServers": { "libucks": { - "command": "libucks", - "args": ["serve"] + "command": "/Users/ecaterina/Developer/libucks/.venv/bin/python", + "args": ["/Users/ecaterina/Developer/libucks/main.py", "serve"], + "env": { + "ANTHROPIC_API_KEY": "sk-ant-YOUR-KEY-HERE", + "PYTHONPATH": "/Users/ecaterina/Developer/libucks" + } } } } ``` -## 5. Init a different local repo +**Important notes:** +- Use absolute paths — Claude Desktop does not inherit your shell environment. +- `ANTHROPIC_API_KEY` **must** be in `"env"` for the same reason. +- `PYTHONPATH` is a belt-and-suspenders fallback; the venv Python is the primary + mechanism. +- After saving, **quit and relaunch** Claude Desktop for the config to take effect. + +## 6. Verify the server starts + +From a fresh terminal (no venv active, any cwd): + +```bash +/Users/ecaterina/Developer/libucks/.venv/bin/python \ + /Users/ecaterina/Developer/libucks/main.py serve +``` + +It should block silently, waiting for MCP stdin. Press `Ctrl-C` to stop. +No "Loading weights" or other text should appear on stdout — those messages are +redirected to stderr. + +## 7. Index a different repo ```bash libucks init --local /absolute/path/to/other/repo ``` -Then run `libucks serve` from inside that repo (it reads `.libucks/` relative to cwd). +Then update the `"args"` in `claude_desktop_config.json` to point `serve` at the +repo that contains `.libucks/` (the server reads `.libucks/` relative to cwd, or +from `paths.repo_root` in `.libucks/config.toml`). -## 6. PYTHONPATH note +--- -**Not required** if installed with `pip install -e .` — the `libucks` console script is on `PATH` and the package is importable. +## Phase 6: Production-Grade Dynamic Engine -If running the entry point directly (e.g. `python main.py serve`), set: +### Step 1 — Start the server ```bash -PYTHONPATH=. python main.py serve +libucks serve ``` + +This starts the MCP server (stdio) **and** automatically starts two background +tasks inside the same process: + +| Background task | What it does | Frequency | +|---|---|---| +| `GitHookReceiver` | Listens on `.libucks/server.sock` for git events | Always-on | +| `HealthMonitor` | Splits overflowing buckets, merges redundant ones | Every 5 min | + +### Step 2 — Install git hooks in a repo + +Run once per repository you want libucks to track: + +```bash +cd /path/to/your/repo +libucks install-hooks +``` + +This **appends** (never overwrites) three trigger lines to `.git/hooks/`: + +``` +post-commit → libucks hook post-commit "$@" || true +post-checkout → libucks hook post-checkout "$@" || true +post-rewrite → libucks hook post-rewrite "$@" || true +``` + +After a `git commit` or `git checkout`, the hook sends a JSON event over the +Unix socket. The server replays any missed diffs and updates `last_indexed_head`. + +### Step 3 — Verify the engine is running + +Use the `libucks_status` MCP tool from Claude Desktop, or call it directly: + +```bash +echo '{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"libucks_status","arguments":{}}}' \ + | libucks serve 2>/dev/null +``` + +The JSON response includes `bucket_count` and `total_tokens`. A healthy system +shows the bucket count stabilising over time as the HealthMonitor splits and +merges. + +**Background watcher confirmation** — check the server stderr log for: + +``` +[libucks] git_hook_receiver.listening sock=.libucks/server.sock +[libucks] health_monitor.started interval=300 +``` + +If you see both lines, all three layers (JIT staleness, git hooks, health monitor) +are active. diff --git a/libucks/_cli.py b/libucks/_cli.py new file mode 100644 index 0000000..0b4348b --- /dev/null +++ b/libucks/_cli.py @@ -0,0 +1,83 @@ +"""CLI entry point — lives inside the package so the console script works from any directory.""" +import asyncio +import json +import socket +import subprocess +from pathlib import Path + +import click + + +def _find_repo_root() -> Path: + """Return the git repo root for cwd, or cwd itself if not in a repo.""" + try: + result = subprocess.run( + ["git", "rev-parse", "--show-toplevel"], + capture_output=True, + text=True, + timeout=5, + ) + if result.returncode == 0: + return Path(result.stdout.strip()) + except Exception: + pass + return Path.cwd() + + +@click.group() +@click.version_option(version="0.1.0", prog_name="libucks") +def cli(): + """libucks — Librarian Buckets, local AI memory server for coding agents.""" + + +@cli.command("init") +@click.option("--local", "local_path", type=click.Path(exists=True, file_okay=False, path_type=Path), + required=True, help="Path to a local repository to index.") +def init_cmd(local_path: Path): + """Seed libucks buckets from a local repository.""" + from libucks.init_orchestrator import InitOrchestrator + + orchestrator = InitOrchestrator(local_path) + asyncio.run(orchestrator.run()) + + +@cli.command("serve") +def serve_cmd(): + """Start the libucks MCP server over stdio.""" + from libucks.mcp_bridge import serve + asyncio.run(serve()) + + +@cli.command("install-hooks") +@click.option("--repo", "repo_path", type=click.Path(exists=True, file_okay=False, path_type=Path), + default=None, help="Path to repository (defaults to git repo containing cwd).") +def install_hooks_cmd(repo_path: Path | None): + """Append libucks git hook triggers to .git/hooks/ (never overwrites).""" + from libucks.git_hook_receiver import install_hooks + + target = repo_path or _find_repo_root() + modified = install_hooks(target) + if modified: + click.echo(f"Installed hooks: {', '.join(modified)}") + else: + click.echo("All hooks already installed — nothing changed.") + + +@cli.command("hook") +@click.argument("event") +@click.argument("args", nargs=-1) +def hook_cmd(event: str, args: tuple): + """Send a git hook event to the running libucks server (called by git hooks).""" + repo_path = _find_repo_root() + sock_path = repo_path / ".libucks" / "server.sock" + if not sock_path.exists(): + return # server not running — silent exit so git is never blocked + + payload = json.dumps({"event": event, "args": list(args)}).encode() + try: + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as s: + s.settimeout(3) + s.connect(str(sock_path)) + s.sendall(payload) + except Exception: + pass # never block git diff --git a/libucks/diff/diff_extractor.py b/libucks/diff/diff_extractor.py index b6e42c1..e2f542e 100644 --- a/libucks/diff/diff_extractor.py +++ b/libucks/diff/diff_extractor.py @@ -17,22 +17,16 @@ class DiffExtractor: def __init__(self, repo_path: Path) -> None: self._repo = git.Repo(str(repo_path), search_parent_directories=True) - def extract(self, filepath: Path) -> List[DiffEvent]: - """Run ``git diff HEAD -- --find-renames`` and parse into DiffEvents.""" - rel = str(filepath) - try: - diff_text = self._repo.git.diff( - "HEAD", "--find-renames", "--", rel - ) - except git.GitCommandError as exc: - log.warning("diff_extractor.git_error", file=rel, error=str(exc)) - return [] + # ------------------------------------------------------------------ + # Internal + # ------------------------------------------------------------------ + def _parse_diff_output(self, diff_text: str, rel: str) -> List[DiffEvent]: + """Parse raw git diff output into DiffEvents. Shared by extract() and extract_between().""" if not diff_text: log.debug("diff_extractor.no_diff", file=rel) return [] - # Detect binary files — git outputs "Binary files … differ" if "Binary files" in diff_text: log.warning("diff_extractor.binary_skipped", file=rel) return [] @@ -87,3 +81,36 @@ def extract(self, filepath: Path) -> List[DiffEvent]: ) return events + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def extract(self, filepath: Path) -> List[DiffEvent]: + """Run ``git diff HEAD -- --find-renames`` and parse into DiffEvents.""" + rel = str(filepath) + try: + diff_text = self._repo.git.diff("HEAD", "--find-renames", "--", rel) + except git.GitCommandError as exc: + log.warning("diff_extractor.git_error", file=rel, error=str(exc)) + return [] + return self._parse_diff_output(diff_text, rel) + + def extract_between(self, filepath: Path, from_sha: str, to_sha: str) -> List[DiffEvent]: + """Run ``git diff -- `` and parse into DiffEvents. + + Used by StartupRecovery to replay commits that occurred while the server was offline. + """ + rel = str(filepath) + try: + diff_text = self._repo.git.diff(from_sha, to_sha, "--find-renames", "--", rel) + except git.GitCommandError as exc: + log.warning( + "diff_extractor.git_error_between", + file=rel, + from_sha=from_sha[:8], + to_sha=to_sha[:8], + error=str(exc), + ) + return [] + return self._parse_diff_output(diff_text, rel) diff --git a/libucks/embeddings/embedding_service.py b/libucks/embeddings/embedding_service.py index 79744f2..69ff505 100644 --- a/libucks/embeddings/embedding_service.py +++ b/libucks/embeddings/embedding_service.py @@ -62,10 +62,10 @@ def reset(cls) -> None: def embed(self, text: str) -> np.ndarray: """Embed a single string and return a normalised float32 vector.""" - raw = self._model.encode(text, convert_to_numpy=True) + raw = self._model.encode(text, convert_to_numpy=True, show_progress_bar=False) return _l2_normalize(raw) def embed_batch(self, texts: List[str]) -> np.ndarray: """Embed a list of strings and return a (N, D) normalised float32 matrix.""" - raw = self._model.encode(texts, convert_to_numpy=True) + raw = self._model.encode(texts, convert_to_numpy=True, show_progress_bar=False) return _l2_normalize(raw) diff --git a/libucks/git_hook_receiver.py b/libucks/git_hook_receiver.py new file mode 100644 index 0000000..f430138 --- /dev/null +++ b/libucks/git_hook_receiver.py @@ -0,0 +1,109 @@ +"""GitHookReceiver — Unix domain socket listener for git hook events. + +Git hook scripts call `libucks hook "$@" || true` which sends a +single JSON line over the Unix socket at `.libucks/server.sock` and exits. + +Supported payload shapes: + {"event": "post-commit"} + {"event": "post-checkout", "args": ["", "", "1"]} + {"event": "post-rewrite", "args": ["rebase"]} + +The server reads the payload, calls *on_event*, and closes the connection. +Hook scripts never wait for a response — they fire-and-forget. +""" +from __future__ import annotations + +import asyncio +import json +from pathlib import Path +from typing import Awaitable, Callable + +import structlog + +log = structlog.get_logger(__name__) + +# Type alias for the callback injected by mcp_bridge. +OnEventFn = Callable[[dict], Awaitable[None]] + + +async def _handle_connection( + reader: asyncio.StreamReader, + writer: asyncio.StreamWriter, + on_event: OnEventFn, +) -> None: + """Read one JSON payload, dispatch, close.""" + try: + data = await asyncio.wait_for(reader.read(4096), timeout=5.0) + payload: dict = json.loads(data.decode()) + log.info("git_hook_receiver.event", hook_event=payload.get("event")) + await on_event(payload) + except asyncio.TimeoutError: + log.warning("git_hook_receiver.timeout") + except Exception as exc: + log.warning("git_hook_receiver.error", error=str(exc)) + finally: + try: + writer.close() + await writer.wait_closed() + except Exception: + pass + + +async def serve_socket(sock_path: Path, on_event: OnEventFn) -> None: + """Listen on *sock_path* for git hook events indefinitely. + + Removes any stale socket file first so bind always succeeds on restart. + Designed to be launched with ``asyncio.ensure_future()`` from mcp_bridge. + """ + sock_path.unlink(missing_ok=True) + + server = await asyncio.start_unix_server( + lambda r, w: _handle_connection(r, w, on_event), + path=str(sock_path), + ) + log.info("git_hook_receiver.listening", sock=str(sock_path)) + async with server: + await server.serve_forever() + + +# --------------------------------------------------------------------------- +# Hook installer (called by `libucks install-hooks`) +# --------------------------------------------------------------------------- + +_HOOK_EVENTS = ["post-commit", "post-checkout", "post-rewrite"] +_HOOK_LINE = "libucks hook {event} \"$@\" || true" + + +def install_hooks(repo_path: Path) -> list[str]: + """Append libucks trigger lines to .git/hooks/. + + Rules: + - If the hook file does not exist: create it with a ``#!/bin/sh`` shebang. + - If it exists but already contains our trigger: skip (idempotent). + - Always appends — never overwrites existing content. + - Sets executable bit on newly created files. + + Returns the list of hook names that were modified. + """ + hooks_dir = repo_path / ".git" / "hooks" + hooks_dir.mkdir(parents=True, exist_ok=True) + + modified: list[str] = [] + for event in _HOOK_EVENTS: + trigger = _HOOK_LINE.format(event=event) + hook_file = hooks_dir / event + + if hook_file.exists(): + existing = hook_file.read_text() + if trigger in existing: + log.debug("git_hook_receiver.hook_already_installed", hook_event=event) + continue + hook_file.write_text(existing.rstrip("\n") + "\n" + trigger + "\n") + else: + hook_file.write_text(f"#!/bin/sh\n{trigger}\n") + hook_file.chmod(0o755) + + modified.append(event) + log.info("git_hook_receiver.hook_installed", hook_event=event, path=str(hook_file)) + + return modified diff --git a/libucks/health_monitor.py b/libucks/health_monitor.py new file mode 100644 index 0000000..f0b1d8e --- /dev/null +++ b/libucks/health_monitor.py @@ -0,0 +1,136 @@ +"""HealthMonitor — periodic quality guardian for the bucket index. + +Runs every *interval* seconds (default 300 = 5 min). Each pass: + + 1. Size check: token_count >= mitosis_threshold → MitosisService.split() + 2. Coherence check: mean chunk-to-centroid similarity < 0.55 → MitosisService.split() + 3. Merge pass: delegate to MergingService to find and collapse redundant pairs. + +Coherence is the mean cosine similarity of each chunk embedding to the bucket +centroid — cheap to compute (one embed_batch + dot product), no pairwise loop. +""" +from __future__ import annotations + +import asyncio +from pathlib import Path +from typing import TYPE_CHECKING, Optional + +import numpy as np +import structlog + +from libucks.mitosis import _read_chunk_content + +if TYPE_CHECKING: + from libucks.embeddings.embedding_service import EmbeddingService + from libucks.merging_service import MergingService + from libucks.mitosis import MitosisService + from libucks.storage.bucket_registry import BucketRegistry + from libucks.storage.bucket_store import BucketStore + +log = structlog.get_logger(__name__) + +_COHERENCE_THRESHOLD = 0.55 + + +class HealthMonitor: + def __init__( + self, + registry: "BucketRegistry", + store: "BucketStore", + mitosis_service: "MitosisService", + merging_service: "MergingService", + embedder: "EmbeddingService", + mitosis_threshold: int = 20_000, + interval: int = 300, + ) -> None: + self._registry = registry + self._store = store + self._mitosis = mitosis_service + self._merging = merging_service + self._embedder = embedder + self._mitosis_threshold = mitosis_threshold + self._interval = interval + + async def run(self) -> None: + """Loop forever, running a health check every *interval* seconds.""" + log.info("health_monitor.started", interval=self._interval) + while True: + await asyncio.sleep(self._interval) + await self._check() + + async def _check(self) -> None: + """Run one health pass over all registered buckets.""" + bucket_ids = list(self._registry.get_all_centroids().keys()) + log.info("health_monitor.check", bucket_count=len(bucket_ids)) + + for bucket_id in bucket_ids: + try: + if self._registry.is_splitting(bucket_id): + continue + except KeyError: + continue # bucket deregistered mid-pass + + # ---- Size check ------------------------------------------------- + try: + tokens = self._registry.get_token_count(bucket_id) + except KeyError: + continue + if tokens >= self._mitosis_threshold: + log.warning( + "health_monitor.size_trigger", + bucket_id=bucket_id, + tokens=tokens, + threshold=self._mitosis_threshold, + ) + await self._mitosis.split(bucket_id) + continue # bucket is gone; skip coherence + + # ---- Coherence check -------------------------------------------- + score = self._compute_coherence(bucket_id) + if score is not None: + # Store for status reporting + entry = self._registry._buckets.get(bucket_id) + if entry is not None: + entry.coherence_score = score + if score < _COHERENCE_THRESHOLD: + log.warning( + "health_monitor.coherence_trigger", + bucket_id=bucket_id, + coherence=round(score, 3), + ) + await self._mitosis.split(bucket_id) + + # ---- Merge pass ----------------------------------------------------- + await self._merging.run_merge_pass() + + def _compute_coherence(self, bucket_id: str) -> Optional[float]: + """Return mean cosine similarity of chunk embeddings to the bucket centroid. + + Returns None if the bucket cannot be read or has fewer than 2 chunks. + """ + try: + front_matter, _ = self._store.read(bucket_id) + except FileNotFoundError: + return None + + chunks = front_matter.chunks + if len(chunks) < 2: + return 1.0 # a single-chunk bucket is trivially coherent + + contents = [_read_chunk_content(c) for c in chunks] + try: + embeddings = self._embedder.embed_batch(contents) # (N, D) + except Exception as exc: + log.warning("health_monitor.embed_failed", bucket_id=bucket_id, error=str(exc)) + return None + + # L2-normalise rows + norms = np.linalg.norm(embeddings, axis=1, keepdims=True) + embeddings = embeddings / np.maximum(norms, 1e-8) + + centroid = embeddings.mean(axis=0) + c_norm = np.linalg.norm(centroid) + if c_norm > 0: + centroid /= c_norm + + return float(np.mean(embeddings @ centroid)) diff --git a/libucks/init_orchestrator.py b/libucks/init_orchestrator.py index ed3a80e..88de9a1 100644 --- a/libucks/init_orchestrator.py +++ b/libucks/init_orchestrator.py @@ -24,6 +24,23 @@ # Extensions considered source code (subset of grammar registry support). _SOURCE_EXTENSIONS = set(SUPPORTED_LANGUAGES.keys()) +# Additional extensions beyond the grammar registry that carry useful context. +_EXTRA_EXTENSIONS = {".md"} + +# Files larger than this are almost certainly generated blobs or data dumps. +_MAX_FILE_BYTES = 500 * 1_024 # 500 KB + +# Directory names that are pure noise — any path component matching one is skipped. +_NOISE_DIRS = { + "__pycache__", "node_modules", ".venv", "venv", # already skipped, now explicit + "dist", "build", "out", # build artifacts + "assets", "static", "images", "media", # non-code assets + "coverage", "htmlcov", # test coverage reports + ".claude", # shadow-clone worktrees / Claude internals + "docs", # documentation folders (multi-language explosion) + "scripts", # internal build scripts +} + # Approximate tokens per word for rough counting. _TOKENS_PER_CHAR = 0.25 @@ -53,15 +70,19 @@ def _to_chunk_metadata(raw: RawChunk) -> ChunkMetadata: def _collect_source_files(repo_path: Path) -> List[Path]: + allowed_exts = _SOURCE_EXTENSIONS | _EXTRA_EXTENSIONS files: List[Path] = [] for p in sorted(repo_path.rglob("*")): if not p.is_file(): continue - if p.suffix.lower() not in _SOURCE_EXTENSIONS: + if p.suffix.lower() not in allowed_exts: continue - # Skip hidden dirs and common non-source trees. + # Skip hidden dirs and known noise directories. parts = p.relative_to(repo_path).parts - if any(part.startswith(".") or part in ("__pycache__", "node_modules", ".venv", "venv") for part in parts): + if any(part.startswith(".") or part in _NOISE_DIRS for part in parts): + continue + # Skip files that are almost certainly generated blobs or data dumps. + if p.stat().st_size > _MAX_FILE_BYTES: continue files.append(p) return files diff --git a/libucks/librarian.py b/libucks/librarian.py index e464037..3bae23f 100644 --- a/libucks/librarian.py +++ b/libucks/librarian.py @@ -11,6 +11,8 @@ import asyncio import base64 import hashlib +import subprocess +from datetime import datetime, timezone from pathlib import Path from typing import TYPE_CHECKING, List, Optional @@ -48,6 +50,28 @@ def _read_chunk_content(meta: ChunkMetadata) -> str: return "" +def _get_head_sha(repo_path: Optional[Path]) -> str: + """Return the current git HEAD SHA, or 'unknown' if unavailable.""" + if repo_path is None: + return "unknown" + try: + result = subprocess.run( + ["git", "-C", str(repo_path), "rev-parse", "HEAD"], + capture_output=True, + text=True, + timeout=5, + ) + if result.returncode == 0: + return result.stdout.strip() + except Exception: + pass + return "unknown" + + +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + class Librarian: def __init__( self, @@ -58,6 +82,7 @@ def __init__( embedder: Optional["EmbeddingService"] = None, mitosis_threshold: int = 20_000, mitosis_service: Optional["MitosisService"] = None, + repo_path: Optional[Path] = None, ) -> None: self.bucket_id = bucket_id self._store = store @@ -66,6 +91,7 @@ def __init__( self._embedder = embedder self._mitosis_threshold = mitosis_threshold self._mitosis_service = mitosis_service + self._repo_path = repo_path self.queue: asyncio.Queue[object] = asyncio.Queue() # ------------------------------------------------------------------ @@ -138,10 +164,30 @@ async def _handle_update(self, event: UpdateEvent) -> None: await self._registry.register( self.bucket_id, centroid.astype(np.float32), new_token_count ) + + # Phase 6-A: stamp chunks from the updated file with current HEAD SHA + # and an indexed_at timestamp, then persist and save registry. + head_sha = _get_head_sha(self._repo_path) + now = _now_iso() + updated_file = event.hunk.file + for chunk in front_matter.chunks: + # Match on suffix to handle relative vs absolute path variations. + if chunk.source_file == updated_file or chunk.source_file.endswith(updated_file): + chunk.git_sha = head_sha + chunk.indexed_at = now + + front_matter.last_indexed_at = now + front_matter.index_head_sha = head_sha + self._store.write_front_matter(self.bucket_id, front_matter) + + self._registry.update_index_timestamp(self.bucket_id, now, head_sha) + self._registry.save() + log.info( "librarian.update.done", bucket_id=self.bucket_id, token_count=new_token_count, + head_sha=head_sha, ) # Check mitosis outside lock. diff --git a/libucks/mcp_bridge.py b/libucks/mcp_bridge.py index 8e31dc5..923446c 100644 --- a/libucks/mcp_bridge.py +++ b/libucks/mcp_bridge.py @@ -6,6 +6,17 @@ """ from __future__ import annotations +# Must be set before any native extension (tokenizers Rust runtime, ObjC) is +# imported — placing them here, at module load, guarantees that. +import os +os.environ.setdefault("OBJC_DISABLE_INITIALIZE_FORK_SAFETY", "YES") # prevents SIGABRT on Apple Silicon +os.environ.setdefault("TOKENIZERS_PARALLELISM", "false") # prevents HF tokenizer deadlock warning +os.environ.setdefault("TRANSFORMERS_VERBOSITY", "error") # suppresses "BertModel report" etc. +os.environ.setdefault("TQDM_DISABLE", "1") # suppresses "Loading weights" progress bars + +import asyncio +import logging +import sys import tomllib from pathlib import Path from typing import Any @@ -16,9 +27,16 @@ from libucks.central_agent import CentralAgent from libucks.config import Config +from libucks.diff.diff_extractor import DiffExtractor from libucks.embeddings.embedding_service import EmbeddingService +from libucks.git_hook_receiver import serve_socket +from libucks.health_monitor import HealthMonitor from libucks.librarian import Librarian +from libucks.merging_service import MergingService +from libucks.mitosis import MitosisService from libucks.query_orchestrator import QueryOrchestrator +from libucks.stale_checker import StaleChecker +from libucks.startup_recovery import StartupRecovery from libucks.storage.bucket_registry import BucketRegistry from libucks.storage.bucket_store import BucketStore from libucks.thinking.text_strategy import TextStrategy @@ -26,30 +44,56 @@ def _load_repo_path() -> Path: - """Find the target repo path from .libucks/config.toml in cwd, or use cwd.""" - cwd = Path.cwd() - config_file = cwd / ".libucks" / "config.toml" - if config_file.exists(): - with open(config_file, "rb") as fh: - data = tomllib.load(fh) - repo = data.get("paths", {}).get("repo_root", None) - if repo: - return Path(repo).expanduser().resolve() - return cwd + """Return the repository root. + + Resolution order: + 1. LIBUCKS_REPO_PATH env var — set this in claude_desktop_config.json "env" + to point the server at any repo you want. + 2. Project root inferred from __file__ — reliable fallback that is never + the filesystem root, even when Claude Desktop launches with cwd='/'. + """ + env_path = os.environ.get("LIBUCKS_REPO_PATH") + if env_path: + return Path(env_path).expanduser().resolve() + # __file__ = libucks/mcp_bridge.py → .parent = libucks/ → .parent = project root + return Path(__file__).parent.parent.resolve() async def serve() -> None: + # Route ALL logging to stderr — stdout is reserved for MCP JSON-RPC. + logging.basicConfig(stream=sys.stderr, level=logging.INFO, force=True) + + import structlog + structlog.configure( + logger_factory=structlog.PrintLoggerFactory(file=sys.stderr), + ) + + try: + import transformers + transformers.logging.set_verbosity_error() + except Exception: + pass + repo_path = _load_repo_path() cfg = Config.load(repo_path) - registry_path = repo_path / cfg.paths.registry_file - bucket_dir = repo_path / cfg.paths.bucket_dir + bucket_dir = repo_path / ".libucks" + print(f"[libucks] repo={repo_path} registry={registry_path} buckets={bucket_dir}", file=sys.stderr) registry = BucketRegistry(registry_path) registry.load() store = BucketStore(bucket_dir) - embedder = EmbeddingService.get_instance(cfg.model.embedding_model) + + # Pre-load the embedding model BEFORE the MCP stdio server starts. + # Temporarily redirect sys.stdout → sys.stderr so any remaining direct + # prints from model loading never reach the MCP pipe. + _real_stdout = sys.stdout + sys.stdout = sys.stderr + try: + embedder = EmbeddingService.get_instance(cfg.model.embedding_model) + finally: + sys.stdout = _real_stdout # restore BEFORE mcp.server.stdio.stdio_server() captures it strategy = TextStrategy.from_env(cfg.model.anthropic_model) agent = CentralAgent(registry, cfg, embed_fn=embedder.embed) @@ -67,13 +111,104 @@ async def serve() -> None: librarians[bucket_id] = lib agent.register_librarian(bucket_id, lib) + translator = Translator(strategy) + + # ------------------------------------------------------------------ + # Startup recovery: replay commits that arrived while server was offline. + # ------------------------------------------------------------------ + recovery: StartupRecovery | None = None + try: + extractor = DiffExtractor(repo_path) + recovery = StartupRecovery( + repo_path=repo_path, + registry=registry, + store=store, + librarians=librarians, + extractor=extractor, + ) + current_head = await recovery.run() + if current_head is not None: + registry._meta["last_indexed_head"] = current_head + registry._meta["watcher_pid"] = os.getpid() + registry.save() + print(f"[libucks] startup recovery complete, HEAD={current_head[:8]}", file=sys.stderr) + except Exception as exc: + # Recovery is best-effort — never block server startup. + print(f"[libucks] startup recovery skipped: {exc}", file=sys.stderr) + + # ------------------------------------------------------------------ + # Git hook socket listener (Phase 6-D). + # ------------------------------------------------------------------ + sock_path = bucket_dir / "server.sock" + + async def _on_hook_event(payload: dict) -> None: + """Handle a JSON event from a git hook and trigger background re-index.""" + if recovery is None: + return + try: + new_head = await recovery.run() + if new_head is not None: + registry._meta["last_indexed_head"] = new_head + registry.save() + print(f"[libucks] hook event '{payload.get('event')}' → re-indexed HEAD={new_head[:8]}", file=sys.stderr) + except Exception as exc: + print(f"[libucks] hook event error: {exc}", file=sys.stderr) + + asyncio.ensure_future(serve_socket(sock_path, _on_hook_event)) + + # ------------------------------------------------------------------ + # HealthMonitor (Phase 6-E/6-F): autonomous quality guardian. + # ------------------------------------------------------------------ + mitosis_svc = MitosisService( + store=store, + registry=registry, + embedder=embedder, + agent=agent, + strategy=strategy, + mitosis_threshold=cfg.routing.mitosis_threshold, + ) + merging_svc = MergingService( + registry=registry, + store=store, + agent=agent, + embedder=embedder, + strategy=strategy, + ) + health_monitor = HealthMonitor( + registry=registry, + store=store, + mitosis_service=mitosis_svc, + merging_service=merging_svc, + embedder=embedder, + mitosis_threshold=cfg.routing.mitosis_threshold, + ) + asyncio.ensure_future(health_monitor.run()) + + # ------------------------------------------------------------------ + # StaleChecker + reindex callback (Phase 6-C JIT invalidation). + # ------------------------------------------------------------------ + stale_checker = StaleChecker(registry=registry, store=store, repo_path=repo_path) + + async def _reindex_stale(stale_bucket_ids: list[str]) -> None: + """Background re-index triggered by stale query results.""" + if recovery is None: + return + try: + new_head = await recovery.run() + if new_head is not None: + registry._meta["last_indexed_head"] = new_head + registry.save() + except Exception as exc: + print(f"[libucks] background reindex error: {exc}", file=sys.stderr) + orchestrator = QueryOrchestrator( central_agent=agent, librarians=librarians, embed_fn=embedder.embed, top_k=cfg.routing.top_k, + stale_checker=stale_checker, + reindex_fn=_reindex_stale, ) - translator = Translator(strategy) server = Server("libucks") diff --git a/libucks/merging_service.py b/libucks/merging_service.py new file mode 100644 index 0000000..6c7435c --- /dev/null +++ b/libucks/merging_service.py @@ -0,0 +1,214 @@ +"""MergingService — merge two semantically similar, under-filled buckets into one. + +Merge conditions (all three must hold): + 1. cosine_similarity(A.centroid, B.centroid) > MERGE_SIMILARITY (0.82) + 2. A.token_count + B.token_count < MERGE_TOKEN_LIMIT (15 000) + 3. Neither bucket appears in _meta.merge_history within the last hour. + +Merge protocol: + - Absorbing bucket = higher token count (keeps its ID). + - Dissolved bucket = smaller (deregistered and deleted). + - New centroid = normalize(mean(embed_batch(all chunks from both))). + - New prose = ThinkingStrategy.reason() over combined chunk content. + - Anti-cycle: append to _meta.merge_history; prune entries > 24 h old on save. +""" +from __future__ import annotations + +import base64 +from datetime import datetime, timedelta, timezone +from typing import TYPE_CHECKING, List, Optional, Set + +import numpy as np +import structlog + +from libucks.mitosis import _read_chunk_content + +if TYPE_CHECKING: + from libucks.central_agent import CentralAgent + from libucks.embeddings.embedding_service import EmbeddingService + from libucks.storage.bucket_registry import BucketRegistry + from libucks.storage.bucket_store import BucketStore + from libucks.thinking.base import ThinkingStrategy + +log = structlog.get_logger(__name__) + +MERGE_SIMILARITY: float = 0.82 +MERGE_TOKEN_LIMIT: int = 15_000 + + +def _encode_centroid(arr: np.ndarray) -> str: + return base64.b64encode(arr.astype(np.float32).tobytes()).decode() + + +class MergingService: + def __init__( + self, + registry: "BucketRegistry", + store: "BucketStore", + agent: "CentralAgent", + embedder: "EmbeddingService", + strategy: "ThinkingStrategy", + ) -> None: + self._registry = registry + self._store = store + self._agent = agent + self._embedder = embedder + self._strategy = strategy + + # ------------------------------------------------------------------ + # Public + # ------------------------------------------------------------------ + + async def run_merge_pass(self) -> None: + """Scan all bucket pairs and merge the first eligible pair found. + + One merge per pass avoids cascading consistency issues. + """ + centroids = self._registry.get_all_centroids() + bucket_ids = list(centroids.keys()) + recent = self._recent_merged_ids() + + for i, a in enumerate(bucket_ids): + for b in bucket_ids[i + 1 :]: + if self._should_merge(a, b, centroids, recent): + await self._merge(a, b) + return + + # ------------------------------------------------------------------ + # Internal + # ------------------------------------------------------------------ + + def _should_merge( + self, + a: str, + b: str, + centroids: dict, + recent: Set[str], + ) -> bool: + if a in recent or b in recent: + return False + sim = float(centroids[a] @ centroids[b]) # L2-normalised → cosine sim + if sim <= MERGE_SIMILARITY: + return False + try: + combined = self._registry.get_token_count(a) + self._registry.get_token_count(b) + except KeyError: + return False + return combined < MERGE_TOKEN_LIMIT + + def _recent_merged_ids(self) -> Set[str]: + """Return bucket IDs involved in any merge within the last hour.""" + cutoff = datetime.now(timezone.utc) - timedelta(hours=1) + result: Set[str] = set() + for entry in self._registry._meta.get("merge_history", []): + try: + merged_at_str: str = entry["merged_at"] + merged_at = datetime.fromisoformat(merged_at_str) + # Make offset-naive comparisons safe + if merged_at.tzinfo is None: + merged_at = merged_at.replace(tzinfo=timezone.utc) + if merged_at > cutoff: + for bid in entry.get("merged_bucket_ids", []): + result.add(bid) + except Exception: + pass + return result + + async def _merge(self, a: str, b: str) -> None: + try: + tokens_a = self._registry.get_token_count(a) + tokens_b = self._registry.get_token_count(b) + except KeyError: + return + + absorbing, dissolved = (a, b) if tokens_a >= tokens_b else (b, a) + log.info("merging.start", absorbing=absorbing, dissolved=dissolved) + + try: + fm_absorb, _ = self._store.read(absorbing) + fm_dissolve, _ = self._store.read(dissolved) + except FileNotFoundError as exc: + log.warning("merging.read_failed", error=str(exc)) + return + + all_chunks = fm_absorb.chunks + fm_dissolve.chunks + contents = [_read_chunk_content(c) for c in all_chunks] + + try: + embeddings = self._embedder.embed_batch(contents) + except Exception as exc: + log.warning("merging.embed_failed", error=str(exc)) + return + + centroid = np.mean(embeddings, axis=0).astype(np.float32) + norm = float(np.linalg.norm(centroid)) + if norm > 0: + centroid /= norm + + domain = f"{fm_absorb.domain_label} + {fm_dissolve.domain_label}" + combined_content = "\n".join(contents) + try: + result = await self._strategy.reason( + f"Summarize these merged code chunks: {domain}", + combined_content[:2000], + ) + prose = str(result) + except Exception as exc: + log.warning("merging.reason_failed", error=str(exc)) + prose = f"# {domain}\n\n(merged bucket)" + + self._store.create( + bucket_id=absorbing, + domain_label=domain, + centroid=_encode_centroid(centroid), + chunks=all_chunks, + prose=prose, + ) + total_tokens = sum(c.token_count for c in all_chunks) + await self._registry.register(absorbing, centroid, total_tokens) + + # Dissolve the smaller bucket + self._agent.unregister_librarian(dissolved) + try: + await self._registry.deregister(dissolved) + except KeyError: + pass + try: + self._store.delete(dissolved) + except FileNotFoundError: + pass + + # Anti-cycle bookkeeping + history: list = self._registry._meta.setdefault("merge_history", []) # type: ignore[assignment] + history.append( + { + "merged_bucket_ids": [absorbing, dissolved], + "result_bucket_id": absorbing, + "merged_at": datetime.now(timezone.utc).isoformat(), + } + ) + self._prune_merge_history() + self._registry.save() + + log.info( + "merging.complete", + absorbing=absorbing, + dissolved=dissolved, + total_tokens=total_tokens, + ) + + def _prune_merge_history(self) -> None: + cutoff = datetime.now(timezone.utc) - timedelta(hours=24) + history: List[dict] = self._registry._meta.get("merge_history", []) # type: ignore[assignment] + kept = [] + for entry in history: + try: + merged_at_str: str = entry["merged_at"] + merged_at = datetime.fromisoformat(merged_at_str) + if merged_at.tzinfo is None: + merged_at = merged_at.replace(tzinfo=timezone.utc) + if merged_at > cutoff: + kept.append(entry) + except Exception: + pass + self._registry._meta["merge_history"] = kept diff --git a/libucks/models/bucket.py b/libucks/models/bucket.py index 6c43f00..89683f7 100644 --- a/libucks/models/bucket.py +++ b/libucks/models/bucket.py @@ -1,4 +1,4 @@ -from typing import List +from typing import List, Optional from pydantic import BaseModel, field_validator @@ -11,6 +11,12 @@ class BucketFrontMatter(BaseModel): centroid_embedding: str # base64-encoded float32 array token_count: int chunks: List[ChunkMetadata] + # Phase 6-A metadata — all optional so old bucket files deserialise without error + last_indexed_at: Optional[str] = None # ISO-8601 UTC of last Librarian write + index_head_sha: Optional[str] = None # git HEAD SHA at last Librarian write + coherence_score: Optional[float] = None # set by HealthMonitor (Phase 6-E) + parent_bucket_id: Optional[str] = None # set on mitosis children + generation: int = 0 # incremented on each mitosis @field_validator("token_count") @classmethod diff --git a/libucks/models/chunk.py b/libucks/models/chunk.py index 7b045de..6930fef 100644 --- a/libucks/models/chunk.py +++ b/libucks/models/chunk.py @@ -1,3 +1,5 @@ +from typing import Optional + from pydantic import BaseModel, field_validator @@ -8,6 +10,7 @@ class ChunkMetadata(BaseModel): end_line: int git_sha: str token_count: int + indexed_at: Optional[str] = None # ISO-8601 UTC; set by Librarian on every update @field_validator("token_count") @classmethod diff --git a/libucks/query_orchestrator.py b/libucks/query_orchestrator.py index a41d073..65fa754 100644 --- a/libucks/query_orchestrator.py +++ b/libucks/query_orchestrator.py @@ -2,15 +2,18 @@ from __future__ import annotations import asyncio -from typing import Callable, List, Optional +from typing import Awaitable, Callable, List, Optional import numpy as np +import structlog from libucks.central_agent import CentralAgent from libucks.librarian import Librarian from libucks.models.events import QueryEvent from libucks.thinking.base import Representation +log = structlog.get_logger(__name__) + class QueryOrchestrator: def __init__( @@ -19,11 +22,15 @@ def __init__( librarians: dict[str, Librarian], embed_fn: Callable[[str], np.ndarray], top_k: int = 3, + stale_checker: object = None, # Optional[StaleChecker] — avoid circular import + reindex_fn: Optional[Callable[[List[str]], Awaitable[None]]] = None, ) -> None: self._agent = central_agent self._librarians = librarians self._embed_fn = embed_fn self._top_k = top_k + self._stale_checker = stale_checker + self._reindex_fn = reindex_fn async def query(self, text: str) -> List[Representation]: embedding = self._embed_fn(text) @@ -31,6 +38,22 @@ async def query(self, text: str) -> List[Representation]: if not bucket_ids: return [] + # ---- JIT stale check (Phase 6-C) ------------------------------------- + if self._stale_checker is not None: + stale_result = await self._stale_checker.check(bucket_ids) + if stale_result.is_stale: + log.warning( + "query.stale_buckets_detected", + level=stale_result.level, + reason=stale_result.reason, + stale_bucket_ids=stale_result.stale_bucket_ids, + ) + if self._reindex_fn is not None: + asyncio.ensure_future( + self._reindex_fn(stale_result.stale_bucket_ids) + ) + # ---- Answer with possibly-stale data (eventual consistency) ---------- + async def _query_one(bucket_id: str) -> Optional[Representation]: librarian = self._librarians.get(bucket_id) if librarian is None: diff --git a/libucks/stale_checker.py b/libucks/stale_checker.py new file mode 100644 index 0000000..b1f5c1c --- /dev/null +++ b/libucks/stale_checker.py @@ -0,0 +1,254 @@ +"""StaleChecker — JIT (Just-In-Time) staleness detection for the query path. + +Runs in <50ms before Librarians are called. If any of the four levels detects +that a bucket's content may be out of sync with the source files, it returns +a StaleCheckResult with is_stale=True so the orchestrator can fire a background +re-index while still answering the query with the stale data (eventual consistency). + +Four levels (ordered cheapest → most specific): + + Level 1 — Process: Is the watcher PID recorded in _meta still alive? + Uses os.kill(pid, 0); ESRCH means the process is gone. + Budget: ~2ms. + + Level 2 — Git HEAD: Does the current HEAD differ from _meta.last_indexed_head? + Uses a subprocess git rev-parse HEAD. + If different, ALL queried buckets are flagged stale and we return early. + Budget: ~15ms. + + Level 3 — File mtime: Are source files newer than the bucket's last_indexed_at? + Uses os.stat().st_mtime on every unique source_file in each queried bucket. + Detects uncommitted saves (the watcher debounce window, IDE saves). + Budget: ~5ms. + + Level 4 — Chunk SHA: Does the bucket's index_head_sha differ from current HEAD? + Detects per-bucket SHA drift (bucket was indexed at an older commit). + Purely in-memory; current_head reused from Level 2. + Budget: ~5ms. + +The three module-level helpers (_process_is_alive, _get_current_head, +_get_file_mtime) are extracted for test-patchability without global mocking. +""" +from __future__ import annotations + +import os +import subprocess +from dataclasses import dataclass, field +from datetime import datetime +from pathlib import Path +from typing import TYPE_CHECKING, List, Optional + +import structlog + +if TYPE_CHECKING: + from libucks.storage.bucket_registry import BucketRegistry + from libucks.storage.bucket_store import BucketStore + +log = structlog.get_logger(__name__) + + +# --------------------------------------------------------------------------- +# Injectable helpers (patched in unit tests) +# --------------------------------------------------------------------------- + +def _process_is_alive(pid: int) -> bool: + """Return True if the process with *pid* exists on this machine.""" + try: + os.kill(pid, 0) + return True + except ProcessLookupError: + return False + except PermissionError: + return True # process exists; we just lack permission to signal it + except OSError: + return False + + +def _get_current_head(repo_path: Path) -> Optional[str]: + """Return current git HEAD SHA via subprocess, or None if git is unavailable.""" + try: + result = subprocess.run( + ["git", "-C", str(repo_path), "rev-parse", "HEAD"], + capture_output=True, + text=True, + timeout=10, + ) + if result.returncode == 0: + return result.stdout.strip() + except Exception: + pass + return None + + +def _get_file_mtime(path: str) -> Optional[float]: + """Return os.stat mtime for *path*, or None if the file cannot be stat'd.""" + try: + return os.stat(path).st_mtime + except OSError: + return None + + +# --------------------------------------------------------------------------- +# Result type +# --------------------------------------------------------------------------- + +@dataclass +class StaleCheckResult: + is_stale: bool + stale_bucket_ids: List[str] + level: int # 0 = fresh; 1–4 = first level that triggered + reason: str + + +# --------------------------------------------------------------------------- +# StaleChecker +# --------------------------------------------------------------------------- + +class StaleChecker: + """Checks the freshness of a set of bucket IDs before they are queried. + + Args: + registry: The live BucketRegistry (read-only; provides per-bucket metadata). + store: The BucketStore (used for Level 3 to read chunk source-file paths). + repo_path: The repository root (used for Level 2 subprocess call). + """ + + def __init__( + self, + registry: "BucketRegistry", + store: "BucketStore", + repo_path: Path, + ) -> None: + self._registry = registry + self._store = store + self._repo_path = repo_path + + # ------------------------------------------------------------------ + # Public + # ------------------------------------------------------------------ + + async def check(self, bucket_ids: List[str]) -> StaleCheckResult: + """Check *bucket_ids* for staleness. Returns within ~50ms.""" + + # ---- Level 1: watcher process alive? --------------------------------- + watcher_pid = self._registry._meta.get("watcher_pid") + if watcher_pid is not None: + try: + pid_int = int(watcher_pid) + except (TypeError, ValueError): + pid_int = None + if pid_int is not None and not _process_is_alive(pid_int): + log.warning( + "stale_checker.watcher_down", + pid=pid_int, + level=1, + ) + return StaleCheckResult( + is_stale=True, + stale_bucket_ids=list(bucket_ids), + level=1, + reason=f"watcher process (PID {pid_int}) is not running", + ) + + # ---- Level 2: git HEAD drift? ---------------------------------------- + current_head = _get_current_head(self._repo_path) + last_indexed_head: Optional[str] = self._registry._meta.get("last_indexed_head") # type: ignore[assignment] + + if current_head is not None and last_indexed_head is not None: + if current_head != last_indexed_head: + log.warning( + "stale_checker.head_drift", + from_sha=last_indexed_head[:8], + to_sha=current_head[:8], + level=2, + ) + return StaleCheckResult( + is_stale=True, + stale_bucket_ids=list(bucket_ids), + level=2, + reason=( + f"global HEAD drifted: " + f"{last_indexed_head[:8]} → {current_head[:8]}" + ), + ) + + # ---- Levels 3 & 4: per-bucket checks --------------------------------- + stale_ids: List[str] = [] + first_trigger_level: Optional[int] = None + + for bucket_id in bucket_ids: + entry = self._registry._buckets.get(bucket_id) + if entry is None: + continue + + # Level 3 — file mtime vs last_indexed_at + if entry.last_indexed_at is not None: + try: + indexed_ts = datetime.fromisoformat(entry.last_indexed_at).timestamp() + if self._any_source_file_newer(bucket_id, indexed_ts): + stale_ids.append(bucket_id) + if first_trigger_level is None: + first_trigger_level = 3 + log.warning( + "stale_checker.file_newer_than_index", + bucket_id=bucket_id, + level=3, + ) + continue + except (ValueError, TypeError): + pass # malformed timestamp — skip Level 3 for this bucket + + # Level 4 — per-bucket index_head_sha vs current HEAD + if ( + current_head is not None + and entry.index_head_sha is not None + and entry.index_head_sha != current_head + ): + stale_ids.append(bucket_id) + if first_trigger_level is None: + first_trigger_level = 4 + log.warning( + "stale_checker.bucket_sha_drift", + bucket_id=bucket_id, + bucket_sha=entry.index_head_sha[:8], + current_sha=current_head[:8], + level=4, + ) + + if stale_ids: + level = first_trigger_level or 3 + return StaleCheckResult( + is_stale=True, + stale_bucket_ids=stale_ids, + level=level, + reason=f"level-{level} staleness in {len(stale_ids)} bucket(s)", + ) + + return StaleCheckResult( + is_stale=False, + stale_bucket_ids=[], + level=0, + reason="all queried buckets are fresh", + ) + + # ------------------------------------------------------------------ + # Internal + # ------------------------------------------------------------------ + + def _any_source_file_newer(self, bucket_id: str, indexed_ts: float) -> bool: + """Return True if any source file in *bucket_id* has mtime > *indexed_ts*.""" + try: + front_matter, _ = self._store.read(bucket_id) + except FileNotFoundError: + return False + + seen: set[str] = set() + for chunk in front_matter.chunks: + src = chunk.source_file + if src in seen: + continue + seen.add(src) + mtime = _get_file_mtime(src) + if mtime is not None and mtime > indexed_ts: + return True + return False diff --git a/libucks/startup_recovery.py b/libucks/startup_recovery.py new file mode 100644 index 0000000..9dad235 --- /dev/null +++ b/libucks/startup_recovery.py @@ -0,0 +1,206 @@ +"""StartupRecovery — replay commits that arrived while libucks serve was offline. + +Algorithm (runs once, synchronously, before the MCP stdio server starts): + + 1. Read registry._meta["last_indexed_head"] — the HEAD SHA at last save. + 2. Run `git rev-parse HEAD` — the current HEAD. + 3. If they differ (gap detected): + a. `git diff --name-only ` — which files changed. + b. For each file whose extension is tracked: + - Resolve which bucket(s) own chunks from that file. + - `DiffExtractor.extract_between(file, last, current)` — get the diff. + - For each hunk, call `librarian.handle(UpdateEvent(…))`. + 4. Always return the current HEAD so the caller can update the baseline. + +If `last_indexed_head` is None (first run after `libucks init`), no recovery is +attempted — the index was just built by INIT, so it is already current. The +current HEAD is still returned so the caller can record it as the new baseline. + +The two module-level git helpers (_git_rev_parse_head, _git_diff_name_only) are +deliberately extracted so tests can patch them without mocking subprocess globally. +""" +from __future__ import annotations + +import subprocess +from pathlib import Path +from typing import TYPE_CHECKING, Dict, List, Optional + +import structlog + +from libucks.diff.diff_extractor import DiffExtractor +from libucks.models.events import UpdateEvent +from libucks.storage.bucket_registry import BucketRegistry +from libucks.storage.bucket_store import BucketStore +from libucks.watchdog_service import _TRACKED_EXTENSIONS + +if TYPE_CHECKING: + from libucks.librarian import Librarian + +log = structlog.get_logger(__name__) + + +# --------------------------------------------------------------------------- +# Injectable git helpers (patched in unit tests) +# --------------------------------------------------------------------------- + +def _git_rev_parse_head(repo_path: Path) -> Optional[str]: + """Return current git HEAD SHA, or None if git is unavailable.""" + try: + result = subprocess.run( + ["git", "-C", str(repo_path), "rev-parse", "HEAD"], + capture_output=True, + text=True, + timeout=10, + ) + if result.returncode == 0: + return result.stdout.strip() + except Exception: + pass + return None + + +def _git_diff_name_only(repo_path: Path, from_sha: str, to_sha: str) -> List[str]: + """Return list of repo-relative file paths changed between two SHAs.""" + try: + result = subprocess.run( + ["git", "-C", str(repo_path), "diff", "--name-only", from_sha, to_sha], + capture_output=True, + text=True, + timeout=30, + ) + if result.returncode == 0: + return [f for f in result.stdout.strip().splitlines() if f] + except Exception: + pass + return [] + + +# --------------------------------------------------------------------------- +# StartupRecovery +# --------------------------------------------------------------------------- + +class StartupRecovery: + def __init__( + self, + repo_path: Path, + registry: BucketRegistry, + store: BucketStore, + librarians: Dict[str, "Librarian"], + extractor: DiffExtractor, + ) -> None: + self._repo_path = repo_path + self._registry = registry + self._store = store + self._librarians = librarians + self._extractor = extractor + + def _find_buckets_for_file(self, rel_filepath: str) -> List[str]: + """Return bucket IDs that own at least one chunk from the given file. + + Matching is done by resolving both paths to absolute form so that + relative-vs-absolute mismatches (common when mixing git output with + stored absolute paths) do not cause missed updates. + """ + try: + abs_target = (self._repo_path / rel_filepath).resolve() + except Exception: + return [] + + matched: List[str] = [] + for bucket_id in self._store.list_all(): + try: + front_matter, _ = self._store.read(bucket_id) + except FileNotFoundError: + continue + for chunk in front_matter.chunks: + try: + chunk_abs = Path(chunk.source_file).resolve() + except Exception: + continue + if chunk_abs == abs_target: + matched.append(bucket_id) + break # one match per bucket is enough + + return matched + + async def run(self) -> Optional[str]: + """Replay any commits missed while the server was offline. + + Returns the current git HEAD SHA if git is reachable, None otherwise. + The caller MUST write this value into registry._meta["last_indexed_head"] + and call registry.save() so the next startup has an accurate baseline. + """ + current_head = _git_rev_parse_head(self._repo_path) + if not current_head: + log.warning("startup_recovery.git_unavailable", repo=str(self._repo_path)) + return None + + last_head: Optional[str] = self._registry._meta.get("last_indexed_head") + + if not last_head: + log.info( + "startup_recovery.no_baseline", + current_head=current_head[:8], + note="recording baseline; no recovery needed", + ) + return current_head + + if last_head == current_head: + log.info("startup_recovery.up_to_date", head=current_head[:8]) + return current_head + + log.info( + "startup_recovery.gap_detected", + from_sha=last_head[:8], + to_sha=current_head[:8], + ) + + changed_files = _git_diff_name_only(self._repo_path, last_head, current_head) + log.info("startup_recovery.changed_files_count", count=len(changed_files)) + + recovered_updates = 0 + for rel_filepath in changed_files: + suffix = Path(rel_filepath).suffix.lower() + if suffix not in _TRACKED_EXTENSIONS: + log.debug("startup_recovery.skip_extension", file=rel_filepath, suffix=suffix) + continue + + bucket_ids = self._find_buckets_for_file(rel_filepath) + if not bucket_ids: + log.debug("startup_recovery.no_bucket_for_file", file=rel_filepath) + continue + + try: + diff_events = self._extractor.extract_between( + self._repo_path / rel_filepath, + last_head, + current_head, + ) + except Exception as exc: + log.warning( + "startup_recovery.extract_failed", + file=rel_filepath, + error=str(exc), + ) + continue + + if not diff_events: + log.debug("startup_recovery.empty_diff", file=rel_filepath) + continue + + for bucket_id in bucket_ids: + librarian = self._librarians.get(bucket_id) + if librarian is None: + continue + for diff_event in diff_events: + for hunk in diff_event.hunks: + update = UpdateEvent(bucket_id=bucket_id, hunk=hunk) + await librarian.handle(update) + recovered_updates += 1 + + log.info( + "startup_recovery.complete", + recovered_updates=recovered_updates, + to_sha=current_head[:8], + ) + return current_head diff --git a/libucks/storage/bucket_registry.py b/libucks/storage/bucket_registry.py index e1de500..13df9a9 100644 --- a/libucks/storage/bucket_registry.py +++ b/libucks/storage/bucket_registry.py @@ -20,19 +20,28 @@ import json import struct from pathlib import Path -from typing import Dict +from typing import Dict, Optional import numpy as np class _BucketEntry: - __slots__ = ("centroid", "token_count", "lock", "is_splitting") + __slots__ = ( + "centroid", "token_count", "lock", "is_splitting", + "last_indexed_at", "index_head_sha", "coherence_score", + "query_hit_count", "last_query_hit_at", + ) def __init__(self, centroid: np.ndarray, token_count: int, lock: asyncio.Lock) -> None: self.centroid = centroid self.token_count = token_count self.lock = lock self.is_splitting = False + self.last_indexed_at: Optional[str] = None + self.index_head_sha: Optional[str] = None + self.coherence_score: Optional[float] = None + self.query_hit_count: int = 0 + self.last_query_hit_at: Optional[str] = None def _encode_centroid(centroid: np.ndarray) -> str: @@ -49,6 +58,13 @@ class BucketRegistry: def __init__(self, registry_path: Path) -> None: self._path = registry_path self._buckets: Dict[str, _BucketEntry] = {} + self._meta: Dict[str, object] = { + "schema_version": 2, + "last_indexed_head": None, + "last_indexed_at": None, + "watcher_pid": None, + "merge_history": [], + } # ------------------------------------------------------------------ # Mutating operations (async so callers can use asyncio.gather) @@ -79,6 +95,16 @@ async def set_splitting(self, bucket_id: str, flag: bool) -> None: raise KeyError(f"Bucket not registered: {bucket_id!r}") self._buckets[bucket_id].is_splitting = flag + def update_index_timestamp( + self, bucket_id: str, last_indexed_at: str, index_head_sha: str + ) -> None: + """Record when and at which git HEAD a bucket was last written by a Librarian.""" + if bucket_id not in self._buckets: + raise KeyError(f"Bucket not registered: {bucket_id!r}") + entry = self._buckets[bucket_id] + entry.last_indexed_at = last_indexed_at + entry.index_head_sha = index_head_sha + # ------------------------------------------------------------------ # Read-only operations (sync — safe because asyncio is single-threaded) # ------------------------------------------------------------------ @@ -106,14 +132,18 @@ def get_lock(self, bucket_id: str) -> asyncio.Lock: # ------------------------------------------------------------------ def save(self) -> None: - data = { - bucket_id: { + data: Dict[str, object] = {"_meta": self._meta} + for bucket_id, entry in self._buckets.items(): + data[bucket_id] = { "centroid_embedding": _encode_centroid(entry.centroid), "token_count": entry.token_count, "is_splitting": entry.is_splitting, + "last_indexed_at": entry.last_indexed_at, + "index_head_sha": entry.index_head_sha, + "coherence_score": entry.coherence_score, + "query_hit_count": entry.query_hit_count, + "last_query_hit_at": entry.last_query_hit_at, } - for bucket_id, entry in self._buckets.items() - } self._path.parent.mkdir(parents=True, exist_ok=True) self._path.write_text(json.dumps(data, indent=2), encoding="utf-8") @@ -121,11 +151,22 @@ def load(self) -> None: if not self._path.exists(): return data = json.loads(self._path.read_text(encoding="utf-8")) - for bucket_id, state in data.items(): + for key, state in data.items(): + if key.startswith("_"): + # Handle registry-level meta blocks; ignore unknown underscore keys. + if key == "_meta" and isinstance(state, dict): + self._meta.update(state) + continue + bucket_id = key centroid = _decode_centroid(state["centroid_embedding"]) # Preserve any existing lock if already in memory (e.g. partial reload). existing = self._buckets.get(bucket_id) lock = existing.lock if existing else asyncio.Lock() entry = _BucketEntry(centroid=centroid, token_count=state["token_count"], lock=lock) entry.is_splitting = state.get("is_splitting", False) + entry.last_indexed_at = state.get("last_indexed_at") + entry.index_head_sha = state.get("index_head_sha") + entry.coherence_score = state.get("coherence_score") + entry.query_hit_count = state.get("query_hit_count", 0) + entry.last_query_hit_at = state.get("last_query_hit_at") self._buckets[bucket_id] = entry diff --git a/libucks/storage/bucket_store.py b/libucks/storage/bucket_store.py index f259ccb..cf4ba8c 100644 --- a/libucks/storage/bucket_store.py +++ b/libucks/storage/bucket_store.py @@ -67,23 +67,35 @@ def _require_exists(self, bucket_id: str) -> Path: @staticmethod def _bfm_to_yaml_block(bfm: BucketFrontMatter) -> str: """Serialise a BucketFrontMatter to the raw YAML block string (no delimiters).""" + chunks_data = [] + for c in bfm.chunks: + chunk_dict: dict = { + "chunk_id": c.chunk_id, + "source_file": c.source_file, + "start_line": c.start_line, + "end_line": c.end_line, + "git_sha": c.git_sha, + "token_count": c.token_count, + } + if c.indexed_at is not None: + chunk_dict["indexed_at"] = c.indexed_at + chunks_data.append(chunk_dict) + data: dict = { "bucket_id": bfm.bucket_id, "domain_label": bfm.domain_label, "centroid_embedding": bfm.centroid_embedding, "token_count": bfm.token_count, - "chunks": [ - { - "chunk_id": c.chunk_id, - "source_file": c.source_file, - "start_line": c.start_line, - "end_line": c.end_line, - "git_sha": c.git_sha, - "token_count": c.token_count, - } - for c in bfm.chunks - ], + "chunks": chunks_data, } + # Include new optional metadata fields only when set — keeps files clean. + for field in ("last_indexed_at", "index_head_sha", "parent_bucket_id", "coherence_score"): + val = getattr(bfm, field) + if val is not None: + data[field] = val + if bfm.generation: + data["generation"] = bfm.generation + return yaml.dump(data, allow_unicode=True, sort_keys=False) @staticmethod @@ -96,6 +108,11 @@ def _yaml_block_to_bfm(yaml_block: str) -> BucketFrontMatter: centroid_embedding=parsed["centroid_embedding"], token_count=parsed["token_count"], chunks=chunks, + last_indexed_at=parsed.get("last_indexed_at"), + index_head_sha=parsed.get("index_head_sha"), + coherence_score=parsed.get("coherence_score"), + parent_bucket_id=parsed.get("parent_bucket_id"), + generation=parsed.get("generation", 0), ) # ------------------------------------------------------------------ diff --git a/main.py b/main.py index ffd94d3..43ea9b3 100644 --- a/main.py +++ b/main.py @@ -1,33 +1,14 @@ -import asyncio +import sys from pathlib import Path -import click - - -@click.group() -@click.version_option(version="0.1.0", prog_name="libucks") -def cli(): - """libucks — Librarian Buckets, local AI memory server for coding agents.""" - - -@cli.command("init") -@click.option("--local", "local_path", type=click.Path(exists=True, file_okay=False, path_type=Path), - required=True, help="Path to a local repository to index.") -def init_cmd(local_path: Path): - """Seed libucks buckets from a local repository.""" - from libucks.init_orchestrator import InitOrchestrator - - orchestrator = InitOrchestrator(local_path) - asyncio.run(orchestrator.run()) - - -@cli.command("serve") -def serve_cmd(): - """Start the libucks MCP server over stdio.""" - import asyncio - from libucks.mcp_bridge import serve - asyncio.run(serve()) +# When run as a script (python /abs/path/main.py serve) the project root is +# not automatically on sys.path. Insert it so `from libucks.xxx import yyy` +# resolves correctly regardless of cwd or how Claude Desktop invokes us. +_PROJECT_ROOT = Path(__file__).parent.resolve() +if str(_PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(_PROJECT_ROOT)) +from libucks._cli import cli # noqa: E402 — path bootstrap must come first if __name__ == "__main__": cli() diff --git a/pyproject.toml b/pyproject.toml index bfb6e9b..f8477cc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -35,7 +35,7 @@ dev = [ ] [project.scripts] -libucks = "main:cli" +libucks = "libucks._cli:cli" [tool.pytest.ini_options] asyncio_mode = "auto" diff --git a/scripts/check_nervous_system.py b/scripts/check_nervous_system.py new file mode 100644 index 0000000..066b73d --- /dev/null +++ b/scripts/check_nervous_system.py @@ -0,0 +1,296 @@ +#!/usr/bin/env python3 +""" +scripts/check_nervous_system.py +================================ +End-to-end vibe check for the libucks production engine. + +Checks (in order): + 1. IPC — is the Unix socket server reachable? + 2. Registry — bucket health: token counts vs mitosis threshold + 3. Stale — JIT staleness detection after touching a tracked source file + 4. Query — direct Python query for the new test function name + +Usage: + python scripts/check_nervous_system.py [--repo /path/to/repo] + +Exit code: 0 = all critical checks passed, 1 = one or more failures. +""" +from __future__ import annotations + +import argparse +import asyncio +import json +import os +import socket +import subprocess +import sys +import textwrap +import time +from datetime import datetime, timezone +from pathlib import Path +from typing import Optional + +# ── Repo / path resolution ──────────────────────────────────────────────────── + +def _find_repo(explicit: Optional[str]) -> Path: + if explicit: + return Path(explicit).resolve() + env = os.environ.get("LIBUCKS_REPO_PATH") + if env: + return Path(env).resolve() + # Walk from script location up to project root + return Path(__file__).parent.parent.resolve() + + +# ── Terminal colours ────────────────────────────────────────────────────────── + +_USE_COLOUR = sys.stdout.isatty() + +def _c(code: str, text: str) -> str: + return f"\033[{code}m{text}\033[0m" if _USE_COLOUR else text + +PASS = lambda: _c("32", "PASS") +FAIL = lambda: _c("31", "FAIL") +SKIP = lambda: _c("33", "SKIP") +INFO = lambda: _c("36", "INFO") +WARN = lambda: _c("33", "WARN") + + +def _hdr(n: int, title: str) -> None: + print(f"\n{'─'*60}") + print(f" [{n}/4] {title}") + print(f"{'─'*60}") + + +def _row(tag_fn, msg: str) -> None: + print(f" {tag_fn():<18} {msg}") + + +# ── 1. IPC check ───────────────────────────────────────────────────────────── + +def check_ipc(sock_path: Path) -> bool: + _hdr(1, "IPC — Unix socket server") + print(f" Socket : {sock_path}") + + if not sock_path.exists(): + _row(SKIP, "socket file not found — start the server with `libucks serve`") + return True # not a hard failure; server may simply not be running + + payload = json.dumps({"event": "vibe-check", "source": "check_nervous_system"}).encode() + try: + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as s: + s.settimeout(3) + s.connect(str(sock_path)) + s.sendall(payload) + _row(PASS, "server acknowledged the ping (connection accepted)") + return True + except OSError as exc: + _row(FAIL, f"could not connect to socket: {exc}") + return False + + +# ── 2. Registry health ──────────────────────────────────────────────────────── + +def check_registry(registry_path: Path, mitosis_threshold: int = 20_000) -> bool: + _hdr(2, "Registry — bucket health") + print(f" File : {registry_path}") + + if not registry_path.exists(): + _row(FAIL, "registry.json not found — run `libucks init --local ` first") + return False + + data = json.loads(registry_path.read_text()) + buckets = {k: v for k, v in data.items() if not k.startswith("_")} + meta = data.get("_meta", {}) + + print(f" Head : {str(meta.get('last_indexed_head', 'unknown'))[:12]}") + print(f" Watcher PID: {meta.get('watcher_pid', 'not set')}") + print() + + oversized = 0 + for bid, state in sorted(buckets.items(), key=lambda x: -x[1].get("token_count", 0)): + tokens = state.get("token_count", 0) + coherence = state.get("coherence_score") + coh_str = f" coherence={coherence:.2f}" if coherence is not None else "" + flag = "" + if tokens >= mitosis_threshold: + flag = f" ← OVER {mitosis_threshold:,} (pending mitosis)" + oversized += 1 + print(f" {bid} {tokens:>7,} tokens{coh_str}{flag}") + + print() + if oversized: + _row(WARN, f"{oversized} bucket(s) over threshold — HealthMonitor will split on next pass") + else: + _row(PASS, f"{len(buckets)} bucket(s) all within token threshold") + + _row(PASS, f"registry loaded successfully ({len(buckets)} buckets)") + return True + + +# ── 3. Stale check ──────────────────────────────────────────────────────────── + +async def _run_stale_check(repo: Path, registry_path: Path) -> bool: + """ + Creates a unique test file, touches a tracked source file to advance its mtime, + then runs StaleChecker to prove Level-3 detection fires. + """ + from libucks.storage.bucket_registry import BucketRegistry + from libucks.storage.bucket_store import BucketStore + from libucks.stale_checker import StaleChecker + + # ── Create the weird test function file ─────────────────────────────────── + test_file = repo / "fastapi" / "internal" / "test_logic.py" + test_file.parent.mkdir(parents=True, exist_ok=True) + test_file.write_text(textwrap.dedent("""\ + # Auto-generated by check_nervous_system.py — safe to delete. + def super_secret_fastapi_magic_999(request): + \"\"\"Validates the incoming request payload against a secret schema.\"\"\" + return {"status": "ok", "magic": 999} + """)) + print(f" Created: {test_file}") + + # ── Load registry + store ───────────────────────────────────────────────── + registry = BucketRegistry(registry_path) + registry.load() + bucket_dir = registry_path.parent / "buckets" + store = BucketStore(bucket_dir) + + bucket_ids = list(registry.get_all_centroids().keys()) + if not bucket_ids: + _row(SKIP, "no buckets registered — index first") + return True + + # ── Touch a file that IS already tracked (Level 3 trigger) ─────────────── + # Find the first tracked source file across all buckets + touched_file: Optional[str] = None + for bid in bucket_ids: + try: + fm, _ = store.read(bid) + if fm.chunks: + src = fm.chunks[0].source_file + p = Path(src) + if p.exists(): + p.touch() # advance mtime + touched_file = src + print(f" Touched: {p.name} (bucket={bid[:8]})") + break + except FileNotFoundError: + continue + + # ── Run StaleChecker ────────────────────────────────────────────────────── + checker = StaleChecker(registry=registry, store=store, repo_path=repo) + result_obj = await checker.check(bucket_ids) + + print() + print(f" StaleCheckResult:") + print(f" is_stale = {result_obj.is_stale}") + print(f" level = {result_obj.level}") + print(f" reason = {result_obj.reason}") + print(f" stale_bucket_ids= {result_obj.stale_bucket_ids}") + print() + + if result_obj.is_stale: + _row(PASS, f"StaleChecker fired (Level {result_obj.level}): {result_obj.reason}") + else: + if touched_file: + _row(WARN, "touched file did not trigger staleness — check last_indexed_at timestamps") + else: + _row(INFO, "no tracked source file found to touch; staleness not verified") + + return True + + +def check_stale(repo: Path, registry_path: Path) -> bool: + _hdr(3, "Stale — JIT staleness detection") + return asyncio.run(_run_stale_check(repo, registry_path)) + + +# ── 4. Query ────────────────────────────────────────────────────────────────── + +def check_query(repo: Path, registry_path: Path) -> bool: + """ + Searches all bucket store files for the weird function name without loading + the embedding model (fast grep-style scan of raw .md files in .libucks/buckets/). + If the function is found, the content has been indexed. + If not, that confirms the StaleChecker was right to flag staleness. + """ + _hdr(4, "Query — content discovery for super_secret_fastapi_magic_999") + target = "super_secret_fastapi_magic_999" + bucket_dir = registry_path.parent / "buckets" + + if not bucket_dir.exists(): + _row(SKIP, "bucket directory not found") + return True + + md_files = list(bucket_dir.glob("*.md")) + print(f" Scanning {len(md_files)} bucket file(s) for: {target!r}") + + found_in: list[str] = [] + for md in md_files: + try: + if target in md.read_text(errors="replace"): + found_in.append(md.stem) + except OSError: + pass + + print() + if found_in: + _row(PASS, f"function name found in bucket(s): {found_in}") + _row(INFO, "content is already indexed — query would return a live answer") + else: + _row(INFO, "function not yet in any bucket (as expected for a brand-new file)") + _row(INFO, "StaleChecker detected the gap; next re-index will absorb it") + + # Also try a subprocess grep over the actual source to confirm the file exists + result = subprocess.run( + ["grep", "-r", target, str(repo / "fastapi" / "internal")], + capture_output=True, text=True, + ) + if result.stdout.strip(): + _row(PASS, f"source file confirmed on disk:\n {result.stdout.strip()}") + + return True + + +# ── Main ───────────────────────────────────────────────────────────────────── + +def main() -> int: + parser = argparse.ArgumentParser(description="libucks nervous system vibe check") + parser.add_argument("--repo", default=None, help="Path to the indexed repository") + args = parser.parse_args() + + repo = _find_repo(args.repo) + libucks_dir = repo / ".libucks" + registry_path = libucks_dir / "registry.json" + sock_path = libucks_dir / "server.sock" + + print(f"\n{'═'*60}") + print(f" libucks nervous system check") + print(f" repo={repo}") + print(f" time={datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')}") + print(f"{'═'*60}") + + failures = 0 + + if not check_ipc(sock_path): + failures += 1 + if not check_registry(registry_path): + failures += 1 + if not check_stale(repo, registry_path): + failures += 1 + if not check_query(repo, registry_path): + failures += 1 + + print(f"\n{'═'*60}") + if failures == 0: + print(f" {PASS()} All checks passed.") + else: + print(f" {FAIL()} {failures} check(s) failed.") + print(f"{'═'*60}\n") + + return 0 if failures == 0 else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/integration/test_health_and_merge.py b/tests/integration/test_health_and_merge.py new file mode 100644 index 0000000..185ef6b --- /dev/null +++ b/tests/integration/test_health_and_merge.py @@ -0,0 +1,262 @@ +"""Integration tests for HealthMonitor (6-E) and MergingService (6-F). + +Uses real registry/store objects with mocked embedder and strategy. +Proves that the size trigger, coherence trigger, and merge trigger all fire. +""" +from __future__ import annotations + +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch + +import numpy as np +import pytest + +from libucks.health_monitor import HealthMonitor, _COHERENCE_THRESHOLD +from libucks.merging_service import MergingService, MERGE_SIMILARITY, MERGE_TOKEN_LIMIT +from libucks.mitosis import MitosisService +from libucks.storage.bucket_registry import BucketRegistry +from libucks.storage.bucket_store import BucketStore + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture() +def registry(tmp_path: Path) -> BucketRegistry: + reg = BucketRegistry(tmp_path / "registry.json") + return reg + + +@pytest.fixture() +def store(tmp_path: Path) -> BucketStore: + return BucketStore(tmp_path / "buckets") + + +def _mock_embedder(dim: int = 8) -> MagicMock: + """Return a fake EmbeddingService whose embed/embed_batch return deterministic unit vectors.""" + embedder = MagicMock() + embedder.embed.side_effect = lambda text: np.ones(dim, dtype=np.float32) / np.sqrt(dim) + embedder.embed_batch.side_effect = lambda texts: np.ones((len(texts), dim), dtype=np.float32) / np.sqrt(dim) + return embedder + + +def _mock_strategy() -> MagicMock: + strategy = MagicMock() + strategy.reason = AsyncMock(return_value="mock prose") + return strategy + + +def _mock_agent() -> MagicMock: + agent = MagicMock() + agent.unregister_librarian = MagicMock() + return agent + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +async def _seed(registry: BucketRegistry, bucket_id: str, tokens: int, centroid: np.ndarray) -> None: + norm = np.linalg.norm(centroid) + if norm > 0: + centroid = centroid / norm + await registry.register(bucket_id, centroid.astype(np.float32), tokens) + + +# --------------------------------------------------------------------------- +# 6-E: HealthMonitor size trigger +# --------------------------------------------------------------------------- + +class TestHealthMonitorSizeTrigger: + @pytest.mark.asyncio + async def test_size_trigger_calls_mitosis_split(self, registry: BucketRegistry, store: BucketStore) -> None: + """A bucket over the token threshold must trigger MitosisService.split().""" + BIG_TOKEN_COUNT = 25_000 # above default 20k threshold + bucket_id = "aabbccdd" + centroid = np.array([1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0], dtype=np.float32) + await _seed(registry, bucket_id, BIG_TOKEN_COUNT, centroid) + + mitosis_svc = MagicMock(spec=MitosisService) + mitosis_svc.split = AsyncMock() + merging_svc = MagicMock(spec=MergingService) + merging_svc.run_merge_pass = AsyncMock() + + monitor = HealthMonitor( + registry=registry, + store=store, + mitosis_service=mitosis_svc, + merging_service=merging_svc, + embedder=_mock_embedder(), + mitosis_threshold=20_000, + ) + + await monitor._check() + + mitosis_svc.split.assert_called_once_with(bucket_id) + merging_svc.run_merge_pass.assert_called_once() + + +# --------------------------------------------------------------------------- +# 6-E: HealthMonitor coherence trigger +# --------------------------------------------------------------------------- + +class TestHealthMonitorCoherenceTrigger: + @pytest.mark.asyncio + async def test_low_coherence_calls_mitosis_split( + self, registry: BucketRegistry, store: BucketStore + ) -> None: + """A bucket whose chunks are incoherent must trigger MitosisService.split().""" + bucket_id = "coh00001" + centroid = np.array([1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0], dtype=np.float32) + await _seed(registry, bucket_id, 5_000, centroid) + + mitosis_svc = MagicMock(spec=MitosisService) + mitosis_svc.split = AsyncMock() + merging_svc = MagicMock(spec=MergingService) + merging_svc.run_merge_pass = AsyncMock() + + # 4 orthogonal unit vectors → mean coherence = 1/√4 = 0.5 < threshold 0.55 + embedder = MagicMock() + dim = 8 + e1 = np.zeros(dim, dtype=np.float32); e1[0] = 1.0 + e2 = np.zeros(dim, dtype=np.float32); e2[1] = 1.0 + e3 = np.zeros(dim, dtype=np.float32); e3[2] = 1.0 + e4 = np.zeros(dim, dtype=np.float32); e4[3] = 1.0 + embedder.embed_batch.side_effect = lambda texts: np.stack([e1, e2, e3, e4]) + + monitor = HealthMonitor( + registry=registry, + store=store, + mitosis_service=mitosis_svc, + merging_service=merging_svc, + embedder=embedder, + mitosis_threshold=20_000, + ) + + # Patch store.read to return a fake bucket with 2 chunks + fake_chunk = MagicMock() + fake_chunk.source_file = "/nonexistent/file.py" + fake_chunk.start_line = 1 + fake_chunk.end_line = 10 + fake_fm = MagicMock() + fake_fm.chunks = [fake_chunk, fake_chunk, fake_chunk, fake_chunk] + + with patch.object(store, "read", return_value=(fake_fm, "prose")): + await monitor._check() + + mitosis_svc.split.assert_called_once_with(bucket_id) + + +# --------------------------------------------------------------------------- +# 6-F: MergingService merge trigger +# --------------------------------------------------------------------------- + +class TestMergingService: + @pytest.mark.asyncio + async def test_merge_fires_when_similar_and_small( + self, registry: BucketRegistry, store: BucketStore, tmp_path: Path + ) -> None: + """Two buckets with high centroid similarity and low combined tokens must be merged.""" + dim = 8 + # Both centroids point in nearly the same direction → similarity > MERGE_SIMILARITY + c1 = np.array([1.0, 0.01, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0], dtype=np.float32) + c2 = np.array([1.0, 0.01, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0], dtype=np.float32) + c1 /= np.linalg.norm(c1) + c2 /= np.linalg.norm(c2) + + await _seed(registry, "bucket_a", 3_000, c1) + await _seed(registry, "bucket_b", 4_000, c2) + + agent = _mock_agent() + svc = MergingService( + registry=registry, + store=store, + agent=agent, + embedder=_mock_embedder(dim), + strategy=_mock_strategy(), + ) + + # Verify _should_merge returns True for this pair + centroids = registry.get_all_centroids() + recent: set = set() + assert svc._should_merge("bucket_a", "bucket_b", centroids, recent), ( + "Expected _should_merge to return True for similar, small buckets" + ) + + @pytest.mark.asyncio + async def test_merge_blocked_by_token_limit(self, registry: BucketRegistry, store: BucketStore) -> None: + """Two large buckets must not be merged even if similar.""" + dim = 8 + c = np.ones(dim, dtype=np.float32) / np.sqrt(dim) + await _seed(registry, "big_a", 10_000, c.copy()) + await _seed(registry, "big_b", 10_000, c.copy()) + + svc = MergingService( + registry=registry, + store=store, + agent=_mock_agent(), + embedder=_mock_embedder(dim), + strategy=_mock_strategy(), + ) + + centroids = registry.get_all_centroids() + assert not svc._should_merge("big_a", "big_b", centroids, set()), ( + "Merge should be blocked: combined tokens >= MERGE_TOKEN_LIMIT" + ) + + @pytest.mark.asyncio + async def test_merge_blocked_by_anti_cycle(self, registry: BucketRegistry, store: BucketStore) -> None: + """A bucket in recent merge_history must not be merged again within 1 hour.""" + from datetime import datetime, timezone + + dim = 8 + c = np.ones(dim, dtype=np.float32) / np.sqrt(dim) + await _seed(registry, "cycle_a", 1_000, c.copy()) + await _seed(registry, "cycle_b", 1_000, c.copy()) + + # Record a recent merge involving cycle_a + registry._meta["merge_history"] = [ + { + "merged_bucket_ids": ["cycle_a", "cycle_x"], + "result_bucket_id": "cycle_a", + "merged_at": datetime.now(timezone.utc).isoformat(), + } + ] + + svc = MergingService( + registry=registry, + store=store, + agent=_mock_agent(), + embedder=_mock_embedder(dim), + strategy=_mock_strategy(), + ) + + centroids = registry.get_all_centroids() + recent = svc._recent_merged_ids() + assert not svc._should_merge("cycle_a", "cycle_b", centroids, recent), ( + "Merge should be blocked by anti-cycle guard" + ) + + @pytest.mark.asyncio + async def test_no_merge_when_dissimilar(self, registry: BucketRegistry, store: BucketStore) -> None: + """Orthogonal centroids must not be merged.""" + dim = 8 + c1 = np.zeros(dim, dtype=np.float32); c1[0] = 1.0 + c2 = np.zeros(dim, dtype=np.float32); c2[1] = 1.0 + + await _seed(registry, "orth_a", 1_000, c1) + await _seed(registry, "orth_b", 1_000, c2) + + svc = MergingService( + registry=registry, + store=store, + agent=_mock_agent(), + embedder=_mock_embedder(dim), + strategy=_mock_strategy(), + ) + + centroids = registry.get_all_centroids() + assert not svc._should_merge("orth_a", "orth_b", centroids, set()), ( + "Orthogonal centroids should not be merged" + ) diff --git a/tests/unit/test_bucket_registry.py b/tests/unit/test_bucket_registry.py index 35bdbf0..d7d8fbf 100644 --- a/tests/unit/test_bucket_registry.py +++ b/tests/unit/test_bucket_registry.py @@ -171,6 +171,101 @@ async def test_save_then_register_then_reload(self, registry_path: Path): assert "bucket-a" in r2.get_all_centroids() assert "bucket-b" not in r2.get_all_centroids() + async def test_save_includes_meta_block( + self, registry: BucketRegistry, registry_path: Path + ): + await registry.register("bucket-a", _centroid(), token_count=10) + registry.save() + data = json.loads(registry_path.read_text()) + assert "_meta" in data + assert data["_meta"]["schema_version"] == 2 + # Bucket entries must still be present and accessible + assert "bucket-a" in data + + async def test_load_ignores_meta_block_as_bucket( + self, registry: BucketRegistry, registry_path: Path + ): + """The _meta block must not be interpreted as a bucket entry.""" + await registry.register("bucket-a", _centroid(), token_count=10) + registry.save() + + r2 = BucketRegistry(registry_path) + r2.load() + centroids = r2.get_all_centroids() + assert "_meta" not in centroids + assert "bucket-a" in centroids + + async def test_load_handles_legacy_format_without_meta(self, registry_path: Path): + """Old registry.json files (no _meta key) must load without error.""" + legacy = { + "bucket-a": { + "centroid_embedding": base64.b64encode( + _centroid(0.5).tobytes() + ).decode(), + "token_count": 42, + "is_splitting": False, + } + } + registry_path.parent.mkdir(parents=True, exist_ok=True) + registry_path.write_text(json.dumps(legacy), encoding="utf-8") + + r = BucketRegistry(registry_path) + r.load() + assert "bucket-a" in r.get_all_centroids() + assert r.get_token_count("bucket-a") == 42 + + async def test_index_timestamp_round_trips(self, registry_path: Path): + r1 = BucketRegistry(registry_path) + await r1.register("bucket-a", _centroid(), token_count=100) + r1.update_index_timestamp("bucket-a", "2026-04-02T10:00:00+00:00", "abc123def456") + r1.save() + + r2 = BucketRegistry(registry_path) + r2.load() + # Access internal entry to verify fields survived round-trip + entry = r2._buckets["bucket-a"] + assert entry.last_indexed_at == "2026-04-02T10:00:00+00:00" + assert entry.index_head_sha == "abc123def456" + + async def test_new_bucket_fields_default_to_none_after_register( + self, registry: BucketRegistry + ): + await registry.register("bucket-a", _centroid(), token_count=100) + entry = registry._buckets["bucket-a"] + assert entry.last_indexed_at is None + assert entry.index_head_sha is None + assert entry.coherence_score is None + assert entry.query_hit_count == 0 + assert entry.last_query_hit_at is None + + +# --------------------------------------------------------------------------- +# update_index_timestamp() +# --------------------------------------------------------------------------- + +class TestIndexTimestamp: + async def test_sets_fields_on_registered_bucket(self, registry: BucketRegistry): + await registry.register("bucket-a", _centroid(), token_count=100) + registry.update_index_timestamp("bucket-a", "2026-04-02T10:00:00+00:00", "sha123") + entry = registry._buckets["bucket-a"] + assert entry.last_indexed_at == "2026-04-02T10:00:00+00:00" + assert entry.index_head_sha == "sha123" + + async def test_unknown_bucket_raises_key_error(self, registry: BucketRegistry): + with pytest.raises(KeyError): + registry.update_index_timestamp("ghost", "2026-04-02T10:00:00+00:00", "sha123") + + async def test_re_register_preserves_timestamp(self, registry: BucketRegistry): + """Re-registering a bucket to update its centroid must not wipe the timestamp.""" + await registry.register("bucket-a", _centroid(0.1), token_count=100) + registry.update_index_timestamp("bucket-a", "2026-04-02T10:00:00+00:00", "sha123") + # Re-register with a new centroid (simulates Librarian recomputing after update) + await registry.register("bucket-a", _centroid(0.9), token_count=200) + entry = registry._buckets["bucket-a"] + # Centroid and token_count updated; timestamp preserved + assert entry.last_indexed_at == "2026-04-02T10:00:00+00:00" + assert entry.index_head_sha == "sha123" + # --------------------------------------------------------------------------- # set_splitting() / is_splitting() diff --git a/tests/unit/test_git_hook_receiver.py b/tests/unit/test_git_hook_receiver.py new file mode 100644 index 0000000..bddab37 --- /dev/null +++ b/tests/unit/test_git_hook_receiver.py @@ -0,0 +1,240 @@ +"""Unit tests for libucks.git_hook_receiver. + +Tests: + - install_hooks: creates new hook files with shebang + trigger line + - install_hooks: appends to existing hook files without overwriting + - install_hooks: idempotent (does not duplicate the trigger line) + - serve_socket / IPC: payload sent over the Unix socket is dispatched to on_event +""" +from __future__ import annotations + +import asyncio +import json +import os +import socket +import stat +import tempfile +from pathlib import Path + +import pytest + +from libucks.git_hook_receiver import _HOOK_EVENTS, _HOOK_LINE, install_hooks, serve_socket + + +@pytest.fixture() +def short_sock_path() -> Path: + """Return a short Unix socket path that fits within macOS's 104-char AF_UNIX limit.""" + # Use /tmp with a short unique suffix — pytest's tmp_path is often too long. + with tempfile.NamedTemporaryFile(suffix=".sock", dir="/tmp", delete=True) as f: + p = Path(f.name) + # File was deleted; return the path so server can create it + return p + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture() +def fake_repo(tmp_path: Path) -> Path: + """A minimal directory tree that looks like a git repo (.git/hooks/).""" + (tmp_path / ".git" / "hooks").mkdir(parents=True) + return tmp_path + + +# --------------------------------------------------------------------------- +# install_hooks — filesystem tests +# --------------------------------------------------------------------------- + +class TestInstallHooks: + def test_creates_new_hook_files(self, fake_repo: Path) -> None: + modified = install_hooks(fake_repo) + + assert set(modified) == set(_HOOK_EVENTS) + hooks_dir = fake_repo / ".git" / "hooks" + for event in _HOOK_EVENTS: + hook_file = hooks_dir / event + assert hook_file.exists(), f"{event} hook not created" + content = hook_file.read_text() + assert content.startswith("#!/bin/sh\n"), f"{event} missing shebang" + assert _HOOK_LINE.format(event=event) in content + + def test_new_hook_files_are_executable(self, fake_repo: Path) -> None: + install_hooks(fake_repo) + hooks_dir = fake_repo / ".git" / "hooks" + for event in _HOOK_EVENTS: + hook_file = hooks_dir / event + mode = hook_file.stat().st_mode + assert mode & stat.S_IXUSR, f"{event} hook is not executable" + + def test_appends_to_existing_hook(self, fake_repo: Path) -> None: + hooks_dir = fake_repo / ".git" / "hooks" + existing_content = "#!/bin/sh\necho 'existing hook'\n" + hook_file = hooks_dir / "post-commit" + hook_file.write_text(existing_content) + + install_hooks(fake_repo) + + result = hook_file.read_text() + assert "echo 'existing hook'" in result, "existing content was erased" + assert _HOOK_LINE.format(event="post-commit") in result, "trigger not appended" + # Shebang should appear only once + assert result.count("#!/bin/sh") == 1 + + def test_idempotent_does_not_double_append(self, fake_repo: Path) -> None: + install_hooks(fake_repo) + install_hooks(fake_repo) # second call should be a no-op + + hooks_dir = fake_repo / ".git" / "hooks" + for event in _HOOK_EVENTS: + content = (hooks_dir / event).read_text() + trigger = _HOOK_LINE.format(event=event) + assert content.count(trigger) == 1, f"trigger duplicated for {event}" + + def test_returns_empty_list_when_already_installed(self, fake_repo: Path) -> None: + install_hooks(fake_repo) + second_pass = install_hooks(fake_repo) + assert second_pass == [], "expected [] on no-op second install" + + def test_creates_hooks_dir_if_missing(self, tmp_path: Path) -> None: + """install_hooks creates .git/hooks/ if it does not exist.""" + (tmp_path / ".git").mkdir() + # hooks dir intentionally not created + modified = install_hooks(tmp_path) + assert set(modified) == set(_HOOK_EVENTS) + + +# --------------------------------------------------------------------------- +# serve_socket / IPC — real Unix socket, async +# --------------------------------------------------------------------------- + +class TestServeSocket: + async def _send_payload(self, sock_path: Path, payload: dict) -> None: + """Helper: open the socket, send JSON, close.""" + # Wait until the server has bound a real Unix socket (not just any file). + for _ in range(40): + if sock_path.exists() and stat.S_ISSOCK(sock_path.stat().st_mode): + break + await asyncio.sleep(0.05) + + loop = asyncio.get_running_loop() + await loop.run_in_executor( + None, + lambda: _sync_send(sock_path, payload), + ) + + @pytest.mark.asyncio + async def test_dispatches_event_to_on_event(self, short_sock_path: Path) -> None: + received: list[dict] = [] + + async def on_event(payload: dict) -> None: + received.append(payload) + + server_task = asyncio.ensure_future(serve_socket(short_sock_path, on_event)) + try: + await self._send_payload(short_sock_path, {"event": "post-commit"}) + # Give the server a moment to invoke on_event + await asyncio.sleep(0.1) + assert received == [{"event": "post-commit"}] + finally: + server_task.cancel() + try: + await server_task + except asyncio.CancelledError: + pass + + @pytest.mark.asyncio + async def test_handles_multiple_connections(self, short_sock_path: Path) -> None: + received: list[dict] = [] + + async def on_event(payload: dict) -> None: + received.append(payload) + + server_task = asyncio.ensure_future(serve_socket(short_sock_path, on_event)) + try: + for event_name in ["post-commit", "post-checkout", "post-rewrite"]: + await self._send_payload(short_sock_path, {"event": event_name}) + await asyncio.sleep(0.05) + + await asyncio.sleep(0.1) + events = [p["event"] for p in received] + assert "post-commit" in events + assert "post-checkout" in events + assert "post-rewrite" in events + finally: + server_task.cancel() + try: + await server_task + except asyncio.CancelledError: + pass + + @pytest.mark.asyncio + async def test_removes_stale_socket_on_start(self, short_sock_path: Path) -> None: + # Create a stale file at the socket path + short_sock_path.write_bytes(b"stale") + + received: list[dict] = [] + + async def on_event(payload: dict) -> None: + received.append(payload) + + server_task = asyncio.ensure_future(serve_socket(short_sock_path, on_event)) + try: + await self._send_payload(short_sock_path, {"event": "post-commit"}) + await asyncio.sleep(0.1) + assert len(received) == 1 + finally: + server_task.cancel() + try: + await server_task + except asyncio.CancelledError: + pass + + @pytest.mark.asyncio + async def test_ignores_malformed_json(self, short_sock_path: Path) -> None: + """Malformed JSON must not crash the server.""" + received: list[dict] = [] + + async def on_event(payload: dict) -> None: + received.append(payload) + + server_task = asyncio.ensure_future(serve_socket(short_sock_path, on_event)) + try: + # Wait for socket to be bound (not just any file) + for _ in range(40): + if short_sock_path.exists() and stat.S_ISSOCK(short_sock_path.stat().st_mode): + break + await asyncio.sleep(0.05) + + loop = asyncio.get_running_loop() + await loop.run_in_executor(None, lambda: _sync_send_raw(short_sock_path, b"not json!!")) + await asyncio.sleep(0.1) + + # Server still alive — send a valid event + await self._send_payload(short_sock_path, {"event": "post-commit"}) + await asyncio.sleep(0.1) + assert received == [{"event": "post-commit"}] + finally: + server_task.cancel() + try: + await server_task + except asyncio.CancelledError: + pass + + +# --------------------------------------------------------------------------- +# Sync helpers (run in executor to avoid blocking the event loop) +# --------------------------------------------------------------------------- + +def _sync_send(sock_path: Path, payload: dict) -> None: + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as s: + s.settimeout(3) + s.connect(str(sock_path)) + s.sendall(json.dumps(payload).encode()) + + +def _sync_send_raw(sock_path: Path, data: bytes) -> None: + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as s: + s.settimeout(3) + s.connect(str(sock_path)) + s.sendall(data) diff --git a/tests/unit/test_models.py b/tests/unit/test_models.py index 31c3347..8bc4086 100644 --- a/tests/unit/test_models.py +++ b/tests/unit/test_models.py @@ -80,7 +80,10 @@ def test_round_trip_dict(self): data = _make_chunk() chunk = ChunkMetadata(**data) result = chunk.model_dump() - assert result == data + # indexed_at defaults to None — present in dump but not in the helper dict + assert result["chunk_id"] == data["chunk_id"] + assert result["git_sha"] == data["git_sha"] + assert result["indexed_at"] is None def test_round_trip_json(self): data = _make_chunk() @@ -88,6 +91,23 @@ def test_round_trip_json(self): restored = ChunkMetadata.model_validate_json(chunk.model_dump_json()) assert restored == chunk + def test_indexed_at_defaults_to_none(self): + chunk = ChunkMetadata(**_make_chunk()) + assert chunk.indexed_at is None + + def test_indexed_at_round_trip_dict(self): + data = _make_chunk(indexed_at="2026-04-02T10:00:00+00:00") + chunk = ChunkMetadata(**data) + assert chunk.indexed_at == "2026-04-02T10:00:00+00:00" + restored = ChunkMetadata.model_validate_json(chunk.model_dump_json()) + assert restored.indexed_at == "2026-04-02T10:00:00+00:00" + + def test_old_chunk_without_indexed_at_deserialises(self): + """Chunks persisted before Phase 6-A (no indexed_at key) must load cleanly.""" + data = _make_chunk() # helper produces no indexed_at key + chunk = ChunkMetadata(**data) + assert chunk.indexed_at is None + def test_missing_chunk_id_raises(self): data = _make_chunk() del data["chunk_id"] @@ -153,6 +173,37 @@ def test_round_trip_dict(self): assert result["token_count"] == data["token_count"] assert len(result["chunks"]) == 1 + def test_new_optional_fields_default_to_none_or_zero(self): + bfm = BucketFrontMatter(**_make_bucket_front_matter()) + assert bfm.last_indexed_at is None + assert bfm.index_head_sha is None + assert bfm.coherence_score is None + assert bfm.parent_bucket_id is None + assert bfm.generation == 0 + + def test_new_fields_round_trip_json(self): + data = _make_bucket_front_matter( + last_indexed_at="2026-04-02T10:00:00+00:00", + index_head_sha="e4f9a3b1c2d3ef45ab67cd89ef012345", + coherence_score=0.82, + parent_bucket_id="deadbeef", + generation=1, + ) + bfm = BucketFrontMatter(**data) + restored = BucketFrontMatter.model_validate_json(bfm.model_dump_json()) + assert restored.last_indexed_at == "2026-04-02T10:00:00+00:00" + assert restored.index_head_sha == "e4f9a3b1c2d3ef45ab67cd89ef012345" + assert restored.coherence_score == pytest.approx(0.82) + assert restored.parent_bucket_id == "deadbeef" + assert restored.generation == 1 + + def test_old_bucket_without_new_fields_deserialises(self): + """BFMs persisted before Phase 6-A (no new fields) must load cleanly.""" + data = _make_bucket_front_matter() + bfm = BucketFrontMatter(**data) + assert bfm.generation == 0 + assert bfm.last_indexed_at is None + def test_centroid_base64_round_trip(self): original = np.array([0.1, 0.2, 0.3, 0.4], dtype=np.float32) encoded = base64.b64encode(original.tobytes()).decode() diff --git a/tests/unit/test_stale_checker.py b/tests/unit/test_stale_checker.py new file mode 100644 index 0000000..116c172 --- /dev/null +++ b/tests/unit/test_stale_checker.py @@ -0,0 +1,629 @@ +"""Phase 6-C Testing Gate — test_stale_checker.py + +Tests the four-level StaleChecker protocol and the QueryOrchestrator +integration (reindex_fn fired, eventual-consistency answer returned). + +All OS and subprocess calls are patched at the module-level helper boundary +so the tests run without a real git repo or real files on disk. +""" +from __future__ import annotations + +import asyncio +import base64 +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch + +import numpy as np +import pytest + +from libucks.models.chunk import ChunkMetadata +from libucks.stale_checker import StaleCheckResult, StaleChecker +from libucks.storage.bucket_registry import BucketRegistry +from libucks.storage.bucket_store import BucketStore + +_MOD = "libucks.stale_checker" + +_BUCKET_ID = "deadbeef" +_OTHER_ID = "cafebabe" +_OLD_SHA = "a" * 40 +_NEW_SHA = "b" * 40 +_SOURCE_FILE = "/repo/libucks/auth.py" +_INDEXED_AT = "2026-04-02T10:00:00+00:00" +from datetime import datetime as _dt +_INDEXED_TS: float = _dt.fromisoformat(_INDEXED_AT).timestamp() # 1775124000.0 + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture +def tmp_registry(tmp_path: Path) -> BucketRegistry: + reg_path = tmp_path / ".libucks" / "registry.json" + reg_path.parent.mkdir(parents=True, exist_ok=True) + return BucketRegistry(reg_path) + + +@pytest.fixture +def tmp_store(tmp_path: Path) -> BucketStore: + return BucketStore(tmp_path / ".libucks" / "buckets") + + +def _make_centroid_b64() -> str: + return base64.b64encode(np.array([1.0, 0.0, 0.0], dtype=np.float32).tobytes()).decode() + + +async def _seed_bucket( + store: BucketStore, + registry: BucketRegistry, + bucket_id: str, + source_file: str = _SOURCE_FILE, + last_indexed_at: str | None = _INDEXED_AT, + index_head_sha: str | None = _OLD_SHA, +) -> None: + """Create a bucket with one chunk and populate the registry entry.""" + chunk = ChunkMetadata( + chunk_id="c001", + source_file=source_file, + start_line=1, + end_line=10, + git_sha=index_head_sha or "init", + token_count=50, + indexed_at=last_indexed_at, + ) + store.create( + bucket_id=bucket_id, + domain_label="test", + centroid=_make_centroid_b64(), + chunks=[chunk], + prose="prose", + ) + await registry.register(bucket_id, np.array([1.0, 0.0, 0.0], dtype=np.float32), 50) + registry.update_index_timestamp(bucket_id, last_indexed_at or "", index_head_sha or "") + + +def _make_checker( + registry: BucketRegistry, + store: BucketStore, + repo_path: Path, +) -> StaleChecker: + return StaleChecker(registry=registry, store=store, repo_path=repo_path) + + +# --------------------------------------------------------------------------- +# Level 1 — watcher process alive? +# --------------------------------------------------------------------------- + +class TestLevel1WatcherProcess: + async def test_dead_watcher_pid_returns_stale_level_1( + self, tmp_path, tmp_registry, tmp_store + ): + tmp_registry._meta["watcher_pid"] = 99999 # fake PID + await _seed_bucket(tmp_store, tmp_registry, _BUCKET_ID) + checker = _make_checker(tmp_registry, tmp_store, tmp_path) + + with patch(f"{_MOD}._process_is_alive", return_value=False), \ + patch(f"{_MOD}._get_current_head", return_value=_OLD_SHA): + result = await checker.check([_BUCKET_ID]) + + assert result.is_stale is True + assert result.level == 1 + assert _BUCKET_ID in result.stale_bucket_ids + + async def test_alive_watcher_pid_does_not_trigger_level_1( + self, tmp_path, tmp_registry, tmp_store + ): + tmp_registry._meta["watcher_pid"] = 12345 + tmp_registry._meta["last_indexed_head"] = _OLD_SHA + await _seed_bucket(tmp_store, tmp_registry, _BUCKET_ID) + checker = _make_checker(tmp_registry, tmp_store, tmp_path) + + with patch(f"{_MOD}._process_is_alive", return_value=True), \ + patch(f"{_MOD}._get_current_head", return_value=_OLD_SHA), \ + patch(f"{_MOD}._get_file_mtime", return_value=_INDEXED_TS - 1): + result = await checker.check([_BUCKET_ID]) + + assert result.is_stale is False + assert result.level == 0 + + async def test_no_watcher_pid_in_meta_skips_level_1( + self, tmp_path, tmp_registry, tmp_store + ): + """If watcher_pid is None (never set), Level 1 is skipped entirely.""" + assert tmp_registry._meta.get("watcher_pid") is None + tmp_registry._meta["last_indexed_head"] = _OLD_SHA + await _seed_bucket(tmp_store, tmp_registry, _BUCKET_ID) + checker = _make_checker(tmp_registry, tmp_store, tmp_path) + + with patch(f"{_MOD}._process_is_alive") as mock_kill, \ + patch(f"{_MOD}._get_current_head", return_value=_OLD_SHA), \ + patch(f"{_MOD}._get_file_mtime", return_value=_INDEXED_TS - 1): + result = await checker.check([_BUCKET_ID]) + + mock_kill.assert_not_called() + assert result.is_stale is False + + async def test_level_1_flags_all_queried_buckets( + self, tmp_path, tmp_registry, tmp_store + ): + tmp_registry._meta["watcher_pid"] = 99999 + await _seed_bucket(tmp_store, tmp_registry, _BUCKET_ID) + await _seed_bucket(tmp_store, tmp_registry, _OTHER_ID, source_file="/repo/other.py") + checker = _make_checker(tmp_registry, tmp_store, tmp_path) + + with patch(f"{_MOD}._process_is_alive", return_value=False): + result = await checker.check([_BUCKET_ID, _OTHER_ID]) + + assert result.is_stale is True + assert set(result.stale_bucket_ids) == {_BUCKET_ID, _OTHER_ID} + + +# --------------------------------------------------------------------------- +# Level 2 — git HEAD drift +# --------------------------------------------------------------------------- + +class TestLevel2HeadDrift: + async def test_head_changed_returns_stale_level_2( + self, tmp_path, tmp_registry, tmp_store + ): + tmp_registry._meta["last_indexed_head"] = _OLD_SHA + await _seed_bucket(tmp_store, tmp_registry, _BUCKET_ID) + checker = _make_checker(tmp_registry, tmp_store, tmp_path) + + with patch(f"{_MOD}._process_is_alive", return_value=True), \ + patch(f"{_MOD}._get_current_head", return_value=_NEW_SHA): + result = await checker.check([_BUCKET_ID]) + + assert result.is_stale is True + assert result.level == 2 + assert _BUCKET_ID in result.stale_bucket_ids + + async def test_head_unchanged_does_not_trigger_level_2( + self, tmp_path, tmp_registry, tmp_store + ): + tmp_registry._meta["last_indexed_head"] = _OLD_SHA + await _seed_bucket(tmp_store, tmp_registry, _BUCKET_ID) + checker = _make_checker(tmp_registry, tmp_store, tmp_path) + + with patch(f"{_MOD}._process_is_alive", return_value=True), \ + patch(f"{_MOD}._get_current_head", return_value=_OLD_SHA), \ + patch(f"{_MOD}._get_file_mtime", return_value=_INDEXED_TS - 1): + result = await checker.check([_BUCKET_ID]) + + assert result.level != 2 + + async def test_level_2_flags_all_queried_buckets( + self, tmp_path, tmp_registry, tmp_store + ): + tmp_registry._meta["last_indexed_head"] = _OLD_SHA + await _seed_bucket(tmp_store, tmp_registry, _BUCKET_ID) + await _seed_bucket(tmp_store, tmp_registry, _OTHER_ID, source_file="/repo/other.py") + checker = _make_checker(tmp_registry, tmp_store, tmp_path) + + with patch(f"{_MOD}._process_is_alive", return_value=True), \ + patch(f"{_MOD}._get_current_head", return_value=_NEW_SHA): + result = await checker.check([_BUCKET_ID, _OTHER_ID]) + + assert result.is_stale is True + assert set(result.stale_bucket_ids) == {_BUCKET_ID, _OTHER_ID} + + async def test_no_last_indexed_head_skips_level_2( + self, tmp_path, tmp_registry, tmp_store + ): + """If last_indexed_head is None (never set), Level 2 cannot fire.""" + assert tmp_registry._meta.get("last_indexed_head") is None + await _seed_bucket(tmp_store, tmp_registry, _BUCKET_ID) + checker = _make_checker(tmp_registry, tmp_store, tmp_path) + + with patch(f"{_MOD}._process_is_alive", return_value=True), \ + patch(f"{_MOD}._get_current_head", return_value=_NEW_SHA), \ + patch(f"{_MOD}._get_file_mtime", return_value=_INDEXED_TS - 1): + result = await checker.check([_BUCKET_ID]) + + # Level 2 did not fire (no baseline to compare against) + assert result.level != 2 + + async def test_git_unavailable_skips_level_2( + self, tmp_path, tmp_registry, tmp_store + ): + tmp_registry._meta["last_indexed_head"] = _OLD_SHA + await _seed_bucket(tmp_store, tmp_registry, _BUCKET_ID) + checker = _make_checker(tmp_registry, tmp_store, tmp_path) + + with patch(f"{_MOD}._process_is_alive", return_value=True), \ + patch(f"{_MOD}._get_current_head", return_value=None), \ + patch(f"{_MOD}._get_file_mtime", return_value=_INDEXED_TS - 1): + result = await checker.check([_BUCKET_ID]) + + # git unavailable → Level 2 skipped, still fresh + assert result.level != 2 + + +# --------------------------------------------------------------------------- +# Level 3 — file mtime vs last_indexed_at +# --------------------------------------------------------------------------- + +class TestLevel3FileMtime: + async def test_file_newer_than_index_triggers_level_3( + self, tmp_path, tmp_registry, tmp_store + ): + tmp_registry._meta["last_indexed_head"] = _OLD_SHA + await _seed_bucket(tmp_store, tmp_registry, _BUCKET_ID, last_indexed_at=_INDEXED_AT) + checker = _make_checker(tmp_registry, tmp_store, tmp_path) + + with patch(f"{_MOD}._process_is_alive", return_value=True), \ + patch(f"{_MOD}._get_current_head", return_value=_OLD_SHA), \ + patch(f"{_MOD}._get_file_mtime", return_value=_INDEXED_TS + 60): # 1 minute newer + result = await checker.check([_BUCKET_ID]) + + assert result.is_stale is True + assert result.level == 3 + assert _BUCKET_ID in result.stale_bucket_ids + + async def test_file_older_than_index_is_fresh( + self, tmp_path, tmp_registry, tmp_store + ): + tmp_registry._meta["last_indexed_head"] = _OLD_SHA + await _seed_bucket(tmp_store, tmp_registry, _BUCKET_ID, last_indexed_at=_INDEXED_AT) + checker = _make_checker(tmp_registry, tmp_store, tmp_path) + + with patch(f"{_MOD}._process_is_alive", return_value=True), \ + patch(f"{_MOD}._get_current_head", return_value=_OLD_SHA), \ + patch(f"{_MOD}._get_file_mtime", return_value=_INDEXED_TS - 60): # older + result = await checker.check([_BUCKET_ID]) + + assert result.is_stale is False + + async def test_missing_last_indexed_at_skips_level_3( + self, tmp_path, tmp_registry, tmp_store + ): + """Bucket with no last_indexed_at (pre-Phase-6-A) skips mtime check.""" + tmp_registry._meta["last_indexed_head"] = _OLD_SHA + await _seed_bucket( + tmp_store, tmp_registry, _BUCKET_ID, + last_indexed_at=None, index_head_sha=_OLD_SHA, + ) + checker = _make_checker(tmp_registry, tmp_store, tmp_path) + + with patch(f"{_MOD}._process_is_alive", return_value=True), \ + patch(f"{_MOD}._get_current_head", return_value=_OLD_SHA), \ + patch(f"{_MOD}._get_file_mtime") as mock_mtime: + result = await checker.check([_BUCKET_ID]) + + mock_mtime.assert_not_called() + assert result.is_stale is False + + async def test_stat_failure_on_deleted_file_does_not_trigger_stale( + self, tmp_path, tmp_registry, tmp_store + ): + """If a source file was deleted (stat returns None), skip that file.""" + tmp_registry._meta["last_indexed_head"] = _OLD_SHA + await _seed_bucket(tmp_store, tmp_registry, _BUCKET_ID, last_indexed_at=_INDEXED_AT) + checker = _make_checker(tmp_registry, tmp_store, tmp_path) + + with patch(f"{_MOD}._process_is_alive", return_value=True), \ + patch(f"{_MOD}._get_current_head", return_value=_OLD_SHA), \ + patch(f"{_MOD}._get_file_mtime", return_value=None): # deleted + result = await checker.check([_BUCKET_ID]) + + assert result.is_stale is False + + async def test_level_3_deduplicates_source_files_across_chunks( + self, tmp_path, tmp_registry, tmp_store + ): + """Two chunks from the same file should call _get_file_mtime only once.""" + from libucks.models.bucket import BucketFrontMatter + + chunk_a = ChunkMetadata( + chunk_id="c001", source_file=_SOURCE_FILE, + start_line=1, end_line=5, git_sha=_OLD_SHA, token_count=10, + indexed_at=_INDEXED_AT, + ) + chunk_b = ChunkMetadata( + chunk_id="c002", source_file=_SOURCE_FILE, # same file + start_line=6, end_line=10, git_sha=_OLD_SHA, token_count=10, + indexed_at=_INDEXED_AT, + ) + tmp_store.create( + bucket_id=_BUCKET_ID, + domain_label="test", + centroid=_make_centroid_b64(), + chunks=[chunk_a, chunk_b], + prose="prose", + ) + await tmp_registry.register(_BUCKET_ID, np.array([1.0, 0.0, 0.0], dtype=np.float32), 20) + tmp_registry.update_index_timestamp(_BUCKET_ID, _INDEXED_AT, _OLD_SHA) + tmp_registry._meta["last_indexed_head"] = _OLD_SHA + + checker = _make_checker(tmp_registry, tmp_store, tmp_path) + + with patch(f"{_MOD}._process_is_alive", return_value=True), \ + patch(f"{_MOD}._get_current_head", return_value=_OLD_SHA), \ + patch(f"{_MOD}._get_file_mtime", return_value=_INDEXED_TS - 1) as mock_mtime: + await checker.check([_BUCKET_ID]) + + # Called once for the deduplicated source_file, not once per chunk + mock_mtime.assert_called_once_with(_SOURCE_FILE) + + +# --------------------------------------------------------------------------- +# Level 4 — per-bucket index_head_sha drift +# --------------------------------------------------------------------------- + +class TestLevel4BucketShaDrift: + async def test_bucket_sha_behind_current_head_triggers_level_4( + self, tmp_path, tmp_registry, tmp_store + ): + tmp_registry._meta["last_indexed_head"] = _NEW_SHA # global HEAD is new + await _seed_bucket( + tmp_store, tmp_registry, _BUCKET_ID, + last_indexed_at=_INDEXED_AT, index_head_sha=_OLD_SHA, # bucket is behind + ) + checker = _make_checker(tmp_registry, tmp_store, tmp_path) + + with patch(f"{_MOD}._process_is_alive", return_value=True), \ + patch(f"{_MOD}._get_current_head", return_value=_NEW_SHA), \ + patch(f"{_MOD}._get_file_mtime", return_value=_INDEXED_TS - 1): + result = await checker.check([_BUCKET_ID]) + + assert result.is_stale is True + assert result.level == 4 + assert _BUCKET_ID in result.stale_bucket_ids + + async def test_bucket_sha_matches_current_head_is_fresh( + self, tmp_path, tmp_registry, tmp_store + ): + tmp_registry._meta["last_indexed_head"] = _NEW_SHA + await _seed_bucket( + tmp_store, tmp_registry, _BUCKET_ID, + last_indexed_at=_INDEXED_AT, index_head_sha=_NEW_SHA, + ) + checker = _make_checker(tmp_registry, tmp_store, tmp_path) + + with patch(f"{_MOD}._process_is_alive", return_value=True), \ + patch(f"{_MOD}._get_current_head", return_value=_NEW_SHA), \ + patch(f"{_MOD}._get_file_mtime", return_value=_INDEXED_TS - 1): + result = await checker.check([_BUCKET_ID]) + + assert result.is_stale is False + + async def test_bucket_with_none_index_sha_skips_level_4( + self, tmp_path, tmp_registry, tmp_store + ): + """Buckets without index_head_sha (pre-Phase-6-A) skip Level 4.""" + tmp_registry._meta["last_indexed_head"] = _NEW_SHA + await _seed_bucket( + tmp_store, tmp_registry, _BUCKET_ID, + last_indexed_at=_INDEXED_AT, index_head_sha=None, + ) + # Manually clear the index_head_sha on the entry (seed sets it to "") + tmp_registry._buckets[_BUCKET_ID].index_head_sha = None + checker = _make_checker(tmp_registry, tmp_store, tmp_path) + + with patch(f"{_MOD}._process_is_alive", return_value=True), \ + patch(f"{_MOD}._get_current_head", return_value=_NEW_SHA), \ + patch(f"{_MOD}._get_file_mtime", return_value=_INDEXED_TS - 1): + result = await checker.check([_BUCKET_ID]) + + assert result.is_stale is False + + async def test_level_4_only_flags_outdated_bucket_not_fresh_one( + self, tmp_path, tmp_registry, tmp_store + ): + """Two buckets: one with current SHA, one with old SHA → only old one flagged.""" + tmp_registry._meta["last_indexed_head"] = _NEW_SHA + + await _seed_bucket(tmp_store, tmp_registry, _BUCKET_ID, index_head_sha=_OLD_SHA) + await _seed_bucket( + tmp_store, tmp_registry, _OTHER_ID, + source_file="/repo/other.py", index_head_sha=_NEW_SHA, + ) + checker = _make_checker(tmp_registry, tmp_store, tmp_path) + + with patch(f"{_MOD}._process_is_alive", return_value=True), \ + patch(f"{_MOD}._get_current_head", return_value=_NEW_SHA), \ + patch(f"{_MOD}._get_file_mtime", return_value=_INDEXED_TS - 1): + result = await checker.check([_BUCKET_ID, _OTHER_ID]) + + assert result.is_stale is True + assert _BUCKET_ID in result.stale_bucket_ids + assert _OTHER_ID not in result.stale_bucket_ids + + +# --------------------------------------------------------------------------- +# Level priority — higher levels don't fire when lower levels already returned +# --------------------------------------------------------------------------- + +class TestLevelPriority: + async def test_level_1_fires_before_level_2( + self, tmp_path, tmp_registry, tmp_store + ): + tmp_registry._meta["watcher_pid"] = 99999 + tmp_registry._meta["last_indexed_head"] = _OLD_SHA + await _seed_bucket(tmp_store, tmp_registry, _BUCKET_ID) + checker = _make_checker(tmp_registry, tmp_store, tmp_path) + + with patch(f"{_MOD}._process_is_alive", return_value=False), \ + patch(f"{_MOD}._get_current_head", return_value=_NEW_SHA) as mock_head: + result = await checker.check([_BUCKET_ID]) + + assert result.level == 1 + # Level 2 subprocess call should NOT have been made since Level 1 returned early + mock_head.assert_not_called() + + async def test_level_2_fires_before_level_3( + self, tmp_path, tmp_registry, tmp_store + ): + tmp_registry._meta["last_indexed_head"] = _OLD_SHA + await _seed_bucket(tmp_store, tmp_registry, _BUCKET_ID, last_indexed_at=_INDEXED_AT) + checker = _make_checker(tmp_registry, tmp_store, tmp_path) + + with patch(f"{_MOD}._process_is_alive", return_value=True), \ + patch(f"{_MOD}._get_current_head", return_value=_NEW_SHA), \ + patch(f"{_MOD}._get_file_mtime") as mock_mtime: + result = await checker.check([_BUCKET_ID]) + + assert result.level == 2 + # Level 3 mtime check should NOT have been made since Level 2 returned early + mock_mtime.assert_not_called() + + async def test_level_3_fires_before_level_4_for_same_bucket( + self, tmp_path, tmp_registry, tmp_store + ): + """If Level 3 already flagged a bucket, Level 4 is not checked for that bucket.""" + tmp_registry._meta["last_indexed_head"] = _NEW_SHA + await _seed_bucket( + tmp_store, tmp_registry, _BUCKET_ID, + last_indexed_at=_INDEXED_AT, index_head_sha=_OLD_SHA, + ) + checker = _make_checker(tmp_registry, tmp_store, tmp_path) + + # File is newer (triggers L3) AND sha is old (would trigger L4) + with patch(f"{_MOD}._process_is_alive", return_value=True), \ + patch(f"{_MOD}._get_current_head", return_value=_NEW_SHA), \ + patch(f"{_MOD}._get_file_mtime", return_value=_INDEXED_TS + 60): + result = await checker.check([_BUCKET_ID]) + + # Level 3 fires first + assert result.level == 3 + + +# --------------------------------------------------------------------------- +# Fresh result +# --------------------------------------------------------------------------- + +class TestFreshResult: + async def test_all_fresh_returns_is_stale_false( + self, tmp_path, tmp_registry, tmp_store + ): + tmp_registry._meta["last_indexed_head"] = _OLD_SHA + await _seed_bucket( + tmp_store, tmp_registry, _BUCKET_ID, + last_indexed_at=_INDEXED_AT, index_head_sha=_OLD_SHA, + ) + checker = _make_checker(tmp_registry, tmp_store, tmp_path) + + with patch(f"{_MOD}._process_is_alive", return_value=True), \ + patch(f"{_MOD}._get_current_head", return_value=_OLD_SHA), \ + patch(f"{_MOD}._get_file_mtime", return_value=_INDEXED_TS - 1): + result = await checker.check([_BUCKET_ID]) + + assert result.is_stale is False + assert result.level == 0 + assert result.stale_bucket_ids == [] + + async def test_empty_bucket_ids_returns_fresh( + self, tmp_path, tmp_registry, tmp_store + ): + tmp_registry._meta["last_indexed_head"] = _OLD_SHA + checker = _make_checker(tmp_registry, tmp_store, tmp_path) + + with patch(f"{_MOD}._process_is_alive", return_value=True), \ + patch(f"{_MOD}._get_current_head", return_value=_OLD_SHA): + result = await checker.check([]) + + assert result.is_stale is False + + +# --------------------------------------------------------------------------- +# QueryOrchestrator integration +# --------------------------------------------------------------------------- + +class TestQueryOrchestratorIntegration: + """Verify the orchestrator fires reindex_fn on stale and still returns results.""" + + def _make_orchestrator(self, stale_result: StaleCheckResult): + from libucks.query_orchestrator import QueryOrchestrator + from libucks.central_agent import CentralAgent + + # Minimal mocks — we just care about the stale-check branch + mock_agent = MagicMock(spec=CentralAgent) + mock_agent.route.return_value = [_BUCKET_ID] + + mock_librarian = AsyncMock() + mock_librarian.handle.return_value = "some representation" + + mock_checker = AsyncMock() + mock_checker.check.return_value = stale_result + + reindex_calls: list = [] + + async def reindex_fn(bucket_ids): + reindex_calls.append(bucket_ids) + + orch = QueryOrchestrator( + central_agent=mock_agent, + librarians={_BUCKET_ID: mock_librarian}, + embed_fn=lambda t: np.zeros(384, dtype=np.float32), + top_k=1, + stale_checker=mock_checker, + reindex_fn=reindex_fn, + ) + return orch, mock_checker, reindex_calls + + async def test_stale_result_fires_reindex_fn(self): + stale = StaleCheckResult( + is_stale=True, + stale_bucket_ids=[_BUCKET_ID], + level=3, + reason="test stale", + ) + orch, mock_checker, reindex_calls = self._make_orchestrator(stale) + + await orch.query("how does auth work?") + + # Give ensure_future time to schedule + await asyncio.sleep(0) + + assert len(reindex_calls) == 1 + assert reindex_calls[0] == [_BUCKET_ID] + + async def test_stale_result_still_returns_answer(self): + """Even when stale, the query must return results (eventual consistency).""" + stale = StaleCheckResult( + is_stale=True, + stale_bucket_ids=[_BUCKET_ID], + level=2, + reason="HEAD drift", + ) + orch, _, _ = self._make_orchestrator(stale) + + results = await orch.query("how does auth work?") + + assert results == ["some representation"] + + async def test_fresh_result_does_not_fire_reindex_fn(self): + fresh = StaleCheckResult( + is_stale=False, + stale_bucket_ids=[], + level=0, + reason="fresh", + ) + orch, mock_checker, reindex_calls = self._make_orchestrator(fresh) + + await orch.query("how does auth work?") + await asyncio.sleep(0) + + assert reindex_calls == [] + + async def test_no_stale_checker_query_still_works(self): + """QueryOrchestrator without a stale_checker must work as before.""" + from libucks.query_orchestrator import QueryOrchestrator + from libucks.central_agent import CentralAgent + + mock_agent = MagicMock(spec=CentralAgent) + mock_agent.route.return_value = [_BUCKET_ID] + mock_librarian = AsyncMock() + mock_librarian.handle.return_value = "answer" + + orch = QueryOrchestrator( + central_agent=mock_agent, + librarians={_BUCKET_ID: mock_librarian}, + embed_fn=lambda t: np.zeros(384, dtype=np.float32), + top_k=1, + stale_checker=None, + reindex_fn=None, + ) + + results = await orch.query("question?") + assert results == ["answer"] diff --git a/tests/unit/test_startup_recovery.py b/tests/unit/test_startup_recovery.py new file mode 100644 index 0000000..7f13fb9 --- /dev/null +++ b/tests/unit/test_startup_recovery.py @@ -0,0 +1,478 @@ +"""Phase 6-B Testing Gate — test_startup_recovery.py + +Tests the StartupRecovery logic that replays commits missed while the server +was offline. All git I/O is mocked at the module-level helper boundary so +the tests run without a real git repo. +""" +from __future__ import annotations + +import asyncio +import base64 +from pathlib import Path +from typing import List +from unittest.mock import AsyncMock, MagicMock, call, patch + +import numpy as np +import pytest + +from libucks.diff.diff_extractor import DiffExtractor +from libucks.librarian import Librarian +from libucks.models.chunk import ChunkMetadata +from libucks.models.events import DiffEvent, DiffHunk, UpdateEvent +from libucks.startup_recovery import StartupRecovery, _git_diff_name_only, _git_rev_parse_head +from libucks.storage.bucket_registry import BucketRegistry +from libucks.storage.bucket_store import BucketStore + + +# --------------------------------------------------------------------------- +# Shared constants +# --------------------------------------------------------------------------- + +_OLD_SHA = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" +_NEW_SHA = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" +_BUCKET_ID = "deadbeef" + +# A .py file tracked by the watcher extension set +_TRACKED_FILE = "libucks/auth.py" +# A .txt file NOT in the tracked extension set +_UNTRACKED_FILE = "README.txt" + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture +def tmp_registry(tmp_path: Path) -> BucketRegistry: + reg_path = tmp_path / ".libucks" / "registry.json" + reg_path.parent.mkdir(parents=True, exist_ok=True) + return BucketRegistry(reg_path) + + +@pytest.fixture +def tmp_store(tmp_path: Path) -> BucketStore: + return BucketStore(tmp_path / ".libucks" / "buckets") + + +def _make_centroid_b64() -> str: + arr = np.array([1.0, 0.0, 0.0], dtype=np.float32) + return base64.b64encode(arr.tobytes()).decode() + + +def _seed_bucket(store: BucketStore, bucket_id: str, source_file: str) -> None: + """Create a minimal bucket file whose single chunk points at source_file.""" + chunk = ChunkMetadata( + chunk_id="c001", + source_file=source_file, + start_line=1, + end_line=10, + git_sha=_OLD_SHA, + token_count=50, + ) + store.create( + bucket_id=bucket_id, + domain_label="test domain", + centroid=_make_centroid_b64(), + chunks=[chunk], + prose="Some prose.", + ) + + +def _make_mock_librarian() -> AsyncMock: + """Return an AsyncMock that mimics Librarian.handle().""" + lib = AsyncMock(spec=Librarian) + lib.bucket_id = _BUCKET_ID + return lib + + +def _make_mock_extractor(events: List[DiffEvent]) -> MagicMock: + """Return a MagicMock DiffExtractor whose extract_between() returns *events*.""" + ext = MagicMock(spec=DiffExtractor) + ext.extract_between.return_value = events + return ext + + +def _make_diff_event(filepath: str, added_lines: List[str] | None = None) -> DiffEvent: + hunk = DiffHunk( + file=filepath, + old_start=1, + old_end=5, + new_start=1, + new_end=6, + added_lines=added_lines or ["def new_func(): pass"], + removed_lines=[], + ) + return DiffEvent(file=filepath, hunks=[hunk], is_rename=False) + + +def _make_recovery( + tmp_path: Path, + registry: BucketRegistry, + store: BucketStore, + librarians: dict, + extractor: MagicMock, +) -> StartupRecovery: + return StartupRecovery( + repo_path=tmp_path, + registry=registry, + store=store, + librarians=librarians, + extractor=extractor, + ) + + +# --------------------------------------------------------------------------- +# Helper to run async tests without friction +# --------------------------------------------------------------------------- + +_MOD = "libucks.startup_recovery" + + +# --------------------------------------------------------------------------- +# Tests: run() return value and baseline recording +# --------------------------------------------------------------------------- + +class TestRunReturnValue: + async def test_returns_current_head_when_up_to_date( + self, tmp_path, tmp_registry, tmp_store + ): + tmp_registry._meta["last_indexed_head"] = _NEW_SHA + recovery = _make_recovery(tmp_path, tmp_registry, tmp_store, {}, MagicMock()) + + with patch(f"{_MOD}._git_rev_parse_head", return_value=_NEW_SHA), \ + patch(f"{_MOD}._git_diff_name_only", return_value=[]): + result = await recovery.run() + + assert result == _NEW_SHA + + async def test_returns_none_when_git_unavailable( + self, tmp_path, tmp_registry, tmp_store + ): + recovery = _make_recovery(tmp_path, tmp_registry, tmp_store, {}, MagicMock()) + + with patch(f"{_MOD}._git_rev_parse_head", return_value=None): + result = await recovery.run() + + assert result is None + + async def test_returns_current_head_when_no_baseline( + self, tmp_path, tmp_registry, tmp_store + ): + # last_indexed_head is None (first boot after init) + assert tmp_registry._meta.get("last_indexed_head") is None + recovery = _make_recovery(tmp_path, tmp_registry, tmp_store, {}, MagicMock()) + + with patch(f"{_MOD}._git_rev_parse_head", return_value=_NEW_SHA), \ + patch(f"{_MOD}._git_diff_name_only") as mock_diff: + result = await recovery.run() + + assert result == _NEW_SHA + # No diff call when there is no baseline to compare against + mock_diff.assert_not_called() + + async def test_returns_current_head_after_successful_recovery( + self, tmp_path, tmp_registry, tmp_store + ): + tmp_registry._meta["last_indexed_head"] = _OLD_SHA + _seed_bucket(tmp_store, _BUCKET_ID, str(tmp_path / _TRACKED_FILE)) + lib = _make_mock_librarian() + extractor = _make_mock_extractor([_make_diff_event(_TRACKED_FILE)]) + recovery = _make_recovery(tmp_path, tmp_registry, tmp_store, {_BUCKET_ID: lib}, extractor) + + with patch(f"{_MOD}._git_rev_parse_head", return_value=_NEW_SHA), \ + patch(f"{_MOD}._git_diff_name_only", return_value=[_TRACKED_FILE]): + result = await recovery.run() + + assert result == _NEW_SHA + + +# --------------------------------------------------------------------------- +# Tests: recovery triggers librarian.handle() +# --------------------------------------------------------------------------- + +class TestRecoveryUpdates: + async def test_handle_called_for_changed_tracked_file( + self, tmp_path, tmp_registry, tmp_store + ): + tmp_registry._meta["last_indexed_head"] = _OLD_SHA + _seed_bucket(tmp_store, _BUCKET_ID, str(tmp_path / _TRACKED_FILE)) + lib = _make_mock_librarian() + extractor = _make_mock_extractor([_make_diff_event(_TRACKED_FILE)]) + recovery = _make_recovery(tmp_path, tmp_registry, tmp_store, {_BUCKET_ID: lib}, extractor) + + with patch(f"{_MOD}._git_rev_parse_head", return_value=_NEW_SHA), \ + patch(f"{_MOD}._git_diff_name_only", return_value=[_TRACKED_FILE]): + await recovery.run() + + lib.handle.assert_called_once() + called_event = lib.handle.call_args[0][0] + assert isinstance(called_event, UpdateEvent) + assert called_event.bucket_id == _BUCKET_ID + + async def test_handle_called_once_per_hunk( + self, tmp_path, tmp_registry, tmp_store + ): + """A DiffEvent with 3 hunks should produce 3 handle() calls.""" + tmp_registry._meta["last_indexed_head"] = _OLD_SHA + _seed_bucket(tmp_store, _BUCKET_ID, str(tmp_path / _TRACKED_FILE)) + lib = _make_mock_librarian() + + hunk = DiffHunk( + file=_TRACKED_FILE, old_start=1, old_end=2, new_start=1, new_end=3, + added_lines=["x"], removed_lines=[], + ) + multi_hunk_event = DiffEvent( + file=_TRACKED_FILE, hunks=[hunk, hunk, hunk], is_rename=False + ) + extractor = _make_mock_extractor([multi_hunk_event]) + recovery = _make_recovery(tmp_path, tmp_registry, tmp_store, {_BUCKET_ID: lib}, extractor) + + with patch(f"{_MOD}._git_rev_parse_head", return_value=_NEW_SHA), \ + patch(f"{_MOD}._git_diff_name_only", return_value=[_TRACKED_FILE]): + await recovery.run() + + assert lib.handle.call_count == 3 + + async def test_handle_not_called_when_head_unchanged( + self, tmp_path, tmp_registry, tmp_store + ): + tmp_registry._meta["last_indexed_head"] = _NEW_SHA # same as current + lib = _make_mock_librarian() + extractor = _make_mock_extractor([]) + recovery = _make_recovery(tmp_path, tmp_registry, tmp_store, {_BUCKET_ID: lib}, extractor) + + with patch(f"{_MOD}._git_rev_parse_head", return_value=_NEW_SHA), \ + patch(f"{_MOD}._git_diff_name_only", return_value=[]): + await recovery.run() + + lib.handle.assert_not_called() + + async def test_handle_not_called_when_no_baseline( + self, tmp_path, tmp_registry, tmp_store + ): + lib = _make_mock_librarian() + extractor = _make_mock_extractor([]) + recovery = _make_recovery(tmp_path, tmp_registry, tmp_store, {_BUCKET_ID: lib}, extractor) + + with patch(f"{_MOD}._git_rev_parse_head", return_value=_NEW_SHA), \ + patch(f"{_MOD}._git_diff_name_only", return_value=[_TRACKED_FILE]): + await recovery.run() + + lib.handle.assert_not_called() + + +# --------------------------------------------------------------------------- +# Tests: file filtering +# --------------------------------------------------------------------------- + +class TestFileFiltering: + async def test_skips_file_with_untracked_extension( + self, tmp_path, tmp_registry, tmp_store + ): + tmp_registry._meta["last_indexed_head"] = _OLD_SHA + lib = _make_mock_librarian() + extractor = _make_mock_extractor([_make_diff_event(_UNTRACKED_FILE)]) + recovery = _make_recovery(tmp_path, tmp_registry, tmp_store, {_BUCKET_ID: lib}, extractor) + + with patch(f"{_MOD}._git_rev_parse_head", return_value=_NEW_SHA), \ + patch(f"{_MOD}._git_diff_name_only", return_value=[_UNTRACKED_FILE]): + await recovery.run() + + lib.handle.assert_not_called() + extractor.extract_between.assert_not_called() + + async def test_skips_file_not_belonging_to_any_bucket( + self, tmp_path, tmp_registry, tmp_store + ): + """A .py file that has no chunks in any bucket must be silently skipped.""" + tmp_registry._meta["last_indexed_head"] = _OLD_SHA + # Note: we do NOT seed any bucket for _TRACKED_FILE here + lib = _make_mock_librarian() + extractor = _make_mock_extractor([_make_diff_event(_TRACKED_FILE)]) + recovery = _make_recovery(tmp_path, tmp_registry, tmp_store, {_BUCKET_ID: lib}, extractor) + + with patch(f"{_MOD}._git_rev_parse_head", return_value=_NEW_SHA), \ + patch(f"{_MOD}._git_diff_name_only", return_value=[_TRACKED_FILE]): + await recovery.run() + + lib.handle.assert_not_called() + + async def test_handles_tracked_file_and_skips_untracked_in_same_diff( + self, tmp_path, tmp_registry, tmp_store + ): + tmp_registry._meta["last_indexed_head"] = _OLD_SHA + _seed_bucket(tmp_store, _BUCKET_ID, str(tmp_path / _TRACKED_FILE)) + lib = _make_mock_librarian() + extractor = _make_mock_extractor([_make_diff_event(_TRACKED_FILE)]) + recovery = _make_recovery(tmp_path, tmp_registry, tmp_store, {_BUCKET_ID: lib}, extractor) + + changed = [_TRACKED_FILE, _UNTRACKED_FILE] + with patch(f"{_MOD}._git_rev_parse_head", return_value=_NEW_SHA), \ + patch(f"{_MOD}._git_diff_name_only", return_value=changed): + await recovery.run() + + # Tracked file: 1 handle call; untracked: skipped + lib.handle.assert_called_once() + + async def test_skips_file_when_extractor_returns_empty_diff( + self, tmp_path, tmp_registry, tmp_store + ): + """extract_between() returning [] (e.g. binary file) must not call handle.""" + tmp_registry._meta["last_indexed_head"] = _OLD_SHA + _seed_bucket(tmp_store, _BUCKET_ID, str(tmp_path / _TRACKED_FILE)) + lib = _make_mock_librarian() + extractor = _make_mock_extractor([]) # empty diff + recovery = _make_recovery(tmp_path, tmp_registry, tmp_store, {_BUCKET_ID: lib}, extractor) + + with patch(f"{_MOD}._git_rev_parse_head", return_value=_NEW_SHA), \ + patch(f"{_MOD}._git_diff_name_only", return_value=[_TRACKED_FILE]): + await recovery.run() + + lib.handle.assert_not_called() + + +# --------------------------------------------------------------------------- +# Tests: extract_between receives correct arguments +# --------------------------------------------------------------------------- + +class TestExtractorArgs: + async def test_extract_between_called_with_correct_shas( + self, tmp_path, tmp_registry, tmp_store + ): + tmp_registry._meta["last_indexed_head"] = _OLD_SHA + _seed_bucket(tmp_store, _BUCKET_ID, str(tmp_path / _TRACKED_FILE)) + lib = _make_mock_librarian() + extractor = _make_mock_extractor([_make_diff_event(_TRACKED_FILE)]) + recovery = _make_recovery(tmp_path, tmp_registry, tmp_store, {_BUCKET_ID: lib}, extractor) + + with patch(f"{_MOD}._git_rev_parse_head", return_value=_NEW_SHA), \ + patch(f"{_MOD}._git_diff_name_only", return_value=[_TRACKED_FILE]): + await recovery.run() + + extractor.extract_between.assert_called_once() + _, kwargs_or_args = extractor.extract_between.call_args[0], extractor.extract_between.call_args + call_args = extractor.extract_between.call_args + # Positional: (filepath, from_sha, to_sha) + assert call_args[0][1] == _OLD_SHA # from_sha + assert call_args[0][2] == _NEW_SHA # to_sha + + async def test_git_diff_name_only_called_with_correct_shas( + self, tmp_path, tmp_registry, tmp_store + ): + tmp_registry._meta["last_indexed_head"] = _OLD_SHA + recovery = _make_recovery(tmp_path, tmp_registry, tmp_store, {}, MagicMock()) + + with patch(f"{_MOD}._git_rev_parse_head", return_value=_NEW_SHA), \ + patch(f"{_MOD}._git_diff_name_only", return_value=[]) as mock_diff: + await recovery.run() + + mock_diff.assert_called_once_with(tmp_path, _OLD_SHA, _NEW_SHA) + + +# --------------------------------------------------------------------------- +# Tests: multiple files in the same diff +# --------------------------------------------------------------------------- + +class TestMultipleFiles: + async def test_two_tracked_files_in_different_buckets( + self, tmp_path, tmp_registry, tmp_store + ): + tmp_registry._meta["last_indexed_head"] = _OLD_SHA + + file_a = "libucks/auth.py" + file_b = "libucks/query.py" + bucket_a = "aabbccdd" + bucket_b = "eeff0011" + + _seed_bucket(tmp_store, bucket_a, str(tmp_path / file_a)) + _seed_bucket(tmp_store, bucket_b, str(tmp_path / file_b)) + + lib_a = _make_mock_librarian() + lib_b = _make_mock_librarian() + + # extractor returns appropriate events per file call + extractor = MagicMock(spec=DiffExtractor) + extractor.extract_between.side_effect = [ + [_make_diff_event(file_a)], + [_make_diff_event(file_b)], + ] + + recovery = _make_recovery( + tmp_path, tmp_registry, tmp_store, + {bucket_a: lib_a, bucket_b: lib_b}, + extractor, + ) + + with patch(f"{_MOD}._git_rev_parse_head", return_value=_NEW_SHA), \ + patch(f"{_MOD}._git_diff_name_only", return_value=[file_a, file_b]): + await recovery.run() + + lib_a.handle.assert_called_once() + lib_b.handle.assert_called_once() + + async def test_same_file_in_two_buckets_calls_both_librarians( + self, tmp_path, tmp_registry, tmp_store + ): + """A file shared across two buckets should trigger updates in both.""" + tmp_registry._meta["last_indexed_head"] = _OLD_SHA + + bucket_a = "aabbccdd" + bucket_b = "eeff0011" + shared_file = _TRACKED_FILE + + _seed_bucket(tmp_store, bucket_a, str(tmp_path / shared_file)) + _seed_bucket(tmp_store, bucket_b, str(tmp_path / shared_file)) + + lib_a = _make_mock_librarian() + lib_b = _make_mock_librarian() + extractor = _make_mock_extractor([_make_diff_event(shared_file)]) + + recovery = _make_recovery( + tmp_path, tmp_registry, tmp_store, + {bucket_a: lib_a, bucket_b: lib_b}, + extractor, + ) + + with patch(f"{_MOD}._git_rev_parse_head", return_value=_NEW_SHA), \ + patch(f"{_MOD}._git_diff_name_only", return_value=[shared_file]): + await recovery.run() + + lib_a.handle.assert_called_once() + lib_b.handle.assert_called_once() + + +# --------------------------------------------------------------------------- +# Tests: resilience +# --------------------------------------------------------------------------- + +class TestResilience: + async def test_extract_exception_skips_file_does_not_raise( + self, tmp_path, tmp_registry, tmp_store + ): + tmp_registry._meta["last_indexed_head"] = _OLD_SHA + _seed_bucket(tmp_store, _BUCKET_ID, str(tmp_path / _TRACKED_FILE)) + lib = _make_mock_librarian() + + extractor = MagicMock(spec=DiffExtractor) + extractor.extract_between.side_effect = RuntimeError("git blew up") + + recovery = _make_recovery(tmp_path, tmp_registry, tmp_store, {_BUCKET_ID: lib}, extractor) + + with patch(f"{_MOD}._git_rev_parse_head", return_value=_NEW_SHA), \ + patch(f"{_MOD}._git_diff_name_only", return_value=[_TRACKED_FILE]): + result = await recovery.run() # must not raise + + assert result == _NEW_SHA + lib.handle.assert_not_called() + + async def test_empty_changed_files_list_completes_cleanly( + self, tmp_path, tmp_registry, tmp_store + ): + tmp_registry._meta["last_indexed_head"] = _OLD_SHA + recovery = _make_recovery(tmp_path, tmp_registry, tmp_store, {}, MagicMock()) + + with patch(f"{_MOD}._git_rev_parse_head", return_value=_NEW_SHA), \ + patch(f"{_MOD}._git_diff_name_only", return_value=[]): + result = await recovery.run() + + assert result == _NEW_SHA