feat(cli): persist-smoke verb + fail-loud duplicate plugin names

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 <subcmd>`."""
     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
     # =========================================================================

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)

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):