fix: gate hybrid graph reads on complete local indexes
This commit is contained in:
@@ -6227,6 +6227,15 @@ def index_connect():
|
|||||||
)
|
)
|
||||||
"""
|
"""
|
||||||
)
|
)
|
||||||
|
conn.execute(
|
||||||
|
"""
|
||||||
|
CREATE TABLE IF NOT EXISTS local_index_state (
|
||||||
|
project_id TEXT PRIMARY KEY,
|
||||||
|
graph_complete INTEGER NOT NULL DEFAULT 0,
|
||||||
|
graph_completed_at TEXT
|
||||||
|
)
|
||||||
|
"""
|
||||||
|
)
|
||||||
conn.execute(
|
conn.execute(
|
||||||
"CREATE INDEX IF NOT EXISTS local_index_edges_source "
|
"CREATE INDEX IF NOT EXISTS local_index_edges_source "
|
||||||
"ON local_index_edges(project_id, source_node_id)"
|
"ON local_index_edges(project_id, source_node_id)"
|
||||||
@@ -6260,6 +6269,52 @@ def index_connect():
|
|||||||
return conn
|
return conn
|
||||||
|
|
||||||
|
|
||||||
|
def local_graph_index_state(project_id, conn=None):
|
||||||
|
def read(connection):
|
||||||
|
row = connection.execute(
|
||||||
|
"SELECT graph_complete, graph_completed_at "
|
||||||
|
"FROM local_index_state WHERE project_id = ?",
|
||||||
|
(project_id,),
|
||||||
|
).fetchone()
|
||||||
|
return {
|
||||||
|
"graph_complete": bool(row and row["graph_complete"]),
|
||||||
|
"graph_completed_at": row["graph_completed_at"] if row else None,
|
||||||
|
}
|
||||||
|
|
||||||
|
if conn is not None:
|
||||||
|
return read(conn)
|
||||||
|
with index_connect() as connection:
|
||||||
|
return read(connection)
|
||||||
|
|
||||||
|
|
||||||
|
def set_local_graph_index_complete(project_id, complete, conn=None):
|
||||||
|
completed_at = iso_now() if complete else None
|
||||||
|
|
||||||
|
def write(connection):
|
||||||
|
connection.execute(
|
||||||
|
"""
|
||||||
|
INSERT INTO local_index_state (
|
||||||
|
project_id, graph_complete, graph_completed_at
|
||||||
|
)
|
||||||
|
VALUES (?, ?, ?)
|
||||||
|
ON CONFLICT(project_id) DO UPDATE SET
|
||||||
|
graph_complete = excluded.graph_complete,
|
||||||
|
graph_completed_at = excluded.graph_completed_at
|
||||||
|
""",
|
||||||
|
(project_id, 1 if complete else 0, completed_at),
|
||||||
|
)
|
||||||
|
|
||||||
|
if conn is not None:
|
||||||
|
write(conn)
|
||||||
|
return
|
||||||
|
lock = index_file_lock()
|
||||||
|
try:
|
||||||
|
with index_connect() as connection:
|
||||||
|
write(connection)
|
||||||
|
finally:
|
||||||
|
release_index_file_lock(lock)
|
||||||
|
|
||||||
|
|
||||||
def flatten_index_values(value):
|
def flatten_index_values(value):
|
||||||
if value is None:
|
if value is None:
|
||||||
return []
|
return []
|
||||||
@@ -6586,6 +6641,11 @@ def local_index_diagnostics(project_id, conn=None):
|
|||||||
"sync_recommended": count == 0,
|
"sync_recommended": count == 0,
|
||||||
}
|
}
|
||||||
)
|
)
|
||||||
|
graph_state = local_graph_index_state(project_id, connection)
|
||||||
|
diagnostics.update(graph_state)
|
||||||
|
diagnostics["sync_recommended"] = bool(
|
||||||
|
diagnostics["sync_recommended"] or not graph_state["graph_complete"]
|
||||||
|
)
|
||||||
return diagnostics
|
return diagnostics
|
||||||
|
|
||||||
if conn is not None:
|
if conn is not None:
|
||||||
@@ -6682,7 +6742,7 @@ def reindex_project(args):
|
|||||||
nodes = []
|
nodes = []
|
||||||
cursor = None
|
cursor = None
|
||||||
while True:
|
while True:
|
||||||
query = {"limit": limit}
|
query = {"limit": limit, "view": "full"}
|
||||||
if cursor:
|
if cursor:
|
||||||
query["cursor"] = cursor
|
query["cursor"] = cursor
|
||||||
page = request_json("GET", f"/v1/projects/{project_id}/nodes?{rest_query_string(query)}")
|
page = request_json("GET", f"/v1/projects/{project_id}/nodes?{rest_query_string(query)}")
|
||||||
@@ -6693,13 +6753,21 @@ def reindex_project(args):
|
|||||||
lock = index_file_lock()
|
lock = index_file_lock()
|
||||||
try:
|
try:
|
||||||
with index_connect() as conn:
|
with index_connect() as conn:
|
||||||
|
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_fts WHERE project_id = ?", (project_id,))
|
||||||
|
conn.execute("DELETE FROM local_index_edges WHERE project_id = ?", (project_id,))
|
||||||
conn.execute("DELETE FROM local_index_nodes WHERE project_id = ?", (project_id,))
|
conn.execute("DELETE FROM local_index_nodes WHERE project_id = ?", (project_id,))
|
||||||
finally:
|
finally:
|
||||||
release_index_file_lock(lock)
|
release_index_file_lock(lock)
|
||||||
for node in nodes:
|
for node in nodes:
|
||||||
index_node(node)
|
index_node(node)
|
||||||
return {"project_id": project_id, "indexed": len(nodes), "index_path": str(INDEX_PATH)}
|
set_local_graph_index_complete(project_id, True)
|
||||||
|
return {
|
||||||
|
"project_id": project_id,
|
||||||
|
"indexed": len(nodes),
|
||||||
|
"index_path": str(INDEX_PATH),
|
||||||
|
"graph_complete": True,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
def sync_local_index(args):
|
def sync_local_index(args):
|
||||||
@@ -6709,6 +6777,15 @@ def sync_local_index(args):
|
|||||||
include_archived = as_bool(args.get("include_archived"), True)
|
include_archived = as_bool(args.get("include_archived"), True)
|
||||||
lookback_seconds = normalize_int(args.get("lookback_seconds"), 5, 0, 3600)
|
lookback_seconds = normalize_int(args.get("lookback_seconds"), 5, 0, 3600)
|
||||||
status_before = local_index_diagnostics(project_id)
|
status_before = local_index_diagnostics(project_id)
|
||||||
|
if not status_before.get("graph_complete"):
|
||||||
|
result = reindex_project({**args, "project_id": project_id, "limit": limit})
|
||||||
|
return {
|
||||||
|
**result,
|
||||||
|
"status": "synced",
|
||||||
|
"mode": "full_reindex",
|
||||||
|
"reason": "local_graph_index_incomplete",
|
||||||
|
"index": local_index_diagnostics(project_id),
|
||||||
|
}
|
||||||
latest_node_updated_at = status_before.get("latest_node_updated_at")
|
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)
|
updated_after = args.get("updated_after") or iso_minus_seconds(latest_node_updated_at, lookback_seconds)
|
||||||
nodes = []
|
nodes = []
|
||||||
@@ -6766,7 +6843,9 @@ def clear_local_index(args):
|
|||||||
try:
|
try:
|
||||||
with index_connect() as conn:
|
with index_connect() as conn:
|
||||||
conn.execute("DELETE FROM local_index_fts WHERE project_id = ?", (project_id,))
|
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_nodes WHERE project_id = ?", (project_id,))
|
conn.execute("DELETE FROM local_index_nodes WHERE project_id = ?", (project_id,))
|
||||||
|
set_local_graph_index_complete(project_id, False, conn)
|
||||||
finally:
|
finally:
|
||||||
release_index_file_lock(lock)
|
release_index_file_lock(lock)
|
||||||
return {"project_id": project_id, "cleared": True}
|
return {"project_id": project_id, "cleared": True}
|
||||||
@@ -11000,6 +11079,7 @@ def local_memory_context(args, seed_ids):
|
|||||||
if direction not in {"outgoing", "incoming", "both"}:
|
if direction not in {"outgoing", "incoming", "both"}:
|
||||||
raise RuntimeError("direction must be outgoing, incoming, or both")
|
raise RuntimeError("direction must be outgoing, incoming, or both")
|
||||||
with index_connect() as conn:
|
with index_connect() as conn:
|
||||||
|
graph_state = local_graph_index_state(project_id, conn)
|
||||||
def node_row(node_id=None, path=None):
|
def node_row(node_id=None, path=None):
|
||||||
if node_id:
|
if node_id:
|
||||||
return conn.execute(
|
return conn.execute(
|
||||||
@@ -11088,6 +11168,11 @@ def local_memory_context(args, seed_ids):
|
|||||||
queue = unique_preserve_order(next_ids)
|
queue = unique_preserve_order(next_ids)
|
||||||
if not queue:
|
if not queue:
|
||||||
break
|
break
|
||||||
|
warnings = []
|
||||||
|
if not graph_state["graph_complete"]:
|
||||||
|
warnings.append(
|
||||||
|
"Local graph index coverage is incomplete; run nap index sync before relying on local-only neighborhoods."
|
||||||
|
)
|
||||||
return {
|
return {
|
||||||
"ok": True,
|
"ok": True,
|
||||||
"seed_node_ids": [row["node_id"] for row in seed_rows],
|
"seed_node_ids": [row["node_id"] for row in seed_rows],
|
||||||
@@ -11099,8 +11184,8 @@ def local_memory_context(args, seed_ids):
|
|||||||
"truncated": bool(truncation_reasons),
|
"truncated": bool(truncation_reasons),
|
||||||
"truncation_reasons": truncation_reasons,
|
"truncation_reasons": truncation_reasons,
|
||||||
"source": "local_index",
|
"source": "local_index",
|
||||||
"freshness": {"indexed_at": indexed_at},
|
"freshness": {"indexed_at": indexed_at, **graph_state},
|
||||||
"warnings": [],
|
"warnings": warnings,
|
||||||
"view": "titles",
|
"view": "titles",
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -11155,6 +11240,13 @@ def get_memory_context(args):
|
|||||||
raise RuntimeError("mode must be hybrid, local, or server")
|
raise RuntimeError("mode must be hybrid, local, or server")
|
||||||
if mode in {"local", "hybrid"}:
|
if mode in {"local", "hybrid"}:
|
||||||
try:
|
try:
|
||||||
|
if mode == "hybrid":
|
||||||
|
project_id = resolve_project_id(args)
|
||||||
|
graph_state = local_graph_index_state(project_id)
|
||||||
|
if not graph_state["graph_complete"]:
|
||||||
|
raise RuntimeError(
|
||||||
|
"local graph index coverage is incomplete; run nap index sync"
|
||||||
|
)
|
||||||
return local_memory_context(args, seed_ids)
|
return local_memory_context(args, seed_ids)
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
if mode == "local":
|
if mode == "local":
|
||||||
|
|||||||
@@ -781,7 +781,8 @@ def test_index_status_reports_sync_high_water_mark(mod):
|
|||||||
status = mod.index_status({})
|
status = mod.index_status({})
|
||||||
assert status["count"] == 1
|
assert status["count"] == 1
|
||||||
assert status["latest_node_updated_at"] == "2026-05-06T12:00:00Z"
|
assert status["latest_node_updated_at"] == "2026-05-06T12:00:00Z"
|
||||||
assert status["sync_recommended"] is False
|
assert status["graph_complete"] is False
|
||||||
|
assert status["sync_recommended"] is True
|
||||||
|
|
||||||
|
|
||||||
def test_index_sync_fetches_incremental_changes_and_removes_archived_nodes(mod):
|
def test_index_sync_fetches_incremental_changes_and_removes_archived_nodes(mod):
|
||||||
@@ -794,6 +795,7 @@ def test_index_sync_fetches_incremental_changes_and_removes_archived_nodes(mod):
|
|||||||
existing["id"] = "existing-node"
|
existing["id"] = "existing-node"
|
||||||
existing["updated_at"] = "2026-05-06T12:00:00Z"
|
existing["updated_at"] = "2026-05-06T12:00:00Z"
|
||||||
mod.index_node(existing)
|
mod.index_node(existing)
|
||||||
|
mod.set_local_graph_index_complete("project-1", True)
|
||||||
|
|
||||||
fresh = node("/notes/fresh", "active", content="fresh sync needle")
|
fresh = node("/notes/fresh", "active", content="fresh sync needle")
|
||||||
fresh["id"] = "fresh-node"
|
fresh["id"] = "fresh-node"
|
||||||
@@ -817,6 +819,8 @@ def test_index_sync_fetches_incremental_changes_and_removes_archived_nodes(mod):
|
|||||||
assert result["seen"] == 2
|
assert result["seen"] == 2
|
||||||
assert result["indexed"] == 1
|
assert result["indexed"] == 1
|
||||||
assert result["removed"] == 1
|
assert result["removed"] == 1
|
||||||
|
assert result["mode"] == "incremental"
|
||||||
|
assert result["index"]["graph_complete"] is True
|
||||||
assert calls
|
assert calls
|
||||||
local = mod.search_local_index({"q": "fresh needle"})
|
local = mod.search_local_index({"q": "fresh needle"})
|
||||||
assert [item["id"] for item in local["items"]] == ["fresh-node"]
|
assert [item["id"] for item in local["items"]] == ["fresh-node"]
|
||||||
|
|||||||
@@ -123,6 +123,67 @@ def test_local_context_returns_title_radius_without_node_bodies(mod):
|
|||||||
assert result["source"] == "local_index"
|
assert result["source"] == "local_index"
|
||||||
|
|
||||||
|
|
||||||
|
def test_hybrid_context_falls_back_until_local_graph_coverage_is_complete(mod):
|
||||||
|
state = tempfile.TemporaryDirectory()
|
||||||
|
mod._surface_fallback_state = state
|
||||||
|
mod.AUTH_DIR = pathlib.Path(state.name)
|
||||||
|
mod.INDEX_PATH = mod.AUTH_DIR / "index.sqlite"
|
||||||
|
mod.INDEX_LOCK_PATH = mod.AUTH_DIR / "index.sqlite.lock"
|
||||||
|
mod.index_node({
|
||||||
|
"id": "root", "project_id": "project-1", "full_path": "/docs/root",
|
||||||
|
"folder_path": "/docs", "name": "root", "type": "note",
|
||||||
|
"metadata": {"title": "Root"}, "tags": [], "links": [],
|
||||||
|
"content_text": "body", "updated_at": "2026-08-16T12:00:00Z",
|
||||||
|
})
|
||||||
|
calls = []
|
||||||
|
|
||||||
|
def server_context(_args, seed_ids):
|
||||||
|
calls.append(seed_ids)
|
||||||
|
return {"ok": True, "nodes": [{"id": "root"}], "edges": [], "source": "server"}
|
||||||
|
|
||||||
|
mod.server_memory_context = server_context
|
||||||
|
|
||||||
|
result = mod.get_memory_context({"node_id": "root", "mode": "hybrid"})
|
||||||
|
|
||||||
|
assert result["source"] == "server"
|
||||||
|
assert result["local_fallback"]
|
||||||
|
assert calls == [["root"]]
|
||||||
|
|
||||||
|
|
||||||
|
def test_index_sync_upgrades_pre_graph_index_with_full_reindex(mod):
|
||||||
|
state = tempfile.TemporaryDirectory()
|
||||||
|
mod._surface_upgrade_state = state
|
||||||
|
mod.AUTH_DIR = pathlib.Path(state.name)
|
||||||
|
mod.INDEX_PATH = mod.AUTH_DIR / "index.sqlite"
|
||||||
|
mod.INDEX_LOCK_PATH = mod.AUTH_DIR / "index.sqlite.lock"
|
||||||
|
mod.index_node({
|
||||||
|
"id": "old", "project_id": "project-1", "full_path": "/docs/old",
|
||||||
|
"folder_path": "/docs", "name": "old", "type": "note",
|
||||||
|
"metadata": {}, "tags": [], "links": [], "content_text": "old",
|
||||||
|
"updated_at": "2026-08-16T12:00:00Z",
|
||||||
|
})
|
||||||
|
fresh = {
|
||||||
|
"id": "fresh", "project_id": "project-1", "full_path": "/docs/fresh",
|
||||||
|
"folder_path": "/docs", "name": "fresh", "type": "note",
|
||||||
|
"metadata": {}, "tags": [],
|
||||||
|
"links": [{"node_id": "target", "relation": "references"}],
|
||||||
|
"content_text": "fresh", "updated_at": "2026-08-16T12:01:00Z",
|
||||||
|
}
|
||||||
|
mod.request_json = lambda *_args, **_kwargs: {"items": [fresh], "next_cursor": None}
|
||||||
|
|
||||||
|
result = mod.sync_local_index({})
|
||||||
|
|
||||||
|
assert result["mode"] == "full_reindex"
|
||||||
|
assert result["reason"] == "local_graph_index_incomplete"
|
||||||
|
assert result["graph_complete"] is True
|
||||||
|
assert result["index"]["graph_complete"] is True
|
||||||
|
with mod.index_connect() as conn:
|
||||||
|
assert conn.execute(
|
||||||
|
"SELECT COUNT(*) FROM local_index_edges WHERE project_id = ?",
|
||||||
|
("project-1",),
|
||||||
|
).fetchone()[0] == 1
|
||||||
|
|
||||||
|
|
||||||
def test_discovery_sort_and_returned_item_facets(mod):
|
def test_discovery_sort_and_returned_item_facets(mod):
|
||||||
mod.list_project_nodes = lambda args: {
|
mod.list_project_nodes = lambda args: {
|
||||||
"items": [
|
"items": [
|
||||||
|
|||||||
Reference in New Issue
Block a user