feat(journey): shared backend for editing and deleting learned nodes

Map journey node ids back to SKILL.md or §-delimited memory chunks and
perform user-initiated edits/deletes. Skill deletes archive (curator-
restorable); memory deletes rewrite MEMORY.md/USER.md in place.
This commit is contained in:
Brooklyn Nicholson 2026-06-30 15:07:19 -05:00
parent 571092ee36
commit a0576560ed
2 changed files with 320 additions and 0 deletions

206
agent/learning_mutations.py Normal file
View file

@ -0,0 +1,206 @@
"""User-initiated edit/delete for journey nodes (learned skills + memories).
The journey graph (``agent.learning_graph``) gives every node a stable id:
- **skills** the skill name (e.g. ``"debugging-hermes-desktop"``)
- **memories** ``memory:<source>:<index>`` where ``source`` is ``memory``
(``MEMORY.md``) or ``profile`` (``USER.md``) and ``index`` is the node's
position in the combined card list (``MEMORY.md`` cards first, then
``USER.md``).
This module maps a node id back to its on-disk home and performs the mutation,
shared by the CLI (``hermes journey delete|edit``), the TUI ``/journey`` overlay
(gateway RPCs), and the desktop GUI (REST). Deleting a skill *archives* it
(recoverable via ``hermes curator restore``); deleting a memory rewrites its
file. Pure stdlib + existing skill/memory helpers.
"""
from __future__ import annotations
from pathlib import Path
from typing import Any
_MEMORY_DELIM = "\n§\n"
_MEMORY_FILES = {"memory": "MEMORY.md", "profile": "USER.md"}
def parse_node_kind(node_id: str) -> str:
return "memory" if node_id.startswith("memory:") else "skill"
def _memories_dir() -> Path:
from hermes_constants import get_hermes_home
return get_hermes_home() / "memories"
def _parse_memory_id(node_id: str) -> tuple[str, int]:
"""``memory:<source>:<index>`` → (source, global_index)."""
parts = node_id.split(":", 2)
if len(parts) != 3 or parts[0] != "memory" or parts[1] not in _MEMORY_FILES:
raise ValueError(f"bad memory node id: {node_id!r}")
try:
return parts[1], int(parts[2])
except ValueError as exc:
raise ValueError(f"bad memory node id: {node_id!r}") from exc
def _memory_local_index(source: str, global_index: int) -> int:
"""Global card index → position within the source's own file.
``_memory_cards`` emits all ``MEMORY.md`` cards before ``USER.md`` cards, so
a profile card's local index is its global index minus the memory count.
"""
from agent.learning_graph import _memory_cards
cards = _memory_cards()
if not 0 <= global_index < len(cards):
raise IndexError(f"memory index {global_index} out of range")
if cards[global_index].get("source") != source:
raise ValueError("memory node id is stale — refresh the graph")
if source == "memory":
return global_index
return global_index - sum(1 for c in cards if c.get("source") == "memory")
def _read_chunks(path: Path) -> list[str]:
"""Raw ``§``-delimited chunks, preserving formatting; empties dropped to
match ``_memory_cards`` indexing."""
text = path.read_text(encoding="utf-8")
return [c for c in text.split(_MEMORY_DELIM) if c.strip()]
def _locate_memory(source: str, gidx: int) -> tuple[Path, list[str], int]:
"""Resolve a memory card to its file, all chunks, and local index."""
path = _memories_dir() / _MEMORY_FILES[source]
if not path.exists():
raise ValueError(f"{path.name} not found")
chunks = _read_chunks(path)
local = _memory_local_index(source, gidx)
if not 0 <= local < len(chunks):
raise ValueError("memory node id is stale — refresh the graph")
return path, chunks, local
# ── Inspect (edit prefill) ──────────────────────────────────────────────────
def node_detail(node_id: str) -> dict[str, Any]:
"""Current content for an edit prefill. ``content`` is the full SKILL.md
(skills) or the raw memory chunk (memories)."""
try:
return _node_detail(node_id)
except (ValueError, IndexError) as exc:
return {"ok": False, "message": str(exc)}
def _node_detail(node_id: str) -> dict[str, Any]:
if parse_node_kind(node_id) == "memory":
source, gidx = _parse_memory_id(node_id)
_, chunks, local = _locate_memory(source, gidx)
body = chunks[local].strip()
return {"ok": True, "kind": "memory", "id": node_id, "label": body.splitlines()[0][:80], "content": body}
from tools.skill_manager_tool import _find_skill
found = _find_skill(node_id)
if not found:
return {"ok": False, "message": f"skill '{node_id}' not found"}
skill_md = Path(found["path"]) / "SKILL.md"
if not skill_md.exists():
return {"ok": False, "message": f"SKILL.md missing for '{node_id}'"}
return {
"ok": True,
"kind": "skill",
"id": node_id,
"label": node_id,
"content": skill_md.read_text(encoding="utf-8"),
}
# ── Delete ──────────────────────────────────────────────────────────────────
def delete_node(node_id: str) -> dict[str, Any]:
try:
return _delete_memory(node_id) if parse_node_kind(node_id) == "memory" else _delete_skill(node_id)
except (ValueError, IndexError) as exc:
return {"ok": False, "message": str(exc)}
def _delete_skill(name: str) -> dict[str, Any]:
from tools import skill_usage
if skill_usage.get_record(name).get("pinned"):
return {"ok": False, "message": f"'{name}' is pinned — unpin it first (hermes curator unpin {name})"}
ok, message = skill_usage.archive_skill(name)
if ok:
_clear_skill_cache()
return {"ok": ok, "message": f"archived '{name}' — restore with: hermes curator restore {name}" if ok else message}
def _delete_memory(node_id: str) -> dict[str, Any]:
source, gidx = _parse_memory_id(node_id)
path, chunks, local = _locate_memory(source, gidx)
del chunks[local]
_write_chunks(path, chunks)
return {"ok": True, "message": f"deleted memory from {path.name}"}
# ── Edit ────────────────────────────────────────────────────────────────────
def edit_node(node_id: str, content: str) -> dict[str, Any]:
try:
return _edit_memory(node_id, content) if parse_node_kind(node_id) == "memory" else _edit_skill(node_id, content)
except (ValueError, IndexError) as exc:
return {"ok": False, "message": str(exc)}
def _edit_skill(name: str, content: str) -> dict[str, Any]:
from tools.skill_manager_tool import _edit_skill as _do_edit
result = _do_edit(name, content)
if result.get("success"):
_clear_skill_cache()
return {"ok": True, "message": f"updated '{name}'"}
return {"ok": False, "message": result.get("error", "edit failed")}
def _edit_memory(node_id: str, content: str) -> dict[str, Any]:
source, gidx = _parse_memory_id(node_id)
body = content.strip()
if not body:
return {"ok": False, "message": "empty memory — use delete to remove it"}
path, chunks, local = _locate_memory(source, gidx)
chunks[local] = body
_write_chunks(path, chunks)
return {"ok": True, "message": f"updated memory in {path.name}"}
# ── Helpers ─────────────────────────────────────────────────────────────────
def _write_chunks(path: Path, chunks: list[str]) -> None:
body = _MEMORY_DELIM.join(c.strip() for c in chunks)
path.write_text(f"{body}\n" if body else "", encoding="utf-8")
def _clear_skill_cache() -> None:
try:
from agent.prompt_builder import clear_skills_system_prompt_cache
clear_skills_system_prompt_cache(clear_snapshot=True)
except Exception:
pass

View file

@ -0,0 +1,114 @@
"""Behavior contracts for journey node edit/delete (agent.learning_mutations).
Exercises the real on-disk resolution (skills dir + MEMORY.md/USER.md chunking)
against a temp HERMES_HOME, never mocks the idfile mapping is the whole point.
"""
from __future__ import annotations
import pytest
from agent import learning_mutations as lm
from hermes_constants import get_hermes_home
_SKILL = """---
name: my-skill
description: A test skill.
---
# My Skill
Body.
"""
@pytest.fixture
def home():
base = get_hermes_home()
(base / "memories").mkdir(parents=True, exist_ok=True)
(base / "memories" / "MEMORY.md").write_text("alpha note\nline two\n§\nbeta note", encoding="utf-8")
(base / "memories" / "USER.md").write_text("user profile note", encoding="utf-8")
skill = base / "skills" / "my-skill"
skill.mkdir(parents=True, exist_ok=True)
(skill / "SKILL.md").write_text(_SKILL, encoding="utf-8")
return base
def test_parse_node_kind():
assert lm.parse_node_kind("memory:memory:0") == "memory"
assert lm.parse_node_kind("memory:profile:3") == "memory"
assert lm.parse_node_kind("debugging-hermes") == "skill"
def test_memory_global_index_maps_across_files(home):
# MEMORY.md → indices 0,1; USER.md → index 2 (global, memory cards first).
assert lm.node_detail("memory:memory:0")["content"].startswith("alpha note")
assert lm.node_detail("memory:memory:1")["content"] == "beta note"
assert lm.node_detail("memory:profile:2")["content"] == "user profile note"
def test_memory_label_is_first_line(home):
assert lm.node_detail("memory:memory:0")["label"] == "alpha note"
def test_delete_memory_rewrites_file(home):
assert lm.delete_node("memory:memory:0")["ok"]
remaining = (home / "memories" / "MEMORY.md").read_text(encoding="utf-8")
assert "alpha note" not in remaining
assert "beta note" in remaining
def test_edit_memory_replaces_chunk(home):
assert lm.edit_node("memory:profile:2", "rewritten profile")["ok"]
assert (home / "memories" / "USER.md").read_text(encoding="utf-8").strip() == "rewritten profile"
def test_edit_memory_empty_is_rejected(home):
res = lm.edit_node("memory:memory:1", " ")
assert not res["ok"]
assert "delete" in res["message"]
def test_stale_memory_index_errors(home):
res = lm.node_detail("memory:memory:9")
assert not res["ok"]
def test_bad_memory_id_returns_error(home):
res = lm.delete_node("memory:bogus:0")
assert not res["ok"]
def test_skill_detail_returns_skill_md(home):
d = lm.node_detail("my-skill")
assert d["ok"] and d["kind"] == "skill"
assert "name: my-skill" in d["content"]
def test_delete_skill_archives_recoverably(home):
res = lm.delete_node("my-skill")
assert res["ok"]
assert not (home / "skills" / "my-skill").exists()
assert (home / "skills" / ".archive" / "my-skill" / "SKILL.md").exists()
def test_delete_pinned_skill_refused(home):
from tools import skill_usage
skill_usage.set_pinned("my-skill", True)
res = lm.delete_node("my-skill")
assert not res["ok"]
assert "pinned" in res["message"]
assert (home / "skills" / "my-skill").exists()
def test_edit_skill_rewrites_and_validates(home):
bad = lm.edit_node("my-skill", "no frontmatter here")
assert not bad["ok"]
good = lm.edit_node("my-skill", _SKILL.replace("A test skill.", "Updated desc."))
assert good["ok"]
assert "Updated desc." in (home / "skills" / "my-skill" / "SKILL.md").read_text(encoding="utf-8")
def test_missing_skill_detail(home):
assert not lm.node_detail("nonexistent-skill")["ok"]