hanzo-mcp: search, ranked code spans from a local index

One call answers a question or a failing test's output with the spans
of the tree most likely to hold the cause, as path:start-end and their
lines, so an agent stops chaining grep and read.

Python is chunked by ast, other text in overlapping 40-line windows.
BM25 over code tokens is fused by reciprocal rank with the spans a
traceback's path:line frames name; when the query names a test file
the index holds, the code under test ranks first. Every search stats
the tree and re-chunks only what moved.

A dense stage (Qwen3-Embedding form, vectors cached by text hash) is
on only when HANZO_SEARCH_MODEL names a model: on the 42 dev tasks it
lifts span MRR .385 to .508, but a CPU query embedding takes seconds.

`python -m hanzo_search.server --root DIR` serves the tool alone over
stdio with the standard library, outside hanzo_mcp, whose import is
seconds long.

The `hanzo` mode lists it, so the default surface carries it; another
mode, `tool disable search` or HANZO_MCP_ENABLED_TOOLS decide it as they
decide every other tool.

LLM.md

@@ -416,6 +416,32 @@ runs them against `conformance/fake.py`, and `uv run python tests/live_conforman
 against api.hanzo.ai and prints pass or fail per rule. Release: tag `hanzo-kai-v<version>`, and
 `https://pypi.org/pypi/hanzo-kai/json` is the proof, not a green run.
 
+## search — ranked code spans from a local index
+
+`hanzo_search` ships in the hanzo-mcp distribution as its own top-level package, because
+`import hanzo_mcp` costs seconds and the standalone server must answer in under one. hanzo-mcp
+registers `search(query, root="", k=5)` in `_register_system_tools`; `python -m
+hanzo_search.server --root DIR` serves it alone over stdio (newline JSON-RPC, standard library
+only, ~0.2 s to the first answer, index included, on a 10K-line tree).
+
+- **Index** (`index.py`): Python chunked by `ast` (each function and method, each class header,
+  code between them in windows), other text in 40-line windows 30 apart; nothing over 1 MB,
+  binaries, hidden dirs, `node_modules`, venvs or build output. Every search stats the tree
+  first: (size, mtime, ctime, inode) moved → read; bytes changed → re-chunked; gone → dropped.
+- **Rank**: BM25 over code tokens (an identifier whole, then its dotted, snake and camel parts),
+  fused by reciprocal rank with the spans a traceback's `path:line` frames name, deepest first.
+  When the query names a test file the index holds, test files rank after the code under test.
+- **Dense** (`dense.py`), off unless `HANZO_SEARCH_MODEL` names a local model directory
+  (extra `search`: torch, transformers): Qwen3-Embedding form, CPU, fp32; chunk vectors by the
+  SHA-256 of their text in `HANZO_SEARCH_CACHE` (default `~/.cache/hanzo/search`), one SQLite
+  file per model. On dev it lifts span MRR .385 → .508 (traceback) and .327 → .472 (short),
+  but a query embedding took 2.5–7 s on dgx's CPU, so the default is BM25 and frames.
+- Dev (42 kai-day code tasks, `~/.cache/kai-day/search.dev.json`), failing-test output as the
+  query: file r@1 .714, r@5 1.0; span r@1 .19, r@5 .667. Warm search on a 72K-line tree p50
+  11 ms, p95 43 ms.
+- Tests: `tests/test_hanzo_search.py` (tokens, chunks, ranking, freshness, caps, schema size,
+  dense cache with a fake embedder, stdio round trip).
+
 ## Plan usage — `hanzoai.usage`
 
 `read_usage(headers)` → `Usage(usage, usage_class, paid_by, fallback, served, reason)` from

pkg/hanzo-mcp/hanzo_mcp/tools/__init__.py

@@ -273,6 +273,18 @@ def register_all_tools(
     else:
         logger.warning("No tool packages discovered via entry points")
 
+    # search ships in this distribution (hanzo_search), so the mode decides it here, as it does a package's
+    if "search" in asked:
+        search_on = asked["search"]
+    elif asked.get("all") or not (mode and mode.tools):
+        search_on = True
+    else:
+        search_on = bool(tool_config.get("search", False))
+    if search_on:
+        from hanzo_search import register as register_search
+
+        register_search(mcp_server)
+
     # Register system tools that are always available
     _register_system_tools(mcp_server, all_tools)
 

pkg/hanzo-mcp/hanzo_mcp/tools/common/personality.py

@@ -142,7 +142,7 @@ ESSENTIAL_TOOLS = [
 # (`tool enable`, HANZO_MCP_ENABLED_TOOLS).
 DEFAULT_TOOLS = [
     "fs", "exec", "zsh", "git", "curl", "code", "lsp", "browser", "playwright", "agent", "fetch",
-    "llm", "kai_decide", "repo", "hanzo", "mode", "tool",
+    "llm", "kai_decide", "repo", "hanzo", "mode", "tool", "search",
 ]
 
 # Process and package helpers the wider personalities keep.

pkg/hanzo-mcp/hanzo_search/__init__.py

@@ -0,0 +1,33 @@
+"""`search`: ranked code spans for a natural-language or error-text query, from a local index."""
+
+from __future__ import annotations
+
+from typing import Any
+
+DESCRIPTION = (
+    "Find code: the spans of a source tree most relevant to a question or to error text "
+    "(a traceback or failing test output works as is), ranked, each as path:start-end and its lines. "
+    "Indexes root (default: working directory) once and follows edits. Use before grep/read."
+)
+
+SCHEMA: dict[str, Any] = {
+    "type": "object",
+    "properties": {
+        "query": {"type": "string"},
+        "root": {"type": "string"},
+        "k": {"type": "integer", "default": 5},
+    },
+    "required": ["query"],
+}
+
+
+def register(mcp: Any) -> None:
+    """Add `search` to a FastMCP server; the index loads on the first call."""
+
+    @mcp.tool(name="search", description=DESCRIPTION)
+    async def search(query: str, root: str = "", k: int = 5) -> str:
+        import asyncio
+
+        from hanzo_search.index import search as run
+
+        return await asyncio.to_thread(run, query, root, k)

pkg/hanzo-mcp/hanzo_search/dense.py

@@ -0,0 +1,125 @@
+"""The optional dense stage: a local embedding model, on only when HANZO_SEARCH_MODEL names one.
+
+The model is read in the Qwen3-Embedding form (last-token pooled, unit length, a query under
+an instruction), on the CPU, offline. Each chunk is embedded once: its vector is kept under the
+SHA-256 of its text in HANZO_SEARCH_CACHE (default ~/.cache/hanzo/search), one SQLite file per
+model, so another process, or another checkout of the same files, embeds nothing again.
+Needs torch and transformers, which the base install does not carry.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import os
+import sqlite3
+import threading
+from typing import TYPE_CHECKING, Any, Callable
+
+if TYPE_CHECKING:
+    from hanzo_search.index import Chunk, Index
+
+PREFIX = "Instruct: Given a failing test's output or a question about code, retrieve the code it concerns\nQuery:"
+LENGTH = 512  # tokens read of a chunk or a query (a query keeps its end)
+BATCH = 16
+
+
+def text(c: "Chunk") -> str:
+    return f"{c.path} {c.name}\n{c.text}"
+
+
+def cache() -> str:
+    return os.environ.get("HANZO_SEARCH_CACHE") or os.path.join(os.path.expanduser("~"), ".cache", "hanzo", "search")
+
+
+class Dense:
+    """Chunk vectors by text hash (memory, then the SQLite file, then embed), ranked by cosine."""
+
+    def __init__(self, embed: Callable[[list[str]], Any], db: str):
+        os.makedirs(os.path.dirname(db), exist_ok=True)
+        self.embed = embed
+        self.db = sqlite3.connect(db, check_same_thread=False)
+        self.db.execute("create table if not exists v (k blob primary key, x blob)")
+        self.lock = threading.Lock()
+        self.held: dict[bytes, Any] = {}
+        self.matrix: dict[int, tuple[tuple[bytes, ...], Any]] = {}
+        self.embedded = 0
+
+    def vectors(self, ix: "Index") -> tuple[list[int], Any]:
+        import numpy as np
+
+        ids = sorted(ix.chunks)
+        keys = [hashlib.sha256(text(ix.chunks[c]).encode()).digest() for c in ids]
+        want = list({k for k in keys if k not in self.held})
+        for i in range(0, len(want), 500):
+            part = want[i : i + 500]
+            for k, x in self.db.execute("select k, x from v where k in (%s)" % ",".join("?" * len(part)), part):
+                self.held[k] = np.frombuffer(x, dtype=np.float16).astype(np.float32)
+        todo = [k for k in want if k not in self.held]
+        if todo:
+            byk = {k: c for c, k in zip(ids, keys)}
+            new = self.embed([text(ix.chunks[byk[k]]) for k in todo])
+            self.embedded += len(todo)
+            self.db.executemany("insert or replace into v values (?, ?)", [(k, v.astype(np.float16).tobytes()) for k, v in zip(todo, new)])
+            self.db.commit()
+            for k, v in zip(todo, new):
+                self.held[k] = np.asarray(v, dtype=np.float16).astype(np.float32)
+        tag = tuple(keys)
+        m = self.matrix.get(id(ix))
+        if m is None or m[0] != tag:
+            m = (tag, np.stack([self.held[k] for k in keys]) if keys else np.zeros((0, 1), np.float32))
+            self.matrix[id(ix)] = m
+        return ids, m[1]
+
+    def rank(self, ix: "Index", query: str) -> list[int]:
+        with self.lock:
+            ids, m = self.vectors(ix)
+            if not ids:
+                return []
+            s = m @ self.embed([PREFIX + query])[0]
+        return [ids[i] for i in (-s).argsort()]
+
+
+def model(path: str, device: str = "cpu") -> tuple[Callable[[list[str]], Any], str]:
+    """An embed function over a local model directory, and the model's identity."""
+    try:
+        import torch
+        from transformers import AutoModel, AutoTokenizer
+    except ImportError as e:
+        raise RuntimeError(f"HANZO_SEARCH_MODEL is set but {e.name} is not installed") from e
+    tok = AutoTokenizer.from_pretrained(path, padding_side="left", truncation_side="left", local_files_only=True)
+    dtype = torch.float32 if device == "cpu" else torch.bfloat16
+    net = AutoModel.from_pretrained(path, dtype=dtype, local_files_only=True).to(device).eval()
+
+    def embed(texts: list[str]) -> Any:
+        import numpy as np
+
+        out = []
+        for i in range(0, len(texts), BATCH):
+            b = tok(texts[i : i + BATCH], padding=True, truncation=True, max_length=LENGTH, return_tensors="pt").to(device)
+            with torch.inference_mode():
+                h = net(**b).last_hidden_state[:, -1].float()
+            out.append(torch.nn.functional.normalize(h, dim=-1).cpu().numpy())
+        return np.concatenate(out)
+
+    ident = hashlib.sha256(open(os.path.join(path, "config.json"), "rb").read())
+    for name in sorted(os.listdir(path)):
+        if name.endswith(".safetensors"):
+            ident.update(f"{name}:{os.path.getsize(os.path.join(path, name))}".encode())
+    return embed, ident.hexdigest()[:16]
+
+
+_dense: Dense | None = None
+_guard = threading.Lock()
+
+
+def load() -> Dense | None:
+    """The process's dense stage, None unless HANZO_SEARCH_MODEL names a model directory."""
+    global _dense
+    path = os.environ.get("HANZO_SEARCH_MODEL")
+    if not path:
+        return None
+    with _guard:
+        if _dense is None:
+            embed, ident = model(path)
+            _dense = Dense(embed, os.path.join(cache(), ident + ".sqlite"))
+        return _dense

pkg/hanzo-mcp/hanzo_search/index.py

@@ -0,0 +1,366 @@
+"""An index of a source tree that ranks code spans for a query and follows the tree.
+
+Every search stats the tree first: a file whose size, mtime, ctime or inode moved is read
+again, re-chunked only when its bytes changed, and a file gone from the tree drops out.
+Ranking fuses, by reciprocal rank, BM25 over code-aware tokens, the spans a traceback in
+the query names by path and line, and (when a model is configured) dense similarity.
+"""
+
+from __future__ import annotations
+
+import ast
+import hashlib
+import math
+import os
+import re
+import threading
+import warnings
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING, Callable
+
+if TYPE_CHECKING:
+    from hanzo_search.dense import Dense
+
+SKIP = {"node_modules", "__pycache__", "venv", "env", "build", "dist", "target", "site-packages"}
+NOISE = (".lock", ".min.js", ".map", ".svg", ".pyc", ".pyo", ".so", ".o", ".a", ".dylib", ".dll", ".exe", ".whl", ".gz", ".zip", ".tar", ".png", ".jpg", ".jpeg", ".gif", ".ico", ".pdf", ".woff", ".woff2", ".ttf", ".npy", ".npz", ".pkl", ".sqlite", ".db")
+LIMIT = 1 << 20  # bytes: a larger file is not indexed
+SPAN = 40  # lines a chunk holds at most
+STRIDE = 30  # a window's step, so windows overlap by SPAN - STRIDE
+WIDTH = 300  # characters a shown line keeps
+K1, B = 1.2, 0.75
+RRF = 60
+
+WORD = re.compile(r"[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)*")
+PART = re.compile(r"[A-Z]+(?![a-z])|[A-Z]?[a-z]+|[0-9]+")
+PATH = re.compile(r"(?:[\w.-]+/)*[\w-]+\.\w+")
+FRAME = re.compile(r"""(?:File "([^"]+)", line (\d+))|((?:[\w.-]+/)*[\w.-]+\.\w+):(\d+)""")
+
+
+def tokens(text: str) -> list[str]:
+    """Lowercased identifiers, each whole, then its dotted segments and their snake/camel parts."""
+    out: list[str] = []
+    for m in WORD.finditer(text):
+        word = m.group()
+        if "." in word:
+            out.append(word.lower())
+        for seg in word.split("."):
+            low = seg.lower()
+            if len(low) > 1:
+                out.append(low)
+            parts = PART.findall(seg)
+            if len(parts) > 1:
+                out.extend(p.lower() for p in parts if len(p) > 1)
+    return out
+
+
+@dataclass
+class Chunk:
+    path: str
+    start: int  # 1-based, inclusive
+    end: int
+    name: str
+    text: str
+    tf: dict[str, int] = field(default_factory=dict)
+    length: int = 0
+
+
+def windows(start: int, end: int) -> list[tuple[int, int]]:
+    """[start, end] as windows of at most SPAN lines, STRIDE apart, the last flush with end."""
+    if end - start + 1 <= SPAN:
+        return [(start, end)]
+    out = []
+    s = start
+    while True:
+        out.append((s, min(s + SPAN - 1, end)))
+        if s + SPAN - 1 >= end:
+            return out
+        s += STRIDE
+
+
+def outline(text: str) -> list[tuple[int, int, str]]:
+    """Python spans: each function and method, each class's header, and the code between them."""
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore")
+        tree = ast.parse(text)
+    spans: list[tuple[int, int, str]] = []
+    defs = (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)
+
+    def first(node: ast.AST) -> int:
+        return min([node.lineno] + [d.lineno for d in node.decorator_list])  # type: ignore[attr-defined]
+
+    def visit(body: list[ast.stmt], prefix: str) -> None:
+        for node in body:
+            if not isinstance(node, defs):
+                continue
+            name = prefix + node.name
+            inner = [n for n in node.body if isinstance(n, defs)] if isinstance(node, ast.ClassDef) else []
+            if inner:
+                spans.append((first(node), min(first(n) for n in inner) - 1, name))
+                visit(node.body, name + ".")
+            else:
+                spans.append((first(node), node.end_lineno or node.lineno, name))
+
+    visit(tree.body, "")
+    return spans
+
+
+def chunk(path: str, text: str) -> list[Chunk]:
+    """A file's chunks: Python by its outline (code outside any def in gap windows), else windows."""
+    lines = text.splitlines()
+    n = len(lines)
+    spans: list[tuple[int, int, str]] = []
+    if path.endswith(".py"):
+        try:
+            spans = outline(text)
+        except (SyntaxError, ValueError):
+            spans = []
+    covered = bytearray(n + 2)
+    out: list[tuple[int, int, str]] = []
+    for s, e, name in spans:
+        for a, b in windows(s, e):
+            out.append((a, b, name))
+        for i in range(s, e + 1):
+            covered[i] = 1
+    i = 1
+    while i <= n:
+        if covered[i]:
+            i += 1
+            continue
+        j = i
+        while j + 1 <= n and not covered[j + 1]:
+            j += 1
+        out.extend((a, b, "") for a, b in windows(i, j))
+        i = j + 1
+    chunks = []
+    for s, e, name in sorted(out):
+        body = "\n".join(lines[s - 1 : e])
+        if not body.strip():
+            continue
+        c = Chunk(path, s, e, name, body)
+        toks = tokens(f"{path} {name}\n{body}")
+        for t in toks:
+            c.tf[t] = c.tf.get(t, 0) + 1
+        c.length = len(toks)
+        chunks.append(c)
+    return chunks
+
+
+def walk(root: str) -> dict[str, os.stat_result]:
+    """Indexable files under root, by path relative to it."""
+    out: dict[str, os.stat_result] = {}
+    stack = [""]
+    while stack:
+        rel = stack.pop()
+        try:
+            it = os.scandir(os.path.join(root, rel) if rel else root)
+        except OSError:
+            continue
+        with it:
+            for e in it:
+                name = e.name
+                if name.startswith(".") or name in SKIP or name.endswith(".egg-info"):
+                    continue
+                path = f"{rel}/{name}" if rel else name
+                try:
+                    if e.is_dir(follow_symlinks=False):
+                        stack.append(path)
+                    elif e.is_file(follow_symlinks=False) and not name.endswith(NOISE):
+                        st = e.stat(follow_symlinks=False)
+                        if st.st_size <= LIMIT:
+                            out[path] = st
+                except OSError:
+                    continue
+    return out
+
+
+def tested(path: str) -> bool:
+    """A test file, by the names pytest and unittest collect."""
+    parts = path.split("/")
+    base = parts[-1]
+    return base.startswith("test_") or base.endswith("_test.py") or base == "conftest.py" or any(p in ("test", "tests") for p in parts[:-1])
+
+
+@dataclass
+class Hit:
+    chunk: Chunk
+    score: float
+
+
+class Index:
+    """Chunks of one tree with BM25 postings, kept current by refresh()."""
+
+    def __init__(self, root: str, dense: "Dense | None" = None):
+        self.root = os.path.realpath(root)
+        self.dense = dense
+        self.files: dict[str, tuple[tuple[int, int, int, int], str, list[int]]] = {}
+        self.chunks: dict[int, Chunk] = {}
+        self.post: dict[str, dict[int, int]] = {}
+        self.total = 0
+        self.serial = 0
+        self.lock = threading.Lock()
+
+    def drop(self, path: str) -> None:
+        for cid in self.files.pop(path)[2]:
+            c = self.chunks.pop(cid)
+            self.total -= c.length
+            for t in c.tf:
+                p = self.post[t]
+                del p[cid]
+                if not p:
+                    del self.post[t]
+
+    def add(self, path: str, key: tuple[int, int, int, int], digest: str, text: str) -> None:
+        ids = []
+        for c in chunk(path, text):
+            cid = self.serial
+            self.serial += 1
+            self.chunks[cid] = c
+            self.total += c.length
+            for t, n in c.tf.items():
+                self.post.setdefault(t, {})[cid] = n
+            ids.append(cid)
+        self.files[path] = (key, digest, ids)
+
+    def refresh(self) -> int:
+        """Bring the index to the tree as it is now; the number of files read again."""
+        seen = walk(self.root)
+        for path in [p for p in self.files if p not in seen]:
+            self.drop(path)
+        read = 0
+        for path, st in seen.items():
+            key = (st.st_size, st.st_mtime_ns, st.st_ctime_ns, st.st_ino)
+            old = self.files.get(path)
+            if old and old[0] == key:
+                continue
+            try:
+                data = open(os.path.join(self.root, path), "rb").read()
+            except OSError:
+                if old:
+                    self.drop(path)
+                continue
+            read += 1
+            if b"\0" in data[:8192]:  # binary: kept by its stat alone, so it is not read again
+                if old:
+                    self.drop(path)
+                self.files[path] = (key, "", [])
+                continue
+            digest = hashlib.sha256(data).hexdigest()
+            if old and old[1] == digest:
+                self.files[path] = (key, digest, old[2])
+                continue
+            if old:
+                self.drop(path)
+            self.add(path, key, digest, data.decode("utf-8", "replace"))
+        return read
+
+    def bm25(self, query: str) -> list[int]:
+        n = len(self.chunks)
+        if not n:
+            return []
+        avg = self.total / n
+        scores: dict[int, float] = {}
+        for t in set(tokens(query)):
+            p = self.post.get(t)
+            if not p:
+                continue
+            idf = math.log(1 + (n - len(p) + 0.5) / (len(p) + 0.5))
+            for cid, f in p.items():
+                norm = K1 * (1 - B + B * self.chunks[cid].length / avg)
+                scores[cid] = scores.get(cid, 0.0) + idf * f * (K1 + 1) / (f + norm)
+        return sorted(scores, key=lambda c: -scores[c])
+
+    def resolver(self) -> Callable[[str], str | None]:
+        """A map from a path as a traceback or a message spells it to the one indexed file it names."""
+        names: dict[str, list[str]] = {}
+        for path in self.files:
+            names.setdefault(path.rsplit("/", 1)[-1], []).append(path)
+
+        def resolve(name: str) -> str | None:
+            name = name.replace("\\", "/")
+            if name.startswith("./"):
+                name = name[2:]
+            fits = [p for p in names.get(name.rsplit("/", 1)[-1], ()) if p == name or p.endswith("/" + name) or name.endswith("/" + p)]
+            return fits[0] if len(fits) == 1 else None
+
+        return resolve
+
+    def frames(self, query: str) -> list[int]:
+        """Chunks holding the lines a traceback in the query names, the deepest frame first."""
+        resolve = self.resolver()
+        order: list[int] = []
+        for m in reversed(list(FRAME.finditer(query))):
+            name, line = (m.group(1), m.group(2)) if m.group(1) else (m.group(3), m.group(4))
+            path = resolve(name)
+            if path is None:
+                continue
+            ln = int(line)
+            for cid in self.files[path][2]:
+                c = self.chunks[cid]
+                if c.start <= ln <= c.end and cid not in order:
+                    order.append(cid)
+        return order
+
+    def rank(self, query: str) -> list[Hit]:
+        """Fused ranks; when the query names a test file it already holds, the code under test first."""
+        lists = [self.bm25(query), self.frames(query)]
+        if self.dense is not None:
+            lists.append(self.dense.rank(self, query))
+        fused: dict[int, float] = {}
+        for ranked in lists:
+            for r, cid in enumerate(ranked):
+                fused[cid] = fused.get(cid, 0.0) + 1.0 / (RRF + r + 1)
+        order = sorted(fused, key=lambda c: -fused[c])
+        resolve = self.resolver()
+        if any(tested(p) for p in filter(None, map(resolve, PATH.findall(query)))):
+            order = [c for c in order if not tested(self.chunks[c].path)] + [c for c in order if tested(self.chunks[c].path)]
+        return [Hit(self.chunks[c], fused[c]) for c in order]
+
+    def search(self, query: str, k: int = 5) -> list[Hit]:
+        with self.lock:
+            self.refresh()
+            return self.rank(query)[: max(1, k)]
+
+
+def render(hits: list[Hit], cap: int = 6000) -> str:
+    """Each hit as `path:start-end name` and its lines, all within cap characters."""
+    blocks: list[str] = []
+    used = 0
+    for h in hits:
+        c = h.chunk
+        lines = [ln if len(ln) <= WIDTH else ln[:WIDTH] + "…" for ln in c.text.splitlines()[:SPAN]]
+        block = f"{c.path}:{c.start}-{c.end}" + (f" {c.name}" if c.name else "") + "\n" + "\n".join(lines)
+        room = cap - used - 2 * len(blocks)
+        if room <= 0:
+            break
+        if len(block) > room:
+            if blocks and room < 400:
+                break
+            block = block[:room]
+        blocks.append(block)
+        used += len(block)
+    return "\n\n".join(blocks) if blocks else "no match"
+
+
+_open: dict[str, Index] = {}
+_guard = threading.Lock()
+
+
+def index(root: str) -> Index:
+    """The process's index of root, made on first use."""
+    real = os.path.realpath(root)
+    with _guard:
+        if real not in _open:
+            from hanzo_search.dense import load
+
+            _open[real] = Index(real, load())
+        return _open[real]
+
+
+def search(query: str, root: str = "", k: int = 5) -> str:
+    """Rendered top-k spans of root (default: the working directory) for query."""
+    if not query.strip():
+        return "empty query"
+    base = root or os.getcwd()
+    if not os.path.isdir(base):
+        return f"not a directory: {base}"
+    return render(index(base).search(query, k))

pkg/hanzo-mcp/hanzo_search/server.py

@@ -0,0 +1,76 @@
+"""A stdio MCP server exposing `search` alone: `python -m hanzo_search.server --root DIR`.
+
+Newline-delimited JSON-RPC 2.0 on stdin/stdout, standard library only, so it answers
+`initialize` within a fraction of a second of starting; the index is built on the first call.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+
+from hanzo_search import DESCRIPTION, SCHEMA
+
+
+def answer(msg: dict, root: str) -> dict | None:
+    """The response to one message, None for a notification."""
+    method = msg.get("method")
+    if "id" not in msg:
+        return None
+    rid = msg["id"]
+    params = msg.get("params") or {}
+    if method == "initialize":
+        result = {
+            "protocolVersion": params.get("protocolVersion", "2025-06-18"),
+            "capabilities": {"tools": {"listChanged": False}},
+            "serverInfo": {"name": "hanzo-search", "version": "1"},
+        }
+    elif method == "ping":
+        result = {}
+    elif method == "tools/list":
+        result = {"tools": [{"name": "search", "description": DESCRIPTION, "inputSchema": SCHEMA}]}
+    elif method == "tools/call":
+        if params.get("name") != "search":
+            return {"jsonrpc": "2.0", "id": rid, "error": {"code": -32602, "message": f"unknown tool: {params.get('name')}"}}
+        args = params.get("arguments") or {}
+        query = args.get("query")
+        if not isinstance(query, str):
+            return {"jsonrpc": "2.0", "id": rid, "result": {"content": [{"type": "text", "text": "query must be a string"}], "isError": True}}
+        base = args.get("root") or root
+        if not os.path.isabs(base):
+            base = os.path.join(root, base)
+        try:
+            from hanzo_search.index import search
+
+            text, failed = search(query, base, int(args.get("k") or 5)), False
+        except Exception as e:  # a failed call is the tool's answer, never the server's exit
+            text, failed = f"{type(e).__name__}: {e}", True
+        result = {"content": [{"type": "text", "text": text}], "isError": failed}
+    else:
+        return {"jsonrpc": "2.0", "id": rid, "error": {"code": -32601, "message": f"method not found: {method}"}}
+    return {"jsonrpc": "2.0", "id": rid, "result": result}
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser(prog="python -m hanzo_search.server")
+    ap.add_argument("--root", default=os.getcwd(), help="tree to search (default: the working directory)")
+    root = os.path.realpath(ap.parse_args().root)
+    out = sys.stdout
+    for line in sys.stdin:
+        if not line.strip():
+            continue
+        try:
+            msg = json.loads(line)
+        except json.JSONDecodeError as e:
+            reply: dict | None = {"jsonrpc": "2.0", "id": None, "error": {"code": -32700, "message": str(e)}}
+        else:
+            reply = answer(msg, root) if isinstance(msg, dict) else {"jsonrpc": "2.0", "id": None, "error": {"code": -32600, "message": "not a request"}}
+        if reply is not None:
+            out.write(json.dumps(reply) + "\n")
+            out.flush()
+
+
+if __name__ == "__main__":
+    main()

pkg/hanzo-mcp/pyproject.toml

@@ -81,6 +81,12 @@ dependencies = [
 # shell = ["hanzo-tools-shell>=0.1.0"]
 # etc.
 
+# Dense stage of the `search` tool (on when HANZO_SEARCH_MODEL names a local model)
+search = [
+    "torch>=2.4",
+    "transformers>=4.51",
+]
+
 # Memory tools with full backend support
 memory = [
     "hanzo-tools-memory[full]>=0.2.2",
@@ -137,7 +143,7 @@ hanzo-mcp-dev = "hanzo_mcp.dev_server:run_dev_server"
 
 [tool.setuptools.packages.find]
 where = ["."]
-include = ["hanzo_mcp*"]
+include = ["hanzo_mcp*", "hanzo_search*"]
 
 [tool.setuptools.package-data]
 hanzo_mcp = ["py.typed"]

pkg/hanzo-mcp/tests/test_hanzo_search.py

@@ -0,0 +1,304 @@
+"""The `search` tool: tokens, chunks, ranking, freshness, rendering, schema, dense cache, stdio."""
+
+import json
+import os
+import shutil
+import subprocess
+import sys
+import time
+
+import pytest
+
+from hanzo_search import DESCRIPTION, SCHEMA
+from hanzo_search import index as X
+from hanzo_search.dense import Dense
+
+PKG = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+PRICE = '''"""Prices."""
+import math
+
+RATE = 0.2
+
+
+def apply_discount(total, percent):
+    """Take percent off total."""
+    return total * (1 - percent / 100)
+
+
+class Invoice:
+    """An invoice."""
+
+    currency = "usd"
+
+    def subtotal(self, lines):
+        return sum(q * p for q, p in lines)
+
+    @property
+    def taxRate(self):
+        return RATE
+'''
+
+PARSE = '''def parse_header(line):
+    key, _, value = line.partition(":")
+    return key.strip().lower(), value.strip()
+
+
+def read_body(stream):
+    return stream.read()
+'''
+
+TEST = '''from shop.price import apply_discount
+
+
+def test_apply_discount():
+    assert apply_discount(100, 10) == 90
+'''
+
+
+@pytest.fixture
+def repo(tmp_path):
+    (tmp_path / "shop").mkdir()
+    (tmp_path / "shop" / "price.py").write_text(PRICE)
+    (tmp_path / "shop" / "parse.py").write_text(PARSE)
+    (tmp_path / "tests").mkdir()
+    (tmp_path / "tests" / "test_price.py").write_text(TEST)
+    (tmp_path / "README.md").write_text("# Shop\n\nPrices and invoices for the shop.\n")
+    (tmp_path / ".git").mkdir()
+    (tmp_path / ".git" / "config").write_text("apply_discount apply_discount")
+    (tmp_path / "node_modules").mkdir()
+    (tmp_path / "node_modules" / "x.js").write_text("apply_discount()")
+    (tmp_path / "blob.bin").write_bytes(b"apply_discount\0\0\0")
+    return tmp_path
+
+
+def top(ix, query):
+    return [(h.chunk.path, h.chunk.name) for h in ix.search(query, 5)]
+
+
+def test_tokens_split_identifiers_and_keep_them_whole():
+    t = X.tokens("parse_header(HTTPServer.getValue) x 42")
+    for want in ("parse_header", "parse", "header", "httpserver.getvalue", "httpserver", "http", "server", "getvalue", "get", "value"):
+        assert want in t, want
+    assert "x" not in t and "42" not in t
+
+
+def test_python_chunks_follow_definitions():
+    chunks = X.chunk("shop/price.py", PRICE)
+    names = [(c.name, c.start, c.end) for c in chunks]
+    assert ("apply_discount", 7, 9) in names
+    assert ("Invoice.subtotal", 17, 18) in names
+    assert ("Invoice.taxRate", 20, 22) in names  # the decorator starts the span
+    assert any(n == "Invoice" and s == 12 for n, s, _ in names)  # the class header
+    assert any(n == "" and s == 1 for n, s, _ in names)  # module code between definitions
+
+
+def test_long_spans_and_other_files_are_overlapping_windows():
+    body = "\n".join(f"line {i}" for i in range(1, 101))
+    spans = [(c.start, c.end) for c in X.chunk("notes.txt", body)]
+    assert spans == [(1, 40), (31, 70), (61, 100)]
+    fn = "def big():\n" + "\n".join(f"    x{i} = {i}" for i in range(60)) + "\n"
+    parts = [c for c in X.chunk("big.py", fn) if c.name == "big"]
+    assert [(c.start, c.end) for c in parts] == [(1, 40), (31, 61)]
+    assert all(c.end - c.start + 1 <= X.SPAN for c in parts)
+
+
+def test_broken_python_still_indexes():
+    chunks = X.chunk("bad.py", "def f(:\n    return 1\n")
+    assert len(chunks) == 1 and chunks[0].start == 1
+
+
+def test_walk_skips_vcs_vendor_and_binaries(repo):
+    ix = X.Index(str(repo))
+    ix.refresh()
+    assert set(ix.files) == {"shop/price.py", "shop/parse.py", "tests/test_price.py", "README.md", "blob.bin"}
+    assert ix.files["blob.bin"][2] == []
+
+
+def test_bm25_ranks_the_named_definition_first(repo):
+    ix = X.Index(str(repo))
+    assert top(ix, "where is the header of a line parsed")[0] == ("shop/parse.py", "parse_header")
+    assert top(ix, "subtotal of the invoice lines")[0] == ("shop/price.py", "Invoice.subtotal")
+
+
+def test_a_traceback_names_its_span_and_the_code_under_test_comes_first(repo):
+    ix = X.Index(str(repo))
+    out = (
+        "tests/test_price.py:5: in test_apply_discount\n    assert apply_discount(100, 10) == 90\n"
+        "shop/price.py:9: in apply_discount\nE   AssertionError\n"
+        "FAILED tests/test_price.py::test_apply_discount - AssertionError"
+    )
+    assert top(ix, out)[0] == ("shop/price.py", "apply_discount")
+    ranked = [h.chunk for h in ix.rank(out)]
+    first = max(i for i, c in enumerate(ranked) if not X.tested(c.path))
+    assert all(X.tested(c.path) for c in ranked[first + 1 :]) and X.tested(ranked[-1].path)
+    # a question that names no test file ranks tests by relevance alone
+    assert top(ix, "test apply discount 100 10 90")[0] == ("tests/test_price.py", "test_apply_discount")
+
+
+def test_freshness_create_modify_delete_and_replace(repo):
+    ix = X.Index(str(repo))
+    assert all(p != "shop/new.py" for p, _ in top(ix, "frobnicate widget"))
+    (repo / "shop" / "new.py").write_text("def frobnicate_widget():\n    return 1\n")
+    assert top(ix, "frobnicate widget")[0] == ("shop/new.py", "frobnicate_widget")
+
+    # rewritten: same size, mtime put back; the ctime still moves
+    p = repo / "shop" / "parse.py"
+    st = p.stat()
+    p.write_text(PARSE.replace("read_body", "zapx_body"))
+    os.utime(p, ns=(st.st_atime_ns, st.st_mtime_ns))
+    assert top(ix, "zapx body")[0] == ("shop/parse.py", "zapx_body")
+    assert all(c.name != "read_body" for c in ix.chunks.values())
+
+    (repo / "shop" / "new.py").unlink()
+    ix.search("x", 1)
+    assert "shop/new.py" not in ix.files and all(c.path != "shop/new.py" for c in ix.chunks.values())
+
+    # a whole directory swapped for another with the same names and new contents
+    shutil.rmtree(repo / "shop")
+    (repo / "shop").mkdir()
+    (repo / "shop" / "price.py").write_text("def quux_total():\n    return 2\n")
+    hits = top(ix, "quux total")
+    assert hits[0] == ("shop/price.py", "quux_total")
+    assert set(ix.files) == {"shop/price.py", "tests/test_price.py", "README.md", "blob.bin"}
+    assert not any(t in ix.post for t in ("subtotal", "parse_header", "zapx_body"))
+    assert sum(c.length for c in ix.chunks.values()) == ix.total
+
+
+def test_unchanged_files_are_not_read_again(repo):
+    ix = X.Index(str(repo))
+    assert ix.refresh() == 5  # blob.bin is read once, found binary, and kept by its stat
+    assert ix.refresh() == 0
+    os.utime(repo / "README.md")  # touched, same bytes: read, not re-chunked
+    before = set(ix.chunks)
+    assert ix.refresh() == 1 and set(ix.chunks) == before
+
+
+def test_render_caps_hits_lines_and_characters(tmp_path):
+    for i in range(12):
+        body = "\n".join(f"    v{j} = 'needle {i} {j}' + '{'w' * 400}'" for j in range(80))
+        (tmp_path / f"m{i}.py").write_text(f"def needle_{i}():\n{body}\n")
+    out = X.search("needle", str(tmp_path), k=50)
+    assert len(out) <= 6000
+    blocks = out.split("\n\n")
+    assert all(len(b.splitlines()) <= X.SPAN + 1 for b in blocks)
+    assert all(len(line) <= X.WIDTH + 1 for b in blocks for line in b.splitlines()[1:])
+    head = blocks[0].splitlines()[0]
+    path, _, rest = head.partition(":")
+    assert path.startswith("m") and rest.split()[0].count("-") == 1
+    assert len(X.search("needle", str(tmp_path), k=3).split("\n\n")) <= 3
+    assert X.search("zzzqqq", str(tmp_path)) == "no match"
+    assert X.search("  ", str(tmp_path)) == "empty query"
+    assert X.search("x", str(tmp_path / "missing")).startswith("not a directory")
+
+
+def test_schema_and_description_stay_small():
+    tokens = len(json.dumps({"name": "search", "description": DESCRIPTION, "inputSchema": SCHEMA})) / 4
+    assert tokens < 250, tokens
+
+
+def test_fastmcp_registration_is_the_same_small_tool():
+    from mcp.server.fastmcp import FastMCP
+
+    from hanzo_search import register
+
+    mcp = FastMCP("t")
+    register(mcp)
+    tool = mcp._tool_manager.get_tool("search")
+    assert tool.description == DESCRIPTION
+    assert set(tool.parameters["properties"]) == {"query", "root", "k"}
+    assert tool.parameters["required"] == ["query"]
+    assert len(json.dumps({"d": tool.description, "s": tool.parameters})) / 4 < 250
+
+
+def test_dense_vectors_are_kept_by_content_across_trees(repo, tmp_path_factory):
+    import numpy as np
+
+    calls = []
+
+    def embed(texts):
+        calls.append(len(texts))
+        out = np.zeros((len(texts), 8), np.float32)
+        for i, t in enumerate(texts):
+            out[i, hash(t) % 8] = 1.0
+        return out
+
+    db = str(tmp_path_factory.mktemp("cache") / "m.sqlite")
+    ix = X.Index(str(repo), Dense(embed, db))
+    ix.search("invoice", 3)
+    first = sum(calls) - 1  # less the query
+    assert first == len(ix.chunks)
+
+    copy = tmp_path_factory.mktemp("copy") / "t"
+    shutil.copytree(repo, copy)
+    calls.clear()
+    other = X.Index(str(copy), Dense(embed, db))  # a fresh process: memory empty, the file kept
+    other.search("invoice", 3)
+    assert calls == [1]  # the query alone
+
+    (copy / "shop" / "parse.py").write_text(PARSE + "\n\ndef added():\n    return 3\n")
+    calls.clear()
+    other.search("invoice", 3)
+    assert sum(calls) == 1 + 1  # the one new chunk, then the query
+
+
+def rpc(proc, msg):
+    proc.stdin.write(json.dumps(msg) + "\n")
+    proc.stdin.flush()
+    if "id" not in msg:
+        return None
+    return json.loads(proc.stdout.readline())
+
+
+def test_stdio_server_round_trip(repo, tmp_path):
+    home = tmp_path / "home"
+    home.mkdir()
+    env = {"PATH": os.environ["PATH"], "HOME": str(home), "PYTHONPATH": PKG}
+    t = time.perf_counter()
+    proc = subprocess.Popen(
+        [sys.executable, "-m", "hanzo_search.server", "--root", str(repo)],
+        stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True, cwd=str(repo.parent), env=env,
+    )
+    try:
+        init = rpc(proc, {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "t", "version": "1"}}})
+        assert time.perf_counter() - t < 1.0
+        assert init["result"]["protocolVersion"] == "2025-06-18" and "tools" in init["result"]["capabilities"]
+        rpc(proc, {"jsonrpc": "2.0", "method": "notifications/initialized"})
+        tools = rpc(proc, {"jsonrpc": "2.0", "id": 2, "method": "tools/list"})["result"]["tools"]
+        assert [x["name"] for x in tools] == ["search"]
+        call = rpc(proc, {"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": "search", "arguments": {"query": "parse a header line", "k": 2}}})
+        text = call["result"]["content"][0]["text"]
+        assert call["result"]["isError"] is False and text.startswith("shop/parse.py:1-3 parse_header")
+        bad = rpc(proc, {"jsonrpc": "2.0", "id": 4, "method": "tools/call", "params": {"name": "search", "arguments": {}}})
+        assert bad["result"]["isError"] is True
+        assert rpc(proc, {"jsonrpc": "2.0", "id": 5, "method": "nope"})["error"]["code"] == -32601
+        assert rpc(proc, {"jsonrpc": "2.0", "id": 6, "method": "ping"})["result"] == {}
+    finally:
+        proc.stdin.close()
+        proc.wait(timeout=5)
+    assert proc.returncode == 0
+
+
+def test_the_default_mode_carries_search_and_the_user_can_turn_it_off(monkeypatch, tmp_path):
+    import asyncio
+
+    from mcp.server.fastmcp import FastMCP
+
+    from hanzo_mcp.tools import register_all_tools
+    from hanzo_mcp.tools.common.mode_loader import ModeLoader
+    from hanzo_mcp.tools.common.permissions import PermissionManager
+
+    monkeypatch.setenv("HOME", str(tmp_path))
+    monkeypatch.delenv("HANZO_MCP_ENABLED_TOOLS", raising=False)
+    monkeypatch.delenv("HANZO_MCP_MODE", raising=False)
+    ModeLoader.initialize_modes()
+
+    def names():
+        server = FastMCP("s")
+        register_all_tools(server, PermissionManager())
+        return [t.name for t in asyncio.run(server.list_tools())]
+
+    assert names().count("search") == 1
+    (tmp_path / ".hanzo" / "mcp").mkdir(parents=True)
+    (tmp_path / ".hanzo" / "mcp" / "tool_states.json").write_text('{"search": false}')
+    assert "search" not in names()