Type node creation and filter local memory by effective type

- nap_create_node accepts `type`; when omitted (also in nap_bulk and
  agent-node creates) the type is inferred from the folder and sent
  explicitly, so older APIs also store typed plans and decisions.
  Explicit spelling variants are canonicalized; custom types pass as is.
- Local browse, full-text and encrypted-shard searches filter on the
  effective type (canonical spelling, or the folder's type for generic
  notes; `type=note` excludes notes whose folder implies a type); server
  rows are post-filtered the same way. Rows expose `effective_type` when
  it differs from the stored type; discover offers an effective_type
  facet.
- Active-only filters also exclude deployed, released,
  released_verified_production and folded.
Mirrors napseer src/vocab.rs; decision
/decisions/2026-09-29-node-type-status-vocabulary.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
HumbleTurtle
2026-09-29 00:32:36 -03:00
co-authored by Claude Opus 5.5
parent 611a4571f3
commit f201cb358b
4 changed files with 624 additions and 22 deletions
+397
View File
@@ -0,0 +1,397 @@
"""Effective node types and release-terminal statuses in the CLI wrapper.
Decision /decisions/2026-09-29-node-type-status-vocabulary: creates infer a
type from the folder when the caller omits one, type filters match effective
types (a generic `note` resolves from its folder), and active-only lists also
exclude release-style terminal statuses. Older APIs do none of this, so the
wrapper must behave the same on its own. Writes go through the real request
path to a fake transport; discovery runs against a disposable HTTP provider
and the real local index.
"""
import copy
import importlib.util
import json
import pathlib
import sys
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlsplit
import pytest
SCRIPT = pathlib.Path(__file__).resolve().parents[1] / "resources" / "scripts" / "napseer_mcp_server.py"
def load_module(name):
sys.path.insert(0, str(SCRIPT.parent))
spec = importlib.util.spec_from_file_location(name, SCRIPT)
module = importlib.util.module_from_spec(spec)
assert spec and spec.loader
spec.loader.exec_module(module)
return module
# -- Writes -----------------------------------------------------------------
class FakeResponse:
def __init__(self, payload):
self._body = json.dumps(payload).encode("utf-8")
def read(self, *_args):
return self._body
def __enter__(self):
return self
def __exit__(self, *_exc):
return False
@pytest.fixture
def writer(monkeypatch, tmp_path):
monkeypatch.setenv("NAPSEER_TOOL_PROFILES", "all")
mod = load_module("napseer_effective_types_write_test")
mod.DEFAULT_PROJECT_ID = "project-1"
mod.TOKEN = "synthetic-access"
mod.TOKEN_EXPIRES_AT = "2099-01-01T00:00:00Z"
mod.AUTH = {**mod.AUTH, "account_mode": "operator_account", "project_id": "project-1"}
mod.CONFIGURED_TOOL_PROFILES = {"all"}
mod.refresh_public_auth_state = lambda: mod.AUTH
mod.project_memory_encryption_active = lambda _project_id: False
mod.AUTH_DIR = tmp_path / "state"
mod.INDEX_PATH = tmp_path / "state" / "index.sqlite"
mod.INDEX_LOCK_PATH = tmp_path / "state" / "index.lock"
mod.resolve_project_id = lambda _args: "project-1"
mod.acquire_project_lock = lambda args=None, **kwargs: {"id": "lock-1", "lease_token": "lease-1"}
mod.cleanup_project_lock = lambda project_id, lock: None
sent = []
def fake_urlopen(request, timeout=None):
body = json.loads(request.data.decode("utf-8")) if request.data else None
sent.append((request.get_method(), urlsplit(request.full_url).path, body))
folder = (body or {}).get("folder_path") or "/"
name = (body or {}).get("name") or "node"
return FakeResponse({
"id": f"node-{len(sent)}", "project_id": "project-1",
"full_path": f"{folder.rstrip('/')}/{name}", "folder_path": folder, "name": name,
# The fake server applies the historical default; the payload is what is asserted.
"type": (body or {}).get("type") or "note",
"content_text": (body or {}).get("content_text", ""), "metadata": {},
"updated_at": "2026-09-29T12:00:00Z",
})
mod.api_urlopen = fake_urlopen
return mod, sent
def created_payload(sent):
posts = [body for method, path, body in sent if method == "POST" and path.endswith("/nodes")]
assert len(posts) == 1, sent
return posts[0]
@pytest.mark.parametrize(
("folder_path", "expected"),
[
("/plans", "plan"),
("/plans/2026", "plan"),
("/decisions", "decision"),
("/product/decisions", "decision"),
("/implementation-notes", "implementation_note"),
("/deployments", "deployment"),
("/runbooks", "runbook"),
("/rules", "rule"),
("/indexes", "index"),
("/analysis", "analysis"),
("/product/analysis", "analysis"),
("/documentation/security", "documentation"),
("/kanban/doing", "kanban_card"),
# The server normalizes folder segments before storing; inference agrees.
("/Implementation Notes/", "implementation_note"),
],
)
def test_create_without_type_sends_the_folder_type(writer, folder_path, expected):
mod, sent = writer
mod.call_tool_impl("nap_create_node", {"folder_path": folder_path, "name": "record", "content_text": "body"})
assert created_payload(sent)["type"] == expected
@pytest.mark.parametrize("folder_path", ["/notes", "/plansx", "/product", "/services/plans"])
def test_create_without_type_outside_typed_folders_leaves_the_server_default(writer, folder_path):
mod, sent = writer
mod.call_tool_impl("nap_create_node", {"folder_path": folder_path, "name": "record", "content_text": "body"})
assert "type" not in created_payload(sent)
@pytest.mark.parametrize(
("given", "expected"),
[
("implementation-note", "implementation_note"),
("Kanban Card", "kanban_card"),
("product-spec", "product_spec"),
("decision", "decision"),
# An explicit generic note stays a note even inside a typed folder.
("note", "note"),
# Unknown custom types are kept exactly as given.
("security-documentation", "security-documentation"),
("Custom Type", "Custom Type"),
],
)
def test_create_with_explicit_type_only_canonicalizes_known_spellings(writer, given, expected):
mod, sent = writer
mod.call_tool_impl(
"nap_create_node",
{"folder_path": "/plans", "name": "record", "content_text": "body", "type": given},
)
assert created_payload(sent)["type"] == expected
def test_create_node_schema_advertises_optional_type(writer):
mod, _sent = writer
tool = next(tool for tool in mod.tools() if tool["name"] == "nap_create_node")
schema = tool["inputSchema"]
assert schema["properties"]["type"]["type"] == "string"
assert "type" not in schema.get("required", [])
def test_bulk_creates_infer_types_per_item(writer):
mod, sent = writer
mod.try_get_node_by_path = lambda path, allow_agent=False: None
mod.call_tool_impl("nap_bulk", {"nodes": [
{"path": "/decisions/choose-store", "content_text": "a"},
{"path": "/scratch/idea", "content_text": "b"},
{"path": "/runbooks/restore", "content_text": "c", "type": "product-spec"},
{"path": "/plans/custom", "content_text": "d", "type": "Custom Type"},
]})
posts = [body for method, path, body in sent if method == "POST" and path.endswith("/nodes")]
assert [body.get("type") for body in posts] == ["decision", None, "product_spec", "Custom Type"]
assert "type" not in posts[1]
def test_bulk_update_without_type_does_not_retype_the_stored_node(writer):
mod, sent = writer
existing = {"id": "node-9", "project_id": "project-1", "full_path": "/plans/legacy",
"folder_path": "/plans", "name": "legacy", "type": "note", "content_text": "old"}
mod.get_node_by_id = lambda args: existing
mod.call_tool_impl("nap_bulk", {"nodes": [{"node_id": "node-9", "content_text": "new"}]})
patches = [body for method, path, body in sent if method == "PATCH"]
assert patches and "type" not in patches[0]
# -- Local discovery ---------------------------------------------------------
def stored_node(folder, name, node_type, *, minute, status=None, content=None):
stamp = f"2026-09-01T12:{minute:02d}:00Z"
return {
"id": name, "project_id": "project-1", "full_path": f"{folder}/{name}",
"folder_path": folder, "name": name, "type": node_type,
"metadata": {"status": status} if status else {}, "tags": [], "aliases": [], "links": [],
"content_text": content or f"{name} vocabularyterm", "updated_at": stamp, "archived_at": None,
}
@pytest.fixture
def local(tmp_path, monkeypatch):
nodes = {}
events = []
class Handler(BaseHTTPRequestHandler):
def log_message(self, *_args):
pass
def reply(self, status, payload):
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps(payload).encode())
def do_GET(self):
url = urlsplit(self.path)
query = parse_qs(url.query)
if url.path.endswith("/changes"):
after = query.get("after", ["head"])[0]
start = len(events) if after == "head" else int(after[1:])
return self.reply(200, {"items": [], "next_cursor": f"c{start}", "has_more": False})
if url.path.endswith("/nodes"):
rows = sorted((copy.deepcopy(n) for n in nodes.values()),
key=lambda n: (n["updated_at"], n["id"]), reverse=True)
return self.reply(200, {"items": rows, "next_cursor": None})
node_id = url.path.rsplit("/", 1)[-1]
if node_id not in nodes:
return self.reply(404, {"error": {"code": "not_found", "message": "missing"}})
return self.reply(200, copy.deepcopy(nodes[node_id]))
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
monkeypatch.setenv("NAPSEER_TOOL_PROFILES", "all")
mod = load_module("napseer_effective_types_local_test")
mod.AUTH_DIR = tmp_path / "project-state"
mod.INDEX_PATH = mod.AUTH_DIR / "index.sqlite"
mod.INDEX_LOCK_PATH = mod.AUTH_DIR / "index.lock"
mod.BASE_URL = f"http://127.0.0.1:{server.server_port}"
mod.TOKEN = "disposable-test-token"
mod.AUTH = {"account_id": "test-account"}
mod.DEFAULT_PROJECT_ID = "project-1"
mod.CONFIGURED_TOOL_PROFILES = {"all"}
mod.refresh_public_auth_state = lambda: None
mod.require_unlocked = lambda _name: None
mod.resolve_project_id = lambda args: args.get("project_id", "project-1")
mod.project_encryption_state = lambda _project_id: "standard"
def seed(*items):
for item in items:
nodes[item["id"]] = copy.deepcopy(item)
events.append(item["id"])
assert mod.reindex_project({})["graph_complete"] is True
try:
yield mod, seed
finally:
server.shutdown()
server.server_close()
thread.join(timeout=2)
TYPED_FIXTURE = [
stored_node("/plans", "legacy-plan", "note", minute=1),
stored_node("/plans/2026", "nested-plan", "note", minute=2),
stored_node("/plans", "typed-plan", "plan", minute=3),
stored_node("/notes", "scratch", "note", minute=4),
stored_node("/implementation-notes", "dashed-note", "implementation-note", minute=5),
stored_node("/implementation-notes", "generic-note", "note", minute=6),
stored_node("/services", "misplaced-plan", "Plan", minute=7),
stored_node("/services", "custom", "security-documentation", minute=8),
# A differently spelled generic note still resolves from its folder.
stored_node("/decisions", "capital-note", "Note", minute=9),
]
def discover(mod, **args):
return mod.call_tool_impl("nap_discover", {"mode": "local", "limit": 50, **args})
@pytest.mark.parametrize("q", [None, "vocabularyterm"])
def test_local_type_filter_matches_effective_types(local, q):
mod, seed = local
seed(*TYPED_FIXTURE)
extra = {"q": q} if q else {}
plans = discover(mod, type="plan", **extra)
by_id = {item["node_id"]: item for item in plans["items"]}
assert set(by_id) == {"legacy-plan", "nested-plan", "typed-plan", "misplaced-plan"}
# The stored type is reported unchanged; effective_type is added when it differs.
assert (by_id["legacy-plan"]["type"], by_id["legacy-plan"]["effective_type"]) == ("note", "plan")
assert (by_id["nested-plan"]["type"], by_id["nested-plan"]["effective_type"]) == ("note", "plan")
assert (by_id["misplaced-plan"]["type"], by_id["misplaced-plan"]["effective_type"]) == ("Plan", "plan")
assert by_id["typed-plan"]["type"] == "plan"
assert "effective_type" not in by_id["typed-plan"]
notes = discover(mod, type="implementation_note", **extra)
by_id = {item["node_id"]: item for item in notes["items"]}
assert set(by_id) == {"dashed-note", "generic-note"}
assert by_id["dashed-note"]["type"] == "implementation-note"
assert {item["effective_type"] for item in notes["items"]} == {"implementation_note"}
# A spelling variant in the filter canonicalizes the same way.
dashed = discover(mod, type="implementation-note", **extra)
assert {item["node_id"] for item in dashed["items"]} == {"dashed-note", "generic-note"}
decisions = discover(mod, type="decision", **extra)
assert [item["node_id"] for item in decisions["items"]] == ["capital-note"]
assert (decisions["items"][0]["type"], decisions["items"][0]["effective_type"]) == ("Note", "decision")
custom = discover(mod, type="security_documentation", **extra)
assert [item["node_id"] for item in custom["items"]] == ["custom"]
@pytest.mark.parametrize("q", [None, "vocabularyterm"])
@pytest.mark.parametrize("requested", ["note", "Note", " NOTE "])
def test_local_note_filter_is_effective_type_only(local, q, requested):
mod, seed = local
seed(*TYPED_FIXTURE)
extra = {"q": q} if q else {}
# Only the note under /notes is a note; notes under /plans, /decisions and
# /implementation-notes take their folder's type and are excluded.
notes = discover(mod, type=requested, **extra)
assert [item["node_id"] for item in notes["items"]] == ["scratch"]
assert "effective_type" not in notes["items"][0]
def test_unfiltered_local_rows_carry_effective_type_and_facets(local):
mod, seed = local
seed(*TYPED_FIXTURE)
result = discover(mod, view="paths", facets=["type", "effective_type"])
by_id = {item["node_id"]: item for item in result["items"]}
assert by_id["legacy-plan"]["type"] == "note"
assert by_id["legacy-plan"]["effective_type"] == "plan"
assert "effective_type" not in by_id["scratch"]
assert by_id["custom"]["effective_type"] == "security_documentation"
counts = result["facets"]["counts"]
# The stored-type facet keeps its historical meaning.
assert {"value": "note", "count": 4} in counts["type"]
assert {"value": "plan", "count": 4} in counts["effective_type"]
assert {"value": "implementation_note", "count": 2} in counts["effective_type"]
RELEASE_TERMINAL = ["deployed", "released", "released_verified_production", "folded"]
@pytest.mark.parametrize("q", [None, "vocabularyterm"])
def test_local_active_only_excludes_release_terminal_statuses(local, q):
mod, seed = local
seed(
stored_node("/deployments", "live", "note", minute=1, status="active"),
*[
stored_node("/deployments", f"release-{status}", "note", minute=2 + index, status=status)
for index, status in enumerate(RELEASE_TERMINAL)
],
stored_node("/plans", "finished", "note", minute=9, status="completed"),
)
extra = {"q": q} if q else {}
active = discover(mod, active_only=True, **extra)
assert [item["node_id"] for item in active["items"]] == ["live"]
# Without active_only the release records remain discoverable.
everything = discover(mod, **extra)
assert {f"release-{status}" for status in RELEASE_TERMINAL} <= {
item["node_id"] for item in everything["items"]
}
def test_server_rows_are_post_filtered_on_effective_type(local):
mod, _seed = local
# An older server matches stored types only, so type=note returns every
# stored note; the wrapper keeps just the effective notes.
rows = [
{"id": "under-plans", "full_path": "/plans/legacy", "folder_path": "/plans", "type": "note"},
{"id": "under-decisions", "full_path": "/decisions/choice", "type": "Note"},
{"id": "under-notes", "full_path": "/notes/scratch", "folder_path": "/notes", "type": "note"},
]
requests = []
def fake_request(method, path, payload=None, **kwargs):
requests.append(path)
return {"items": copy.deepcopy(rows), "next_cursor": None}
mod.request_json = fake_request
notes = mod.list_project_nodes({"type": "note", "view": "paths"})
assert "type=note" in requests[-1]
assert [item["node_id"] for item in notes["items"]] == ["under-notes"]
rows[:] = [{"id": "typed", "full_path": "/plans/typed", "type": "plan"},
{"id": "legacy", "full_path": "/plans/legacy", "type": "note"}]
plans = mod.list_project_nodes({"type": "plan", "view": "paths"})
assert [item["node_id"] for item in plans["items"]] == ["typed", "legacy"]
def test_active_only_filter_on_server_rows_excludes_release_statuses(local):
mod, _seed = local
rows = [{"id": status, "metadata": {"status": status}} for status in [*RELEASE_TERMINAL, "active", "done"]]
filters = mod.status_filter_from_args({"active_only": True})
# "done" keeps its existing CLI handling (already excluded).
assert [row["id"] for row in mod.apply_status_filters(rows, filters)] == ["active"]
explicit = mod.status_filter_from_args({"status": "deployed"})
assert explicit["warnings"] == []
+20 -4
View File
@@ -52,18 +52,18 @@ def unlock(mod, secret=None):
cache_secret(mod, secret or memory_secret())
def plain_node(name, minute, content=None):
def plain_node(name, minute, content=None, folder="/notes"):
stamp = f"2026-09-01T12:{minute:02d}:00Z"
return {
"id": name, "project_id": PROJECT, "full_path": f"/notes/{name}", "folder_path": "/notes",
"id": name, "project_id": PROJECT, "full_path": f"{folder}/{name}", "folder_path": folder,
"name": name, "type": "note", "metadata": {}, "tags": [], "aliases": [], "links": [],
"content_text": content if content is not None else name, "updated_at": stamp, "archived_at": None,
}
def e2e_node(mod, name, minute, content, metadata=None, secret=None, archived=False):
def e2e_node(mod, name, minute, content, metadata=None, secret=None, archived=False, folder="/notes"):
secret = secret or memory_secret()
node = plain_node(name, minute)
node = plain_node(name, minute, folder=folder)
context = {"full_path": node["full_path"]}
node.update({
"content_text": "", "metadata": {}, "encryption_state": "encrypted",
@@ -229,6 +229,22 @@ def test_e2e_body_term_is_findable_with_snippet_while_unlocked(env):
assert not discovered.get("warnings")
def test_e2e_shard_hits_filter_on_effective_type(env):
mod, provider, _ = env
provider.put(e2e_node(mod, "legacyplan", 1, BODY, folder="/plans"), "node_created")
provider.put(e2e_node(mod, "scratchnote", 2, BODY), "node_created")
assert mod.reindex_project({})["e2e_index"]["status"] == "applied"
# Bodies are only in the shard, so these hits come from the E2E path.
plans = mod.search_local_index({"q": BODY_TERM, "limit": 10, "type": "plan"})
assert hit_ids(plans) == ["legacyplan"]
assert plans["items"][0]["search_source"] == "local_e2e"
assert (plans["items"][0]["type"], plans["items"][0]["effective_type"]) == ("note", "plan")
notes = mod.search_local_index({"q": BODY_TERM, "limit": 10, "type": "note"})
assert hit_ids(notes) == ["scratchnote"]
assert notes["items"][0]["search_source"] == "local_e2e"
def test_locked_secret_yields_no_e2e_hits_and_an_explicit_warning(env):
mod, provider, _ = env
seed(mod, provider)
+2
View File
@@ -305,6 +305,8 @@ def run():
"node_id": "done",
"full_path": "/plans/done",
"type": "note",
# A generic note under /plans reports its effective type as well.
"effective_type": "plan",
"status": "completed",
}]