feat(plugins): register_command(override=True) to shadow a built-in command

cli.py

@@ -12001,6 +12001,37 @@ def process_command(self, command: str) -> bool:
                 platform="cli",
             )
 
+        # Authorized built-in override (register_command(override=True)):
+        # a plugin command that the operator opted in to shadow the built-in
+        # wins BEFORE the built-in dispatch below. This keeps override
+        # precedence identical across CLI, gateway and TUI. Unauthorized or
+        # non-override plugin commands are NOT returned here, so built-ins keep
+        # their normal precedence and reach the post-built-in plugin fallback.
+        try:
+            from hermes_cli.plugins import (
+                get_plugin_command_override_handler,
+                resolve_plugin_command_result,
+            )
+            _override_handler = get_plugin_command_override_handler(canonical)
+            if _override_handler is None and _base_word != canonical:
+                _override_handler = get_plugin_command_override_handler(
+                    _base_word
+                )
+        except Exception:
+            _override_handler = None
+        if _override_handler is not None:
+            _ov_parts = cmd_original.split(None, 1)
+            _ov_args = _ov_parts[1].strip() if len(_ov_parts) > 1 else ""
+            try:
+                result = resolve_plugin_command_result(
+                    _override_handler(_ov_args)
+                )
+                if result:
+                    _cprint(str(result))
+            except Exception as e:
+                _cprint(f"\033[1;31mPlugin command error: {e}{_RST}")
+            return True
+
         # A bare `/resume` prompt is one-shot: any command other than the
         # resume/sessions handlers (which manage the pending state themselves)
         # disarms it so a later number isn't swallowed as a stale selection.

gateway/run.py

@@ -17876,6 +17876,40 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]:
                     canonical = _cmd_def.name if _cmd_def else command
                     break
 
+        # Authorized built-in override (register_command(override=True)):
+        # an operator-approved plugin command that shadows the built-in wins
+        # BEFORE built-in dispatch, so override precedence is identical to CLI
+        # and TUI. Non-override / unauthorized plugin commands are not returned
+        # here and keep reaching the post-built-in plugin fallback below.
+        if command:
+            try:
+                from hermes_cli.plugins import (
+                    get_plugin_command_override_handler,
+                )
+                # Telegram autocomplete sends underscores; plugin commands
+                # register with hyphens (mirrors the fallback normalization).
+                _override_handler = get_plugin_command_override_handler(
+                    canonical.replace("_", "-") if canonical else canonical
+                )
+                if _override_handler is None and command != canonical:
+                    _override_handler = get_plugin_command_override_handler(
+                        command.replace("_", "-")
+                    )
+            except Exception:
+                _override_handler = None
+            if _override_handler is not None:
+                try:
+                    user_args = event.get_command_args().strip()
+                    result = _override_handler(user_args)
+                    if asyncio.iscoroutine(result):
+                        result = await result
+                    return str(result) if result else None
+                except Exception as e:
+                    logger.warning(
+                        "Plugin command override dispatch failed: %s", e
+                    )
+                    return f"Plugin command error: {e}"
+
         if canonical == "pause":
             return await self._handle_pause_command(event)
 

hermes_cli/plugin_capabilities.py

@@ -128,6 +128,14 @@ class CapabilitySpec:
                 "(add reactions, rename threads) via ctx.platform_actions"
             ),
         ),
+        CapabilitySpec(
+            id="commands.override",
+            legacy_path=("allow_command_override",),
+            description=(
+                "Shadow a built-in slash command (e.g. /new, /model), an "
+                "override intercepts what users invoke through that command"
+            ),
+        ),
     )
 }
 

hermes_cli/plugins.py

@@ -102,6 +102,14 @@ class PluginToolOverrideError(PermissionError):
     """
 
 
+class PluginCommandOverrideError(PermissionError):
+    """Raised when a plugin attempts to override a built-in slash command
+    without operator opt-in via
+    ``plugins.entries.<plugin_id>.allow_command_override`` (or the
+    ``commands.override`` granted capability).
+    """
+
+
 logger = logging.getLogger(__name__)
 
 
@@ -1983,6 +1991,35 @@ def _tool_override_allowed(self, tool_name: str) -> bool:
         # manager's home, never the active profile's (#65593 constraint).
         return plugin_capability_granted(plugin_id, "tools.override", config=cfg)
 
+    def _command_override_allowed(self, command_name: str) -> bool:
+        """Return True if this plugin may shadow a built-in command.
+
+        Mirrors :meth:`_tool_override_allowed` exactly so command overrides
+        get the SAME fail-closed operator opt-in as tool overrides. Bundled
+        plugins are trusted; every other source must be granted the
+        ``commands.override`` capability, satisfied by EITHER the
+        consent-flow grant
+        (``plugins.entries.<plugin_id>.granted_capabilities``) OR the
+        deprecated legacy key ``allow_command_override: true``. Any failure
+        to read consent state returns False (fail closed).
+        """
+        source = getattr(self.manifest, "source", "") or ""
+        if source == "bundled":
+            return True
+        try:
+            from hermes_cli.config import load_config
+
+            with _plugin_home_scope(self._manager.home_path):
+                cfg = load_config() or {}
+        except Exception:
+            # If we can't load config, fail closed, better to break the
+            # override than silently grant it.
+            return False
+        plugin_id = self.manifest.key or self.manifest.name
+        return plugin_capability_granted(
+            plugin_id, "commands.override", config=cfg
+        )
+
     # -- message injection --------------------------------------------------
 
     def inject_message(
@@ -2124,6 +2161,7 @@ def register_command(
         handler: Callable,
         description: str = "",
         args_hint: str = "",
+        override: bool = False,
     ) -> Optional[PluginRegistration]:
         """Register a slash command (e.g. ``/lcm``) available in CLI and gateway sessions.
 
@@ -2141,7 +2179,18 @@ def register_command(
         parameterless in Discord and still accept trailing text when invoked
         as free-form chat.
 
-        Names conflicting with built-in commands are rejected with a warning.
+        Names conflicting with built-in commands are rejected with a warning,
+        unless ``override=True`` is passed, in which case the plugin command
+        shadows the built-in on every surface (CLI, gateway, TUI). Use this to
+        enhance or wrap a built-in command from a plugin (e.g. add lineage
+        metadata to ``/new``).
+
+        ``override=True`` against a built-in command requires the operator to
+        opt in via ``plugins.entries.<plugin_id>.allow_command_override: true``
+        in config.yaml (or the ``commands.override`` granted capability), and
+        mirrors the fail-closed trust gate that ``register_tool(override=True)``
+        already enforces. Without that gate any enabled plugin could silently
+        replace a privileged built-in like ``/model`` or ``/config``.
         """
         clean = name.lower().strip().lstrip("/").replace(" ", "-")
         if not clean:
@@ -2151,16 +2200,37 @@ def register_command(
             )
             return
 
-        # Reject if it conflicts with a built-in command
+        # Resolve built-in collision. Without override=True the plugin command
+        # is rejected (unchanged). With override=True it may shadow the
+        # built-in, but ONLY when the operator has opted this plugin in,
+        # fail-closed (mirrors the register_tool override gate).
+        overrides_builtin = False
         try:
             from hermes_cli.commands import resolve_command
             if resolve_command(clean) is not None:
+                if not override:
+                    logger.warning(
+                        "Plugin '%s' tried to register command '/%s' which "
+                        "conflicts with a built-in command. Skipping.",
+                        self.manifest.name, clean,
+                    )
+                    return
+                if not self._command_override_allowed(clean):
+                    plugin_id = self.manifest.key or self.manifest.name
+                    raise PluginCommandOverrideError(
+                        f"Plugin {self.manifest.name!r} cannot override "
+                        f"built-in command '/{clean}'. Set "
+                        f"plugins.entries.{plugin_id}.allow_command_override: "
+                        f"true in config.yaml to allow this plugin to shadow "
+                        f"built-in commands."
+                    )
+                overrides_builtin = True
                 logger.warning(
-                    "Plugin '%s' tried to register command '/%s' which conflicts "
-                    "with a built-in command. Skipping.",
+                    "Plugin '%s' is OVERRIDING built-in command '/%s'.",
                     self.manifest.name, clean,
                 )
-                return
+        except PluginCommandOverrideError:
+            raise
         except Exception:
             pass  # If commands module isn't available, skip the check
 
@@ -2171,6 +2241,9 @@ def register_command(
             "plugin": self.manifest.name,
             "plugin_key": self.manifest.key or self.manifest.name,
             "args_hint": (args_hint or "").strip(),
+            # Authorized shadow of a built-in, every surface consults this
+            # flag so precedence is identical on CLI, gateway and TUI.
+            "override_builtin": overrides_builtin,
         }
         self._manager._plugin_commands[clean] = entry
         handle = self._track_replacement(
@@ -2183,7 +2256,11 @@ def register_command(
                 self._manager._plugin_commands, clean, entry, replacement
             ),
         )
-        logger.debug("Plugin %s registered command: /%s", self.manifest.name, clean)
+        logger.debug(
+            "Plugin %s registered command: /%s%s",
+            self.manifest.name, clean,
+            " (override)" if overrides_builtin else "",
+        )
         return handle
 
     # -- tool dispatch -------------------------------------------------------
@@ -6648,6 +6725,23 @@ def get_plugin_command_handler(name: str) -> Optional[Callable]:
     return entry["handler"] if entry else None
 
 
+def get_plugin_command_override_handler(name: str) -> Optional[Callable]:
+    """Return the handler for a plugin command that is an AUTHORIZED override
+    of a built-in, or ``None``.
+
+    Every dispatch surface (CLI, gateway, TUI) calls this BEFORE resolving the
+    built-in so an operator-approved ``register_command(override=True)`` wins
+    consistently. A plugin command that merely shares a name without the
+    ``override_builtin`` flag (i.e. it was registered before the built-in
+    existed, or without authorization) is NOT returned here, so built-ins keep
+    their normal precedence in every other case.
+    """
+    entry = _ensure_plugins_discovered()._plugin_commands.get(name)
+    if entry and entry.get("override_builtin"):
+        return entry["handler"]
+    return None
+
+
 _PLUGIN_COMMAND_AWAIT_TIMEOUT_SECS = 30.0
 
 

tests/cli/test_plugin_command_override.py

@@ -0,0 +1,64 @@
+"""Behavioral coverage for plugin overrides through CLI dispatch."""
+
+from unittest.mock import MagicMock
+
+import pytest
+import yaml
+
+from hermes_cli.plugins import PluginCommandOverrideError, PluginContext, PluginManager, PluginManifest
+
+
+def _register_help_override(tmp_path, monkeypatch, *, granted: bool):
+    from hermes_cli import plugins as plugins_mod
+
+    home = tmp_path / "home"
+    home.mkdir()
+    entry = {"granted_capabilities": ["commands.override"]} if granted else {}
+    (home / "config.yaml").write_text(
+        yaml.safe_dump({"plugins": {"entries": {"help-plugin": entry}}}), encoding="utf-8"
+    )
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    manager = PluginManager(scope_key=str(home))
+    context = PluginContext(
+        PluginManifest(name="help-plugin", key="help-plugin", source="user"), manager
+    )
+    handler = MagicMock(return_value="plugin help result")
+    if granted:
+        context.register_command("help", handler, override=True)
+    else:
+        with pytest.raises(PluginCommandOverrideError):
+            context.register_command("help", handler, override=True)
+    monkeypatch.setattr(plugins_mod, "_ensure_plugins_discovered", lambda: manager)
+    return handler
+
+
+def _make_cli():
+    import cli as cli_mod
+
+    instance = object.__new__(cli_mod.HermesCLI)
+    instance.session_id = "plugin-override-cli"
+    instance._pending_resume_sessions = None
+    return instance
+
+
+def test_authorized_plugin_help_override_wins_cli_dispatch(tmp_path, monkeypatch, capsys):
+    handler = _register_help_override(tmp_path, monkeypatch, granted=True)
+    cli = _make_cli()
+    cli.show_help = MagicMock(side_effect=AssertionError("built-in /help ran"))
+
+    assert cli.process_command("/help raw arguments") is True
+
+    handler.assert_called_once_with("raw arguments")
+    cli.show_help.assert_not_called()
+    assert "plugin help result" in capsys.readouterr().out
+
+
+def test_ungranted_plugin_cannot_replace_cli_help(tmp_path, monkeypatch):
+    handler = _register_help_override(tmp_path, monkeypatch, granted=False)
+    cli = _make_cli()
+    cli.show_help = MagicMock()
+
+    assert cli.process_command("/help") is True
+
+    handler.assert_not_called()
+    cli.show_help.assert_called_once()

tests/gateway/test_plugin_command_override.py

@@ -0,0 +1,59 @@
+"""Behavioral coverage for plugin overrides through gateway dispatch."""
+
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+import yaml
+
+from hermes_cli.plugins import PluginCommandOverrideError, PluginContext, PluginManager, PluginManifest
+from tests.gateway.test_gateway_command_dispatch_minimal import _make_event, _make_runner
+
+
+def _register_help_override(tmp_path, monkeypatch, *, granted: bool):
+    from hermes_cli import plugins as plugins_mod
+
+    home = tmp_path / "home"
+    home.mkdir()
+    entry = {"granted_capabilities": ["commands.override"]} if granted else {}
+    (home / "config.yaml").write_text(
+        yaml.safe_dump({"plugins": {"entries": {"help-plugin": entry}}}), encoding="utf-8"
+    )
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    manager = PluginManager(scope_key=str(home))
+    context = PluginContext(
+        PluginManifest(name="help-plugin", key="help-plugin", source="user"), manager
+    )
+    handler = MagicMock(return_value="gateway plugin help")
+    if granted:
+        context.register_command("help", handler, override=True)
+    else:
+        with pytest.raises(PluginCommandOverrideError):
+            context.register_command("help", handler, override=True)
+    monkeypatch.setattr(plugins_mod, "_ensure_plugins_discovered", lambda: manager)
+    return handler
+
+
+@pytest.mark.asyncio
+async def test_authorized_plugin_help_override_wins_gateway_dispatch(tmp_path, monkeypatch):
+    handler = _register_help_override(tmp_path, monkeypatch, granted=True)
+    runner, _adapter = _make_runner()
+    runner._handle_help_command = AsyncMock(side_effect=AssertionError("built-in /help ran"))
+
+    result = await runner._handle_message(_make_event("/help raw arguments"))
+
+    assert result == "gateway plugin help"
+    handler.assert_called_once_with("raw arguments")
+    runner._handle_help_command.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_ungranted_plugin_cannot_replace_gateway_help(tmp_path, monkeypatch):
+    handler = _register_help_override(tmp_path, monkeypatch, granted=False)
+    runner, _adapter = _make_runner()
+    runner._handle_help_command = AsyncMock(return_value="built-in gateway help")
+
+    result = await runner._handle_message(_make_event("/help"))
+
+    assert result == "built-in gateway help"
+    handler.assert_not_called()
+    runner._handle_help_command.assert_called_once()