Require node identity for kanban mutations

This commit is contained in:
HumbleTurtle
2026-05-22 23:05:32 -03:00
parent 58162f3f59
commit 411a7b73e1
2 changed files with 60 additions and 27 deletions
+40 -24
View File
@@ -8549,6 +8549,17 @@ def kanban_card_by_path(path):
return node return node
def kanban_card_by_ref(args):
if args.get("path"):
if args.get("_internal_path_ref"):
return kanban_card_by_path(args["path"])
raise RuntimeError("path is an index/route only; resolve it with nap_node_by_path, then use node_id or uri")
node = resolve_existing_node_ref(args)
if not normalize_node_path(node.get("full_path") or "").startswith("/kanban/"):
raise RuntimeError("kanban command node_id/uri must reference a /kanban card")
return node
def kanban_target_path(node, column): def kanban_target_path(node, column):
column = normalize_kanban_column(column) column = normalize_kanban_column(column)
title_slug = normalize_kanban_slug(node.get("name") or kanban_title(node)) title_slug = normalize_kanban_slug(node.get("name") or kanban_title(node))
@@ -8805,10 +8816,7 @@ def plan_to_kanban_card(args):
def update_kanban_card(args): def update_kanban_card(args):
args = dict(args or {}) args = dict(args or {})
path = args.get("path") existing = kanban_card_by_ref(args)
if not path:
raise RuntimeError("path is required")
existing = kanban_card_by_path(path)
metadata = kanban_card_metadata(args, existing=existing, column=kanban_column(existing)) metadata = kanban_card_metadata(args, existing=existing, column=kanban_column(existing))
if "clear_blocked_by" in args and as_bool(args.get("clear_blocked_by"), False): if "clear_blocked_by" in args and as_bool(args.get("clear_blocked_by"), False):
metadata.pop("blocked_by", None) metadata.pop("blocked_by", None)
@@ -8822,17 +8830,25 @@ def update_kanban_card(args):
} }
if "content_text" in args or "body" in 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") payload["content_text"] = args.get("content_text") if "content_text" in args else args.get("body")
updated = update_node_by_path({**payload, "_internal_path_ref": True}) project_id = resolve_project_id(args)
updated_node = request_project_write(
"PATCH",
f"/v1/projects/{project_id}/nodes/{existing['id']}",
encrypt_node_payload_for_write(project_id, node_payload_from_args(payload, existing=existing), existing=existing),
project_id,
f"update kanban card {existing['full_path']}",
"node",
existing["id"],
)
index_node(updated_node)
updated = node_write_reference(decrypt_node_for_return(updated_node, project_id=project_id))
return {"ok": True, "changed": True, "updated": True, "path": updated.get("path") or existing["full_path"], "column": kanban_column(existing)} return {"ok": True, "changed": True, "updated": True, "path": updated.get("path") or existing["full_path"], "column": kanban_column(existing)}
def move_kanban_card(args, target_column): def move_kanban_card(args, target_column):
args = dict(args or {}) args = dict(args or {})
path = args.get("path")
if not path:
raise RuntimeError("path is required")
target_column = normalize_kanban_column(args.get("column") or target_column) target_column = normalize_kanban_column(args.get("column") or target_column)
existing = kanban_card_by_path(path) existing = kanban_card_by_ref(args)
source = existing["full_path"] source = existing["full_path"]
target = kanban_target_path(existing, target_column) target = kanban_target_path(existing, target_column)
metadata = kanban_card_metadata(args, existing=existing, column=target_column) metadata = kanban_card_metadata(args, existing=existing, column=target_column)
@@ -8845,8 +8861,8 @@ def move_kanban_card(args, target_column):
encrypt_node_payload_for_write(project_id, payload, existing=existing), encrypt_node_payload_for_write(project_id, payload, existing=existing),
project_id, project_id,
f"move kanban card {source} to {target}", f"move kanban card {source} to {target}",
"path", "node",
source, existing["id"],
) )
remove_indexed_node(existing["id"]) remove_indexed_node(existing["id"])
index_node(updated) index_node(updated)
@@ -10586,20 +10602,20 @@ def kanban_workflow_tool_schemas():
}, },
"additionalProperties": False, "additionalProperties": False,
} }
card_path_schema = { card_ref_schema = {
"type": "object", "type": "object",
"required": ["path"],
"properties": { "properties": {
"path": {"type": "string"}, "node_id": {"type": "string", "description": "Canonical node UUID for the Kanban card."},
"uri": {"type": "string", "description": "Canonical nap://project-name/project-id/node-id URI for the Kanban card."},
"rank": {"type": "string"}, "rank": {"type": "string"},
}, },
"additionalProperties": False, "additionalProperties": False,
} }
block_schema = { block_schema = {
"type": "object", "type": "object",
"required": ["path"],
"properties": { "properties": {
"path": {"type": "string"}, "node_id": {"type": "string", "description": "Canonical node UUID for the Kanban card."},
"uri": {"type": "string", "description": "Canonical nap://project-name/project-id/node-id URI for the Kanban card."},
"blocked_by": {"type": "string"}, "blocked_by": {"type": "string"},
"blocked_reason": {"type": "string"}, "blocked_reason": {"type": "string"},
}, },
@@ -10607,9 +10623,9 @@ def kanban_workflow_tool_schemas():
} }
update_schema = { update_schema = {
"type": "object", "type": "object",
"required": ["path"],
"properties": { "properties": {
"path": {"type": "string"}, "node_id": {"type": "string", "description": "Canonical node UUID for the Kanban card."},
"uri": {"type": "string", "description": "Canonical nap://project-name/project-id/node-id URI for the Kanban card."},
"title": {"type": "string"}, "title": {"type": "string"},
"content_text": {"type": "string"}, "content_text": {"type": "string"},
"body": {"type": "string"}, "body": {"type": "string"},
@@ -10661,13 +10677,13 @@ def kanban_workflow_tool_schemas():
"additionalProperties": False, "additionalProperties": False,
}, },
}, },
{"name": "nap_kanban_start", "description": "Move a Kanban card to working state, canonical /kanban/doing.", "inputSchema": card_path_schema}, {"name": "nap_kanban_start", "description": "Move a Kanban card to working state by node_id/uri, canonical /kanban/doing.", "inputSchema": card_ref_schema},
{"name": "nap_kanban_send_review", "description": "Move a Kanban card to /kanban/review.", "inputSchema": card_path_schema}, {"name": "nap_kanban_send_review", "description": "Move a Kanban card to /kanban/review by node_id/uri.", "inputSchema": card_ref_schema},
{"name": "nap_kanban_complete", "description": "Move a Kanban card to /kanban/done.", "inputSchema": card_path_schema}, {"name": "nap_kanban_complete", "description": "Move a Kanban card to /kanban/done by node_id/uri.", "inputSchema": card_ref_schema},
{"name": "nap_kanban_block", "description": "Mark a Kanban card blocked and optionally record a reason.", "inputSchema": block_schema}, {"name": "nap_kanban_block", "description": "Mark a Kanban card blocked and optionally record a reason.", "inputSchema": block_schema},
{"name": "nap_kanban_unblock", "description": "Clear a Kanban card blocked state and reason.", "inputSchema": {"type": "object", "required": ["path"], "properties": {"path": {"type": "string"}}, "additionalProperties": False}}, {"name": "nap_kanban_unblock", "description": "Clear a Kanban card blocked state and reason by node_id/uri.", "inputSchema": {"type": "object", "properties": {"node_id": {"type": "string"}, "uri": {"type": "string"}}, "additionalProperties": False}},
{"name": "nap_kanban_update", "description": "Update common Kanban card fields while preserving the node-backed model.", "inputSchema": update_schema}, {"name": "nap_kanban_update", "description": "Update common Kanban card fields by node_id/uri while preserving the node-backed model.", "inputSchema": update_schema},
{"name": "nap_kanban_archive", "description": "Archive a node-backed Kanban card.", "inputSchema": {"type": "object", "required": ["path"], "properties": {"path": {"type": "string"}, "dry_run": {"type": "boolean", "default": False}}, "additionalProperties": False}}, {"name": "nap_kanban_archive", "description": "Archive a node-backed Kanban card by node_id/uri.", "inputSchema": {"type": "object", "properties": {"node_id": {"type": "string"}, "uri": {"type": "string"}, "dry_run": {"type": "boolean", "default": False}}, "additionalProperties": False}},
] ]
+20 -3
View File
@@ -278,6 +278,20 @@ def test_tools_include_node_by_path_descriptor(mod):
assert "uri" in schema["properties"] assert "uri" in schema["properties"]
assert "path" not in schema["properties"] assert "path" not in schema["properties"]
for name in [
"nap_kanban_start",
"nap_kanban_send_review",
"nap_kanban_complete",
"nap_kanban_block",
"nap_kanban_unblock",
"nap_kanban_update",
"nap_kanban_archive",
]:
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): def test_mutation_tools_reject_path_and_patch_by_node_id(mod):
found = node("/notes/current", "active", node_type="note") found = node("/notes/current", "active", node_type="note")
@@ -670,7 +684,7 @@ def test_kanban_workflow_create_move_and_block(mod):
"content_text": payload.get("content_text", existing["content_text"]), "content_text": payload.get("content_text", existing["content_text"]),
"tags": payload.get("tags", existing["tags"]), "tags": payload.get("tags", existing["tags"]),
} }
mod.get_node_by_path = lambda args, allow_agent=False: existing mod.get_node_by_id = lambda args: {**existing, "id": "id-add-mcp-commands"}
mod.index_node = lambda node: None mod.index_node = lambda node: None
mod.remove_indexed_node = lambda node_id: None mod.remove_indexed_node = lambda node_id: None
@@ -682,7 +696,7 @@ def test_kanban_workflow_create_move_and_block(mod):
assert writes[0][2]["metadata"]["priority"] == "high" assert writes[0][2]["metadata"]["priority"] == "high"
assert writes[0][2]["metadata"]["rank"] == "00001024" assert writes[0][2]["metadata"]["rank"] == "00001024"
moved = mod.move_kanban_card({"path": "/kanban/todo/add-mcp-commands"}, "doing") moved = mod.move_kanban_card({"node_id": "id-add-mcp-commands"}, "doing")
assert moved["to"] == "/kanban/doing/add-mcp-commands" assert moved["to"] == "/kanban/doing/add-mcp-commands"
assert "node" not in moved assert "node" not in moved
assert writes[1][0] == "PATCH" assert writes[1][0] == "PATCH"
@@ -690,12 +704,15 @@ def test_kanban_workflow_create_move_and_block(mod):
assert writes[1][2]["metadata"]["column"] == "doing" assert writes[1][2]["metadata"]["column"] == "doing"
assert writes[1][2]["metadata"]["lifecycle_state"] == "active" assert writes[1][2]["metadata"]["lifecycle_state"] == "active"
blocked = mod.block_kanban_card({"path": "/kanban/todo/add-mcp-commands", "blocked_reason": "waiting"}, True) blocked = mod.block_kanban_card({"node_id": "id-add-mcp-commands", "blocked_reason": "waiting"}, True)
assert blocked["updated"] is True assert blocked["updated"] is True
assert "node" not in blocked assert "node" not in blocked
assert writes[2][2]["metadata"]["blocked"] is True assert writes[2][2]["metadata"]["blocked"] is True
assert writes[2][2]["metadata"]["blocked_by"] == "waiting" 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): def test_kanban_create_deduplicates_overlapping_titles(mod):
writes = [] writes = []