Keep the local index fresh from the change feed and improve ranking

- Track a feed cursor in the local index: reindex takes the head cursor
  before listing and commits it with the snapshot; reads refresh from
  /changes (bounded, throttled) and apply creates, updates, archives,
  restores and hard deletes atomically with the cursor. Existing
  cursorless indexes bootstrap tracking on first read. Failures answer
  locally with explicit freshness warnings; 400 triggers a full
  reindex, 404 falls back to timestamp sync.
- Rank by a weighted score: exact identity (including aliases and
  paths), term coverage including body, intent, bm25, authority, a
  status penalty for superseded/completed records and mild recency.
- Return query-centred FTS5 snippets (never for E2E nodes).
- FTS schema v2: porter stemming, prefix index, title column, rowid
  deletes, two-character terms, domain words no longer stopwords;
  v1 indexes upgrade in place.
- Hyphenated alias/path literals match as adjacent-token phrases and
  rank as exact identity.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

tests/test_mcp_index_feed_sync.py

@@ -0,0 +1,404 @@
+"""Change-feed driven local index freshness against a disposable HTTP provider.
+
+The provider implements the node list, single-node read and change-feed
+routes with synthetic data. The real request, index and search code run.
+"""
+
+import copy
+import importlib.util
+import json
+import pathlib
+import re
+import threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from urllib.parse import parse_qs, urlsplit
+
+import pytest
+
+SCRIPT = pathlib.Path(__file__).resolve().parents[1] / "resources/scripts/napseer_mcp_server.py"
+
+
+def make_node(name, minute, *, content=None, archived=False):
+    stamp = f"2026-09-01T12:{minute:02d}:00Z"
+    return {
+        "id": name, "project_id": "project-1", "full_path": f"/notes/{name}",
+        "folder_path": "/notes", "name": name, "type": "note",
+        "metadata": {}, "tags": [], "aliases": [], "links": [],
+        "content_text": content if content is not None else name,
+        "updated_at": stamp, "archived_at": stamp if archived else None,
+    }
+
+
+class Provider:
+    def __init__(self):
+        self.nodes = {}
+        self.events = []
+        self.requests = []
+        self.fail = {}  # route kind -> HTTP status
+        self.on_list = None
+
+    def put(self, node, kind):
+        self.nodes[node["id"]] = copy.deepcopy(node)
+        self.events.append({"type": kind, "node_id": node["id"], "full_path": node["full_path"],
+                            "new_revision": node["updated_at"]})
+
+    def delete(self, node_id):
+        node = self.nodes.pop(node_id)
+        self.events.append({"type": "node_deleted", "node_id": node_id, "full_path": node["full_path"],
+                            "new_revision": None})
+
+    def count(self, kind):
+        return sum(1 for item in self.requests if item[0] == kind)
+
+
+@pytest.fixture
+def feed(tmp_path, monkeypatch):
+    provider = Provider()
+
+    class Handler(BaseHTTPRequestHandler):
+        def log_message(self, *_args):
+            pass
+
+        def reply(self, status, payload):
+            self.send_response(status)
+            self.send_header("Content-Type", "application/json")
+            self.end_headers()
+            self.wfile.write(json.dumps(payload).encode())
+
+        def do_GET(self):
+            url = urlsplit(self.path)
+            query = parse_qs(url.query)
+            if url.path.endswith("/changes"):
+                kind = "changes"
+            elif url.path.endswith("/nodes"):
+                kind = "list"
+            else:
+                kind = "get"
+            provider.requests.append((kind, url.path, query))
+            if kind in provider.fail:
+                status = provider.fail[kind]
+                return self.reply(status, {"error": {"code": "test_failure", "message": "injected"}})
+            if kind == "changes":
+                after = query.get("after", ["head"])[0]
+                if after == "head":
+                    start = len(provider.events)
+                elif re.fullmatch(r"c\d+", after):
+                    start = int(after[1:])
+                else:
+                    return self.reply(400, {"error": {"code": "invalid_cursor", "message": "bad cursor"}})
+                limit = int(query.get("limit", ["100"])[0])
+                page = provider.events[start:start + limit]
+                items = [{**event, "cursor": f"c{start + index + 1}", "project_id": "project-1",
+                          "changed_fields": [], "created_at": "2026-09-01T12:00:00Z"}
+                         for index, event in enumerate(page)]
+                end = start + len(page)
+                return self.reply(200, {"items": items, "next_cursor": f"c{end}",
+                                        "has_more": end < len(provider.events)})
+            if kind == "list":
+                if provider.on_list is not None:
+                    hook, provider.on_list = provider.on_list, None
+                    hook()
+                rows = [copy.deepcopy(n) for n in provider.nodes.values() if not n["archived_at"]]
+                rows.sort(key=lambda n: (n["updated_at"], n["id"]), reverse=True)
+                after = query.get("updated_after", [""])[0]
+                if after:
+                    rows = [n for n in provider.nodes.values() if n["updated_at"] > after]
+                return self.reply(200, {"items": rows, "next_cursor": None})
+            node_id = url.path.rsplit("/", 1)[-1]
+            if node_id not in provider.nodes:
+                return self.reply(404, {"error": {"code": "not_found", "message": "missing"}})
+            return self.reply(200, copy.deepcopy(provider.nodes[node_id]))
+
+    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    monkeypatch.setenv("NAPSEER_TOOL_PROFILES", "all")
+    monkeypatch.syspath_prepend(str(SCRIPT.parent))
+    spec = importlib.util.spec_from_file_location("feed_sync_test_worker", SCRIPT)
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    mod.AUTH_DIR = tmp_path / "project-state"
+    mod.INDEX_PATH = mod.AUTH_DIR / "index.sqlite"
+    mod.INDEX_LOCK_PATH = mod.AUTH_DIR / "index.lock"
+    mod.BASE_URL = f"http://127.0.0.1:{server.server_port}"
+    mod.TOKEN = "disposable-test-token"
+    mod.AUTH = {"account_id": "test-account"}
+    mod.DEFAULT_PROJECT_ID = "project-1"
+    mod.CONFIGURED_TOOL_PROFILES = {"all"}
+    mod.refresh_public_auth_state = lambda: None
+    mod.require_unlocked = lambda _name: None
+    mod.resolve_project_id = lambda args: args.get("project_id", "project-1")
+    mod.project_encryption_state = lambda _project_id: "standard"
+    try:
+        yield mod, provider
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=2)
+
+
+def local_ids(mod, text):
+    return [item["id"] for item in mod.search_local_index({"q": text, "limit": 10})["items"]]
+
+
+def seed(mod, provider, *names):
+    for index, name in enumerate(names):
+        provider.put(make_node(name, index), "node_created")
+    result = mod.reindex_project({})
+    assert result["graph_complete"] is True
+    return result
+
+
+def test_reindex_takes_head_cursor_before_listing_and_replays_idempotently(feed):
+    mod, provider = feed
+    provider.put(make_node("alpha", 1), "node_created")
+
+    def concurrent_write():
+        provider.put(make_node("alpha", 2, content="alpha concurrent revision"), "node_updated")
+
+    provider.on_list = concurrent_write
+    mod.reindex_project({})
+
+    kinds = [item[0] for item in provider.requests]
+    assert kinds[:2] == ["changes", "list"]
+    assert provider.requests[0][2]["after"] == ["head"]
+    assert mod.local_index_feed_state("project-1")["feed_cursor"] == "c1"
+    assert local_ids(mod, "concurrent") == ["alpha"]
+
+    result = mod.sync_local_index_from_feed("project-1")
+
+    assert (result["events"], result["skipped_current"], result["indexed"]) == (1, 1, 0)
+    assert provider.count("get") == 0  # revision already indexed: no refetch
+    assert mod.local_index_feed_state("project-1")["feed_cursor"] == "c2"
+
+
+def test_feed_applies_create_update_archive_restore_and_delete(feed):
+    mod, provider = feed
+    seed(mod, provider, "keepme", "archiveme", "deleteme", "restoreme")
+    provider.put(make_node("restoreme", 5, archived=True), "node_archived")
+    provider.put(make_node("freshnode", 10, content="brandnewterm"), "node_created")
+    provider.put(make_node("keepme", 11, content="keepme revisedterm"), "node_updated")
+    provider.put(make_node("archiveme", 12, archived=True), "node_archived")
+    provider.delete("deleteme")
+    provider.put(make_node("restoreme", 13), "node_restored")
+
+    result = mod.sync_local_index_from_feed("project-1")
+
+    assert result["status"] == "synced"
+    assert local_ids(mod, "brandnewterm") == ["freshnode"]
+    assert local_ids(mod, "revisedterm") == ["keepme"]
+    assert local_ids(mod, "archiveme") == []
+    assert local_ids(mod, "deleteme") == []
+    assert local_ids(mod, "restoreme") == ["restoreme"]
+    with mod.index_connect() as conn:
+        ids = {row[0] for row in conn.execute("SELECT node_id FROM local_index_nodes")}
+    # Archived rows match a full rebuild, which lists only active nodes.
+    assert ids == {"keepme", "freshnode", "restoreme"}
+    assert mod.local_index_feed_state("project-1")["feed_cursor"] == f"c{len(provider.events)}"
+
+
+def test_feed_failure_never_advances_cursor_and_retry_recovers(feed):
+    mod, provider = feed
+    seed(mod, provider, "stable")
+    before = mod.local_index_feed_state("project-1")["feed_cursor"]
+    provider.put(make_node("newcomer", 10, content="latecomerterm"), "node_created")
+    provider.fail["get"] = 503
+
+    with pytest.raises(mod.SafeToolError):
+        mod.sync_local_index_from_feed("project-1")
+
+    assert mod.local_index_feed_state("project-1")["feed_cursor"] == before
+    assert local_ids(mod, "latecomerterm") == []
+    del provider.fail["get"]
+    mod.sync_local_index_from_feed("project-1")
+    assert local_ids(mod, "latecomerterm") == ["newcomer"]
+
+
+def test_invalid_feed_cursor_triggers_full_reindex(feed):
+    mod, provider = feed
+    seed(mod, provider, "stable")
+    with mod.index_connect() as conn:
+        conn.execute("UPDATE local_index_state SET feed_cursor = 'expired-cursor'")
+    provider.nodes["hardgone"] = make_node("hardgone", 3)  # present only in the full listing
+
+    result = mod.sync_local_index({})
+
+    assert result["mode"] == "full_reindex"
+    assert result["reason"] == "feed_cursor_invalid"
+    assert local_ids(mod, "hardgone") == ["hardgone"]
+    assert mod.local_index_feed_state("project-1")["feed_cursor"] == f"c{len(provider.events)}"
+
+
+def test_old_server_without_feed_falls_back_to_timestamp_sync(feed):
+    mod, provider = feed
+    seed(mod, provider, "stable")
+    provider.fail["changes"] = 404
+    provider.nodes["later"] = make_node("later", 30, content="timestampterm")
+
+    result = mod.sync_local_index({})
+
+    assert result["mode"] == "incremental"
+    assert result["feed"] == "unsupported"
+    assert local_ids(mod, "timestampterm") == ["later"]
+
+
+def test_reindex_on_old_server_without_feed_still_rebuilds(feed):
+    mod, provider = feed
+    provider.fail["changes"] = 404
+    provider.put(make_node("legacy", 1), "node_created")
+
+    result = mod.reindex_project({})
+
+    assert result["graph_complete"] is True
+    assert local_ids(mod, "legacy") == ["legacy"]
+    assert mod.local_index_feed_state("project-1")["feed_cursor"] is None
+
+
+def test_search_auto_syncs_from_feed_and_throttles(feed):
+    mod, provider = feed
+    seed(mod, provider, "stable")
+    provider.put(make_node("autosync", 10, content="autosyncterm"), "node_created")
+
+    first = mod.call_tool_impl("nap_discover", {"q": "autosyncterm", "mode": "local"})
+    provider.put(make_node("throttled", 11, content="throttledterm"), "node_created")
+    second = mod.call_tool_impl("nap_discover", {"q": "throttledterm", "mode": "local"})
+
+    assert [item["node_id"] for item in first["items"]] == ["autosync"]
+    assert first["freshness"]["stale"] is False
+    assert first["freshness"]["feed_synced_at"]
+    assert second["items"] == []  # within the throttle window no second feed read
+    assert provider.count("changes") == 2  # reindex head + one automatic sync
+    assert second["freshness"]["feed_sync"] == "throttled"
+
+
+def test_search_answers_locally_and_marks_stale_when_feed_unavailable(feed):
+    mod, provider = feed
+    seed(mod, provider, "stable")
+    provider.fail["changes"] = 503
+
+    result = mod.call_tool_impl("nap_discover", {"q": "stable", "mode": "local"})
+
+    assert [item["node_id"] for item in result["items"]] == ["stable"]
+    assert result["freshness"]["stale"] is True
+    assert result["freshness"]["reason"] == "server_unavailable"
+    assert result["warnings"][0]["code"] == "server_unavailable"
+
+
+def test_context_local_read_auto_syncs_from_feed(feed):
+    mod, provider = feed
+    seed(mod, provider, "stable")
+    provider.put(make_node("contextnew", 10), "node_created")
+
+    result = mod.call_tool_impl("nap_context", {"paths": ["/notes/contextnew"], "mode": "hybrid"})
+
+    assert provider.count("changes") == 2
+    assert "contextnew" in json.dumps(result)
+    assert result["freshness"]["feed_sync"] == "synced"
+
+
+def cursorless_complete_index(mod, provider, *names):
+    for index, name in enumerate(names):
+        node = make_node(name, index)
+        provider.put(node, "node_created")
+        mod.index_node(node)
+    mod.set_local_graph_index_complete("project-1", True)
+
+
+def test_cursorless_index_bootstraps_on_first_search_then_refreshes(feed):
+    mod, provider = feed
+    cursorless_complete_index(mod, provider, "manual")
+    provider.nodes["gapnode"] = make_node("gapnode", 20, content="gapterm")  # changed before tracking
+
+    first = mod.call_tool_impl("nap_discover", {"q": "gapterm", "mode": "local"})
+
+    assert [kind for kind, *_ in provider.requests] == ["changes", "list"]
+    assert provider.requests[0][2]["after"] == ["head"]
+    assert [item["node_id"] for item in first["items"]] == ["gapnode"]
+    assert first["freshness"]["feed_sync"] == "tracking_initialized"
+    assert not first.get("warnings")
+    assert mod.local_index_feed_state("project-1")["feed_cursor"] == "c1"
+
+    provider.put(make_node("manual", 30, content="manual remoteupdateterm"), "node_updated")
+    throttled = mod.call_tool_impl("nap_discover", {"q": "remoteupdateterm", "mode": "local"})
+    assert throttled["items"] == [] and len(provider.requests) == 2  # no HTTP inside the window
+
+    mod.FEED_AUTO_SYNC_STATE.clear()  # the throttle window elapses
+    later = mod.call_tool_impl("nap_discover", {"q": "remoteupdateterm", "mode": "local"})
+    assert [item["node_id"] for item in later["items"]] == ["manual"]
+    assert later["freshness"]["feed_sync"] == "synced"
+
+
+def test_bootstrap_failure_answers_locally_warns_and_does_not_retry(feed):
+    mod, provider = feed
+    cursorless_complete_index(mod, provider, "manual")
+    provider.fail["changes"] = 503
+
+    first = mod.call_tool_impl("nap_discover", {"q": "manual", "mode": "local"})
+    second = mod.call_tool_impl("nap_discover", {"q": "manual", "mode": "local"})
+
+    for result in (first, second):
+        assert [item["node_id"] for item in result["items"]] == ["manual"]
+        assert [w["code"] for w in result["warnings"]] == ["index_not_tracking"]
+        assert result["warnings"][0]["next_action"] == "nap_index_sync"
+        assert result["freshness"]["stale"] is True
+    assert provider.count("changes") == 1  # one attempt per retry window
+    assert provider.count("list") == 0
+    assert mod.local_index_feed_state("project-1")["feed_cursor"] is None
+
+
+def test_bootstrap_budget_exhaustion_stays_untracked_and_applies_nothing(feed, monkeypatch):
+    mod, provider = feed
+    cursorless_complete_index(mod, provider, "manual")
+    provider.nodes["gapnode"] = make_node("gapnode", 20, content="gapterm")
+    monkeypatch.setattr(mod, "FEED_AUTO_SYNC_MAX_PAGES", 0)
+
+    result = mod.call_tool_impl("nap_discover", {"q": "gapterm", "mode": "local"})
+
+    assert result["items"] == []
+    assert [w["code"] for w in result["warnings"]] == ["index_not_tracking"]
+    assert mod.local_index_feed_state("project-1")["feed_cursor"] is None
+
+
+def test_bootstrap_on_old_server_marks_unsupported_without_retry(feed):
+    mod, provider = feed
+    cursorless_complete_index(mod, provider, "manual")
+    provider.fail["changes"] = 404
+
+    first = mod.call_tool_impl("nap_discover", {"q": "manual", "mode": "local"})
+    mod.call_tool_impl("nap_discover", {"q": "manual", "mode": "local"})
+
+    assert first["freshness"]["feed_sync"] == "unsupported"
+    assert not first.get("warnings")
+    assert provider.count("changes") == 1 and provider.count("list") == 0
+
+
+def test_feed_sync_is_bounded_by_pages_and_resumes(feed):
+    mod, provider = feed
+    seed(mod, provider, "stable")
+    start = len(provider.events)
+    for index in range(3):
+        provider.put(make_node(f"bounded{index}", 10 + index), "node_created")
+
+    first = mod.sync_local_index_from_feed("project-1", max_pages=2, page_limit=1)
+
+    assert (first["status"], first["pages"], first["has_more"]) == ("partial", 2, True)
+    assert mod.local_index_feed_state("project-1")["feed_cursor"] == f"c{start + 2}"
+    second = mod.sync_local_index_from_feed("project-1", max_pages=5, page_limit=1)
+    assert second["status"] == "synced"
+    assert local_ids(mod, "bounded2") == ["bounded2"]
+
+
+def test_explicit_sync_initializes_feed_tracking_for_cursorless_index(feed):
+    mod, provider = feed
+    provider.put(make_node("manual", 1), "node_created")
+    mod.index_node(make_node("manual", 1))
+    mod.set_local_graph_index_complete("project-1", True)
+
+    result = mod.sync_local_index({})
+
+    assert result["mode"] == "incremental"
+    assert result["feed"] == "initialized"
+    assert mod.local_index_feed_state("project-1")["feed_cursor"] == "c1"
+    provider.delete("manual")
+    mod.sync_local_index({})
+    assert local_ids(mod, "manual") == []  # hard delete observed through the feed

tests/test_mcp_index_rebuild_recovery.py

@@ -34,6 +34,12 @@ def memory(tmp_path, monkeypatch):
             pass
 
         def do_GET(self):
+            if urlsplit(self.path).path.endswith("/changes"):
+                # Model a server without the change feed; the rebuild must not depend on it.
+                self.send_response(404)
+                self.end_headers()
+                self.wfile.write(b'{"error":{"code":"not_found","message":"no change feed"}}')
+                return
             query = parse_qs(urlsplit(self.path).query)
             provider["requests"].append(query)
             offset = int(query.get("cursor", ["0"])[0])

tests/test_mcp_memory_ergonomics_helpers.py

@@ -886,6 +886,9 @@ def test_index_sync_fetches_incremental_changes_and_removes_archived_nodes(mod):
         calls = []
 
         def fake_request(method, path, payload=None, **kwargs):
+            if "/changes?" in path:
+                # Server without the change feed: sync uses the timestamp path.
+                raise mod.SafeToolError("not_found", "no change feed", status=404)
             calls.append(path)
             assert method == "GET"
             assert "view=full" in path

tests/test_mcp_reliability_contracts.py

@@ -211,11 +211,19 @@ def record_revoked_grant(mod):
 
 def test_local_answers_warn_when_login_grant_is_revoked(mod):
     build_complete_index(mod)
-    mod.request_json = lambda *a, **k: (_ for _ in ()).throw(AssertionError("no HTTP expected"))
 
+    def empty_feed(_method, path, *_args, **_kwargs):
+        if "/changes?" in path:
+            return {"items": [], "next_cursor": "c0", "has_more": False}
+        return {"items": [], "next_cursor": None}
+
+    # A healthy read starts change-feed tracking for the cursorless index.
+    mod.request_json = empty_feed
     healthy = mod.discover_memory({"folder_path": "/plans", "view": "paths"})
     assert not healthy.get("warnings")
     assert not (healthy.get("freshness") or {}).get("stale")
+    assert healthy["freshness"]["feed_sync"] == "tracking_initialized"
+    mod.request_json = lambda *a, **k: (_ for _ in ()).throw(AssertionError("no HTTP expected"))
 
     record_revoked_grant(mod)
     state_text = (mod.AUTH_DIR / "auth-failure.json").read_text()

tests/test_mcp_search_latency.py

@@ -153,6 +153,18 @@ def indexed_node(path, *, node_type="note", tags=None, status="active", content=
     }
 
 
+def feed_provider(calls):
+    """Synthetic server: empty change feed and no newer nodes."""
+
+    def request(method, path, payload=None, **_kwargs):
+        calls.append((method, path))
+        if "/changes?" in path:
+            return {"items": [], "next_cursor": "c0", "has_more": False}
+        return {"items": [], "next_cursor": None}
+
+    return request
+
+
 def configure_local_index(mod, tmp_path):
     mod.AUTH_DIR = tmp_path
     mod.INDEX_PATH = tmp_path / "index.sqlite"
@@ -208,11 +220,14 @@ def test_complete_local_compact_browse_uses_keyset_cursor_without_http(tmp_path)
         item["updated_at"] = f"2026-08-2{index}T00:00:00Z"
         mod.index_node(item)
     mod.set_local_graph_index_complete("project-1", True)
+    calls = []
+    mod.TOKEN = "disposable-test-token"  # synthetic login; requests go to feed_provider only
+    mod.request_json = feed_provider(calls)
+    # The first read starts change-feed tracking for this cursorless index.
+    mod.discover_memory({"folder_path": "/plans", "limit": 1, "view": "paths"})
+    assert [path.split("?")[0].rsplit("/", 1)[-1] for _method, path in calls] == ["changes", "nodes"]
+    calls.clear()
 
-    def unexpected_request(*_args, **_kwargs):
-        raise AssertionError("complete local compact browse must not make an HTTP request")
-
-    mod.request_json = unexpected_request
     first = mod.discover_memory(
         {
             "folder_path": "/plans",
@@ -241,6 +256,9 @@ def test_complete_local_compact_browse_uses_keyset_cursor_without_http(tmp_path)
     )
     assert not ({item["node_id"] for item in first["items"]} & {item["node_id"] for item in second["items"]})
     assert all(item["node_id"] for item in first["items"] + second["items"])
+    # Steady state inside the refresh throttle window: no HTTP at all.
+    assert calls == []
+    assert first["freshness"]["feed_sync"] == "throttled"
 
 
 def test_local_compact_browse_cursor_is_bound_to_filters(tmp_path):
@@ -482,11 +500,13 @@ def test_complete_filtered_local_search_stays_off_network(tmp_path):
         )
     )
     mod.set_local_graph_index_complete("project-1", True)
+    calls = []
+    mod.TOKEN = "disposable-test-token"  # synthetic login; requests go to feed_provider only
+    mod.request_json = feed_provider(calls)
+    mod.search_memory({"q": "facturación", "tag": "billing", "limit": 1, "view": "summary"})
+    assert len(calls) == 2  # one-time change-feed bootstrap: head + catch-up
+    calls.clear()
 
-    def unexpected_request(*_args, **_kwargs):
-        raise AssertionError("a complete filtered local result must not use the network")
-
-    mod.request_json = unexpected_request
     result = mod.search_memory(
         {
             "q": "facturación",
@@ -500,6 +520,8 @@ def test_complete_filtered_local_search_stays_off_network(tmp_path):
     assert [item["full_path"] for item in result["items"]] == [
         "/documentation/facturacion-electronica"
     ]
+    assert calls == []  # a complete, tracked, recently refreshed index stays off the network
+    assert not result.get("warnings")
 
 
 def test_discovery_schema_documents_dynamic_sort_and_intent():

tests/test_mcp_search_quality.py

@@ -0,0 +1,308 @@
+"""Local search quality: FTS schema v2, weighted ranking and query snippets.
+
+Every test runs the real index, query planner and ranking code against a
+disposable SQLite file; no network is used.
+"""
+
+import importlib.util
+import pathlib
+import sqlite3
+import sys
+
+import pytest
+
+SCRIPT = pathlib.Path(__file__).resolve().parents[1] / "resources" / "scripts" / "napseer_mcp_server.py"
+
+
+def load_module(tmp_path):
+    sys.path.insert(0, str(SCRIPT.parent))
+    spec = importlib.util.spec_from_file_location("napseer_mcp_server_search_quality_test", SCRIPT)
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    mod.AUTH_DIR = tmp_path
+    mod.INDEX_PATH = tmp_path / "index.sqlite"
+    mod.INDEX_LOCK_PATH = tmp_path / "index.lock"
+    mod.resolve_project_id = lambda _args: "project-1"
+    mod.project_encryption_state = lambda _project_id: "standard"
+
+    def unexpected_request(*_args, **_kwargs):
+        raise AssertionError("local search quality tests must not use the network")
+
+    mod.request_json = unexpected_request
+    return mod
+
+
+def indexed_node(path, *, content="", status="active", updated_at="2026-08-22T00:00:00Z", **extra):
+    name = path.rsplit("/", 1)[-1]
+    node = {
+        "id": f"node-{path.strip('/').replace('/', '-')}",
+        "project_id": "project-1",
+        "full_path": path,
+        "folder_path": path.rsplit("/", 1)[0] or None,
+        "name": name,
+        "title": name,
+        "type": "note",
+        "tags": [],
+        "aliases": [],
+        "links": [],
+        "metadata": {"status": status},
+        "content_text": content,
+        "encryption_state": "legacy_plaintext",
+        "updated_at": updated_at,
+    }
+    node.update(extra)
+    return node
+
+
+def paths(result):
+    return [item["full_path"] for item in result["items"]]
+
+
+# --- FTS schema v2 ---------------------------------------------------------
+
+
+def test_stemming_recalls_other_inflections(tmp_path):
+    mod = load_module(tmp_path)
+    mod.index_node(indexed_node("/notes/release-log", content="The deployment pipeline promotes images."))
+
+    assert paths(mod.search_local_index({"q": "deploying", "limit": 5})) == ["/notes/release-log"]
+    assert paths(mod.search_local_index({"q": "configured deployments", "limit": 5})) == ["/notes/release-log"]
+
+
+def test_prefix_query_uses_prefix_index_and_matches(tmp_path):
+    mod = load_module(tmp_path)
+    mod.index_node(indexed_node("/notes/cluster", content="kubernetes control plane"))
+
+    with mod.index_connect() as conn:
+        sql = conn.execute(
+            "SELECT sql FROM sqlite_master WHERE name = 'local_index_fts'"
+        ).fetchone()["sql"]
+    assert "porter unicode61 remove_diacritics 2" in sql
+    assert "prefix = '2 3'" in sql
+    assert paths(mod.search_local_index({"q": "kuber", "limit": 5})) == ["/notes/cluster"]
+
+
+def test_two_character_terms_are_searchable(tmp_path):
+    mod = load_module(tmp_path)
+    mod.index_node(indexed_node("/notes/note-a", content="db migration runbook for postgres cutover steps"))
+    mod.index_node(indexed_node("/notes/note-b", content="migration", updated_at="2026-09-01T00:00:00Z"))
+
+    analysis = mod.analyze_search_query("db migration")
+    result = mod.search_local_index({"q": "db migration", "limit": 5})
+
+    assert "db" in analysis["terms"]
+    assert paths(result)[0] == "/notes/note-a"
+
+
+def test_domain_words_are_not_stop_words():
+    mod = load_module(pathlib.Path("/nonexistent"))
+    concepts = mod.analyze_search_query("plan work decide")["concepts"]
+
+    assert {"plan", "work", "decide"} <= set(concepts)
+
+
+def test_single_node_fts_delete_uses_rowid_not_scan(tmp_path, monkeypatch):
+    mod = load_module(tmp_path)
+    node = indexed_node("/notes/rowid-target", content="first version")
+    mod.index_node(node)
+    statements = []
+    original_open = mod.open_index_connection
+
+    def traced_open():
+        conn = original_open()
+        conn.set_trace_callback(statements.append)
+        return conn
+
+    monkeypatch.setattr(mod, "open_index_connection", traced_open)
+    mod.index_node({**node, "content_text": "second version"})
+    mod.remove_indexed_node(node["id"])
+
+    fts_deletes = [s for s in statements if "DELETE FROM local_index_fts" in s]
+    assert fts_deletes
+    assert all("rowid" in s and "node_id" not in s for s in fts_deletes)
+    with mod.index_connect() as conn:
+        assert conn.execute("SELECT COUNT(*) FROM local_index_fts").fetchone()[0] == 0
+        assert conn.execute("SELECT COUNT(*) FROM local_index_nodes").fetchone()[0] == 0
+
+
+def test_reindexing_a_node_replaces_its_single_fts_row(tmp_path):
+    mod = load_module(tmp_path)
+    node = indexed_node("/notes/replaced", content="alpha")
+    mod.index_node(node)
+    mod.index_node({**node, "content_text": "omega"})
+
+    with mod.index_connect() as conn:
+        rows = conn.execute("SELECT rowid, content_text FROM local_index_fts").fetchall()
+        mapped = conn.execute("SELECT fts_rowid FROM local_index_nodes").fetchone()[0]
+    assert [row["content_text"] for row in rows] == ["omega"]
+    assert rows[0]["rowid"] == mapped
+
+
+V1_SCHEMA = [
+    """CREATE TABLE local_index_nodes (
+        node_id TEXT PRIMARY KEY, project_id TEXT NOT NULL, full_path TEXT NOT NULL,
+        name TEXT NOT NULL, folder_path TEXT, node_type TEXT NOT NULL, tags_json TEXT NOT NULL,
+        updated_at TEXT NOT NULL, indexed_at TEXT NOT NULL, title TEXT, status TEXT,
+        archived INTEGER NOT NULL DEFAULT 0, aliases_json TEXT NOT NULL DEFAULT '[]')""",
+    """CREATE TABLE local_index_state (
+        project_id TEXT PRIMARY KEY, graph_complete INTEGER NOT NULL DEFAULT 0, graph_completed_at TEXT)""",
+    """CREATE VIRTUAL TABLE local_index_fts USING fts5(
+        node_id UNINDEXED, project_id UNINDEXED, full_path, name, folder_path, node_type,
+        tags, aliases, links, metadata_text, content_text)""",
+]
+
+
+def test_v1_index_upgrades_in_place_and_keeps_completeness(tmp_path):
+    index_path = tmp_path / "index.sqlite"
+    conn = sqlite3.connect(str(index_path))
+    for statement in V1_SCHEMA:
+        conn.execute(statement)
+    conn.execute(
+        "INSERT INTO local_index_nodes VALUES ('n1','project-1','/notes/legacy','legacy','/notes','note','[]',"
+        "'2026-08-01T00:00:00Z','2026-08-01T00:00:00Z','Legacy title','active',0,'[]')"
+    )
+    conn.execute(
+        "INSERT INTO local_index_fts VALUES ('n1','project-1','/notes/legacy','legacy','/notes','note','','','',"
+        "'','the deployment runbook body')"
+    )
+    conn.execute("INSERT INTO local_index_state VALUES ('project-1', 1, '2026-08-01T00:00:00Z')")
+    conn.commit()
+    conn.close()
+
+    mod = load_module(tmp_path)
+    result = mod.search_local_index({"q": "deploying", "limit": 5})
+
+    assert paths(result) == ["/notes/legacy"]
+    with mod.index_connect() as conn:
+        assert conn.execute("PRAGMA user_version").fetchone()[0] == mod.INDEX_SCHEMA_VERSION
+        assert conn.execute("SELECT COUNT(*) FROM local_index_fts").fetchone()[0] == 1
+        assert "porter" in conn.execute(
+            "SELECT sql FROM sqlite_master WHERE name = 'local_index_fts'"
+        ).fetchone()[0]
+        assert conn.execute("SELECT fts_rowid FROM local_index_nodes").fetchone()[0] is not None
+    assert mod.local_index_diagnostics("project-1")["graph_complete"] is True
+    # The upgrade drops no rows and a legacy-migrated index has no feed cursor yet.
+    assert mod.local_index_feed_state("project-1")["feed_cursor"] is None
+
+
+# --- ranking --------------------------------------------------------------
+
+
+def test_body_full_coverage_beats_authority_path_partial_match(tmp_path):
+    mod = load_module(tmp_path)
+    mod.index_node(indexed_node("/rules/quasar", content="unrelated governance text"))
+    mod.index_node(
+        indexed_node(
+            "/work/current/notes-a",
+            content="The quasar rollout checklist is kept here with every gate listed.",
+        )
+    )
+
+    result = mod.search_local_index({"q": "quasar rollout checklist", "limit": 5})
+
+    assert paths(result) == ["/work/current/notes-a", "/rules/quasar"]
+
+
+def test_superseded_status_is_demoted_unless_requested(tmp_path):
+    mod = load_module(tmp_path)
+    mod.index_node(
+        indexed_node("/decisions/retry-a", content="gateway retry budget", status="superseded",
+                     updated_at="2026-09-10T00:00:00Z")
+    )
+    mod.index_node(
+        indexed_node("/decisions/retry-b", content="gateway retry budget", status="active",
+                     updated_at="2026-08-01T00:00:00Z")
+    )
+
+    default = mod.search_local_index({"q": "gateway retry budget", "limit": 5})
+    requested = mod.search_local_index({"q": "superseded gateway retry budget", "limit": 5})
+
+    assert paths(default) == ["/decisions/retry-b", "/decisions/retry-a"]
+    assert paths(requested)[0] == "/decisions/retry-a"
+
+
+def test_explicit_exact_identity_stays_on_top_of_requested_intent(tmp_path):
+    mod = load_module(tmp_path)
+    mod.index_node(indexed_node("/notes/cache-invalidation", content="cache invalidation"))
+    mod.index_node(indexed_node("/decisions/cache-layer", content="cache invalidation " * 5))
+
+    result = mod.search_local_index({"q": "cache invalidation", "intent": "decision", "limit": 5})
+
+    assert paths(result)[0] == "/notes/cache-invalidation"
+    assert "exact_identity" in result["items"][0]["match_reason"]
+
+
+# --- snippets -------------------------------------------------------------
+
+
+def test_local_search_returns_query_centred_snippet(tmp_path):
+    mod = load_module(tmp_path)
+    filler = "Unrelated operational background sentence. " * 20
+    mod.index_node(
+        indexed_node("/notes/long-note", content=f"{filler}The zanzibar failover drill runs weekly. {filler}")
+    )
+    mod.set_local_graph_index_complete("project-1", True)
+
+    local = mod.search_local_index({"q": "zanzibar", "limit": 5})
+    envelope = mod.search_memory({"q": "zanzibar", "limit": 5, "view": "summary", "mode": "local"})
+
+    assert "zanzibar" in local["items"][0]["snippet"].lower()
+    assert "zanzibar" in envelope["items"][0]["snippet"].lower()
+    assert "zanzibar" in envelope["items"][0]["preview"].lower()
+
+
+def test_e2e_nodes_have_no_snippet_or_body_index(tmp_path):
+    mod = load_module(tmp_path)
+    mod.index_node(
+        indexed_node(
+            "/notes/zanzibar-sealed",
+            content="plaintext-secret-marker zanzibar",
+            encryption_state="encrypted",
+            encrypted_content_envelope={"ciphertext": "opaque"},
+        )
+    )
+
+    assert mod.search_local_index({"q": "plaintext-secret-marker", "limit": 5})["items"] == []
+    result = mod.search_local_index({"q": "zanzibar", "limit": 5})
+    assert paths(result) == ["/notes/zanzibar-sealed"]
+    assert not result["items"][0].get("snippet")
+
+
+@pytest.mark.parametrize("query", ["memory-source-policy", "memory source policy"])
+def test_hyphenated_identity_query_matches_tokenized_path(tmp_path, query):
+    mod = load_module(tmp_path)
+    mod.index_node(indexed_node("/rules/memory-source-policy", content="Canonical memory rule."))
+    mod.index_node(indexed_node("/notes/other", content="policy about memory elsewhere"))
+
+    assert paths(mod.search_local_index({"q": query, "limit": 5}))[0] == "/rules/memory-source-policy"
+
+
+@pytest.mark.parametrize(
+    "query, expected",
+    [
+        ("existing-alias", "node-notes-first"),
+        ("secondaryhandle", "node-notes-second"),
+        ("/notes/existing-work", "node-notes-existing-work"),
+        ("existing-work", "node-notes-existing-work"),
+    ],
+)
+def test_discover_local_literal_alias_or_path_is_top_exact_identity(tmp_path, query, expected):
+    mod = load_module(tmp_path)
+    mod.CONFIGURED_TOOL_PROFILES = {"all"}
+    mod.require_unlocked = lambda _name: None
+    mod.index_node(indexed_node("/notes/first", content="body", aliases=["existing-alias"]))
+    mod.index_node(indexed_node("/notes/second", content="body", aliases=["secondaryhandle"]))
+    mod.index_node(indexed_node("/notes/existing-work", content="body"))
+    # Authoritative distractor whose body repeats the words of every literal.
+    mod.index_node(
+        indexed_node(
+            "/rules/distractor",
+            content="existing alias existing work secondaryhandle notes " * 4,
+        )
+    )
+    mod.set_local_graph_index_complete("project-1", True)
+
+    result = mod.call_tool_impl("nap_discover", {"q": query, "mode": "local", "view": "summary"})
+
+    assert result["items"][0]["node_id"] == expected
+    assert "exact_identity" in result["items"][0]["match_reason"]