nap-cli/tests/test_mcp_search_quality.py

"""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"]