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