harness

diff --git hermes-agent/tests/gateway/test_methods_harness_imports.py harness/tests/gateway/test_methods_harness_imports.py new file mode 100644 index 0000000000000000000000000000000000000000..df58051420b7d203cb43234a304a0f64bb5cb1d5 --- /dev/null +++ harness/tests/gateway/test_methods_harness_imports.py @@ -0,0 +1,136 @@ +"""Every name the split handler modules reference must actually resolve. + +Regression guard for the class of bug that killed every wiki page: the +upstream rebase's handler split carried wiki handlers into +``methods_harness.py`` without their ``wiki_api`` imports. Python resolves +function-body names at CALL time, so the module imported cleanly and nothing +failed until the first ``wiki.scan`` — which died with ``NameError: name +'resolve_wiki' is not defined`` and the desktop showed "Failed to load page" +on every wiki page. + +This test finds every bare name loaded inside a function body of each split +module and asserts it resolves against the module's own globals, its inline +(function-local) imports, Python builtins, or ``tui_gateway.server``'s +globals (handler bodies are rebound onto server's namespace at install time +— see ``method_ctx.py`` — so server globals are legitimately reachable). +A name none of those provide is exactly the wiki bug waiting for its first +caller. +""" + +import ast +import builtins +from pathlib import Path + +import pytest + +_GATEWAY_DIR = Path(__file__).resolve().parents[2] / "tui_gateway" +_SPLIT_MODULES = sorted(_GATEWAY_DIR.glob("methods_*.py")) + + +def _module_scope_names(tree: ast.Module) -> set[str]: + """Names bound at module scope: imports, assignments, defs, classes.""" + names: set[str] = set() + for node in tree.body: + if isinstance(node, ast.Import): + for alias in node.names: + names.add((alias.asname or alias.name).split(".")[0]) + elif isinstance(node, ast.ImportFrom): + for alias in node.names: + names.add(alias.asname or alias.name) + elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): + names.add(node.name) + elif isinstance(node, ast.Assign): + for target in node.targets: + for n in ast.walk(target): + if isinstance(n, ast.Name): + names.add(n.id) + elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name): + names.add(node.target.id) + elif isinstance(node, ast.Try): + # try/except ImportError fallback blocks bind in both arms. + for sub in ast.walk(node): + if isinstance(sub, (ast.Import, ast.ImportFrom)): + for alias in sub.names: + names.add((alias.asname or alias.name).split(".")[0]) + elif isinstance(sub, ast.Assign): + for target in sub.targets: + for n in ast.walk(target): + if isinstance(n, ast.Name): + names.add(n.id) + return names + + +def _function_unresolved_names(func: ast.AST, module_names: set[str]) -> set[str]: + """Bare Name loads in ``func`` that neither local bindings nor + ``module_names`` nor builtins provide.""" + bound: set[str] = set() + loads: list[str] = [] + for node in ast.walk(func): + if isinstance(node, ast.Lambda): + args = node.args + for a in ( + list(args.posonlyargs) + list(args.args) + list(args.kwonlyargs) + ): + bound.add(a.arg) + if args.vararg: + bound.add(args.vararg.arg) + if args.kwarg: + bound.add(args.kwarg.arg) + elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + bound.add(node.name) + args = node.args + for a in ( + list(args.posonlyargs) + list(args.args) + list(args.kwonlyargs) + ): + bound.add(a.arg) + if args.vararg: + bound.add(args.vararg.arg) + if args.kwarg: + bound.add(args.kwarg.arg) + elif isinstance(node, (ast.Import, ast.ImportFrom)): + for alias in node.names: + bound.add((alias.asname or alias.name).split(".")[0]) + elif isinstance(node, ast.Name): + if isinstance(node.ctx, (ast.Store, ast.Del)): + bound.add(node.id) + else: + loads.append(node.id) + elif isinstance(node, ast.ExceptHandler) and node.name: + bound.add(node.name) + elif isinstance(node, (ast.comprehension,)): + for n in ast.walk(node.target): + if isinstance(n, ast.Name): + bound.add(n.id) + elif isinstance(node, ast.ClassDef): + bound.add(node.name) + elif isinstance(node, (ast.Global, ast.Nonlocal)): + bound.update(node.names) + builtin_names = set(dir(builtins)) | {"__name__", "__file__", "__doc__"} + return { + n for n in loads + if n not in bound and n not in module_names and n not in builtin_names + } + + +def _server_globals() -> set[str]: + """Names bound at module scope in tui_gateway/server.py — the namespace + handler bodies are rebound onto at install time.""" + tree = ast.parse((_GATEWAY_DIR / "server.py").read_text()) + return _module_scope_names(tree) + + +@pytest.mark.parametrize("module_path", _SPLIT_MODULES, ids=lambda p: p.name) +def test_handler_names_resolve(module_path: Path) -> None: + tree = ast.parse(module_path.read_text()) + reachable = _module_scope_names(tree) | _server_globals() + unresolved: dict[str, set[str]] = {} + for node in tree.body: + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + missing = _function_unresolved_names(node, reachable) + if missing: + unresolved[f"{node.name}:{node.lineno}"] = missing + assert not unresolved, ( + f"{module_path.name} references names that resolve nowhere — these are " + f"NameErrors waiting for their first caller (the wiki.scan bug): " + f"{unresolved}" + )

diff --git hermes-agent/tests/test_tui_gateway_server.py harness/tests/test_tui_gateway_server.py index 43341730059f8398cff9ad9a950aa6abe8f80c71..23789e10f8150d2ec823c508e2e05f1132f6d09a 100644 --- hermes-agent/tests/test_tui_gateway_server.py +++ harness/tests/test_tui_gateway_server.py @@ -324,7 +324,7 @@ monkeypatch.setattr(server, "_wait_agent_for_prompt", lambda _session, _rid, _sid: None) monkeypatch.setattr( server, "_run_prompt_submit", - lambda rid, sid, _session, text: inline_calls.append((rid, sid, text)), + lambda rid, sid, _session, text, chat_mode=False: inline_calls.append((rid, sid, text)), ) monkeypatch.setattr(server.threading, "Thread", _ImmediateThread)   @@ -2064,12 +2064,14 @@ # _get_platform_tools (a non-configurable platform toolset in hermes-cli's # universe); `project` is GUI-only, folded in by _load_enabled_toolsets. # Toolsets inside their first release (_RECENTLY_SHIPPED_TOOLSETS) are # back-filled onto saved lists that never offered them — allow those too. + # `learning` joined the core tool list (toolsets.py) and is recovered onto + # a saved list the same way `kanban` is. from hermes_cli.tools_config import _RECENTLY_SHIPPED_TOOLSETS   result = server._load_enabled_toolsets() assert result is not None - assert {"kanban", "memory", "project"} <= set(result) - assert set(result) - {"kanban", "memory", "project"} <= _RECENTLY_SHIPPED_TOOLSETS + assert {"kanban", "memory", "project", "learning"} <= set(result) + assert set(result) - {"kanban", "memory", "project", "learning"} <= _RECENTLY_SHIPPED_TOOLSETS err = capsys.readouterr().err assert "ignoring disabled MCP servers" in err assert "mcp-off" in err @@ -2094,8 +2096,8 @@ from hermes_cli.tools_config import _RECENTLY_SHIPPED_TOOLSETS   result = server._load_enabled_toolsets() assert result is not None - assert {"kanban", "memory", "project"} <= set(result) - assert set(result) - {"kanban", "memory", "project"} <= _RECENTLY_SHIPPED_TOOLSETS + assert {"kanban", "memory", "project", "learning"} <= set(result) + assert set(result) - {"kanban", "memory", "project", "learning"} <= _RECENTLY_SHIPPED_TOOLSETS assert "using configured CLI toolsets" in capsys.readouterr().err   @@ -5070,6 +5072,75 @@ def run_conversation(self, prompt, conversation_history=None, stream_callback=None, **_kwargs): self._turns.append(prompt) return {"final_response": "", "messages": []} + + +class _NoToolsCapturingAgent(_RecordingAgent): + """Agent whose ``run_conversation`` declares ``no_tools`` explicitly — like + the real ``AIAgent`` — so the server's signature-gated forwarding matches + and we can observe the value that reaches the turn.""" + + def __init__(self, turns): + super().__init__(turns) + self.observed_no_tools = "unset" + + def run_conversation( + self, prompt, conversation_history=None, stream_callback=None, no_tools=False, **_kwargs + ): + self.observed_no_tools = no_tools + return super().run_conversation(prompt) + + +def test_run_prompt_submit_chat_mode_runs_the_turn_tool_less(monkeypatch, tmp_path): + """``chat_mode=True`` forwards ``no_tools=True`` into the agent turn.""" + _configure_immediate_prompt_run(monkeypatch, tmp_path) + agent = _NoToolsCapturingAgent([]) + session = _session(session_key="s-chat", agent=agent, running=True) + server._sessions["sid-chat"] = session + try: + server._run_prompt_submit("rid", "sid-chat", session, "hey", chat_mode=True) + assert agent.observed_no_tools is True + finally: + server._sessions.pop("sid-chat", None) + + +def test_run_prompt_submit_default_keeps_tools(monkeypatch, tmp_path): + """Without chat mode the turn runs the normal tool-enabled agent.""" + _configure_immediate_prompt_run(monkeypatch, tmp_path) + agent = _NoToolsCapturingAgent([]) + session = _session(session_key="s-normal", agent=agent, running=True) + server._sessions["sid-normal"] = session + try: + server._run_prompt_submit("rid", "sid-normal", session, "hey") + assert agent.observed_no_tools is False + finally: + server._sessions.pop("sid-normal", None) + + +def test_prompt_submit_mode_chat_dispatches_tool_less(monkeypatch, tmp_path): + """End-to-end: ``prompt.submit`` with ``mode: "chat"`` reaches the turn as + a tool-less run (proves the handler reads the mode flag).""" + _configure_immediate_prompt_run(monkeypatch, tmp_path) + monkeypatch.setattr(server, "_ensure_session_db_row", lambda _session: None) + monkeypatch.setattr(server, "_persist_branch_seed", lambda _session: None) + monkeypatch.setattr(server, "_load_cfg", lambda: {"dashboard": {"turn_isolation": False}}) + fake_title = types.ModuleType("agent.title_generator") + setattr(fake_title, "maybe_auto_title", lambda *args, **kwargs: None) + monkeypatch.setitem(sys.modules, "agent.title_generator", fake_title) + + agent = _NoToolsCapturingAgent([]) + server._sessions["sid-e2e"] = _session(session_key="s-e2e", agent=agent) + try: + resp = server.handle_request( + { + "id": "turn-1", + "method": "prompt.submit", + "params": {"session_id": "sid-e2e", "text": "hey", "mode": "chat"}, + } + ) + assert resp["result"]["status"] == "streaming" + assert agent.observed_no_tools is True + finally: + server._sessions.pop("sid-e2e", None)   @pytest.mark.parametrize("exit_code", [0, 7])

diff --git hermes-agent/tui_gateway/entry.py harness/tui_gateway/entry.py index 00b801011b30699ae031897ce15a315567ce2a6b..073fe2d4fe67f32f643a5932b103c451715b8878 100644 --- hermes-agent/tui_gateway/entry.py +++ harness/tui_gateway/entry.py @@ -431,6 +431,13 @@ # gate inside ensure_mcp_discovery_started keeps the ~200ms MCP SDK # import cost entirely off the path for users with no mcp_servers. ensure_mcp_discovery_started()   + # Load user plugin handlers before the first RPC can arrive. + try: + from tui_gateway.artifact_plugin_loader import initial_load as _plugin_load + _plugin_load() + except Exception: + logger.warning("plugin initial_load failed", exc_info=True) + if not write_json({ "jsonrpc": "2.0", "method": "event",

diff --git hermes-agent/tui_gateway/methods_harness.py harness/tui_gateway/methods_harness.py new file mode 100644 index 0000000000000000000000000000000000000000..d41f13982b32f7efe4e200128175a139855bc323 --- /dev/null +++ harness/tui_gateway/methods_harness.py @@ -0,0 +1,1243 @@ +"""Harness-specific JSON-RPC handlers preserved across upstream rebases.""" + +import json +import os +import sys +from pathlib import Path +from typing import Any, Optional + +from hermes_constants import get_hermes_home + +from .method_ctx import HandlerRegistry + +# The wiki handlers below were carried into this module during the upstream +# rebase's handler split WITHOUT their imports — every wiki RPC (wiki.scan, +# wiki.page, wiki.list, wiki.taxonomy, wiki.expand_links, wiki.changesets) +# died with `NameError: name 'resolve_wiki' is not defined`, which the +# desktop rendered as "Failed to load page" on every wiki page. NameError, +# not ImportError, because Python resolves function-body names at CALL time — +# so the module imported cleanly and the break was invisible until the first +# wiki call. tests/gateway/test_methods_harness_imports.py now pins every +# name these handlers reference to an actual import. +from tui_gateway.wiki_api import ( + resolve_wiki, + wiki_changesets, + wiki_expand_links, + wiki_flatten_taxonomy, + wiki_list, + wiki_page, + wiki_scan, + wiki_taxonomy, +) + +_registry = HandlerRegistry() +method = _registry.method +_profile_scoped = _registry.profile_scoped + + +# Skills helpers recovered from the pre-rebase server.py (60b71f5a6) — the +# handler split carried their callers here but dropped the definitions, so +# the skills.* handlers NameError'd on first call, same class as the wiki +# import loss above. +def _find_local_skill_md(skill_name: str) -> Optional[Path]: + """Find a locally installed skill's SKILL.md file by name. + + Searches ~/.hermes/skills/ recursively for a SKILL.md whose YAML + frontmatter ``name`` field matches *skill_name*. Returns the + absolute Path or None. + """ + import yaml + + skills_dir = Path(get_hermes_home()) / "skills" + if not skills_dir.is_dir(): + return None + for candidate in sorted(skills_dir.rglob("SKILL.md")): + try: + raw = candidate.read_text(encoding="utf-8")[:4096] + except (OSError, UnicodeDecodeError): + continue + # Extract YAML frontmatter between --- markers + if not raw.startswith("---"): + continue + end = raw.find("---", 3) + if end == -1: + continue + try: + fm = yaml.safe_load(raw[3:end]) + except Exception: + continue + if isinstance(fm, dict) and fm.get("name") == skill_name: + return candidate + return None + + +def _parse_skill_frontmatter(content: str) -> dict[str, Any]: + """Parse YAML frontmatter from a SKILL.md string.""" + import yaml + + if not content.startswith("---"): + return {} + end = content.find("---", 3) + if end == -1: + return {} + try: + return yaml.safe_load(content[3:end]) or {} + except Exception: + return {} + + +def _skill_info_from_path(skill_md: Path, fm: dict[str, Any]) -> dict[str, Any]: + """Build a skill info dict consumable by HermesNative's SkillInfo.""" + category = skill_md.parent.name if skill_md.parent.parent != skill_md.parent.parent.parent else "general" + # Use the relative path from skills/ dir as the category path + try: + skills_dir = Path(get_hermes_home()) / "skills" + rel = skill_md.parent.relative_to(skills_dir) + # category is the top-level directory name + parts = rel.parts + if parts: + category = parts[0] if len(parts) >= 1 else "general" + except ValueError: + category = "general" + + info: dict[str, Any] = { + "name": fm.get("name", skill_md.parent.name), + "description": fm.get("description", ""), + "category": category, + "source": "local", + "id": fm.get("name", skill_md.parent.name), + "identifier": fm.get("name", skill_md.parent.name), + "tags": fm.get("metadata", {}).get("hermes", {}).get("tags", []) if isinstance(fm.get("metadata"), dict) else [], + "path": str(skill_md), + "skill_md_preview": "", # populated by caller after reading content + } + return info + +@method("session.prompt_breakdown") +def _(rid, params: dict) -> dict: + """Return the session's system prompt decomposed into sections with token counts.""" + session, err = _sess_nowait(params, rid) + if err: + return err + agent = session.get("agent") + home = Path(get_hermes_home()) + + try: + import tiktoken + _enc = tiktoken.get_encoding("cl100k_base") + _count = lambda t: len(_enc.encode(t)) if t else 0 + except Exception: + _count = lambda t: max(1, int(len(t) / 3.5)) if t else 0 + + def _section(name, source, content, color): + text = (content or "").strip() + return dict(name=name, source=source, + content_preview=text[:200], full_content=text[:10240], + token_count=_count(text), char_count=len(text), color=color) + + persona = "" + for p in [home / "SOUL.md", home / "persona.md"]: + if p.exists(): + persona = p.read_text().strip() + break + memory = (home / "memories" / "MEMORY.md").read_text().strip() if (home / "memories" / "MEMORY.md").exists() else "" + user = (home / "memories" / "USER.md").read_text().strip() if (home / "memories" / "USER.md").exists() else "" + ephemeral = str(getattr(agent, "ephemeral_system_prompt", "") or "").strip() + + skills = "" + if agent is not None and hasattr(agent, "tools"): + from run_agent import get_toolset_for_tool + toolsets = {ts for t in getattr(agent, "valid_tool_names", set()) if (ts := get_toolset_for_tool(t))} + if toolsets: + from agent.prompt_builder import build_skills_system_prompt + skills = build_skills_system_prompt(getattr(agent, "valid_tool_names", set()), toolsets) + + tools_json = json.dumps(getattr(agent, "tools", []) or [], ensure_ascii=False) + history = list(session.get("history", [])) + hist_json = json.dumps(history, ensure_ascii=False) + + sections = [ + _section("Persona", str(home / "SOUL.md"), persona, "#7c7cff"), + _section("Memory", str(home / "memories" / "MEMORY.md"), memory, "#ff7c7c"), + _section("User Profile", str(home / "memories" / "USER.md"), user, "#ffb87c"), + _section("Ephemeral Prompt", "(session personality/prompt)", ephemeral, "#7cff7c"), + _section("Active Skills", "~/.hermes/skills/", skills, "#ffd700"), + ] + + return _ok(rid, dict( + session_id=params.get("session_id", ""), + model=getattr(agent, "model", "") if agent else "", + context_limit=131072, + total_system_tokens=sum(s["token_count"] for s in sections), + sections=sections, + tool_definitions=dict(token_count=_count(tools_json), count=len(getattr(agent, "tools", []) or [])), + conversation_history=dict(token_count=_count(hist_json), message_count=len(history)), + )) + +@method("credits.view") +def _(rid, params: dict) -> dict: + """Structured Nous credit view for the TUI /credits command. + + Account-independent (a portal fetch gated on "a Nous account is logged in"), + so it works with no live agent / on a resumed session — same as the /usage + credits block. Returns the surface-agnostic CreditsView fields so the TUI can + render a clickable top-up <Link>. Fail-open: a portal hiccup or logged-out + account yields {logged_in: false}, never an error the user has to parse. + """ + try: + from agent.account_usage import build_credits_view + + view = build_credits_view() + return _ok( + rid, + { + "logged_in": bool(view.logged_in), + "balance_lines": [ + line for line in view.balance_lines if not line.lstrip().startswith("📈") + ], + "identity_line": view.identity_line, + "topup_url": view.topup_url, + "depleted": bool(view.depleted), + }, + ) + except Exception: + # Fail-open: TUI treats this as "not logged in" and shows the prompt. + return _ok(rid, {"logged_in": False, "balance_lines": [], "identity_line": None, "topup_url": None, "depleted": False}) + +@method("session.timeline") +def _(rid, params: dict) -> dict: + """Return temporally-structured session events for playback visualization.""" + session, _ = _sess_nowait(params, rid) + session_key = (session or {}).get("session_key") or params.get("session_id", "") + messages = list((session or {}).get("history", [])) + + db = _get_db() + if db and session_key: + try: + messages = db.get_messages_as_conversation(session_key, include_ancestors=True) + except Exception: + pass + + events = [] + tool_starts = {} + tool_count = 0 + in_tokens = out_tokens = 0 + + for m in messages: + role = m.get("role") + ts = m.get("timestamp") or time.time() + tc = m.get("token_count") + if isinstance(tc, (int, float)): + if role == "user": in_tokens += int(tc) + elif role == "assistant": out_tokens += int(tc) + + if role == "user": + if events and events[-1].get("type") != "turn_boundary": + events.append(dict(type="turn_boundary", timestamp=ts)) + events.append(dict(type="user_message", timestamp=ts, + content=_coerce_message_text(m.get("content")), + token_count=tc)) + elif role == "assistant": + for tc_item in m.get("tool_calls") or []: + fn = tc_item.get("function", {}) + tid = tc_item.get("id", "") + name = fn.get("name", "") + evt = dict(type="tool_start", timestamp=ts, tool_name=name, + tool_id=tid, summary=_tool_ctx(name, {})) + events.append(evt) + tool_starts[tid] = evt + tool_count += 1 + content = _coerce_message_text(m.get("content")) + if content.strip(): + events.append(dict(type="assistant_message", timestamp=ts, + content=content, token_count=tc)) + elif role == "tool": + tid = m.get("tool_call_id", "") + start = tool_starts.get(tid, {}) + dur = round((ts - start.get("timestamp", ts)) * 1000) if start.get("timestamp") and ts else None + events.append(dict(type="tool_end", timestamp=ts, + tool_name=m.get("tool_name", ""), tool_id=tid, + content=_coerce_message_text(m.get("content"))[:300], + duration_ms=dur)) + + tss = [e["timestamp"] for e in events if isinstance(e.get("timestamp"), (int, float))] + total_duration = round(tss[-1] - tss[0], 1) if len(tss) >= 2 else 0.0 + + cost = None + if db and session_key: + try: + s = db.get_session(session_key) or {} + cost = s.get("actual_cost_usd") or s.get("estimated_cost_usd") + except Exception: + pass + + return _ok(rid, dict( + session_id=session_key, events=events, + total_duration_seconds=total_duration, tool_calls=tool_count, + input_tokens=in_tokens, output_tokens=out_tokens, cost_usd=cost, + )) + +@method("skills.get") +def _(rid, params: dict) -> dict: + """Read a locally installed skill's SKILL.md content.""" + skill_name = params.get("skill_id", "") + file_path = params.get("file_path") # optional: relative path within skill dir + + if not skill_name: + return _err(rid, 4018, "missing skill_id") + + try: + skill_md = _find_local_skill_md(skill_name) + if skill_md is None: + return _err(rid, 4019, f"skill '{skill_name}' not found locally") + + # If a file_path is given, read a specific file from the skill dir + if file_path: + target = (skill_md.parent / file_path).resolve() + # Safety: only allow reading within the skill directory + try: + target.relative_to(skill_md.parent) + except ValueError: + return _err(rid, 4020, f"path '{file_path}' escapes skill directory") + if not target.is_file(): + return _err(rid, 4021, f"file '{file_path}' not found in skill dir") + content = target.read_text(encoding="utf-8") + return _ok(rid, { + "skill": _skill_info_from_path(skill_md, _parse_skill_frontmatter(content)), + "file_path": file_path, + "content": content, + "read_only": not os.access(target, os.W_OK), + }) + + content = skill_md.read_text(encoding="utf-8") + fm = _parse_skill_frontmatter(content) + + info = _skill_info_from_path(skill_md, fm) + info["skill_md_preview"] = content[:2000] + + read_only = not os.access(skill_md, os.W_OK) + + return _ok(rid, { + "skill": info, + "file_path": "SKILL.md", + "content": content, + "read_only": read_only, + }) + except Exception as e: + return _err(rid, 5024, str(e)) + +@method("skills.update") +def _(rid, params: dict) -> dict: + """Update a locally installed skill's SKILL.md content.""" + skill_name = params.get("skill_id", "") + new_content = params.get("content", "") + + if not skill_name: + return _err(rid, 4022, "missing skill_id") + if not new_content: + return _err(rid, 4023, "missing content") + + try: + skill_md = _find_local_skill_md(skill_name) + if skill_md is None: + return _err(rid, 4019, f"skill '{skill_name}' not found locally") + + if not os.access(skill_md, os.W_OK): + return _err(rid, 4024, f"skill '{skill_name}' is read-only") + + skill_md.write_text(new_content, encoding="utf-8") + fm = _parse_skill_frontmatter(new_content) + info = _skill_info_from_path(skill_md, fm) + info["skill_md_preview"] = new_content[:2000] + + # Reload skills so the agent picks up changes + try: + from agent.skill_commands import reload_skills + reload_skills() + except Exception: + pass # best-effort; don't fail the write + + return _ok(rid, {"skill": info}) + except Exception as e: + return _err(rid, 5024, str(e)) + +@method("files.list") +def _(rid, params: dict) -> dict: + """List a browsable root's file tree (the repo checkout or ~/.hermes). + + With no ``root`` param, returns the set of available root names so the + desktop can render the top level; with a ``root``, returns that root's + bounded, containment-checked tree. All logic lives in the pure + tui_gateway.files_browse module (locally imported so it resolves against + the real import system regardless of the handler's rebound globals).""" + try: + from tui_gateway.files_browse import FileBrowseError, file_roots, list_tree + + root = params.get("root") + if not root: + return _ok(rid, {"roots": sorted(file_roots().keys())}) + try: + return _ok(rid, list_tree(root, params.get("path", "") or "")) + except FileBrowseError as fe: + return _err(rid, fe.code, fe.message) + except Exception as e: + logger.exception("files.list failed") + return _err(rid, 5060, str(e)) + + +@method("files.read") +def _(rid, params: dict) -> dict: + """Read one file (read-only, UTF-8 text) from a browsable root.""" + try: + from tui_gateway.files_browse import FileBrowseError, read_file + + root = params.get("root") + path = params.get("path") + if not root: + return _err(rid, 4001, "root is required") + try: + return _ok(rid, read_file(root, path)) + except FileBrowseError as fe: + return _err(rid, fe.code, fe.message) + except Exception as e: + logger.exception("files.read failed") + return _err(rid, 5061, str(e)) + + +@method("session.set_prompt") +def _(rid, params: dict) -> dict: + """Set an ephemeral system prompt append on the live agent. + + The prompt is appended to the agent's system prompt on every API call + but is NOT persisted to trajectories or the session database. + Setting an empty string clears the ephemeral prompt. + """ + session, err = _sess(params, rid) + if err: + return err + agent = session.get("agent") + if not agent: + return _err(rid, 4001, "session not ready") + if session.get("running"): + return _err( + rid, + 4009, + "session busy — /interrupt the current turn before setting prompt", + ) + prompt = str(params.get("prompt", "") or "").strip() + agent.ephemeral_system_prompt = prompt or None + agent._cached_system_prompt = None + return _ok(rid, {"prompt": prompt}) + +@method("wiki.scan") +def _(rid, params: dict) -> dict: + try: + from tui_gateway.wiki_api import resolve_wiki, wiki_scan + wiki_name = params.get("wiki") or params.get("path") + wiki_path = resolve_wiki(wiki_name) + result = wiki_scan(wiki_path) + return _ok(rid, result) + except Exception as e: + logger.exception("wiki.scan failed") + return _err(rid, 5050, str(e)) + +@method("wiki.page") +def _(rid, params: dict) -> dict: + try: + from tui_gateway.wiki_api import resolve_wiki, wiki_page + page_path = params.get("path") + if not page_path: + return _err(rid, 4001, "path is required") + wiki_name = params.get("wiki") + wiki_path = resolve_wiki(wiki_name) + result = wiki_page(page_path, wiki_path) + if result is None: + return _err(rid, 4040, f"page not found: {page_path}") + return _ok(rid, result) + except Exception as e: + logger.exception("wiki.page failed") + return _err(rid, 5051, str(e)) + +@method("wiki.list") +def _(rid, params: dict) -> dict: + try: + result = wiki_list() + return _ok(rid, result) + except Exception as e: + logger.exception("wiki.list failed") + return _err(rid, 5052, str(e)) + + +@method("wiki.glossary") +def _(rid, params: dict) -> dict: + """Read the validated glossary for one configured wiki.""" + try: + from tui_gateway.wiki_glossary import ( + GlossaryValidationError, + load_glossary, + resolve_configured_wiki, + ) + + try: + unknown = set(params) - {"wiki"} + if unknown: + raise GlossaryValidationError( + f"unknown parameter(s): {', '.join(sorted(map(str, unknown)))}" + ) + wiki_path = resolve_configured_wiki(params.get("wiki")) + return _ok(rid, load_glossary(wiki_path)) + except GlossaryValidationError as exc: + return _err(rid, 4001, str(exc)) + except Exception as e: + logger.exception("wiki.glossary failed") + return _err(rid, 5062, str(e)) + + +@method("wiki.glossary.update") +def _(rid, params: dict) -> dict: + """Validate and atomically replace a configured wiki's glossary.""" + try: + from tui_gateway.wiki_glossary import ( + GlossaryConflictError, + GlossaryValidationError, + resolve_configured_wiki, + update_glossary, + ) + + try: + allowed = {"wiki", "version", "mode", "proper_nouns", "if_match"} + unknown = set(params) - allowed + if unknown: + raise GlossaryValidationError( + f"unknown parameter(s): {', '.join(sorted(map(str, unknown)))}" + ) + wiki_path = resolve_configured_wiki(params.get("wiki")) + if_match = params.get("if_match") + if if_match is not None and not isinstance(if_match, str): + raise GlossaryValidationError("if_match must be a string") + payload = { + "version": params.get("version"), + "mode": params.get("mode"), + "proper_nouns": params.get("proper_nouns"), + } + return _ok( + rid, + update_glossary(wiki_path, payload, if_match=if_match), + ) + except GlossaryConflictError as exc: + return _err(rid, 409, str(exc)) + except GlossaryValidationError as exc: + return _err(rid, 4001, str(exc)) + except Exception as e: + logger.exception("wiki.glossary.update failed") + return _err(rid, 5063, str(e)) + + +@method("wiki.taxonomy") +def _(rid, params: dict) -> dict: + """Return the hierarchical taxonomy tree from taxonomy.yaml.""" + try: + from tui_gateway.wiki_api import resolve_wiki, wiki_taxonomy, wiki_flatten_taxonomy + wiki_name = params.get("wiki") + wiki_path = resolve_wiki(wiki_name) + tax = wiki_taxonomy(wiki_path) + if tax is None: + return _err(rid, 4040, "taxonomy.yaml not found") + flat = wiki_flatten_taxonomy(wiki_path) + return _ok(rid, {"taxonomy": tax, "flat_paths": flat}) + except Exception as e: + logger.exception("wiki.taxonomy failed") + return _err(rid, 5053, str(e)) + +@method("wiki.expand_links") +def _(rid, params: dict) -> dict: + """Expand integration_links for a wiki page to live status objects.""" + try: + from tui_gateway.wiki_api import resolve_wiki, wiki_expand_links + slug = params.get("slug") + if not slug: + return _err(rid, 4001, "slug is required") + wiki_name = params.get("wiki") + wiki_path = resolve_wiki(wiki_name) + result = wiki_expand_links(slug, wiki_path) + if "error" in result: + return _err(rid, 4040, result["error"]) + return _ok(rid, result) + except Exception as e: + logger.exception("wiki.expand_links failed") + return _err(rid, 5054, str(e)) + +@method("wiki.changesets") +def _(rid, params: dict) -> dict: + """Return paginated wiki changesets (timeline view).""" + try: + from tui_gateway.wiki_api import resolve_wiki, wiki_changesets + wiki_name = params.get("wiki") + wiki_path = resolve_wiki(wiki_name) + result = wiki_changesets( + wiki_path=wiki_path, + page=params.get("page"), + action=params.get("action"), + trigger=params.get("trigger"), + limit=params.get("limit", 50), + offset=params.get("offset", 0), + since=params.get("since"), + until=params.get("until"), + ) + return _ok(rid, result) + except Exception as e: + logger.exception("wiki.changesets failed") + return _err(rid, 5055, str(e)) + +@method("wiki.events") +def _(rid, params: dict) -> dict: + """Return the ingestion event log — what caused wiki updates. + + A join over data already on disk: raw sources under ``raw/`` are the + events, and the changeset index records which events caused which page + writes. Each event carries the changesets it produced, so a client can + walk event → changeset → page and back. + + Params: + - ``kind`` (str, optional): filter by event kind. Kinds are defined by + ``type: event-type`` wiki pages, not a fixed list. + - ``limit`` (int, default 200, max 1000) / ``offset`` (int, default 0) + - ``since`` / ``until`` (ISO timestamps, optional) + - ``wiki`` (str, optional): wiki name (omit for default). + """ + try: + from tui_gateway.wiki_api import resolve_wiki, wiki_events + wiki_path = resolve_wiki(params.get("wiki")) + + result = wiki_events( + wiki_path=wiki_path, + kind=params.get("kind"), + limit=params.get("limit", 200), + offset=params.get("offset", 0), + since=params.get("since"), + until=params.get("until"), + ) + return _ok(rid, result) + except Exception as e: + logger.exception("wiki.events failed") + return _err(rid, 5059, str(e)) + +@method("wiki.changeset_diff") +def _(rid, params: dict) -> dict: + """Return the unified git diff for a single changeset. + + Params: + - ``id`` (str, required): changeset id from wiki.changesets. + - ``wiki`` (str, optional): wiki name (omit for default). + + Returns ``{"diff": "<unified diff>", "changeset": {...}}``. When the wiki + isn't git-initialized (older captures), returns error code 5057 with the + changeset attached so clients can still show metadata. + """ + try: + changeset_id = params.get("id") + if not changeset_id or not isinstance(changeset_id, str): + return _err(rid, 4001, "id must be a non-empty string") + from tui_gateway.wiki_api import resolve_wiki, wiki_changeset_diff + wiki_path = resolve_wiki(params.get("wiki")) + + result = wiki_changeset_diff(changeset_id, wiki_path=wiki_path) + if "error" in result: + return _err(rid, 5057, result["error"]) + return _ok(rid, result) + except Exception as e: + logger.exception("wiki.changeset_diff failed") + return _err(rid, 5056, str(e)) + +@method("wiki.update") +def _(rid, params: dict) -> dict: + """Write a wiki page — the one mutating method on the wiki surface. + + Full-replace write with optimistic concurrency (see the ``wiki.update`` + semantics in Portal's docs/rpc-reference.md): + + - ``path`` (str, required): page path relative to the wiki root (.md, + must resolve inside the root). + - ``body`` (str, required): FULL replacement markdown body. + - ``frontmatter`` (object, optional): REPLACES the entire frontmatter + block when present; omitted preserves it. ``updated`` is always set + server-side. + - ``if_match`` (str, optional): optimistic-concurrency precondition — + the ``updated`` the client read at load. Stale → error 409 with + ``data.latest`` carrying the server's current page. + - ``force`` (bool, default false): bypass the if_match precondition. + - ``trigger`` (str, default "manual"): what kind of change this is. + Was hardcoded to "manual", so every write through this method looked + identical in the timeline regardless of what made it. + - ``source_events`` (array of str, optional): the ingestion events that + caused this write, as wiki-relative raw source paths. Recorded as the + changeset's provenance; omitted reads downstream as *unknown*. + - ``summary`` (str, optional): changeset summary. + - ``wiki`` (str, optional): wiki name (omit for default). + + Records a changeset (action: update|create) with the usual git commit + capture, so edits appear in wiki.changesets. + """ + try: + page_path = params.get("path") + if not page_path or not isinstance(page_path, str): + return _err(rid, 4001, "path is required") + body = params.get("body") + if not isinstance(body, str): + return _err(rid, 4001, "body is required") + frontmatter = params.get("frontmatter") + if frontmatter is not None and not isinstance(frontmatter, dict): + return _err(rid, 4001, "frontmatter must be an object") + if_match = params.get("if_match") + if if_match is not None and not isinstance(if_match, str): + return _err(rid, 4001, "if_match must be a string") + trigger = params.get("trigger", "manual") + if not isinstance(trigger, str) or not trigger.strip(): + return _err(rid, 4001, "trigger must be a non-empty string") + source_events = params.get("source_events") + if source_events is not None and not isinstance(source_events, list): + return _err(rid, 4001, "source_events must be an array of strings") + summary = params.get("summary") + if summary is not None and not isinstance(summary, str): + return _err(rid, 4001, "summary must be a string") + + from tui_gateway.wiki_api import resolve_wiki, wiki_update + + wiki_path = resolve_wiki(params.get("wiki")) + + result = wiki_update( + page_path, + body, + frontmatter=frontmatter, + if_match=if_match, + force=bool(params.get("force", False)), + trigger=trigger.strip(), + source_events=source_events, + summary=summary, + wiki_path=wiki_path, + ) + if "error" in result: + if result.get("code") == "conflict": + return _err(rid, 409, result["error"], + data={"latest": result.get("latest")}) + return _err(rid, 4001, result["error"]) + return _ok(rid, result) + except Exception as e: + logger.exception("wiki.update failed") + return _err(rid, 5058, str(e)) + +@method("artifact.set") +def _(rid, params: dict) -> dict: + """Upsert a living artifact (a named model in the client render + dialects — map/chart/graph/stats/table/markdown — that any writer + maintains). Merges per kind server-side (map: markers union by label; + others replace) unless replace=true; appends a revision; emits + `artifact.changed` so connected clients stream the update live.""" + try: + from tui_gateway.artifact_store import set_artifact + + raw_actions = params.get("actions") + actions = raw_actions if isinstance(raw_actions, list) else None + raw_queries = params.get("queries") + queries = raw_queries if isinstance(raw_queries, list) else None + stored = set_artifact( + artifact_id=str(params.get("id", "")), + kind=str(params.get("kind", "")), + content=str(params.get("content", "")), + title=params.get("title"), + updated_by=str(params.get("updated_by", "")), + replace=bool(params.get("replace", False)), + actions=actions, + queries=queries, + ) + # A new revision may have changed what the page's queries mean; let + # subscribed slots re-run now rather than at their next tick. + from tui_gateway.artifact_queries import mark_changed as _queries_mark_changed + _queries_mark_changed(artifact_id=stored["id"]) + _emit("artifact.changed", "", { + "id": stored["id"], "kind": stored["kind"], + "title": stored["title"], "rev": stored["rev"], + "updated_at": stored["updated_at"], + "updated_by": stored["updated_by"], + }) + return _ok(rid, {"artifact": stored}) + except ValueError as e: + return _err(rid, 4001, str(e)) + except Exception as e: + logger.exception("artifact.set failed") + return _err(rid, 5210, str(e)) + +@method("artifact.get") +def _(rid, params: dict) -> dict: + """Fetch one artifact with content.""" + try: + from tui_gateway.artifact_store import get_artifact + + artifact = get_artifact(str(params.get("id", ""))) + if artifact is None: + return _err(rid, 4004, "artifact not found") + return _ok(rid, {"artifact": artifact}) + except Exception as e: + logger.exception("artifact.get failed") + return _err(rid, 5211, str(e)) + +@method("artifact.list") +def _(rid, params: dict) -> dict: + """All artifacts without content, newest first.""" + try: + from tui_gateway.artifact_store import list_artifacts + + return _ok(rid, {"artifacts": list_artifacts()}) + except Exception as e: + logger.exception("artifact.list failed") + return _err(rid, 5212, str(e)) + +@method("artifact.delete") +def _(rid, params: dict) -> dict: + """Remove an artifact (and its revisions); emits artifact.changed with + deleted=true.""" + try: + from tui_gateway.artifact_store import delete_artifact + + artifact_id = str(params.get("id", "")) + if not delete_artifact(artifact_id): + return _err(rid, 4004, "artifact not found") + _emit("artifact.changed", "", {"id": artifact_id, "deleted": True}) + return _ok(rid, {"deleted": artifact_id}) + except Exception as e: + logger.exception("artifact.delete failed") + return _err(rid, 5213, str(e)) + +@method("artifact.revisions") +def _(rid, params: dict) -> dict: + """Revision metadata for an artifact (no content), newest first — + the audit trail: who changed what, when.""" + try: + from tui_gateway.artifact_store import get_artifact, list_revisions + + artifact_id = str(params.get("id", "")) + if get_artifact(artifact_id) is None: + return _err(rid, 4004, "artifact not found") + return _ok(rid, {"revisions": list_revisions(artifact_id)}) + except Exception as e: + logger.exception("artifact.revisions failed") + return _err(rid, 5214, str(e)) + +@method("artifact.revision") +def _(rid, params: dict) -> dict: + """One revision's full content (time-travel view / restore source).""" + try: + from tui_gateway.artifact_store import get_revision + + revision = get_revision(str(params.get("id", "")), int(params.get("rev", 0))) + if revision is None: + return _err(rid, 4004, "revision not found") + return _ok(rid, {"revision": revision}) + except (TypeError, ValueError): + return _err(rid, 4001, "rev must be an integer") + except Exception as e: + logger.exception("artifact.revision failed") + return _err(rid, 5215, str(e)) + +@method("artifact.action.invoke") +def _(rid, params: dict) -> dict: + """Invoke a backend intent declared in an artifact's action manifest. + + The client sends only stable identifiers — artifact ID, pinned revision, + binding ID, entity ref, an idempotency key, and optional bounded human + context for contained-session intents. The server resolves the + registered handler from the artifact's declarations at that revision; + a forged binding or substituted intent name is rejected because the + server never trusts the caller's intent string. + + Returns: {"status": "needs_confirmation"|"succeeded"|"failed"| + "conflict"|"unsupported", ...} + """ + try: + from tui_gateway.artifact_actions import invoke + + result = invoke( + artifact_id=str(params.get("artifact_id", "")), + artifact_rev=int(params.get("artifact_rev", 0)), + binding_id=str(params.get("binding_id", "")), + entity_ref=str(params.get("entity_ref", "")), + idempotency_key=str(params.get("idempotency_key", "")), + user_context=params.get("user_context"), + ) + if result.get("status") == "succeeded": + # Emit artifact.changed so the client refreshes live. + from tui_gateway.artifact_store import get_artifact + artifact = get_artifact(str(params.get("artifact_id", ""))) + if artifact: + _emit("artifact.changed", "", { + "id": artifact["id"], "kind": artifact.get("kind", ""), + "title": artifact.get("title", ""), "rev": artifact.get("rev", 0), + "updated_at": artifact.get("updated_at", ""), + "updated_by": artifact.get("updated_by", ""), + }) + return _ok(rid, result) + except (TypeError, ValueError) as e: + return _err(rid, 4001, str(e)) + except Exception as e: + logger.exception("artifact.action.invoke failed") + return _err(rid, 5216, str(e)) + +@method("artifact.action.confirm") +def _(rid, params: dict) -> dict: + """Complete a pending destructive intent after native confirmation. + + ``challenge`` is the short-lived token issued by artifact.action.invoke + when the handler requires confirmation. It is bound to actor, artifact + revision, binding, resolved target, and expiry on the server — the + artifact cannot weaken confirmation policy by declaring confirm: false. + """ + try: + from tui_gateway.artifact_actions import confirm + + result = confirm( + artifact_id=str(params.get("artifact_id", "")), + challenge=str(params.get("challenge", "")), + ) + if result.get("status") == "succeeded": + from tui_gateway.artifact_store import get_artifact + artifact = get_artifact(str(params.get("artifact_id", ""))) + if artifact: + _emit("artifact.changed", "", { + "id": artifact["id"], "kind": artifact.get("kind", ""), + "title": artifact.get("title", ""), "rev": artifact.get("rev", 0), + "updated_at": artifact.get("updated_at", ""), + "updated_by": artifact.get("updated_by", ""), + }) + return _ok(rid, result) + except Exception as e: + logger.exception("artifact.action.confirm failed") + return _err(rid, 5217, str(e)) + +@method("artifact.action.log") +def _(rid, params: dict) -> dict: + """Query the invocation ledger for an artifact. + + Params: + artifact_id (str, required) + binding_id (str, optional) — filter to one binding + entity_ref (str, optional) — filter to one entity + limit (int, optional, default 50, max 200) + + Returns newest-first list of invocation records. Native uses this to + re-hydrate badge state on artifact-pane open after app restart. + """ + try: + artifact_id = (params.get("artifact_id") or "").strip() + if not artifact_id: + return _err(rid, 4001, "artifact_id required") + from tui_gateway.artifact_invocation_ledger import query as _ledger_query + records = _ledger_query( + artifact_id=artifact_id, + binding_id=params.get("binding_id"), + entity_ref=params.get("entity_ref"), + limit=int(params.get("limit", 50)), + ) + return _ok(rid, {"records": records}) + except Exception as e: + logger.exception("artifact.action.log failed") + return _err(rid, 5219, str(e)) + +@method("artifact.query.invoke") +def _(rid, params: dict) -> dict: + """Run a query the artifact declares, with page-supplied parameters. + + The read side of ``artifact.action.invoke``. The client sends the + artifact ID, its pinned revision, the ``query_id`` from the artifact's + ``queries`` manifest, a ``params`` object and an optional ``cursor``. The + server resolves the registered handler from the manifest at that + revision, validates every parameter against the artifact's declared + schema and the handler's own, runs the handler, and returns JSON data + with an ``etag``. The caller never names a handler and never sends query + text. + + Returns: {"status": "ok"|"failed"|"conflict"|"unsupported", ...} + """ + try: + from tui_gateway.artifact_queries import invoke as _query_invoke + + raw_params = params.get("params") + if raw_params is not None and not isinstance(raw_params, dict): + return _err(rid, 4001, "params must be an object") + rev = params.get("artifact_rev") + result = _query_invoke( + artifact_id=str(params.get("artifact_id", "")), + artifact_rev=int(rev) if rev is not None else None, + query_id=str(params.get("query_id", "")), + params=raw_params, + cursor=params.get("cursor"), + ) + return _ok(rid, result) + except (TypeError, ValueError) as e: + return _err(rid, 4001, str(e)) + except Exception as e: + logger.exception("artifact.query.invoke failed") + return _err(rid, 5220, str(e)) + +@method("artifact.query.subscribe") +def _(rid, params: dict) -> dict: + """Follow a declared query: the gateway re-runs it on the artifact's + declared ``live`` cadence (or when a plugin reports a change) and emits + ``artifact.query.changed`` only when the result's etag differs. Returns + the current result plus a ``subscription`` handle for + ``artifact.query.unsubscribe``. Same params as ``artifact.query.invoke``. + """ + try: + from tui_gateway import artifact_queries as _queries + + # Broadcasts go through the server's own emitter; installed here so the + # poller thread has one by the time it has anything to say. + _queries.set_emitter(lambda event, payload: _emit(event, "", payload)) + raw_params = params.get("params") + if raw_params is not None and not isinstance(raw_params, dict): + return _err(rid, 4001, "params must be an object") + rev = params.get("artifact_rev") + result = _queries.subscribe( + artifact_id=str(params.get("artifact_id", "")), + artifact_rev=int(rev) if rev is not None else None, + query_id=str(params.get("query_id", "")), + params=raw_params, + ) + return _ok(rid, result) + except (TypeError, ValueError) as e: + return _err(rid, 4001, str(e)) + except Exception as e: + logger.exception("artifact.query.subscribe failed") + return _err(rid, 5221, str(e)) + +@method("artifact.query.unsubscribe") +def _(rid, params: dict) -> dict: + """Drop a subscription handle from ``artifact.query.subscribe``. The slot + stops being polled once its last subscriber leaves.""" + try: + from tui_gateway.artifact_queries import unsubscribe as _query_unsubscribe + + handle = str(params.get("subscription", "")).strip() + if not handle: + return _err(rid, 4001, "subscription required") + return _ok(rid, _query_unsubscribe(handle)) + except Exception as e: + logger.exception("artifact.query.unsubscribe failed") + return _err(rid, 5222, str(e)) + +@method("artifact.query.handlers") +def _(rid, params: dict) -> dict: + """The registered query handler names and their parameter schemas — what + an artifact author (human or agent) may declare against. Built-ins plus + whatever the loaded plugins registered.""" + try: + from tui_gateway.artifact_queries import _QUERY_HANDLERS + + handlers = [ + {"name": name, "params": entry.get("params") or {}} + for name, entry in sorted(_QUERY_HANDLERS.items()) + ] + return _ok(rid, {"handlers": handlers}) + except Exception as e: + logger.exception("artifact.query.handlers failed") + return _err(rid, 5223, str(e)) + +@method("actions.reload") +def _(rid, params: dict) -> dict: + """Reload plugin action handlers from ~/.hermes/plugins/actions/. + + Returns a diff of added/changed/removed handler names and the list of + files that were loaded. On any parse/exec error the live registry is + unchanged and the traceback is returned in ``error``. + + Safe to call from an agent tool, CLI, or RPC — the security boundary is + the plugins directory's non-agent-writability, not the trigger. + """ + try: + from tui_gateway.artifact_plugin_loader import reload as _reload + result = _reload() + return _ok(rid, result) + except Exception as e: + logger.exception("actions.reload failed") + return _err(rid, 5218, str(e)) + +@method("gateway.restart") +def _(rid, params: dict) -> dict: + """Re-exec the gateway process in place, loading all updated source files. + + Use after pulling new code onto the gateway host — the response is sent + before the process replaces itself, so callers receive it reliably. + + The native app will experience a WebSocket disconnect followed by the + standard reconnect sequence. On reconnect, gateway.capabilities will + reflect any new capabilities added by the update. + + Safe to call from an agent tool (e.g. after ``git pull`` updates the + fork). The process re-execs with the same ``sys.argv`` and inherits + the environment, so all env vars (LINEAR_API_KEY, HERMES_HOME, etc.) + are preserved. + """ + import threading + + def _do_restart(): + import time + time.sleep(0.15) # let the response frame flush through the transport + os.execv(sys.executable, [sys.executable] + sys.argv) + + threading.Thread(target=_do_restart, daemon=True).start() + return _ok(rid, {"status": "restarting"}) + +@method("gateway.capabilities") +def _(rid, params: dict) -> dict: + """Report gateway capabilities so native clients can feature-gate + controls without trial-and-error method calls. + + ``capability_names`` is the authoritative set — clients check for + substring matches (e.g. ``artifact.action``) rather than exact values + so adding sub-capabilities doesn't break old clients. + """ + try: + from hermes_cli import __version__, __release_date__ + version = f"{__version__}+{__release_date__}" + except Exception: + version = "unknown" + + return _ok(rid, { + "gateway_version": version, + "capability_names": [ + "artifact.set", + "artifact.get", + "artifact.list", + "artifact.delete", + "artifact.revisions", + "artifact.revision", + "artifact.action", + "artifact.action.invoke", + "artifact.action.confirm", + "artifact.action.reload", + "artifact.action.log", + "artifact.query", + "artifact.query.invoke", + "artifact.query.subscribe", + "artifact.query.handlers", + "gateway.restart", + "wiki.scan", + "wiki.page", + "wiki.list", + "wiki.glossary", + "wiki.glossary.update", + "wiki.changesets", + "wiki.events", + "learning.course", + "learning.deck", + "learning.progress", + "learning.review", + "prompt.chat_mode", + ], + }) + +@method("feed.get") +def _(rid, params: dict) -> dict: + """Return curated news feed from digest pipelines.""" + try: + from tui_gateway.digest_store import get_feed as _feed_get + sources = params.get("sources") + if sources is not None and not isinstance(sources, list): + return _err(rid, 4001, "sources must be a list or null") + result = _feed_get( + sources=sources, since=params.get("since"), + limit=params.get("limit", 50), offset=params.get("offset", 0), + ) + return _ok(rid, result) + except Exception as e: + logger.exception("feed.get failed") + return _err(rid, 5200, str(e)) + +@method("feed.sources") +def _(rid, params: dict) -> dict: + """Return available feed sources and article counts.""" + try: + from tui_gateway.digest_store import get_sources as _feed_sources + result = _feed_sources() + return _ok(rid, result) + except Exception as e: + logger.exception("feed.sources failed") + return _err(rid, 5201, str(e)) + +@method("feed.publish") +def _(rid, params: dict) -> dict: + """Append articles to the news feed store (the producer side of feed.get). + + Params: + - ``source`` (str, required): feed source name (e.g. ``"ai-digest"``), + shown as a filter tab and used as the dedup key. + - ``articles`` (list[dict], required): each may carry ``title``, + ``url``, ``summary``, ``tags``, ``image_url``. Articles are deduped + against what was already stored for the same source. + + Returns ``{"total": N}`` — the feed size after the append. + """ + try: + from tui_gateway.digest_store import append_digest as _feed_publish + source = params.get("source") + if not source or not isinstance(source, str): + return _err(rid, 4001, "source must be a non-empty string") + articles = params.get("articles") + if not isinstance(articles, list): + return _err(rid, 4001, "articles must be a list") + total = _feed_publish(source, articles) + return _ok(rid, {"total": total}) + except Exception as e: + logger.exception("feed.publish failed") + return _err(rid, 5202, str(e)) + +@method("push.register") +def _(rid, params: dict) -> dict: + """Register an APNs device token for remote push notifications. + + Params: + - ``token`` (str, required): hex APNs device token. + - ``platform`` (str): "macos" (default) or "ios". + - ``device_name`` (str, optional): human-readable device label. + - ``bundle_id`` (str, optional): per-device topic override (macOS and + iOS builds have different bundle ids). + + Returns the stored entry plus ``apns_configured`` so clients can tell the + user when the gateway has no APNs credentials. + """ + try: + from tui_gateway.apns_sender import is_configured + from tui_gateway.push_store import register_token + + token = params.get("token") + if not token or not isinstance(token, str): + return _err(rid, 4001, "token must be a non-empty string") + entry = register_token( + token, + platform=params.get("platform", "macos"), + device_name=params.get("device_name", ""), + bundle_id=params.get("bundle_id"), + ) + if "error" in entry: + return _err(rid, 4001, entry["error"]) + return _ok(rid, {"registered": True, "apns_configured": is_configured(), "entry": entry}) + except Exception as e: + logger.exception("push.register failed") + return _err(rid, 5210, str(e)) + +@method("push.unregister") +def _(rid, params: dict) -> dict: + """Remove an APNs device token (e.g. on sign-out).""" + try: + from tui_gateway.push_store import unregister_token + + token = params.get("token") + if not token or not isinstance(token, str): + return _err(rid, 4001, "token must be a non-empty string") + return _ok(rid, {"removed": unregister_token(token)}) + except Exception as e: + logger.exception("push.unregister failed") + return _err(rid, 5211, str(e)) + + +def register(server) -> None: + _registry.install(server) \ No newline at end of file

diff --git hermes-agent/tui_gateway/server.py harness/tui_gateway/server.py index 14d6dbc2f2fad96654c3a3c6b1307d602281ea89..6cd96f0f2a65893e2f52b6e4e03ebe4179f9eb4c 100644 --- hermes-agent/tui_gateway/server.py +++ harness/tui_gateway/server.py @@ -249,6 +249,9 @@ # dead after a few skin switches. The handler serializes concurrent # reloads via _mcp_reload_lock. "reload.mcp", "process.list", + # Service collectors and application probes run subprocess/network I/O; + # keep graph health refreshes off the gateway reader thread. + "cron.graph", "projects.discover_repos", "projects.record_repos", "projects.for_cwd", @@ -280,6 +283,10 @@ "session.workspace.move", "shell.exec", "skills.manage", "slash.exec", + # Glossary reads/writes touch per-wiki YAML and fsync updates. Keep that + # filesystem I/O off the JSON-RPC reader thread. + "wiki.glossary", + "wiki.glossary.update", } )   @@ -1299,6 +1306,36 @@   atexit.register(_shutdown_sessions) _start_idle_reaper() + + +def _start_wiki_watcher() -> None: + """Programmatic wiki-event enforcement: capture changesets for page + writes that bypass wiki.update (agent file tools, humans, scripts) and + emit wiki.changed so the desktop feed updates without relying on the + agent being prompted to run the capture CLI. See wiki_watch.py.""" + try: + from tui_gateway import wiki_api, wiki_watch + + def _roots() -> list: + roots = set() + try: + roots.add(wiki_api.resolve_wiki(None)) + except Exception: + pass + try: + for w in wiki_api.wiki_list().get("wikis", []): + if w.get("path"): + roots.add(w["path"]) + except Exception: + pass + return sorted(roots) + + wiki_watch.start_wiki_watcher(emit=_emit, wiki_roots=_roots) + except Exception: + logger.debug("wiki watcher failed to start", exc_info=True) + + +_start_wiki_watcher()   # ── Plumbing ────────────────────────────────────────────────────────── @@ -9714,6 +9751,7 @@ display_kind: str | None = None, display_metadata: dict | None = None, image_paths: list[str] | None = None, queued_prompt_generation: int | None = None, + chat_mode: bool = False, ) -> None: with session["history_lock"]: if ( @@ -10010,6 +10048,12 @@ except (TypeError, ValueError): _run_params = {} if "task_id" in _run_params: run_kwargs["task_id"] = session["session_key"] + # Chat mode: route this turn through the tool-less path (voice + # conversation replies — plain completion, no tool loop, no action + # side effects). Signature-gated so an older agent without the + # parameter simply runs the normal tool-enabled turn. + if chat_mode and "no_tools" in _run_params: + run_kwargs["no_tools"] = True if display_kind and "persist_user_display_kind" in _run_params: run_kwargs["persist_user_display_kind"] = display_kind run_kwargs["persist_user_display_metadata"] = display_metadata @@ -14417,6 +14461,8 @@ methods_config as _methods_config, methods_prompt as _methods_prompt, methods_session as _methods_session, methods_tools as _methods_tools, + methods_harness as _methods_harness, + methods_learning as _methods_learning, )   for _m in ( @@ -14425,6 +14471,8 @@ _methods_prompt, _methods_config, _methods_complete, _methods_tools, + _methods_harness, + _methods_learning, ): _m.register(sys.modules[__name__]) del _m

diff --git hermes-agent/tui_gateway/methods_tools.py harness/tui_gateway/methods_tools.py index dc12130c6fd303a556b1bdda39acd5d766c13a2d..49f6386a9cfb7e80f927a2e547ba0c0153cbdb48 100644 --- hermes-agent/tui_gateway/methods_tools.py +++ harness/tui_gateway/methods_tools.py @@ -1632,25 +1632,272 @@ action, jid = params.get("action", "list"), params.get("name", "") try: from tools.cronjob_tools import cronjob   - if action == "list": - return _ok(rid, json.loads(cronjob(action="list"))) - if action == "add": - return _ok( - rid, - json.loads( - cronjob( - action="create", - name=jid, - schedule=params.get("schedule", ""), - prompt=params.get("prompt", ""), + # This method is the UI's door to cron — Portal and the TUI call it when + # a person clicks. The model reaches the same code in-process through its + # own tool, which attributes itself to the agent, so binding the human + # here is what keeps the recorded actor from naming the wrong author of + # every change made by hand. `params.get("actor")` lets a caller that + # knows better say so; anything unrecognized is passed through verbatim + # rather than folded into "unknown". + from cron.changesets import use_changeset_origin + + actor = str(params.get("actor") or "human").strip() or "human" + with use_changeset_origin(actor): + if action == "list": + return _ok(rid, json.loads(cronjob(action="list"))) + if action == "add": + return _ok( + rid, + json.loads( + cronjob( + action="create", + name=jid, + schedule=params.get("schedule", ""), + prompt=params.get("prompt", ""), + ) + ), + ) + if action in {"remove", "pause", "resume"}: + return _ok(rid, json.loads(cronjob(action=action, job_id=jid))) + if action in {"describe", "history"}: + # Read paths. `list` caps `prompt_preview` at 100 chars for the + # model's benefit; a person expanding a job card needs the whole + # prompt (`describe`) and the execution ledger (`history`). + if not jid: + return _err(rid, 4001, "name (job id) is required") + from cron.jobs import get_job, resolve_job_ref + + job = get_job(jid) or resolve_job_ref(jid) + if job is None: + return _err(rid, 4404, f"cron job '{jid}' not found") + if action == "describe": + from cron.jobs import job_source_files + from tools.cronjob_tools import _format_job + + detail = _format_job(job) + detail["prompt"] = job.get("prompt") or "" + for field in ("inputs", "outputs", "side_effects", "source_files", "context_from"): + detail[field] = list(job.get(field) or []) + detail["source_files_resolved"] = job_source_files(job) + return _ok(rid, {"success": True, "job": detail}) + from cron.executions import list_executions + + try: + limit = max(1, min(int(params.get("limit") or 50), 500)) + except (TypeError, ValueError): + limit = 50 + runs = list_executions(job_id=job["id"], limit=limit) + return _ok( + rid, + { + "success": True, + "job_id": job["id"], + "job_name": job.get("name"), + "count": len(runs), + "runs": runs, + }, + ) + if action == "update": + if not jid: + return _err(rid, 4001, "name (job id) is required") + # `name` is already the job identifier on this method, so the + # new friendly name travels as `job_name`. Every other field is + # passed straight to the tool, which normalizes + validates. + kwargs = {} + if "prompt" in params: + kwargs["prompt"] = params["prompt"] + if "job_name" in params: + kwargs["name"] = params["job_name"] + for key in ( + "schedule", "deliver", "repeat", "skills", "script", + "monitor_script", "monitor_url", "context_from", "workdir", + "enabled_toolsets", "inputs", "outputs", "side_effects", + "source_files", + ): + if key in params: + kwargs[key] = params[key] + if not kwargs: + return _err(rid, 4001, "update needs at least one field to change") + result = json.loads(cronjob(action="update", job_id=jid, **kwargs)) + if result.get("success") is False: + return _err(rid, 4017, str(result.get("error") or "update failed")) + return _ok(rid, result) + return _err(rid, 4016, f"unknown cron action: {action}") + except Exception as e: + return _err(rid, 5023, str(e)) + + +@method("cron.graph") +def _(rid, params: dict) -> dict: + """Cron interflow dataflow graph — nodes (crons/sources/artifacts/sinks) + + typed edges — for Portal to render. Same edge shape as ``wiki.scan``.""" + try: + from cron.jobs import build_cron_graph + + # Overlay live long-running services (dashboards, APIs) that self-declared + # their dataflow — they meet crons on shared resource nodes. Four liveness + # providers feed the same merge: tracked background processes (the process + # IS the lease), Docker containers (labels + `docker ps`), Nomad jobs + # (meta + running allocations), and launchd services (sidecar registry + + # `launchctl print` state probe). Best-effort per provider — a hiccup in + # any must never sink the cron graph itself. + # Independent provider probes run concurrently. Each collector is already + # bounded and fail-open; parallelism keeps Portal's 10-second health + # refresh near the slowest provider rather than the sum of four timeouts. + from concurrent.futures import ThreadPoolExecutor, as_completed + from tools.process_registry import process_registry + from tools.docker_services import collect_docker_services + from tools.nomad_services import collect_nomad_services + from tools.launchd_services import collect_launchd_services + + collectors = { + "process": process_registry.collect_service_declarations, + "docker": collect_docker_services, + "nomad": collect_nomad_services, + "launchd": collect_launchd_services, + } + services = [] + with ThreadPoolExecutor(max_workers=len(collectors)) as executor: + pending = {executor.submit(fn): name for name, fn in collectors.items()} + for future in as_completed(pending): + provider = pending[future] + try: + services.extend(future.result()) + except Exception: + logger.exception( + "cron.graph: %s service overlay unavailable", provider + ) + + return _ok(rid, build_cron_graph(services=services)) + except Exception as e: + logger.exception("cron.graph failed") + return _err(rid, 5024, str(e)) + + +@method("code.graph") +def _(rid, params: dict) -> dict: + """Code knowledge graph for one service — nodes (modules/classes/functions) + and typed edges (imports/calls) built from the service's declared + ``source_files``, in the same ``{source, target, type, class}`` shape + ``cron.graph``/``wiki.scan`` use so Portal's renderer is reused. + + Derived on read and content-digest cached, so it tracks the current service + definition without riding the changeset log. Fail-open: an unknown service, + a service with no readable in-root files, or a graphify hiccup returns a + soft error, never a 500 that would sink the surface.""" + try: + service_id = (params or {}).get("service") + if not isinstance(service_id, str) or not service_id.strip(): + return _err(rid, 4029, "code.graph needs a 'service' id") + service_id = service_id.strip() + + # Resolve the service the same way cron.graph does — the four concurrent + # liveness collectors — then pick the one whose id matches. + from concurrent.futures import ThreadPoolExecutor, as_completed + from tools.process_registry import process_registry + from tools.docker_services import collect_docker_services + from tools.nomad_services import collect_nomad_services + from tools.launchd_services import collect_launchd_services + + collectors = { + "process": process_registry.collect_service_declarations, + "docker": collect_docker_services, + "nomad": collect_nomad_services, + "launchd": collect_launchd_services, + } + services = [] + with ThreadPoolExecutor(max_workers=len(collectors)) as executor: + pending = {executor.submit(fn): name for name, fn in collectors.items()} + for future in as_completed(pending): + try: + services.extend(future.result()) + except Exception: + logger.exception( + "code.graph: %s service overlay unavailable", pending[future] ) - ), + + service = next((s for s in services if s.get("id") == service_id), None) + if service is None: + return _err(rid, 4030, f"unknown service: {service_id}") + + from cron.jobs import job_source_files + from cron.code_graph import build_service_code_graph, CodeGraphUnavailable + + try: + graph = build_service_code_graph( + service_id, + source_files=job_source_files(service), + code_control=service.get("code_control"), ) - if action in {"remove", "pause", "resume"}: - return _ok(rid, json.loads(cronjob(action=action, job_id=jid))) - return _err(rid, 4016, f"unknown cron action: {action}") + except CodeGraphUnavailable as exc: + return _err(rid, 4031, f"code graph unavailable: {exc}") + return _ok(rid, graph) + except Exception as e: + logger.exception("code.graph failed") + return _err(rid, 5038, str(e)) + + +@method("cron.changesets") +def _(rid, params: dict) -> dict: + """Recorded history of the cron configuration, newest first. + + The counterpart to ``cron.graph``: that one says what the wiring is now, this + one says when it changed, who changed it, and which session caused it — the + three things a client polling ``cron.graph`` can only guess at. See + ``cron/changesets.py`` for what counts as a change (configuration only, never + liveness) and for the digest's compatibility with Portal's own. + """ + try: + from cron.changesets import ensure_baseline, read_changesets + + # Open the log on the current configuration if nothing has written to it + # yet, so the first recorded change has a real "before" to be compared + # against instead of being reported against the empty graph. Best-effort: + # a store that can't be written to can still be read from. + try: + ensure_baseline() + except Exception: + logger.debug("cron.changesets: baseline not established", exc_info=True) + + return _ok( + rid, + read_changesets( + limit=max(1, min(int(params.get("limit") or 50), 500)), + offset=max(0, int(params.get("offset") or 0)), + since=params.get("since") or None, + until=params.get("until") or None, + job=params.get("job") or None, + ), + ) except Exception as e: - return _err(rid, 5023, str(e)) + logger.exception("cron.changesets failed") + return _err(rid, 5036, str(e)) + + +@method("cron.changeset_diff") +def _(rid, params: dict) -> dict: + """The configurations on either side of one recorded change. + + Graphs, not prose. The client derives its change statements from these with + the same code it runs over its own observed history; a second account of + "what changed", written here, would drift from that one. ``before`` is + omitted — not empty — when the previous revision isn't in the log (the + baseline row, or a parent trimmed off the end), because a client that read an + empty graph there would report a steady-state configuration as freshly built. + """ + try: + from cron.changesets import read_changeset_diff + + changeset_id = str(params.get("id") or "").strip() + if not changeset_id: + return _err(rid, 4020, "cron.changeset_diff requires an id") + payload = read_changeset_diff(changeset_id) + if payload is None: + return _err(rid, 4021, f"unknown cron changeset: {changeset_id}") + return _ok(rid, payload) + except Exception as e: + logger.exception("cron.changeset_diff failed") + return _err(rid, 5037, str(e))   @method("learning.frames")

diff --git hermes-agent/agent/conversation_loop.py harness/agent/conversation_loop.py index 1a236e6eacc55a9a64837e4435a08cf4f48ec421..e2da31989cbe5fc75cd2846a6b653c2a28bcf4dd 100644 --- hermes-agent/agent/conversation_loop.py +++ harness/agent/conversation_loop.py @@ -1431,6 +1431,7 @@ persist_user_timestamp: Optional[float] = None, persist_user_display_kind: Optional[str] = None, persist_user_display_metadata: Optional[Dict[str, Any]] = None, moa_config: Optional[dict[str, Any]] = None, + no_tools: bool = False, ) -> Dict[str, Any]: """ Run a complete conversation with tool calling until completion. @@ -2107,7 +2108,13 @@ # bytes on consecutive turns, which breaks the prefix match at # exactly the point the breakpoints were meant to protect. Marking # last also keeps breakpoints off messages that the orphan sweep or # the thinking-only drop is about to remove or merge away. - tools_for_api = agent.tools + # Chat mode (no_tools): send the turn with an empty tool list so the + # model returns a plain completion — no tool loop, no action side + # effects, lower latency. Adapters treat an empty list as "no tools" + # and omit the ``tools`` key entirely. ``agent.tools`` is the canonical + # per-session registry, so we substitute a fresh empty list here rather + # than mutating it — the next (normal) turn still sees the full set. + tools_for_api = [] if no_tools else agent.tools if agent._use_prompt_caching and agent.provider != "moa": _static_system_prefix = getattr(agent, "_cached_system_prompt_static", None) _initial_cache_plan = build_prompt_cache_plan(

diff --git hermes-agent/run_agent.py harness/run_agent.py index 9589606745211eb6bb5c8339103815ede5616b8d..922d9ec3c99b83db70b2b20ee7867b226e9b79e4 100644 --- hermes-agent/run_agent.py +++ harness/run_agent.py @@ -7904,6 +7904,7 @@ persist_user_timestamp: Optional[float] = None, persist_user_display_kind: Optional[str] = None, persist_user_display_metadata: Optional[Dict[str, Any]] = None, moa_config: Optional[dict[str, Any]] = None, + no_tools: bool = False, ) -> Dict[str, Any]: """Forwarder — see ``agent.conversation_loop.run_conversation``.""" from agent.aux_accounting import ( @@ -7998,6 +7999,7 @@ persist_user_timestamp=persist_user_timestamp, persist_user_display_kind=persist_user_display_kind, persist_user_display_metadata=persist_user_display_metadata, moa_config=moa_config, + no_tools=no_tools, ) terminal = result if isinstance(result, dict) else {} if terminal.get("interrupted") is True:

diff --git hermes-agent/tui_gateway/methods_prompt.py harness/tui_gateway/methods_prompt.py index d0efe89f9f057c7c08699be9bbff0bf187781545..15aed9ee3953ec98ae57e754d8671abce205a49f 100644 --- hermes-agent/tui_gateway/methods_prompt.py +++ harness/tui_gateway/methods_prompt.py @@ -71,6 +71,12 @@ sid = params.get("session_id", "") raw_text = params.get("text", "") text = sanitize_user_prompt_text(raw_text) if isinstance(raw_text, str) else raw_text + # Tool-less chat turn (opt-in via ``mode: "chat"``). Used by the desktop's + # voice conversation loop: replies should be a plain completion — no tool + # loop, no action side effects, lower latency — because a spoken + # back-and-forth is a conversation, not an action request. Absent/other + # values keep the normal tool-enabled agent turn. + chat_mode = params.get("mode") == "chat" # Typed bare stop phrase while backend voice mode is active ends the # voice chat instead of sending "stop" to the agent — the typed twin of # the spoken stop phrase (PR #73106), applied at the ONE server-side @@ -386,7 +392,7 @@ else "Session no longer running before the agent was ready" }, ) return - _run_prompt_submit(rid, sid, session, text) + _run_prompt_submit(rid, sid, session, text, chat_mode=chat_mode)   run_thread = threading.Thread(target=run_after_agent_ready, daemon=True) # Keep a handle so session.interrupt can tell a live turn from a stuck

diff --git hermes-agent/tests/tui_gateway/test_wiki.py harness/tests/tui_gateway/test_wiki.py new file mode 100644 index 0000000000000000000000000000000000000000..d2951106a4192b1a6eb17b2e6248480ce9201e56 --- /dev/null +++ harness/tests/tui_gateway/test_wiki.py @@ -0,0 +1,326 @@ +import tempfile +from pathlib import Path + +from tui_gateway import wiki_api as wiki + + +def _make_wiki(): + d = tempfile.mkdtemp() + root = Path(d) + (root / "entities").mkdir() + (root / "concepts").mkdir() + return root + + +class TestScan: + def test_empty_dir(self): + root = _make_wiki() + result = wiki.wiki_scan(str(root)) + assert result["pages"] == [] + assert result["links"] == [] + + def test_missing_dir(self): + result = wiki.wiki_scan("/nonexistent/wiki/path") + assert result == {"pages": [], "links": []} + + def test_scan_entities_and_concepts(self): + root = _make_wiki() + (root / "entities" / "dflash-mlx.md").write_text( + "---\ntitle: dflash-mlx\ntype: entity\ntags: [optimization]\n---\n\nBody here. [[speculative-decoding]]\n", + encoding="utf-8", + ) + (root / "concepts" / "speculative-decoding.md").write_text( + "---\ntitle: Speculative Decoding\ntype: concept\n---\n\nConcept body.\n", + encoding="utf-8", + ) + + result = wiki.wiki_scan(str(root)) + pages = {p["id"]: p for p in result["pages"]} + assert len(pages) == 2 + assert pages["dflash-mlx"]["type"] == "entity" + assert pages["dflash-mlx"]["tags"] == ["optimization"] + assert pages["speculative-decoding"]["type"] == "concept" + + links = result["links"] + assert len(links) == 1 + assert links[0]["source"] == "dflash-mlx" + assert links[0]["target"] == "speculative-decoding" + assert links[0]["type"] == "wikilink" + + def test_no_frontmatter(self): + root = _make_wiki() + (root / "entities" / "plain.md").write_text( + "No frontmatter. [[other]]", encoding="utf-8" + ) + result = wiki.wiki_scan(str(root)) + assert len(result["pages"]) == 1 + assert result["pages"][0]["title"] == "plain" + # No frontmatter -> the default page type is "concept". + assert result["pages"][0]["type"] == "concept" + assert result["pages"][0]["tags"] == [] + + +class TestRootPages: + def test_root_level_pages_scanned(self): + root = _make_wiki() + (root / "log.md").write_text( + "# Wiki Log\n\nRecent changes. [[dflash-mlx]]\n", encoding="utf-8" + ) + (root / "index.md").write_text( + "---\ntitle: Index\ntype: index\n---\n\n[[dflash-mlx]]\n", encoding="utf-8" + ) + (root / "entities" / "dflash-mlx.md").write_text( + "---\ntitle: dflash-mlx\ntype: entity\n---\n\nBody.\n", encoding="utf-8" + ) + + result = wiki.wiki_scan(str(root)) + pages = {p["id"]: p for p in result["pages"]} + assert "log" in pages + assert "index" in pages + # Root pages live at the wiki root — no subdir prefix. + assert pages["log"]["path"] == "log.md" + # No frontmatter type on a root page -> "meta"; explicit type wins. + assert pages["log"]["type"] == "meta" + assert pages["index"]["type"] == "index" + + # Root pages participate in the link graph. + link_pairs = {(l["source"], l["target"]) for l in result["links"]} + assert ("log", "dflash-mlx") in link_pairs + assert ("index", "dflash-mlx") in link_pairs + + def test_root_page_readable_via_wiki_page(self): + root = _make_wiki() + (root / "log.md").write_text("# Log\n\nEntries.\n", encoding="utf-8") + result = wiki.wiki_page("log.md", str(root)) + assert result is not None + assert result["path"] == "log.md" + assert "Entries." in result["body"] + + def test_non_markdown_root_files_ignored(self): + root = _make_wiki() + (root / "taxonomy.yaml").write_text("categories: {}\n", encoding="utf-8") + (root / "notes.txt").write_text("not a page", encoding="utf-8") + result = wiki.wiki_scan(str(root)) + assert result["pages"] == [] + + +class TestPage: + def test_read_page(self): + root = _make_wiki() + (root / "entities" / "swiftlm.md").write_text( + "---\ntitle: SwiftLM\ntype: entity\n---\n\nBenchmark suite.\n", + encoding="utf-8", + ) + result = wiki.wiki_page("entities/swiftlm.md", str(root)) + assert result is not None + assert result["path"] == "entities/swiftlm.md" + assert result["frontmatter"]["title"] == "SwiftLM" + assert result["body"].strip() == "Benchmark suite." + + def test_page_not_found(self): + root = _make_wiki() + assert wiki.wiki_page("entities/missing.md", str(root)) is None + + def test_path_traversal_blocked(self): + root = _make_wiki() + # Try to escape the wiki root + assert wiki.wiki_page("../outside.md", str(root)) is None + + def test_absolute_path_rejected(self): + root = _make_wiki() + assert wiki.wiki_page("/etc/passwd", str(root)) is None + + +class TestFrontmatter: + def test_valid_frontmatter(self): + text = "---\ntitle: Foo\ntype: concept\n---\n\nBody" + fm, body = wiki._parse_frontmatter(text) + assert fm["title"] == "Foo" + assert body.strip() == "Body" + + def test_no_frontmatter(self): + text = "Just markdown" + fm, body = wiki._parse_frontmatter(text) + assert fm == {} + assert body == "Just markdown" + + def test_invalid_yaml_treated_as_none(self): + text = "---\n[bad yaml\n---\n\nBody" + fm, body = wiki._parse_frontmatter(text) + assert fm == {} + assert body.strip() == "Body" + + +class TestUpdate: + def _seed_page(self, root, updated="2026-07-01T00:00:00Z"): + (root / "entities" / "dflash-mlx.md").write_text( + "---\n" + "title: dflash-mlx\n" + "type: entity\n" + "tags: [optimization]\n" + "tag_path:\n" + " - ml/inference\n" + f"updated: {updated}\n" + "custom: keepme\n" + "---\n" + "\n" + "Original body. [[speculative-decoding]]\n", + encoding="utf-8", + ) + + def test_create_new_page(self): + root = _make_wiki() + result = wiki.wiki_update( + "entities/new-page.md", + "\nFresh body.\n", + frontmatter={"title": "New Page", "type": "entity"}, + wiki_path=str(root), + ) + assert "error" not in result + assert result["frontmatter"]["title"] == "New Page" + assert result["updated"] != "" + # created is stamped on new pages + assert result["frontmatter"]["created"] == result["updated"] + # The file round-trips through the reader. + page = wiki.wiki_page("entities/new-page.md", str(root)) + assert page is not None + assert page["frontmatter"]["title"] == "New Page" + assert page["body"].strip() == "Fresh body." + + def test_update_preserves_frontmatter_when_omitted(self): + root = _make_wiki() + self._seed_page(root) + result = wiki.wiki_update( + "entities/dflash-mlx.md", "\nReplaced body.\n", wiki_path=str(root) + ) + assert "error" not in result + fm = result["frontmatter"] + assert fm["title"] == "dflash-mlx" + assert fm["custom"] == "keepme" + # Server bumps updated past the seeded value. + assert fm["updated"] != "2026-07-01T00:00:00Z" + page = wiki.wiki_page("entities/dflash-mlx.md", str(root)) + assert page["body"].strip() == "Replaced body." + + def test_frontmatter_replacement_drops_absent_keys(self): + root = _make_wiki() + self._seed_page(root) + result = wiki.wiki_update( + "entities/dflash-mlx.md", + "\nBody.\n", + frontmatter={"title": "Retitled", "type": "entity"}, + wiki_path=str(root), + ) + fm = result["frontmatter"] + assert fm["title"] == "Retitled" + assert "custom" not in fm # absent from the replacement → dropped + + def test_if_match_allows_write(self): + root = _make_wiki() + self._seed_page(root) + result = wiki.wiki_update( + "entities/dflash-mlx.md", + "\nNew body.\n", + if_match="2026-07-01T00:00:00Z", + wiki_path=str(root), + ) + assert "error" not in result + + def test_stale_if_match_conflicts_with_latest(self): + root = _make_wiki() + self._seed_page(root) + result = wiki.wiki_update( + "entities/dflash-mlx.md", + "\nClobber body.\n", + if_match="1999-01-01T00:00:00Z", + wiki_path=str(root), + ) + assert result.get("code") == "conflict" + assert result["latest"]["updated"] == "2026-07-01T00:00:00Z" + assert "Original body" in result["latest"]["body"] + # The file is untouched. + page = wiki.wiki_page("entities/dflash-mlx.md", str(root)) + assert "Original body" in page["body"] + + def test_force_bypasses_conflict(self): + root = _make_wiki() + self._seed_page(root) + result = wiki.wiki_update( + "entities/dflash-mlx.md", + "\nForced body.\n", + if_match="1999-01-01T00:00:00Z", + force=True, + wiki_path=str(root), + ) + assert "error" not in result + page = wiki.wiki_page("entities/dflash-mlx.md", str(root)) + assert "Forced body" in page["body"] + + def test_list_keys_survive_string_only_clients(self): + root = _make_wiki() + self._seed_page(root) + # A string-only client round-trips tag_path as "" — the current list + # must be preserved, not wiped. + result = wiki.wiki_update( + "entities/dflash-mlx.md", + "\nBody.\n", + frontmatter={"title": "dflash-mlx", "type": "entity", "tag_path": ""}, + wiki_path=str(root), + ) + assert result["frontmatter"]["tag_path"] == ["ml/inference"] + # A non-empty scalar is comma-split into a list. + result = wiki.wiki_update( + "entities/dflash-mlx.md", + "\nBody.\n", + frontmatter={"title": "dflash-mlx", "tag_path": "a/b, c"}, + wiki_path=str(root), + ) + assert result["frontmatter"]["tag_path"] == ["a/b", "c"] + # And the written file parses back to a list. + page = wiki.wiki_page("entities/dflash-mlx.md", str(root)) + assert page["frontmatter"]["tag_path"] == ["a/b", "c"] + + def test_traversal_rejected(self): + root = _make_wiki() + result = wiki.wiki_update("../outside.md", "\nNope.\n", wiki_path=str(root)) + assert result.get("code") == "invalid" + assert not (root.parent / "outside.md").exists() + + def test_non_markdown_rejected(self): + root = _make_wiki() + result = wiki.wiki_update("entities/notes.txt", "\nNope.\n", wiki_path=str(root)) + assert result.get("code") == "invalid" + + def test_changeset_recorded(self): + root = _make_wiki() + self._seed_page(root) + wiki.wiki_update("entities/dflash-mlx.md", "\nTracked.\n", wiki_path=str(root)) + index = (root / "changesets" / "index.json") + assert index.exists() + import json + entries = json.loads(index.read_text(encoding="utf-8")) + assert len(entries) == 1 + assert entries[0]["action"] == "update" + assert entries[0]["page"] == "entities/dflash-mlx.md" + # The full changeset file carries the trigger. + cs = json.loads( + (root / "changesets" / f"{entries[0]['id']}.json").read_text(encoding="utf-8") + ) + assert cs["trigger"] == "manual" + + def test_serialized_lists_parse_back(self): + root = _make_wiki() + result = wiki.wiki_update( + "entities/lists.md", + "\nBody.\n", + frontmatter={ + "title": "Lists", + "tag_path": ["a/b", "c"], + "integration_links": ["github:org/repo#1"], + }, + wiki_path=str(root), + ) + assert "error" not in result + page = wiki.wiki_page("entities/lists.md", str(root)) + assert page["frontmatter"]["tag_path"] == ["a/b", "c"] + assert page["frontmatter"]["integration_links"] == ["github:org/repo#1"]

diff --git hermes-agent/tests/tui_gateway/test_wiki_nested_scan.py harness/tests/tui_gateway/test_wiki_nested_scan.py new file mode 100644 index 0000000000000000000000000000000000000000..8cb76bbd4eb07ab395d30242917c37af7bbc288c --- /dev/null +++ harness/tests/tui_gateway/test_wiki_nested_scan.py @@ -0,0 +1,235 @@ +"""Contract tests for wiki_scan over NESTED wiki layouts and typed relations. + +These lock four behaviours that a real two-level wiki depends on. Each one was +a live bug: a 971-page wiki scanned as 5 pages, every nested page 404'd in +wiki.page, every page reported zero tags, and every structural edge arrived +untyped. + +The assertions are invariants (a scanned page's advertised path must load; a +declared tag axis must survive to the client), not frozen counts, so they stay +meaningful as the wiki grows. +""" + +from tui_gateway.wiki_api import wiki_page, wiki_scan + + +def _write(path, text): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + + +def _nested_wiki(tmp_path): + """A wiki shaped like a real controlled-taxonomy corpus: two levels deep, + nested `tags:` mapping, typed relations, and path-style wikilinks.""" + _write( + tmp_path / "entities" / "org" / "stableenrich.md", + """--- +title: "StableEnrich" +type: org +tags: + protocol: [mpp, x402] + chain: [base, solana] + maturity: deployed +--- + +## Relations + +<!-- RELATIONS:x402 — GENERATED, do not hand-edit --> +- settles_through: [[coinbase]] +- implements: [[x402]] +- deployed_on: [[base]] +<!-- /RELATIONS:x402 --> + +## Links + +- [[queries/x402-market]] +""", + ) + _write( + tmp_path / "entities" / "org" / "coinbase.md", + """--- +title: "Coinbase" +type: org +tags: + protocol: [x402] + maturity: deployed +--- +Facilitator. +""", + ) + _write( + tmp_path / "entities" / "protocol" / "x402.md", + """--- +title: "x402" +type: protocol +tags: + protocol: [x402] +--- +Protocol. +""", + ) + _write( + tmp_path / "entities" / "chain" / "base.md", + """--- +title: "Base" +type: chain +tags: + chain: [base] +--- +Chain. +""", + ) + _write( + tmp_path / "queries" / "x402-market.md", + """--- +title: "x402 Market" +type: query +--- +Projection. +""", + ) + return tmp_path + + +def test_scan_finds_pages_nested_more_than_one_level_deep(tmp_path): + """Pages under entities/<bucket>/ must be scanned. + + Regression: _iter_page_files used iterdir() (non-recursive). Since nothing + lives directly in entities/, every entity page was invisible and a large + wiki scanned as only its root + queries files. + """ + wiki = _nested_wiki(tmp_path) + ids = {p["id"] for p in wiki_scan(str(wiki))["pages"]} + assert {"stableenrich", "coinbase", "x402", "base", "x402-market"} <= ids + + +def test_every_scanned_path_is_loadable_by_wiki_page(tmp_path): + """The path wiki.scan advertises must be the path wiki.page accepts. + + Regression: rel_path joined only the top-level bucket, so a file at + entities/org/foo.md was advertised as entities/foo.md and the client's + follow-up wiki.page call 404'd ("Failed to load page"). + """ + wiki = _nested_wiki(tmp_path) + pages = wiki_scan(str(wiki))["pages"] + assert pages, "scan returned no pages" + unloadable = [p["path"] for p in pages if wiki_page(p["path"], str(wiki)) is None] + assert unloadable == [] + + +def test_nested_tags_mapping_reaches_the_client_as_axis_values(tmp_path): + """A nested `tags:` mapping must survive to the client. + + Regression: the flat frontmatter parser hoisted the children to top level + and left `tags` itself None, so every page reported tags == [] and clients + had nothing to colour or filter nodes by. + """ + wiki = _nested_wiki(tmp_path) + page = next(p for p in wiki_scan(str(wiki))["pages"] if p["id"] == "stableenrich") + assert page["tags"], "nested tags mapping produced no tags" + assert "protocol:x402" in page["tags"] + assert "protocol:mpp" in page["tags"] + assert "maturity:deployed" in page["tags"] + + +def test_flat_tag_shapes_still_supported(tmp_path): + """Nested-mapping support must not regress the simpler shapes.""" + _write( + tmp_path / "entities" / "bracketed.md", + '---\ntitle: B\ntype: concept\ntags: [ml, research]\n---\nbody\n', + ) + _write( + tmp_path / "entities" / "bare.md", + "---\ntitle: C\ntype: concept\ntags: one, two\n---\nbody\n", + ) + pages = {p["id"]: p for p in wiki_scan(str(tmp_path))["pages"]} + assert pages["bracketed"]["tags"] == ["ml", "research"] + assert pages["bare"]["tags"] == ["one", "two"] + + +def test_typed_relations_are_emitted_with_their_predicate_as_edge_type(tmp_path): + """Structural edges must carry their predicate, not a generic label. + + Regression: wiki_scan hardcoded type="wikilink" for every edge, so the + graph could not distinguish "settles through" from "mentions". + """ + wiki = _nested_wiki(tmp_path) + links = wiki_scan(str(wiki))["links"] + typed = {(l["source"], l["type"], l["target"]) for l in links} + assert ("stableenrich", "settles_through", "coinbase") in typed + assert ("stableenrich", "implements", "x402") in typed + assert ("stableenrich", "deployed_on", "base") in typed + + +def test_typed_relation_is_not_also_duplicated_as_a_plain_wikilink(tmp_path): + """One relation is one edge: the predicate edge, not predicate + wikilink.""" + wiki = _nested_wiki(tmp_path) + links = wiki_scan(str(wiki))["links"] + coinbase_edges = [ + l for l in links if l["source"] == "stableenrich" and l["target"] == "coinbase" + ] + assert len(coinbase_edges) == 1 + assert coinbase_edges[0]["type"] == "settles_through" + + +def test_path_style_wikilink_targets_resolve_to_their_page(tmp_path): + """[[queries/x402-market]] must resolve to the x402-market page. + + Regression: targets were matched only against bare page ids, so every + path-style link resolved to nothing and vanished from the graph. + """ + wiki = _nested_wiki(tmp_path) + links = wiki_scan(str(wiki))["links"] + assert any( + l["source"] == "stableenrich" and l["target"] == "x402-market" for l in links + ) + + +def test_links_never_reference_a_nonexistent_page(tmp_path): + """Every edge endpoint must be a real scanned page (no dangling edges).""" + wiki = _nested_wiki(tmp_path) + result = wiki_scan(str(wiki)) + ids = {p["id"] for p in result["pages"]} + for link in result["links"]: + assert link["source"] in ids + assert link["target"] in ids + + +def test_unresolvable_relation_target_is_dropped_not_dangling(tmp_path): + """A relation naming a page that doesn't exist yields no edge.""" + _write( + tmp_path / "entities" / "org" / "solo.md", + """--- +title: Solo +type: org +--- + +## Relations + +<!-- RELATIONS:x402 — GENERATED --> +- settles_through: [[ghost-facilitator]] +<!-- /RELATIONS:x402 --> +""", + ) + links = wiki_scan(str(tmp_path))["links"] + assert not [l for l in links if l["target"] == "ghost-facilitator"] + + +def test_self_referencing_relation_is_not_emitted(tmp_path): + """A page relating to itself must not produce a self-loop edge.""" + _write( + tmp_path / "entities" / "org" / "selfy.md", + """--- +title: Selfy +type: org +--- + +## Relations + +<!-- RELATIONS:x402 — GENERATED --> +- implements: [[selfy]] +<!-- /RELATIONS:x402 --> +""", + ) + links = wiki_scan(str(tmp_path))["links"] + assert not [l for l in links if l["source"] == l["target"]]

diff --git hermes-agent/tui_gateway/wiki_api.py harness/tui_gateway/wiki_api.py new file mode 100644 index 0000000000000000000000000000000000000000..8909d3b3fb64dfcb4132e620580b44f8cbcabd7e --- /dev/null +++ harness/tui_gateway/wiki_api.py @@ -0,0 +1,1021 @@ +""" +Wiki scanning API for the TUI gateway. + +Provides filesystem-level wiki introspection for native clients that +render graph views or page detail. Supports a wiki name (e.g. "d-inference") +that resolves to a path via ~/.hermes/wikis.yaml, falling back to +$WIKI_PATH or ~/wiki. + +Multi-wiki support via ~/.hermes/wikis.yaml registry. + +v2 (2026-06-12): Adds hierarchical taxonomy (tag_path), taxonomy tree serving, +and integration link expansion for project management systems. +""" +import logging +import os +import re +from datetime import datetime, timezone +from pathlib import Path +from typing import Optional +import yaml + +logger = logging.getLogger(__name__) + + +def _load_wiki_registry() -> dict: + """Load ~/.hermes/wikis.yaml, returning {name: path} dict. + Returns empty dict if file doesn't exist or is unparseable. + """ + registry_path = Path(os.path.expanduser("~/.hermes/wikis.yaml")) + if not registry_path.exists(): + return {} + try: + with open(registry_path, encoding="utf-8") as f: + data = yaml.safe_load(f) or {} + except Exception: + return {} + if not isinstance(data, dict): + return {} + wikis = data.get("wikis", {}) + if not isinstance(wikis, dict): + return {} + resolved = {} + for name, path in wikis.items(): + if isinstance(path, str): + resolved[str(name)] = os.path.expanduser(path) + return resolved + + +def resolve_wiki(name: Optional[str] = None) -> str: + """Resolve a wiki name to a filesystem path. + + Resolution order: + 1. If name matches a key in ~/.hermes/wikis.yaml -> use that path + 2. If name looks like a path (~ or / prefix) -> expand and use directly + 3. If name is None/empty -> use registry's 'default' key + 4. Fall back to $WIKI_PATH env var + 5. Final fallback: ~/wiki + """ + registry = _load_wiki_registry() + + if name: + # Try registry name match + if name in registry: + return registry[name] + # Try raw path + if name.startswith("~") or name.startswith("/"): + return os.path.expanduser(name) + + # No name or name not found - use default + if registry: + # Read raw YAML to get the default key + registry_path = Path(os.path.expanduser("~/.hermes/wikis.yaml")) + try: + with open(registry_path, encoding="utf-8") as f: + data = yaml.safe_load(f) or {} + default_name = data.get("default") + if default_name and default_name in registry: + return registry[default_name] + except Exception: + pass + + # Fallbacks + env = os.environ.get("WIKI_PATH", "") + if env: + return env + return os.path.expanduser("~/wiki") + + +def wiki_list() -> dict: + """Return list of available wikis from ~/.hermes/wikis.yaml.""" + registry = _load_wiki_registry() + wikis = [] + for name, path in registry.items(): + wikis.append({"name": name, "path": path}) + return {"wikis": wikis} + + +def _default_wiki_path() -> str: + return resolve_wiki(None) + + +def _parse_frontmatter(content: str) -> tuple[dict, str]: + """Parse YAML frontmatter — returns (metadata, body). + + Handles both simple key:value and multi-line YAML list fields + (tag_path, integration_links, sources). + """ + if not content.startswith("---"): + return {}, content + parts = content.split("---", 2) + if len(parts) < 3: + return {}, content + metadata = {} + current_key = None + current_list = None + # Nested mapping support: `tags:` with indented `key: value` children is a + # real YAML shape used by controlled-taxonomy wikis. The flat parser + # hoisted those children to top level and left `tags` itself None, so every + # page reported `tags: []` — the graph had nothing to colour nodes by and + # the taxonomy filter was empty. Children are now BOTH kept nested under + # their parent (as a dict) and flattened into "parent.child"/bare-key + # aliases, so existing readers of e.g. fm["maturity"] keep working. + nested_parent = None + nested_map: dict = {} + + def _flush_nested(): + nonlocal nested_parent, nested_map + if nested_parent and nested_map: + metadata[nested_parent] = dict(nested_map) + # flat list of "key:value" strings for clients expecting a sequence + metadata[f"{nested_parent}_flat"] = [ + f"{k}:{v}" for k, v in nested_map.items() + ] + nested_parent, nested_map = None, {} + + for raw_line in parts[1].split("\n"): + line = raw_line + # Indented `key: value` under a parent mapping (2+ spaces, not a list item) + if nested_parent and re.match(r"^\s+[^\s-][^:]*:", line): + k, v = line.split(":", 1) + k, v = k.strip(), v.strip() + if len(v) >= 2 and v[0] == v[-1] and v[0] in ('"', "'"): + v = v[1:-1] + nested_map[k] = v + # also expose bare key at top level (back-compat for fm["maturity"]) + metadata.setdefault(k, v) + continue + if nested_parent and not line.strip(): + continue + if nested_parent and not line.startswith((" ", "\t")): + _flush_nested() + + # Check for indented list item (YAML list) + if line.startswith(" - ") and current_key: + if current_list is None: + current_list = [] + value = line.strip()[2:].strip().strip('"').strip("'") + current_list.append(value) + continue + + # Flush previous key's list + if current_key and current_list is not None: + metadata[current_key] = current_list + current_key = None + current_list = None + + line = line.strip() + if not line: + current_key = None + current_list = None + continue + if ":" in line: + key, val = line.split(":", 1) + key = key.strip() + val = val.strip() + # strip outer quotes + if len(val) >= 2 and val[0] == val[-1] and val[0] in ('"', "'"): + val = val[1:-1] + if val: # scalar value + metadata[key] = val + else: + # An empty value opens either a YAML list ("- item" lines) or a + # nested mapping (" child: value" lines). We can't know which + # until the next line, so arm both and let the line shape decide. + current_key = key + current_list = None + nested_parent = key + nested_map = {} + + # Flush final key's list / nested mapping + if current_key and current_list is not None: + metadata[current_key] = current_list + _flush_nested() + + return metadata, parts[2] + + +def _extract_wikilinks(body: str) -> list[str]: + """Extract [[wikilinks]] from markdown body.""" + pattern = r'\[\[([^\]|]+)(?:\|[^\]]+)?\]\]' + matches = re.findall(pattern, body) + return [m.strip().lower().replace(" ", "-") for m in matches] + + +#: Typed relation predicates (wiki SCHEMA §3.3). Structural edges are written as +#: "- <predicate>: [[target]]" inside a generated RELATIONS fence, so the graph +#: can label an edge with what it MEANS instead of a generic "wikilink". +RELATION_PREDICATES = ( + "settles_through", + "implements", + "deployed_on", + "listed_in", + "operates", + "routes_to", +) + +_TYPED_RELATION_RX = re.compile( + r"^\s*-\s+(" + "|".join(RELATION_PREDICATES) + r")\s*:\s*\[\[([^\]|]+)(?:\|[^\]]+)?\]\]", + re.MULTILINE, +) + + +def _extract_typed_relations(body: str) -> list[tuple]: + """Yield (predicate, target) for typed relation lines in a page body.""" + return [ + (m.group(1), m.group(2).strip().lower().replace(" ", "-")) + for m in _TYPED_RELATION_RX.finditer(body) + ] + + +def _resolve_link_target(target: str, page_ids: set) -> Optional[str]: + """Resolve a wikilink target to a page id, or None if it names no page. + + Accepts bare slugs ("coinbase") and path-style targets + ("entities/protocol/mpp", "queries/x402-market"). Path-style links are + written by the projectors and by hand; matching them only against bare ids + meant they resolved to nothing and disappeared from the graph. + """ + if target in page_ids: + return target + tail = target.rstrip("/").rsplit("/", 1)[-1] + if tail.endswith(".md"): + tail = tail[:-3] + return tail if tail in page_ids else None + + +#: Content subdirectories scanned for wiki pages. Root-level *.md files +#: (index.md, log.md, ...) are scanned too — see _iter_page_files. +WIKI_SUBDIRS = ["entities", "concepts", "comparisons", "queries", "raw", + "projects", "goals", "life", "issues"] + + +def _iter_page_files(wiki: Path): + """Yield (subdir, file) for every wiki page markdown file. + + Covers the content subdirectories plus root-level pages (subdir "" — + e.g. index.md, log.md), which previously never appeared in wiki.scan + and were therefore invisible in graph clients. + + RECURSES into nested subdirectories. Real taxonomies are two levels deep + (``entities/org/foo.md``, ``entities/protocol/x402.md``, + ``events/capital/snapshot.md``), and a non-recursive ``iterdir()`` silently + dropped every one of them: a 971-page wiki scanned as 5 pages (3 root + 2 + queries) because nothing lives directly in ``entities/``. The reported + subdir stays the TOP-LEVEL bucket so taxonomy filters keep working, while + ``rel`` (added by the caller) carries the full relative path. + """ + for subdir in [""] + WIKI_SUBDIRS: + dir_path = wiki / subdir if subdir else wiki + if not dir_path.exists(): + continue + # Root level is deliberately NOT recursive: its children are the + # taxonomy buckets themselves, which are walked on their own pass. + files = sorted(dir_path.glob("*.md")) if not subdir else sorted(dir_path.rglob("*.md")) + for file in files: + if file.suffix != ".md" or not file.is_file(): + continue + yield subdir, file + + +def wiki_scan(wiki_path: Optional[str] = None) -> dict: + """Scan wiki directory and return graph structure.""" + wiki = Path(wiki_path or _default_wiki_path()) + if not wiki.exists(): + return {"pages": [], "links": []} + + pages: list[dict] = [] + page_ids: set[str] = set() + links: list[dict] = [] + + # First pass: collect all pages + for subdir, file in _iter_page_files(wiki): + try: + content = file.read_text(encoding="utf-8") + except Exception: + continue + fm, _ = _parse_frontmatter(content) + slug = file.stem + # Path must be the FULL path relative to the wiki root. Joining only the + # top-level bucket (f"{subdir}/{file.name}") advertised + # "entities/stableenrich.md" for a file at "entities/org/stableenrich.md", + # so every nested page 404'd in wiki.page ("Failed to load page") the + # moment the scan started recursing. + try: + rel_path = file.relative_to(wiki).as_posix() + except ValueError: + rel_path = f"{subdir}/{file.name}" if subdir else file.name + + # Parse tags. Two shapes are supported: + # tags: [a, b] → flat list (legacy / simple wikis) + # tags: → nested controlled taxonomy + # protocol: [x402] + # maturity: deployed + # The nested form is returned by _parse_frontmatter as a dict; flatten + # it to "axis:value" strings so clients get one uniform list to colour + # and filter by (previously this called .strip() on the dict's absence + # and every page reported no tags at all). + raw_tags = fm.get("tags", "") + tags: list[str] = [] + if isinstance(raw_tags, dict): + for axis, val in raw_tags.items(): + cleaned = str(val).strip().strip("[]").replace("'", "").replace('"', "") + for v in (x.strip() for x in cleaned.split(",")): + if v: + tags.append(f"{axis}:{v}") + elif isinstance(raw_tags, list): + tags = [str(t).strip() for t in raw_tags if str(t).strip()] + elif raw_tags: + cleaned = str(raw_tags).strip().strip("[]").replace("'", "").replace('"', "") + tags = [t.strip() for t in cleaned.split(",") if t.strip()] + + # Root-level pages (index/log) are meta pages unless frontmatter + # says otherwise; subdir pages keep the old "concept" default. + default_type = "meta" if not subdir else "concept" + + pages.append( + { + "id": slug, + "title": fm.get("title", slug), + "type": fm.get("type", default_type), + "tags": tags, + "tag_path": fm.get("tag_path", []) if isinstance(fm.get("tag_path"), list) else [], + "integration_links": fm.get("integration_links", []) if isinstance(fm.get("integration_links"), list) else [], + # Already parsed as a list key (LIST_FRONTMATTER_KEYS) and + # already written by the ingest skill — it was simply never + # forwarded, so page-level provenance sat on disk unreadable + # by any client. Forwarding it is the whole fix. + "sources": fm.get("sources", []) if isinstance(fm.get("sources"), list) else [], + "path": rel_path, + "created": fm.get("created", ""), + "updated": fm.get("updated", ""), + "confidence": fm.get("confidence", ""), + "contested": fm.get("contested", "").lower() == "true", + } + ) + page_ids.add(slug) + + # Second pass: extract links. Typed relations (SCHEMA §3.3) are emitted + # with their PREDICATE as the edge type so clients can style/filter by + # relationship kind; everything else stays a plain "wikilink". + # + # Also resolves path-style targets ([[entities/protocol/mpp]]) to their + # slug, which previously matched no page id and so silently vanished from + # the graph — dropping every cross-protocol edge on the floor. + for _subdir, file in _iter_page_files(wiki): + try: + content = file.read_text(encoding="utf-8") + except Exception: + continue + _, body = _parse_frontmatter(content) + slug = file.stem + typed: set[tuple] = set() + for pred, target in _extract_typed_relations(body): + resolved = _resolve_link_target(target, page_ids) + if resolved and resolved != slug: + typed.add((resolved, pred)) + links.append({"source": slug, "target": resolved, "type": pred}) + for target in _extract_wikilinks(body): + resolved = _resolve_link_target(target, page_ids) + if not resolved or resolved == slug: + continue + # don't duplicate an edge already emitted with its real predicate + if any((resolved, p) in typed for p in RELATION_PREDICATES): + continue + links.append({"source": slug, "target": resolved, "type": "wikilink"}) + + return {"pages": pages, "links": links} + + +def wiki_page(path: str, wiki_path: Optional[str] = None) -> Optional[dict]: + """Read a single wiki page by relative path (e.g. 'entities/dflash-mlx.md').""" + wiki = Path(wiki_path or _default_wiki_path()) + target = wiki / path + # Security: refuse to escape the wiki directory + try: + target = target.resolve() + wiki = wiki.resolve() + except Exception: + return None + if not str(target).startswith(str(wiki)): + return None + if not target.exists() or target.suffix != ".md": + return None + try: + content = target.read_text(encoding="utf-8") + except Exception: + return None + fm, body = _parse_frontmatter(content) + return {"frontmatter": fm, "body": body, "path": path} + + +#: Frontmatter keys that hold YAML lists. Clients whose frontmatter model is +#: string-only (Portal's [String: String]) round-trip these as empty strings; +#: an empty scalar for a list key therefore means "couldn't represent it" — +#: preserve the current list rather than wiping it. A non-empty scalar is +#: comma-split; a real list is used as-is. +LIST_FRONTMATTER_KEYS = {"tag_path", "integration_links", "sources"} + +#: Preferred key order when serializing frontmatter (rest alphabetical), so +#: hand-edited and client-written files produce stable, reviewable diffs. +_FRONTMATTER_KEY_ORDER = [ + "title", "type", "tags", "tag_path", "created", "updated", + "confidence", "contested", "integration_links", "sources", +] + + +def _serialize_frontmatter(meta: dict) -> str: + """Serialize a frontmatter dict back to YAML-ish text `_parse_frontmatter` + can read: scalars as `key: value`, lists as `key:` + ` - item` lines.""" + def sort_key(k: str): + return (_FRONTMATTER_KEY_ORDER.index(k) if k in _FRONTMATTER_KEY_ORDER + else len(_FRONTMATTER_KEY_ORDER), k) + + def scalar(v) -> str: + s = str(v) + if any(c in s for c in (":", "#", '"')) or s != s.strip(): + s = '"' + s.replace('"', '\\"') + '"' + return s + + lines = [] + for key in sorted(meta.keys(), key=sort_key): + value = meta[key] + if isinstance(value, list): + lines.append(f"{key}:") + lines.extend(f" - {scalar(item)}" for item in value) + else: + lines.append(f"{key}: {scalar(value)}") + return "\n".join(lines) + "\n" + + +def wiki_update( + path: str, + body: str, + frontmatter: Optional[dict] = None, + if_match: Optional[str] = None, + force: bool = False, + trigger: str = "manual", + source_events: Optional[list] = None, + summary: Optional[str] = None, + wiki_path: Optional[str] = None, +) -> dict: + """Write a wiki page (full replace) with optimistic concurrency. + + The one write method on the wiki surface — see the `wiki.update` + semantics in Portal's docs/rpc-reference.md. + + Args: + path: Page path relative to the wiki root (must end in .md and + resolve INSIDE the root — traversal is rejected). + body: FULL replacement markdown body (no patch mode). + frontmatter: When a dict, REPLACES the entire frontmatter block + (absent keys are dropped); when None, the existing frontmatter + is preserved. `updated` is always set server-side; `created` + is set on new pages. + if_match: Optimistic-concurrency precondition — the `updated` value + the client read at load. When it differs from the server's + current `updated`, the write is rejected with a conflict. + force: Bypass the if_match precondition ("save anyway"). + trigger: What kind of change this is. Previously hardcoded to + "manual" here, which made every write through this method + indistinguishable — an automated ingest and a hand edit in the + desktop app landed in the timeline identically. Callers that + know better can now say so. + source_events: The ingestion events that caused this write, as + wiki-relative raw source paths. Recorded on the changeset as + provenance; omitted means unrecorded, which reads as *unknown*. + summary: Human-readable summary for the changeset. Defaults to a + generic one derived from the trigger. + wiki_path: Wiki root path override. + + Returns: + {"frontmatter": ..., "body": ..., "path": ..., "updated": ...} on + success, or {"error": msg, "code": "invalid"|"conflict", ...} — + conflicts also carry "latest": the server's current page. + """ + wiki = Path(wiki_path or _default_wiki_path()) + target = wiki / path + # Security: refuse to escape the wiki directory (mirrors wiki_page). + try: + target = target.resolve() + wiki = wiki.resolve() + except Exception: + return {"error": "path resolution failed", "code": "invalid"} + if not str(target).startswith(str(wiki)): + return {"error": f"path escapes wiki: {path}", "code": "invalid"} + if target.suffix != ".md": + return {"error": f"not a markdown page: {path}", "code": "invalid"} + + # Read the current page (if any) for the precondition + preservation. + exists = target.exists() + current_fm: dict = {} + if exists: + try: + current_fm, _ = _parse_frontmatter(target.read_text(encoding="utf-8")) + except Exception as e: + return {"error": f"could not read existing page: {e}", "code": "invalid"} + + # Optimistic concurrency: stale read → reject with the server's latest. + current_updated = current_fm.get("updated", "") + if exists and if_match is not None and not force and if_match != current_updated: + latest = wiki_page(path, str(wiki)) + if latest is not None: + latest["updated"] = current_updated + return { + "error": f"conflict: page changed since read (updated {current_updated!r})", + "code": "conflict", + "latest": latest, + } + + action = "update" if exists else "create" + + # Build the new frontmatter block. + new_fm = dict(current_fm) if frontmatter is None else dict(frontmatter) + # Coerce list-valued keys so string-only clients can't wipe them. + for key in LIST_FRONTMATTER_KEYS: + if key not in new_fm: + continue + value = new_fm[key] + if isinstance(value, list): + continue + if isinstance(value, str): + if not value.strip(): + # Empty scalar = "couldn't represent" → keep the current list. + if key in current_fm: + new_fm[key] = current_fm[key] + else: + del new_fm[key] + else: + new_fm[key] = [t.strip() for t in value.split(",") if t.strip()] + + now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + if not exists and not new_fm.get("created"): + new_fm["created"] = now + new_fm["updated"] = now # server-authoritative + + # Serialize: frontmatter block + body (exactly one blank line between). + normalized_body = body if body.startswith("\n") else "\n" + body + content = f"---\n{_serialize_frontmatter(new_fm)}---{normalized_body}" + + try: + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(content, encoding="utf-8") + except Exception as e: + return {"error": f"write failed: {e}", "code": "invalid"} + + # Record the change (git commit + changeset index) — best effort: the + # write itself already succeeded, so a capture hiccup only loses the + # audit entry, never the page. + try: + module = _load_wiki_changeset_module("wiki_capture_changeset") + module.wiki_capture_changeset( + page_path=path, + action=action, + summary=summary or f"{trigger} edit via wiki.update ({action})", + trigger=trigger, + source_events=source_events, + wiki_path=str(wiki), + ) + except Exception: + logger.warning("wiki.update: changeset capture failed for %s", path, exc_info=True) + + return {"frontmatter": new_fm, "body": body, "path": path, "updated": now} + + +def wiki_taxonomy(wiki_path: Optional[str] = None) -> Optional[dict]: + """Load and return the hierarchical taxonomy tree from taxonomy.yaml. + + Returns the full taxonomy dict with categories and nested children, + or None if taxonomy.yaml doesn't exist.""" + wiki = Path(wiki_path or _default_wiki_path()) + taxonomy_path = wiki / "taxonomy.yaml" + if not taxonomy_path.exists(): + return None + try: + with open(taxonomy_path, encoding="utf-8") as f: + return yaml.safe_load(f) or {} + except Exception: + return None + + +def wiki_flatten_taxonomy(wiki_path: Optional[str] = None) -> list[str]: + """Return a flat list of all valid taxonomy paths from taxonomy.yaml.""" + tree = wiki_taxonomy(wiki_path) + if not tree: + return [] + + def _flatten(categories, prefix=""): + paths = [] + for name, node in categories.items(): + if not isinstance(node, dict): + continue + path = f"{prefix}{name}" if prefix else name + paths.append(path) + if "children" in node and isinstance(node["children"], dict): + paths.extend(_flatten(node["children"], f"{path}/")) + return paths + + return sorted(_flatten(tree.get("categories", {}))) + + +def _load_wiki_changeset_module(required_attr: str): + """Load the wiki_changeset helper, preferring the repo-bundled copy. + + The previous sys.path approach let a STALE deployed copy in + ~/.hermes/scripts shadow the repo's updated module (and once cached in + sys.modules it kept winning) — surfacing to clients as + "cannot import name 'wiki_changeset_diff' from 'wiki_changeset'". + + Load by explicit file path via importlib instead: repo copy first, user + copy as fallback — and only accept a copy that actually has the symbol + the caller needs, so version skew degrades to the next candidate rather + than a confusing ImportError from the wrong file. + """ + import importlib.util + import os as _os + + repo_copy = _os.path.join( + _os.path.dirname(_os.path.dirname(_os.path.abspath(__file__))), + "scripts", "wiki_changeset.py", + ) + user_copy = _os.path.join( + _os.path.expanduser("~"), ".hermes", "scripts", "wiki_changeset.py" + ) + + tried = [] + for path in (repo_copy, user_copy): + if not _os.path.exists(path): + continue + tried.append(path) + try: + spec = importlib.util.spec_from_file_location("_hermes_wiki_changeset", path) + if spec is None or spec.loader is None: + continue + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + except Exception: + continue + if hasattr(module, required_attr): + return module + raise ImportError( + f"no wiki_changeset module providing {required_attr!r} found " + f"(tried: {tried or [repo_copy, user_copy]}) — " + "the gateway install may predate this feature" + ) + + +def wiki_changesets( + wiki_path: Optional[str] = None, + page: Optional[str] = None, + action: Optional[str] = None, + trigger: Optional[str] = None, + limit: int = 50, + offset: int = 0, + since: Optional[str] = None, + until: Optional[str] = None, +) -> dict: + """Query wiki changesets (timeline view). + + Args: + wiki_path: Wiki root path override + page: Filter by page path + action: Filter by action ('create', 'update', 'archive', 'delete') + trigger: Filter by trigger ('ingest', 'query', 'lint', 'process-inbox') + limit: Max results (default 50) + offset: Pagination offset + since: ISO timestamp filter (after) + until: ISO timestamp filter (before) + + Returns: + {"changesets": [...], "total": N, "limit": L, "offset": O} + """ + module = _load_wiki_changeset_module("wiki_query_changesets") + return module.wiki_query_changesets( + wiki_path=wiki_path, + page=page, + action=action, + trigger=trigger, + limit=limit, + offset=offset, + since=since, + until=until, + ) + + +#: Subdirectory holding raw ingested sources. Each file in here IS an event: +#: immutable, path-identified, and already carrying its own url + ingest time. +RAW_SUBDIR = "raw" + + +def _parse_event_time(value: str) -> Optional[datetime]: + """Parse an ``ingested`` frontmatter value into an aware UTC datetime. + + ``ingested`` is written by whatever ingested the source — sometimes by hand — + so it is not reliably strict RFC3339. A bare ``datetime.isoformat()`` + (no zone), a space separator, or a plain date are all common. Each denotes a + real instant, so each should be parsed rather than treated as "no time". + + A value with no zone is read as UTC: a wiki timestamp nobody attached a zone + to is one nobody chose a zone for, and being off by an offset beats losing + the event. + + Returns None when the value genuinely isn't a time, which callers treat as + undated — never as "now", which would be inventing data. + """ + text = (value or "").strip() + if not text: + return None + # fromisoformat handles the space separator, microseconds, and offsets; + # "Z" only from 3.11, so normalize it for older interpreters. + candidate = text[:-1] + "+00:00" if text.endswith(("Z", "z")) else text + try: + parsed = datetime.fromisoformat(candidate) + except ValueError: + return None + return parsed.replace(tzinfo=timezone.utc) if parsed.tzinfo is None else parsed + + +def _normalize_event_time(value: str) -> str: + """Render an ``ingested`` value as strict RFC3339 UTC, or "" if unparseable. + + The wire contract is one format, so clients don't each have to re-derive + what a wiki might contain. An unparseable value becomes "" — the same thing + a missing field produces, which is what "undated" already means on the wire. + """ + parsed = _parse_event_time(value) + if parsed is None: + return "" + return parsed.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def wiki_events( + wiki_path: Optional[str] = None, + kind: Optional[str] = None, + limit: int = 200, + offset: int = 0, + since: Optional[str] = None, + until: Optional[str] = None, +) -> dict: + """The ingestion event log — every event that caused a wiki update. + + Reads the **materialized event log** (``changesets/events.json``), the + write-time store populated by ``wiki_record_event`` and + ``wiki_capture_changeset``. Events are captured when they happen, not + reconstructed by scanning ``raw/`` after the fact — the old derivation + assumed top-level ``.md`` files with frontmatter and silently returned + nothing for the JSON snapshots the real pipeline writes under + ``raw/snapshots/`` and ``raw/mpp/``. + + Each event reports the changesets it caused, so a client can navigate + event → changeset → page as well as the reverse. The raw-scan derivation + survives only as ``_wiki_events_from_raw_scan`` — a one-time backfill seed, + never the live read. + + Args: + wiki_path: Wiki root path override + kind: Filter by event kind (the recorded ``kind``). Kinds are defined + by ``type: event-type`` wiki pages, not by a fixed list here. + limit: Max results (default 200, max 1000) + offset: Pagination offset + since: ISO timestamp, only events at or after this + until: ISO timestamp, only events at or before this + + Returns: + {"events": [...], "total": N, "limit": L, "offset": O} + + Each event's ``timestamp`` is strict RFC3339 UTC, or ``""`` when no + time could be established. ``time_estimated`` is True when the + timestamp came from the record's ``recorded_at`` (when it was first + materialized) rather than the source's own ``ingested_at``. Bounds are + compared as instants. + """ + try: + module = _load_wiki_changeset_module("wiki_query_events") + except ImportError: + # A gateway whose bundled wiki_changeset predates the materialized log + # still has the derivable events in raw/ — fall back to the scan so an + # un-migrated wiki isn't blank. This is the ONLY live use of the scan. + logger.warning( + "wiki.events: materialized log unavailable, deriving from raw/ scan", + exc_info=True, + ) + return _wiki_events_from_raw_scan( + wiki_path=wiki_path, kind=kind, limit=limit, + offset=offset, since=since, until=until, + ) + wiki = Path(wiki_path or _default_wiki_path()) + return module.wiki_query_events( + wiki_path=str(wiki), kind=kind, limit=limit, + offset=offset, since=since, until=until, + ) + + +def _wiki_events_from_raw_scan( + wiki_path: Optional[str] = None, + kind: Optional[str] = None, + limit: int = 200, + offset: int = 0, + since: Optional[str] = None, + until: Optional[str] = None, +) -> dict: + """DERIVED event log — reconstruct events by scanning ``raw/*.md``. + + Retained only as the backfill seed and the fallback for a gateway whose + ``wiki_changeset`` module predates the materialized log. NOT the live read + path: it sees only top-level ``.md`` files with frontmatter and misses the + JSON snapshots in ``raw/`` subdirs that the real pipeline writes, which is + exactly the fragility this refactor removes. + """ + wiki = Path(wiki_path or _default_wiki_path()) + raw_dir = wiki / RAW_SUBDIR + if not raw_dir.exists(): + return {"events": [], "total": 0, "limit": limit, "offset": offset} + + # Window bounds as instants, parsed once. An unparseable bound is treated as + # absent rather than as an impossible one, so a malformed `since` widens the + # query instead of silently returning nothing. + since_dt = _parse_event_time(since or "") + until_dt = _parse_event_time(until or "") + + # Which changesets each event caused. Built once from the index rather than + # per event, so the join stays linear in changeset count. + caused: dict = {} + try: + module = _load_wiki_changeset_module("wiki_query_changesets") + # Pull the whole index: an event's effects can be arbitrarily far back + # in the timeline, so a windowed read would under-report them. + known = module.wiki_query_changesets(wiki_path=str(wiki), limit=1000) + for changeset in known.get("changesets", []): + for key in changeset.get("source_event_keys") or []: + caused.setdefault(key, []).append( + { + "id": changeset.get("id", ""), + "page": changeset.get("page", ""), + "title": changeset.get("title", ""), + "action": changeset.get("action", ""), + "timestamp": changeset.get("timestamp", ""), + } + ) + except Exception: + # A wiki whose gateway install predates changesets still has raw + # sources, and a log of events with no effects recorded beats no log. + logger.warning("wiki.events: changeset join unavailable", exc_info=True) + + events: list[dict] = [] + for file in sorted(raw_dir.iterdir()): + if file.suffix != ".md" or not file.is_file(): + continue + try: + fm, _ = _parse_frontmatter(file.read_text(encoding="utf-8")) + except Exception: + continue + key = f"{RAW_SUBDIR}/{file.name}" + # `ingested` is when the event happened, normalized to one wire format so + # a client isn't left guessing which of its several shapes it received. + # + # mtime is the fallback for a source written before the field existed — + # an event with no time at all can't be plotted, so a real observation + # beats none. But it is flagged rather than passed off as the event's + # own time: `git clone` rewrites every mtime to checkout time, which + # would otherwise pile a whole wiki's history onto one bogus instant. + # `time_estimated` is what lets the client draw it as estimated. + event_time = _parse_event_time(str(fm.get("ingested", ""))) + time_estimated = False + if event_time is None: + try: + event_time = datetime.fromtimestamp(file.stat().st_mtime, timezone.utc) + time_estimated = True + except OSError: + event_time = None + timestamp = ( + event_time.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + if event_time + else "" + ) + # The kind is the page's own declared type, matched against event-type + # pages client-side. Absent means undeclared, not "manual". + event_kind = str(fm.get("event_kind", "") or fm.get("type", "")).strip() + + if kind and event_kind != kind: + continue + # Compared as instants, not strings. Lexical comparison silently drops + # events that ARE in the window: "2026-07-20 12:00:00" sorts below + # "2026-07-20T00:00:00Z" because a space sorts below "T", and a date-only + # value sorts below every timestamp on its own day. It also keeps events + # that aren't, since a non-UTC offset doesn't sort by real time. + if since_dt and event_time and event_time < since_dt: + continue + if until_dt and event_time and event_time > until_dt: + continue + + events.append( + { + "key": key, + "kind": event_kind, + "title": str(fm.get("title", file.stem)), + "timestamp": timestamp, + # True when `timestamp` came from the file's mtime rather than + # from `ingested` — a real observation, but not the event's own + # time, and a client should say so rather than imply precision. + "time_estimated": time_estimated, + "source_url": str(fm.get("source_url", "")), + "sha256": str(fm.get("sha256", "")), + "changesets": caused.get(key, []), + } + ) + + # Newest first, matching the changeset timeline. Sorting the emitted strings + # is sound now that they're all one normalized format; it was not when the + # field passed through verbatim. An empty timestamp still sorts below every + # real one under reverse ordering, so a genuinely undated event lands at the + # end rather than being silently dated to now. + events.sort(key=lambda e: e["timestamp"], reverse=True) + + total = len(events) + window = events[offset : offset + min(max(limit, 0), 1000)] + return {"events": window, "total": total, "limit": limit, "offset": offset} + + +def wiki_changeset_diff(changeset_id: str, wiki_path: Optional[str] = None) -> dict: + """Return the unified git diff for one changeset (timeline detail view). + + Args: + changeset_id: Changeset id from wiki.changesets (e.g. '2026-06-28T140819-001') + wiki_path: Wiki root path override + + Returns: + {"diff": "<unified diff>", "changeset": {...}} or {"error": ...} + """ + module = _load_wiki_changeset_module("wiki_changeset_diff") + return module.wiki_changeset_diff(changeset_id, wiki_path=wiki_path) + + +def wiki_expand_links(page_slug: str, wiki_path: Optional[str] = None) -> dict: + """Expand integration_links for a wiki page into live status. + + Currently resolves GitHub and Linear links. Returns a dict + mapping each link to a status object. Other link types return + a 'pending' status with the original value. + + Example return: + {"github:hermes-agent#456": {"status": "merged", "title": "Fix wiki...", "url": "..."}} + """ + wiki = Path(wiki_path or _default_wiki_path()) + + # Find the page by slug (root-level pages like index/log included) + for subdir in [""] + WIKI_SUBDIRS: + file_path = (wiki / subdir if subdir else wiki) / f"{page_slug}.md" + if file_path.exists(): + break + else: + return {"error": f"Page '{page_slug}' not found"} + + try: + content = file_path.read_text(encoding="utf-8") + except Exception: + return {"error": f"Could not read '{page_slug}'"} + + fm, _ = _parse_frontmatter(content) + links = fm.get("integration_links", []) + if not isinstance(links, list): + return {} + + result = {} + for link in links: + if not isinstance(link, str) or ":" not in link: + continue + prefix, rest = link.split(":", 1) + prefix = prefix.lower() + + if prefix == "github": + # Parse org/repo#num + if "#" in rest: + repo_path, num = rest.rsplit("#", 1) + result[link] = { + "type": "github", + "repo": repo_path, + "number": num, + "url": f"https://github.com/{repo_path}/pull/{num}", + "status": "unknown", + "title": f"{repo_path}#{num}", + } + else: + result[link] = {"type": "github", "repo": rest, "status": "unknown", "title": rest} + elif prefix == "linear": + result[link] = { + "type": "linear", + "issue_id": rest, + "url": f"https://linear.app/issue/{rest}", + "status": "unknown", + "title": rest, + } + elif prefix == "notion": + result[link] = {"type": "notion", "url": rest, "status": "unknown", "title": "Notion page"} + elif prefix == "obsidian": + result[link] = {"type": "obsidian", "note": rest, "status": "unknown", "title": rest} + elif prefix == "slack": + result[link] = {"type": "slack", "channel_msg": rest, "status": "unknown", "title": rest} + else: + result[link] = {"type": prefix, "raw": rest, "status": "unknown", "title": f"{prefix}:{rest}"} + + return result

diff --git hermes-agent/docs/api/wiki-changesets.md harness/docs/api/wiki-changesets.md new file mode 100644 index 0000000000000000000000000000000000000000..cab2b40c749985e171379200adb8a8378ed2a60e --- /dev/null +++ harness/docs/api/wiki-changesets.md @@ -0,0 +1,339 @@ +# wiki.changesets — API Spec for HermesNative Timeline + +## Endpoint + +``` +JSON-RPC method: wiki.changesets +Gateway: tui_gateway/server.py (already wired, no server-side changes needed) +``` + +## Request + +```jsonc +{ + "method": "wiki.changesets", + "params": { + // ALL optional — omit for full timeline + "wiki": "main", // wiki name from wikis.yaml (omit for default) + "page": "entities/llama-cpp.md", // filter to one page + "action": "update", // "create" | "update" | "archive" | "delete" + "trigger": "ingest", // "ingest" | "query" | "lint" | "process-inbox" | "manual" + "limit": 50, // default 50, max 200 + "offset": 0, // pagination offset + "since": "2026-06-01T00:00:00Z", // ISO 8601, only after this + "until": "2026-06-28T00:00:00Z" // ISO 8601, only before this + } +} +``` + +All params are optional. Omit everything for the full timeline (newest first, 50 per page). + +## Response + +```jsonc +{ + "changesets": [ + { + "id": "2026-06-28T140819-001", // unique, sortable + "timestamp": "2026-06-28T14:08:19Z", // ISO 8601 UTC + "action": "update", // "create" | "update" | "archive" | "delete" + "page": "entities/hermes-agent.md", // relative path in wiki + "title": "Hermes Agent", // page title from frontmatter + "type": "entity", // page type from frontmatter + "summary": "Added speculative decoding benchmarks and updated to b4820", + "diff_stats": { + "lines_added": 45, + "lines_removed": 12 + }, + "trigger": "ingest", // what caused the change + "source": "raw/articles/llama-cpp-release.md", // LEGACY single source; prefer source_event_keys + "source_event_keys": [ // provenance: the events that caused this + "raw/articles/llama-cpp-release.md", + "raw/papers/spec-decoding.md" + ], + "git_commit": "218f565a", // short git hash (empty if no git) + "after_sha256": "a7185eefbeca4d2f..." // page content hash after change + } + // ... more changesets + ], + "total": 7, // total matching (for pagination) + "limit": 50, + "offset": 0 +} +``` + +## Page type values + +The `type` field is the `type:` frontmatter value from each page: + +| type | directory | description | +|------|-----------|-------------| +| `entity` | entities/ | person, org, model, product | +| `concept` | concepts/ | idea, technique, topic | +| `comparison` | comparisons/ | side-by-side analysis | +| `query` | queries/ | filed question/answer | +| `project` | projects/ | personal project | +| `goal` | goals/ | life/project goal | +| `life` | life/ | life tracking entry | +| `issue` | issues/ | problem/blocker/task | + +## Action values + +| action | meaning | +|--------|---------| +| `create` | new page written | +| `update` | existing page modified | +| `archive` | page moved to _archive/ | +| `delete` | page removed | + +## Trigger values + +These five are conventional, not exhaustive. A trigger is a free string, and +what each one *is* is declared by a `type: event-type` wiki page — so adding an +ingestion source is a page commit, not a gateway or client release. Clients +resolve unrecognized values against those pages and fall back to a stable +derived presentation, so a new trigger is never a broken one. + +| trigger | meaning | +|---------|---------| +| `ingest` | from a source ingest (article, paper, URL) | +| `query` | filed from a query answer | +| `lint` | auto-fix during linting | +| `process-inbox` | from human inbox thoughts | +| `manual` | direct agent action | + +## Provenance — `source_event_keys` + +`source_event_keys` is the edge from a change back to the events that caused +it: an ordered list of wiki-relative raw source paths. Raw sources are +immutable files carrying their own `source_url` / `ingested` / `sha256`, so the +path is a stable identity and provenance needs no new storage. + +It is **always present**, so a reader never has to distinguish "field missing +because this changeset is old" from "field missing because nobody recorded it": + +| value | meaning | +|-------|---------| +| `["raw/a.md", "raw/b.md"]` | recorded — these events caused the change | +| `[]` | **unknown**. Not a claim that nothing caused it | + +Empty means *unrecorded*, deliberately not split into "no cause" vs "cause not +written down". Those two are indistinguishable on disk — the legacy `source` +defaults to `""` for both — and inferring which from the `trigger` would be +guesswork presented as fact. So the log refuses to guess. + +That makes `unknown` trustworthy only if it stays rare going forward, which is +why every write path now accepts provenance and the count of unknowns only +shrinks: `unknown` comes to mean precisely "predates provenance". + +Changesets written before this field existed are migrated **on read** — the +legacy `source` becomes the first key. An existing KB needs a newer gateway, +not a migration script. + +## Error response + +```jsonc +{ + "error": { + "code": 5055, + "message": "error description" + } +} +``` + +## Swift model (suggested) + +```swift +struct WikiChangeset: Codable, Identifiable { + let id: String + let timestamp: String + let action: String // "create" | "update" | "archive" | "delete" + let page: String // "entities/llama-cpp.md" + let title: String + let type: String // "entity" | "concept" | ... + let summary: String + let diffStats: DiffStats + let trigger: String + let source: String + /// Absent on pre-provenance payloads, so decode it optionally and treat + /// nil and [] identically — both are "unknown". + let sourceEventKeys: [String]? + let gitCommit: String + let afterSha256: String + + struct DiffStats: Codable { + let linesAdded: Int + let linesRemoved: Int + } + + enum CodingKeys: String, CodingKey { + case id, timestamp, action, page, title, type, summary, trigger, source + case diffStats = "diff_stats" + case sourceEventKeys = "source_event_keys" + case gitCommit = "git_commit" + case afterSha256 = "after_sha256" + } +} + +struct WikiChangesetsResponse: Codable { + let changesets: [WikiChangeset] + let total: Int + let limit: Int + let offset: Int +} +``` + +## Usage notes + +1. **Pagination**: use `limit` + `offset`. `total` tells you how many more exist. +2. **Filter by page**: pass `"page": "entities/llama-cpp.md"` to see the edit history of one page. +3. **Date range**: `since`/`until` for week/month views. ISO 8601 with Z suffix. +4. **Empty git_commit**: means the wiki isn't git-initialized or git wasn't available when the changeset was captured. Don't crash, just hide the commit link. +5. **Timeline order**: newest first by default (index.json is prepended). Already correct, no client-side sorting needed. + +## Example: "last 20 changes across the whole wiki" + +```jsonc +{ + "method": "wiki.changesets", + "params": { "limit": 20 } +} +``` + +## Example: "all changes to llama.cpp since June 1" + +```jsonc +{ + "method": "wiki.changesets", + "params": { + "page": "entities/llama-cpp.md", + "since": "2026-06-01T00:00:00Z", + "limit": 50 + } +} +``` + +## Example: "creates only, this week" + +```jsonc +{ + "method": "wiki.changesets", + "params": { + "action": "create", + "since": "2026-06-22T00:00:00Z", + "until": "2026-06-28T23:59:59Z" + } +} +``` +## wiki.changeset_diff — per-changeset unified diff + +```jsonc +{ + "method": "wiki.changeset_diff", + "params": { + "id": "2026-06-28T140819-001", // required — from wiki.changesets + "wiki": "main" // optional + } +} +``` + +Response: + +```jsonc +{ + "diff": "diff --git a/entities/x.md b/entities/x.md\n--- a/...\n+++ b/...\n@@ -4,3 +4,4 @@ ...\n+Line two added.\n", + "changeset": { /* same shape as a wiki.changesets entry */ } +} +``` + +Errors: `4001` bad/missing id · `5057` diff unavailable (changeset unknown, or +the wiki wasn't git-initialized when it was captured — the message says which) +· `5056` unexpected failure. Diffs are truncated at 200KB. + +## wiki.events — the ingestion event log + +Every event that caused a wiki update, newest first. This is a **join over data +already on disk**, not new storage: files under `raw/` are the events, and the +changeset index records which events caused which page writes. So it is +accurate for history that predates it. + +```jsonc +{ + "method": "wiki.events", + "params": { + // ALL optional + "wiki": "main", + "kind": "ingest", // filter by event kind (see below) + "limit": 200, // default 200, max 1000 + "offset": 0, + "since": "2026-06-01T00:00:00Z", + "until": "2026-06-28T00:00:00Z" + } +} +``` + +Response: + +```jsonc +{ + "events": [ + { + "key": "raw/articles/llama-cpp-release.md", // stable identity; the provenance join key + "kind": "ingest", // event_kind, else the page's type + "title": "llama.cpp b4820 release notes", + "timestamp": "2026-06-28T14:05:00Z", // `ingested`; "" when never recorded + "source_url": "https://github.com/ggml-org/llama.cpp/releases/tag/b4820", + "sha256": "a7185eefbeca4d2f...", + "changesets": [ // what this event caused + { + "id": "2026-06-28T140819-001", + "page": "entities/llama-cpp.md", + "title": "llama.cpp", + "action": "update", + "timestamp": "2026-06-28T14:08:19Z" + } + ] + } + ], + "total": 12, + "limit": 200, + "offset": 0 +} +``` + +`kind` is **not** a fixed enum. It's the raw source's `event_kind` (falling back +to its `type`), matched against `type: event-type` wiki pages that declare what +each kind is and how to draw it. An undeclared kind is still a valid kind. + +Notes: + +- **An event that caused nothing still appears** with `"changesets": []`. An + ingested source nobody synthesized from is exactly the gap worth seeing. +- **`timestamp` is never invented.** A source with no `ingested` reports `""` + and sorts last, rather than being silently dated to now. +- A wiki with no `raw/` returns an empty log, not an error. + +Errors: `5059` unexpected failure. + +## wiki.update — provenance on write + +`wiki.update` accepts three additional optional params so a write can declare +what caused it: + +```jsonc +{ + "method": "wiki.update", + "params": { + "path": "entities/llama-cpp.md", + "body": "...", + "trigger": "ingest", // default "manual" + "source_events": ["raw/articles/llama-cpp-release.md"], // provenance + "summary": "Added b4820 speculative-decoding benchmarks" + } +} +``` + +`trigger` was previously hardcoded to `"manual"` on this path, which made every +write through it indistinguishable in the timeline regardless of what made it. +Omitting `source_events` records `[]` — *unknown* — which is the honest result +for a hand edit that genuinely had no ingestion event.

diff --git hermes-agent/scripts/wiki_changeset.py harness/scripts/wiki_changeset.py new file mode 100644 index 0000000000000000000000000000000000000000..71b6f7d6fbfa563d52db66bf3a7f2e7e693915a4 --- /dev/null +++ harness/scripts/wiki_changeset.py @@ -0,0 +1,886 @@ +#!/usr/bin/env python3 +""" +Wiki changeset tracking module. + +Captures before/after state of wiki pages on every write, stores structured +changeset JSON files, and maintains a chronological index for fast timeline +queries. Integrates with git for raw diff storage. + +Storage layout: + wiki/changesets/ + ├── index.json # chronological list of all changesets + ├── 2026-06-28T143000-001.json # individual changeset files + └── ... + +Usage from wiki_api.py: + from scripts.wiki_changeset import wiki_capture_changeset, wiki_query_changesets + +Usage from the agent (after writing pages): + wiki_capture_changeset("entities/llama-cpp.md", "update", + "Added speculative decoding benchmarks", "ingest", + "raw/articles/source.md") +""" + +import hashlib +import json +import os +import subprocess +import time +from datetime import datetime, timezone +from pathlib import Path +from typing import Optional + + +def _wiki_root(wiki_path: Optional[str] = None) -> Path: + """Resolve wiki root path.""" + if wiki_path: + return Path(os.path.expanduser(wiki_path)) + return Path(os.path.expanduser(os.environ.get("WIKI_PATH", "~/wiki"))) + + +def _changesets_dir(wiki_path: Optional[str] = None) -> Path: + """Get or create the changesets directory.""" + d = _wiki_root(wiki_path) / "changesets" + d.mkdir(parents=True, exist_ok=True) + return d + + +def _index_path(wiki_path: Optional[str] = None) -> Path: + return _changesets_dir(wiki_path) / "index.json" + + +def _load_index(wiki_path: Optional[str] = None) -> list: + """Load the changeset index, or return empty list.""" + ip = _index_path(wiki_path) + if not ip.exists(): + return [] + try: + with open(ip, encoding="utf-8") as f: + data = json.load(f) + return data if isinstance(data, list) else [] + except (json.JSONDecodeError, OSError): + return [] + + +def _save_index(index: list, wiki_path: Optional[str] = None): + """Save the changeset index atomically.""" + ip = _index_path(wiki_path) + tmp = ip.with_suffix(".tmp") + with open(tmp, "w", encoding="utf-8") as f: + json.dump(index, f, indent=2, sort_keys=True) + os.replace(tmp, ip) + + +def _events_path(wiki_path: Optional[str] = None) -> Path: + """Location of the materialized event log (a machine-layer store). + + Lives alongside the changeset index so the two provenance halves — what + changed (index.json) and what caused it (events.json) — sit together and + are backed up, git-tracked, and reasoned about as one unit. + """ + return _changesets_dir(wiki_path) / "events.json" + + +def _load_events(wiki_path: Optional[str] = None) -> dict: + """Load the event store as a {key: record} map, or an empty map. + + Keyed by the event key (the raw source path / opaque source id) because + the key IS the event's identity: the same source ingested twice is one + event, not two, so a map dedupes for free and makes upsert O(1). + """ + ep = _events_path(wiki_path) + if not ep.exists(): + return {} + try: + with open(ep, encoding="utf-8") as f: + data = json.load(f) + return data if isinstance(data, dict) else {} + except (json.JSONDecodeError, OSError): + return {} + + +def _save_events(events: dict, wiki_path: Optional[str] = None): + """Persist the event store atomically (temp file + os.replace).""" + ep = _events_path(wiki_path) + tmp = ep.with_suffix(".tmp") + with open(tmp, "w", encoding="utf-8") as f: + json.dump(events, f, indent=2, sort_keys=True) + os.replace(tmp, ep) + + +def wiki_record_event( + key: str, + kind: Optional[str] = None, + source_url: Optional[str] = None, + sha256: Optional[str] = None, + ingested_at: Optional[str] = None, + trigger: Optional[str] = None, + wiki_path: Optional[str] = None, +) -> dict: + """Upsert an ingestion event into the materialized event log. + + An event is a thing the wiki ingested — a raw snapshot, an article, an + MPP dump — identified by ``key`` (its raw path or opaque source id). This + is the write-time counterpart to ``wiki_capture_changeset``: instead of + reconstructing events after the fact by scanning ``raw/`` (fragile — it + assumed top-level ``.md`` files with frontmatter, and missed the JSON + snapshots the real pipeline writes), the ingester declares each event as + it happens. + + ``key`` is the identity. The first write with a given key CREATES the + record; later writes are idempotent — they only fill in fields that were + previously empty, and never clobber a value already recorded. So one + snapshot that causes 40 page writes records ONE event (upserted 40 times), + not 40. Re-running a backfill is safe for the same reason. + + Fields other than ``key`` are optional because the write path often can't + infer them (an ``article:<hash>`` key carries no url on its own); an + ingester that knows them passes them, and a later call that learns them + fills the blanks. + + Returns the stored record, or ``{"error": ...}`` for a blank key. + """ + k = (key or "").strip() + if not k: + return {"error": "event key is required"} + + events = _load_events(wiki_path) + existing = events.get(k) + if not isinstance(existing, dict): + existing = {} + + def _pick(new, old): + # Idempotent fill-forward: a non-empty new value only when there is no + # non-empty old value. Keeps re-runs and re-emits from clobbering + # richer data recorded earlier. + new_s = new.strip() if isinstance(new, str) else (new or "") + old_s = old.strip() if isinstance(old, str) else (old or "") + return new_s if not old_s and new_s else (old or "") + + record = { + "key": k, + "kind": _pick(kind, existing.get("kind")), + "source_url": _pick(source_url, existing.get("source_url")), + "sha256": _pick(sha256, existing.get("sha256")), + "ingested_at": _pick(ingested_at, existing.get("ingested_at")), + "trigger": _pick(trigger, existing.get("trigger")), + # First time this key was recorded, so a genuinely undated event still + # has a real instant to sort by (distinct from ingested_at, which is + # when the SOURCE says it was ingested). + "recorded_at": existing.get("recorded_at") + or datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), + } + events[k] = record + _save_events(events, wiki_path) + return record + + +def _sha256_file(path: Path) -> str: + """Compute SHA256 of a file's contents.""" + h = hashlib.sha256() + with open(path, "rb") as f: + for chunk in iter(lambda: f.read(65536), b""): + h.update(chunk) + return h.hexdigest() + + +def _next_changeset_id(wiki_path: Optional[str] = None) -> str: + """Generate a unique changeset ID: ISO-timestamp-NNN.""" + ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H%M%S") + csd = _changesets_dir(wiki_path) + # Count existing changesets with this timestamp prefix + existing = list(csd.glob(f"{ts}-*.json")) + n = len(existing) + 1 + return f"{ts}-{n:03d}" + + +def _parse_frontmatter(content: str) -> tuple[dict, str]: + """Parse YAML frontmatter — returns (metadata, body).""" + if not content.startswith("---"): + return {}, content + parts = content.split("---", 2) + if len(parts) < 3: + return {}, content + metadata = {} + current_key = None + current_list = None + for line in parts[1].split("\n"): + if line.startswith(" - ") and current_key: + if current_list is None: + current_list = [] + value = line.strip()[2:].strip().strip('"').strip("'") + current_list.append(value) + continue + if current_key and current_list is not None: + metadata[current_key] = current_list + current_key = None + current_list = None + line = line.strip() + if not line: + current_key = None + current_list = None + continue + if ":" in line: + key, val = line.split(":", 1) + key = key.strip() + val = val.strip() + if len(val) >= 2 and val[0] == val[-1] and val[0] in ('"', "'"): + val = val[1:-1] + if val: + metadata[key] = val + else: + current_key = key + current_list = None + if current_key and current_list is not None: + metadata[current_key] = current_list + return metadata, parts[2] + + +def normalize_provenance( + source: str = "", + source_events: Optional[list] = None, +) -> list: + """Normalize the two provenance inputs into one ordered list of event keys. + + An event key is the wiki-relative path of the raw source that caused the + change (``raw/articles/llama-cpp-release.md``). Raw sources are immutable + files, so the path is a stable identity and the file itself already carries + the event's URL and ingest time — provenance needs no new storage, only the + edge. + + ``source`` is the legacy single-value form and is folded in as the first + key, so every existing caller keeps working and gains a list for free. + Blanks and duplicates are dropped: an empty list and an absent field mean + the same thing to a reader, and the client collapses both to ``unknown``. + """ + keys: list[str] = [] + candidates = [source] if isinstance(source, str) else [] + if isinstance(source_events, (list, tuple)): + candidates.extend(source_events) + elif isinstance(source_events, str): + # A single string where a list was expected — the shape a shell caller + # or a JSON-lite client most easily produces. Accept it rather than + # silently recording nothing. + candidates.append(source_events) + for candidate in candidates: + if not isinstance(candidate, str): + continue + key = candidate.strip() + if key and key not in keys: + keys.append(key) + return keys + + +def wiki_capture_changeset( + page_path: str, + action: str, + summary: str, + trigger: str = "manual", + source: str = "", + source_events: Optional[list] = None, + wiki_path: Optional[str] = None, +) -> dict: + """Capture a changeset for a wiki page modification. + + Captures the current state of the page (after the write), computes a + SHA256 hash, records git commit info, and stores a structured changeset + JSON file. Updates the chronological index. + + Args: + page_path: Relative path within wiki (e.g. 'entities/llama-cpp.md') + action: One of 'create', 'update', 'archive', 'delete' + summary: Human-readable summary of what changed + trigger: What triggered this change ('ingest', 'query', 'lint', + 'process-inbox', 'manual') + source: Legacy single source file (e.g. 'raw/articles/source.md'). + Folded into source_event_keys as the first entry. + source_events: The events that caused this change, as wiki-relative + raw source paths. A synthesis usually has several, which the + single ``source`` could never express. + wiki_path: Optional wiki root path override + + Returns: + The changeset dict that was stored, or error dict. + + Provenance is recorded, never inferred. A capture that declares no events + stores an empty ``source_event_keys``, which reads downstream as *unknown* + — "nobody recorded this" — not as "nothing caused it". Those two are + indistinguishable here, so the honest move is to refuse to guess and let + the count of unknowns shrink as callers start declaring. + """ + wiki = _wiki_root(wiki_path) + target = wiki / page_path + + # Resolve to prevent path traversal + try: + target = target.resolve() + wiki_resolved = wiki.resolve() + except Exception: + return {"error": "path resolution failed"} + + if not str(target).startswith(str(wiki_resolved)): + return {"error": f"path escapes wiki: {page_path}"} + + if action not in ("create", "update", "archive", "delete"): + return {"error": f"invalid action: {action}"} + + # Compute after-hash (page must exist unless it's a delete) + after_hash = "" + diff_stats = {"lines_added": 0, "lines_removed": 0} + page_title = "" + page_type = "" + + if target.exists() and target.suffix == ".md": + after_hash = _sha256_file(target) + try: + content = target.read_text(encoding="utf-8") + fm, _ = _parse_frontmatter(content) + page_title = fm.get("title", target.stem) + page_type = fm.get("type", "concept") + except Exception: + page_title = target.stem + elif action == "delete": + after_hash = "" + page_title = target.stem + else: + return {"error": f"page not found: {page_path}"} + + # Get git diff stats if git is available + git_commit = "" + git_root = wiki + while git_root != git_root.parent and not (git_root / ".git").exists(): + git_root = git_root.parent + + if (git_root / ".git").exists(): + try: + # Stage the file + subprocess.run( + ["git", "add", str(target)], + cwd=str(wiki), + capture_output=True, + timeout=10, + ) + + # Try to commit; if nothing staged, just grab HEAD + commit_msg = f"[{action}] {page_path}: {summary}"[:72] + result = subprocess.run( + ["git", "commit", "-m", commit_msg], + cwd=str(wiki), + capture_output=True, + text=True, + timeout=10, + ) + # Get HEAD hash (works whether or not we made a new commit) + hash_result = subprocess.run( + ["git", "rev-parse", "HEAD"], + cwd=str(wiki), + capture_output=True, + text=True, + timeout=5, + ) + git_commit = hash_result.stdout.strip()[:8] + + # Get diff stats: if we made a new commit, diff HEAD~1..HEAD; + # otherwise diff against the initial commit for baseline stats + if result.returncode == 0: + # New commit was created — diff against parent + diff_target = "HEAD~1" + else: + # Nothing to commit — file hasn't changed since last commit. + # Diff against the root commit to capture total file size. + root_hash = subprocess.run( + ["git", "rev-list", "--max-parents=0", "HEAD"], + cwd=str(wiki), + capture_output=True, + text=True, + timeout=5, + ).stdout.strip() + diff_target = root_hash if root_hash else "HEAD~1" + + diff_result = subprocess.run( + ["git", "diff", "--stat", diff_target, "HEAD", "--", str(target)], + cwd=str(wiki), + capture_output=True, + text=True, + timeout=5, + ) + stat_line = diff_result.stdout.strip() + if "insertion" in stat_line or "deletion" in stat_line: + import re + ins = re.search(r"(\d+)\s+insertion", stat_line) + dels = re.search(r"(\d+)\s+deletion", stat_line) + diff_stats["lines_added"] = int(ins.group(1)) if ins else 0 + diff_stats["lines_removed"] = int(dels.group(1)) if dels else 0 + except Exception: + pass + + # Build changeset + csid = _next_changeset_id(wiki_path) + now = datetime.now(timezone.utc) + + changeset = { + "id": csid, + "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), + "action": action, + "page": page_path, + "title": page_title, + "type": page_type, + "summary": summary, + "diff_stats": diff_stats, + "trigger": trigger, + "source": source, + # The provenance edge: which ingestion events caused this change. + # Always present (possibly empty) so a reader never has to distinguish + # "field missing because old" from "field missing because unrecorded". + "source_event_keys": normalize_provenance(source, source_events), + "git_commit": git_commit, + "after_sha256": after_hash, + } + + # Write changeset file + cs_file = _changesets_dir(wiki_path) / f"{csid}.json" + with open(cs_file, "w", encoding="utf-8") as f: + json.dump(changeset, f, indent=2) + + # Update index (prepend — newest first) + index = _load_index(wiki_path) + index_entry = { + "id": csid, + "timestamp": changeset["timestamp"], + "action": action, + "page": page_path, + "title": page_title, + "type": page_type, + "summary": summary, + "git_commit": git_commit, + } + index.insert(0, index_entry) + _save_index(index, wiki_path) + + # Emit an event record for each source that caused this change (create-if- + # absent, idempotent). This is the write-time capture that replaces the + # old after-the-fact raw/ scan: one snapshot causing 40 page writes + # upserts ONE event 40 times, not 40 events. The write path can't infer an + # event's kind/url from an opaque key, so it records only what it knows — + # the key and the trigger — and leaves an ingester (or a later call with + # --event-kind/--event-url) to enrich the rest. + for event_key in changeset["source_event_keys"]: + try: + wiki_record_event(event_key, trigger=trigger, wiki_path=wiki_path) + except Exception: + # A capture that succeeded must not fail because the event log + # hiccuped; the changeset is the source of truth and a backfill + # can reconstruct any missed event from it. + pass + + return changeset + + +def wiki_changeset_diff(changeset_id: str, wiki_path: Optional[str] = None) -> dict: + """Return the unified git diff for a single changeset. + + Uses the ``git_commit`` recorded at capture time: the capture path commits + each page write, so ``git show <commit> -- <page>`` reproduces exactly what + changed. Returns ``{"diff": <unified diff>, "changeset": {...}}`` or + ``{"error": ...}`` when the changeset is unknown or the wiki has no git + history (older captures with empty git_commit). + """ + csid = (changeset_id or "").strip() + # IDs are timestamp-shaped (e.g. 2026-06-28T140819-001); reject separators + # so a crafted id can't traverse out of the changesets dir. + if not csid or "/" in csid or "\\" in csid or ".." in csid: + return {"error": f"invalid changeset id: {changeset_id!r}"} + + cs_file = _changesets_dir(wiki_path) / f"{csid}.json" + if not cs_file.exists(): + return {"error": f"changeset not found: {csid}"} + try: + with open(cs_file, encoding="utf-8") as f: + changeset = _with_provenance(json.load(f)) + except (json.JSONDecodeError, OSError) as exc: + return {"error": f"changeset unreadable: {exc}"} + + commit = (changeset.get("git_commit") or "").strip() + page = changeset.get("page", "") + if not commit: + return { + "error": "no git commit recorded for this changeset " + "(wiki was not git-initialized at capture time)", + "changeset": changeset, + } + + wiki = _wiki_root(wiki_path) + try: + # --format="" drops the commit header, leaving just the diff body; + # scoping to the page keeps a multi-file commit focused. + result = subprocess.run( + ["git", "show", "--format=", "--no-color", commit, "--", page], + cwd=str(wiki), + capture_output=True, + text=True, + timeout=10, + ) + except (subprocess.SubprocessError, OSError) as exc: + return {"error": f"git show failed: {exc}", "changeset": changeset} + + if result.returncode != 0: + return { + "error": f"git show failed: {result.stderr.strip()[:200]}", + "changeset": changeset, + } + + diff = result.stdout + # Cap pathological diffs; the client renders line-by-line. + if len(diff) > 200_000: + diff = diff[:200_000] + "\n… (diff truncated at 200KB)\n" + + return {"diff": diff, "changeset": changeset} + + +def _with_provenance(changeset: dict) -> dict: + """Ensure a changeset read from disk carries ``source_event_keys``. + + Changesets written before provenance existed have a ``source`` string and + no list. Deriving the list on read migrates them in place, at no cost and + with no rewrite pass: a KB adopting this needs no migration script, only a + newer gateway. Everything with neither field reads as an empty list, which + the client renders as *unknown*. + """ + if not isinstance(changeset, dict): + return changeset + if isinstance(changeset.get("source_event_keys"), list): + return changeset + enriched = dict(changeset) + enriched["source_event_keys"] = normalize_provenance( + changeset.get("source", "") or "" + ) + return enriched + + +def wiki_query_changesets( + wiki_path: Optional[str] = None, + page: Optional[str] = None, + action: Optional[str] = None, + trigger: Optional[str] = None, + limit: int = 50, + offset: int = 0, + since: Optional[str] = None, + until: Optional[str] = None, +) -> dict: + """Query changesets with optional filters. + + Args: + wiki_path: Wiki root path override + page: Filter by page path (e.g. 'entities/llama-cpp.md') + action: Filter by action ('create', 'update', 'archive', 'delete') + trigger: Filter by trigger ('ingest', 'query', etc.) + limit: Max results (default 50, max 200) + offset: Pagination offset + since: ISO timestamp, only return changesets after this + until: ISO timestamp, only return changesets before this + + Returns: + {"changesets": [...], "total": N, "limit": L, "offset": O} + """ + index = _load_index(wiki_path) + + # Apply filters + filtered = [] + for entry in index: + if page and entry.get("page") != page: + continue + if action and entry.get("action") != action: + continue + if trigger: + # Trigger is only in the full changeset, not index. + # Load full changeset to check. + cs_file = _changesets_dir(wiki_path) / f"{entry['id']}.json" + if cs_file.exists(): + try: + with open(cs_file, encoding="utf-8") as f: + cs = json.load(f) + if cs.get("trigger") != trigger: + continue + except Exception: + continue + else: + continue + if since and entry.get("timestamp", "") < since: + continue + if until and entry.get("timestamp", "") > until: + continue + filtered.append(entry) + + total = len(filtered) + page_slice = filtered[offset : offset + min(limit, 200)] + + # Enrich with full changeset data + enriched = [] + for entry in page_slice: + cs_file = _changesets_dir(wiki_path) / f"{entry['id']}.json" + if cs_file.exists(): + try: + with open(cs_file, encoding="utf-8") as f: + enriched.append(_with_provenance(json.load(f))) + except Exception: + enriched.append(_with_provenance(entry)) + else: + enriched.append(_with_provenance(entry)) + + return { + "changesets": enriched, + "total": total, + "limit": limit, + "offset": offset, + } + + +def _parse_event_time(value: str) -> Optional["datetime"]: + """Parse a timestamp into an aware UTC datetime, or None if unparseable. + + Accepts strict RFC3339, a bare ``datetime.isoformat()`` (no zone), a space + separator, microseconds, a plain date, and a trailing ``Z``. A value with + no zone is read as UTC. This mirrors the tolerance the raw-scan path needed + for hand-written ``ingested`` fields, kept here so an event's timestamp is + normalized the same way whether it came from an ingester or a backfill. + """ + text = (value or "").strip() + if not text: + return None + candidate = text[:-1] + "+00:00" if text.endswith(("Z", "z")) else text + try: + parsed = datetime.fromisoformat(candidate) + except ValueError: + return None + return parsed.replace(tzinfo=timezone.utc) if parsed.tzinfo is None else parsed + + +def _normalize_event_time(value: str) -> str: + """Render a timestamp as strict RFC3339 UTC, or "" if unparseable.""" + parsed = _parse_event_time(value) + if parsed is None: + return "" + return parsed.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _caused_by_key(wiki_path: Optional[str] = None) -> dict: + """Map each event key → the list of changesets that declared it as a cause. + + Built once from the changeset index (paging past the 200-per-call cap) so + the event→changeset join stays linear. Each caused entry is a compact + changeset summary the client can navigate to. + """ + caused: dict = {} + offset = 0 + while True: + page = wiki_query_changesets(wiki_path=wiki_path, limit=200, offset=offset) + rows = page.get("changesets", []) + for changeset in rows: + for key in changeset.get("source_event_keys") or []: + caused.setdefault(key, []).append( + { + "id": changeset.get("id", ""), + "page": changeset.get("page", ""), + "title": changeset.get("title", ""), + "action": changeset.get("action", ""), + "timestamp": changeset.get("timestamp", ""), + } + ) + offset += len(rows) + if not rows or offset >= page.get("total", 0): + break + return caused + + +def wiki_query_events( + wiki_path: Optional[str] = None, + kind: Optional[str] = None, + limit: int = 200, + offset: int = 0, + since: Optional[str] = None, + until: Optional[str] = None, +) -> dict: + """Read the materialized event log, joined to the changesets each caused. + + This is the live read path: it serves the emitted event records (written + at tool-call time by ``wiki_record_event`` / ``wiki_capture_changeset``), + NOT a reconstruction by scanning ``raw/``. The raw scan is kept only as a + one-time backfill seed, because the real pipeline writes JSON snapshots in + subdirs that the scan never saw. + + Each event's ``timestamp`` is strict RFC3339 UTC, taken from + ``ingested_at`` when present and falling back to ``recorded_at`` (the + instant the event was first materialized), which is flagged with + ``time_estimated``. Window bounds are compared as instants. Newest first. + + Returns ``{"events": [...], "total": N, "limit": L, "offset": O}``. + """ + events_map = _load_events(wiki_path) + caused = _caused_by_key(wiki_path) + + since_dt = _parse_event_time(since or "") + until_dt = _parse_event_time(until or "") + + rows: list[dict] = [] + for key, rec in events_map.items(): + if not isinstance(rec, dict): + continue + # `kind` is optional enrichment, while every capture already carries + # the open-taxonomy wire value as `trigger`. Most writers only call + # capture, so fall back to that value or the UI would label correctly + # classified events "Unclassified" forever. Keep the stored fields + # separate so a later enrichment can still add a more specific kind. + event_kind = str(rec.get("kind") or rec.get("trigger") or "").strip() + if kind and event_kind != kind: + continue + + ingested = str(rec.get("ingested_at", "") or "") + event_dt = _parse_event_time(ingested) + time_estimated = False + if event_dt is None: + # Fall back to when the event was first recorded — a real instant, + # but not the source's own ingest time, so flag it as estimated. + event_dt = _parse_event_time(str(rec.get("recorded_at", "") or "")) + time_estimated = event_dt is not None + + if since_dt and event_dt and event_dt < since_dt: + continue + if until_dt and event_dt and event_dt > until_dt: + continue + + timestamp = ( + event_dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + if event_dt + else "" + ) + rows.append( + { + "key": rec.get("key", key), + "kind": event_kind, + "title": str(rec.get("title", "") or key), + "timestamp": timestamp, + "time_estimated": time_estimated, + "source_url": str(rec.get("source_url", "") or ""), + "sha256": str(rec.get("sha256", "") or ""), + "trigger": str(rec.get("trigger", "") or ""), + "changesets": caused.get(key, []), + } + ) + + # Newest first; an empty timestamp sorts last under reverse ordering, so a + # genuinely undated event lands at the end rather than dated to now. + rows.sort(key=lambda e: e["timestamp"], reverse=True) + + total = len(rows) + window = rows[offset : offset + min(max(limit, 0), 1000)] + return {"events": window, "total": total, "limit": limit, "offset": offset} + + +def wiki_backfill_events(wiki_path: Optional[str] = None) -> dict: + """Materialize an event record per distinct source key across all changesets. + + A one-time (idempotent) migration for a wiki whose events were previously + only *derivable* from the changeset index. Walks every changeset — paging + past the 200-per-call cap — collects each distinct ``source_event_keys`` + entry, and upserts it via ``wiki_record_event`` (create-if-absent, so + re-running never duplicates or clobbers). The write path can't infer a + key's kind/url, so backfilled records carry only key + trigger; an + ingester enriches the rest on its next real write. + + Returns ``{"scanned_changesets": N, "distinct_keys": K, "created": C, + "already_present": P}``. + """ + caused = _caused_by_key(wiki_path) + before = set(_load_events(wiki_path).keys()) + + scanned = 0 + offset = 0 + while True: + page = wiki_query_changesets(wiki_path=wiki_path, limit=200, offset=offset) + rows = page.get("changesets", []) + scanned += len(rows) + offset += len(rows) + if not rows or offset >= page.get("total", 0): + break + + created = 0 + already = 0 + for key, changesets in caused.items(): + # The earliest changeset that names this key gives a plausible ingest + # time when nothing better is known — better than the backfill's own + # clock, since it reflects when the effect actually landed. + stamps = [c.get("timestamp", "") for c in changesets if c.get("timestamp")] + earliest = min(stamps) if stamps else None + if key in before: + already += 1 + else: + created += 1 + wiki_record_event( + key, + ingested_at=earliest, + wiki_path=wiki_path, + ) + + return { + "scanned_changesets": scanned, + "distinct_keys": len(caused), + "created": created, + "already_present": already, + } + + +# ── CLI entry point for testing ──────────────────────────────────────── +if __name__ == "__main__": + import sys + if len(sys.argv) < 2: + print("usage: wiki-changeset.py <capture|query> [...]") + sys.exit(1) + + cmd = sys.argv[1] + if cmd == "capture": + # wiki-changeset.py capture <page_path> <action> <summary> [trigger] [source] + if len(sys.argv) < 4: + print("usage: wiki-changeset.py capture <page_path> <action> <summary> [trigger] [source]") + sys.exit(1) + result = wiki_capture_changeset( + page_path=sys.argv[2], + action=sys.argv[3], + summary=sys.argv[4] if len(sys.argv) > 4 else "", + trigger=sys.argv[5] if len(sys.argv) > 5 else "manual", + source=sys.argv[6] if len(sys.argv) > 6 else "", + # Remaining args are additional event keys, so a synthesis drawing + # on several sources can be captured in one call. + source_events=list(sys.argv[7:]), + ) + print(json.dumps(result, indent=2)) + elif cmd == "query": + result = wiki_query_changesets( + page=sys.argv[2] if len(sys.argv) > 2 else None, + limit=int(sys.argv[3]) if len(sys.argv) > 3 else 50, + ) + print(json.dumps(result, indent=2)) + elif cmd == "record": + # wiki-changeset.py record <key> [kind] [source_url] [sha256] [ingested_at] [trigger] + if len(sys.argv) < 3: + print("usage: wiki-changeset.py record <key> [kind] [source_url] [sha256] [ingested_at] [trigger]") + sys.exit(1) + result = wiki_record_event( + key=sys.argv[2], + kind=sys.argv[3] if len(sys.argv) > 3 else None, + source_url=sys.argv[4] if len(sys.argv) > 4 else None, + sha256=sys.argv[5] if len(sys.argv) > 5 else None, + ingested_at=sys.argv[6] if len(sys.argv) > 6 else None, + trigger=sys.argv[7] if len(sys.argv) > 7 else None, + ) + print(json.dumps(result, indent=2)) + elif cmd == "events": + result = wiki_query_events( + limit=int(sys.argv[2]) if len(sys.argv) > 2 else 200, + ) + print(json.dumps(result, indent=2)) + elif cmd == "backfill-events": + result = wiki_backfill_events() + print(json.dumps(result, indent=2)) + else: + print(f"unknown command: {cmd}") + sys.exit(1) \ No newline at end of file

diff --git hermes-agent/tests/gateway/test_wiki_watch.py harness/tests/gateway/test_wiki_watch.py new file mode 100644 index 0000000000000000000000000000000000000000..2785dbe0119a0836492b1bc3e71e570f2d33118f --- /dev/null +++ harness/tests/gateway/test_wiki_watch.py @@ -0,0 +1,131 @@ +"""The wiki watcher: programmatic changeset capture for out-of-band writes. + +The feed used to depend on the agent being PROMPTED to run the capture CLI +after editing pages with file tools — no enforcement on insertion, so pages +written by terminal sessions/humans/scripts never reached the desktop's wiki +timeline. These tests pin the sweeper's contract: baseline captures nothing, +out-of-band writes capture exactly once, already-captured writes dedup by +content hash, and half-written files wait for quiescence. +""" + +import json +import time +from pathlib import Path + +import pytest + +from tui_gateway.wiki_watch import _QUIESCENT_NS, sweep_once + + +class CaptureSpy: + def __init__(self): + self.calls = [] + + def __call__(self, **kwargs): + self.calls.append(kwargs) + return {"id": f"cs-{len(self.calls)}", "page": kwargs["page_path"], **kwargs} + + +def _settled_ns() -> int: + """A 'now' far enough in the future that every file is quiescent.""" + return time.time_ns() + 2 * _QUIESCENT_NS + + +@pytest.fixture() +def wiki(tmp_path: Path) -> Path: + (tmp_path / "entities").mkdir() + (tmp_path / "raw").mkdir() + (tmp_path / "changesets").mkdir() + (tmp_path / "entities" / "alpha.md").write_text("# Alpha\n", encoding="utf-8") + return tmp_path + + +def test_baseline_sweep_captures_nothing(wiki: Path): + spy = CaptureSpy() + snapshot, captured = sweep_once(None, wiki, spy, now_ns=_settled_ns()) + assert captured == [] + assert spy.calls == [] + assert "entities/alpha.md" in snapshot + + +def test_out_of_band_write_is_captured_once(wiki: Path): + spy = CaptureSpy() + snapshot, _ = sweep_once(None, wiki, spy, now_ns=_settled_ns()) + + (wiki / "entities" / "beta.md").write_text("# Beta\n", encoding="utf-8") + (wiki / "entities" / "alpha.md").write_text("# Alpha v2\n", encoding="utf-8") + + snapshot, captured = sweep_once(snapshot, wiki, spy, now_ns=_settled_ns()) + actions = {c["page_path"]: c["action"] for c in spy.calls} + assert actions == {"entities/beta.md": "create", "entities/alpha.md": "update"} + assert all(c["trigger"] == "auto" for c in spy.calls) + assert len(captured) == 2 + + # Steady state: nothing new, nothing captured. + _, captured = sweep_once(snapshot, wiki, spy, now_ns=_settled_ns()) + assert captured == [] + assert len(spy.calls) == 2 + + +def test_already_captured_write_dedups_by_hash(wiki: Path): + """A write that came WITH its own changeset (wiki.update / capture CLI) + must not be recorded twice.""" + spy = CaptureSpy() + snapshot, _ = sweep_once(None, wiki, spy, now_ns=_settled_ns()) + + page = wiki / "entities" / "alpha.md" + page.write_text("# Alpha captured elsewhere\n", encoding="utf-8") + + import hashlib + sha = hashlib.sha256(page.read_bytes()).hexdigest() + (wiki / "changesets" / "cs-external.json").write_text( + json.dumps({"id": "cs-external", "page": "entities/alpha.md", "after_sha256": sha}), + encoding="utf-8", + ) + (wiki / "changesets" / "index.json").write_text( + json.dumps([{"id": "cs-external", "page": "entities/alpha.md"}]), + encoding="utf-8", + ) + + _, captured = sweep_once(snapshot, wiki, spy, now_ns=_settled_ns()) + assert captured == [] + assert spy.calls == [] + + +def test_half_written_file_waits_for_quiescence(wiki: Path): + spy = CaptureSpy() + snapshot, _ = sweep_once(None, wiki, spy, now_ns=_settled_ns()) + + (wiki / "entities" / "fresh.md").write_text("# partial", encoding="utf-8") + + # Sweep with 'now' ~equal to the write time: inside the quiescence window. + snapshot, captured = sweep_once(snapshot, wiki, spy, now_ns=time.time_ns()) + assert captured == [] + assert spy.calls == [] + assert "entities/fresh.md" not in snapshot, "must be re-examined next sweep" + + # Once quiescent, it captures. + snapshot, captured = sweep_once(snapshot, wiki, spy, now_ns=_settled_ns()) + assert [c["page_path"] for c in spy.calls] == ["entities/fresh.md"] + assert "entities/fresh.md" in snapshot + + +def test_raw_and_changesets_dirs_are_not_pages(wiki: Path): + spy = CaptureSpy() + snapshot, _ = sweep_once(None, wiki, spy, now_ns=_settled_ns()) + + (wiki / "raw" / "event.md").write_text("source\n", encoding="utf-8") + (wiki / "changesets" / "junk.md").write_text("x\n", encoding="utf-8") + + _, captured = sweep_once(snapshot, wiki, spy, now_ns=_settled_ns()) + assert captured == [] + assert spy.calls == [] + + +def test_deleted_page_leaves_snapshot_without_capture(wiki: Path): + spy = CaptureSpy() + snapshot, _ = sweep_once(None, wiki, spy, now_ns=_settled_ns()) + (wiki / "entities" / "alpha.md").unlink() + snapshot, captured = sweep_once(snapshot, wiki, spy, now_ns=_settled_ns()) + assert captured == [] + assert "entities/alpha.md" not in snapshot

diff --git hermes-agent/tests/tui_gateway/test_changeset_diff.py harness/tests/tui_gateway/test_changeset_diff.py new file mode 100644 index 0000000000000000000000000000000000000000..42172267bc13083ef683973fa202bdcd8bcda3c8 --- /dev/null +++ harness/tests/tui_gateway/test_changeset_diff.py @@ -0,0 +1,71 @@ +"""Tests for wiki_changeset_diff — the timeline's git-style diff view.""" + +import subprocess +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "scripts")) +import wiki_changeset # noqa: E402 + + +@pytest.fixture +def git_wiki(tmp_path, monkeypatch): + """A git-initialized scratch wiki, with WIKI_PATH pointed at it.""" + wiki = tmp_path / "wiki" + (wiki / "entities").mkdir(parents=True) + subprocess.run(["git", "init", "-q"], cwd=wiki, check=True) + subprocess.run(["git", "config", "user.email", "t@t"], cwd=wiki, check=True) + subprocess.run(["git", "config", "user.name", "t"], cwd=wiki, check=True) + monkeypatch.setenv("WIKI_PATH", str(wiki)) + return wiki + + +def _write(wiki: Path, rel: str, text: str) -> None: + (wiki / rel).write_text(text, encoding="utf-8") + + +class TestChangesetDiff: + def test_update_diff_shows_added_line(self, git_wiki): + _write(git_wiki, "entities/x.md", "---\ntitle: X\n---\nLine one.\n") + wiki_changeset.wiki_capture_changeset("entities/x.md", "create", "initial") + _write(git_wiki, "entities/x.md", "---\ntitle: X\n---\nLine one.\nLine two.\n") + cs = wiki_changeset.wiki_capture_changeset("entities/x.md", "update", "add line") + + res = wiki_changeset.wiki_changeset_diff(cs["id"]) + assert "error" not in res + assert "+Line two." in res["diff"] + assert res["changeset"]["id"] == cs["id"] + + def test_create_diff_is_all_additions(self, git_wiki): + _write(git_wiki, "entities/y.md", "---\ntitle: Y\n---\nBody.\n") + cs = wiki_changeset.wiki_capture_changeset("entities/y.md", "create", "new page") + + res = wiki_changeset.wiki_changeset_diff(cs["id"]) + assert "error" not in res + assert "+Body." in res["diff"] + assert "new file mode" in res["diff"] + + def test_unknown_id(self, git_wiki): + res = wiki_changeset.wiki_changeset_diff("2099-01-01T000000-001") + assert "not found" in res["error"] + + def test_traversal_rejected(self, git_wiki): + for bad in ("../../../etc/passwd", "a/b", "a\\b", ".."): + res = wiki_changeset.wiki_changeset_diff(bad) + assert "invalid changeset id" in res["error"], bad + + def test_no_git_commit_recorded(self, git_wiki, tmp_path, monkeypatch): + # A wiki without git: capture records empty git_commit; diff must + # return a structured error (with the changeset) rather than crash. + bare = tmp_path / "bare-wiki" + (bare / "entities").mkdir(parents=True) + monkeypatch.setenv("WIKI_PATH", str(bare)) + _write(bare, "entities/z.md", "---\ntitle: Z\n---\nBody.\n") + cs = wiki_changeset.wiki_capture_changeset("entities/z.md", "create", "no-git page") + assert cs.get("git_commit", "") == "" + + res = wiki_changeset.wiki_changeset_diff(cs["id"]) + assert "no git commit" in res["error"] + assert res["changeset"]["id"] == cs["id"]

diff --git hermes-agent/tests/tui_gateway/test_wiki_changeset_loader.py harness/tests/tui_gateway/test_wiki_changeset_loader.py new file mode 100644 index 0000000000000000000000000000000000000000..fcc2472758534e7af77a4f84147cffa556318369 --- /dev/null +++ harness/tests/tui_gateway/test_wiki_changeset_loader.py @@ -0,0 +1,47 @@ +"""Regression: a stale ~/.hermes/scripts/wiki_changeset.py must not shadow +the repo-bundled module. + +Deployed gateways carry an old copy of wiki_changeset.py under +~/.hermes/scripts. The previous sys.path bootstrap let that stale copy win, +so new symbols (wiki_changeset_diff) raised +"cannot import name 'wiki_changeset_diff' from 'wiki_changeset'" even though +the repo copy had them. The loader now imports by explicit file path and +skips candidates lacking the required symbol. +""" + +import pytest + +from tui_gateway import wiki_api + + +@pytest.fixture +def stale_user_copy(tmp_path, monkeypatch): + """A fake $HOME whose ~/.hermes/scripts/wiki_changeset.py is outdated.""" + scripts = tmp_path / ".hermes" / "scripts" + scripts.mkdir(parents=True) + (scripts / "wiki_changeset.py").write_text( + "def wiki_query_changesets(**kw):\n" + " return {'changesets': [], 'stale': True, 'total': 0, 'limit': 50, 'offset': 0}\n", + encoding="utf-8", + ) + monkeypatch.setenv("HOME", str(tmp_path)) + return scripts + + +class TestLoaderPrefersRepoCopy: + def test_new_symbol_resolves_despite_stale_shadow(self, stale_user_copy): + module = wiki_api._load_wiki_changeset_module("wiki_changeset_diff") + assert hasattr(module, "wiki_changeset_diff") + + def test_changesets_not_hijacked_by_stale_copy(self, stale_user_copy, tmp_path, monkeypatch): + wiki = tmp_path / "wiki" + (wiki / "entities").mkdir(parents=True) + result = wiki_api.wiki_changesets(wiki_path=str(wiki)) + # The stale copy tags its result; the repo copy never does. + assert "stale" not in result + + def test_missing_symbol_reports_paths_tried(self, monkeypatch, tmp_path): + monkeypatch.setenv("HOME", str(tmp_path)) # no user copy at all + with pytest.raises(ImportError) as exc: + wiki_api._load_wiki_changeset_module("nonexistent_function_xyz") + assert "nonexistent_function_xyz" in str(exc.value)

diff --git hermes-agent/tests/tui_gateway/test_wiki_provenance.py harness/tests/tui_gateway/test_wiki_provenance.py new file mode 100644 index 0000000000000000000000000000000000000000..861dd9d10e98d3198ac63c368245c743278ac48a --- /dev/null +++ harness/tests/tui_gateway/test_wiki_provenance.py @@ -0,0 +1,525 @@ +"""Tests for wiki provenance — the event→changeset→page edge. + +Covers the three things that would fail silently: provenance normalization +(what "unknown" means and what it doesn't), the read-time migration that lets +an existing KB adopt this with no rewrite pass, and the wiki.events join. +""" + +import subprocess +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "scripts")) +import wiki_changeset # noqa: E402 + +from tui_gateway import wiki_api # noqa: E402 + + +@pytest.fixture +def git_wiki(tmp_path, monkeypatch): + """A git-initialized scratch wiki with raw/ present, WIKI_PATH pointed at it.""" + wiki = tmp_path / "wiki" + (wiki / "entities").mkdir(parents=True) + (wiki / "raw").mkdir(parents=True) + subprocess.run(["git", "init", "-q"], cwd=wiki, check=True) + subprocess.run(["git", "config", "user.email", "t@t"], cwd=wiki, check=True) + subprocess.run(["git", "config", "user.name", "t"], cwd=wiki, check=True) + monkeypatch.setenv("WIKI_PATH", str(wiki)) + return wiki + + +def _write(wiki: Path, rel: str, text: str) -> None: + target = wiki / rel + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(text, encoding="utf-8") + + +class TestNormalizeProvenance: + def test_no_inputs_is_empty_not_a_guess(self): + # Empty means unrecorded. The function never invents a source, because + # "nothing caused this" and "nobody wrote down what caused this" are + # indistinguishable here and guessing would make the log untrustworthy. + assert wiki_changeset.normalize_provenance() == [] + assert wiki_changeset.normalize_provenance("", []) == [] + assert wiki_changeset.normalize_provenance(" ", [" "]) == [] + + def test_legacy_single_source_becomes_the_first_key(self): + assert wiki_changeset.normalize_provenance("raw/a.md") == ["raw/a.md"] + + def test_multiple_events_keep_wire_order(self): + # The whole reason for a list: a synthesis draws on several sources, + # which the single `source` string could never express. + keys = wiki_changeset.normalize_provenance( + "raw/first.md", ["raw/second.md", "raw/third.md"] + ) + assert keys == ["raw/first.md", "raw/second.md", "raw/third.md"] + + def test_duplicates_collapse(self): + keys = wiki_changeset.normalize_provenance("raw/a.md", ["raw/a.md", "raw/b.md"]) + assert keys == ["raw/a.md", "raw/b.md"] + + def test_a_bare_string_where_a_list_was_expected_still_records(self): + # The shape a shell or JSON-lite caller most easily produces. Accepting + # it beats silently recording no provenance at all. + assert wiki_changeset.normalize_provenance("", "raw/a.md") == ["raw/a.md"] + + def test_non_string_entries_are_skipped_not_stringified(self): + keys = wiki_changeset.normalize_provenance("", ["raw/a.md", None, 7, {}]) + assert keys == ["raw/a.md"] + + +class TestCaptureRecordsProvenance: + def test_capture_stores_declared_events(self, git_wiki): + _write(git_wiki, "entities/x.md", "---\ntitle: X\n---\nBody.\n") + cs = wiki_changeset.wiki_capture_changeset( + "entities/x.md", "create", "synthesized", + trigger="ingest", + source_events=["raw/one.md", "raw/two.md"], + ) + assert cs["source_event_keys"] == ["raw/one.md", "raw/two.md"] + assert cs["trigger"] == "ingest" + + def test_capture_without_provenance_is_empty_not_absent(self, git_wiki): + # Always present, so a reader never distinguishes "field missing + # because old" from "field missing because unrecorded". + _write(git_wiki, "entities/y.md", "---\ntitle: Y\n---\nBody.\n") + cs = wiki_changeset.wiki_capture_changeset("entities/y.md", "create", "no source") + assert cs["source_event_keys"] == [] + + def test_query_round_trips_provenance(self, git_wiki): + _write(git_wiki, "entities/z.md", "---\ntitle: Z\n---\nBody.\n") + wiki_changeset.wiki_capture_changeset( + "entities/z.md", "create", "from a source", source_events=["raw/src.md"] + ) + result = wiki_changeset.wiki_query_changesets() + assert result["changesets"][0]["source_event_keys"] == ["raw/src.md"] + + +class TestReadTimeMigration: + """A KB adopting this needs a newer gateway, not a migration script.""" + + def test_a_pre_provenance_changeset_gains_keys_from_its_legacy_source(self, git_wiki): + _write(git_wiki, "entities/old.md", "---\ntitle: Old\n---\nBody.\n") + cs = wiki_changeset.wiki_capture_changeset( + "entities/old.md", "create", "legacy", source="raw/legacy.md" + ) + # Rewrite the stored file into the old shape — no source_event_keys. + import json + cs_file = git_wiki / "changesets" / f"{cs['id']}.json" + stored = json.loads(cs_file.read_text(encoding="utf-8")) + del stored["source_event_keys"] + cs_file.write_text(json.dumps(stored), encoding="utf-8") + + result = wiki_changeset.wiki_query_changesets() + assert result["changesets"][0]["source_event_keys"] == ["raw/legacy.md"] + + def test_a_changeset_with_neither_field_reads_as_unknown(self): + assert wiki_changeset._with_provenance({"id": "x"})["source_event_keys"] == [] + + def test_migration_never_overwrites_recorded_provenance(self): + recorded = {"id": "x", "source": "raw/a.md", "source_event_keys": ["raw/b.md"]} + assert wiki_changeset._with_provenance(recorded)["source_event_keys"] == ["raw/b.md"] + + +class TestRecordEvent: + """The materialized event log — events captured at write-time, not derived. + + The identity is the key; first write creates, later writes are idempotent + fill-forward (never clobbering a richer earlier value). + """ + + def test_record_creates_an_event(self, git_wiki): + rec = wiki_changeset.wiki_record_event( + "raw/snapshots/2026-08-18.json", + kind="snapshot", + source_url="https://example.invalid/s", + sha256="abc123", + ingested_at="2026-08-18T06:00:00Z", + trigger="ingest", + ) + assert rec["key"] == "raw/snapshots/2026-08-18.json" + assert rec["kind"] == "snapshot" + assert rec["source_url"] == "https://example.invalid/s" + assert rec["sha256"] == "abc123" + assert rec["ingested_at"] == "2026-08-18T06:00:00Z" + + def test_a_blank_key_is_refused(self, git_wiki): + assert "error" in wiki_changeset.wiki_record_event(" ") + + def test_second_write_with_same_key_is_idempotent(self, git_wiki): + wiki_changeset.wiki_record_event("raw/mpp/a.json", kind="mpp") + again = wiki_changeset.wiki_record_event("raw/mpp/a.json", kind="mpp") + events = wiki_changeset._load_events() + # One event, not two — the key is the identity. + assert list(events.keys()) == ["raw/mpp/a.json"] + assert again["kind"] == "mpp" + + def test_later_write_fills_blanks_but_never_clobbers(self, git_wiki): + # A capture records only the key + trigger; a later enriching call adds + # the url it learned. But a value already recorded is never overwritten. + wiki_changeset.wiki_record_event("raw/x.json", trigger="ingest") + enriched = wiki_changeset.wiki_record_event( + "raw/x.json", source_url="https://example.invalid/x", kind="snapshot" + ) + assert enriched["source_url"] == "https://example.invalid/x" + assert enriched["kind"] == "snapshot" + assert enriched["trigger"] == "ingest" + # A conflicting later kind must not clobber the recorded one. + again = wiki_changeset.wiki_record_event("raw/x.json", kind="OTHER") + assert again["kind"] == "snapshot" + + +class TestCaptureEmitsEvents: + """The write path emits events, so one snapshot → many pages is one event.""" + + def test_capture_materializes_an_event_per_source(self, git_wiki): + _write(git_wiki, "entities/x.md", "---\ntitle: X\n---\nBody.\n") + wiki_changeset.wiki_capture_changeset( + "entities/x.md", "create", "synthesized", + trigger="ingest", source_events=["raw/one.json", "raw/two.json"], + ) + events = wiki_changeset._load_events() + assert set(events.keys()) == {"raw/one.json", "raw/two.json"} + assert events["raw/one.json"]["trigger"] == "ingest" + + def test_one_snapshot_causing_many_pages_is_one_event(self, git_wiki): + # 3 page writes all caused by the same snapshot must record ONE event, + # not three — the upsert dedupes by key. + for i in range(3): + _write(git_wiki, f"entities/p{i}.md", f"---\ntitle: P{i}\n---\nB.\n") + wiki_changeset.wiki_capture_changeset( + f"entities/p{i}.md", "create", "from the snapshot", + trigger="ingest", source_events=["raw/snapshots/day.json"], + ) + events = wiki_changeset._load_events() + assert list(events.keys()) == ["raw/snapshots/day.json"] + + def test_capture_with_no_source_emits_no_event(self, git_wiki): + # A hand edit with no provenance records no event — the log stays a + # record of ingestion, not of every keystroke. + _write(git_wiki, "entities/y.md", "---\ntitle: Y\n---\nBody.\n") + wiki_changeset.wiki_capture_changeset("entities/y.md", "create", "no source") + assert wiki_changeset._load_events() == {} + + +class TestWikiEvents: + """wiki.events reads the MATERIALIZED log and joins the changesets caused. + + These pin the emitted-event contract (not the old raw/ derivation) while + preserving the invariants the derivation legitimately had: window/since/ + until as instants, RFC3339 normalization, newest-first sort, pagination, + kind filter, and the caused-nothing view. + """ + + def test_events_join_the_changesets_they_caused(self, git_wiki): + # An ingester records the event with its domain metadata (the write + # path can't infer a url from an opaque key)… + wiki_changeset.wiki_record_event( + "raw/article.json", kind="ingest", + source_url="https://example.invalid/x", sha256="abc123", + ingested_at="2026-07-01T10:00:00Z", + ) + # …and a capture declares it as the cause of a page write. + _write(git_wiki, "entities/x.md", "---\ntitle: X\n---\nBody.\n") + wiki_changeset.wiki_capture_changeset( + "entities/x.md", "create", "from the article", + trigger="ingest", source_events=["raw/article.json"], + ) + + result = wiki_api.wiki_events(wiki_path=str(git_wiki)) + assert result["total"] == 1 + event = result["events"][0] + assert event["key"] == "raw/article.json" + assert event["kind"] == "ingest" + assert event["source_url"] == "https://example.invalid/x" + assert event["sha256"] == "abc123" + # The edge the client navigates: event → the changesets it caused. + assert [c["page"] for c in event["changesets"]] == ["entities/x.md"] + + def test_capture_trigger_classifies_event_when_kind_was_not_enriched(self, git_wiki): + """The capture path knows the open event-kind wire value as `trigger`. + + Most wiki writers only call capture; they do not separately enrich the + materialized event. The read contract must therefore expose that trigger + as the event kind instead of rendering a classified source as blank. + """ + _write(git_wiki, "entities/x.md", "---\ntitle: X\n---\nBody.\n") + wiki_changeset.wiki_capture_changeset( + "entities/x.md", "create", "from Telegram", + trigger="telegram", source_events=["raw/telegram/item.md"], + ) + + event = wiki_api.wiki_events(wiki_path=str(git_wiki))["events"][0] + + assert event["kind"] == "telegram" + assert event["trigger"] == "telegram" + + def test_an_event_that_caused_nothing_still_appears(self, git_wiki): + # An ingester fetched a source but hasn't synthesized from it yet. It + # records the event directly (wiki_record_event is public/standalone), + # and it must show in the feed as caused-nothing — that's the work + # queue, not an error. + wiki_changeset.wiki_record_event( + "raw/unused.json", kind="ingest", ingested_at="2026-07-02T10:00:00Z" + ) + result = wiki_api.wiki_events(wiki_path=str(git_wiki)) + assert result["total"] == 1 + assert result["events"][0]["changesets"] == [] + + def test_events_are_newest_first(self, git_wiki): + for name, ingested in [ + ("old", "2026-07-01T10:00:00Z"), + ("new", "2026-07-03T10:00:00Z"), + ]: + wiki_changeset.wiki_record_event(f"raw/{name}.json", ingested_at=ingested) + events = wiki_api.wiki_events(wiki_path=str(git_wiki))["events"] + assert [e["key"] for e in events] == ["raw/new.json", "raw/old.json"] + + def test_an_event_with_no_ingested_at_falls_back_to_recorded_at_and_says_so( + self, git_wiki + ): + # An event captured with no source ingest time still has a real instant + # to plot — when it was first materialized — but that is NOT the source's + # own time, so it is flagged rather than presented as precise. + wiki_changeset.wiki_record_event("raw/nostamp.json") + event = wiki_api.wiki_events(wiki_path=str(git_wiki))["events"][0] + assert event["timestamp"] != "" + assert event["time_estimated"] is True + + def test_a_dated_event_is_not_marked_estimated(self, git_wiki): + wiki_changeset.wiki_record_event( + "raw/dated.json", ingested_at="2026-07-01T10:00:00Z" + ) + event = wiki_api.wiki_events(wiki_path=str(git_wiki))["events"][0] + assert event["timestamp"] == "2026-07-01T10:00:00Z" + assert event["time_estimated"] is False + + @pytest.mark.parametrize( + "written", + [ + "2026-07-20T12:00:00Z", # strict RFC3339 + "2026-07-20T12:00:00+00:00", # explicit UTC offset + "2026-07-20T12:00:00", # bare datetime.isoformat() + "2026-07-20T12:00:00.123456", # ...with microseconds + "2026-07-20 12:00:00", # space separator + " 2026-07-20T12:00:00Z ", # padded + ], + ) + def test_ingested_at_is_normalized_to_one_wire_format(self, git_wiki, written): + # ingested_at is supplied by whatever recorded the event, sometimes by + # hand, so it is not reliably strict RFC3339. Every one of these denotes + # the same instant and must reach the client as the same string. + wiki_changeset.wiki_record_event("raw/a.json", ingested_at=written) + event = wiki_api.wiki_events(wiki_path=str(git_wiki))["events"][0] + assert event["timestamp"] == "2026-07-20T12:00:00Z" + assert event["time_estimated"] is False + + def test_a_non_utc_offset_is_converted_not_truncated(self, git_wiki): + # 05:00-07:00 is 12:00Z. Dropping the offset would misplace the event. + wiki_changeset.wiki_record_event( + "raw/a.json", ingested_at="2026-07-20T05:00:00-07:00" + ) + event = wiki_api.wiki_events(wiki_path=str(git_wiki))["events"][0] + assert event["timestamp"] == "2026-07-20T12:00:00Z" + + def test_an_unparseable_ingested_at_falls_back_and_is_flagged(self, git_wiki): + # An unusable value tells us nothing about when the event happened, so + # recorded_at stands in and is marked estimated — never passed through. + wiki_changeset.wiki_record_event("raw/a.json", ingested_at="whenever") + event = wiki_api.wiki_events(wiki_path=str(git_wiki))["events"][0] + assert event["timestamp"] != "whenever" + assert event["time_estimated"] is True + assert wiki_api._parse_event_time(event["timestamp"]) is not None + + @pytest.mark.parametrize( + "written", + [ + "2026-07-20T12:00:00Z", + "2026-07-20 12:00:00", # a space sorts BELOW "T" lexically + "2026-07-20", # sorts below every stamp on its own day + "2026-07-20T05:00:00-07:00", # doesn't sort by real time at all + ], + ) + def test_the_window_keeps_events_inside_it_whatever_the_format( + self, git_wiki, written + ): + # Bounds are compared as INSTANTS, so an event genuinely inside the + # window is kept whatever shape its timestamp was written in. + wiki_changeset.wiki_record_event("raw/a.json", ingested_at=written) + result = wiki_api.wiki_events( + wiki_path=str(git_wiki), + since="2026-07-20T00:00:00Z", + until="2026-07-21T00:00:00Z", + ) + assert [e["key"] for e in result["events"]] == ["raw/a.json"] + + def test_the_window_still_excludes_events_outside_it(self, git_wiki): + for name, ingested in [ + ("before", "2019-01-01T00:00:00Z"), + ("inside", "2026-07-20T12:00:00Z"), + ("after", "2031-01-01T00:00:00Z"), + ]: + wiki_changeset.wiki_record_event(f"raw/{name}.json", ingested_at=ingested) + result = wiki_api.wiki_events( + wiki_path=str(git_wiki), + since="2026-07-20T00:00:00Z", + until="2026-07-21T00:00:00Z", + ) + assert [e["key"] for e in result["events"]] == ["raw/inside.json"] + + def test_a_non_utc_bound_is_compared_as_an_instant(self, git_wiki): + # until = 05:00-07:00 = 12:00Z, so a 13:00Z event is after it. + wiki_changeset.wiki_record_event( + "raw/late.json", ingested_at="2026-07-20T13:00:00Z" + ) + wiki_changeset.wiki_record_event( + "raw/early.json", ingested_at="2026-07-20T11:00:00Z" + ) + result = wiki_api.wiki_events( + wiki_path=str(git_wiki), until="2026-07-20T05:00:00-07:00" + ) + assert [e["key"] for e in result["events"]] == ["raw/early.json"] + + def test_an_unparseable_bound_widens_rather_than_empties(self, git_wiki): + wiki_changeset.wiki_record_event( + "raw/a.json", ingested_at="2026-07-20T12:00:00Z" + ) + result = wiki_api.wiki_events(wiki_path=str(git_wiki), since="garbage") + assert [e["key"] for e in result["events"]] == ["raw/a.json"] + + def test_an_estimated_time_still_participates_in_the_window(self, git_wiki): + # An event with no usable ingested_at gets recorded_at (now), a real + # time the window applies to like any other. + wiki_changeset.wiki_record_event("raw/a.json", ingested_at="nonsense") + past = wiki_api.wiki_events( + wiki_path=str(git_wiki), + since="2019-01-01T00:00:00Z", + until="2019-12-31T00:00:00Z", + ) + assert past["events"] == [] + allof = wiki_api.wiki_events(wiki_path=str(git_wiki)) + assert [e["key"] for e in allof["events"]] == ["raw/a.json"] + assert allof["events"][0]["time_estimated"] is True + + def test_kind_filter_matches_the_declared_kind(self, git_wiki): + wiki_changeset.wiki_record_event("raw/a.json", kind="github_pr") + wiki_changeset.wiki_record_event("raw/b.json", kind="ingest") + prs = wiki_api.wiki_events(wiki_path=str(git_wiki), kind="github_pr") + assert [e["key"] for e in prs["events"]] == ["raw/a.json"] + assert prs["events"][0]["kind"] == "github_pr" + + def test_a_wiki_without_events_has_an_empty_log_not_an_error(self, tmp_path): + bare = tmp_path / "bare" + (bare / "entities").mkdir(parents=True) + result = wiki_api.wiki_events(wiki_path=str(bare)) + assert result == {"events": [], "total": 0, "limit": 200, "offset": 0} + + def test_pagination_reports_the_full_total(self, git_wiki): + for i in range(5): + wiki_changeset.wiki_record_event( + f"raw/s{i}.json", ingested_at=f"2026-07-0{i + 1}T10:00:00Z" + ) + result = wiki_api.wiki_events(wiki_path=str(git_wiki), limit=2, offset=1) + assert result["total"] == 5 + assert len(result["events"]) == 2 + assert [e["key"] for e in result["events"]] == ["raw/s3.json", "raw/s2.json"] + + +class TestBackfillEvents: + """Backfill materializes one event per distinct source key, idempotently.""" + + def test_backfill_materializes_events_for_existing_changesets(self, git_wiki): + # Simulate a wiki whose changesets carry source keys but whose event + # log was never populated (the pre-refactor state): capture, then wipe + # the event store the capture emitted. + _write(git_wiki, "entities/x.md", "---\ntitle: X\n---\nBody.\n") + wiki_changeset.wiki_capture_changeset( + "entities/x.md", "create", "s", trigger="ingest", + source_events=["raw/one.json", "raw/two.json"], + ) + (git_wiki / "changesets" / "events.json").unlink() + + res = wiki_changeset.wiki_backfill_events(wiki_path=str(git_wiki)) + assert res["distinct_keys"] == 2 + assert res["created"] == 2 + assert set(wiki_changeset._load_events(str(git_wiki)).keys()) == { + "raw/one.json", "raw/two.json" + } + + def test_backfill_is_idempotent_on_rerun(self, git_wiki): + _write(git_wiki, "entities/x.md", "---\ntitle: X\n---\nBody.\n") + wiki_changeset.wiki_capture_changeset( + "entities/x.md", "create", "s", trigger="ingest", + source_events=["raw/one.json"], + ) + first = wiki_changeset.wiki_backfill_events(wiki_path=str(git_wiki)) + second = wiki_changeset.wiki_backfill_events(wiki_path=str(git_wiki)) + # The events already existed (emitted at capture), so re-running never + # creates duplicates. + assert second["created"] == 0 + assert second["already_present"] == first["distinct_keys"] + assert list(wiki_changeset._load_events(str(git_wiki)).keys()) == ["raw/one.json"] + + def test_backfill_dates_an_event_by_its_earliest_caused_changeset(self, git_wiki): + _write(git_wiki, "entities/x.md", "---\ntitle: X\n---\nBody.\n") + wiki_changeset.wiki_capture_changeset( + "entities/x.md", "create", "s", trigger="ingest", + source_events=["raw/one.json"], + ) + (git_wiki / "changesets" / "events.json").unlink() + wiki_changeset.wiki_backfill_events(wiki_path=str(git_wiki)) + # The backfilled event carries a real timestamp (the changeset's), so + # it's plottable rather than dumped at "now". + event = wiki_api.wiki_events(wiki_path=str(git_wiki))["events"][0] + assert event["timestamp"] != "" + + +class TestScanForwardsSources: + def test_page_level_sources_reach_the_client(self, git_wiki): + # Parsed as a list key and written by the ingest skill, but previously + # dropped from the payload — so page-level provenance was unreadable. + _write( + git_wiki, "entities/x.md", + "---\ntitle: X\nsources:\n - raw/a.md\n - raw/b.md\n---\nBody.\n", + ) + pages = wiki_api.wiki_scan(wiki_path=str(git_wiki))["pages"] + page = next(p for p in pages if p["id"] == "x") + assert page["sources"] == ["raw/a.md", "raw/b.md"] + + def test_a_page_without_sources_reports_an_empty_list(self, git_wiki): + _write(git_wiki, "entities/y.md", "---\ntitle: Y\n---\nBody.\n") + pages = wiki_api.wiki_scan(wiki_path=str(git_wiki))["pages"] + page = next(p for p in pages if p["id"] == "y") + assert page["sources"] == [] + + +class TestUpdateThreadsProvenance: + def test_update_records_the_trigger_it_was_given(self, git_wiki): + # The bug: trigger was hardcoded "manual" here, so an automated ingest + # and a hand edit in the desktop app were indistinguishable. + wiki_api.wiki_update( + "entities/x.md", "Body.\n", + frontmatter={"title": "X"}, + trigger="ingest", + source_events=["raw/src.md"], + summary="ingested the release notes", + wiki_path=str(git_wiki), + ) + cs = wiki_changeset.wiki_query_changesets(wiki_path=str(git_wiki))["changesets"][0] + assert cs["trigger"] == "ingest" + assert cs["source_event_keys"] == ["raw/src.md"] + assert cs["summary"] == "ingested the release notes" + + def test_update_defaults_to_manual_with_unknown_provenance(self, git_wiki): + wiki_api.wiki_update( + "entities/y.md", "Body.\n", + frontmatter={"title": "Y"}, + wiki_path=str(git_wiki), + ) + cs = wiki_changeset.wiki_query_changesets(wiki_path=str(git_wiki))["changesets"][0] + assert cs["trigger"] == "manual" + # A hand edit in the app genuinely has no ingestion event, and the + # honest record of that is an empty list, not a fabricated source. + assert cs["source_event_keys"] == []

diff --git hermes-agent/tui_gateway/wiki_watch.py harness/tui_gateway/wiki_watch.py new file mode 100644 index 0000000000000000000000000000000000000000..a585cd55ba785c03400c483bb9a858beb7f77d3f --- /dev/null +++ harness/tui_gateway/wiki_watch.py @@ -0,0 +1,203 @@ +"""Programmatic changeset capture for out-of-band wiki writes. + +The wiki's event feed is a join over ``raw/`` ingestion sources and the +changeset index — and until now the index only gained entries when a write +went through the ``wiki.update`` RPC or when the agent REMEMBERED to run the +capture CLI after editing pages with file tools. There was no programmatic +enforcement on insertion: a page written by the agent's terminal session, +a human editor, or any script simply never appeared in the feed, so the +desktop's wiki timeline rendered only whatever the prompting happened to +capture ("best faith prompting"). + +This module is the enforcement. A daemon thread snapshots every registered +wiki's page tree and, on each sweep, captures a changeset for any page whose +content changed OUTSIDE the capture machinery — then emits a ``wiki.changed`` +gateway event so connected clients know the feed moved. + +Dedup, so RPC- and CLI-captured writes don't double-record: before capturing, +the sweeper compares the page's current SHA256 against the ``after_sha256`` +of the page's most recent changeset. A match means the write was already +captured by whoever made it; the sweeper just refreshes its snapshot. + +The first sweep of each wiki is a BASELINE: it records state without +capturing, so a gateway restart over a 200-page wiki does not mint 200 +spurious "auto" changesets. Writes made while the gateway was down are +therefore not back-filled — the enforcement is live-forward, matching what +an event feed is for. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +import os +import threading +import time +from pathlib import Path +from typing import Callable, Optional + +logger = logging.getLogger(__name__) + +# Directories under the wiki root that are not pages: changeset storage and +# the raw ingestion sources (raw/ files ARE events — wiki_events reads them +# from disk directly, no changeset needed). +_SKIP_DIRS = {"changesets", "raw", ".git"} + +_SWEEP_INTERVAL_S = float(os.environ.get("HERMES_WIKI_WATCH_INTERVAL", "20")) + + +def _snapshot(wiki_root: Path) -> dict[str, tuple[int, int]]: + """Relative page path → (mtime_ns, size) for every .md page.""" + out: dict[str, tuple[int, int]] = {} + if not wiki_root.is_dir(): + return out + for base, dirs, files in os.walk(wiki_root): + rel_base = Path(base).relative_to(wiki_root) + parts = rel_base.parts + if parts and parts[0] in _SKIP_DIRS: + dirs[:] = [] + continue + dirs[:] = [d for d in dirs if not (not parts and d in _SKIP_DIRS)] + for name in files: + if not name.endswith(".md"): + continue + p = Path(base) / name + try: + st = p.stat() + except OSError: + continue + out[str(p.relative_to(wiki_root))] = (st.st_mtime_ns, st.st_size) + return out + + +def _sha256(path: Path) -> Optional[str]: + try: + h = hashlib.sha256() + with open(path, "rb") as f: + for chunk in iter(lambda: f.read(65536), b""): + h.update(chunk) + return h.hexdigest() + except OSError: + return None + + +def _latest_captured_sha(page: str, wiki_root: Path) -> Optional[str]: + """The ``after_sha256`` of the page's newest changeset, or None.""" + try: + index_path = wiki_root / "changesets" / "index.json" + entries = json.loads(index_path.read_text(encoding="utf-8")) + for entry in entries: # newest first + if entry.get("page") == page: + cs_path = wiki_root / "changesets" / f"{entry.get('id')}.json" + cs = json.loads(cs_path.read_text(encoding="utf-8")) + return cs.get("after_sha256") + except Exception: + return None + return None + + +# A page whose mtime is younger than this is possibly still being written +# (multi-chunk agent edits); leave it for the next sweep so a half-written +# page is never captured as a changeset. +_QUIESCENT_NS = int(3e9) + + +def sweep_once( + prev: Optional[dict[str, tuple[int, int]]], + wiki_root: Path, + capture: Callable[..., dict], + now_ns: Optional[int] = None, +) -> tuple[dict[str, tuple[int, int]], list[dict]]: + """One sweep of one wiki. Returns (new snapshot, captured changesets). + + ``prev is None`` marks the baseline sweep: snapshot only, capture nothing. + Deletions are NOT captured (the capture helper hashes the post-write file, + which a delete no longer has) — a deleted page simply leaves the snapshot. + Pure enough to test directly — the thread below owns time and emission. + """ + now = now_ns if now_ns is not None else time.time_ns() + current = _snapshot(wiki_root) + if prev is None: + return current, [] + + captured: list[dict] = [] + settled = dict(prev) + for page, stamp in current.items(): + known = prev.get(page) + if known == stamp: + settled[page] = stamp + continue + if now - stamp[0] < _QUIESCENT_NS: + # Still (possibly) being written — keep the OLD record so the next + # sweep re-examines this page instead of silently adopting the + # half-written state. + if known is not None: + settled[page] = known + else: + settled.pop(page, None) + continue + settled[page] = stamp + action = "create" if known is None else "update" + sha = _sha256(wiki_root / page) + if sha is not None and sha == _latest_captured_sha(page, wiki_root): + # Already captured by wiki.update or the capture CLI — the write + # brought its own audit entry; recording it again would double + # every RPC write in the feed. + continue + try: + cs = capture( + page_path=page, + action=action, + summary=f"auto-captured: page {action} outside wiki.update", + trigger="auto", + wiki_path=str(wiki_root), + ) + if isinstance(cs, dict) and "error" not in cs: + captured.append(cs) + except Exception: + logger.warning("wiki_watch: capture failed for %s", page, exc_info=True) + # Deleted pages fall out: settled starts from prev, so drop anything no + # longer on disk. + settled = {k: v for k, v in settled.items() if k in current} + return settled, captured + + +def start_wiki_watcher( + emit: Callable[[str, str, dict], None], + wiki_roots: Callable[[], list[str]], + interval: float = _SWEEP_INTERVAL_S, +) -> threading.Thread: + """Start the sweeper daemon. ``emit(type, sid, payload)`` matches the + gateway's ``_emit``; ``wiki_roots`` is called per sweep so registry edits + take effect without a restart.""" + + def _loop() -> None: + # The version-skew-safe loader from wiki_api (repo copy first, only + # accepting a module that actually has the capture symbol). + from tui_gateway.wiki_api import _load_wiki_changeset_module + wiki_capture_changeset = _load_wiki_changeset_module( + "wiki_capture_changeset" + ).wiki_capture_changeset + + states: dict[str, Optional[dict[str, tuple[int, int]]]] = {} + while True: + try: + for root_str in wiki_roots(): + root = Path(os.path.expanduser(root_str)) + prev = states.get(root_str) + snapshot, captured = sweep_once(prev, root, wiki_capture_changeset) + states[root_str] = snapshot + if captured: + emit("wiki.changed", "", { + "wiki": root_str, + "pages": [c.get("page") for c in captured], + "changesets": [c.get("id") for c in captured], + }) + except Exception: + logger.debug("wiki_watch sweep failed", exc_info=True) + time.sleep(interval) + + thread = threading.Thread(target=_loop, name="wiki-watch", daemon=True) + thread.start() + return thread

diff --git hermes-agent/docs/api/wiki-glossary.md harness/docs/api/wiki-glossary.md new file mode 100644 index 0000000000000000000000000000000000000000..6976800315c93d56b14e1eb58b38ca2ed66e2a98 --- /dev/null +++ harness/docs/api/wiki-glossary.md @@ -0,0 +1,115 @@ +# Per-wiki glossary API + +Each wiki configured in `$HERMES_HOME/wikis.yaml` may define a proper-noun +glossary at `<wiki-root>/glossary.yaml`. The glossary is optional. It is owned +by the wiki rather than by an installed skill, so gateway clients and other +Harness code use the same validated data. + +## `glossary.yaml` schema + +```yaml +version: 1 +mode: canonicalize # canonicalize | strict +proper_nouns: + - canonical: Nous Research + aliases: + - Nous + description: AI research organization + - canonical: Hermes Agent +``` + +The root fields are required: + +- `version` must be the integer `1`. +- `mode` must be `canonicalize` or `strict`. +- `proper_nouns` must be an array. Each entry requires a non-empty `canonical` + string and may contain an `aliases` array of non-empty strings and a + non-empty `description` string. + +Canonical spellings and aliases share one case-insensitive namespace. A +spelling cannot duplicate another canonical spelling or alias, including an +alias in its own entry. Unknown fields are rejected. If the file is absent, +the glossary is disabled; if it is present but unreadable, malformed, or +invalid, loading fails closed. + +Resource bounds are enforced at validation time: at most 2,000 entries, 50 +aliases per entry, 256 characters per spelling, and 2,000 characters per +description. + +Canonicalization maps a canonical spelling or alias to its `canonical` value. +An unknown spelling passes through in `canonicalize` mode and is rejected in +`strict` mode. + +Harness consumers may call `canonicalize_text` to normalize configured forms +inside generated text and `canonicalize_inventory` to normalize and deduplicate +a model-declared proper-noun inventory. In strict mode, the inventory helper +raises on any unlisted name. Producers must invoke these helpers at their +post-generation boundary; publishing a policy does not implicitly rewrite +unrelated `wiki.update` calls. + +## Wiki selection and authority + +Both RPCs accept `wiki`, which must be a name in +`$HERMES_HOME/wikis.yaml`. If `wiki` is omitted, that registry must contain an +explicit `default` whose name exists in `wikis`. Raw paths, unknown names, +empty names, invalid configured paths, and environment/home-directory +fallbacks are rejected. + +## `wiki.glossary` + +Read one glossary: + +```jsonc +{ + "method": "wiki.glossary", + "params": { "wiki": "main" } +} +``` + +Result when absent: + +```json +{ + "enabled": false, + "version": 1, + "mode": "canonicalize", + "proper_nouns": [], + "revision": "" +} +``` + +A present valid file returns the same shape with `enabled: true`, normalized +entries, and `revision` set to the SHA-256 hex digest of the stored bytes. +Invalid selection or an invalid present glossary returns JSON-RPC error `4001`. +Unexpected I/O/runtime failures return `5062`. + +## `wiki.glossary.update` + +Atomically replace one glossary: + +```jsonc +{ + "method": "wiki.glossary.update", + "params": { + "wiki": "main", + "version": 1, + "mode": "strict", + "proper_nouns": [ + { "canonical": "OpenAI", "aliases": ["Open AI"] } + ], + "if_match": "sha256-from-the-last-read" + } +} +``` + +`version`, `mode`, and `proper_nouns` are required. `if_match` is optional; when +present it must equal the current revision. Use the empty string to require +that no glossary currently exists. A stale revision returns error `409` and +does not write. Invalid params or schema return `4001`; unexpected failures +return `5063`. A successful update returns the normalized read shape, including +the new revision. + +Writes use a same-directory temporary file, `fsync`, and atomic replacement. +Clients should retain the returned revision and provide it on their next edit. + +Both method names are advertised by `gateway.capabilities`.

diff --git hermes-agent/tests/tui_gateway/test_wiki_glossary.py harness/tests/tui_gateway/test_wiki_glossary.py new file mode 100644 index 0000000000000000000000000000000000000000..e701b40037e84446db76f2f8e113cc58412526b8 --- /dev/null +++ harness/tests/tui_gateway/test_wiki_glossary.py @@ -0,0 +1,372 @@ +from __future__ import annotations + +import hashlib +import subprocess +import sys +from pathlib import Path + +import pytest +import yaml + +from tui_gateway import server, wiki_glossary as glossary_module +from tui_gateway.wiki_glossary import ( + GlossaryConflictError, + GlossaryValidationError, + canonicalize_proper_noun, + canonicalize_text, + canonicalize_inventory, + load_glossary, + load_configured_wikis, + normalize_glossary, + update_glossary, +) + + +def _configure(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> tuple[Path, Path]: + home = tmp_path / "home" + primary = tmp_path / "primary" + other = tmp_path / "other" + primary.mkdir() + other.mkdir() + home.mkdir() + (home / "wikis.yaml").write_text( + yaml.safe_dump( + { + "default": "primary", + "wikis": {"primary": str(primary), "other": str(other)}, + }, + sort_keys=False, + ), + encoding="utf-8", + ) + monkeypatch.setenv("HERMES_HOME", str(home)) + return primary, other + + +def _call(method: str, params: dict, rid: int = 7) -> dict: + return server._methods[method](rid, params) + + +def test_missing_glossary_is_disabled_default(tmp_path: Path) -> None: + glossary = load_glossary(tmp_path) + assert glossary == { + "enabled": False, + "version": 1, + "mode": "canonicalize", + "proper_nouns": [], + "revision": "", + } + + +def test_loader_normalizes_and_canonicalizes_aliases(tmp_path: Path) -> None: + raw = ( + "version: 1\n" + "mode: strict\n" + "proper_nouns:\n" + " - canonical: OpenAI\n" + " aliases: [open ai, OPEN-AI]\n" + " description: Model provider\n" + ) + (tmp_path / "glossary.yaml").write_text(raw, encoding="utf-8") + + glossary = load_glossary(tmp_path) + assert glossary["enabled"] is True + assert glossary["revision"] == hashlib.sha256(raw.encode()).hexdigest() + assert glossary["proper_nouns"] == [ + { + "canonical": "OpenAI", + "aliases": ["open ai", "OPEN-AI"], + "description": "Model provider", + } + ] + assert canonicalize_proper_noun("OPEN AI", glossary) == "OpenAI" + assert canonicalize_proper_noun("unknown", glossary) is None + assert canonicalize_text("Open ai uses OPEN-AI.", glossary) == "OpenAI uses OpenAI." + assert canonicalize_inventory(["OPEN AI", "OpenAI"], glossary) == ["OpenAI"] + with pytest.raises(GlossaryValidationError, match="unknown proper nouns"): + canonicalize_inventory(["OpenAI", "Acme Cloud"], glossary) + + +@pytest.mark.parametrize( + "payload, message", + [ + ( + { + "version": 1, + "mode": "canonicalize", + "proper_nouns": [{"canonical": "x" * 257}], + }, + "at most 256", + ), + ( + { + "version": 1, + "mode": "canonicalize", + "proper_nouns": [ + { + "canonical": "OpenAI", + "aliases": [f"alias-{i}" for i in range(51)], + } + ], + }, + "at most 50 aliases", + ), + ( + { + "version": 1, + "mode": "canonicalize", + "proper_nouns": [{"canonical": "OpenAI", "description": "x" * 2001}], + }, + "at most 2000", + ), + ( + { + "version": 1, + "mode": "canonicalize", + "proper_nouns": [{"canonical": f"Term {i}"} for i in range(2001)], + }, + "at most 2000 entries", + ), + ], +) +def test_normalize_glossary_enforces_resource_bounds( + payload: dict, message: str +) -> None: + with pytest.raises(GlossaryValidationError, match=message): + normalize_glossary(payload) + + +@pytest.mark.parametrize( + "payload", + [ + {"version": 2, "mode": "canonicalize", "proper_nouns": []}, + {"version": 1, "mode": "permissive", "proper_nouns": []}, + { + "version": 1, + "mode": "canonicalize", + "proper_nouns": [ + {"canonical": "OpenAI", "aliases": ["OA"]}, + {"canonical": "oa"}, + ], + }, + { + "version": 1, + "mode": "canonicalize", + "proper_nouns": [{"canonical": "OpenAI", "aliases": ["openai"]}], + }, + ], +) +def test_malformed_or_ambiguous_present_glossary_fails_closed( + tmp_path: Path, payload: dict +) -> None: + (tmp_path / "glossary.yaml").write_text(yaml.safe_dump(payload), encoding="utf-8") + with pytest.raises(GlossaryValidationError): + load_glossary(tmp_path) + + +def test_update_is_atomic_and_optimistically_concurrent(tmp_path: Path) -> None: + first = update_glossary( + tmp_path, + {"version": 1, "mode": "canonicalize", "proper_nouns": []}, + if_match="", + ) + assert first["enabled"] is True + assert first["revision"] + assert load_glossary(tmp_path) == first + + with pytest.raises(GlossaryConflictError): + update_glossary( + tmp_path, + {"version": 1, "mode": "strict", "proper_nouns": []}, + if_match="stale", + ) + assert load_glossary(tmp_path) == first + assert not list(tmp_path.glob(".glossary.yaml.*.tmp")) + + +def test_update_returns_the_revision_and_content_it_wrote( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + calls = 0 + + def fake_load(_root: Path) -> dict: + nonlocal calls + calls += 1 + if calls == 1: + return { + "enabled": False, + "version": 1, + "mode": "canonicalize", + "proper_nouns": [], + "revision": "", + } + return { + "enabled": True, + "version": 1, + "mode": "strict", + "proper_nouns": [{"canonical": "Foreign"}], + "revision": "foreign", + } + + monkeypatch.setattr(glossary_module, "load_glossary", fake_load) + result = update_glossary( + tmp_path, + { + "version": 1, + "mode": "canonicalize", + "proper_nouns": [{"canonical": "Own write"}], + }, + if_match="", + ) + + assert result["proper_nouns"] == [{"canonical": "Own write"}] + assert result["mode"] == "canonicalize" + assert calls == 1 + + +def test_interprocess_lock_blocks_a_second_process(tmp_path: Path) -> None: + script = """ +import sys +from pathlib import Path +from tui_gateway.wiki_glossary import _interprocess_write_lock +with _interprocess_write_lock(Path(sys.argv[1])): + print('acquired') +""" + with glossary_module._interprocess_write_lock(tmp_path): + process = subprocess.Popen( + [sys.executable, "-c", script, str(tmp_path)], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + try: + with pytest.raises(subprocess.TimeoutExpired): + process.communicate(timeout=0.2) + except BaseException: + process.kill() + process.communicate() + raise + + stdout, stderr = process.communicate(timeout=5) + assert process.returncode == 0, stderr + assert stdout.strip() == "acquired" + + +def test_case_insensitive_regex_never_indexes_an_unmapped_casefold() -> None: + glossary = { + "enabled": True, + "version": 1, + "mode": "canonicalize", + "proper_nouns": [{"canonical": "Item", "aliases": ["i"]}], + "revision": "test", + } + assert canonicalize_text("İ ı i", glossary) == "İ ı Item" + + +def test_duplicate_yaml_keys_fail_closed(tmp_path: Path) -> None: + (tmp_path / "glossary.yaml").write_text( + "version: 1\nmode: strict\nmode: canonicalize\nproper_nouns: []\n", + encoding="utf-8", + ) + with pytest.raises(GlossaryValidationError, match="duplicate key"): + load_glossary(tmp_path) + + +def test_duplicate_registry_keys_fail_closed( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + home = tmp_path / "home" + home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + (home / "wikis.yaml").write_text( + "default: primary\nwikis:\n primary: /one\n primary: /two\n", + encoding="utf-8", + ) + with pytest.raises(GlossaryValidationError, match="duplicate key"): + load_configured_wikis() + + +def test_rpc_methods_registered_and_advertised() -> None: + assert "wiki.glossary" in server._methods + assert "wiki.glossary.update" in server._methods + assert "wiki.glossary" in server._LONG_HANDLERS + assert "wiki.glossary.update" in server._LONG_HANDLERS + capabilities = _call("gateway.capabilities", {})["result"]["capability_names"] + assert "wiki.glossary" in capabilities + assert "wiki.glossary.update" in capabilities + + +def test_rpc_read_uses_only_configured_name_or_explicit_default( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + primary, _ = _configure(monkeypatch, tmp_path) + + assert _call("wiki.glossary", {})["result"]["enabled"] is False + assert _call("wiki.glossary", {"wiki": "primary"})["result"]["enabled"] is False + + for forbidden_params in ( + {"wiki": str(primary)}, + {"wiki": "~/wiki"}, + {"wiki": "unknown"}, + {"wiki": ""}, + {"wiki": 3}, + {"path": str(primary)}, + ): + response = _call("wiki.glossary", forbidden_params) + assert response["error"]["code"] == 4001 + + +def test_rpc_rejects_missing_default_without_fallback( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + _configure(monkeypatch, tmp_path) + home = Path(str(tmp_path / "home")) + (home / "wikis.yaml").write_text( + yaml.safe_dump({"wikis": {"primary": str(tmp_path / "primary")}}), + encoding="utf-8", + ) + monkeypatch.setenv("WIKI_PATH", str(tmp_path / "forbidden-fallback")) + + response = _call("wiki.glossary", {}) + assert response["error"]["code"] == 4001 + + +def test_rpc_read_fails_closed_for_present_invalid_file( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + primary, _ = _configure(monkeypatch, tmp_path) + (primary / "glossary.yaml").write_text("version: [unterminated", encoding="utf-8") + + response = _call("wiki.glossary", {"wiki": "primary"}) + assert response["error"]["code"] == 4001 + + +def test_rpc_update_validates_payload_and_maps_conflict( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + _configure(monkeypatch, tmp_path) + valid = { + "wiki": "primary", + "version": 1, + "mode": "canonicalize", + "proper_nouns": [{"canonical": "Nous Research", "aliases": ["Nous"]}], + "if_match": "", + } + created = _call("wiki.glossary.update", valid) + assert created["result"]["proper_nouns"][0]["canonical"] == "Nous Research" + + stale = _call("wiki.glossary.update", {**valid, "mode": "strict"}) + assert stale["error"]["code"] == 409 + + ambiguous = _call( + "wiki.glossary.update", + { + **valid, + "if_match": created["result"]["revision"], + "proper_nouns": [ + {"canonical": "Alpha", "aliases": ["shared"]}, + {"canonical": "Beta", "aliases": ["SHARED"]}, + ], + }, + ) + assert ambiguous["error"]["code"] == 4001

diff --git hermes-agent/tui_gateway/wiki_glossary.py harness/tui_gateway/wiki_glossary.py new file mode 100644 index 0000000000000000000000000000000000000000..a6890d7a97aa064b776f5fe1b1589f447a804088 --- /dev/null +++ harness/tui_gateway/wiki_glossary.py @@ -0,0 +1,425 @@ +"""Validated per-wiki proper-noun glossary storage. + +This module is deliberately part of the gateway rather than a skill: native +clients and any Harness caller share one schema, validator, canonicalizer, and +optimistic-concurrency implementation. +""" + +from __future__ import annotations + +import hashlib +import os +import re +import tempfile +import threading +from contextlib import contextmanager +from pathlib import Path +from typing import Any, Mapping + +import yaml + +from hermes_constants import get_hermes_home + +GLOSSARY_FILENAME = "glossary.yaml" +GLOSSARY_VERSION = 1 +GLOSSARY_MODES = frozenset({"canonicalize", "strict"}) +MAX_GLOSSARY_ENTRIES = 2_000 +MAX_ALIASES_PER_ENTRY = 50 +MAX_TERM_LENGTH = 256 +MAX_DESCRIPTION_LENGTH = 2_000 +_WRITE_LOCK = threading.Lock() + + +class _UniqueKeyLoader(yaml.SafeLoader): + """Safe YAML loader that rejects duplicate mapping keys.""" + + +def _construct_unique_mapping( + loader: yaml.SafeLoader, node: yaml.nodes.MappingNode, deep: bool = False +) -> dict: + mapping: dict[Any, Any] = {} + for key_node, value_node in node.value: + key = loader.construct_object(key_node, deep=deep) + try: + duplicate = key in mapping + except TypeError as exc: + raise yaml.constructor.ConstructorError( + "while constructing a mapping", + node.start_mark, + "found an unhashable key", + key_node.start_mark, + ) from exc + if duplicate: + raise yaml.constructor.ConstructorError( + "while constructing a mapping", + node.start_mark, + f"found duplicate key {key!r}", + key_node.start_mark, + ) + mapping[key] = loader.construct_object(value_node, deep=deep) + return mapping + + +_UniqueKeyLoader.add_constructor( + yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, + _construct_unique_mapping, +) + + +def _safe_load_unique(value: str) -> Any: + return yaml.load(value, Loader=_UniqueKeyLoader) + + +@contextmanager +def _interprocess_write_lock(root: Path): + """Serialize glossary compare-and-replace across gateway processes.""" + lock_path = root / f".{GLOSSARY_FILENAME}.lock" + descriptor = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600) + try: + if os.name == "nt": + import msvcrt + + if os.fstat(descriptor).st_size == 0: + os.write(descriptor, b"\0") + os.lseek(descriptor, 0, os.SEEK_SET) + msvcrt.locking(descriptor, msvcrt.LK_LOCK, 1) + try: + yield + finally: + os.lseek(descriptor, 0, os.SEEK_SET) + msvcrt.locking(descriptor, msvcrt.LK_UNLCK, 1) + else: + import fcntl + + fcntl.flock(descriptor, fcntl.LOCK_EX) + try: + yield + finally: + fcntl.flock(descriptor, fcntl.LOCK_UN) + finally: + os.close(descriptor) + + +class GlossaryError(ValueError): + """Base class for client-actionable glossary failures.""" + + +class GlossaryValidationError(GlossaryError): + """The registry or glossary does not satisfy its public schema.""" + + +class GlossaryConflictError(GlossaryError): + """An optimistic-concurrency revision did not match current storage.""" + + +def _disabled_glossary() -> dict[str, Any]: + return { + "enabled": False, + "version": GLOSSARY_VERSION, + "mode": "canonicalize", + "proper_nouns": [], + "revision": "", + } + + +def _nonempty_string( + value: Any, field: str, *, max_length: int = MAX_TERM_LENGTH +) -> str: + if not isinstance(value, str) or not value.strip(): + raise GlossaryValidationError(f"{field} must be a non-empty string") + normalized = value.strip() + if len(normalized) > max_length: + raise GlossaryValidationError( + f"{field} must be at most {max_length} characters" + ) + return normalized + + +def normalize_glossary(data: Any) -> dict[str, Any]: + """Validate and normalize the persisted/update glossary payload. + + Proper-noun spellings form one case-insensitive namespace. A canonical + spelling or alias therefore cannot duplicate any other canonical spelling + or alias, including another spelling in the same entry. + """ + if not isinstance(data, Mapping): + raise GlossaryValidationError("glossary must be an object") + allowed_root = {"version", "mode", "proper_nouns"} + unknown = set(data) - allowed_root + if unknown: + raise GlossaryValidationError( + f"unknown glossary field(s): {', '.join(sorted(map(str, unknown)))}" + ) + + version = data.get("version") + if isinstance(version, bool) or version != GLOSSARY_VERSION: + raise GlossaryValidationError("version must be 1") + mode = data.get("mode") + if mode not in GLOSSARY_MODES: + raise GlossaryValidationError("mode must be 'canonicalize' or 'strict'") + raw_entries = data.get("proper_nouns") + if not isinstance(raw_entries, list): + raise GlossaryValidationError("proper_nouns must be an array") + if len(raw_entries) > MAX_GLOSSARY_ENTRIES: + raise GlossaryValidationError( + f"proper_nouns must contain at most {MAX_GLOSSARY_ENTRIES} entries" + ) + + seen: dict[str, str] = {} + entries: list[dict[str, Any]] = [] + for index, raw in enumerate(raw_entries): + field = f"proper_nouns[{index}]" + if not isinstance(raw, Mapping): + raise GlossaryValidationError(f"{field} must be an object") + allowed_entry = {"canonical", "aliases", "description"} + entry_unknown = set(raw) - allowed_entry + if entry_unknown: + raise GlossaryValidationError( + f"unknown {field} field(s): " + + ", ".join(sorted(map(str, entry_unknown))) + ) + + canonical = _nonempty_string(raw.get("canonical"), f"{field}.canonical") + aliases_raw = raw.get("aliases", []) + if not isinstance(aliases_raw, list): + raise GlossaryValidationError( + f"{field}.aliases must be an array of strings" + ) + if len(aliases_raw) > MAX_ALIASES_PER_ENTRY: + raise GlossaryValidationError( + f"{field} must contain at most {MAX_ALIASES_PER_ENTRY} aliases" + ) + aliases = [ + _nonempty_string(alias, f"{field}.aliases[{alias_index}]") + for alias_index, alias in enumerate(aliases_raw) + ] + description = raw.get("description") + if description is not None: + description = _nonempty_string( + description, + f"{field}.description", + max_length=MAX_DESCRIPTION_LENGTH, + ) + + for spelling in [canonical, *aliases]: + folded = spelling.casefold() + if folded in seen: + raise GlossaryValidationError( + f"ambiguous proper noun spelling {spelling!r}; " + f"already used by {seen[folded]!r}" + ) + seen[folded] = canonical + + entry: dict[str, Any] = {"canonical": canonical} + if aliases: + entry["aliases"] = aliases + if description is not None: + entry["description"] = description + entries.append(entry) + + return {"version": GLOSSARY_VERSION, "mode": mode, "proper_nouns": entries} + + +def load_glossary(wiki_root: str | os.PathLike[str]) -> dict[str, Any]: + """Load ``glossary.yaml`` or return the disabled state when absent. + + A present file is never treated as absent after a parse, decode, or schema + error; callers receive :class:`GlossaryValidationError` (fail closed). + """ + path = Path(wiki_root) / GLOSSARY_FILENAME + try: + raw = path.read_bytes() + except FileNotFoundError: + return _disabled_glossary() + except OSError as exc: + raise GlossaryValidationError( + f"cannot read {GLOSSARY_FILENAME}: {exc}" + ) from exc + + try: + decoded = raw.decode("utf-8") + parsed = _safe_load_unique(decoded) + except (UnicodeDecodeError, yaml.YAMLError) as exc: + raise GlossaryValidationError(f"invalid {GLOSSARY_FILENAME}: {exc}") from exc + + normalized = normalize_glossary(parsed) + return { + "enabled": True, + **normalized, + "revision": hashlib.sha256(raw).hexdigest(), + } + + +def canonicalize_proper_noun(value: str, glossary: Mapping[str, Any]) -> str | None: + """Resolve a canonical/alias spelling according to a loaded glossary. + + Unknown spellings pass through in ``canonicalize`` mode and are rejected + (``None``) in ``strict`` mode. A disabled glossary is a no-op. + """ + if not isinstance(value, str): + raise TypeError("value must be a string") + if not glossary.get("enabled", False): + return value + folded = value.casefold() + for entry in glossary.get("proper_nouns", []): + spellings = [entry["canonical"], *entry.get("aliases", [])] + if any(folded == spelling.casefold() for spelling in spellings): + return entry["canonical"] + return None if glossary.get("mode") == "strict" else value + + +def _canonical_mapping(glossary: Mapping[str, Any]) -> dict[str, str]: + mapping: dict[str, str] = {} + for entry in glossary.get("proper_nouns", []): + canonical = entry["canonical"] + for spelling in [canonical, *entry.get("aliases", [])]: + mapping[spelling.casefold()] = canonical + return mapping + + +def canonicalize_text(value: str, glossary: Mapping[str, Any]) -> str: + """Normalize every configured canonical spelling or alias in free text.""" + if not isinstance(value, str): + raise TypeError("value must be a string") + if not glossary.get("enabled", False): + return value + mapping = _canonical_mapping(glossary) + forms = sorted(mapping, key=lambda form: (-len(form), form)) + if not forms: + return value + pattern = re.compile( + r"(?<!\w)(?:" + "|".join(re.escape(form) for form in forms) + r")(?!\w)", + re.IGNORECASE, + ) + return pattern.sub( + lambda match: mapping.get(match.group(0).casefold(), match.group(0)), + value, + ) + + +def canonicalize_inventory(values: Any, glossary: Mapping[str, Any]) -> list[str]: + """Normalize a declared proper-noun inventory and enforce strict mode.""" + if not isinstance(values, list): + raise GlossaryValidationError("proper-noun inventory must be an array") + if len(values) > MAX_GLOSSARY_ENTRIES: + raise GlossaryValidationError( + f"proper-noun inventory must contain at most {MAX_GLOSSARY_ENTRIES} entries" + ) + result: list[str] = [] + unknown: list[str] = [] + seen: set[str] = set() + for index, value in enumerate(values): + normalized = _nonempty_string(value, f"proper_nouns[{index}]") + canonical = canonicalize_proper_noun(normalized, glossary) + if canonical is None: + unknown.append(normalized) + continue + folded = canonical.casefold() + if folded not in seen: + seen.add(folded) + result.append(canonical) + if unknown: + raise GlossaryValidationError( + "unknown proper nouns in strict mode: " + + ", ".join(sorted(set(unknown), key=str.casefold)) + ) + return result + + +def update_glossary( + wiki_root: str | os.PathLike[str], + data: Any, + *, + if_match: str | None = None, +) -> dict[str, Any]: + """Validate and atomically replace one wiki glossary.""" + if if_match is not None and not isinstance(if_match, str): + raise GlossaryValidationError("if_match must be a string") + normalized = normalize_glossary(data) + root = Path(wiki_root) + path = root / GLOSSARY_FILENAME + + root.mkdir(parents=True, exist_ok=True) + with _WRITE_LOCK, _interprocess_write_lock(root): + current = load_glossary(root) + if if_match is not None and if_match != current["revision"]: + raise GlossaryConflictError("glossary revision conflict") + + rendered = yaml.safe_dump( + normalized, + allow_unicode=True, + sort_keys=False, + ).encode("utf-8") + fd, temporary = tempfile.mkstemp( + prefix=f".{GLOSSARY_FILENAME}.", suffix=".tmp", dir=root + ) + try: + with os.fdopen(fd, "wb") as stream: + stream.write(rendered) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, path) + try: + directory_fd = os.open(root, os.O_RDONLY) + try: + os.fsync(directory_fd) + finally: + os.close(directory_fd) + except OSError: + # Some platforms/filesystems do not support directory fsync. + pass + except BaseException: + try: + os.unlink(temporary) + except FileNotFoundError: + pass + raise + + return { + "enabled": True, + **normalized, + "revision": hashlib.sha256(rendered).hexdigest(), + } + + +def load_configured_wikis() -> tuple[str | None, dict[str, str]]: + """Return one validated registry snapshot as ``(default, wikis)``.""" + registry_path = Path(get_hermes_home()) / "wikis.yaml" + try: + parsed = _safe_load_unique(registry_path.read_text(encoding="utf-8")) + except FileNotFoundError as exc: + raise GlossaryValidationError("no configured wiki registry") from exc + except (OSError, UnicodeDecodeError, yaml.YAMLError) as exc: + raise GlossaryValidationError(f"invalid wiki registry: {exc}") from exc + if not isinstance(parsed, Mapping): + raise GlossaryValidationError("wiki registry must be an object") + raw_wikis = parsed.get("wikis") + if not isinstance(raw_wikis, Mapping): + raise GlossaryValidationError("wiki registry wikis must be an object") + + wikis: dict[str, str] = {} + for raw_name, raw_path in raw_wikis.items(): + name = _nonempty_string(raw_name, "wiki name") + path = _nonempty_string(raw_path, f"wiki {name!r} path") + wikis[name] = os.path.expanduser(path) + default = parsed.get("default") + if default is not None: + default = _nonempty_string(default, "wiki registry default") + return default, wikis + + +def resolve_configured_wiki(name: Any = None) -> str: + """Resolve only a configured wiki name, never a raw path or fallback.""" + default, wikis = load_configured_wikis() + if name is None: + if default is None: + raise GlossaryValidationError("wiki is required; no default is configured") + name = default + elif not isinstance(name, str) or not name.strip(): + raise GlossaryValidationError("wiki must be a configured non-empty name") + if name not in wikis: + raise GlossaryValidationError(f"unknown configured wiki: {name!r}") + root = Path(wikis[name]) + if not root.is_dir(): + raise GlossaryValidationError(f"configured wiki is not a directory: {name!r}") + return str(root)

diff --git hermes-agent/skills/research/llm-wiki-native/SKILL.md harness/skills/research/llm-wiki-native/SKILL.md new file mode 100644 index 0000000000000000000000000000000000000000..d69ac65b2c03113e7ce075df8a2b54a0109b14c0 --- /dev/null +++ harness/skills/research/llm-wiki-native/SKILL.md @@ -0,0 +1,201 @@ +--- +name: llm-wiki-native +description: "Scan, query, and record changesets via the native wiki API." +version: 1.0.0 +author: Hermes Agent +license: MIT +platforms: [linux, macos, windows] +metadata: + hermes: + tags: [wiki, knowledge-base, research, notes, markdown, native-api, changesets] + category: research + related_skills: [llm-wiki, obsidian, arxiv] +--- + +# LLM Wiki (Native API) + +Build and maintain [Karpathy's LLM Wiki](https://gist.github.com/karpathy/442a6bf555914893e9891c11519de94f) +— a persistent, compounding knowledge base of interlinked markdown files — using +Hermes' **native wiki API** instead of raw filesystem walks. + +This is the API-native companion to the [`llm-wiki`](../llm-wiki/SKILL.md) skill. +The conventions (three layers, frontmatter, taxonomy, page thresholds, update +policy) are identical — read that skill for the full curation philosophy. **This +skill changes only the mechanics: how you orient, search, and — critically — how +you record what changed.** + +## Why use this instead of plain filesystem ops + +The Hermes gateway exposes the wiki through `tui_gateway/wiki_api.py`, and the +desktop/native app renders it: a graph view (`wiki.scan`), page detail +(`wiki.page`), taxonomy filtering, and a **Timeline tab** (`wiki.changesets`). + +Those views only reflect your work if changes go through the native code path. +In particular, **the Timeline is empty unless you capture a changeset after every +write.** Editing files directly with `write_file` keeps the markdown correct but +leaves the app's timeline blank and its graph stale until the next manual rescan. + +This skill drives the same code path the app reads: +- **Read/orient** through `wiki scan` / `wiki page` / `wiki changesets` — one call + returns the structured graph or timeline, no manual link-walking. +- **Record** every create/update/archive/delete through `wiki capture`, which + appends to `<wiki>/changesets/index.json` (and commits to git when available) — + so `wiki.changesets` and the app's Timeline populate in real time. + +## When to Use + +Prefer this skill over `llm-wiki` when **the user runs the Hermes desktop/native +app** (or any client that renders `wiki.scan` / `wiki.changesets`) and wants the +graph and timeline to stay live. Otherwise the two are interchangeable. Activates when the user: + +- Asks to ingest a source, file a query, or lint a wiki **and** uses the native app +- Asks "what changed in the wiki?", "show the timeline", or about recent wiki activity +- References their wiki/graph/timeline in the desktop app + +## Prerequisites + +- **Wiki location** — resolved exactly like the gateway: a name via + `~/.hermes/wikis.yaml`, else `$WIKI_PATH`, else `~/wiki`. Pass `--wiki NAME` to + target a specific registered wiki; omit it for the default. +- **Helper modules** — the CLI imports `tui_gateway/wiki_api.py` and + `scripts/wiki_changeset.py`. It finds them automatically in an installed Hermes + (`~/.hermes/scripts`) or a repo checkout. No third-party packages. +- **Git (optional)** — if the wiki dir is a git repo, `capture` records a commit + hash per change. Without git it still records changesets (empty `git_commit`). + +## How to Run + +All operations go through the bundled CLI via the `terminal` tool. From the skill +directory (`~/.hermes/skills/research/llm-wiki-native/`): + +```bash +python3 scripts/wiki.py <command> [args] [--wiki NAME] +``` + +## Quick Reference + +| Goal | Command | +|------|---------| +| Graph (pages + links) | `python3 scripts/wiki.py scan` | +| Read one page | `python3 scripts/wiki.py page entities/llama-cpp.md` | +| Valid taxonomy paths | `python3 scripts/wiki.py taxonomy` | +| Expand a page's integration links | `python3 scripts/wiki.py expand llama-cpp` | +| Recent timeline (newest first) | `python3 scripts/wiki.py changesets --limit 20` | +| One page's history | `python3 scripts/wiki.py changesets --page entities/llama-cpp.md` | +| Creates this week | `python3 scripts/wiki.py changesets --action create --since 2026-06-22T00:00:00Z` | +| Event log (what caused updates) | `python3 scripts/wiki.py events --limit 20` | +| Sources that produced nothing yet | `python3 scripts/wiki.py events --json` → events with empty `changesets` | +| **Record a change** | `python3 scripts/wiki.py capture entities/llama-cpp.md update "Added speculative decoding benchmarks" --trigger ingest --source-event raw/articles/src.md` | + +`--json` on `scan` / `page` / `changesets` / `events` returns raw JSON for +programmatic use. `capture` actions: `create` · `update` · `archive` · `delete`. +Triggers are conventionally `ingest` · `query` · `lint` · `process-inbox` · +`manual`, but the set is **open**: what an event kind is gets declared by a +`type: event-type` wiki page, so a new ingestion source is a page you write, +not a code change. + +## Provenance — always declare what caused a change + +`capture` takes `--source-event RAW_PATH`, repeatable, naming each event that +caused the change. **Declare it on every capture that has one.** A synthesis +drawing on three sources passes it three times. + +Omitting it records provenance as `unknown`. That is the honest answer for a +hand edit with no ingestion behind it, and a real gap for anything else — +`unknown` is never inferred or backfilled, precisely so that its presence +means something. The value of the whole log depends on `unknown` staying rare: +it should only ever describe changes that predate this field. + +Two things follow that are worth using: + +- `python3 scripts/wiki.py events` lists sources with **no** resulting + changesets — ingested material nobody has synthesized from yet. That's a + work queue, not an error. +- A page's `sources:` frontmatter is page-level provenance and is now returned + by `scan`. Keep it in sync with what you capture. + +## Procedure + +### Orient (every session — do this first) + +The native scan replaces manual SCHEMA/index/log reading for structure, but you +still read `SCHEMA.md` for conventions: + +```bash +python3 scripts/wiki.py scan # what pages/links exist +read_file "$WIKI/SCHEMA.md" # domain conventions + taxonomy +python3 scripts/wiki.py changesets --limit 30 # what changed recently +``` + +This prevents duplicate pages and missed cross-references — the same goal as the +`llm-wiki` orientation, fewer reads. + +### Ingest a source + +1. Capture the raw source to `raw/` (`web_extract` → `write_file`), with raw + frontmatter (`source_url`, `ingested`, `sha256`) — see `llm-wiki` for the format. +2. `python3 scripts/wiki.py scan` and `search_files` to find existing pages for + the entities/concepts mentioned. +3. Write/update pages with `write_file`, following SCHEMA.md (frontmatter, + `[[wikilinks]]` ≥ 2, taxonomy tags, page thresholds, update policy). +4. **Record each change** — once per page touched: + ```bash + python3 scripts/wiki.py capture entities/llama-cpp.md update \ + "Added b4820 speculative-decoding benchmarks" --trigger ingest \ + --source-event raw/articles/llama-cpp-release.md + ``` +5. Update `index.md` and `log.md` as usual. Report every file created/updated. + +> A single ingest commonly touches 5–15 pages. Capture a changeset for **each** — +> that's what fills the app's timeline and keeps the graph current. + +### Query + +1. `python3 scripts/wiki.py scan` (and `search_files` on large wikis) to find + relevant pages; `python3 scripts/wiki.py page <path>` to read them. +2. Synthesize, citing `[[pages]]`. +3. If the answer is worth keeping, write it to `queries/` or `comparisons/` and + `capture … create … --trigger query`. + +### Lint + +Run the `llm-wiki` lint checks (orphans, broken links, frontmatter, contradictions, +stale pages). Two are easier here: +- **Orphans / broken links:** `python3 scripts/wiki.py scan --json` gives the full + page set and resolved links — diff link targets against page ids in one pass. +- **Recent activity / log rotation:** `python3 scripts/wiki.py changesets` is the + authoritative history. + +Record fixes you make with `capture … --trigger lint`. + +## Pitfalls + +- **Capture after writing, not before.** `capture` hashes the page's current + on-disk state, so write the file first, then capture. (`delete` is the exception + — capture after removing the file.) +- **One capture per page, per logical change.** Don't capture the same file twice + for one edit; do capture each distinct page an ingest touches. +- **Don't invent provenance.** If you don't know which source caused a change, + leave `--source-event` off and let it record `unknown`. Naming a plausible + source you didn't actually read is worse than admitting the gap — it makes + every other recorded provenance untrustworthy. +- **`capture` doesn't write content.** It records a changeset for a page you wrote + with `write_file`. It is not a substitute for writing the markdown. +- **Never modify `raw/`.** Sources are immutable; corrections go in wiki pages. +- **`--wiki NAME` must match `wikis.yaml`** (or be a path). A name the gateway + can't resolve falls back to `$WIKI_PATH`/`~/wiki` — verify with `scan` first. +- **Everything else is the `llm-wiki` skill.** Frontmatter, taxonomy discipline, + page thresholds, cross-reference minimums, and the contradiction/update policy + are unchanged — follow them. + +## Verification + +After an operation, confirm it landed in the native path: + +```bash +python3 scripts/wiki.py changesets --limit 5 # your change appears, newest first +python3 scripts/wiki.py scan | head # new pages/links present +``` + +In the desktop app, the **Wiki → Timeline** tab should now show the changes, and +the graph should include any new pages/links.

diff --git hermes-agent/skills/research/llm-wiki-native/scripts/wiki.py harness/skills/research/llm-wiki-native/scripts/wiki.py new file mode 100755 index 0000000000000000000000000000000000000000..fba27ff72f3cf2a49f9cdcc65e48260d1c8e9bc8 --- /dev/null +++ harness/skills/research/llm-wiki-native/scripts/wiki.py @@ -0,0 +1,320 @@ +#!/usr/bin/env python3 +""" +wiki.py — thin CLI over the native Hermes wiki API. + +Wraps tui_gateway/wiki_api.py and scripts/wiki_changeset.py so the agent can +drive the structured wiki (scan / page / taxonomy / changesets / expand-links) +and record changesets through the SAME code path the native app reads — instead +of re-implementing graph walks and changeset capture with raw filesystem tools. + +This is what makes the native app's wiki views (graph + Timeline tab) reflect +the agent's work: every page write goes through `capture`, which appends to +`<wiki>/changesets/index.json`, which `wiki.changesets` serves. + +Usage (through the `terminal` tool): + python3 wiki.py scan [--wiki NAME] [--json] + python3 wiki.py page PATH [--wiki NAME] [--json] + python3 wiki.py taxonomy [--wiki NAME] + python3 wiki.py expand SLUG [--wiki NAME] + python3 wiki.py changesets [--wiki NAME] [--page PATH] [--action A] + [--trigger T] [--since ISO] [--until ISO] + [--limit N] [--offset N] [--json] + python3 wiki.py events [--wiki NAME] [--kind K] [--since ISO] [--until ISO] + [--limit N] [--offset N] [--json] + python3 wiki.py capture PATH ACTION SUMMARY [--trigger T] + [--source-event RAW_PATH]... [--wiki NAME] + +ACTION ∈ create | update | archive | delete +TRIGGER — conventionally ingest | query | lint | process-inbox | manual + (default: manual), but any kind the wiki declares via a + `type: event-type` page is valid; the taxonomy is the wiki's. + +Pass --source-event once per event that caused the change. Omitting it records +provenance as *unknown*, which is honest for a hand edit and a gap for an +ingest — `events` shows which sources have produced nothing. + +The wiki is resolved by NAME via ~/.hermes/wikis.yaml, else $WIKI_PATH, else +~/wiki — identical to the gateway, so a name here means the same wiki there. +""" +import argparse +import json +import os +import sys + + +def _bootstrap_imports(): + """Make tui_gateway.wiki_api and scripts.wiki_changeset importable. + + The skill ships under ~/.hermes/skills/...; the helper modules live in the + hermes-agent repo (and a copy under ~/.hermes/scripts). Probe both so the + skill works whether run from a checkout or an installed Hermes. + """ + candidates = [] + # Installed layout: ~/.hermes/scripts holds wiki_changeset.py + candidates.append(os.path.join(os.path.expanduser("~"), ".hermes", "scripts")) + # Repo layout: walk up looking for a dir containing tui_gateway/wiki_api.py + here = os.path.dirname(os.path.abspath(__file__)) + node = here + for _ in range(8): + if os.path.exists(os.path.join(node, "tui_gateway", "wiki_api.py")): + candidates.append(node) + candidates.append(os.path.join(node, "scripts")) + break + parent = os.path.dirname(node) + if parent == node: + break + node = parent + for d in candidates: + if d and d not in sys.path and os.path.isdir(d): + sys.path.insert(0, d) + + +_bootstrap_imports() + +try: + from tui_gateway import wiki_api +except Exception: # pragma: no cover - import shape varies by install + wiki_api = None +try: + import wiki_changeset +except Exception: # pragma: no cover + wiki_changeset = None + + +def _need_api(): + if wiki_api is None: + sys.exit( + "error: could not import tui_gateway.wiki_api — run this from a " + "hermes-agent checkout or an installed Hermes (~/.hermes/scripts)." + ) + + +def _print(obj, as_json): + if as_json: + print(json.dumps(obj, indent=2, ensure_ascii=False)) + return + print(_human(obj)) + + +def _human(obj) -> str: + """Compact human-readable rendering for the common shapes.""" + if isinstance(obj, dict) and "pages" in obj and "links" in obj: + lines = [f"{len(obj['pages'])} pages, {len(obj['links'])} links"] + for p in obj["pages"]: + lines.append(f" [{p.get('type','?'):11}] {p.get('path','')} — {p.get('title','')}") + return "\n".join(lines) + if isinstance(obj, dict) and "changesets" in obj: + lines = [f"{obj.get('total', len(obj['changesets']))} changesets " + f"(showing {len(obj['changesets'])}, offset {obj.get('offset', 0)})"] + for c in obj["changesets"]: + stats = c.get("diff_stats", {}) or {} + keys = c.get("source_event_keys") or [] + # Surface unknown provenance rather than leaving a blank: the point + # of the field is that its absence is visible. + provenance = f" ← {', '.join(keys)}" if keys else " ← unknown" + lines.append( + f" {c.get('timestamp','')} {c.get('action','?'):7} " + f"{c.get('page','')} +{stats.get('lines_added',0)}/-{stats.get('lines_removed',0)} " + f"[{c.get('trigger','')}] {c.get('summary','')}{provenance}" + ) + return "\n".join(lines) + if isinstance(obj, dict) and "events" in obj: + lines = [f"{obj.get('total', len(obj['events']))} events " + f"(showing {len(obj['events'])}, offset {obj.get('offset', 0)})"] + for e in obj["events"]: + caused = e.get("changesets") or [] + effect = f"→ {len(caused)} change(s)" if caused else "→ nothing yet" + lines.append( + f" {e.get('timestamp','') or '(undated)':20} [{e.get('kind','') or '?'}] " + f"{e.get('key','')} {effect}" + ) + return "\n".join(lines) + return json.dumps(obj, indent=2, ensure_ascii=False) + + +def cmd_scan(a): + _need_api() + _print(wiki_api.wiki_scan(wiki_path=_resolve(a)), a.json) + + +def cmd_page(a): + _need_api() + res = wiki_api.wiki_page(a.path, wiki_path=_resolve(a)) + if res is None: + sys.exit(f"error: page not found or outside wiki: {a.path}") + if a.json: + print(json.dumps(res, indent=2, ensure_ascii=False)) + else: + print(f"# {res['frontmatter'].get('title', a.path)} ({a.path})") + print(res["body"].strip()) + + +def cmd_taxonomy(a): + _need_api() + paths = wiki_api.wiki_flatten_taxonomy(wiki_path=_resolve(a)) + if not paths: + print("(no taxonomy.yaml — taxonomy is optional)") + return + print("\n".join(paths)) + + +def cmd_expand(a): + _need_api() + _print(wiki_api.wiki_expand_links(a.slug, wiki_path=_resolve(a)), True) + + +def cmd_changesets(a): + _need_api() + res = wiki_api.wiki_changesets( + wiki_path=_resolve(a), + page=a.page, + action=a.action, + trigger=a.trigger, + limit=a.limit, + offset=a.offset, + since=a.since, + until=a.until, + ) + _print(res, a.json) + + +def cmd_events(a): + _need_api() + res = wiki_api.wiki_events( + wiki_path=_resolve(a), + kind=a.kind, + limit=a.limit, + offset=a.offset, + since=a.since, + until=a.until, + ) + _print(res, a.json) + + +def cmd_capture(a): + if wiki_changeset is None: + sys.exit( + "error: could not import wiki_changeset — ensure scripts/wiki_changeset.py " + "is on the path (~/.hermes/scripts or a repo checkout)." + ) + wiki_path = _resolve(a) + res = wiki_changeset.wiki_capture_changeset( + page_path=a.path, + action=a.action, + summary=a.summary, + trigger=a.trigger, + source=a.source, + source_events=getattr(a, "source_events", None), + wiki_path=wiki_path, + ) + if isinstance(res, dict) and res.get("error"): + sys.exit(f"error: {res['error']}") + cid = res.get("id") if isinstance(res, dict) else None + keys = res.get("source_event_keys") or [] if isinstance(res, dict) else [] + # Domain metadata the write path can't infer (an opaque `article:<hash>` + # key carries no url on its own). The ingester supplies it here and we fill + # it onto each event record the capture just materialized — idempotent, so + # it enriches without clobbering anything an earlier call recorded. + event_kind = getattr(a, "event_kind", None) + event_url = getattr(a, "event_url", None) + if (event_kind or event_url) and hasattr(wiki_changeset, "wiki_record_event"): + for key in keys: + wiki_changeset.wiki_record_event( + key, kind=event_kind, source_url=event_url, + trigger=a.trigger, wiki_path=wiki_path, + ) + # Say when provenance is missing. A silent capture is how a KB accumulates + # unknowns nobody notices until the whole log is untrustworthy. + provenance = f" ← {', '.join(keys)}" if keys else " [provenance: unknown]" + print(f"captured changeset {cid or ''}: {a.action} {a.path}{provenance}".rstrip()) + + +def cmd_backfill_events(a): + if wiki_changeset is None or not hasattr(wiki_changeset, "wiki_backfill_events"): + sys.exit( + "error: wiki_changeset.wiki_backfill_events unavailable — the bundled " + "wiki_changeset.py may predate the materialized event log." + ) + res = wiki_changeset.wiki_backfill_events(wiki_path=_resolve(a)) + if getattr(a, "json", False): + print(json.dumps(res, indent=2, ensure_ascii=False)) + else: + print( + f"backfill-events: scanned {res.get('scanned_changesets', 0)} changesets, " + f"{res.get('distinct_keys', 0)} distinct source keys " + f"({res.get('created', 0)} created, {res.get('already_present', 0)} already present)" + ) + + +def _resolve(a): + """Resolve --wiki NAME to a path via the gateway's own resolver when present.""" + name = getattr(a, "wiki", None) + if not name: + return None + if wiki_api is not None: + return wiki_api.resolve_wiki(name) + if name.startswith("~") or name.startswith("/"): + return os.path.expanduser(name) + return name + + +def main(argv=None): + p = argparse.ArgumentParser(description="CLI over the native Hermes wiki API") + p.add_argument("--wiki", help="wiki name (wikis.yaml) or path; default resolves like the gateway") + sub = p.add_subparsers(dest="cmd", required=True) + + s = sub.add_parser("scan", help="graph structure: pages + links") + s.add_argument("--json", action="store_true"); s.set_defaults(func=cmd_scan) + + s = sub.add_parser("page", help="read one page by relative path") + s.add_argument("path"); s.add_argument("--json", action="store_true"); s.set_defaults(func=cmd_page) + + s = sub.add_parser("taxonomy", help="flat list of valid taxonomy paths") + s.set_defaults(func=cmd_taxonomy) + + s = sub.add_parser("expand", help="expand a page's integration_links") + s.add_argument("slug"); s.set_defaults(func=cmd_expand) + + s = sub.add_parser("changesets", help="query the edit timeline") + s.add_argument("--page"); s.add_argument("--action"); s.add_argument("--trigger") + s.add_argument("--since"); s.add_argument("--until") + s.add_argument("--limit", type=int, default=50); s.add_argument("--offset", type=int, default=0) + s.add_argument("--json", action="store_true"); s.set_defaults(func=cmd_changesets) + + s = sub.add_parser("events", help="ingestion event log — what caused wiki updates") + s.add_argument("--kind"); s.add_argument("--since"); s.add_argument("--until") + s.add_argument("--limit", type=int, default=200); s.add_argument("--offset", type=int, default=0) + s.add_argument("--json", action="store_true"); s.set_defaults(func=cmd_events) + + s = sub.add_parser("capture", help="record a changeset after writing a page") + s.add_argument("path"); s.add_argument("action", choices=["create", "update", "archive", "delete"]) + s.add_argument("summary") + # Deliberately not `choices=[...]`: what an event kind IS belongs to the + # wiki (a `type: event-type` page), not to this parser. A closed list here + # would mean adding an ingestion source requires editing this file. + s.add_argument("--trigger", default="manual", + help="event kind — conventionally ingest | query | lint | " + "process-inbox | manual, but any kind a wiki declares") + s.add_argument("--source", default="", help="legacy single source path") + s.add_argument("--source-event", action="append", default=[], dest="source_events", + metavar="RAW_PATH", + help="event that caused this change (repeatable) — " + "e.g. --source-event raw/articles/src.md") + # Domain metadata the write path can't infer from an opaque event key. + # Threaded onto the materialized event record for each --source-event. + s.add_argument("--event-kind", default=None, + help="kind for the caused event(s), e.g. github_pr | arxiv | snapshot") + s.add_argument("--event-url", default=None, + help="source URL for the caused event(s)") + s.set_defaults(func=cmd_capture) + + s = sub.add_parser("backfill-events", + help="materialize an event record per source key across all changesets (idempotent)") + s.add_argument("--json", action="store_true"); s.set_defaults(func=cmd_backfill_events) + + a = p.parse_args(argv) + a.func(a) + + +if __name__ == "__main__": + main()

diff --git hermes-agent/cron/jobs.py harness/cron/jobs.py index 7d91ec35ed1e6eb6e198e7d08b1efa19d6f0851c..bf7043e3d5187cf26ce494977c7d1cbee64c7f28 100644 --- hermes-agent/cron/jobs.py +++ harness/cron/jobs.py @@ -476,6 +476,12 @@ # half-paused record (enabled=true + state/paused_at) cannot render as # "paused" while the fleet is still live. See effective_job_state(). normalized["state"] = effective_job_state(normalized)   + # Backfill dataflow metadata for records written before these fields existed + # so graph/API consumers can read them unconditionally. + for df_field in ("inputs", "outputs", "side_effects", "source_files"): + value = normalized.get(df_field) + normalized[df_field] = value if isinstance(value, list) else [] + return normalized   @@ -1246,6 +1252,28 @@ ) return jobs + recovered   +def _record_configuration_changeset(jobs: List[Dict[str, Any]]) -> None: + """Record a changeset iff this save moved the dataflow *configuration*. + + Every cron mutation funnels through ``_save_jobs_unlocked``, which makes this + the one place a history can be written from and cover the CLI, the gateway, + the tool and the scheduler alike. It is also why the recording has to be + gated: most saves reaching here are scheduler bookkeeping (``last_run_at``, + ``next_run_at``, preflight flags, the due-scan sweep), and a row per save + would be a log of the tick loop. ``cron.changesets`` compares a digest taken + over the configuration form only and returns None when nothing moved. + + Best-effort by construction: a history that can fail a save would be a worse + feature than no history. + """ + try: + from cron.changesets import record_change + + record_change(jobs) + except Exception: + logger.debug("cron changeset not recorded for this save", exc_info=True) + + def _save_jobs_unlocked( jobs: List[Dict[str, Any]], *, @@ -1350,6 +1378,7 @@ # and deterministically clobber the nested create; it also races # a degraded sibling landing between replace and stat. Later # saves in this section simply take the full merge (fail-safe). _record_load_stamp(None) + _record_configuration_changeset(jobs) return   # Exhausted retries — last merge + write without another re-peek. @@ -1371,6 +1400,7 @@ atomic_replace(tmp_path, jobs_file) tmp_path = None _secure_file(jobs_file) _preserve_file_ownership(jobs_file, _stat_before) + _record_configuration_changeset(jobs) except BaseException: if tmp_path is not None: try: @@ -1395,6 +1425,35 @@ with _jobs_lock(): _save_jobs_unlocked(jobs, removed_ids=removed_ids, replace=replace)   +def _scheduler_save( + jobs: List[Dict[str, Any]], + *, + note: str = "", + removed_ids: Optional[Collection[str]] = None, +) -> None: + """``save_jobs`` for a write the *scheduler itself* makes. + + Most scheduler writes are runtime bookkeeping and record no changeset at all + (``_record_configuration_changeset``), but a few genuinely change the + configuration: a finished one-shot is disabled, a wedged claim is removed, a + contradictory half-paused record self-disables. "The scheduler disabled this + job" and "someone disabled this job" are exactly the distinction a history is + opened to settle, and only the call site knows which one this is — so the + attribution is bound here rather than inferred later from the shape of the + change. + + Attribution must never be able to block a save: an unimportable changeset + module falls through to a plain save. + """ + try: + from cron.changesets import use_changeset_origin + except Exception: + save_jobs(jobs, removed_ids=removed_ids) + return + with use_changeset_origin("scheduler", note=note): + save_jobs(jobs, removed_ids=removed_ids) + + def _normalize_workdir(workdir: Optional[str]) -> Optional[str]: """Normalize and validate a cron job workdir.   @@ -1536,6 +1595,837 @@ bool(job.get("no_agent")), )   +# --------------------------------------------------------------------------- +# Dataflow metadata (interflow graph) +# +# Crons never dispatch to each other — they communicate through data: one job +# writes an artifact, a later job wakes and reads it. So the cron graph is a +# *dataflow* graph, and cron→cron edges are INFERRED by matching one job's +# `cron-output:<id>` input to the producing job. Each job declares three typed +# `scheme:value` lists (the creating agent fills them — the mechanism can't see +# semantic reads/writes): +# inputs — what it reads (external sources + upstream cron output). Edges IN. +# outputs — consumable data it writes (join keys). Edges OUT. +# side_effects — terminal actions (telegram/pr/…). Sink leaves, never edges. +# The output-vs-side-effect split keeps the graph readable: outputs make edges, +# side effects make sinks. +_INPUT_SCHEMES = frozenset( + {"url", "http", "https", "file", "wiki", "postgres", "cron-output"} +) +_OUTPUT_SCHEMES = frozenset({"wiki", "file", "postgres"}) +_SIDE_EFFECT_SCHEMES = frozenset( + {"telegram", "slack", "email", "notify", "pr", "github", "webhook"} +) +# Service resources are graph metadata, not executable capabilities. Their +# namespace is intentionally open so a service can declare the boundary it +# actually hosts or consumes (http, redis, kafka, s3, grpc, …) without a Harness +# release adding that scheme to a global allowlist first. URI/RFC 3986 scheme +# syntax keeps refs canonical; the closed side-effect vocabulary below preserves +# the data-vs-terminal-action boundary. +_RESOURCE_SCHEME_RE = re.compile(r"^[A-Za-z][A-Za-z0-9+.-]*$") +_RELATIONSHIP_PREDICATE_RE = re.compile(r"^[a-z][a-z0-9_]{0,63}$") +_RESERVED_EDGE_PREDICATES = frozenset( + {"reads", "writes", "feeds", "hosts"} | set(_SIDE_EFFECT_SCHEMES) +) + +# `deliver` values that are genuine EXTERNAL side effects (vs local/origin +# in-band delivery, which is not a graph sink). Maps a deliver target to the +# side-effect scheme it mechanically implies, for the declared-vs-derived +# cross-check in _validate_dataflow_shape. +_DELIVER_SIDE_EFFECT_SCHEME = { + "telegram": "telegram", + "slack": "slack", + "email": "email", +} + + +# --------------------------------------------------------------------------- +# Source files (the code a job runs) +# +# A job's dataflow says what it reads and writes; its *source files* say which +# code does the reading and writing. Two are known mechanically — `script` and +# `monitor_script` — and the rest is agentic: a prompt that says "run +# ~/.hermes/scripts/ingest.py" or "use the classifier in indexing/x402.py" +# names code the mechanism can't see, so the creating agent declares it in +# `source_files`. `build_cron_graph` merges both layers onto the cron node +# (`source_files`) and resolves each path onto the file-browser roots +# (`tui_gateway.files_browse`) so Portal can open the file in place with +# `files.read`. +# +# Deliberately NOT part of the graph commitment: `cron/changesets.py` and +# Portal's `CronGraphDigest` hash the node row independently and must agree +# byte-for-byte, so growing that row is a coordinated change, not a side effect +# of adding metadata here. +def _normalize_source_files(values: Any, *, field_name: str = "source_files") -> List[str]: + """Normalize a declared source-file list to sorted, deduped path strings. + + Accepts a single string or a list; ``None``/empty → ``[]``. Entries are + filesystem paths, not typed refs: absolute, ``~``-relative, or relative + (relative resolves under ``HERMES_HOME/scripts/``, mirroring ``script``). A + leading ``file:`` scheme is tolerated and stripped so an agent that reaches + for the dataflow vocabulary isn't rejected for it. Only shape is enforced — + existence is *reported* by the graph (``exists``), never required at write + time, because a job is routinely declared before its script is committed. + """ + if values is None: + return [] + if isinstance(values, str): + raw: List[Any] = [values] + elif isinstance(values, (list, tuple)): + raw = list(values) + else: + raise ValueError( + f"{field_name} must be a string or list of path strings, got " + f"{type(values).__name__}." + ) + + seen: Set[str] = set() + out: List[str] = [] + for item in raw: + if not isinstance(item, str): + raise ValueError( + f"{field_name} entries must be path strings, got {type(item).__name__}." + ) + text = item.strip() + if text.lower().startswith("file:"): + text = text[5:].strip() + if not text: + continue + if "\x00" in text: + raise ValueError(f"{field_name} entry contains a NUL byte and cannot name a file.") + if "://" in text: + raise ValueError( + f"{field_name} entry '{text}' is a URL — source files are filesystem " + "paths; declare remote reads under `inputs` instead." + ) + if text not in seen: + seen.add(text) + out.append(text) + return sorted(out) + + +def _resolve_source_file_path(raw: str) -> Optional[Path]: + """The absolute path a source-file value names, by the scheduler's rule for + ``script``: absolute and ``~`` paths as-is, anything relative under + ``HERMES_HOME/scripts/``. ``None`` when the value can't be a path (NUL byte, + unexpandable ``~``) — same ingestion contract as ``cron.lifecycle_guard``. + """ + if not raw or "\x00" in raw: + return None + try: + path = Path(raw).expanduser() + except (ValueError, RuntimeError, OSError): + return None + if not path.is_absolute(): + path = get_hermes_home() / "scripts" / path + try: + return path.resolve() + except (RuntimeError, OSError): + return path + + +def _source_file_roots() -> Dict[str, Path]: + """The browse roots a source file can be opened under, name → absolute path. + + Sourced from the file browser so the two agree by construction; when that + module isn't importable (a stripped CLI install) the data home alone is + offered, which is where relative scripts live anyway. + """ + try: + from tui_gateway.files_browse import file_roots + + return {name: Path(root).resolve() for name, root in file_roots().items()} + except Exception: + try: + return {"hermes": get_hermes_home().resolve()} + except Exception: + return {} + + +def _browse_root_for(path: Path, roots: Dict[str, Path]) -> Tuple[Optional[str], Optional[str]]: + """``(root name, relative posix path)`` of the browse root containing ``path``. + + Prefers the *deepest* root when roots nest — the repo checkout routinely + lives inside ``~/.hermes`` — so a repo file is addressed as ``repo:…`` rather + than ``hermes:hermes-agent/…``. ``(None, None)`` when no root contains it: + the file is still listed on the node, just not openable from the client. + """ + best: Optional[Tuple[str, Path, Path]] = None + for name, root in roots.items(): + try: + rel = path.relative_to(root) + except ValueError: + continue + if best is None or len(root.parts) > len(best[1].parts): + best = (name, root, rel) + if best is None: + return None, None + return best[0], best[2].as_posix() + + +def job_source_files(job: Dict[str, Any]) -> List[Dict[str, Any]]: + """Every source file a job runs or declares, resolved for the graph. + + Mechanical first (``script`` → role ``script``, ``monitor_script`` → + ``monitor``), then the declared ``source_files`` (``declared``); deduped on + the resolved path with the mechanical role winning, since a declared entry + that is also the job's script is one file with the stronger claim. Each + entry carries the absolute ``path``, the value as ``declared``, its + ``role``, the browse ``root`` + ``rel`` path it can be read under (or + ``None``), and whether it ``exists`` on this host right now. + """ + candidates: List[Tuple[str, str]] = [] + if job.get("script"): + candidates.append((str(job["script"]), "script")) + if job.get("monitor_script"): + candidates.append((str(job["monitor_script"]), "monitor")) + for raw in job.get("source_files") or []: + if isinstance(raw, str) and raw.strip(): + candidates.append((raw, "declared")) + if not candidates: + return [] + + roots = _source_file_roots() + entries: List[Dict[str, Any]] = [] + seen: Set[str] = set() + for raw, role in candidates: + resolved = _resolve_source_file_path(raw) + if resolved is None: + continue + key = str(resolved) + if key in seen: + continue + seen.add(key) + root_name, rel = _browse_root_for(resolved, roots) + try: + exists = resolved.is_file() + except OSError: + exists = False + entries.append( + { + "path": key, + "declared": raw, + "role": role, + "root": root_name, + "rel": rel, + "exists": exists, + } + ) + return entries + + +def _normalize_resource_list( + values: Any, + *, + allowed_schemes: Optional[Collection[str]], + field_name: str, + forbidden_schemes: Collection[str] = (), + reserved_schemes: Collection[str] = (), +) -> List[str]: + """Normalize a dataflow resource list to sorted, deduped ``scheme:value``. + + ``allowed_schemes`` supplies a closed vocabulary (used by persisted cron + declarations and terminal side effects). ``None`` creates an open, + declaration-local resource namespace (used by live services). Open schemes + must still follow RFC 3986 syntax and cannot cross the explicit forbidden or + reserved boundaries. + """ + if values is None: + return [] + if isinstance(values, str): + raw: List[Any] = [values] + elif isinstance(values, (list, tuple)): + raw = list(values) + else: + raise ValueError( + f"{field_name} must be a string or list of 'scheme:value' strings, " + f"got {type(values).__name__}." + ) + + seen: Set[str] = set() + out: List[str] = [] + allowed_sorted = sorted(allowed_schemes) if allowed_schemes is not None else [] + forbidden = set(forbidden_schemes) + reserved = set(reserved_schemes) + for item in raw: + if not isinstance(item, str): + raise ValueError( + f"{field_name} entries must be 'scheme:value' strings, got " + f"{type(item).__name__}." + ) + text = item.strip() + if not text: + continue + if ":" not in text: + expected = ( + f"'scheme:value' with scheme in {allowed_sorted}" + if allowed_schemes is not None + else "'scheme:value'" + ) + raise ValueError( + f"{field_name} entry '{text}' is not a typed reference — expected " + f"{expected}." + ) + scheme, value = text.split(":", 1) + scheme = scheme.strip().lower() + value = value.strip() + if not _RESOURCE_SCHEME_RE.fullmatch(scheme): + raise ValueError( + f"{field_name} entry '{text}' has invalid scheme '{scheme}' — " + "use RFC 3986 scheme syntax (letter, then letters/digits/+.-)." + ) + if scheme in forbidden: + raise ValueError( + f"{field_name} entry '{text}' uses terminal side-effect scheme " + f"'{scheme}'. Declare terminal actions in service_side_effects." + ) + if scheme in reserved: + raise ValueError( + f"{field_name} entry '{text}' uses reserved scheme '{scheme}'; " + "only service inputs may reference an upstream cron output." + ) + if allowed_schemes is not None and scheme not in allowed_schemes: + raise ValueError( + f"{field_name} entry '{text}' has unknown scheme '{scheme}' — " + f"allowed schemes: {allowed_sorted}." + ) + if not value: + raise ValueError( + f"{field_name} entry '{text}' is missing a value after '{scheme}:'." + ) + canonical = f"{scheme}:{value}" + if canonical not in seen: + seen.add(canonical) + out.append(canonical) + return sorted(out) + + +def _normalize_service_relationships(values: Any) -> List[Dict[str, str]]: + """Normalize service subject-predicate-object relationship declarations. + + The declaring service is the implicit subject. Each entry supplies a stable + machine predicate and a typed object ref. These are topology/control facts, + not data reads/writes or terminal actions. + """ + if values is None: + return [] + if not isinstance(values, (list, tuple)): + raise ValueError("service relationships must be a list of objects.") + + seen: Set[tuple[str, str]] = set() + normalized: List[Dict[str, str]] = [] + for item in values: + if not isinstance(item, dict): + raise ValueError("service relationship entries must be objects.") + predicate = item.get("predicate") + object_ref = item.get("object") + if not isinstance(predicate, str) or not _RELATIONSHIP_PREDICATE_RE.fullmatch(predicate): + raise ValueError( + "service relationship predicate must match " + "[a-z][a-z0-9_]{0,63}." + ) + if predicate in _RESERVED_EDGE_PREDICATES: + raise ValueError( + f"service relationship predicate '{predicate}' is reserved by the " + "dataflow/side-effect graph vocabulary." + ) + unexpected = set(item) - {"predicate", "object"} + if unexpected: + raise ValueError( + "service relationship entries only accept predicate and object; " + f"unexpected keys: {sorted(unexpected)}." + ) + try: + objects = _normalize_resource_list( + [object_ref], + allowed_schemes=None, + field_name="service relationship typed object ref", + ) + except ValueError as exc: + raise ValueError(f"invalid service relationship typed object ref: {exc}") from exc + key = (predicate, objects[0]) + if key not in seen: + seen.add(key) + normalized.append({"predicate": predicate, "object": objects[0]}) + return sorted(normalized, key=lambda relation: (relation["predicate"], relation["object"])) + + +def _cron_output_input_ids(inputs: Any) -> List[str]: + """Job IDs referenced by ``cron-output:<id>`` inputs (the inferred edges).""" + if not isinstance(inputs, list): + return [] + ids: List[str] = [] + for ref in inputs: + if isinstance(ref, str) and ref.startswith("cron-output:"): + job_id = ref.split(":", 1)[1].strip() + if job_id: + ids.append(job_id) + return ids + + +def _validate_dataflow_shape(deliver: Optional[str], side_effects: List[str]) -> None: + """Cross-check declared side_effects against the derivable delivery sink. + + ``deliver`` mechanically implies an external side effect for the recognized + external channels (telegram/slack/email). When the job both delivers through + one of those AND declares side effects, the declaration must include the + matching ``scheme:`` entry so the graph's sinks match reality. Only enforced + when side_effects is declared — an empty declaration is the legacy/undeclared + path left to the tool-layer prompt and the store-wide doctor sweep. + """ + if not deliver or not side_effects: + return + implied = _DELIVER_SIDE_EFFECT_SCHEME.get(str(deliver).strip().lower()) + if not implied: + return + declared_schemes = {ref.split(":", 1)[0] for ref in side_effects} + if implied not in declared_schemes: + raise ValueError( + f"deliver='{deliver}' is an external side effect but side_effects " + f"declares no '{implied}:' entry. Add e.g. '{implied}:<target>' to " + f"side_effects so the dataflow graph's sinks match delivery." + ) + + +def _validate_dataflow_context( + context_from: Optional[List[str]], + inputs: List[str], +) -> None: + """Require every hard dep (``context_from``) to appear as a declared input. + + ``context_from`` is the mechanical hard dependency; the declared ``inputs`` + should be a superset, so the derived backbone is a subset of the declared + dataflow. Only enforced when inputs is declared (non-empty). + """ + if not context_from or not inputs: + return + declared_ids = set(_cron_output_input_ids(inputs)) + missing = [cid for cid in context_from if cid not in declared_ids] + if missing: + raise ValueError( + "context_from job(s) not declared as inputs: " + f"{', '.join(sorted(missing))}. Add 'cron-output:<id>' entries to " + "inputs so the dataflow graph reflects the context dependency." + ) + + +def _dataflow_dependency_edges(jobs: List[Dict[str, Any]]) -> Dict[str, List[str]]: + """Map job id → upstream job ids it consumes via ``cron-output`` inputs.""" + edges: Dict[str, List[str]] = {} + for job in jobs: + jid = job.get("id") + if not jid: + continue + edges[jid] = _cron_output_input_ids(job.get("inputs")) + return edges + + +def _find_dependency_cycle(edges: Dict[str, List[str]]) -> Optional[List[str]]: + """Return one cron-output dependency cycle as an id path, or None if acyclic.""" + WHITE, GRAY, BLACK = 0, 1, 2 + color: Dict[str, int] = {node: WHITE for node in edges} + stack: List[str] = [] + + def visit(node: str) -> Optional[List[str]]: + color[node] = GRAY + stack.append(node) + for dep in edges.get(node, []): + if dep not in edges: + continue # dangling target — reported by referential check + if color.get(dep) == GRAY: + return stack[stack.index(dep):] + [dep] + if color.get(dep, WHITE) == WHITE: + found = visit(dep) + if found: + return found + stack.pop() + color[node] = BLACK + return None + + for node in list(edges): + if color.get(node, WHITE) == WHITE: + found = visit(node) + if found: + return found + return None + + +def _reaches(start: str, target: str, edges: Dict[str, List[str]]) -> bool: + """True if ``target`` is reachable from ``start`` by following edges.""" + seen: Set[str] = set() + stack: List[str] = list(edges.get(start, [])) + while stack: + node = stack.pop() + if node == target: + return True + if node in seen: + continue + seen.add(node) + stack.extend(edges.get(node, [])) + return False + + +def _validate_candidate_dataflow( + candidate: Dict[str, Any], + jobs: List[Dict[str, Any]], +) -> None: + """Referential integrity + acyclicity for a single created/updated job. + + ``jobs`` is the current store (for create, ``candidate`` is not yet in it; + for update, the stale record is still at its index and is overridden here). + Every ``cron-output:<id>`` input must point at an existing job, and the + candidate's edges must not close a cycle. Run inside ``_jobs_lock()``. + """ + cid = candidate.get("id") + dep_ids = _cron_output_input_ids(candidate.get("inputs")) + if not dep_ids: + return + known_ids = {j.get("id") for j in jobs if j.get("id")} + known_ids.add(cid) + missing = [d for d in dep_ids if d not in known_ids] + if missing: + raise ValueError( + f"cron job declares cron-output input(s) for unknown job(s): " + f"{', '.join(sorted(missing))}. Use cronjob(action='list') to see " + "available jobs." + ) + edges = _dataflow_dependency_edges(jobs) + edges[cid] = dep_ids # reflect the candidate's (possibly new) edges + if _reaches(cid, cid, edges): + raise ValueError( + "cron dataflow dependency cycle: this job's cron-output inputs would " + "form a cycle (a job cannot transitively depend on its own output)." + ) + + +def validate_store() -> List[str]: + """Store-wide dataflow consistency sweep for ``hermes cron doctor`` / CI. + + Non-raising: returns a list of human-readable issue strings (empty = clean). + Catches drift that per-job create/update validation can't — a producer + deleted after consumers referenced it, a cycle formed across independent + edits, or a malformed stored ref from a hand-edited jobs.json. + """ + jobs = load_jobs() + issues: List[str] = [] + known_ids = {j.get("id") for j in jobs if j.get("id")} + edges: Dict[str, List[str]] = {} + for job in jobs: + jid = job.get("id") or "<no-id>" + for field, schemes in ( + ("inputs", _INPUT_SCHEMES), + ("outputs", _OUTPUT_SCHEMES), + ("side_effects", _SIDE_EFFECT_SCHEMES), + ): + try: + _normalize_resource_list( + job.get(field), allowed_schemes=schemes, field_name=field + ) + except ValueError as exc: + issues.append(f"job '{jid}': {exc}") + dep_ids = _cron_output_input_ids(job.get("inputs")) + edges[jid] = dep_ids + for dep in dep_ids: + if dep not in known_ids: + issues.append( + f"job '{jid}': cron-output input references unknown job '{dep}'." + ) + cycle = _find_dependency_cycle(edges) + if cycle: + issues.append("cron dataflow dependency cycle: " + " → ".join(cycle)) + return issues + + +def normalize_service_declaration( + name: Any, + description: Any, + inputs: Any = None, + outputs: Any = None, + side_effects: Any = None, + relationships: Any = None, + source_files: Any = None, + code_control: Any = None, +) -> Dict[str, Any]: + """Validate & normalize a long-running service's dataflow declaration. + + A service (a dashboard, an API — anything the agent backgrounds and that + outlives the turn) declares itself so it appears in the cron interflow graph + alongside the crons it shares data with. Service input/output schemes are an + open, declaration-local namespace: any valid typed resource (HTTP, Redis, + Kafka, S3, a domain-specific boundary, etc.) can join graph nodes without a + Harness release adding it to a global vocabulary. Terminal action schemes + remain reserved for ``side_effects``, whose vocabulary is deliberately closed. + ``relationships`` adds subject-predicate-object topology facts: the service is + the implicit subject, each machine predicate is explicit, and each object is a + typed ref. Relationships never imply data movement or a terminal action. + ``source_files`` is the code behind the service — filesystem paths (not typed + refs) to the scripts/modules it runs, mirroring a cron's ``source_files`` so + the graph node is browsable. Resolved and reported the same way: files under + a browse root (typically the service's repo checkout) become openable; a path + that doesn't exist yet is listed, not rejected. + Both a non-empty ``name`` and a non-empty ``description`` are REQUIRED — the + description is markdown surfaced in Portal's node detail card, so expanding a + service node always answers "what is this / what does it do" rather than + showing a bare id. Raises ValueError on any malformed field. + """ + if not isinstance(name, str) or not name.strip(): + raise ValueError("service name is required and must be a non-empty string.") + if not isinstance(description, str) or not description.strip(): + raise ValueError( + "service description is required and must be a non-empty markdown " + "string — it is shown in the graph's node detail card so the service " + "is self-explanatory. Describe what the service is and what it does." + ) + declaration: Dict[str, Any] = { + "name": name.strip(), + "description": description.strip(), + "inputs": _normalize_resource_list( + inputs, + allowed_schemes=None, + forbidden_schemes=_SIDE_EFFECT_SCHEMES, + field_name="service inputs", + ), + "outputs": _normalize_resource_list( + outputs, + allowed_schemes=None, + forbidden_schemes=_SIDE_EFFECT_SCHEMES, + reserved_schemes={"cron-output"}, + field_name="service outputs", + ), + "side_effects": _normalize_resource_list( + side_effects, + allowed_schemes=_SIDE_EFFECT_SCHEMES, + field_name="service side_effects", + ), + "source_files": _normalize_source_files( + source_files, field_name="service source_files" + ), + } + normalized_relationships = _normalize_service_relationships(relationships) + if normalized_relationships: + declaration["relationships"] = normalized_relationships + if code_control is not None: + from tools.service_code_control import normalize_service_code_control + + declaration["code_control"] = normalize_service_code_control(code_control) + return declaration + + +def build_cron_graph( + jobs: Optional[List[Dict[str, Any]]] = None, + services: Optional[List[Dict[str, Any]]] = None, +) -> Dict[str, List[Dict[str, Any]]]: + """Assemble the cron interflow dataflow graph as ``{nodes, edges}``. + + Crons never dispatch to each other — they communicate through data — so this + is a dataflow graph. Edges use the wiki graph's typed-edge shape + (``{source, target, type}``) so Portal's typed-edge renderer can be reused. + + ``services`` are long-running processes (a dashboard, an API) that are NOT + crons but participate in the same dataflow: each declares ``inputs`` / + ``outputs`` / ``side_effects`` in the identical ``scheme:value`` vocabulary, + so a service reading ``postgres:analytics.events`` dedupes onto the very node + a cron writing that ref produces — service and cron meet at the shared store. + Each service dict carries ``id`` (its tracked-process session id), ``label``, + ``description`` (markdown, for the Portal detail card) and the three dataflow + lists. They are rendered as ``service`` nodes; liveness is the caller's + concern (only currently-running services should be passed in). + + Node kinds: + - ``cron`` one per job (node id = job id). + - ``service`` one per live long-running process that declared dataflow. + - ``source`` a resource READ by ≥1 cron/service and written by none — an + external/upstream input (url/http/https/file/wiki/postgres). + - ``artifact`` a resource WRITTEN by ≥1 cron/service (consumable output). + When another cron reads the same ref, that shared node IS + the cron→cron link (outputs make edges). + - ``sink`` a side_effect target — a terminal action (side effects make + sinks, never edges onward). + + Edge types: + - ``reads`` source/artifact → cron + - ``writes`` cron → artifact + - ``feeds`` cron → cron — a ``cron-output:<id>`` input; the explicit + dependency backbone (mirrors ``context_from``). + - ``<scheme>`` cron → sink — the delivered action kind (telegram/pr/…). + + Resource/sink node ids are the ``scheme:value`` ref itself (always contains + a colon); cron node ids are the bare job id — so the two id spaces never + collide, and a shared data ref naturally dedupes to one node. + """ + if jobs is None: + jobs = [_normalize_job_record(job) for job in load_jobs()] + services = services or [] + + job_ids = {job.get("id") for job in jobs if job.get("id")} + + # A resource written by any cron OR service is an artifact even if also read + # elsewhere — a service writing postgres makes the same store a cron reads. + produced: Set[str] = set() + for producer in (*jobs, *services): + for ref in producer.get("outputs") or []: + produced.add(ref) + + resource_kind: Dict[str, str] = {} + + def _ensure_resource(ref: str, kind: str) -> None: + existing = resource_kind.get(ref) + if existing is None: + resource_kind[ref] = kind + elif existing == "source" and kind == "artifact": + # A produced resource outranks a bare read: promote source→artifact. + resource_kind[ref] = "artifact" + + relationship_objects: Set[str] = set() + + nodes: List[Dict[str, Any]] = [] + edges: List[Dict[str, Any]] = [] + + for job in jobs: + jid = job.get("id") + if not jid: + continue + nodes.append( + { + "id": jid, + "kind": "cron", + "type": "cron", + "label": job.get("name") or jid, + "schedule": job.get("schedule_display"), + "enabled": bool(job.get("enabled", True)), + "state": job.get("state"), + "uses_llm": not bool(job.get("no_agent")), + "last_status": job.get("last_status"), + "deliver": job.get("deliver"), + # The code behind the node — mechanical script fields merged + # with the declared list, each resolved onto a browse root so + # the client can open it. Node metadata, not nodes: a script is + # what a job *is made of*, not something it exchanges data with. + "source_files": job_source_files(job), + } + ) + + for ref in job.get("inputs") or []: + if ref.startswith("cron-output:"): + upstream = ref.split(":", 1)[1].strip() + if upstream in job_ids: + edges.append( + {"source": upstream, "target": jid, "type": "feeds"} + ) + # A dangling upstream is surfaced by validate_store(), not drawn. + else: + _ensure_resource(ref, "artifact" if ref in produced else "source") + edges.append({"source": ref, "target": jid, "type": "reads"}) + + for ref in job.get("outputs") or []: + _ensure_resource(ref, "artifact") + edges.append({"source": jid, "target": ref, "type": "writes"}) + + for ref in job.get("side_effects") or []: + _ensure_resource(ref, "sink") + edges.append( + {"source": jid, "target": ref, "type": ref.split(":", 1)[0]} + ) + + # Services join the same dataflow: a service reading/writing a ref meets the + # cron on the shared resource node. Same edge grammar as crons — a + # cron-output input becomes a feeds edge, everything else reads/writes/sink. + for service in services: + sid = service.get("id") + if not sid: + continue + node = { + "id": sid, + "kind": "service", + "type": "service", + "label": service.get("label") or sid, + "description": service.get("description") or "", + # The code behind the service — same resolver as cron nodes, so a + # file under a browse root (its repo checkout) is openable. Node + # metadata, not nodes, and — like services themselves — outside the + # configuration digest. + "source_files": job_source_files(service), + } + if isinstance(service.get("health"), dict): + node["health"] = service["health"] + code_control = service.get("code_control") + if isinstance(code_control, dict) and code_control.get("status") == "verified": + node["code_control"] = code_control + # A change-token for the code knowledge graph behind this service: a + # cheap digest of its source files (no reads) that flips when the code + # changes, so the client knows when to refetch the graph. The graph + # itself is built lazily by ``code.graph``; this only references it. + from cron.code_graph import service_code_graph_stamp + + stamp = service_code_graph_stamp(node["source_files"]) + if stamp: + node["code_graph"] = {"ref": sid, "digest": stamp} + nodes.append(node) + for ref in service.get("inputs") or []: + if ref.startswith("cron-output:"): + upstream = ref.split(":", 1)[1].strip() + if upstream in job_ids: + edges.append({"source": upstream, "target": sid, "type": "feeds"}) + else: + _ensure_resource(ref, "artifact" if ref in produced else "source") + edges.append({"source": ref, "target": sid, "type": "reads"}) + for ref in service.get("outputs") or []: + _ensure_resource(ref, "artifact") + edges.append({"source": sid, "target": ref, "type": "writes"}) + for ref in service.get("side_effects") or []: + _ensure_resource(ref, "sink") + edges.append({"source": sid, "target": ref, "type": ref.split(":", 1)[0]}) + for relationship in service.get("relationships") or []: + object_ref = relationship["object"] + relationship_objects.add(object_ref) + edges.append({ + "source": sid, + "target": object_ref, + "type": relationship["predicate"], + "class": "relationship", + }) + if isinstance(code_control, dict) and code_control.get("status") == "verified": + repository_ref = f"github:{code_control['repository']}" + revision_ref = f"git:{code_control['revision']}" + pull = code_control["pull_request"] + pull_ref = f"pr:{code_control['repository']}#{pull['number']}" + for target, predicate in ( + (repository_ref, "source_repository"), + (pull_ref, "released_via"), + (revision_ref, "runs_revision"), + ): + relationship_objects.add(target) + edges.append({ + "source": sid, + "target": target, + "type": predicate, + "class": "control", + }) + + for ref in sorted(resource_kind): + scheme, _, value = ref.partition(":") + nodes.append( + { + "id": ref, + "kind": resource_kind[ref], + "type": scheme, + "label": value or ref, + } + ) + + existing_ids = {node["id"] for node in nodes} + for ref in sorted(relationship_objects - existing_ids): + scheme, _, value = ref.partition(":") + nodes.append({ + "id": ref, + "kind": "object", + "type": scheme, + "label": value or ref, + }) + + return {"nodes": nodes, "edges": edges} + + def _validate_job_mode_invariants( monitor_script: Optional[str], monitor_url: Optional[str], @@ -1586,6 +2476,10 @@ no_agent: bool = False, attach_to_session: Optional[bool] = None, monitor_script: Optional[str] = None, monitor_url: Optional[str] = None, + inputs: Optional[Union[str, List[str]]] = None, + outputs: Optional[Union[str, List[str]]] = None, + side_effects: Optional[Union[str, List[str]]] = None, + source_files: Optional[Union[str, List[str]]] = None, ) -> Dict[str, Any]: """ Create a new cron job. @@ -1643,6 +2537,20 @@ with ``monitor_url``; incompatible with ``no_agent=True``. monitor_url: Optional http(s) URL used as the monitor source instead of a script — fetched with a bounded GET each tick. Same hash-suppression semantics as ``monitor_script``. + inputs: Optional dataflow declaration — typed ``scheme:value`` refs for + what the job reads (schemes: url/http/https/file/wiki/cron-output). + ``cron-output:<id>`` inputs form the inferred cron→cron edges. + outputs: Optional typed refs for consumable data the job writes + (schemes: wiki/file) — the join keys another job's input matches. + side_effects: Optional typed refs for terminal actions the job performs + (schemes: telegram/slack/email/notify/pr/github/webhook) — graph + sink leaves, not edges. See the dataflow metadata block above. + source_files: Optional paths to the code this job runs or relies on + beyond ``script`` / ``monitor_script`` (which are picked up + mechanically) — e.g. a module the prompt tells the agent to + execute. Absolute, ``~``, or relative to ~/.hermes/scripts/. + Shown on the job's graph node so the file can be opened from + Portal. See the source-files block above.   Returns: The created job dict @@ -1701,6 +2609,22 @@ context_from = [str(j).strip() for j in context_from if str(j).strip()] or None else: context_from = None   + # Normalize + validate dataflow metadata (interflow graph). Shape and the + # declared-vs-derived cross-checks are pure per-job invariants; referential + # integrity + acyclicity need the whole store and run inside the lock below. + normalized_inputs = _normalize_resource_list( + inputs, allowed_schemes=_INPUT_SCHEMES, field_name="inputs" + ) + normalized_outputs = _normalize_resource_list( + outputs, allowed_schemes=_OUTPUT_SCHEMES, field_name="outputs" + ) + normalized_side_effects = _normalize_resource_list( + side_effects, allowed_schemes=_SIDE_EFFECT_SCHEMES, field_name="side_effects" + ) + _validate_dataflow_shape(deliver, normalized_side_effects) + _validate_dataflow_context(context_from, normalized_inputs) + normalized_source_files = _normalize_source_files(source_files) + prompt_text = _coerce_job_text(prompt)   # Reject cron jobs that schedule gateway-lifecycle commands. Prevents @@ -1756,6 +2680,14 @@ # Hash-suppression state for monitor jobs: {"last_output_hash": ..., # "last_changed_at": ...}. None until the first monitor tick. "monitor_state": None, "context_from": context_from, + # Dataflow metadata (interflow graph). Typed scheme:value lists; see the + # dataflow block near _validate_dataflow_shape. + "inputs": normalized_inputs, + "outputs": normalized_outputs, + "side_effects": normalized_side_effects, + # Code the job runs beyond `script` / `monitor_script`; see the + # source-files block near job_source_files. + "source_files": normalized_source_files, "schedule": parsed_schedule, "schedule_display": parsed_schedule.get("display", schedule), "repeat": { @@ -1786,6 +2718,7 @@ job["attach_to_session"] = normalized_attach   with _jobs_lock(): jobs = load_jobs() + _validate_candidate_dataflow(job, jobs) jobs.append(job) save_jobs(jobs)   @@ -1908,6 +2841,38 @@ updated.get("monitor_url") or None, bool(updated.get("no_agent")), _upd_script or None, ) + # Normalize + re-validate dataflow metadata so create-time + # invariants can't be bypassed through the update door. Normalize + # any dataflow field present in this update; re-run the shape / + # context cross-checks when a dataflow field OR its derived source + # (deliver / context_from) changes. + _dataflow_fields = {"inputs", "outputs", "side_effects"} + for _df_field, _df_schemes in ( + ("inputs", _INPUT_SCHEMES), + ("outputs", _OUTPUT_SCHEMES), + ("side_effects", _SIDE_EFFECT_SCHEMES), + ): + if _df_field in updates: + updated[_df_field] = _normalize_resource_list( + updates[_df_field], + allowed_schemes=_df_schemes, + field_name=_df_field, + ) + if _dataflow_fields.intersection(updates) or { + "context_from", + "deliver", + }.intersection(updates): + _validate_dataflow_shape( + updated.get("deliver"), updated.get("side_effects") or [] + ) + _validate_dataflow_context( + updated.get("context_from") or None, updated.get("inputs") or [] + ) + if "inputs" in updates: + _validate_candidate_dataflow(updated, jobs) + if "source_files" in updates: + updated["source_files"] = _normalize_source_files(updates["source_files"]) + schedule_changed = "schedule" in updates inference_fields_changed = bool( {"provider", "model", "base_url", "no_agent"}.intersection(updates) @@ -2185,7 +3150,7 @@ # COMPLETED_ONESHOT_RETENTION_DAYS. job["enabled"] = False job["state"] = "completed" job["next_run_at"] = None - save_jobs(jobs) + _scheduler_save(jobs, note="repeat limit reached") return   # Compute next run @@ -2220,7 +3185,7 @@ job["state"] = "completed" elif job.get("state") != "paused": job["state"] = "scheduled"   - save_jobs(jobs) + _scheduler_save(jobs, note="after a job run") return   logger.warning("mark_job_run: job_id %s not found, skipping save", job_id) @@ -2311,7 +3276,7 @@ # deleting the job and its final status/delivery error. job["enabled"] = False job["state"] = "completed" job["next_run_at"] = None - save_jobs(jobs) + _scheduler_save(jobs, note="dispatch limit reached") logger.info( "Job '%s': dispatch limit reached (%d/%d) — marking completed", job.get("name", job.get("id", "?")), @@ -2324,7 +3289,11 @@ # completed (#73973) — a genuinely wedged claim. Remove it so # it stops appearing as due, and leave an operator-visible # diagnostic instead of vanishing silently. jobs.pop(i) - save_jobs(jobs, removed_ids={job_id}) + _scheduler_save( + jobs, + note="wedged one-shot removed", + removed_ids={job_id}, + ) _write_wedged_oneshot_diagnostic(job) logger.info( "Job '%s': dispatch limit reached (%d/%d) — removing", @@ -2992,7 +3961,11 @@ ) continue   if needs_save: - save_jobs(raw_jobs, removed_ids=intentionally_removed or None) + _scheduler_save( + raw_jobs, + note="due-job scan", + removed_ids=intentionally_removed or None, + )   return due