diff --git a/hermes_cli/main.py b/hermes_cli/main.py index 90fe76c3f2bd..f5540725eccb 100644 --- a/hermes_cli/main.py +++ b/hermes_cli/main.py @@ -296,6 +296,7 @@ def _try_termux_ultrafast_version() -> bool: from hermes_cli.subcommands.webhook import build_webhook_parser from hermes_cli.subcommands.hooks import build_hooks_parser from hermes_cli.subcommands.doctor import build_doctor_parser +from hermes_cli.subcommands.persist_smoke import build_persist_smoke_parser from hermes_cli.subcommands.security import build_security_parser from hermes_cli.subcommands.dump import build_dump_parser from hermes_cli.subcommands.debug import build_debug_parser @@ -4380,6 +4381,13 @@ def cmd_doctor(args): run_doctor(args) +def cmd_persist_smoke(args): + """Smoke-test the governed persistence funnel (pre_persist_write).""" + from hermes_cli.persist_smoke import run_persist_smoke + + run_persist_smoke(args) + + def cmd_security(args): """Dispatch `hermes security `.""" sub = getattr(args, "security_command", None) @@ -12709,7 +12717,7 @@ def _build_provider_choices() -> list[str]: "dump", "fallback", "gateway", "hooks", "import", "insights", "gui", "desktop", "kanban", "login", "logout", "logs", "lsp", "mcp", "memory", "migrate", "moa", "journey", "memory-graph", "learning", - "model", "pairing", "pets", "plugins", "portal", "postinstall", "profile", + "model", "pairing", "persist-smoke", "pets", "plugins", "portal", "postinstall", "profile", "project", "proxy", "prompt-size", "send", "sessions", "setup", @@ -13541,6 +13549,12 @@ def _dispatch_secrets(args): # noqa: ANN001 # ========================================================================= build_doctor_parser(subparsers, cmd_doctor=cmd_doctor) + # ========================================================================= + # persist-smoke command — governed persistence funnel smoke test + # (parser built in hermes_cli/subcommands/persist_smoke.py) + # ========================================================================= + build_persist_smoke_parser(subparsers, cmd_persist_smoke=cmd_persist_smoke) + # ========================================================================= # security command — on-demand supply-chain audit # ========================================================================= diff --git a/hermes_cli/persist_smoke.py b/hermes_cli/persist_smoke.py new file mode 100644 index 000000000000..bb92c4dcfa07 --- /dev/null +++ b/hermes_cli/persist_smoke.py @@ -0,0 +1,248 @@ +"""``hermes persist-smoke`` -- operator-facing smoke test for the governed +persistence funnel (``agent.persist_boundary.governed_persist``). + +Confirms, in one shot, that: + +1. plugin discovery actually finds a policy enforcer that registered the + required ``pre_persist_write`` hook (distinguishing "no agent-lineage + plugin discovered at all" from "discovered, but this hook is missing -- + a stale plugin version?"), and +2. a real probe write routed through ``governed_persist`` comes back + genuinely staged (governed) rather than silently falling through to a + canonical write, OR falling through to ``governed_persist``'s own + decision-less local fallback (``persist_boundary._stage_local``) -- which + also reports ``staged=True`` but reflects an unreachable/malformed + enforcer, not a real policy decision, and must not be reported as green + either (a mid-crash worker is not a healthy one). + +This is deliberately a read-mostly diagnostic. On a genuine "staged" outcome +nothing new lands on the canonical path (agent-lineage's own quarantine is +outside this fork's filesystem view); on "denied" nothing is written either. +The "local_fallback" outcome writes durably under this fork's own +``$HERMES_HOME/persist-quarantine-local/`` -- that's ``governed_persist``'s +concern, not this command's, and is left alone. Only the "passthrough" +outcome -- no enforcer wired at all, so ``governed_persist`` performs the +pre-cutover canonical write itself -- actually creates a file under this +command's own probe path, and this command deletes it (and the now-empty +``persist-smoke/`` directory it lived in) before returning, so an ungoverned +smoke run leaves the working tree exactly as it found it. + +Plugin discovery and registry introspection run inside a broad +``try/except``: a fail-loud duplicate-required-plugin-name abort +(``RequiredPluginError``) or any other discovery-time exception is reported +as a clean ``discovery-failed`` JSON diagnostic (exit 1) instead of an +uncaught traceback -- the same condition would abort real Hermes startup, so +this smoke's job is to say so legibly, not to crash trying to say it. + +Not to be confused with agent-lineage's OWN ``hermes-persist smoke`` +(``framework/tools/cli.py`` in the agent-lineage repo): that is a black-box +harness which shells out to a real ``hermes`` binary (``--hermes-bin``) and +judges a live chat turn from NEW hash-chained policy-ledger events written +during it. This command is the fork-side white box: it runs in-process, +inspects the plugin manager's own registries directly, and drives +``governed_persist`` with a synthetic probe rather than a real model turn. +""" + +from __future__ import annotations + +import json +import sys +from datetime import datetime, timezone +from pathlib import Path +from typing import Any +from uuid import uuid4 + +__all__ = ["run_persist_smoke"] + +_HOOK_NAME = "pre_persist_write" +_PLUGIN_NAME = "agent-lineage" +_PROBE_DIR = "persist-smoke" +_PROBE_PATH = f"{_PROBE_DIR}/probe.md" + + +def _probe_content() -> bytes: + stamp = datetime.now(timezone.utc).isoformat() + return f"hermes persist-smoke probe @ {stamp}\n".encode("utf-8") + + +def _plugin_discovered(manager: Any, name: str) -> bool: + """True when *name* was discovered by plugin scanning at all -- loaded, + disabled, or otherwise -- as long as it showed up in the registry.""" + plugins = getattr(manager, "_plugins", None) + if not isinstance(plugins, dict): + return False + if name in plugins: + return True + return any( + getattr(getattr(loaded, "manifest", None), "name", None) == name + for loaded in plugins.values() + ) + + +def _required_hook_registered(manager: Any, hook_name: str) -> bool: + required_hooks = getattr(manager, "_required_hooks", None) + if not isinstance(required_hooks, dict): + return False + return bool(required_hooks.get(hook_name)) + + +def _emit(payload: dict, *, as_json: bool) -> None: + if as_json: + print(json.dumps(payload)) + return + mode = payload.get("mode", "?") + if payload.get("ok"): + print( + f"OK: persist-smoke probe was genuinely staged " + f"(mode={mode}, digest={payload.get('digest')})" + ) + return + detail = ( + payload.get("message") + or payload.get("reason") + or payload.get("error") + or "" + ) + print(f"FAIL: persist-smoke ({mode}): {detail}") + + +def run_persist_smoke(args) -> None: + """Implementation of ``hermes persist-smoke``. + + Exits 0 ONLY when a probe write through ``governed_persist`` comes back + genuinely staged -- ``staged=True``, ``denied=False``, and no fallback + message attached. Every other outcome exits 1: the required hook isn't + registered (``no_hook``), plugin discovery/registry introspection itself + raised (``discovery-failed``), the write was refused (``denied``), + ``governed_persist`` fell through to its own decision-less local staging + because the enforcer was unreachable or malformed (``local_fallback``), + or no enforcer is wired at all so the pre-cutover canonical write ran + (``passthrough``). + """ + as_json = bool(getattr(args, "json", False)) + + from hermes_cli.plugins import discover_plugins, get_plugin_manager + + try: + discover_plugins(force=True) + manager = get_plugin_manager() + hooks = sorted(getattr(manager, "_required_hooks", None) or {}) + except Exception as exc: + # Whatever would abort real Hermes startup (a fail-loud duplicate + # required-plugin-name RequiredPluginError, or any other discovery- + # time failure) must not crash this diagnostic too -- report it as + # a legible, still-valid-JSON verdict instead of a bare traceback. + _emit( + { + "ok": False, + "mode": "discovery-failed", + "error": str(exc), + "hint": "hermes startup itself would abort — fix plugin discovery first", + }, + as_json=as_json, + ) + sys.exit(1) + + if not _required_hook_registered(manager, _HOOK_NAME): + if _plugin_discovered(manager, _PLUGIN_NAME): + message = ( + f"'{_PLUGIN_NAME}' plugin was discovered but registered no " + f"required '{_HOOK_NAME}' hook -- stale plugin version?" + ) + else: + message = ( + f"no '{_PLUGIN_NAME}' plugin was discovered -- install it " + f"under ~/.hermes/plugins/{_PLUGIN_NAME}/ and add it to " + f"plugins.enabled / plugins.required in config.yaml" + ) + _emit( + {"ok": False, "hooks": hooks, "mode": "no_hook", "message": message}, + as_json=as_json, + ) + sys.exit(1) + + from agent.persist_boundary import governed_persist + + # A unique session id per invocation, never a shared fixed default. + # governed_persist threads meta["session_id"] straight through to the + # pre_persist_write hook, and agent-lineage's durable budgets meter + # external_side_effects PER SESSION with a cap (currently 100). A + # periodic/cron-driven smoke that reused one fixed session across every + # run would eventually exhaust that budget purely from its OWN + # accumulated history and start self-denying -- a false "denied" that + # says nothing about whether governance is actually working right now. + # Minting a fresh session per probe keeps the smoke's own call history + # from ever being the thing that fails it. + session_id = f"persist-smoke:{uuid4().hex[:12]}" + result = governed_persist( + "memory", + _PROBE_PATH, + _probe_content(), + {"origin": "persist-smoke", "session_id": session_id}, + ) + + if result.denied: + _emit( + {"ok": False, "mode": "denied", "message": result.message}, + as_json=as_json, + ) + sys.exit(1) + + if result.staged and not result.message: + # Genuine governance: a real enforcer made a real "stage this" + # decision, with nothing left unsaid. + _emit( + { + "ok": True, + "hooks": hooks, + "staged": True, + "digest": result.digest, + "mode": "staged", + }, + as_json=as_json, + ) + return + + if result.staged: + # governed_persist's OWN decision-less local fallback + # (persist_boundary._stage_local) also reports staged=True -- for + # an unreachable hook, a non-dict directive, or a registered + # enforcer's malformed bare "allow" -- but it is not a policy + # decision at all, just durable loss-prevention while governance + # itself is broken. The non-empty message is the tell; reporting + # this as green would invert the whole point of the smoke. + _emit( + { + "ok": False, + "mode": "local_fallback", + "digest": result.digest, + "message": result.message, + }, + as_json=as_json, + ) + sys.exit(1) + + # Neither denied nor staged: governed_persist performed the real + # pre-cutover canonical write (no enforcer registered for this hook). + # Clean up the probe file -- and the now-empty persist-smoke/ directory + # it lived in -- so an ungoverned run leaves the working tree exactly + # as it found it, then fail the smoke -- it exists precisely to prove + # governance is wired, and here it isn't. + probe = Path(_PROBE_PATH) + try: + probe.unlink(missing_ok=True) + except OSError: + pass + try: + probe.parent.rmdir() + except OSError: + pass + _emit( + { + "ok": False, + "mode": "passthrough", + "reason": "no enforcer registered — governed config required for the smoke", + }, + as_json=as_json, + ) + sys.exit(1) diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py index 5a000e1b15e9..84da5f897131 100644 --- a/hermes_cli/plugins.py +++ b/hermes_cli/plugins.py @@ -1503,9 +1503,9 @@ def _discover_and_load_inner(self) -> None: # don't collide even when both manifests say ``name: openai``. disabled = _get_disabled_plugins() enabled = _get_enabled_plugins() # None = opt-in default (nothing enabled) - winners: Dict[str, PluginManifest] = {} - for manifest in manifests: - winners[manifest.key or manifest.name] = manifest + winners: Dict[str, PluginManifest] = self._resolve_manifest_winners( + manifests, required + ) for manifest in winners.values(): lookup_key = manifest.key or manifest.name @@ -1601,6 +1601,77 @@ def _discover_and_load_inner(self) -> None: sum(1 for p in self._plugins.values() if p.enabled), ) + def _resolve_manifest_winners( + self, manifests: List[PluginManifest], required: Set[str] + ) -> Dict[str, PluginManifest]: + """Collapse discovered manifests to one winner per lookup key. + + Later *sources* legitimately override earlier ones on key collision + (a user plugin replacing a bundled one, a project plugin replacing a + user one — see the module docstring) — at most one manifest per + source for a given key, so that case just keeps the last one seen. + + A collision with more than one manifest from the *same* source for + the same key is a different, ambiguous situation: two plugin + directories independently declared the same manifest ``name`` (keys + are path-derived and fall back to the bare ``name`` for flat + top-level plugins, so this is exactly "two directories, same + declared name"). There is no principled way to pick a winner, and + silently doing so by directory-iteration order (the old behavior) + can silently shadow the plugin an operator actually intended to + run. Handle it loudly and deterministically instead: + + * if the colliding name/key is in ``plugins.required``, abort + startup with a :class:`RequiredPluginError` naming every + conflicting directory — a mandatory policy enforcer must never + load from an ambiguous source; + * otherwise, refuse to load *either* copy and log one warning + naming every conflicting directory, so the operator can fix it + without either copy silently winning. + + Unique names/keys are entirely unaffected. + """ + by_key: Dict[str, List[PluginManifest]] = {} + for manifest in manifests: + by_key.setdefault(manifest.key or manifest.name, []).append(manifest) + + winners: Dict[str, PluginManifest] = {} + for lookup_key, group in by_key.items(): + by_source: Dict[str, List[PluginManifest]] = {} + for manifest in group: + by_source.setdefault(manifest.source, []).append(manifest) + dup_source = next( + (src for src, entries in by_source.items() if len(entries) > 1), + None, + ) + if dup_source is None: + # At most one manifest per source -- a single declaration, + # or a legitimate cross-source override. Preserve the + # original "last source wins" semantics. + winners[lookup_key] = group[-1] + continue + + dup_manifests = by_source[dup_source] + conflict_dirs = sorted(str(m.path) for m in dup_manifests) + conflict_desc = ( + f"plugin name {lookup_key!r} is declared by " + f"{len(dup_manifests)} {dup_source} plugin directories: " + f"{' and '.join(conflict_dirs)}" + ) + if lookup_key in required or group[0].name in required: + raise RequiredPluginError( + f"required {conflict_desc} -- ambiguous which " + "directory is authoritative; remove the duplicate " + "before startup" + ) + logger.warning( + "Refusing to load either copy of a duplicate plugin: %s " + "(ambiguous -- remove or rename one of the directories)", + conflict_desc, + ) + # Neither copy is added to `winners` -- both are skipped. + return winners + def _validate_required_plugins(self, required: Set[str]) -> None: """Fail startup when any configured mandatory enforcer is ineffective.""" for required_name in sorted(required): diff --git a/hermes_cli/subcommands/persist_smoke.py b/hermes_cli/subcommands/persist_smoke.py new file mode 100644 index 000000000000..cd0ff26b851f --- /dev/null +++ b/hermes_cli/subcommands/persist_smoke.py @@ -0,0 +1,34 @@ +"""``hermes persist-smoke`` subcommand parser. + +Follows the same extraction pattern as ``hermes_cli/subcommands/doctor.py``: +the parser lives here, the handler is injected by ``main.py`` to avoid an +import cycle, and the actual implementation lives in +``hermes_cli/persist_smoke.py``. +""" + +from __future__ import annotations + +from typing import Callable + + +def build_persist_smoke_parser(subparsers, *, cmd_persist_smoke: Callable) -> None: + """Attach the ``persist-smoke`` subcommand to ``subparsers``.""" + persist_smoke_parser = subparsers.add_parser( + "persist-smoke", + help="Smoke-test the governed persistence funnel (pre_persist_write)", + description=( + "Discovers plugins, asserts the 'pre_persist_write' required " + "hook is registered, then routes one probe write through " + "agent.persist_boundary.governed_persist and confirms it comes " + "back genuinely staged (governed) rather than silently passing " + "through to a canonical write. Exits 0 only on a confirmed " + "staged outcome; any other result -- hook not registered, " + "denied, or an ungoverned passthrough -- exits 1." + ), + ) + persist_smoke_parser.add_argument( + "--json", + action="store_true", + help="Emit machine-readable JSON instead of human-readable text", + ) + persist_smoke_parser.set_defaults(func=cmd_persist_smoke) diff --git a/tests/hermes_cli/test_persist_smoke.py b/tests/hermes_cli/test_persist_smoke.py new file mode 100644 index 000000000000..21999c2eeefa --- /dev/null +++ b/tests/hermes_cli/test_persist_smoke.py @@ -0,0 +1,470 @@ +"""Tests for the ``hermes persist-smoke`` CLI subcommand. + +Mirrors the monkeypatched-dependency style of tests/hermes_cli/ +test_required_enforcers.py and tests/agent/test_persist_boundary.py: +``hermes_cli.plugins.discover_plugins`` / ``get_plugin_manager`` and +``agent.persist_boundary.governed_persist`` are patched directly rather than +exercising the real plugin runtime, since ``run_persist_smoke`` imports each +of them locally (at call time) precisely so this works. +""" + +from __future__ import annotations + +import json +from types import SimpleNamespace + +import pytest + +from agent.persist_boundary import PersistResult +from hermes_cli import persist_smoke + + +class _FakeManager: + def __init__(self, required_hooks=None, plugins=None): + self._required_hooks = required_hooks or {} + self._plugins = plugins or {} + + +def _loaded(name, enabled=True, error=None): + return SimpleNamespace( + manifest=SimpleNamespace(name=name), enabled=enabled, error=error + ) + + +def _args(as_json=True): + return SimpleNamespace(json=as_json) + + +def _patch_discovery(monkeypatch, manager, *, spy=None): + def fake_discover(force=False): + if spy is not None: + spy["force"] = force + + monkeypatch.setattr("hermes_cli.plugins.discover_plugins", fake_discover) + monkeypatch.setattr("hermes_cli.plugins.get_plugin_manager", lambda: manager) + + +_REGISTERED = {"pre_persist_write": [("agent-lineage", lambda **kw: None)]} + + +# --------------------------------------------------------------------------- +# Hook registered -- governed_persist is called and its outcome dictates +# the exit code / JSON envelope. +# --------------------------------------------------------------------------- + + +class TestHookRegistered: + def test_discover_plugins_called_with_force_true(self, monkeypatch, capsys): + manager = _FakeManager(required_hooks=_REGISTERED) + spy: dict = {} + _patch_discovery(monkeypatch, manager, spy=spy) + monkeypatch.setattr( + "agent.persist_boundary.governed_persist", + lambda kind, path, content, meta=None: PersistResult( + staged=True, digest="sha256:" + "a" * 64, denied=False, message="" + ), + ) + + persist_smoke.run_persist_smoke(_args()) + + assert spy["force"] is True + + def test_staged_result_exits_zero_with_digest(self, monkeypatch, capsys): + manager = _FakeManager(required_hooks=_REGISTERED) + _patch_discovery(monkeypatch, manager) + digest = "sha256:" + "a" * 64 + seen = {} + + def fake_governed_persist(kind, path, content, meta=None): + seen["kind"] = kind + seen["path"] = path + seen["meta"] = meta + return PersistResult(staged=True, digest=digest, denied=False, message="") + + monkeypatch.setattr( + "agent.persist_boundary.governed_persist", fake_governed_persist + ) + + # exit(0) doesn't raise SystemExit at all -- run_persist_smoke just + # returns on the staged/ok path. + persist_smoke.run_persist_smoke(_args()) + + out = json.loads(capsys.readouterr().out) + assert out == { + "ok": True, + "hooks": ["pre_persist_write"], + "staged": True, + "digest": digest, + "mode": "staged", + } + assert seen["kind"] == "memory" + assert seen["path"] == "persist-smoke/probe.md" + # Each invocation mints its own session id (see + # TestUniqueSession below) -- only "origin" is fixed. + assert seen["meta"]["origin"] == "persist-smoke" + assert seen["meta"]["session_id"].startswith("persist-smoke:") + + def test_probe_content_includes_an_iso_timestamp(self, monkeypatch): + manager = _FakeManager(required_hooks=_REGISTERED) + _patch_discovery(monkeypatch, manager) + seen = {} + + def fake_governed_persist(kind, path, content, meta=None): + seen["content"] = content + return PersistResult( + staged=True, digest="sha256:" + "b" * 64, denied=False, message="" + ) + + monkeypatch.setattr( + "agent.persist_boundary.governed_persist", fake_governed_persist + ) + + persist_smoke.run_persist_smoke(_args()) + + assert isinstance(seen["content"], bytes) + text = seen["content"].decode("utf-8") + # datetime.isoformat() always contains a literal "T" separator. + assert "T" in text + + def test_denied_result_exits_one(self, monkeypatch, capsys): + manager = _FakeManager(required_hooks=_REGISTERED) + _patch_discovery(monkeypatch, manager) + monkeypatch.setattr( + "agent.persist_boundary.governed_persist", + lambda kind, path, content, meta=None: PersistResult( + staged=False, digest=None, denied=True, message="POLICY_DENIED: nope" + ), + ) + + with pytest.raises(SystemExit) as exc: + persist_smoke.run_persist_smoke(_args()) + + assert exc.value.code == 1 + out = json.loads(capsys.readouterr().out) + assert out == {"ok": False, "mode": "denied", "message": "POLICY_DENIED: nope"} + + def test_passthrough_result_exits_one_and_cleans_up_probe( + self, monkeypatch, capsys, tmp_path + ): + manager = _FakeManager(required_hooks=_REGISTERED) + _patch_discovery(monkeypatch, manager) + monkeypatch.chdir(tmp_path) + + def fake_governed_persist(kind, path, content, meta=None): + # Simulate governed_persist's own pre-cutover canonical write -- + # exactly the file run_persist_smoke is responsible for cleaning + # up on this outcome. + probe = tmp_path / path + probe.parent.mkdir(parents=True, exist_ok=True) + probe.write_bytes(content) + return PersistResult(staged=False, digest=None, denied=False, message="") + + monkeypatch.setattr( + "agent.persist_boundary.governed_persist", fake_governed_persist + ) + + with pytest.raises(SystemExit) as exc: + persist_smoke.run_persist_smoke(_args()) + + assert exc.value.code == 1 + out = json.loads(capsys.readouterr().out) + assert out == { + "ok": False, + "mode": "passthrough", + "reason": "no enforcer registered — governed config required for the smoke", + } + assert not (tmp_path / "persist-smoke" / "probe.md").exists() + # The now-empty persist-smoke/ directory is cleaned up too, not + # just the probe file inside it. + assert not (tmp_path / "persist-smoke").exists() + + def test_passthrough_leaves_other_files_in_probe_dir_alone( + self, monkeypatch, capsys, tmp_path + ): + # If persist-smoke/ isn't empty after the probe is removed (an + # unrelated file happens to live alongside it), rmdir must fail + # silently rather than raising -- the cleanup is best-effort. + manager = _FakeManager(required_hooks=_REGISTERED) + _patch_discovery(monkeypatch, manager) + monkeypatch.chdir(tmp_path) + sibling = tmp_path / "persist-smoke" / "keepme.txt" + + def fake_governed_persist(kind, path, content, meta=None): + probe = tmp_path / path + probe.parent.mkdir(parents=True, exist_ok=True) + probe.write_bytes(content) + sibling.write_text("do not delete") + return PersistResult(staged=False, digest=None, denied=False, message="") + + monkeypatch.setattr( + "agent.persist_boundary.governed_persist", fake_governed_persist + ) + + with pytest.raises(SystemExit) as exc: + persist_smoke.run_persist_smoke(_args()) + + assert exc.value.code == 1 + assert not (tmp_path / "persist-smoke" / "probe.md").exists() + assert sibling.exists() + + def test_human_mode_does_not_crash_and_is_not_json(self, monkeypatch, capsys): + manager = _FakeManager(required_hooks=_REGISTERED) + _patch_discovery(monkeypatch, manager) + monkeypatch.setattr( + "agent.persist_boundary.governed_persist", + lambda kind, path, content, meta=None: PersistResult( + staged=True, digest="sha256:" + "c" * 64, denied=False, message="" + ), + ) + + persist_smoke.run_persist_smoke(_args(as_json=False)) + + out = capsys.readouterr().out + with pytest.raises(json.JSONDecodeError): + json.loads(out) + assert "OK" in out + + +# --------------------------------------------------------------------------- +# governed_persist reports staged=True for TWO different reasons: a genuine +# policy-decision stage (message == "") and its own decision-less local +# fallback (a non-empty message). Only the former is green. +# --------------------------------------------------------------------------- + + +class TestLocalFallback: + def test_local_fallback_result_exits_one_not_zero(self, monkeypatch, capsys): + manager = _FakeManager(required_hooks=_REGISTERED) + _patch_discovery(monkeypatch, manager) + digest = "sha256:" + "d" * 64 + fallback_message = ( + "staged locally: no policy enforcer reachable for pre_persist_write" + ) + monkeypatch.setattr( + "agent.persist_boundary.governed_persist", + lambda kind, path, content, meta=None: PersistResult( + staged=True, digest=digest, denied=False, message=fallback_message + ), + ) + + with pytest.raises(SystemExit) as exc: + persist_smoke.run_persist_smoke(_args()) + + assert exc.value.code == 1 + out = json.loads(capsys.readouterr().out) + assert out == { + "ok": False, + "mode": "local_fallback", + "digest": digest, + "message": fallback_message, + } + + def test_local_fallback_human_mode_reports_fail_not_ok(self, monkeypatch, capsys): + manager = _FakeManager(required_hooks=_REGISTERED) + _patch_discovery(monkeypatch, manager) + monkeypatch.setattr( + "agent.persist_boundary.governed_persist", + lambda kind, path, content, meta=None: PersistResult( + staged=True, + digest="sha256:" + "d" * 64, + denied=False, + message="staged locally: no policy enforcer reachable", + ), + ) + + with pytest.raises(SystemExit) as exc: + persist_smoke.run_persist_smoke(_args(as_json=False)) + + assert exc.value.code == 1 + out = capsys.readouterr().out + assert "FAIL" in out + assert "OK" not in out + + +# --------------------------------------------------------------------------- +# Each invocation must mint its own session id rather than reuse a fixed +# default -- otherwise a periodic smoke exhausts its own shared budget over +# time and starts self-denying. +# --------------------------------------------------------------------------- + + +class TestUniqueSession: + def test_two_invocations_get_two_distinct_session_ids(self, monkeypatch): + manager = _FakeManager(required_hooks=_REGISTERED) + _patch_discovery(monkeypatch, manager) + seen_sessions = [] + + def fake_governed_persist(kind, path, content, meta=None): + seen_sessions.append((meta or {}).get("session_id")) + return PersistResult( + staged=True, digest="sha256:" + "e" * 64, denied=False, message="" + ) + + monkeypatch.setattr( + "agent.persist_boundary.governed_persist", fake_governed_persist + ) + + persist_smoke.run_persist_smoke(_args()) + persist_smoke.run_persist_smoke(_args()) + + assert len(seen_sessions) == 2 + assert all(seen_sessions) + assert seen_sessions[0] != seen_sessions[1] + assert all(s.startswith("persist-smoke:") for s in seen_sessions) + + +# --------------------------------------------------------------------------- +# Hook absent -- exits 1 before ever calling governed_persist, with a +# diagnosis distinguishing "no plugin discovered" from "plugin present but +# this hook is missing". +# --------------------------------------------------------------------------- + + +class TestHookAbsent: + def test_plugin_not_discovered_diagnosis(self, monkeypatch, capsys): + manager = _FakeManager() + _patch_discovery(monkeypatch, manager) + called = {"governed_persist": False} + monkeypatch.setattr( + "agent.persist_boundary.governed_persist", + lambda *a, **kw: called.__setitem__("governed_persist", True), + ) + + with pytest.raises(SystemExit) as exc: + persist_smoke.run_persist_smoke(_args()) + + assert exc.value.code == 1 + assert called["governed_persist"] is False + out = json.loads(capsys.readouterr().out) + assert out["ok"] is False + assert out["mode"] == "no_hook" + assert "no 'agent-lineage' plugin was discovered" in out["message"] + + def test_plugin_present_hook_missing_diagnosis(self, monkeypatch, capsys): + manager = _FakeManager(plugins={"agent-lineage": _loaded("agent-lineage")}) + _patch_discovery(monkeypatch, manager) + + with pytest.raises(SystemExit) as exc: + persist_smoke.run_persist_smoke(_args()) + + assert exc.value.code == 1 + out = json.loads(capsys.readouterr().out) + assert out["ok"] is False + assert out["mode"] == "no_hook" + assert "'agent-lineage' plugin was discovered" in out["message"] + assert "stale plugin version" in out["message"] + + def test_plugin_present_under_different_key_is_still_found_by_name( + self, monkeypatch, capsys + ): + # A plugin discovered under a path-derived key (e.g. nested under a + # category) rather than the bare name must still be recognized by + # its manifest.name -- the diagnosis distinguishes "not discovered" + # from "discovered, hook missing", not "found under this exact key". + manager = _FakeManager( + plugins={"custom/agent-lineage": _loaded("agent-lineage")} + ) + _patch_discovery(monkeypatch, manager) + + with pytest.raises(SystemExit): + persist_smoke.run_persist_smoke(_args()) + + out = json.loads(capsys.readouterr().out) + assert "stale plugin version" in out["message"] + + +# --------------------------------------------------------------------------- +# discover_plugins(force=True) (or registry introspection right after it) +# raising -- a fail-loud duplicate-required-plugin-name RequiredPluginError, +# or any other discovery-time exception -- must produce a clean JSON +# diagnostic, exit 1, and never reach governed_persist. Mirrors main.py's own +# CLI-startup handling of the same call. +# --------------------------------------------------------------------------- + + +class TestDiscoveryFailed: + def test_required_plugin_error_is_diagnosed_not_raised(self, monkeypatch, capsys): + from hermes_cli.plugins import RequiredPluginError + + def fake_discover(force=False): + raise RequiredPluginError( + "required plugin name 'agent-lineage' is ambiguous; configure " + "its path-derived plugin key" + ) + + monkeypatch.setattr("hermes_cli.plugins.discover_plugins", fake_discover) + called = {"governed_persist": False} + monkeypatch.setattr( + "agent.persist_boundary.governed_persist", + lambda *a, **kw: called.__setitem__("governed_persist", True), + ) + + with pytest.raises(SystemExit) as exc: + persist_smoke.run_persist_smoke(_args()) + + assert exc.value.code == 1 + assert called["governed_persist"] is False + out = json.loads(capsys.readouterr().out) + assert out["ok"] is False + assert out["mode"] == "discovery-failed" + assert "agent-lineage" in out["error"] + assert "ambiguous" in out["error"] + assert "hint" in out + + def test_generic_discovery_exception_is_diagnosed_not_raised( + self, monkeypatch, capsys + ): + def fake_discover(force=False): + raise RuntimeError("plugin directory unreadable") + + monkeypatch.setattr("hermes_cli.plugins.discover_plugins", fake_discover) + + with pytest.raises(SystemExit) as exc: + persist_smoke.run_persist_smoke(_args()) + + assert exc.value.code == 1 + out = json.loads(capsys.readouterr().out) + assert out == { + "ok": False, + "mode": "discovery-failed", + "error": "plugin directory unreadable", + "hint": "hermes startup itself would abort — fix plugin discovery first", + } + + def test_registry_introspection_failure_after_discovery_is_diagnosed( + self, monkeypatch, capsys + ): + # discover_plugins() itself can succeed while get_plugin_manager() + # (or reading its registry) is what actually blows up -- both are + # inside the same guarded block. + def fake_discover(force=False): + return None + + def fake_get_manager(): + raise RuntimeError("plugin manager singleton not initialized") + + monkeypatch.setattr("hermes_cli.plugins.discover_plugins", fake_discover) + monkeypatch.setattr("hermes_cli.plugins.get_plugin_manager", fake_get_manager) + + with pytest.raises(SystemExit) as exc: + persist_smoke.run_persist_smoke(_args()) + + assert exc.value.code == 1 + out = json.loads(capsys.readouterr().out) + assert out["mode"] == "discovery-failed" + assert "not initialized" in out["error"] + + def test_human_mode_discovery_failed_is_fail_not_json(self, monkeypatch, capsys): + def fake_discover(force=False): + raise RuntimeError("boom") + + monkeypatch.setattr("hermes_cli.plugins.discover_plugins", fake_discover) + + with pytest.raises(SystemExit): + persist_smoke.run_persist_smoke(_args(as_json=False)) + + out = capsys.readouterr().out + with pytest.raises(json.JSONDecodeError): + json.loads(out) + assert "FAIL" in out + assert "boom" in out diff --git a/tests/hermes_cli/test_required_enforcers.py b/tests/hermes_cli/test_required_enforcers.py index 3ba269e306da..779a658d9c93 100644 --- a/tests/hermes_cli/test_required_enforcers.py +++ b/tests/hermes_cli/test_required_enforcers.py @@ -46,6 +46,23 @@ def _plugin(home: Path, name: str, register_body: str) -> Path: return directory +def _plugin_at( + home: Path, dirname: str, declared_name: str, register_body: str = "pass" +) -> Path: + """Like ``_plugin`` but the directory name and the manifest's declared + ``name`` can differ -- needed to build a same-source duplicate-name + collision (two directories independently declaring the same name).""" + directory = home / "plugins" / dirname + directory.mkdir(parents=True) + (directory / "plugin.yaml").write_text( + yaml.safe_dump({"name": declared_name, "version": "0.1.0"}), encoding="utf-8" + ) + (directory / "__init__.py").write_text( + f"def register(ctx):\n {register_body}\n", encoding="utf-8" + ) + return directory + + def test_required_plugin_must_be_discovered(tmp_path, monkeypatch): home = tmp_path / "home" _write_config(home, enabled=["missing"], required=["missing"]) @@ -359,6 +376,61 @@ def test_required_model_and_run_boundaries_are_supported(tmp_path, monkeypatch): } +# --------------------------------------------------------------------------- +# Duplicate plugin names across directories -- discovery must fail loudly, +# not silently let directory-iteration order pick an arbitrary winner. +# --------------------------------------------------------------------------- + + +def test_duplicate_required_plugin_name_aborts_startup_naming_both_dirs( + tmp_path, monkeypatch +): + home = tmp_path / "home" + dir_a = _plugin_at(home, "dup-plug-a", "dup-plug") + dir_b = _plugin_at(home, "dup-plug-b", "dup-plug") + _write_config(home, required=["dup-plug"]) + monkeypatch.setenv("HERMES_HOME", str(home)) + + with pytest.raises(RequiredPluginError) as excinfo: + PluginManager().discover_and_load() + + message = str(excinfo.value) + assert "dup-plug" in message + assert str(dir_a) in message + assert str(dir_b) in message + + +def test_duplicate_optional_plugin_name_loads_neither_copy( + tmp_path, monkeypatch, caplog +): + home = tmp_path / "home" + dir_a = _plugin_at(home, "dup-plug-a", "dup-plug") + dir_b = _plugin_at(home, "dup-plug-b", "dup-plug") + _write_config(home, enabled=["dup-plug"]) + monkeypatch.setenv("HERMES_HOME", str(home)) + + manager = PluginManager() + with caplog.at_level("WARNING"): + manager.discover_and_load() + + assert "dup-plug" not in manager._plugins + assert str(dir_a) in caplog.text + assert str(dir_b) in caplog.text + + +def test_unique_plugin_names_are_unaffected_by_duplicate_check(tmp_path, monkeypatch): + home = tmp_path / "home" + _plugin(home, "solo", "ctx.register_hook('on_session_start', lambda **kw: None)") + _write_config(home, enabled=["solo"]) + monkeypatch.setenv("HERMES_HOME", str(home)) + + manager = PluginManager() + manager.discover_and_load() + + assert "solo" in manager._plugins + assert manager._plugins["solo"].enabled is True + + def test_conflicting_required_run_bindings_fail_closed(tmp_path, monkeypatch): home = tmp_path / "home" _write_config(home, required=["first", "second"])