napseer/nap-cli
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
611a4571f3bb3e358b13a664e236d4a70739af44
nap-cli/tests/test_mcp_memory_ergonomics_helpers.py
T
HumbleTurtleandClaude Opus 5.5 0938b3026e Keep the local index fresh from the change feed and improve ranking 
- Track a feed cursor in the local index: reindex takes the head cursor
  before listing and commits it with the snapshot; reads refresh from
  /changes (bounded, throttled) and apply creates, updates, archives,
  restores and hard deletes atomically with the cursor. Existing
  cursorless indexes bootstrap tracking on first read. Failures answer
  locally with explicit freshness warnings; 400 triggers a full
  reindex, 404 falls back to timestamp sync.
- Rank by a weighted score: exact identity (including aliases and
  paths), term coverage including body, intent, bm25, authority, a
  status penalty for superseded/completed records and mild recency.
- Return query-centred FTS5 snippets (never for E2E nodes).
- FTS schema v2: porter stemming, prefix index, title column, rowid
  deletes, two-character terms, domain words no longer stopwords;
  v1 indexes upgrade in place.
- Hyphenated alias/path literals match as adjacent-token phrases and
  rank as exact identity.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-28 22:09:12 -03:00

1383 lines
56 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""Smoke-test active plan, kanban, and plan lifecycle MCP helpers."""
import importlib.util
import pathlib
import sys
import tempfile
import pytest
def load_module():
script_path = pathlib.Path(__file__).resolve().parents[1] / "resources" / "scripts" / "napseer_mcp_server.py"
sys.path.insert(0, str(script_path.parent))
spec = importlib.util.spec_from_file_location("napseer_mcp_server_ergonomics_test", script_path)
module = importlib.util.module_from_spec(spec)
assert spec and spec.loader
spec.loader.exec_module(module)
# Keep project resolution independent from any real .napseer/auth.json in
# the checkout. Store the handle on the module so the directory remains
# alive for the complete test invocation.
module._test_state = tempfile.TemporaryDirectory()
module.AUTH_PATH = pathlib.Path(module._test_state.name) / "auth.json"
module.AUTH = {"account_id": "acct-1", "token": "token", "token_expires_at": "later"}
module.DEFAULT_PROJECT_ID = "project-1"
# Keep broad workflow coverage in this legacy suite. Dedicated profile
# tests exercise the production core-only default.
module.CONFIGURED_TOOL_PROFILES = {"all"}
module.project_memory_encryption_active = lambda project_id: False
return module
@pytest.fixture
def mod():
return load_module()
def node(path, status="planned", metadata=None, tags=None, content="body text", archived_at=None, node_type="plan", links=None):
folder, name = path.rsplit("/", 1)
merged_metadata = {"status": status}
if metadata:
merged_metadata.update(metadata)
return {
"id": f"id-{name}",
"project_id": "project-1",
"full_path": path,
"folder_path": folder,
"name": name,
"type": node_type,
"tags": tags or [],
"links": links or [],
"metadata": merged_metadata,
"content_text": content,
"updated_at": "2026-05-06T12:00:00Z",
"archived_at": archived_at,
"encryption_state": "plaintext",
}
def test_active_plan_listing(mod):
active = node("/plans/active", "in_progress", tags=["plan"])
completed = node("/plans/completed", "completed", tags=["plan", "completed"])
archived = node("/plans/archived", "planned", tags=["plan"], archived_at="2026-05-06T12:01:00Z")
def fake_request(method, path, payload=None, **kwargs):
assert method == "GET"
assert "/nodes?" in path
assert "folder_path=%2Fplans" in path
return {"items": [active, completed, archived], "next_cursor": None}
mod.request_json = fake_request
result = mod.list_active_plans({})
assert result["ok"] is True
assert result["view"] == "summary"
assert [item["full_path"] for item in result["items"]] == ["/plans/active"]
assert "content_text" not in result["items"][0]
assert result["items"][0]["status"] == "in_progress"
def test_node_discovery_allows_large_requested_limit(mod):
calls = []
def fake_request(method, path, payload=None, **kwargs):
calls.append(path)
assert method == "GET"
assert "limit=250" in path
return {"items": [node(f"/plans/item-{index}") for index in range(250)], "next_cursor": None}
mod.request_json = fake_request
result = mod.list_project_nodes({"folder_path": "/plans", "limit": 250})
assert result["ok"] is True
assert len(result["items"]) == 250
assert result["budget"]["limit"] == 250
assert result["budget"]["max_limit"] > 200
assert "content_text" not in result["items"][0]
assert len(calls) == 1
def test_node_by_path_returns_canonical_identity_uri(mod):
mod.AUTH["project_slug"] = "napseer"
found = node("/documentation/product/prd", "active", node_type="product_spec")
found["id"] = "node-123"
def fake_request(method, path, payload=None, **kwargs):
assert method == "GET"
assert "/nodes/by-path?" in path
assert "path=%2Fdocumentation%2Fproduct%2Fprd" in path
assert "view=render" not in path
return {
"identity": {
"node_id": "node-123",
"canonical_uri": "nap://napseer/project-1/node-123",
"project_id": "project-1",
"project_name": "napseer",
"lookup_kind": "path",
"lookup_value": "/documentation/product/prd",
"legacy_full_path": "/documentation/product/prd",
},
"route": {
"id": "node-123",
"project_id": "project-1",
"full_path": "/documentation/product/prd",
"type": "product_spec",
"name": "prd",
"folder_path": "/documentation/product",
},
"warnings": ["Path lookup is index/route resolution only."],
}
mod.request_json = fake_request
result = mod.node_by_path({"path": "documentation/product/prd"})
assert result["ok"] is True
assert result["status"] == "resolved"
assert result["identity"] == {
"node_id": "node-123",
"canonical_uri": "nap://napseer/project-1/node-123",
"project_id": "project-1",
"project_name": "napseer",
"lookup_kind": "path",
"lookup_value": "/documentation/product/prd",
"legacy_full_path": "/documentation/product/prd",
}
assert result["route"]["full_path"] == "/documentation/product/prd"
assert "content_text" not in result["route"]
assert "preview" not in result["route"]
assert "read_fingerprint" not in result["route"]
assert "Path lookup is index/route resolution only" in result["warnings"][0]
def test_node_by_path_resolves_project_anchor_shortcut(mod):
def fake_request(method, path, payload=None, **kwargs):
assert method == "GET"
assert path.endswith("/nodes/by-path?path=%40project")
return {
"identity": {
"node_id": "anchor-1",
"canonical_uri": "nap://napseer/project-1/anchor-1",
"project_id": "project-1",
"lookup_kind": "shortcut",
"lookup_value": "@project",
"legacy_full_path": "/routes/project/core",
},
"route": {
"id": "anchor-1",
"project_id": "project-1",
"full_path": "/routes/project/core",
},
}
mod.request_json = fake_request
result = mod.node_by_path({"shortcut": "@Project"})
assert result["identity"]["lookup_kind"] == "shortcut"
assert result["identity"]["lookup_value"] == "@project"
assert result["route"]["full_path"] == "/routes/project/core"
def test_path_resolver_then_read_uses_node_id_route(mod):
calls = []
def fake_request(method, path, payload=None, **kwargs):
calls.append(path)
assert method == "GET"
if "/nodes/by-path?" in path:
assert "view=" not in path
return {
"identity": {"node_id": "node-123", "project_id": "project-1"},
"route": {"id": "node-123", "project_id": "project-1", "full_path": "/documentation/product/prd"},
}
assert path == "/v1/projects/project-1/nodes/node-123"
return {**node("/documentation/product/prd", "active", node_type="product_spec"), "id": "node-123"}
mod.request_json = fake_request
result = mod.read_node_by_path({"path": "/documentation/product/prd"})
assert result["id"] == "node-123"
assert calls == [
"/v1/projects/project-1/nodes/by-path?path=%2Fdocumentation%2Fproduct%2Fprd",
"/v1/projects/project-1/nodes/node-123",
]
def test_node_by_id_returns_canonical_identity_uri(mod):
mod.AUTH["project_slug"] = "napseer"
found = node("/documentation/product/prd", "active", node_type="product_spec")
found["id"] = "node-123"
def fake_request(method, path, payload=None, **kwargs):
assert method == "GET"
assert path == "/v1/projects/project-1/nodes/node-123"
return found
mod.request_json = fake_request
result = mod.node_by_id({"node_id": "node-123"})
assert result["ok"] is True
assert result["identity"]["lookup_kind"] == "node_id"
assert result["identity"]["canonical_uri"] == "nap://napseer/project-1/node-123"
assert result["node"]["id"] == "node-123"
def test_node_by_uri_resolves_canonical_uri(mod):
mod.AUTH["project_slug"] = "napseer"
found = node("/documentation/product/prd", "active", node_type="product_spec")
found["id"] = "node-123"
def fake_request(method, path, payload=None, **kwargs):
assert method == "GET"
assert path == "/v1/projects/project-1/nodes/node-123"
return found
mod.request_json = fake_request
result = mod.node_by_uri({"uri": "nap://napseer/project-1/node-123"})
assert result["ok"] is True
assert result["identity"]["lookup_kind"] == "uri"
assert result["identity"]["lookup_value"] == "nap://napseer/project-1/node-123"
assert result["identity"]["canonical_uri"] == "nap://napseer/project-1/node-123"
def test_node_get_accepts_node_id_without_path(mod):
found = node("/documentation/product/prd", "active", node_type="product_spec")
found["id"] = "node-123"
def fake_request(method, path, payload=None, **kwargs):
assert method == "GET"
assert path == "/v1/projects/project-1/nodes/node-123"
return found
mod.request_json = fake_request
result = mod.node_get({"node_id": "node-123", "view": "content"})
assert result["id"] == "node-123"
assert result["full_path"] == "/documentation/product/prd"
def test_node_get_edit_view_uses_id_route_not_by_path(mod):
edit_payload = {
"node": node("/documentation/product/prd", "active", node_type="product_spec"),
"revision": "2026-05-06T12:00:00Z",
"read_fingerprint": "fingerprint-1",
"content_lines": ["body text"],
"line_count": 1,
"ends_with_newline": False,
}
edit_payload["node"]["id"] = "node-123"
def fake_request(method, path, payload=None, **kwargs):
assert method == "GET"
assert path == "/v1/projects/project-1/nodes/node-123?view=edit"
return edit_payload
mod.request_json = fake_request
result = mod.node_get({"node_id": "node-123", "view": "edit"})
assert result["revision"] == "2026-05-06T12:00:00Z"
assert result["node"]["id"] == "node-123"
def test_node_get_rejects_path_reads(mod):
try:
mod.node_get({"path": "/documentation/product/prd", "view": "content"})
except RuntimeError as exc:
assert "path is an index/route only" in str(exc)
else:
raise AssertionError("path reads should be rejected")
def test_tools_include_node_by_path_descriptor(mod):
tool_by_name = {item["name"]: item for item in mod.tools()}
assert len(tool_by_name) == 60
assert len(tool_by_name) == len(mod.tools())
get_tool = next(item for item in mod.tools() if item["name"] == "nap_node_get")
assert get_tool["inputSchema"].get("required") is None
assert "node_id" in get_tool["inputSchema"]["properties"]
assert "uri" in get_tool["inputSchema"]["properties"]
assert "path" not in get_tool["inputSchema"]["properties"]
assert "node_id returned by nap_discover" in get_tool["description"]
discover_tool = next(item for item in mod.tools() if item["name"] == "nap_discover")
assert "node_id on every row" in discover_tool["description"]
assert "directly to nap_node_get" in discover_tool["description"]
tool = next(item for item in mod.tools() if item["name"] == "nap_node_by_path")
assert tool["inputSchema"].get("required") is None
assert tool["inputSchema"]["oneOf"] == [
{"required": ["path"]},
{"required": ["shortcut"]},
]
assert "shortcut" in tool["inputSchema"]["properties"]
assert "preview_chars" not in tool["inputSchema"]["properties"]
assert "does not read full node content" in tool["description"]
for hidden_name in [
"nap_cat",
"nap_node_by_id",
"nap_node_by_uri",
"nap_tee",
"nap_patch",
"nap_find",
"nap_index_sync",
"nap_recent",
"nap_find_related",
"nap_backlinks",
"nap_ls_folders",
"nap_ls_tags",
]:
assert hidden_name not in tool_by_name
for name in ["nap_discover", "nap_context"]:
schema = next(item for item in mod.tools() if item["name"] == name)["inputSchema"]
view = schema["properties"].get("view")
if view:
assert "full" not in view["enum"]
for name in ["nap_node_patch", "nap_rm"]:
schema = next(item for item in mod.tools() if item["name"] == name)["inputSchema"]
assert "node_id" in schema["properties"]
assert "uri" in schema["properties"]
assert "path" not in schema["properties"]
bulk_node_schema = next(item for item in mod.tools() if item["name"] == "nap_bulk")["inputSchema"]["properties"]["nodes"]["items"]
assert bulk_node_schema.get("required") is None
assert "path" in bulk_node_schema["properties"]
assert "node_id" in bulk_node_schema["properties"]
assert "uri" in bulk_node_schema["properties"]
for required_name in [
"nap_library_ls",
"nap_library_attach",
"nap_library_detach",
"nap_library_nodes",
"nap_library_node_get",
"nap_auth_refresh",
"nap_node_restore",
"nap_plan_transition",
"nap_index_status",
]:
assert required_name in tool_by_name
def test_mutation_tools_reject_path_and_patch_by_node_id(mod):
found = node("/notes/current", "active", node_type="note")
found["id"] = "node-123"
writes = []
mod.get_node_by_id = lambda args: found
mod.request_project_write = lambda method, path, payload, *args, **kwargs: writes.append((method, path, payload, args)) or {
**found,
"content_text": payload.get("content_text", found["content_text"]) if payload else found["content_text"],
}
mod.index_node = lambda node: None
mod.remove_indexed_node = lambda node_id: None
updated = mod.update_node_by_path({"node_id": "node-123", "content_text": "updated"})
assert updated["updated"] is True
assert writes[-1][0] == "PATCH"
assert writes[-1][1] == "/v1/projects/project-1/nodes/node-123"
assert writes[-1][3][2:] == ("node", "node-123")
patched = mod.guarded_node_patch({
"node_id": "node-123",
"precondition": {"revision": "r1", "read_fingerprint": "fp1"},
"set": {"metadata": {"status": "active"}},
})
assert patched["id"] == "node-123"
assert writes[-1][1] == "/v1/projects/project-1/nodes/node-123/patch"
archived = mod.archive_node_by_path({"node_id": "node-123"})
assert archived["archived"] is True
assert writes[-1][0] == "DELETE"
assert writes[-1][1] == "/v1/projects/project-1/nodes/node-123"
with pytest.raises(RuntimeError, match="path is an index/route only"):
mod.update_node_by_path({"path": "/notes/current", "content_text": "bad"})
def test_graph_tools_use_node_identity_instead_of_paths(mod):
source = node("/notes/source", "active", node_type="note")
source["id"] = "source-123"
target = node("/notes/target", "active", node_type="note")
target["id"] = "target-123"
writes = []
reads = []
def fake_get_node_by_id(args):
reads.append(args)
node_id = args.get("node_id")
if node_id == "source-123":
return source
if node_id == "target-123":
return target
raise AssertionError(f"unexpected node_id {node_id}")
def fake_write(method, path, payload, *args, **kwargs):
writes.append((method, path, payload, args))
return {**source, "full_path": "/notes/source-renamed" if payload.get("name") == "source-renamed" else source["full_path"], "links": payload.get("links", source["links"])}
def fake_request(method, path, payload=None, **kwargs):
assert method == "GET"
assert path == "/v1/projects/project-1/nodes/target-123/backlinks"
return {"items": [source]}
mod.get_node_by_id = fake_get_node_by_id
mod.request_project_write = fake_write
mod.request_json = fake_request
mod.index_node = lambda node: None
mod.inbound_reference_plan = lambda path: [{"id": "other-1", "path": "/notes/other"}]
mod.remove_indexed_node = lambda node_id: None
linked = mod.link_nodes({"source_node_id": "source-123", "target_node_id": "target-123", "relation": "references"})
assert linked["created"] is True
assert writes[-1][0] == "PATCH"
assert writes[-1][1] == "/v1/projects/project-1/nodes/source-123"
assert writes[-1][2]["links"] == [{"path": "/notes/target", "relation": "references"}]
assert writes[-1][3][2:] == ("node", "source-123")
backlinks = mod.get_backlinks_by_path({"node_id": "target-123"})
assert backlinks["target"]["id"] == "target-123"
moved = mod.move_node({"node_id": "source-123", "new_path": "/notes/source-renamed"})
assert moved["moved"] is True
assert writes[-1][1] == "/v1/projects/project-1/nodes/source-123"
assert writes[-1][2]["folder_path"] == "/notes"
assert writes[-1][2]["name"] == "source-renamed"
assert writes[-1][3][2:] == ("node", "source-123")
with pytest.raises(RuntimeError, match="paths are index/routes only"):
mod.link_nodes({"source_path": "/notes/source", "target_path": "/notes/target"})
with pytest.raises(RuntimeError, match="path is an index/route only"):
mod.get_backlinks_by_path({"path": "/notes/target"})
with pytest.raises(RuntimeError, match="path is an index/route only"):
mod.move_node({"path": "/notes/source", "new_path": "/notes/source-renamed"})
def test_tee_and_bulk_create_by_path_update_by_identity(mod):
existing = node("/notes/current", "active", node_type="note")
existing["id"] = "node-123"
writes = []
mod.try_get_node_by_path = lambda path, allow_agent=False: existing if path == "/notes/current" else None
mod.get_node_by_id = lambda args: existing
mod.request_project_write = lambda method, path, payload, *args, **kwargs: writes.append((method, path, payload, args)) or {
**existing,
"id": "node-123" if method == "PATCH" else "node-new",
"full_path": "/notes/new" if method == "POST" else existing["full_path"],
"content_text": payload.get("content_text", existing["content_text"]),
}
mod.index_node = lambda node: None
mod.acquire_project_lock = lambda args, **kwargs: {"id": "lock-1", "lease_token": "lease-1"}
mod.lock_headers = lambda lock: {"X-Lock": lock["id"]}
mod.release_project_lock = lambda args, **kwargs: None
mod.request_json = lambda method, path, payload=None, **kwargs: writes.append((method, path, payload, ())) or {
**existing,
"content_text": payload.get("content_text", existing["content_text"]) if payload else existing["content_text"],
}
created = mod.upsert_node({"path": "/notes/new", "content_text": "new"})
assert created["created"] is True
assert writes[-1][0] == "POST"
updated = mod.upsert_node({"node_id": "node-123", "content_text": "updated"})
assert updated["updated"] is True
assert writes[-1][0] == "PATCH"
assert writes[-1][1] == "/v1/projects/project-1/nodes/node-123"
compatibility_update = mod.upsert_node(
{"path": "/notes/current", "content_text": "compatibility update"}
)
assert compatibility_update["updated"] is True
assert writes[-1][0] == "PATCH"
bulk = mod.bulk_upsert_nodes({"nodes": [{"node_id": "node-123", "content_text": "bulk update"}]})
assert bulk["items"][0]["created"] is False
assert writes[-1][1] == "/v1/projects/project-1/nodes/node-123"
compatibility_bulk_update = mod.bulk_upsert_nodes(
{"nodes": [{"path": "/notes/current", "content_text": "compatibility update"}]}
)
assert compatibility_bulk_update["items"][0]["created"] is False
assert writes[-1][0] == "PATCH"
def test_agent_tools_create_by_route_and_update_by_identity(mod):
agent_file = node("/agents/alice/memory/current", "active", node_type="agent-file")
agent_file["id"] = "agent-node-123"
target = node("/notes/target", "active", node_type="note")
target["id"] = "target-123"
writes = []
def fake_get_node_by_id(args):
node_id = args.get("node_id")
if node_id == "agent-node-123":
return agent_file
if node_id == "target-123":
return target
raise AssertionError(f"unexpected node_id {node_id}")
mod.get_node_by_id = fake_get_node_by_id
mod.try_get_node_by_path = lambda path, allow_agent=False: agent_file if path == "/agents/alice/memory/current" else None
mod.request_project_write = lambda method, path, payload, *args, **kwargs: writes.append((method, path, payload, args)) or {
**agent_file,
"id": "agent-node-new" if method == "POST" else agent_file["id"],
"full_path": f"{payload.get('folder_path')}/{payload.get('name')}" if method == "POST" else agent_file["full_path"],
"content_text": payload.get("content_text", agent_file["content_text"]) if payload else agent_file["content_text"],
"links": payload.get("links", agent_file["links"]) if payload else agent_file["links"],
}
mod.index_node = lambda node: None
mod.remove_indexed_node = lambda node_id: None
read = mod.get_agent_node({"agent_slug": "alice", "node_id": "agent-node-123"})
assert read["id"] == "agent-node-123"
created = mod.upsert_agent_node({"agent_slug": "alice", "path": "/memory/new", "content_text": "new"})
assert created["created"] is True
assert writes[-1][0] == "POST"
assert writes[-1][2]["folder_path"] == "/agents/alice/memory"
updated = mod.upsert_agent_node({"agent_slug": "alice", "node_id": "agent-node-123", "content_text": "updated"})
assert updated["updated"] is True
assert writes[-1][1] == "/v1/projects/project-1/nodes/agent-node-123"
assert writes[-1][3][2:] == ("node", "agent-node-123")
patched = mod.update_agent_node({"agent_slug": "alice", "node_id": "agent-node-123", "content_text": "patched"})
assert patched["updated"] is True
archived = mod.archive_agent_node({"agent_slug": "alice", "node_id": "agent-node-123"})
assert archived["archived"] is True
assert writes[-1][0] == "DELETE"
assert writes[-1][3][2:] == ("node", "agent-node-123")
linked = mod.link_agent_node({"agent_slug": "alice", "source_node_id": "agent-node-123", "target_node_id": "target-123"})
assert linked["created"] is True
assert writes[-1][2]["links"] == [{"path": "/notes/target", "relation": "references"}]
assert writes[-1][3][2:] == ("node", "agent-node-123")
with pytest.raises(RuntimeError, match="path already resolves"):
mod.upsert_agent_node({"agent_slug": "alice", "path": "/memory/current", "content_text": "bad"})
with pytest.raises(RuntimeError, match="path is an index/route only"):
mod.update_agent_node({"agent_slug": "alice", "path": "/memory/current", "content_text": "bad"})
with pytest.raises(RuntimeError, match="paths are index/routes only"):
mod.link_agent_node({"agent_slug": "alice", "source_path": "/memory/current", "target_path": "/notes/target"})
def test_context_and_related_use_identity_inputs(mod):
root = node("/notes/root", "active", node_type="note", links=[{"path": "/notes/linked", "relation": "references"}], tags=["topic"])
root["id"] = "root-123"
linked = node("/notes/linked", "active", node_type="note", tags=["topic"])
linked["id"] = "linked-123"
backlink = node("/notes/backlink", "active", node_type="note")
backlink["id"] = "backlink-123"
reads = []
def fake_get_node_by_id(args):
reads.append(args)
node_id = args.get("node_id")
if node_id == "root-123":
return root
if node_id == "linked-123":
return linked
raise AssertionError(f"unexpected node_id {node_id}")
def fake_read_by_path(args, allow_agent=False):
if args["path"] == "/notes/root":
return root
if args["path"] == "/notes/linked":
return linked
if args["path"] == "/notes/backlink":
return backlink
raise RuntimeError("HTTP 404")
def fake_request(method, path, payload=None, **kwargs):
assert method == "GET"
if "/nodes/root-123/neighborhood?" in path:
return {
"nodes": [
{"id": "root-123", "full_path": "/notes/root", "title": "root", "depth": 0},
{"id": "linked-123", "full_path": "/notes/linked", "title": "linked", "depth": 1},
{"id": "backlink-123", "full_path": "/notes/backlink", "title": "backlink", "depth": 1},
],
"edges": [
{"source": "root-123", "target": "linked-123", "relation": "references"},
{"source": "backlink-123", "target": "root-123", "relation": "references"},
],
"levels": [{"depth": 0, "node_ids": ["root-123"]}, {"depth": 1, "node_ids": ["linked-123", "backlink-123"]}],
"frontier": [],
"truncated": False,
}
if path == "/v1/projects/project-1/nodes/root-123/backlinks":
return {"items": [backlink]}
if path == "/v1/projects/project-1/nodes/linked-123/backlinks":
return {"items": []}
raise AssertionError(path)
mod.get_node_by_id = fake_get_node_by_id
mod.read_node_by_path = fake_read_by_path
mod.request_json = fake_request
mod.node_by_path = lambda args: {"identity": {"node_id": "root-123"}}
mod.list_project_nodes = lambda args: {"items": [root, linked]}
context = mod.get_memory_context({"node_id": "root-123", "depth": 1, "mode": "server"})
assert [item["full_path"] for item in context["items"]] == ["/notes/root", "/notes/linked", "/notes/backlink"]
assert {"source": "root-123", "target": "linked-123", "relation": "references"} in context["edges"]
assert {"source": "backlink-123", "target": "root-123", "relation": "references"} in context["edges"]
related = mod.find_related_nodes({"node_id": "root-123", "view": "paths"})
assert related["target"]["full_path"] == "/notes/root"
assert {item["full_path"] for item in related["items"]} >= {"/notes/backlink", "/notes/linked"}
path_context = mod.get_memory_context({"path": "/notes/root", "depth": 0, "mode": "server"})
assert path_context["items"][0]["full_path"] == "/notes/root"
with pytest.raises(RuntimeError, match="path is an index/route only"):
mod.find_related_nodes({"path": "/notes/root"})
def test_hybrid_context_resolves_path_inside_complete_local_snapshot(mod):
with tempfile.TemporaryDirectory() as tmp:
state_dir = pathlib.Path(tmp)
mod.AUTH_DIR = state_dir
mod.INDEX_PATH = state_dir / "index.sqlite"
mod.INDEX_LOCK_PATH = state_dir / "index.lock"
indexed = node("/notes/local-root", "active", node_type="note")
indexed["id"] = "local-root-123"
mod.index_node(indexed)
mod.set_local_graph_index_complete("project-1", True)
def unexpected_request(*_args, **_kwargs):
raise AssertionError("complete local context must not make an HTTP request")
mod.request_json = unexpected_request
result = mod.get_memory_context({
"path": "/notes/local-root",
"depth": 0,
"mode": "hybrid",
})
assert result["source"] == "local_index"
assert result["seed_node_ids"] == ["local-root-123"]
assert result["items"][0]["full_path"] == "/notes/local-root"
with mod.index_connect() as conn:
plan = conn.execute(
"EXPLAIN QUERY PLAN SELECT node_id FROM local_index_nodes "
"WHERE project_id = ? AND full_path = ? AND archived = 0",
("project-1", "/notes/local-root"),
).fetchall()
assert any("local_index_nodes_project_path" in row["detail"] for row in plan)
def test_local_helper_ui_forms_submit_identity_for_existing_node_actions():
source = (pathlib.Path(__file__).resolve().parents[1] / "resources" / "scripts" / "napseer_mcp_server.py").read_text()
assert "name='node_id' value='{html.escape(item.get('id') or '')}'" in source
assert 'name="source_node_id"' in source
assert 'name="target_node_id"' in source
assert 'archive_node_by_path({"node_id": form.get("node_id", [""])[0]})' in source
assert '"source_node_id": form.get("source_node_id", [""])[0]' in source
assert '"target_node_id": form.get("target_node_id", [""])[0]' in source
assert 'name="source_path"' not in source
assert 'name="target_path"' not in source
def test_discovery_rejects_full_content_view(mod):
with pytest.raises(RuntimeError, match="full node bodies require nap_node_get"):
mod.list_project_nodes({"view": "full"})
def test_include_content_is_legacy_noop(mod):
found = node("/plans/found", "planned", content="body must not be returned")
def fake_request(method, path, payload=None, **kwargs):
assert method == "GET"
assert "view=paths" in path
return {"items": [found], "next_cursor": None}
mod.request_json = fake_request
result = mod.list_project_nodes({"include_content": True, "view": "paths"})
assert result["view"] == "paths"
assert "content_text" not in result["items"][0]
def test_recent_memory_defaults_to_paths(mod):
recent = node(
"/notes/recent-title",
"active",
metadata={"title": "Readable Recent Title", "priority": "high"},
tags=["recent"],
content="large body that should not appear in the default recent response",
node_type="note",
)
def fake_request(method, path, payload=None, **kwargs):
assert method == "GET"
assert "/nodes?" in path
assert "view=paths" in path
return {"items": [recent], "next_cursor": None}
mod.request_json = fake_request
result = mod.recent_memory({})
assert result["ok"] is True
assert result["view"] == "paths"
assert result["items"] == [{
"node_id": "id-recent-title",
"full_path": "/notes/recent-title",
"type": "note",
"status": "active",
"updated_at": "2026-05-06T12:00:00Z",
}]
assert "content_text" not in result["items"][0]
assert "metadata_summary" not in result["items"][0]
def test_recent_memory_accepts_date_aliases_and_returns_dates(mod):
recent = node(
"/notes/recent-title",
"active",
metadata={"date": "2026-05-06"},
tags=["recent"],
node_type="note",
)
recent["created_at"] = "2026-05-05T10:00:00Z"
def fake_request(method, path, payload=None, **kwargs):
assert method == "GET"
assert "updated_after=2026-05-05T00%3A00%3A00Z" in path
assert "updated_before=2026-05-06T23%3A59%3A59Z" in path
return {"items": [recent], "next_cursor": None}
mod.request_json = fake_request
result = mod.recent_memory({"since": "2026-05-05", "until": "2026-05-06"})
assert result["items"] == [{
"node_id": "id-recent-title",
"full_path": "/notes/recent-title",
"type": "note",
"status": "active",
"date": "2026-05-06",
"created_at": "2026-05-05T10:00:00Z",
"updated_at": "2026-05-06T12:00:00Z",
}]
def test_find_defaults_to_paths_and_requests_compact_rest_view(mod):
found = node(
"/plans/found",
"planned",
content="large body that should not be fetched by default",
)
def fake_request(method, path, payload=None, **kwargs):
assert method == "GET"
assert "/nodes?" in path
assert "folder_path=%2Fplans" in path
assert "view=paths" in path
return {"items": [found], "next_cursor": None}
mod.request_json = fake_request
result = mod.list_project_nodes({"folder_path": "/plans"})
assert result["ok"] is True
assert result["view"] == "paths"
assert result["items"] == [{
"node_id": "id-found",
"full_path": "/plans/found",
"type": "plan",
"status": "planned",
"updated_at": "2026-05-06T12:00:00Z",
}]
assert "content_text" not in result["items"][0]
def test_discover_uses_search_when_keywords_present(mod):
calls = []
def fake_search(args):
calls.append(args)
return {"ok": True, "items": [], "view": args.get("view")}
mod.search_memory = fake_search
result = mod.discover_memory({"keywords": "gateway relay", "since": "2026-05-01"})
assert result["tool_intent"] == "search"
assert result["range"]["updated_after"] == "2026-05-01T00:00:00Z"
assert calls and calls[-1]["q"] == "gateway relay"
assert calls[-1]["view"] == "summary"
assert "date" in result["awareness"]["date_fields_in_rows"]
assert "item.node_id" in result["awareness"]["next_step"]
assert "outside discovery" in result["awareness"]["next_step"]
def test_classify_query_points_exact_paths_to_context(mod):
result = mod.classify_memory_query({"q": "explain /documentation/security/overview since May", "since": "2026-05-01"})
assert result["classification"]["primary_tool"] == "nap_context"
assert result["classification"]["exact_paths"] == ["/documentation/security/overview"]
assert result["classification"]["range"]["updated_after"] == "2026-05-01T00:00:00Z"
def test_title_view_requests_summary_rest_view_without_content(mod):
found = node(
"/notes/titled",
"active",
metadata={"title": "Readable Title"},
content="large body that should not be fetched by title view",
)
def fake_request(method, path, payload=None, **kwargs):
assert method == "GET"
assert "/nodes?" in path
assert "view=summary" in path
return {"items": [found], "next_cursor": None}
mod.request_json = fake_request
result = mod.list_project_nodes({"view": "titles"})
assert result["ok"] is True
assert result["view"] == "titles"
assert result["items"] == [{
"node_id": "id-titled",
"title": "Readable Title",
"full_path": "/notes/titled",
}]
assert "content_text" not in result["items"][0]
def test_rest_query_string_serializes_booleans_lowercase(mod):
query = mod.rest_query_string({
"folder_path": "/kanban",
"include_archived": False,
"archived_only": False,
"active_only": True,
"status": ["todo", "doing"],
})
assert "include_archived=false" in query
assert "archived_only=false" in query
assert "active_only=true" in query
assert "include_archived=False" not in query
assert "active_only=True" not in query
assert "status=todo" in query
assert "status=doing" in query
def test_index_status_reports_sync_high_water_mark(mod):
with tempfile.TemporaryDirectory() as tmp:
state_dir = pathlib.Path(tmp)
mod.AUTH_DIR = state_dir
mod.INDEX_PATH = state_dir / "index.sqlite"
mod.INDEX_LOCK_PATH = state_dir / "index.lock"
mod.index_node(node("/notes/old", "active", content="old body"))
status = mod.index_status({})
assert status["count"] == 1
assert status["latest_node_updated_at"] == "2026-05-06T12:00:00Z"
assert status["graph_complete"] is False
assert status["sync_recommended"] is True
def test_index_sync_fetches_incremental_changes_and_removes_archived_nodes(mod):
with tempfile.TemporaryDirectory() as tmp:
state_dir = pathlib.Path(tmp)
mod.AUTH_DIR = state_dir
mod.INDEX_PATH = state_dir / "index.sqlite"
mod.INDEX_LOCK_PATH = state_dir / "index.lock"
existing = node("/notes/existing", "active", content="existing body")
existing["id"] = "existing-node"
existing["updated_at"] = "2026-05-06T12:00:00Z"
mod.index_node(existing)
mod.set_local_graph_index_complete("project-1", True)
fresh = node("/notes/fresh", "active", content="fresh sync needle")
fresh["id"] = "fresh-node"
fresh["updated_at"] = "2026-05-06T12:05:00Z"
archived = node("/notes/existing", "active", content="archived body", archived_at="2026-05-06T12:06:00Z")
archived["id"] = "existing-node"
archived["updated_at"] = "2026-05-06T12:06:00Z"
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
assert "include_archived=true" in path
assert "updated_after=2026-05-06T11%3A59%3A55Z" in path
return {"items": [fresh, archived], "next_cursor": None}
mod.request_json = fake_request
result = mod.sync_local_index({"lookback_seconds": 5})
assert result["status"] == "synced"
assert result["seen"] == 2
assert result["indexed"] == 1
assert result["removed"] == 1
assert result["mode"] == "incremental"
assert result["index"]["graph_complete"] is True
assert calls
local = mod.search_local_index({"q": "fresh needle"})
assert [item["id"] for item in local["items"]] == ["fresh-node"]
assert mod.search_local_index({"q": "existing"})["items"] == []
def test_tools_include_read_only_index_status_descriptor(mod):
tool = next(item for item in mod.tools() if item["name"] == "nap_index_status")
assert "local search index" in tool["description"]
assert tool["inputSchema"]["properties"] == {}
contract = mod.tool_contract_metadata("nap_index_status")
assert contract["side_effect"] == "none"
assert contract["auth_mode"] == "local-only"
def test_kanban_list_grouping_and_filters(mod):
doing = node(
"/kanban/doing/fix-wrapper",
"doing",
metadata={"column": "doing", "priority": "high", "owner": "codex", "blocked": True},
tags=["kanban"],
node_type="kanban_card",
)
todo = node(
"/kanban/todo/add-tests",
"todo",
metadata={"column": "todo", "priority": "normal", "owner": "codex", "blocked": False},
tags=["kanban"],
node_type="kanban_card",
)
done = node(
"/kanban/done/old-card",
"done",
metadata={"column": "done", "owner": "codex"},
tags=["kanban"],
node_type="kanban_card",
)
outside = node("/plans/not-kanban", "planned")
def fake_request(method, path, payload=None, **kwargs):
assert method == "GET"
assert "q=" in path
assert "include_archived=false" in path
assert "archived_only=false" in path
return {"items": [doing, todo, done, outside], "next_cursor": None}
mod.request_json = fake_request
result = mod.list_kanban_cards({"owner": "codex", "blocked": True, "group_by": "column"})
assert result["ok"] is True
assert result["kanban_root"] == "/kanban"
assert [item["full_path"] for item in result["items"]] == ["/kanban/doing/fix-wrapper"]
assert "content_text" not in result["items"][0]
assert result["items"][0]["column"] == "doing"
assert result["items"][0]["blocked"] is True
assert list(result["groups"]) == ["doing"]
def test_kanban_dependency_summary_filters_ready_and_blocking(mod):
plan = node(
"/kanban/backlog/plan",
"backlog",
metadata={"column": "backlog", "priority": "urgent", "blocked": False},
tags=["kanban"],
node_type="kanban_card",
links=[{"path": "/kanban/backlog/implementation", "relation": "blocks"}],
)
implementation = node(
"/kanban/todo/implementation",
"todo",
metadata={
"column": "todo",
"priority": "high",
"blocked_by": "/kanban/backlog/plan",
"external_blockers": [{"id": "deploy", "state": "open", "strength": "hard"}],
},
tags=["kanban"],
node_type="kanban_card",
links=[
{"path": "/kanban/backlog/plan", "relation": "blocked-by"},
{"path": "/kanban/backlog/context", "relation": "relates-to"},
{"path": "/kanban/backlog/child", "relation": "parent_of"},
{"path": "/kanban/backlog/canonical", "relation": "duplicate-of"},
],
)
def fake_request(method, path, payload=None, **kwargs):
assert method == "GET"
return {"items": [plan, implementation], "next_cursor": None}
mod.request_json = fake_request
blocked = mod.list_kanban_cards({"dependency_view": "blocked", "limit": 10})
assert [item["full_path"] for item in blocked["items"]] == ["/kanban/todo/implementation"]
assert blocked["items"][0]["dependency_summary"]["hard_blockers"] == 1
assert blocked["items"][0]["dependency_summary"]["external_blockers"] == 1
assert blocked["items"][0]["hard_blockers"] == ["/kanban/backlog/plan"]
assert blocked["items"][0]["related"] == ["/kanban/backlog/context"]
assert blocked["items"][0]["children"] == ["/kanban/backlog/child"]
assert blocked["items"][0]["duplicates"] == ["/kanban/backlog/canonical"]
ready = mod.list_kanban_cards({"dependency_view": "ready", "limit": 10})
assert [item["full_path"] for item in ready["items"]] == ["/kanban/backlog/plan"]
blocking = mod.list_kanban_cards({"dependency_view": "blocking", "limit": 10})
assert blocking["items"][0]["blocking"] == ["/kanban/backlog/implementation"]
def test_kanban_list_reads_beyond_first_source_page(mod):
first_page = [
node(
"/kanban/todo/other-card",
"todo",
metadata={"column": "todo", "owner": "codex", "blocked": False},
tags=["kanban"],
node_type="kanban_card",
)
]
second_page = [
node(
"/kanban/doing/blocked-card",
"doing",
metadata={"column": "doing", "owner": "codex", "blocked": True},
tags=["kanban"],
node_type="kanban_card",
)
]
calls = []
def fake_request(method, path, payload=None, **kwargs):
calls.append(path)
assert method == "GET"
assert "q=%2Fkanban" in path
assert "limit=10000" in path
assert "include_archived=false" in path
assert "archived_only=false" in path
if "cursor=page-2" in path:
return {"items": second_page, "next_cursor": None}
return {"items": first_page, "next_cursor": "page-2"}
mod.request_json = fake_request
result = mod.list_kanban_cards({"blocked": True, "limit": 10})
assert [item["full_path"] for item in result["items"]] == ["/kanban/doing/blocked-card"]
assert result["truncated"] is False
assert len(calls) == 2
def test_kanban_workflow_reads_pending_and_pick_next(mod):
backlog_urgent = node(
"/kanban/backlog/later-urgent",
"backlog",
metadata={"column": "backlog", "priority": "urgent", "rank": "00001024", "blocked": False},
tags=["kanban"],
node_type="kanban_card",
)
todo_high = node(
"/kanban/todo/next-high",
"todo",
metadata={"column": "todo", "priority": "high", "rank": "00002048", "blocked": False},
tags=["kanban"],
node_type="kanban_card",
)
todo_blocked = node(
"/kanban/todo/blocked",
"todo",
metadata={"column": "todo", "priority": "urgent", "rank": "00001024", "blocked": True},
tags=["kanban"],
node_type="kanban_card",
)
def fake_request(method, path, payload=None, **kwargs):
assert method == "GET"
return {"items": [backlog_urgent, todo_high, todo_blocked], "next_cursor": None}
mod.request_json = fake_request
pending = mod.get_pending_kanban_cards({})
assert [item["full_path"] for item in pending["items"]] == ["/kanban/todo/next-high", "/kanban/backlog/later-urgent"]
picked = mod.pick_next_kanban_card({})
assert picked["selected"]["full_path"] == "/kanban/todo/next-high"
assert picked["selection"]["blocked_excluded"] is True
def test_kanban_workflow_create_move_and_block(mod):
writes = []
existing = node(
"/kanban/todo/add-mcp-commands",
"todo",
metadata={"column": "todo", "priority": "normal", "rank": "00001024", "blocked": False, "title": "Add MCP commands"},
tags=["kanban"],
node_type="kanban_card",
)
mod.list_kanban_cards = lambda args: {"items": []}
mod.request_project_write = lambda method, path, payload, *args, **kwargs: writes.append((method, path, payload)) or {
**existing,
"id": "id-add-mcp-commands",
"full_path": f"{payload.get('folder_path', '/kanban/todo')}/{payload.get('name', 'add-mcp-commands')}".replace("//", "/"),
"folder_path": payload.get("folder_path", "/kanban/todo"),
"name": payload.get("name", "add-mcp-commands"),
"metadata": payload.get("metadata", existing["metadata"]),
"content_text": payload.get("content_text", existing["content_text"]),
"tags": payload.get("tags", existing["tags"]),
}
mod.get_node_by_id = lambda args: {**existing, "id": "id-add-mcp-commands"}
mod.index_node = lambda node: None
mod.remove_indexed_node = lambda node_id: None
created = mod.create_kanban_card({"title": "Add MCP commands", "priority": "HIGH"})
assert created["created"] is True
assert "node" not in created
assert writes[0][0] == "POST"
assert writes[0][2]["folder_path"] == "/kanban/todo"
assert writes[0][2]["metadata"]["priority"] == "high"
assert writes[0][2]["metadata"]["rank"] == "00001024"
moved = mod.move_kanban_card({"node_id": "id-add-mcp-commands"}, "doing")
assert moved["to"] == "/kanban/doing/add-mcp-commands"
assert "node" not in moved
assert writes[1][0] == "PATCH"
assert writes[1][2]["folder_path"] == "/kanban/doing"
assert writes[1][2]["metadata"]["column"] == "doing"
assert writes[1][2]["metadata"]["lifecycle_state"] == "active"
blocked = mod.block_kanban_card({"node_id": "id-add-mcp-commands", "blocked_reason": "waiting"}, True)
assert blocked["updated"] is True
assert "node" not in blocked
assert writes[2][2]["metadata"]["blocked"] is True
assert writes[2][2]["metadata"]["blocked_by"] == "waiting"
with pytest.raises(RuntimeError, match="path is an index/route only"):
mod.move_kanban_card({"path": "/kanban/todo/add-mcp-commands"}, "doing")
def test_kanban_create_deduplicates_overlapping_titles(mod):
writes = []
existing_cards = [
node("/kanban/todo/add-mcp-commands", "todo", metadata={"column": "todo"}, tags=["kanban"], node_type="kanban_card"),
node("/kanban/todo/add-mcp-commands-20260510123456", "todo", metadata={"column": "todo"}, tags=["kanban"], node_type="kanban_card"),
node("/kanban/doing/add-mcp-commands", "doing", metadata={"column": "doing"}, tags=["kanban"], node_type="kanban_card"),
]
mod.list_kanban_cards = lambda args: {"items": existing_cards if args.get("column") == "todo" else []}
mod.request_project_write = lambda method, path, payload, *args, **kwargs: writes.append((method, path, payload)) or {
**existing_cards[0],
"id": "id-deduped",
"full_path": f"{payload.get('folder_path')}/{payload.get('name')}",
"folder_path": payload.get("folder_path"),
"name": payload.get("name"),
"metadata": payload.get("metadata", {}),
"tags": payload.get("tags", []),
}
mod.index_node = lambda node: None
created = mod.create_kanban_card({"title": "Add MCP commands", "column": "todo", "now": "2026-05-10T12:34:56Z"})
assert created["path"] == "/kanban/todo/add-mcp-commands-20260510123456-2"
assert "node" not in created
assert writes[0][2]["name"] == "add-mcp-commands-20260510123456-2"
def test_lineage_status_reports_generic_plan_and_card_coverage(mod):
linked_plan = node("/plans/linked", "planned", links=[{"path": "/decisions/source", "relation": "implements"}])
unlinked_plan = node("/plans/unlinked", "planned")
linked_card = node(
"/kanban/todo/linked",
"todo",
metadata={"column": "todo"},
tags=["kanban"],
node_type="kanban_card",
links=[{"path": "/plans/linked", "relation": "derived-from"}],
)
unlinked_card = node(
"/kanban/todo/unlinked",
"todo",
metadata={"column": "todo"},
tags=["kanban"],
node_type="kanban_card",
)
def fake_list(args):
if args.get("folder_path") == "/plans":
return {"items": [linked_plan, unlinked_plan]}
if args.get("q") == "/kanban":
return {"items": [linked_card, unlinked_card, node("/notes/not-kanban", "active")]}
return {"items": []}
mod.list_project_nodes = fake_list
result = mod.lineage_status({})
assert result["status"] == "warnings"
assert result["counts"]["plans"] == 2
assert result["counts"]["plans_with_source_record"] == 1
assert result["counts"]["kanban_cards"] == 2
assert result["counts"]["kanban_cards_with_origin_plan"] == 1
assert [warning["type"] for warning in result["warnings"]] == [
"plan_missing_source_record",
"kanban_card_missing_origin_plan",
]
def test_plan_to_kanban_creates_card_with_origin_plan_link(mod):
writes = []
plan = node(
"/plans/generic-work",
"planned",
metadata={"title": "Generic Work", "priority": "high", "owner": "codex"},
tags=["plan"],
content="Implement a generic record to plan to kanban flow.",
)
mod.get_node_by_id = lambda args: plan
mod.list_kanban_cards = lambda args: {"items": []}
mod.request_project_write = lambda method, path, payload, *args, **kwargs: writes.append((method, path, payload)) or {
"id": "id-generic-work",
"project_id": "project-1",
"full_path": f"{payload.get('folder_path')}/{payload.get('name')}",
"folder_path": payload.get("folder_path"),
"name": payload.get("name"),
"type": payload.get("type"),
"metadata": payload.get("metadata", {}),
"tags": payload.get("tags", []),
"links": payload.get("links", []),
"content_text": payload.get("content_text", ""),
"updated_at": "2026-05-06T12:00:00Z",
}
mod.index_node = lambda node: None
created = mod.plan_to_kanban_card({"plan_node_id": plan["id"], "column": "todo"})
assert created["path"] == "/kanban/todo/generic-work"
assert created["source_plan_path"] == "/plans/generic-work"
assert writes[0][0] == "POST"
assert writes[0][2]["links"] == [{"path": "/plans/generic-work", "relation": "derived-from"}]
assert writes[0][2]["metadata"]["source_plan_path"] == "/plans/generic-work"
assert writes[0][2]["metadata"]["priority"] == "high"
assert writes[0][2]["metadata"]["owner"] == "codex"
def test_plan_lifecycle_dry_run_and_write(mod):
plan = node("/plans/example", "in_progress", tags=["plan"])
plan["id"] = "plan-123"
outcome = node("/implementation-notes/example", "completed", node_type="implementation-note")
outcome["id"] = "outcome-123"
writes = []
def fake_get_by_id(args):
if args["node_id"] == plan["id"]:
return plan
if args["node_id"] == outcome["id"]:
return outcome
raise AssertionError(f"unexpected node id {args['node_id']}")
mod.get_node_by_id = fake_get_by_id
mod.update_node_by_path = lambda args: writes.append(("update", args)) or {"node": {**plan, **args}}
mod.archive_node_by_path = lambda args: writes.append(("archive", args)) or {"archived": True}
dry = mod.complete_plan({"plan_node_id": plan["id"], "outcome_node_id": outcome["id"], "dry_run": True})
assert dry["changed"] is True
assert dry["dry_run"] is True
assert dry["status"] == "completed"
assert "archive node" in dry["planned_changes"]
assert writes == []
written = mod.complete_plan({"plan_node_id": plan["id"], "outcome_node_id": outcome["id"], "reason": "implemented"})
assert written["changed"] is True
assert written["archived"] is True
assert writes[0][0] == "update"
assert writes[0][1]["metadata"]["status"] == "completed"
assert writes[0][1]["metadata"]["outcome_path"] == "/implementation-notes/example"
assert "completed" in writes[0][1]["tags"]
assert "archived" in writes[0][1]["tags"]
assert writes[0][1]["links"] == [{"path": "/implementation-notes/example", "relation": "implemented-by"}]
assert writes[1][0] == "archive"
assert writes[1][1]["node_id"] == plan["id"]
with pytest.raises(RuntimeError, match="path is an index/route only"):
mod.complete_plan({"path": "/plans/example", "outcome_path": "/implementation-notes/example"})
def test_plan_lifecycle_archived_idempotent(mod):
outcome = node("/implementation-notes/example", "completed", node_type="implementation-note")
outcome["id"] = "outcome-123"
archived = node(
"/plans/example",
"completed",
metadata={"outcome_path": "/implementation-notes/example"},
tags=["plan", "completed", "archived"],
archived_at="2026-05-06T12:01:00Z",
)
archived["id"] = "plan-123"
archived["links"] = [{"path": "/implementation-notes/example", "relation": "implemented-by"}]
def fake_get_by_id(args):
if args["node_id"] == outcome["id"]:
return outcome
raise RuntimeError("HTTP 404")
mod.get_node_by_id = fake_get_by_id
mod.list_project_nodes = lambda args: {"items": [archived]}
result = mod.complete_plan({"plan_node_id": archived["id"], "outcome_node_id": outcome["id"]})
assert result["ok"] is True
assert result["changed"] is False
assert result["archived"] is True
def test_tool_registry(mod):
names = {tool["name"] for tool in mod.raw_tools()}
for name in [
"nap_plan_list_active",
"nap_lineage_status",
"nap_kanban_list",
"nap_kanban_get_pending",
"nap_kanban_pick_next",
"nap_kanban_start",
"nap_kanban_complete",
"nap_plan_complete",
"nap_plan_supersede",
"nap_plan_cancel",
"nap_plan_to_kanban",
]:
assert name in names
assert mod.tool_contract_metadata("nap_plan_complete")["dry_run_supported"] is True
assert mod.tool_contract_metadata("nap_plan_list_active")["side_effect"] == "none"
assert mod.tool_contract_metadata("nap_lineage_status")["side_effect"] == "none"
assert mod.tool_contract_metadata("nap_kanban_get_pending")["side_effect"] == "none"
assert mod.tool_contract_metadata("nap_plan_to_kanban")["side_effect"] == "remote-write"
assert mod.tool_contract_metadata("nap_kanban_start")["side_effect"] == "remote-write"
def run():
mod = load_module()
test_active_plan_listing(mod)
mod = load_module()
test_node_discovery_allows_large_requested_limit(mod)
mod = load_module()
test_node_by_path_returns_canonical_identity_uri(mod)
mod = load_module()
test_node_by_id_returns_canonical_identity_uri(mod)
mod = load_module()
test_node_by_uri_resolves_canonical_uri(mod)
mod = load_module()
test_node_get_accepts_node_id_without_path(mod)
mod = load_module()
test_node_get_rejects_path_reads(mod)
mod = load_module()
test_tools_include_node_by_path_descriptor(mod)
mod = load_module()
test_recent_memory_defaults_to_paths(mod)
mod = load_module()
test_find_defaults_to_paths_and_requests_compact_rest_view(mod)
mod = load_module()
test_title_view_requests_summary_rest_view_without_content(mod)
mod = load_module()
test_kanban_list_grouping_and_filters(mod)
mod = load_module()
test_kanban_dependency_summary_filters_ready_and_blocking(mod)
mod = load_module()
test_kanban_list_reads_beyond_first_source_page(mod)
mod = load_module()
test_kanban_workflow_reads_pending_and_pick_next(mod)
mod = load_module()
test_kanban_workflow_create_move_and_block(mod)
mod = load_module()
test_kanban_create_deduplicates_overlapping_titles(mod)
mod = load_module()
test_lineage_status_reports_generic_plan_and_card_coverage(mod)
mod = load_module()
test_plan_to_kanban_creates_card_with_origin_plan_link(mod)
mod = load_module()
test_plan_lifecycle_dry_run_and_write(mod)
mod = load_module()
test_plan_lifecycle_archived_idempotent(mod)
mod = load_module()
test_tool_registry(mod)
print("ok: MCP memory ergonomics helpers smoke passed")
if __name__ == "__main__":
run()
Reference in New Issue View Git Blame Copy Permalink