From 0938b3026edb46e477b87814bff16fc9486747aa Mon Sep 17 00:00:00 2001 From: HumbleTurtle Date: Mon, 28 Sep 2026 22:09:12 -0300 Subject: [PATCH] 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) --- resources/scripts/napseer_mcp_server.py | 1032 +++++++++++++++++-- tests/test_mcp_index_feed_sync.py | 404 ++++++++ tests/test_mcp_index_rebuild_recovery.py | 6 + tests/test_mcp_memory_ergonomics_helpers.py | 3 + tests/test_mcp_reliability_contracts.py | 10 +- tests/test_mcp_search_latency.py | 38 +- tests/test_mcp_search_quality.py | 308 ++++++ 7 files changed, 1708 insertions(+), 93 deletions(-) create mode 100644 tests/test_mcp_index_feed_sync.py create mode 100644 tests/test_mcp_search_quality.py diff --git a/resources/scripts/napseer_mcp_server.py b/resources/scripts/napseer_mcp_server.py index 53ceaf9..d3426dd 100755 --- a/resources/scripts/napseer_mcp_server.py +++ b/resources/scripts/napseer_mcp_server.py @@ -6229,6 +6229,8 @@ def summarize_node(node, view="summary", preview_chars=240): item["preview"] = preview_text(content, preview_chars) for key in ( "search_source", + "snippet", + "matched_terms", "matched_query", "matched_strategy", "matched_fields", @@ -6751,6 +6753,12 @@ def initialize_index_schema(conn): ensure_sqlite_column(conn, "local_index_nodes", "status", "TEXT") ensure_sqlite_column(conn, "local_index_nodes", "archived", "INTEGER NOT NULL DEFAULT 0") ensure_sqlite_column(conn, "local_index_nodes", "aliases_json", "TEXT NOT NULL DEFAULT '[]'") + # Schema v2: server revision (skip redundant feed refetches), the FTS + # rowid (single-node FTS deletes by rowid, not a scan) and whether the + # body was withheld from the index (E2E: never snippet or index plaintext). + ensure_sqlite_column(conn, "local_index_nodes", "revision", "TEXT") + ensure_sqlite_column(conn, "local_index_nodes", "fts_rowid", "INTEGER") + ensure_sqlite_column(conn, "local_index_nodes", "content_withheld", "INTEGER NOT NULL DEFAULT 0") conn.execute( "CREATE INDEX IF NOT EXISTS local_index_nodes_project_path " "ON local_index_nodes(project_id, full_path, archived)" @@ -6821,6 +6829,10 @@ def initialize_index_schema(conn): ) """ ) + # Opaque change-feed cursor committed with the rows it covers. It is + # separate from the per-worktree nap_changes_since agent cursor. + ensure_sqlite_column(conn, "local_index_state", "feed_cursor", "TEXT") + ensure_sqlite_column(conn, "local_index_state", "feed_synced_at", "TEXT") conn.execute( "CREATE INDEX IF NOT EXISTS local_index_edges_source " "ON local_index_edges(project_id, source_node_id)" @@ -6833,11 +6845,29 @@ def initialize_index_schema(conn): "CREATE INDEX IF NOT EXISTS local_index_edges_target_path " "ON local_index_edges(project_id, target_path)" ) - conn.execute( - """ - CREATE VIRTUAL TABLE IF NOT EXISTS local_index_fts USING fts5( + conn.commit() + if conn.execute("PRAGMA user_version").fetchone()[0] < INDEX_SCHEMA_VERSION: + migrate_index_fts_schema(conn) + else: + conn.execute(LOCAL_INDEX_FTS_DDL.format(table="local_index_fts")) + conn.commit() + + +# v1: default unicode61 FTS keyed by an UNINDEXED node_id (deletes scanned). +# v2: porter stemming, diacritic folding, 2/3-char prefix indexes, a title +# column and an integer rowid recorded in local_index_nodes.fts_rowid. +INDEX_SCHEMA_VERSION = 2 +LOCAL_INDEX_FTS_COLUMNS = ( + "node_id", "project_id", "title", "full_path", "name", "folder_path", "node_type", + "tags", "aliases", "links", "metadata_text", "content_text", +) +LOCAL_INDEX_FTS_METADATA_COLUMN = LOCAL_INDEX_FTS_COLUMNS.index("metadata_text") +LOCAL_INDEX_FTS_CONTENT_COLUMN = LOCAL_INDEX_FTS_COLUMNS.index("content_text") +LOCAL_INDEX_FTS_DDL = """ + CREATE VIRTUAL TABLE IF NOT EXISTS {table} USING fts5( node_id UNINDEXED, project_id UNINDEXED, + title, full_path, name, folder_path, @@ -6846,10 +6876,62 @@ def initialize_index_schema(conn): aliases, links, metadata_text, - content_text + content_text, + tokenize = 'porter unicode61 remove_diacritics 2', + prefix = '2 3' ) """ - ) +# bm25 weights in LOCAL_INDEX_FTS_COLUMNS order (UNINDEXED columns are 0). +LOCAL_INDEX_FTS_BM25 = "0.0, 0.0, 12.0, 8.0, 12.0, 4.0, 5.0, 8.0, 6.0, 1.0, 2.0, 1.0" + + +def migrate_index_fts_schema(conn): + """Rebuild the FTS table in the v2 layout inside one write transaction. + + Stored v1 text is re-tokenized locally, so completeness and the node + rows survive and no network is needed. A failure rolls back to the + intact v1 index; readers never observe a partial table. + """ + conn.execute("BEGIN IMMEDIATE") + try: + if conn.execute("PRAGMA user_version").fetchone()[0] >= INDEX_SCHEMA_VERSION: + conn.execute("ROLLBACK") + return + legacy = conn.execute( + "SELECT 1 FROM sqlite_master WHERE name = 'local_index_fts'" + ).fetchone() is not None + conn.execute("DROP TABLE IF EXISTS local_index_fts_v2_build") + conn.execute(LOCAL_INDEX_FTS_DDL.format(table="local_index_fts_v2_build")) + conn.execute("UPDATE local_index_nodes SET fts_rowid = NULL") + if legacy: + rows = conn.execute( + """ + SELECT f.node_id, f.project_id, COALESCE(n.title, n.name) AS title, + f.full_path, f.name, f.folder_path, f.node_type, f.tags, + f.aliases, f.links, f.metadata_text, f.content_text + FROM local_index_fts f + JOIN local_index_nodes n ON n.node_id = f.node_id + """ + ).fetchall() + for row in rows: + cursor = conn.execute( + f"INSERT INTO local_index_fts_v2_build ({', '.join(LOCAL_INDEX_FTS_COLUMNS)}) " + f"VALUES ({', '.join('?' for _ in LOCAL_INDEX_FTS_COLUMNS)})", + tuple(row), + ) + conn.execute( + "UPDATE local_index_nodes SET fts_rowid = ? WHERE node_id = ?", + (cursor.lastrowid, row["node_id"]), + ) + conn.execute("DROP TABLE local_index_fts") + conn.execute("ALTER TABLE local_index_fts_v2_build RENAME TO local_index_fts") + # Rows migrated from v1 predate feed tracking: no cursor is trusted. + conn.execute("UPDATE local_index_state SET feed_cursor = NULL, feed_synced_at = NULL") + conn.execute(f"PRAGMA user_version = {INDEX_SCHEMA_VERSION}") + conn.execute("COMMIT") + except BaseException: + conn.execute("ROLLBACK") + raise def index_storage_identity(): @@ -7035,13 +7117,15 @@ def index_node(node, *, connection=None): ) status = node_status(node) project_id = node.get("project_id") or DEFAULT_PROJECT_ID or "" + delete_node_fts_row(conn, node["id"]) conn.execute( """ INSERT INTO local_index_nodes ( node_id, project_id, full_path, name, folder_path, node_type, - tags_json, updated_at, indexed_at, title, status, archived, aliases_json + tags_json, updated_at, indexed_at, title, status, archived, aliases_json, + revision, content_withheld ) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(node_id) DO UPDATE SET project_id = excluded.project_id, full_path = excluded.full_path, @@ -7054,7 +7138,9 @@ def index_node(node, *, connection=None): title = excluded.title, status = excluded.status, archived = excluded.archived, - aliases_json = excluded.aliases_json + aliases_json = excluded.aliases_json, + revision = excluded.revision, + content_withheld = excluded.content_withheld """, ( node["id"], @@ -7070,6 +7156,8 @@ def index_node(node, *, connection=None): status, 1 if node_archived(node) else 0, compact_json_text(aliases), + node_revision_token(node), + 0 if standard_content_for_index else 1, ), ) conn.execute( @@ -7112,18 +7200,13 @@ def index_node(node, *, connection=None): iso_now(), ), ) - conn.execute("DELETE FROM local_index_fts WHERE node_id = ?", (node["id"],)) - conn.execute( - """ - INSERT INTO local_index_fts ( - node_id, project_id, full_path, name, folder_path, node_type, - tags, aliases, links, metadata_text, content_text - ) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - """, + fts_row = conn.execute( + f"INSERT INTO local_index_fts ({', '.join(LOCAL_INDEX_FTS_COLUMNS)}) " + f"VALUES ({', '.join('?' for _ in LOCAL_INDEX_FTS_COLUMNS)})", ( node["id"], node.get("project_id") or DEFAULT_PROJECT_ID or "", + title or "", node.get("full_path") or "", node.get("name") or "", node.get("folder_path") or "", @@ -7135,25 +7218,95 @@ def index_node(node, *, connection=None): content_text, ), ) + conn.execute( + "UPDATE local_index_nodes SET fts_rowid = ? WHERE node_id = ?", + (fts_row.lastrowid, node["id"]), + ) + + +def node_revision_token(node): + """The server revision of a node row (its RFC 3339 updated_at).""" + value = node.get("revision") or node.get("updated_at") + return str(value) if value not in (None, "") else None + + +def same_node_revision(stored, incoming): + if not stored or not incoming: + return False + if str(stored) == str(incoming): + return True + left, right = parse_iso_datetime(stored), parse_iso_datetime(incoming) + return left is not None and left == right + + +def delete_node_fts_row(conn, node_id): + """Delete one node's FTS row by its recorded integer rowid.""" + row = conn.execute( + "SELECT fts_rowid FROM local_index_nodes WHERE node_id = ?", (node_id,) + ).fetchone() + if row is not None and row["fts_rowid"] is not None: + conn.execute("DELETE FROM local_index_fts WHERE rowid = ?", (row["fts_rowid"],)) + elif row is not None: + # Unmapped row (written by an older worker during an update window). + conn.execute("DELETE FROM local_index_fts WHERE node_id = ?", (node_id,)) + + +def delete_project_index_rows(conn, project_id): + conn.execute( + "DELETE FROM local_index_fts WHERE rowid IN " + "(SELECT fts_rowid FROM local_index_nodes WHERE project_id = ? AND fts_rowid IS NOT NULL)", + (project_id,), + ) + conn.execute("DELETE FROM local_index_edges WHERE project_id = ?", (project_id,)) + conn.execute("DELETE FROM local_index_node_tags WHERE project_id = ?", (project_id,)) + conn.execute("DELETE FROM local_index_nodes WHERE project_id = ?", (project_id,)) + + +def remove_indexed_node_rows(conn, node_id): + delete_node_fts_row(conn, node_id) + conn.execute("DELETE FROM local_index_node_tags WHERE node_id = ?", (node_id,)) + conn.execute( + "DELETE FROM local_index_edges " + "WHERE source_node_id = ? OR target_node_id = ?", + (node_id, node_id), + ) + conn.execute("DELETE FROM local_index_nodes WHERE node_id = ?", (node_id,)) def remove_indexed_node(node_id): lock = index_file_lock() try: with index_connect() as conn: - conn.execute("DELETE FROM local_index_fts WHERE node_id = ?", (node_id,)) - conn.execute("DELETE FROM local_index_node_tags WHERE node_id = ?", (node_id,)) - conn.execute( - "DELETE FROM local_index_edges " - "WHERE source_node_id = ? OR target_node_id = ?", - (node_id, node_id), - ) - conn.execute("DELETE FROM local_index_nodes WHERE node_id = ?", (node_id,)) + remove_indexed_node_rows(conn, node_id) finally: release_index_file_lock(lock) +# Function words only. Domain words (plan, work, decide, done, ...) are +# searchable terms; intent inference reads the raw query separately. SEARCH_STOP_WORDS = { + "am", + "an", + "as", + "at", + "be", + "by", + "do", + "he", + "if", + "in", + "is", + "it", + "me", + "my", + "no", + "of", + "on", + "or", + "so", + "to", + "up", + "us", "about", "again", "also", @@ -7167,11 +7320,8 @@ SEARCH_STOP_WORDS = { "but", "can", "could", - "decide", - "decided", "did", "does", - "done", "for", "from", "had", @@ -7182,13 +7332,11 @@ SEARCH_STOP_WORDS = { "its", "just", "like", - "long", "need", "not", "our", "out", "over", - "plan", "please", "should", "that", @@ -7207,7 +7355,6 @@ SEARCH_STOP_WORDS = { "who", "why", "with", - "work", "you", } @@ -7277,7 +7424,8 @@ def search_concepts(value): concepts = [] for token in raw_tokens: cleaned = token.strip("._:/-") - if len(cleaned) < 3 or cleaned in SEARCH_STOP_WORDS: + # Two-character alphanumeric terms (ui, db, s3) are meaningful. + if len(cleaned) < 2 or cleaned in SEARCH_STOP_WORDS: continue concepts.append(cleaned) return unique_preserve_order(concepts) @@ -7307,6 +7455,45 @@ def fts_prefix_term(value): return f"{term}*" +SEARCH_ROOT_SUFFIXES = ( + "ations", "ation", "ments", "ment", "ings", "ing", "ers", "ied", "ies", "ed", "es", "er", "s", +) + + +def search_root_prefix(term): + """A conservative inflection root for prefix recall across Porter stems. + + Porter maps "deploying" to "deploi" but "deployment" to "deploy", so a + stemmed prefix alone misses sibling forms; "deplo*" matches both. + """ + root = term + for suffix in SEARCH_ROOT_SUFFIXES: + if root.endswith(suffix) and len(root) - len(suffix) >= 4: + root = root[: -len(suffix)] + break + if root and root[-1] in "eiy" and len(root) > 4: + root = root[:-1] + return root if len(root) >= 4 and root != term else "" + + +def fts_term_expression(value): + """FTS5 expression for one query term under the porter tokenizer.""" + parts = re.findall(r"\w+", str(value).lower(), flags=re.UNICODE) + if not parts: + return "" + if len(parts) > 1: + # memory-source-policy is tokenized as adjacent tokens. + return quote_fts(" ".join(parts)) + term = parts[0] + if len(term) <= 2: + return quote_fts(term) # ui/db/s3: exact, not every word starting "ui" + alternatives = [f"{term}*"] + root = search_root_prefix(term) + if root: + alternatives.append(f"{root}*") + return alternatives[0] if len(alternatives) == 1 else f"({' OR '.join(alternatives)})" + + def normalize_search_intent(value): intent = str(value or "any").strip().lower() if intent not in SEARCH_INTENTS: @@ -7365,7 +7552,7 @@ def compile_fts_candidate_query(analysis): components = [quote_fts(phrase) for phrase in analysis.get("phrases") or []] components.extend( term - for term in (fts_prefix_term(value) for value in analysis.get("selected_terms") or []) + for term in (fts_term_expression(value) for value in analysis.get("selected_terms") or []) if term ) return " OR ".join(unique_preserve_order(components)) or quote_fts(analysis["raw"]) @@ -7407,7 +7594,12 @@ def concept_matches_text(concept, text): if not concept or not text: return False variants = unique_preserve_order([concept, singular_search_token(concept)]) + parts = canonical_search_text(concept).split() + if len(parts) > 1: + return f" {' '.join(parts)} " in f" {text} " field_tokens = text.split() + if len(concept) <= 2: + return concept in field_tokens return any(token.startswith(variant) for token in field_tokens for variant in variants) @@ -7429,35 +7621,78 @@ def node_matches_search_intent(node, intent): ) -def node_search_rank_features(node, analysis): +SEARCH_DEMOTED_STATUSES = frozenset(ACTIVE_EXCLUDED_STATUSES | {"deprecated", "obsolete"}) +# Weighted relevance. Exact identity (the query is a node's name, title or +# basename) is decisive; everything else is additive so that one strong +# signal (an authoritative path) cannot hide a node that answers every term. +SEARCH_WEIGHTS = { + "exact_identity": 1000.0, + "identity_substring": 30.0, + "identity_phrase": 12.0, + "explicit_intent": 45.0, + "inferred_intent": 8.0, + "coverage": 40.0, + "identity_coverage": 10.0, + "bm25": 15.0, + "authority": 4.0, + "status_penalty": -30.0, + "recency": 3.0, +} +SEARCH_RECENCY_HALF_LIFE_DAYS = 90.0 + + +def search_requested_statuses(args): + """Statuses the caller asked for, which therefore are not demoted.""" + requested = set(normalize_string_list((args or {}).get("status"))) + if as_bool((args or {}).get("include_archived"), False) or as_bool((args or {}).get("archived_only"), False): + requested.add("archived") + return requested + + +def node_search_rank_features(node, analysis, *, requested_statuses=(), best_bm25=None, now=None): fields = search_field_values(node) raw = canonical_search_text(analysis.get("raw")) identity_values = [fields.get("name", ""), fields.get("title", "")] full_path = fields.get("full_path", "") basename = canonical_search_text(pathlib.PurePosixPath(str(node.get("full_path") or "")).name) + raw_aliases = node.get("aliases") + alias_values = [ + canonical_search_text(alias) + for alias in (raw_aliases if isinstance(raw_aliases, list) else []) + if canonical_search_text(alias) + ] exact_identity = 0 - if raw and raw in unique_preserve_order(identity_values + [basename]): + # A literal name, title, basename, full path or stored alias (hyphenated + # or not) is an exact identity hit. + if raw and raw in unique_preserve_order(identity_values + [basename, full_path] + alias_values): exact_identity = 3 - elif raw and any(raw in value for value in identity_values + [full_path, basename]): + elif raw and any(raw in value for value in identity_values + [full_path, basename] + alias_values): exact_identity = 2 elif any( canonical_search_text(phrase) - and any(canonical_search_text(phrase) in value for value in identity_values + [full_path]) + and any(canonical_search_text(phrase) in value for value in identity_values + [full_path] + alias_values) for phrase in analysis.get("phrases") or [] ): exact_identity = 1 concepts = analysis.get("concepts") or [] matched_fields = [] - matched_concepts = set() + identity_concepts = set() for field, text in fields.items(): field_matched = False for concept in concepts: if concept_matches_text(concept, text): - matched_concepts.add(concept) + identity_concepts.add(concept) field_matched = True if field_matched: matched_fields.append(field) + # Concepts the local FTS row matched in any indexed column, including the + # body (search_local_index sets matched_terms; server rows lack it). + fts_concepts = set(node.get("matched_terms") or ()) & set(concepts) + if fts_concepts - identity_concepts: + matched_fields.append("body_or_metadata") + matched_concepts = identity_concepts | fts_concepts + concept_count = max(1, len(concepts)) requested_intent = analysis.get("requested_intent") or "any" inferred_intent = analysis.get("inferred_intent") or "any" @@ -7470,14 +7705,33 @@ def node_search_rank_features(node, analysis): and node_matches_search_intent(node, inferred_intent) ) authority = node_search_authority(node) - coverage = len(matched_concepts) + status = str(node_status(node) or "").strip().lower() + requested = set(requested_statuses or ()) | (set(concepts) & SEARCH_DEMOTED_STATUSES) + demoted_status = None + if node_archived(node) and "archived" not in requested: + demoted_status = "archived" + elif status in SEARCH_DEMOTED_STATUSES and status not in requested: + demoted_status = status + try: + bm25_rank = float(node.get("rank")) + except (TypeError, ValueError): + bm25_rank = 0.0 + bm25_norm = bm25_rank / best_bm25 if best_bm25 and best_bm25 < 0 and bm25_rank < 0 else 0.0 + updated = search_updated_timestamp(node) + age_days = max(0.0, ((now or time.time()) - updated) / 86400.0) if updated else None + recency = 0.5 ** (age_days / SEARCH_RECENCY_HALF_LIFE_DAYS) if age_days is not None else 0.0 + + weights = SEARCH_WEIGHTS search_score = ( - explicit_intent_match * 1_000_000 - + exact_identity * 100_000 - + coverage * 10_000 - + inferred_intent_match * 5_000 - + len(matched_fields) * 100 - + authority + {3: weights["exact_identity"], 2: weights["identity_substring"], 1: weights["identity_phrase"]}.get(exact_identity, 0.0) + + explicit_intent_match * weights["explicit_intent"] + + inferred_intent_match * weights["inferred_intent"] + + weights["coverage"] * len(matched_concepts) / concept_count + + weights["identity_coverage"] * len(identity_concepts) / concept_count + + weights["bm25"] * bm25_norm + + weights["authority"] * authority / 100.0 + + (weights["status_penalty"] if demoted_status else 0.0) + + weights["recency"] * recency ) reasons = [] if explicit_intent_match: @@ -7488,14 +7742,19 @@ def node_search_rank_features(node, analysis): reasons.append("exact_identity") if matched_fields: reasons.append(f"matched_{','.join(matched_fields)}") + if concepts: + reasons.append(f"coverage_{len(matched_concepts)}/{len(concepts)}") + if demoted_status: + reasons.append(f"demoted_{demoted_status}") return { "explicit_intent_match": explicit_intent_match, "exact_identity": exact_identity, - "coverage": coverage, + "coverage": len(matched_concepts), "inferred_intent_match": inferred_intent_match, "matched_fields": matched_fields, "authority": authority, - "search_score": search_score, + "demoted_status": demoted_status, + "search_score": round(search_score, 3), "match_reason": "; ".join(reasons) or "body_or_metadata_match", } @@ -7510,10 +7769,21 @@ def search_updated_timestamp(node): def rank_search_items(items, analysis, limit=None): + requested_statuses = analysis.get("requested_statuses") or () + ranks = [] + for item in items: + try: + ranks.append(float(item.get("rank"))) + except (TypeError, ValueError): + pass + best_bm25 = min(ranks) if ranks else None + now = time.time() ranked = [] for item in items: node = dict(item) - features = node_search_rank_features(node, analysis) + features = node_search_rank_features( + node, analysis, requested_statuses=requested_statuses, best_bm25=best_bm25, now=now + ) node["matched_fields"] = features["matched_fields"] node["match_reason"] = features["match_reason"] node["search_score"] = features["search_score"] @@ -7522,12 +7792,7 @@ def rank_search_items(items, analysis, limit=None): except (TypeError, ValueError): bm25_rank = 0.0 sort_key = ( - -features["explicit_intent_match"], - -features["exact_identity"], - -features["coverage"], - -features["inferred_intent_match"], - -len(features["matched_fields"]), - -features["authority"], + -features["search_score"], bm25_rank, -search_updated_timestamp(node), str(node.get("full_path") or ""), @@ -7564,6 +7829,10 @@ def local_index_diagnostics(project_id, conn=None): ) graph_state = local_graph_index_state(project_id, connection) diagnostics.update(graph_state) + feed_state = local_index_feed_state(project_id, connection) + diagnostics["feed_tracking"] = bool(feed_state["feed_cursor"]) + diagnostics["feed_synced_at"] = feed_state["feed_synced_at"] + diagnostics["schema_version"] = INDEX_SCHEMA_VERSION diagnostics["sync_recommended"] = bool( diagnostics["sync_recommended"] or not graph_state["graph_complete"] ) @@ -7608,22 +7877,38 @@ def local_answer_staleness(project_id, *, server_error=None, diagnostics=None): return None code = problem["code"] cause = {key: problem[key] for key in ("reason", "message", "recorded_at") if problem.get(key)} + return build_local_staleness(project_id, code, cause, diagnostics=diagnostics) + + +LOCAL_STALENESS_MESSAGES = { + "auth_required": "Answered from the local index because Napseer login is not usable; results may be stale. Run `nap auth login` (or `nap auth repair`).", + "server_unavailable": "Answered from the local index because the Napseer server was unavailable; results may be stale. Retry when the service recovers.", + "feed_backlog": "Answered from the local index while its change-feed catch-up is still in progress; results may be stale. Run nap_index_sync to finish catching up.", + "index_resync_required": "Answered from the local index, but the server rejected its change-feed cursor; results may be stale. Run nap_index_sync (or `nap reindex`) to rebuild it.", + "feed_sync_failed": "Answered from the local index because its change-feed refresh failed; results may be stale. Run nap_index_sync to retry.", + "index_not_tracking": "Answered from the local index, which is not yet tracking the change feed; results may be stale. Run nap_index_sync to start tracking.", +} + + +def build_local_staleness(project_id, code, cause=None, *, diagnostics=None): + """(warning, freshness) for a local-index answer that could not be refreshed.""" if diagnostics is None: try: diagnostics = local_index_diagnostics(project_id) except Exception: diagnostics = {} indexed_at = diagnostics.get("latest_indexed_at") - last_sync_at = diagnostics.get("graph_completed_at") + last_sync_at = max( + (value for value in (diagnostics.get("graph_completed_at"), diagnostics.get("feed_synced_at")) if value), + key=lambda value: parse_iso_datetime(value) or datetime.min.replace(tzinfo=timezone.utc), + default=None, + ) newest = max( (value for value in (parse_iso_datetime(indexed_at), parse_iso_datetime(last_sync_at)) if value is not None), default=None, ) age_seconds = max(0, int((datetime.now(timezone.utc) - newest).total_seconds())) if newest else None - if code == "auth_required": - message = "Answered from the local index because Napseer login is not usable; results may be stale. Run `nap auth login` (or `nap auth repair`)." - else: - message = "Answered from the local index because the Napseer server was unavailable; results may be stale. Retry when the service recovers." + message = LOCAL_STALENESS_MESSAGES.get(code, LOCAL_STALENESS_MESSAGES["server_unavailable"]) warning = { "code": code, "message": message, @@ -7717,9 +8002,9 @@ def search_local_index(args): SELECT n.node_id, n.project_id, n.full_path, n.name, n.folder_path, n.node_type, n.tags_json, n.updated_at, n.indexed_at, n.title, n.status, n.archived,{LOCAL_INDEX_EDGE_COUNT_COLUMNS}, + n.content_withheld, n.aliases_json, local_index_fts.rowid AS fts_rowid, local_index_fts.aliases AS aliases_text, - bm25(local_index_fts, 0.0, 0.0, 8.0, 12.0, 4.0, 5.0, - 8.0, 6.0, 1.0, 2.0, 1.0) AS rank + bm25(local_index_fts, {LOCAL_INDEX_FTS_BM25}) AS rank FROM local_index_fts JOIN local_index_nodes n ON n.node_id = local_index_fts.node_id WHERE {' AND '.join(where)} @@ -7756,6 +8041,7 @@ def search_local_index(args): "type": row["node_type"], "tags": json.loads(row["tags_json"] or "[]"), "aliases_text": row["aliases_text"], + "aliases": json.loads(row["aliases_json"] or "[]"), "status": row["status"], "archived": bool(row["archived"]), "updated_at": row["updated_at"], @@ -7764,10 +8050,23 @@ def search_local_index(args): "backlinks_count": row["backlinks_count"], "rank": row["rank"], "matched_strategy": variant["strategy"], + "_fts_rowid": row["fts_rowid"], + "_content_withheld": bool(row["content_withheld"]), } for row in rows ] - items = rank_search_items(candidates, query_analysis, limit=limit) + # One candidate per node even if an older worker left a stale FTS row. + candidates = unique_nodes(candidates) + attach_local_term_coverage(conn, candidates, query_analysis) + items = rank_search_items( + candidates, + {**query_analysis, "requested_statuses": sorted(search_requested_statuses(args))}, + limit=limit, + ) + attach_local_snippets(conn, items, variant["query"]) + for item in items: + item.pop("_fts_rowid", None) + item.pop("_content_withheld", None) attempts = [ { "strategy": variant["strategy"], @@ -7805,6 +8104,71 @@ def search_local_index(args): } +LOCAL_SNIPPET_TOKENS = 24 + + +def attach_local_term_coverage(conn, candidates, analysis): + """Record which query concepts each candidate matches in any FTS column. + + One bounded FTS lookup per concept, restricted to the candidate rowids, + so body matches count toward coverage with the same stemming as recall. + """ + rowids = [item["_fts_rowid"] for item in candidates if item.get("_fts_rowid") is not None] + concepts = list(analysis.get("concepts") or []) + for item in candidates: + item["matched_terms"] = [] + if not rowids or not concepts: + return + by_rowid = {item["_fts_rowid"]: item for item in candidates if item.get("_fts_rowid") is not None} + placeholders = ",".join("?" for _ in rowids) + for concept in concepts: + expression = fts_term_expression(concept) + if not expression: + continue + try: + matched = conn.execute( + f"SELECT rowid FROM local_index_fts WHERE local_index_fts MATCH ? AND rowid IN ({placeholders})", + [expression, *rowids], + ).fetchall() + except sqlite3.Error: + continue + for row in matched: + by_rowid[row[0]]["matched_terms"].append(concept) + + +def attach_local_snippets(conn, items, match_query): + """Attach a query-centred body snippet; never for E2E (withheld) bodies.""" + rowids = [ + item["_fts_rowid"] + for item in items + if item.get("_fts_rowid") is not None and not item.get("_content_withheld") + ] + if not rowids: + return + try: + rows = conn.execute( + f""" + SELECT rowid, + snippet(local_index_fts, {LOCAL_INDEX_FTS_CONTENT_COLUMN}, '', '', '…', {LOCAL_SNIPPET_TOKENS}) AS body, + snippet(local_index_fts, {LOCAL_INDEX_FTS_METADATA_COLUMN}, '', '', '…', {LOCAL_SNIPPET_TOKENS}) AS meta + FROM local_index_fts + WHERE local_index_fts MATCH ? AND rowid IN ({','.join('?' for _ in rowids)}) + """, + [match_query, *rowids], + ).fetchall() + except sqlite3.Error: + return + snippets = {} + for row in rows: + text = (row["body"] or "").strip() or (row["meta"] or "").strip() + if text: + snippets[row["rowid"]] = " ".join(text.split()) + for item in items: + snippet = snippets.get(item.get("_fts_rowid")) + if snippet and not item.get("_content_withheld"): + item["snippet"] = snippet + + def validate_reindex_page(page, project_id, seen_cursors): """Reject incomplete/malformed inventory before replacing a usable index.""" def invalid(): @@ -7840,6 +8204,10 @@ def validate_reindex_page(page, project_id, seen_cursors): def reindex_project(args): project_id = resolve_project_id(args) limit = max(1, min(int(args.get("limit", 200)), 200)) + # Take the change-feed head before listing: every change after it is + # replayed by the next feed sync, and replaying an already-listed change + # is idempotent. An old server without the feed still rebuilds. + feed_cursor, feed_status = fetch_change_feed_head(project_id) nodes = [] cursor = None seen_cursors = set() @@ -7863,15 +8231,13 @@ def reindex_project(args): raise_if_mcp_request_cancelled() conn.execute("BEGIN IMMEDIATE") set_local_graph_index_complete(project_id, False, conn) - conn.execute("DELETE FROM local_index_fts WHERE project_id = ?", (project_id,)) - conn.execute("DELETE FROM local_index_edges WHERE project_id = ?", (project_id,)) - conn.execute("DELETE FROM local_index_node_tags WHERE project_id = ?", (project_id,)) - conn.execute("DELETE FROM local_index_nodes WHERE project_id = ?", (project_id,)) + delete_project_index_rows(conn, project_id) for node in nodes: raise_if_mcp_request_cancelled() index_node(node, connection=conn) raise_if_mcp_request_cancelled() set_local_graph_index_complete(project_id, True, conn) + set_local_index_feed_cursor(conn, project_id, feed_cursor) finally: conn.close() finally: @@ -7881,6 +8247,7 @@ def reindex_project(args): "indexed": len(nodes), "index_path": str(INDEX_PATH), "graph_complete": True, + "feed": feed_status, } @@ -7888,8 +8255,6 @@ def sync_local_index(args): args = normalize_discovery_ranges(args or {}) project_id = resolve_project_id(args) limit = max(1, min(int(args.get("limit", 200)), 200)) - include_archived = as_bool(args.get("include_archived"), True) - lookback_seconds = normalize_int(args.get("lookback_seconds"), 5, 0, 3600) status_before = local_index_diagnostics(project_id) if not status_before.get("graph_complete"): result = reindex_project({**args, "project_id": project_id, "limit": limit}) @@ -7900,12 +8265,100 @@ def sync_local_index(args): "reason": "local_graph_index_incomplete", "index": local_index_diagnostics(project_id), } + if args.get("updated_after"): + return sync_local_index_by_timestamp(args, project_id, status_before, feed_status="not_used") + feed_state = local_index_feed_state(project_id) + if feed_state["feed_cursor"]: + try: + result = sync_local_index_from_feed(project_id) + except SafeToolError as exc: + if exc.code == "feed_cursor_invalid": + rebuilt = reindex_project({**args, "project_id": project_id, "limit": limit}) + return { + **rebuilt, + "status": "synced", + "mode": "full_reindex", + "reason": "feed_cursor_invalid", + "index": local_index_diagnostics(project_id), + } + if exc.code != "feed_unsupported": + raise + return sync_local_index_by_timestamp(args, project_id, status_before, feed_status="unsupported") + return {**result, "index": local_index_diagnostics(project_id)} + return bootstrap_local_index_feed(args, project_id, status_before) + + +def bootstrap_local_index_feed(args, project_id, status_before, *, deadline=None, max_pages=None, + require_head=False): + """Start change-feed tracking for a complete index that has no cursor. + + Takes the head cursor first, catches up by timestamp, then commits the + cursor only if the index is still complete and still untracked. Changes + between the head and the catch-up are replayed idempotently later. With + a deadline/max_pages, an unfinished catch-up applies nothing and leaves + the index untracked (status "partial"). + """ + def remaining_timeout(): + if deadline is None: + return None + return min(HTTP_TIMEOUT_SECONDS, max(deadline - time.monotonic(), 0.5)) + + head_cursor, feed_status = fetch_change_feed_head(project_id, timeout_seconds=remaining_timeout()) + if require_head and not head_cursor: + return {"project_id": project_id, "status": "not_tracking", "mode": "bootstrap", "feed": feed_status} + result = sync_local_index_by_timestamp( + args, project_id, status_before, feed_status=feed_status, deadline=deadline, max_pages=max_pages + ) + if head_cursor and result["status"] == "synced": + lock = index_file_lock() + try: + conn = index_connect() + try: + with conn: + conn.execute("BEGIN IMMEDIATE") + current = local_index_feed_state(project_id, conn) + if current["graph_complete"] and not current["feed_cursor"]: + set_local_index_feed_cursor(conn, project_id, head_cursor) + finally: + conn.close() + finally: + release_index_file_lock(lock) + result["feed"] = "initialized" + result["index"] = local_index_diagnostics(project_id) + return result + + +def sync_local_index_by_timestamp(args, project_id, status_before, *, feed_status, deadline=None, max_pages=None): + """Legacy catch-up by updated_at. It cannot observe hard deletes. + + Bounded by an optional deadline/max_pages. Pages arrive newest first, so + an unfinished catch-up applies nothing: applying a prefix would raise the + updated_at high-water mark past nodes that were never fetched. + """ + limit = max(1, min(int(args.get("limit", 200)), 200)) + include_archived = as_bool(args.get("include_archived"), True) + lookback_seconds = normalize_int(args.get("lookback_seconds"), 5, 0, 3600) latest_node_updated_at = status_before.get("latest_node_updated_at") updated_after = args.get("updated_after") or iso_minus_seconds(latest_node_updated_at, lookback_seconds) nodes = [] cursor = None pages = 0 while True: + if (max_pages is not None and pages >= max_pages) or ( + deadline is not None and time.monotonic() >= deadline + ): + return { + "project_id": project_id, + "status": "partial", + "mode": "incremental", + "feed": feed_status, + "updated_after": updated_after, + "pages": pages, + "seen": len(nodes), + "indexed": 0, + "removed": 0, + "index_path": str(INDEX_PATH), + } query = { "limit": limit, "view": "full", @@ -7915,7 +8368,10 @@ def sync_local_index(args): query["updated_after"] = updated_after if cursor: query["cursor"] = cursor - page = request_json("GET", f"/v1/projects/{project_id}/nodes?{rest_query_string(query)}") + timeout = None if deadline is None else min(HTTP_TIMEOUT_SECONDS, max(deadline - time.monotonic(), 0.5)) + page = request_json( + "GET", f"/v1/projects/{project_id}/nodes?{rest_query_string(query)}", timeout_seconds=timeout + ) pages += 1 nodes.extend(filter_project_nodes(page.get("items", []))) cursor = page.get("next_cursor") @@ -7924,7 +8380,9 @@ def sync_local_index(args): indexed = 0 removed = 0 for node in nodes: - if node.get("archived_at"): + # A full rebuild lists only active nodes, so archived nodes leave the + # index here too; the change-feed sync applies the same rule. + if node_archived(node): remove_indexed_node(node.get("id")) removed += 1 continue @@ -7934,6 +8392,7 @@ def sync_local_index(args): "project_id": project_id, "status": "synced", "mode": "incremental", + "feed": feed_status, "updated_after": updated_after, "lookback_seconds": lookback_seconds, "include_archived": include_archived, @@ -7951,16 +8410,403 @@ def index_status(args): return local_index_diagnostics(project_id) +CHANGE_FEED_NODE_REFRESH_TYPES = {"node_created", "node_updated", "node_restored", "node_changed"} +CHANGE_FEED_NODE_REMOVE_TYPES = {"node_archived", "node_deleted"} +FEED_SYNC_MAX_PAGES = 50 +FEED_SYNC_BUDGET_SECONDS = 60.0 +FEED_AUTO_SYNC_INTERVAL_SECONDS = 10.0 +FEED_AUTO_SYNC_UNSUPPORTED_INTERVAL_SECONDS = float("inf") # old server: not retried in this process +FEED_AUTO_SYNC_BUDGET_SECONDS = 2.0 +FEED_AUTO_SYNC_MAX_PAGES = 3 +FEED_AUTO_SYNC_PAGE_LIMIT = 50 +FEED_AUTO_BOOTSTRAP_RETRY_SECONDS = 600.0 +FEED_AUTO_SYNC_STATE = {} +FEED_AUTO_SYNC_LOCK = threading.Lock() + + +def local_index_feed_state(project_id, conn=None): + def read(connection): + row = connection.execute( + "SELECT graph_complete, feed_cursor, feed_synced_at FROM local_index_state WHERE project_id = ?", + (project_id,), + ).fetchone() + return { + "graph_complete": bool(row and row["graph_complete"]), + "feed_cursor": (row["feed_cursor"] or None) if row else None, + "feed_synced_at": row["feed_synced_at"] if row else None, + } + + if conn is not None: + return read(conn) + with index_connect() as connection: + return read(connection) + + +def set_local_index_feed_cursor(conn, project_id, cursor): + conn.execute( + """ + INSERT INTO local_index_state (project_id, feed_cursor, feed_synced_at) + VALUES (?, ?, ?) + ON CONFLICT(project_id) DO UPDATE SET + feed_cursor = excluded.feed_cursor, + feed_synced_at = excluded.feed_synced_at + """, + (project_id, cursor or None, iso_now() if cursor else None), + ) + + +def change_feed_error(exc): + """Map a /changes failure to feed_unsupported, feed_cursor_invalid, or itself.""" + if isinstance(exc, SafeToolError) and exc.status == 404: + error = SafeToolError( + "feed_unsupported", + "This Napseer server does not provide the change feed; the local index used timestamp sync.", + status=404, + service_code=exc.service_code, + ) + return error + if isinstance(exc, SafeToolError) and exc.status == 400: + return SafeToolError( + "feed_cursor_invalid", + "The server rejected the local index feed cursor; a full reindex is required.", + status=400, + service_code=exc.service_code, + ) + return exc + + +def request_change_feed(project_id, after, limit, *, timeout_seconds=None): + query = rest_query_string({"after": after, "limit": limit}) + try: + page = request_json( + "GET", f"/v1/projects/{project_id}/changes?{query}", timeout_seconds=timeout_seconds + ) + except SafeToolError as exc: + mapped = change_feed_error(exc) + if mapped is exc: + raise + raise mapped from exc + if (not isinstance(page, dict) or not isinstance(page.get("items"), list) + or not isinstance(page.get("next_cursor"), str) or not page["next_cursor"].strip()): + raise SafeToolError( + "invalid_index_response", + "The change feed returned an unexpected page; the local index cursor was not changed.", + ) + return page + + +def fetch_change_feed_head(project_id, *, timeout_seconds=None): + """Return (head_cursor, status) without blocking a rebuild on the feed. + + An old server (404) returns (None, 'unsupported'); another feed failure + returns (None, 'unavailable') so the index is rebuilt but untracked + until a later nap_index_sync takes a head cursor. + """ + try: + page = request_change_feed(project_id, "head", 1, timeout_seconds=timeout_seconds) + except SafeToolError as exc: + if exc.code == "feed_unsupported": + return None, "unsupported" + if exc.code == "request_cancelled" or server_fallback_code(exc) == "auth_required": + raise + return None, "unavailable" + return page["next_cursor"], "tracking" + + +def fetch_node_for_index(project_id, node_id, *, timeout_seconds=None): + """Fetch the raw stored node (never decrypted) or None when it no longer exists.""" + try: + node = request_json( + "GET", + f"/v1/projects/{project_id}/nodes/{urllib.parse.quote(node_id, safe='')}", + timeout_seconds=timeout_seconds, + ) + except SafeToolError as exc: + if exc.status == 404: + return None + raise + if (not isinstance(node, dict) or node.get("id") != node_id + or node.get("project_id") != project_id + or not isinstance(node.get("full_path"), str) + or not node["full_path"].startswith("/")): + raise SafeToolError( + "invalid_index_response", + "Napseer returned an invalid node for the local index; the feed cursor was not changed.", + ) + return node + + +def sync_local_index_from_feed(project_id, *, max_pages=FEED_SYNC_MAX_PAGES, + budget_seconds=FEED_SYNC_BUDGET_SECONDS, page_limit=200): + """Apply change-feed pages to a complete local index. + + Each page's rows and its cursor commit in one transaction, so the cursor + never advances past a change that was not applied. Network reads happen + outside the writer lock. Bounded by pages and a time budget. + """ + state = local_index_feed_state(project_id) + if not state["graph_complete"]: + raise SafeToolError("local_graph_incomplete", "The local index is incomplete; run a full reindex.") + cursor = state["feed_cursor"] + if not cursor: + raise SafeToolError("feed_not_initialized", "The local index has no change-feed cursor yet.") + deadline = time.monotonic() + max(0.1, float(budget_seconds)) + totals = {"events": 0, "indexed": 0, "removed": 0, "skipped_current": 0, "refetched": 0} + pages = 0 + has_more = False + status = "synced" + while True: + remaining = deadline - time.monotonic() + if pages >= max_pages or remaining <= 0: + status = "partial" if has_more else status + break + raise_if_mcp_request_cancelled() + page = request_change_feed( + project_id, cursor, page_limit, timeout_seconds=min(HTTP_TIMEOUT_SECONDS, max(remaining, 0.5)) + ) + latest = {} + for item in page["items"]: + if not isinstance(item, dict): + raise SafeToolError("invalid_index_response", "The change feed returned a malformed item.") + kind = str(item.get("type") or "") + if not kind.startswith("node_"): + continue # operation_* lifecycle events do not change nodes + node_id = item.get("node_id") + if not isinstance(node_id, str) or not node_id.strip(): + raise SafeToolError("invalid_index_response", "The change feed returned an item without node_id.") + latest.pop(node_id, None) + latest[node_id] = item + totals["events"] += len(latest) + stored = {} + if latest: + with index_connect() as conn: + ids = list(latest) + for row in conn.execute( + f"SELECT node_id, revision FROM local_index_nodes WHERE project_id = ? " + f"AND node_id IN ({','.join('?' for _ in ids)})", + [project_id, *ids], + ).fetchall(): + stored[row["node_id"]] = row["revision"] + operations = [] + budget_exhausted = False + for node_id, item in latest.items(): + kind = item["type"] + if kind in CHANGE_FEED_NODE_REMOVE_TYPES: + operations.append(("remove", node_id)) + continue + if kind not in CHANGE_FEED_NODE_REFRESH_TYPES: + continue + if node_id in stored and same_node_revision(stored[node_id], item.get("new_revision")): + totals["skipped_current"] += 1 + continue + raise_if_mcp_request_cancelled() + remaining = deadline - time.monotonic() + if remaining <= 0: + # A page commits only as a whole; leave it for the next call. + budget_exhausted = True + break + node = fetch_node_for_index( + project_id, node_id, timeout_seconds=min(HTTP_TIMEOUT_SECONDS, max(remaining, 0.5)) + ) + totals["refetched"] += 1 + if node is None or node_archived(node) or is_agent_namespace_path(node["full_path"]): + operations.append(("remove", node_id)) + else: + operations.append(("index", node)) + if budget_exhausted: + status, has_more = "partial", True + break + next_cursor = page["next_cursor"] + lock = index_file_lock() + try: + conn = index_connect() + try: + with conn: + conn.execute("BEGIN IMMEDIATE") + current = local_index_feed_state(project_id, conn) + if not current["graph_complete"] or current["feed_cursor"] != cursor: + # A concurrent reindex or sync moved the snapshot; it owns freshness now. + status = "superseded" + break + for action, value in operations: + if action == "remove": + remove_indexed_node_rows(conn, value) + totals["removed"] += 1 + else: + index_node(value, connection=conn) + totals["indexed"] += 1 + set_local_index_feed_cursor(conn, project_id, next_cursor) + finally: + conn.close() + finally: + release_index_file_lock(lock) + pages += 1 + cursor = next_cursor + has_more = page.get("has_more") is True + if not has_more: + break + return { + "project_id": project_id, + "status": status, + "mode": "feed", + "pages": pages, + "has_more": has_more, + "feed_cursor_advanced": pages > 0, + **totals, + "index_path": str(INDEX_PATH), + } + + +def auto_refresh_local_index(project_id): + """Throttled, time-bounded change-feed refresh before a local read. + + Returns (freshness_facts, staleness). Never raises: a failed refresh + leaves the previous snapshot readable and reports it as stale. + """ + if not INDEX_PATH.exists(): + return {}, None + try: + state = local_index_feed_state(project_id) + except Exception: + return {}, None + if not state["graph_complete"]: + return {}, None + if current_auth_problem(): + # local_answer_staleness reports the auth problem for this read. + return {"feed_sync": "skipped_auth", "feed_synced_at": state["feed_synced_at"]}, None + key = (str(INDEX_PATH), project_id) + now = time.monotonic() + with FEED_AUTO_SYNC_LOCK: + entry = FEED_AUTO_SYNC_STATE.get(key) + if entry and now - entry["at"] < entry["interval"]: + facts = dict(entry["facts"]) + if facts.get("feed_sync") in {"synced", "tracking_initialized"}: + facts["feed_sync"] = "throttled" + facts["feed_synced_at"] = state["feed_synced_at"] + return facts, entry["staleness"] + FEED_AUTO_SYNC_STATE[key] = { + "at": now, + "interval": FEED_AUTO_SYNC_INTERVAL_SECONDS, + "facts": {"feed_sync": "in_progress"}, + "staleness": None, + } + if not state["feed_cursor"]: + return auto_bootstrap_local_index_feed(project_id, key) + interval = FEED_AUTO_SYNC_INTERVAL_SECONDS + staleness = None + try: + result = sync_local_index_from_feed( + project_id, + max_pages=FEED_AUTO_SYNC_MAX_PAGES, + budget_seconds=FEED_AUTO_SYNC_BUDGET_SECONDS, + page_limit=FEED_AUTO_SYNC_PAGE_LIMIT, + ) + facts = {"feed_sync": "synced" if result["status"] != "partial" else "partial"} + if result["status"] == "partial": + facts["feed_backlog"] = True + staleness = build_local_staleness(project_id, "feed_backlog", {"pages": result["pages"]}) + except SafeToolError as exc: + if exc.code == "request_cancelled": + with FEED_AUTO_SYNC_LOCK: + FEED_AUTO_SYNC_STATE.pop(key, None) + raise + if exc.code == "feed_unsupported": + facts = {"feed_sync": "unsupported"} + interval = FEED_AUTO_SYNC_UNSUPPORTED_INTERVAL_SECONDS + elif exc.code == "feed_cursor_invalid": + facts = {"feed_sync": "resync_required"} + staleness = build_local_staleness(project_id, "index_resync_required", safe_exception_payload(exc)) + else: + facts = {"feed_sync": "failed"} + code = server_fallback_code(exc) or "feed_sync_failed" + staleness = build_local_staleness(project_id, code, safe_exception_payload(exc)) + except Exception as exc: + facts = {"feed_sync": "failed"} + code = server_fallback_code(exc) or "feed_sync_failed" + staleness = build_local_staleness(project_id, code, safe_exception_payload(exc)) + try: + facts["feed_synced_at"] = local_index_feed_state(project_id)["feed_synced_at"] + except Exception: + facts["feed_synced_at"] = state["feed_synced_at"] + with FEED_AUTO_SYNC_LOCK: + FEED_AUTO_SYNC_STATE[key] = { + "at": time.monotonic(), "interval": interval, "facts": dict(facts), "staleness": staleness, + } + return facts, staleness + + +def index_not_tracking_staleness(project_id, cause): + staleness = build_local_staleness(project_id, "index_not_tracking", cause) + staleness[0]["next_action"] = "nap_index_sync" + staleness[1]["next_action"] = "nap_index_sync" + return staleness + + +def auto_bootstrap_local_index_feed(project_id, key): + """Automatic, bounded start of feed tracking for a cursorless complete index. + + Called with the throttle slot already reserved. Success starts normal + feed refreshes; failure or budget exhaustion keeps the index untracked, + answers locally with an index_not_tracking warning, and does not retry + for FEED_AUTO_BOOTSTRAP_RETRY_SECONDS. An old server is marked + unsupported and not retried. + """ + interval = FEED_AUTO_BOOTSTRAP_RETRY_SECONDS + staleness = None + try: + result = bootstrap_local_index_feed( + {"limit": FEED_AUTO_SYNC_PAGE_LIMIT}, + project_id, + local_index_diagnostics(project_id), + deadline=time.monotonic() + FEED_AUTO_SYNC_BUDGET_SECONDS, + max_pages=FEED_AUTO_SYNC_MAX_PAGES, + require_head=True, + ) + if result.get("feed") == "initialized": + facts = {"feed_sync": "tracking_initialized"} + interval = FEED_AUTO_SYNC_INTERVAL_SECONDS + elif result.get("feed") == "unsupported": + facts = {"feed_sync": "unsupported"} + interval = FEED_AUTO_SYNC_UNSUPPORTED_INTERVAL_SECONDS + else: + facts = {"feed_sync": "not_tracking"} + staleness = index_not_tracking_staleness( + project_id, {"feed": result.get("feed"), "status": result.get("status")} + ) + except SafeToolError as exc: + if exc.code == "request_cancelled": + with FEED_AUTO_SYNC_LOCK: + FEED_AUTO_SYNC_STATE.pop(key, None) + raise + facts = {"feed_sync": "not_tracking"} + staleness = index_not_tracking_staleness(project_id, safe_exception_payload(exc)) + except Exception as exc: + facts = {"feed_sync": "not_tracking"} + staleness = index_not_tracking_staleness(project_id, safe_exception_payload(exc)) + try: + facts["feed_synced_at"] = local_index_feed_state(project_id)["feed_synced_at"] + except Exception: + facts["feed_synced_at"] = None + with FEED_AUTO_SYNC_LOCK: + FEED_AUTO_SYNC_STATE[key] = { + "at": time.monotonic(), "interval": interval, "facts": dict(facts), "staleness": staleness, + } + return facts, staleness + + +def local_read_freshness(facts): + """Freshness block for a local-index answer; apply_local_staleness adds the verdict.""" + return {"source": "local_index", "stale": False, **(facts or {})} + + def clear_local_index(args): project_id = resolve_project_id(args) lock = index_file_lock() try: with index_connect() as conn: - conn.execute("DELETE FROM local_index_fts WHERE project_id = ?", (project_id,)) - conn.execute("DELETE FROM local_index_edges WHERE project_id = ?", (project_id,)) - conn.execute("DELETE FROM local_index_node_tags WHERE project_id = ?", (project_id,)) - conn.execute("DELETE FROM local_index_nodes WHERE project_id = ?", (project_id,)) + delete_project_index_rows(conn, project_id) set_local_graph_index_complete(project_id, False, conn) + set_local_index_feed_cursor(conn, project_id, None) finally: release_index_file_lock(lock) return {"project_id": project_id, "cleared": True} @@ -11066,12 +11912,15 @@ def browse_project_nodes(args, *, default_view="summary"): return list_project_nodes(args) project_id = resolve_project_id(args) + feed_facts, feed_staleness = auto_refresh_local_index(project_id) diagnostics = local_index_diagnostics(project_id) local_supported = bool(diagnostics.get("graph_complete")) if local_supported: + response = browse_local_index(args, view=view, sort=sort) + response["freshness"] = {**local_read_freshness(feed_facts), **(response.get("freshness") or {})} return apply_local_staleness( - browse_local_index(args, view=view, sort=sort), - local_answer_staleness(project_id, diagnostics=diagnostics), + response, + feed_staleness or local_answer_staleness(project_id, diagnostics=diagnostics), ) if local_cursor: raise RuntimeError("local discovery cursor requires the same complete local snapshot routing") @@ -13182,7 +14031,7 @@ def strip_node_content(node, include_content=False): return node slim = dict(node) content = slim.pop("content_text", "") - if content: + if content and not slim.get("snippet"): slim["snippet"] = content[:240] return slim @@ -13220,6 +14069,9 @@ def search_memory(args): sources = [] server_search = {} staleness = None + feed_facts, feed_staleness = ( + auto_refresh_local_index(project_id) if mode in {"hybrid", "local"} else ({}, None) + ) diagnostics = { "query": query_analysis, "local_index": local_index_diagnostics(project_id), @@ -13293,7 +14145,11 @@ def search_memory(args): items = [node for node in unique_nodes(collected) if node_in_updated_range(node, args)] items = apply_status_filters(items, filters) - items = rank_search_items(items, query_analysis, limit=limit) + items = rank_search_items( + items, + {**query_analysis, "requested_statuses": sorted(search_requested_statuses(args))}, + limit=limit, + ) if not items and query_analysis["is_long_query"]: diagnostics["hint"] = "Long natural-language query produced no matches; retry with quoted exact phrases or 2-5 distinctive nouns if the target memory exists." response = discovery_envelope( @@ -13322,8 +14178,10 @@ def search_memory(args): response["search_projection"] = "sqlite_fts5_current" elif server_needed: response.update(server_search) + if local_used: + response["freshness"] = local_read_freshness(feed_facts) if staleness is None and local_used and not server_needed: - staleness = local_answer_staleness(project_id, diagnostics=diagnostics.get("local_index")) + staleness = feed_staleness or local_answer_staleness(project_id, diagnostics=diagnostics.get("local_index")) return apply_local_staleness(response, staleness) @@ -13863,6 +14721,10 @@ def get_memory_context(args): mode = str(args.get("mode") or "hybrid").strip().lower() if mode not in {"hybrid", "local", "server"}: raise RuntimeError("mode must be hybrid, local, or server") + feed_facts, feed_staleness = {}, None + if mode in {"local", "hybrid"}: + # Refresh before seed routes resolve against the local snapshot. + feed_facts, feed_staleness = auto_refresh_local_index(resolve_project_id(args)) seed_ids = context_seed_ids(args, mode=mode) if mode in {"local", "hybrid"}: try: @@ -13876,9 +14738,11 @@ def get_memory_context(args): "until the managed local projection refresh completes.", ) result = local_memory_context(args, seed_ids) + finalized = finalize_memory_context(args, result, seed_ids) + finalized["freshness"] = {**local_read_freshness(feed_facts), **(finalized.get("freshness") or {})} return apply_local_staleness( - finalize_memory_context(args, result, seed_ids), - local_answer_staleness(resolve_project_id(args)), + finalized, + feed_staleness or local_answer_staleness(resolve_project_id(args)), ) except Exception as exc: if mode == "local": @@ -15751,7 +16615,7 @@ def raw_tools(): }, { "name": "nap_reindex_project", - "description": "Rebuild the local SQLite FTS index from readable project nodes.", + "description": "Rebuild the local SQLite FTS index from readable project nodes. Records the change-feed head first so later reads refresh from the feed.", "inputSchema": { "type": "object", "properties": {"limit": {"type": "integer", "minimum": 1, "maximum": 200}}, @@ -15760,7 +16624,7 @@ def raw_tools(): }, { "name": "nap_index_sync", - "description": "Incrementally synchronize the local SQLite FTS index from remote nodes changed since the latest indexed node update. Archived nodes are removed from the local index.", + "description": "Incrementally synchronize the local SQLite FTS index. Uses the project change feed from the index cursor (creates, updates, restores, archives and hard deletes; a rejected cursor triggers a full reindex); servers without the feed, or an explicit updated_after, use the updated_at high-water mark. Archived nodes are removed from the local index, as in a full rebuild.", "inputSchema": { "type": "object", "properties": { diff --git a/tests/test_mcp_index_feed_sync.py b/tests/test_mcp_index_feed_sync.py new file mode 100644 index 0000000..1c719c5 --- /dev/null +++ b/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 diff --git a/tests/test_mcp_index_rebuild_recovery.py b/tests/test_mcp_index_rebuild_recovery.py index 5fb91d8..d7d9d15 100644 --- a/tests/test_mcp_index_rebuild_recovery.py +++ b/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]) diff --git a/tests/test_mcp_memory_ergonomics_helpers.py b/tests/test_mcp_memory_ergonomics_helpers.py index 713145e..8895a32 100644 --- a/tests/test_mcp_memory_ergonomics_helpers.py +++ b/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 diff --git a/tests/test_mcp_reliability_contracts.py b/tests/test_mcp_reliability_contracts.py index ae9df49..9cde96a 100644 --- a/tests/test_mcp_reliability_contracts.py +++ b/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() diff --git a/tests/test_mcp_search_latency.py b/tests/test_mcp_search_latency.py index 8fecace..6ac5280 100644 --- a/tests/test_mcp_search_latency.py +++ b/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(): diff --git a/tests/test_mcp_search_quality.py b/tests/test_mcp_search_quality.py new file mode 100644 index 0000000..5f7f93b --- /dev/null +++ b/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"]