From ddab2dccc8a3016a624c456d8e175b0046fc7a3a Mon Sep 17 00:00:00 2001 From: HumbleTurtle Date: Fri, 22 May 2026 22:47:30 -0300 Subject: [PATCH] Require node identity for mutation tools --- resources/scripts/napseer_mcp_server.py | 78 +++++++++++++-------- tests/test_mcp_memory_ergonomics_helpers.py | 42 +++++++++++ tests/test_mcp_memory_node_encryption.py | 7 +- 3 files changed, 95 insertions(+), 32 deletions(-) diff --git a/resources/scripts/napseer_mcp_server.py b/resources/scripts/napseer_mcp_server.py index 6e03e11..81b82a6 100755 --- a/resources/scripts/napseer_mcp_server.py +++ b/resources/scripts/napseer_mcp_server.py @@ -7662,8 +7662,9 @@ def node_by_path(args): def guarded_node_patch(args): - reject_agent_namespace_path(args["path"]) project_id = resolve_project_id(args) + existing = resolve_existing_node_ref(args, view="edit") + node_id = existing["id"] precondition = args.get("precondition") or {} if not args.get("dry_run"): if not precondition.get("revision") or not precondition.get("read_fingerprint"): @@ -7690,28 +7691,28 @@ def guarded_node_patch(args): ]: if key in args: payload[key] = args[key] - path = normalize_node_path(args["path"]) result = request_project_write( "PATCH", - f"/v1/projects/{project_id}/nodes/by-path?{rest_query_string({'path': path})}", + f"/v1/projects/{project_id}/nodes/{urllib.parse.quote(node_id, safe='')}", payload, project_id, - f"guarded patch node {path}", - "path", - path, + f"guarded patch node {existing.get('full_path') or node_id}", + "node", + node_id, ) node = ((result.get("data") or {}).get("node") or {}) if isinstance(result, dict) else {} if node: index_node(node) if not isinstance(result, dict): - return {"ok": True, "changed": True, "path": path} + return {"ok": True, "changed": True, "id": node_id, "path": existing.get("full_path")} data = result.get("data") if isinstance(result.get("data"), dict) else {} compact = { "ok": bool(result.get("ok", True)), "status": result.get("status") or ("dry_run" if args.get("dry_run") else "updated"), "changed": not bool(data.get("dry_run")), "dry_run": bool(data.get("dry_run")), - "path": node.get("full_path") or path, + "id": node.get("id") or node_id, + "path": node.get("full_path") or existing.get("full_path"), } if data.get("revision"): compact["revision"] = data["revision"] @@ -7734,11 +7735,13 @@ def try_get_node_by_path(path, allow_agent=False): def node_payload_from_args(args, existing=None): - parts = path_parts(args["path"]) payload = {} - if not existing or "path" in args: + if "path" in args: + parts = path_parts(args["path"]) payload["folder_path"] = parts["folder_path"] payload["name"] = parts["name"] + elif not existing: + raise RuntimeError("path is required when creating a node") for source, target in [ ("type", "type"), ("content_text", "content_text"), @@ -7764,6 +7767,22 @@ def node_payload_from_args(args, existing=None): return payload +def resolve_existing_node_ref(args, view="render"): + if args.get("path"): + if args.get("_internal_path_ref"): + return read_node_by_path({"path": args["path"]}) + raise RuntimeError("path is an index/route only; resolve it with nap_node_by_path, then use node_id or uri") + if args.get("uri"): + resolved = node_by_uri({"uri": args["uri"], "view": view})["node"] + elif args.get("node_id") or args.get("id"): + resolved = get_node_by_id({"node_id": args.get("node_id") or args.get("id"), "view": view}) + else: + raise RuntimeError("one of node_id or uri is required") + if is_agent_namespace_path(resolved.get("full_path") or ""): + reject_agent_namespace_path(resolved.get("full_path") or "") + return resolved + + def upsert_node(args): reject_agent_namespace_path(args["path"]) project_id = resolve_project_id(args) @@ -7797,17 +7816,16 @@ def upsert_node(args): def update_node_by_path(args): - reject_agent_namespace_path(args["path"]) project_id = resolve_project_id(args) - existing = read_node_by_path({"path": args["path"]}) + existing = resolve_existing_node_ref(args) updated = request_project_write( "PATCH", f"/v1/projects/{project_id}/nodes/{existing['id']}", encrypt_node_payload_for_write(project_id, node_payload_from_args(args, existing=existing), existing=existing), project_id, f"update node {existing['full_path']}", - "path", - existing["full_path"], + "node", + existing["id"], ) index_node(updated) node = decrypt_node_for_return(updated, project_id=project_id) @@ -7815,9 +7833,8 @@ def update_node_by_path(args): def archive_node_by_path(args): - reject_agent_namespace_path(args["path"]) project_id = resolve_project_id(args) - existing = read_node_by_path({"path": args["path"]}) + existing = resolve_existing_node_ref(args) if args.get("dry_run"): return {"ok": True, "changed": False, "dry_run": True, "archived": False, "paths": [existing["full_path"]]} request_project_write( @@ -7826,8 +7843,8 @@ def archive_node_by_path(args): None, project_id, f"archive node {existing['full_path']}", - "path", - existing["full_path"], + "node", + existing["id"], ) remove_indexed_node(existing["id"]) return {"ok": True, "changed": True, "dry_run": False, "archived": True, "paths": [existing["full_path"]], "id": existing["id"]} @@ -8805,7 +8822,7 @@ def update_kanban_card(args): } if "content_text" in args or "body" in args: payload["content_text"] = args.get("content_text") if "content_text" in args else args.get("body") - updated = update_node_by_path(payload) + updated = update_node_by_path({**payload, "_internal_path_ref": True}) return {"ok": True, "changed": True, "updated": True, "path": updated.get("path") or existing["full_path"], "column": kanban_column(existing)} @@ -9036,12 +9053,12 @@ def plan_lifecycle_update(args, *, action, terminal_status, required_target_key, updated = node if planned_changes != ["archive node"]: - update_node_by_path({"path": path, "metadata": metadata, "tags": tags, "links": links}) + update_node_by_path({"path": path, "metadata": metadata, "tags": tags, "links": links, "_internal_path_ref": True}) updated = dict(node) updated["metadata"] = metadata updated["tags"] = tags updated["links"] = links - archive_result = archive_node_by_path({"path": path}) + archive_result = archive_node_by_path({"path": path, "_internal_path_ref": True}) return plan_lifecycle_result( action, path, @@ -11292,12 +11309,12 @@ def raw_tools(): }, { "name": "nap_patch", - "description": "Update an existing node by full path using an auto-acquired project lock. Fails if the node does not exist.", + "description": "Update an existing node by canonical node_id or nap:// URI using an auto-acquired node lock. Fails if the node does not exist.", "inputSchema": { "type": "object", - "required": ["path"], "properties": { - "path": {"type": "string"}, + "node_id": {"type": "string", "description": "Canonical node UUID."}, + "uri": {"type": "string", "description": "Canonical nap://project-name/project-id/node-id URI."}, "content_text": {"type": "string"}, "type": {"type": "string"}, "tags": {"type": "array", "items": {"type": "string"}}, @@ -11364,9 +11381,9 @@ def raw_tools(): "description": "Fingerprint-guarded object-first node patch. Non-dry-run calls require precondition.revision and precondition.read_fingerprint from nap_node_get.", "inputSchema": { "type": "object", - "required": ["path"], "properties": { - "path": {"type": "string"}, + "node_id": {"type": "string", "description": "Canonical node UUID."}, + "uri": {"type": "string", "description": "Canonical nap://project-name/project-id/node-id URI."}, "mode": {"type": "string", "enum": ["object_patch", "replace_editable_fields"], "default": "object_patch"}, "precondition": { "type": "object", @@ -11393,11 +11410,14 @@ def raw_tools(): }, { "name": "nap_rm", - "description": "Archive an existing node by full path using an auto-acquired project lock.", + "description": "Archive an existing node by canonical node_id or nap:// URI using an auto-acquired node lock.", "inputSchema": { "type": "object", - "required": ["path"], - "properties": {"path": {"type": "string"}, "dry_run": {"type": "boolean", "default": False}}, + "properties": { + "node_id": {"type": "string", "description": "Canonical node UUID."}, + "uri": {"type": "string", "description": "Canonical nap://project-name/project-id/node-id URI."}, + "dry_run": {"type": "boolean", "default": False}, + }, "additionalProperties": False, }, }, diff --git a/tests/test_mcp_memory_ergonomics_helpers.py b/tests/test_mcp_memory_ergonomics_helpers.py index 5a7a758..0c85248 100644 --- a/tests/test_mcp_memory_ergonomics_helpers.py +++ b/tests/test_mcp_memory_ergonomics_helpers.py @@ -272,6 +272,48 @@ def test_tools_include_node_by_path_descriptor(mod): if view: assert "full" not in view["enum"] + for name in ["nap_patch", "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"] + + +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" + + 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_discovery_rejects_full_content_view(mod): with pytest.raises(RuntimeError, match="full node bodies require nap_node_get"): diff --git a/tests/test_mcp_memory_node_encryption.py b/tests/test_mcp_memory_node_encryption.py index 185efbb..0935109 100644 --- a/tests/test_mcp_memory_node_encryption.py +++ b/tests/test_mcp_memory_node_encryption.py @@ -185,8 +185,9 @@ def run(): else: raise AssertionError("encrypted projects must fail closed when memory secret fetch fails") try: + mod.get_node_by_id = lambda args: encrypted_node(mod, full_path="/notes/alpha") mod.guarded_node_patch({ - "path": "/notes/alpha", + "node_id": "node-1", "precondition": {"revision": "r1", "read_fingerprint": "fp1"}, "content_op": {"op": "replace_all", "content_text": "plaintext"}, }) @@ -363,9 +364,9 @@ def run(): assert tee_body["encrypted_content_envelope"]["aad_subject"].startswith("encryption_id:") existing = encrypted_node(mod, full_path="/notes/beta", content="old") - mod.get_node_by_path = lambda args, allow_agent=False: dict(existing) + mod.get_node_by_id = lambda args: dict(existing) writes.clear() - patched = mod.update_node_by_path({"path": "/notes/beta", "content_text": "patch secret"}) + patched = mod.update_node_by_path({"node_id": "node-1", "content_text": "patch secret"}) patch_body = writes[-1][2] assert patched["ok"] is True assert patched["updated"] is True