HumbleTurtleandClaude Opus 5.5
0938b3026e
- 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>
1383 lines
56 KiB
Python
1383 lines
56 KiB
Python
#!/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()
|