napseer/nap-cli
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Require identity for plan lifecycle tools

Browse Source
This commit is contained in:
HumbleTurtle
2026-05-22 23:23:57 -03:00
parent 65448aee0b
commit 1813e9f4e3
2 changed files with 123 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
resources/scripts/napseer_mcp_server.py
+72 -41
View File
@@ -8797,10 +8797,11 @@ def lineage_status(args):
def plan_to_kanban_card(args):
args = dict(args or {})
plan_path = normalize_node_path(args.get("plan_path") or args.get("path") or "")
if not is_plan_node_path(plan_path):
raise RuntimeError("plan_path must be a /plans/<plan-slug> path")
plan = read_node_by_path({"path": plan_path})
if args.get("plan_path") or args.get("path"):
raise RuntimeError("plan_path/path are index/routes only; resolve with nap_node_by_path, then use plan_node_id or plan_uri")
plan = resolve_existing_node_ref({"node_id": args.get("plan_node_id"), "uri": args.get("plan_uri")})
if not is_plan_node_path(plan.get("full_path") or ""):
raise RuntimeError("plan node must be under /plans")
metadata = node_metadata(plan)
title = str(args.get("title") or metadata.get("title") or plan.get("name") or "").strip()
if not title:
@@ -8927,20 +8928,54 @@ def find_archived_node_by_path(path):
return None
def get_plan_for_lifecycle(path):
normalized = normalize_node_path(path)
if not normalized.startswith("/plans/"):
raise RuntimeError("plan lifecycle helpers only operate on /plans/* nodes")
def find_archived_node_by_id(node_id):
result = list_project_nodes({"archived_only": True, "limit": 500, "view": "metadata"})
for node in result.get("items", []):
if str(node.get("id") or "") == str(node_id):
return node
return None
def node_ref_id_from_args(args):
if args.get("node_id") or args.get("id"):
return str(args.get("node_id") or args.get("id")).strip()
if args.get("plan_node_id"):
return str(args.get("plan_node_id")).strip()
if args.get("uri") or args.get("plan_uri"):
return parse_nap_node_uri(args.get("uri") or args.get("plan_uri"))["node_id"]
return ""
def get_plan_for_lifecycle(args):
if args.get("path") or args.get("plan_path"):
raise RuntimeError("path is an index/route only; resolve it with nap_node_by_path, then use plan_node_id or plan_uri")
node_id = node_ref_id_from_args(args)
try:
node = read_node_by_path({"path": normalized})
return node, False
node = resolve_existing_node_ref({"node_id": args.get("plan_node_id") or args.get("node_id") or args.get("id"), "uri": args.get("plan_uri") or args.get("uri")})
archived = False
except RuntimeError as exc:
if "HTTP 404" not in str(exc) and "node not found" not in str(exc):
if not node_id or ("HTTP 404" not in str(exc) and "node not found" not in str(exc)):
raise
archived = find_archived_node_by_path(normalized)
if archived:
return archived, True
node = find_archived_node_by_id(node_id)
if not node:
raise
archived = True
if not is_plan_node_path(node.get("full_path") or ""):
raise RuntimeError("plan lifecycle helpers only operate on /plans/* nodes")
return node, archived
def resolve_lifecycle_target(args, required_target_key):
if not required_target_key:
return None
if args.get(required_target_key):
raise RuntimeError(f"{required_target_key} is an index/route only; use {required_target_key.replace('_path', '_node_id')} or {required_target_key.replace('_path', '_uri')}")
node_key = required_target_key.replace("_path", "_node_id")
uri_key = required_target_key.replace("_path", "_uri")
if not args.get(node_key) and not args.get(uri_key):
raise RuntimeError(f"{node_key} or {uri_key} is required")
target = resolve_existing_node_ref({"node_id": args.get(node_key), "uri": args.get(uri_key)})
return normalize_node_path(target["full_path"])
def plan_lifecycle_matches(node, *, status, target_path=None, metadata_key=None, link_relation=None):
@@ -9013,17 +9048,13 @@ def cancel_plan(args):
def plan_lifecycle_update(args, *, action, terminal_status, required_target_key, metadata_target_key, timestamp_key, reason_key, link_relation, terminal_tags, require_reason=False):
path = normalize_node_path(args.get("path") or "")
if not path or path == "/":
raise RuntimeError("path is required")
target_path = normalize_node_path(args.get(required_target_key)) if required_target_key else None
if required_target_key and not args.get(required_target_key):
raise RuntimeError(f"{required_target_key} is required")
target_path = resolve_lifecycle_target(args, required_target_key)
reason = str(args.get("reason") or args.get("notes") or "").strip()
if require_reason and not reason:
raise RuntimeError("reason is required")
dry_run = as_bool(args.get("dry_run"), False)
node, already_archived = get_plan_for_lifecycle(path)
node, already_archived = get_plan_for_lifecycle(args)
path = node["full_path"]
already_terminal = plan_lifecycle_matches(
node,
@@ -9034,10 +9065,6 @@ def plan_lifecycle_update(args, *, action, terminal_status, required_target_key,
)
if already_terminal and already_archived:
return plan_lifecycle_result(action, path, node, True, False, dry_run)
if target_path and not already_terminal:
if try_get_node_by_path(target_path, allow_agent=is_agent_namespace_path(target_path)) is None:
raise RuntimeError(f"target node not found: {target_path}")
metadata = dict(node_metadata(node))
links = node.get("links") if isinstance(node.get("links"), list) else []
tags = list(node.get("tags") or [])
@@ -9088,12 +9115,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, "_internal_path_ref": True})
update_node_by_path({"node_id": node["id"], "metadata": metadata, "tags": tags, "links": links})
updated = dict(node)
updated["metadata"] = metadata
updated["tags"] = tags
updated["links"] = links
archive_result = archive_node_by_path({"path": path, "_internal_path_ref": True})
archive_result = archive_node_by_path({"node_id": node["id"]})
return plan_lifecycle_result(
action,
path,
@@ -11473,13 +11500,14 @@ def raw_tools():
},
{
"name": "nap_plan_complete",
"description": "Mark a /plans node completed, link it to an outcome, add terminal tags, and archive it so active plan discovery excludes it.",
"description": "Mark a /plans node completed by plan node identity, link it to an outcome node identity, add terminal tags, and archive it so active plan discovery excludes it.",
"inputSchema": {
"type": "object",
"required": ["path", "outcome_path"],
"properties": {
"path": {"type": "string"},
"outcome_path": {"type": "string"},
"plan_node_id": {"type": "string", "description": "Canonical node id of the /plans node."},
"plan_uri": {"type": "string", "description": "Canonical nap:// URI of the /plans node."},
"outcome_node_id": {"type": "string", "description": "Canonical node id of the outcome node."},
"outcome_uri": {"type": "string", "description": "Canonical nap:// URI of the outcome node."},
"reason": {"type": "string"},
"notes": {"type": "string"},
"dry_run": {"type": "boolean", "default": False},
@@ -11490,13 +11518,15 @@ def raw_tools():
},
{
"name": "nap_plan_supersede",
"description": "Mark a /plans node superseded by another plan/node, add terminal tags and link, then archive it.",
"description": "Mark a /plans node superseded by another node identity, add terminal tags and link, then archive it.",
"inputSchema": {
"type": "object",
"required": ["path", "replacement_path", "reason"],
"required": ["reason"],
"properties": {
"path": {"type": "string"},
"replacement_path": {"type": "string"},
"plan_node_id": {"type": "string", "description": "Canonical node id of the /plans node."},
"plan_uri": {"type": "string", "description": "Canonical nap:// URI of the /plans node."},
"replacement_node_id": {"type": "string", "description": "Canonical node id of the replacement node."},
"replacement_uri": {"type": "string", "description": "Canonical nap:// URI of the replacement node."},
"reason": {"type": "string"},
"notes": {"type": "string"},
"dry_run": {"type": "boolean", "default": False},
@@ -11507,12 +11537,13 @@ def raw_tools():
},
{
"name": "nap_plan_cancel",
"description": "Mark a /plans node cancelled with a reason, add terminal tags, and archive it.",
"description": "Mark a /plans node cancelled by plan node identity with a reason, add terminal tags, and archive it.",
"inputSchema": {
"type": "object",
"required": ["path", "reason"],
"required": ["reason"],
"properties": {
"path": {"type": "string"},
"plan_node_id": {"type": "string", "description": "Canonical node id of the /plans node."},
"plan_uri": {"type": "string", "description": "Canonical nap:// URI of the /plans node."},
"reason": {"type": "string"},
"notes": {"type": "string"},
"dry_run": {"type": "boolean", "default": False},
@@ -11523,12 +11554,12 @@ def raw_tools():
},
{
"name": "nap_plan_to_kanban",
"description": "Create a node-backed kanban card from any /plans node and preserve the origin plan with a derived-from link.",
"description": "Create a node-backed kanban card from a /plans node identity and preserve the origin plan with a derived-from link.",
"inputSchema": {
"type": "object",
"required": ["plan_path"],
"properties": {
"plan_path": {"type": "string"},
"plan_node_id": {"type": "string", "description": "Canonical node id of the /plans node."},
"plan_uri": {"type": "string", "description": "Canonical nap:// URI of the /plans node."},
"title": {"type": "string"},
"slug": {"type": "string"},
"column": {"type": "string", "enum": DEFAULT_KANBAN_COLUMNS, "default": "todo"},
tests/test_mcp_memory_ergonomics_helpers.py
+50 -12
View File
@@ -323,6 +323,31 @@ def test_tools_include_node_by_path_descriptor(mod):
assert "uri" in backlinks_schema["properties"]
assert "path" not in backlinks_schema["properties"]
complete_schema = next(item for item in mod.tools() if item["name"] == "nap_plan_complete")["inputSchema"]
assert "plan_node_id" in complete_schema["properties"]
assert "plan_uri" in complete_schema["properties"]
assert "outcome_node_id" in complete_schema["properties"]
assert "outcome_uri" in complete_schema["properties"]
assert "path" not in complete_schema["properties"]
assert "outcome_path" not in complete_schema["properties"]
supersede_schema = next(item for item in mod.tools() if item["name"] == "nap_plan_supersede")["inputSchema"]
assert supersede_schema["required"] == ["reason"]
assert "plan_node_id" in supersede_schema["properties"]
assert "replacement_node_id" in supersede_schema["properties"]
assert "path" not in supersede_schema["properties"]
assert "replacement_path" not in supersede_schema["properties"]
cancel_schema = next(item for item in mod.tools() if item["name"] == "nap_plan_cancel")["inputSchema"]
assert cancel_schema["required"] == ["reason"]
assert "plan_node_id" in cancel_schema["properties"]
assert "path" not in cancel_schema["properties"]
plan_to_kanban_schema = next(item for item in mod.tools() if item["name"] == "nap_plan_to_kanban")["inputSchema"]
assert "plan_node_id" in plan_to_kanban_schema["properties"]
assert "plan_uri" in plan_to_kanban_schema["properties"]
assert "plan_path" not in plan_to_kanban_schema["properties"]
def test_mutation_tools_reject_path_and_patch_by_node_id(mod):
found = node("/notes/current", "active", node_type="note")
@@ -920,7 +945,7 @@ def test_plan_to_kanban_creates_card_with_origin_plan_link(mod):
content="Implement a generic record to plan to kanban flow.",
)
mod.get_node_by_path = lambda args, allow_agent=False: plan
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",
@@ -937,7 +962,7 @@ def test_plan_to_kanban_creates_card_with_origin_plan_link(mod):
}
mod.index_node = lambda node: None
created = mod.plan_to_kanban_card({"plan_path": "/plans/generic-work", "column": "todo"})
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"
@@ -949,29 +974,30 @@ def test_plan_to_kanban_creates_card_with_origin_plan_link(mod):
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(args, allow_agent=False):
if args["path"] == "/plans/example":
def fake_get_by_id(args):
if args["node_id"] == plan["id"]:
return plan
if args["path"] == "/implementation-notes/example":
if args["node_id"] == outcome["id"]:
return outcome
raise RuntimeError("HTTP 404")
raise AssertionError(f"unexpected node id {args['node_id']}")
mod.get_node_by_path = fake_get
mod.try_get_node_by_path = lambda path, allow_agent=False: outcome if path == "/implementation-notes/example" else None
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({"path": "/plans/example", "outcome_path": "/implementation-notes/example", "dry_run": 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({"path": "/plans/example", "outcome_path": "/implementation-notes/example", "reason": "implemented"})
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"
@@ -981,9 +1007,15 @@ def test_plan_lifecycle_dry_run_and_write(mod):
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",
@@ -991,11 +1023,17 @@ def test_plan_lifecycle_archived_idempotent(mod):
tags=["plan", "completed", "archived"],
archived_at="2026-05-06T12:01:00Z",
)
archived["id"] = "plan-123"
archived["links"] = [{"path": "/implementation-notes/example", "relation": "implemented-by"}]
mod.get_node_by_path = lambda args, allow_agent=False: (_ for _ in ()).throw(RuntimeError("HTTP 404"))
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({"path": "/plans/example", "outcome_path": "/implementation-notes/example"})
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