feat(plugins): add approval transport interface

cli-config.yaml.example

@@ -396,6 +396,9 @@ terminal:
 #   tirith_path: "tirith"       # Path to tirith binary (supports ~ expansion)
 #   tirith_timeout: 5           # Scan timeout in seconds
 #   tirith_fail_open: true      # Allow commands if tirith unavailable
+#   approval:
+#     transport: builtin        # Or an explicitly enabled plugin transport name
+#     transport_fallback: deny  # Set builtin to opt into fallback on transport failure
 
 # =============================================================================
 # Browser Tool Configuration

hermes_cli/approval_transport.py

@@ -0,0 +1,216 @@
+"""Host-owned contract for plugin-provided human approval transports.
+
+Transports only present an immutable, redacted request and return a correlated
+human decision. They do not participate in command detection or authorization
+policy. The host validates scope, request binding, and timeout fail-closed.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import inspect
+import json
+import logging
+import queue
+import threading
+import time
+import uuid
+from dataclasses import dataclass
+from typing import Awaitable, Callable, Literal
+
+logger = logging.getLogger(__name__)
+
+_MAX_ACTIVE_TRANSPORT_WORKERS = 8
+_transport_worker_slots = threading.BoundedSemaphore(_MAX_ACTIVE_TRANSPORT_WORKERS)
+
+ApprovalChoice = Literal["once", "session", "always", "deny"]
+ApprovalPresentFn = Callable[
+    ["ApprovalRequest"], "ApprovalDecision | Awaitable[ApprovalDecision]"
+]
+
+
+@dataclass(frozen=True)
+class ApprovalDecision:
+    """A transport response bound to one exact host-created request."""
+
+    request_id: str
+    request_digest: str
+    choice: str
+
+
+@dataclass(frozen=True)
+class ApprovalRequest:
+    """Immutable, display-only approval request passed to a transport plugin."""
+
+    schema_version: int
+    request_id: str
+    digest: str
+    command: str
+    description: str
+    pattern_key: str
+    pattern_keys: tuple[str, ...]
+    surface: str
+    timeout_seconds: float
+    allowed_choices: tuple[ApprovalChoice, ...]
+
+    @classmethod
+    def create(
+        cls,
+        *,
+        command: str,
+        description: str,
+        pattern_key: str,
+        pattern_keys: tuple[str, ...],
+        session_key: str,
+        surface: str,
+        allow_session: bool,
+        allow_permanent: bool,
+        timeout_seconds: float = 300,
+    ) -> "ApprovalRequest":
+        request_id = uuid.uuid4().hex
+        choices: list[ApprovalChoice] = ["once"]
+        if allow_session:
+            choices.append("session")
+        if allow_permanent:
+            choices.append("always")
+        choices.append("deny")
+        canonical = {
+            "schema_version": 1,
+            "request_id": request_id,
+            "command": command,
+            "description": description,
+            "pattern_key": pattern_key,
+            "pattern_keys": list(pattern_keys),
+            "session_key": session_key,
+            "surface": surface,
+            "timeout_seconds": timeout_seconds,
+            "allowed_choices": choices,
+        }
+        digest = hashlib.sha256(
+            json.dumps(canonical, sort_keys=True, separators=(",", ":")).encode("utf-8")
+        ).hexdigest()
+        return cls(
+            schema_version=1,
+            request_id=request_id,
+            digest=digest,
+            command=command,
+            description=description,
+            pattern_key=pattern_key,
+            pattern_keys=pattern_keys,
+            surface=surface,
+            timeout_seconds=timeout_seconds,
+            allowed_choices=tuple(choices),
+        )
+
+    def respond(self, choice: ApprovalChoice | str) -> ApprovalDecision:
+        """Build the correlated response a transport should return."""
+        return ApprovalDecision(
+            request_id=self.request_id,
+            request_digest=self.digest,
+            choice=choice,
+        )
+
+
+@dataclass(frozen=True)
+class ApprovalTransportResult:
+    """Normalized host result. Any failure is represented as a denial."""
+
+    choice: ApprovalChoice
+    failure: str | None = None
+
+
+@dataclass(frozen=True)
+class RegisteredApprovalTransport:
+    """Plugin-owned registration retained by one profile's PluginManager."""
+
+    name: str
+    present: ApprovalPresentFn
+    plugin_id: str
+    profile_home: str
+
+
+def invoke_approval_transport(
+    present: ApprovalPresentFn,
+    request: ApprovalRequest,
+    *,
+    timeout_seconds: float,
+    poll_interval: float = 1.0,
+    on_poll: Callable[[], None] | None = None,
+    is_interrupted: Callable[[], bool] | None = None,
+) -> ApprovalTransportResult:
+    """Run a sync or async transport on a bounded daemon worker.
+
+    Async callbacks are awaited with ``asyncio.run`` on that worker, never on a
+    gateway or TUI event loop. A callback must return before the host timeout;
+    late results are discarded and cannot authorize another request.
+    """
+
+    if not _transport_worker_slots.acquire(blocking=False):
+        logger.warning("Approval transport worker capacity exhausted")
+        return ApprovalTransportResult("deny", "busy")
+
+    results: queue.Queue[tuple[str, object]] = queue.Queue(maxsize=1)
+
+    async def _await_value(value):
+        return await value
+
+    def _run() -> None:
+        try:
+            value = present(request)
+            if inspect.isawaitable(value):
+                value = asyncio.run(_await_value(value))
+            results.put_nowait(("result", value))
+        except BaseException as exc:  # fail closed even for unusual callback exits
+            try:
+                results.put_nowait(("error", exc))
+            except queue.Full:
+                pass
+        finally:
+            _transport_worker_slots.release()
+
+    worker = threading.Thread(
+        target=_run,
+        name=f"approval-transport-{request.request_id[:8]}",
+        daemon=True,
+    )
+    try:
+        worker.start()
+    except BaseException:
+        _transport_worker_slots.release()
+        logger.warning("Could not start approval transport worker")
+        return ApprovalTransportResult("deny", "error")
+    deadline = time.monotonic() + max(float(timeout_seconds), 0.0)
+    while True:
+        if is_interrupted is not None and is_interrupted():
+            logger.info("Approval transport wait interrupted for %s", request.request_id)
+            return ApprovalTransportResult("deny", "interrupted")
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            logger.warning("Approval transport timed out for request %s", request.request_id)
+            return ApprovalTransportResult("deny", "timeout")
+        try:
+            kind, value = results.get(
+                timeout=min(max(float(poll_interval), 0.001), remaining)
+            )
+            break
+        except queue.Empty:
+            if on_poll is not None:
+                try:
+                    on_poll()
+                except Exception:
+                    logger.debug("Approval transport poll callback failed", exc_info=True)
+
+    if kind == "error":
+        logger.warning("Approval transport failed for request %s", request.request_id)
+        return ApprovalTransportResult("deny", "error")
+    if not isinstance(value, ApprovalDecision):
+        logger.warning("Approval transport returned an invalid decision type")
+        return ApprovalTransportResult("deny", "invalid")
+    if value.request_id != request.request_id or value.request_digest != request.digest:
+        logger.warning("Approval transport returned a stale or mismatched decision")
+        return ApprovalTransportResult("deny", "stale")
+    if value.choice not in request.allowed_choices:
+        logger.warning("Approval transport returned a disallowed choice")
+        return ApprovalTransportResult("deny", "invalid")
+    return ApprovalTransportResult(value.choice)

hermes_cli/config_defaults.py

@@ -2205,6 +2205,16 @@ DEFAULT_CONFIG = {
     "security": {
         "allow_private_urls": False,  # Allow requests to private/internal IPs (for OpenWrt, proxies, VPNs)
         "redact_secrets": True,
+        # Human approval presentation transport. "builtin" preserves the
+        # current CLI/TUI/gateway/ACP surfaces. A plugin transport is used only
+        # when named explicitly here. Transport timeout/error/invalid response
+        # denies unless transport_fallback is explicitly set to "builtin".
+        # This is presentation only: plugins cannot detect, suppress, or
+        # auto-approve commands outside a correlated human response.
+        "approval": {
+            "transport": "builtin",
+            "transport_fallback": "deny",
+        },
         # Writes to agent-instruction files (AGENTS.md/CLAUDE.md/SOUL.md/
         # .cursorrules, project-local .hermes config) always require human
         # approval — even under auto-approve/yolo. Extra patterns are

hermes_cli/plugins.py

@@ -718,6 +718,23 @@ class PluginContext:
         except Exception:
             return "default"
 
+    # -- approval transport registration ------------------------------------
+
+    def register_approval_transport(self, name: str, present_fn: Callable) -> None:
+        """Register a human approval presentation transport.
+
+        The transport is inactive until the operator explicitly selects
+        ``security.approval.transport: <name>``. It receives a host-created,
+        redacted ``ApprovalRequest`` and may only return a correlated human
+        decision; command policy and approval persistence remain host-owned.
+        ``present_fn`` may be synchronous or async.
+        """
+        self._manager.register_approval_transport(
+            name,
+            present_fn,
+            plugin_id=self.manifest.key or self.manifest.name,
+        )
+
     # -- tool registration --------------------------------------------------
 
     def register_tool(
@@ -1613,6 +1630,8 @@ class PluginManager:
         # Plugin-registered auxiliary tasks: key → {key, display_name,
         # description, defaults, plugin}. See PluginContext.register_auxiliary_task.
         self._aux_tasks: Dict[str, Dict[str, Any]] = {}
+        # Explicitly-selected, profile-scoped human approval transports.
+        self._approval_transports: Dict[str, Any] = {}
         # Slack Block Kit action handlers registered by plugins. Each entry
         # is (matcher, callback, plugin_name); the Slack adapter wires them
         # into its slack_bolt App at connect() time. ``matcher`` is whatever
@@ -1649,6 +1668,7 @@ class PluginManager:
             self._plugin_skills.clear()
             self._portable_mcp_servers.clear()
             self._aux_tasks.clear()
+            self._approval_transports.clear()
             self._slack_action_handlers.clear()
             self._context_engine = None
         # Set the flag up front as a re-entrancy guard (a plugin's register()
@@ -1839,6 +1859,49 @@ class PluginManager:
                 sum(1 for p in self._plugins.values() if p.enabled),
             )
 
+    def register_approval_transport(
+        self,
+        name: str,
+        present_fn: Callable,
+        *,
+        plugin_id: str,
+    ) -> None:
+        """Register one plugin-owned approval transport for this profile."""
+        import re
+
+        from hermes_cli.approval_transport import RegisteredApprovalTransport
+
+        clean = str(name).strip().lower()
+        if clean == "builtin":
+            raise ValueError("approval transport name 'builtin' is reserved")
+        if not re.fullmatch(r"[a-z0-9][a-z0-9_-]{0,63}", clean):
+            raise ValueError(
+                "approval transport name must match [a-z0-9][a-z0-9_-]{0,63}"
+            )
+        if not callable(present_fn):
+            raise TypeError("approval transport present_fn must be callable")
+        if clean in self._approval_transports:
+            owner = self._approval_transports[clean].plugin_id
+            raise ValueError(
+                f"approval transport {clean!r} is already registered by {owner!r}"
+            )
+        self._approval_transports[clean] = RegisteredApprovalTransport(
+            name=clean,
+            present=present_fn,
+            plugin_id=plugin_id,
+            profile_home=str(get_hermes_home().resolve()),
+        )
+        logger.info("Plugin %s registered approval transport: %s", plugin_id, clean)
+
+    def get_approval_transport(self, name: str):
+        """Return a transport only inside the profile that registered it."""
+        registered = self._approval_transports.get(str(name).strip().lower())
+        if registered is None:
+            return None
+        if registered.profile_home != str(get_hermes_home().resolve()):
+            return None
+        return registered
+
     def _collect_directory_manifests(self) -> List[PluginManifest]:
         """Collect directory manifests in the same order as full discovery.
 

tests/hermes_cli/test_approval_transport.py

@@ -0,0 +1,509 @@
+"""Approval transport plugin contract and fail-closed host routing."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import threading
+import time
+
+import pytest
+import yaml
+
+from hermes_cli.plugins import PluginContext, PluginManager, PluginManifest
+
+
+def _manifest(name: str = "fixture-approval") -> PluginManifest:
+    return PluginManifest(
+        name=name,
+        version="1.0.0",
+        description="fixture",
+        source="user",
+        key=name,
+    )
+
+
+def _context(manager: PluginManager, name: str = "fixture-approval") -> PluginContext:
+    return PluginContext(_manifest(name), manager)
+
+
+def _request():
+    from hermes_cli.approval_transport import ApprovalRequest
+
+    return ApprovalRequest.create(
+        command="rm -rf /tmp/example",
+        description="recursive delete",
+        pattern_key="rm_recursive",
+        pattern_keys=("rm_recursive",),
+        session_key="session-a",
+        surface="cli",
+        allow_session=True,
+        allow_permanent=True,
+    )
+
+
+def test_context_registers_owned_approval_transport():
+    manager = PluginManager()
+    callback = lambda request: request.respond("deny")
+
+    _context(manager).register_approval_transport("phone", callback)
+
+    registered = manager.get_approval_transport("phone")
+    assert registered is not None
+    assert registered.name == "phone"
+    assert registered.plugin_id == "fixture-approval"
+    assert registered.present is callback
+
+
+def test_transport_names_are_unique_and_builtin_is_reserved():
+    manager = PluginManager()
+    _context(manager, "first").register_approval_transport("phone", lambda request: None)
+
+    with pytest.raises(ValueError, match="already registered"):
+        _context(manager, "second").register_approval_transport("phone", lambda request: None)
+    with pytest.raises(ValueError, match="reserved"):
+        _context(manager).register_approval_transport("builtin", lambda request: None)
+
+
+def test_force_reload_clears_transport_registry(monkeypatch):
+    manager = PluginManager()
+    _context(manager).register_approval_transport("phone", lambda request: None)
+    manager._discovered = True
+    monkeypatch.setattr(manager, "_discover_and_load_inner", lambda: None)
+
+    manager.discover_and_load(force=True)
+
+    assert manager.get_approval_transport("phone") is None
+
+
+def test_transport_registry_is_manager_and_profile_isolated(monkeypatch, tmp_path):
+    first = PluginManager()
+    second = PluginManager()
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path / "profiles" / "work"))
+    _context(first).register_approval_transport("phone", lambda request: None)
+
+    assert first.get_approval_transport("phone") is not None
+    assert second.get_approval_transport("phone") is None
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path / "profiles" / "personal"))
+    assert first.get_approval_transport("phone") is None
+
+
+def test_host_accepts_bound_sync_and_async_decisions():
+    from hermes_cli.approval_transport import invoke_approval_transport
+
+    request = _request()
+    sync_result = invoke_approval_transport(
+        lambda received: received.respond("session"), request, timeout_seconds=1
+    )
+
+    async def present(received):
+        await asyncio.sleep(0)
+        return received.respond("once")
+
+    async_result = invoke_approval_transport(present, request, timeout_seconds=1)
+
+    assert sync_result.choice == "session"
+    assert sync_result.failure is None
+    assert async_result.choice == "once"
+    assert async_result.failure is None
+
+
+def test_host_rejects_scope_not_offered_by_request():
+    from hermes_cli.approval_transport import ApprovalRequest, invoke_approval_transport
+
+    request = ApprovalRequest.create(
+        command="dangerous",
+        description="dangerous",
+        pattern_key="danger",
+        pattern_keys=("danger",),
+        session_key="session-a",
+        surface="cli",
+        allow_session=False,
+        allow_permanent=False,
+    )
+
+    result = invoke_approval_transport(
+        lambda received: received.respond("always"), request, timeout_seconds=1
+    )
+
+    assert result.choice == "deny"
+    assert result.failure == "invalid"
+
+
+@pytest.mark.parametrize(
+    ("present", "failure"),
+    [
+        (lambda request: {"choice": "once"}, "invalid"),
+        (lambda request: request.respond("bogus"), "invalid"),
+        (
+            lambda request: type(request.respond("once"))(
+                request_id="stale",
+                request_digest=request.digest,
+                choice="once",
+            ),
+            "stale",
+        ),
+        (
+            lambda request: type(request.respond("once"))(
+                request_id=request.request_id,
+                request_digest="changed",
+                choice="once",
+            ),
+            "stale",
+        ),
+    ],
+)
+def test_host_rejects_invalid_or_stale_decisions(present, failure):
+    from hermes_cli.approval_transport import invoke_approval_transport
+
+    result = invoke_approval_transport(present, _request(), timeout_seconds=1)
+
+    assert result.choice == "deny"
+    assert result.failure == failure
+
+
+def test_host_timeout_and_exception_deny_without_waiting_forever():
+    from hermes_cli.approval_transport import invoke_approval_transport
+
+    def hangs(_request):
+        time.sleep(1)
+
+    def crashes(_request):
+        raise RuntimeError("transport offline")
+
+    started = time.monotonic()
+    timeout = invoke_approval_transport(hangs, _request(), timeout_seconds=0.02)
+    exception = invoke_approval_transport(crashes, _request(), timeout_seconds=1)
+
+    assert time.monotonic() - started < 0.5
+    assert (timeout.choice, timeout.failure) == ("deny", "timeout")
+    assert (exception.choice, exception.failure) == ("deny", "error")
+
+
+def test_host_transport_wait_is_interruptible_and_pollable():
+    from hermes_cli.approval_transport import invoke_approval_transport
+
+    polls = []
+
+    def hangs(_request):
+        time.sleep(1)
+
+    result = invoke_approval_transport(
+        hangs,
+        _request(),
+        timeout_seconds=1,
+        poll_interval=0.01,
+        on_poll=lambda: polls.append(1),
+        is_interrupted=lambda: len(polls) >= 1,
+    )
+
+    assert result.choice == "deny"
+    assert result.failure == "interrupted"
+    assert polls
+
+
+def test_host_caps_hung_transport_workers():
+    from hermes_cli.approval_transport import (
+        _MAX_ACTIVE_TRANSPORT_WORKERS,
+        invoke_approval_transport,
+    )
+
+    release = threading.Event()
+
+    def hangs(_request):
+        release.wait()
+
+    try:
+        results = [
+            invoke_approval_transport(hangs, _request(), timeout_seconds=0.001)
+            for _ in range(_MAX_ACTIVE_TRANSPORT_WORKERS + 1)
+        ]
+    finally:
+        release.set()
+
+    assert results[-1].choice == "deny"
+    assert results[-1].failure == "busy"
+
+
+def _configure_manual_guard(monkeypatch, approval_module, manager, *, fallback=None):
+    monkeypatch.setattr(approval_module, "_get_approval_mode", lambda: "manual")
+    monkeypatch.setattr(approval_module, "_is_interactive_cli", lambda: True)
+    monkeypatch.setattr(approval_module, "_is_gateway_approval_context", lambda: False)
+    monkeypatch.setattr(approval_module, "detect_hardline_command", lambda command: (False, ""))
+    monkeypatch.setattr(approval_module, "_check_sudo_stdin_guard", lambda command: (False, ""))
+    monkeypatch.setattr(approval_module, "_match_user_deny_rule", lambda command: None)
+    monkeypatch.setattr(approval_module, "_command_matches_permanent_allowlist", lambda command: False)
+    monkeypatch.setattr(approval_module, "detect_dangerous_command", lambda command: (True, "danger", "dangerous"))
+    monkeypatch.setattr(
+        approval_module,
+        "get_current_session_key",
+        lambda *args, **kwargs: "session-a",
+    )
+    monkeypatch.setattr(approval_module, "is_approved", lambda *args: False)
+    monkeypatch.setattr(approval_module, "get_plugin_manager", lambda: manager, raising=False)
+    monkeypatch.setattr(
+        approval_module,
+        "_get_approval_transport_config",
+        lambda: ("phone", fallback),
+        raising=False,
+    )
+    monkeypatch.setattr("tools.tirith_security.check_command_security", lambda command: {"action": "allow"})
+
+
+def test_cli_selected_transport_replaces_builtin_prompt(monkeypatch):
+    from tools import approval
+
+    manager = PluginManager()
+    seen = []
+    _context(manager).register_approval_transport(
+        "phone", lambda request: seen.append(request) or request.respond("once")
+    )
+    _configure_manual_guard(monkeypatch, approval, manager)
+
+    def builtin(*args, **kwargs):
+        raise AssertionError("builtin prompt must not materialize")
+
+    result = approval.check_all_command_guards(
+        "rm -rf /tmp/example", "local", approval_callback=builtin
+    )
+
+    assert result["approved"] is True
+    assert result["user_approved"] is True
+    assert len(seen) == 1
+    assert seen[0].command == "rm -rf /tmp/example"
+    assert seen[0].allowed_choices == ("once", "session", "always", "deny")
+
+
+def test_gateway_selected_transport_does_not_require_gateway_notifier(monkeypatch):
+    from tools import approval
+
+    manager = PluginManager()
+    seen = []
+    _context(manager).register_approval_transport(
+        "phone", lambda request: seen.append(request) or request.respond("once")
+    )
+    _configure_manual_guard(monkeypatch, approval, manager)
+    monkeypatch.setattr(approval, "_is_interactive_cli", lambda: False)
+    monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: True)
+    monkeypatch.setattr(approval, "_gateway_notify_cbs", {})
+
+    result = approval.check_all_command_guards("rm -rf /tmp/example", "local")
+
+    assert result["approved"] is True
+    assert len(seen) == 1
+    assert seen[0].surface == "gateway"
+
+
+def test_execute_code_gateway_uses_selected_transport(monkeypatch):
+    from tools import approval
+
+    manager = PluginManager()
+    seen = []
+    _context(manager).register_approval_transport(
+        "phone", lambda request: seen.append(request) or request.respond("once")
+    )
+    monkeypatch.setattr(approval, "_get_approval_mode", lambda: "manual")
+    monkeypatch.setattr(approval, "_is_gateway_approval_context", lambda: True)
+    monkeypatch.setattr(approval, "_is_cron_approval_context", lambda: False)
+    monkeypatch.setattr(
+        approval, "get_current_session_key", lambda *args, **kwargs: "session-a"
+    )
+    monkeypatch.setattr(approval, "is_approved", lambda *args: False)
+    monkeypatch.setattr(approval, "get_plugin_manager", lambda: manager)
+    monkeypatch.setattr(
+        approval, "_get_approval_transport_config", lambda: ("phone", None)
+    )
+    monkeypatch.setattr(approval, "_gateway_notify_cbs", {})
+
+    result = approval.check_execute_code_guard("print('ok')", "local")
+
+    assert result["approved"] is True
+    assert len(seen) == 1
+    assert seen[0].pattern_key == "execute_code"
+    assert seen[0].surface == "gateway"
+
+
+def test_transport_failure_denies_without_builtin_fallback(monkeypatch):
+    from tools import approval
+
+    manager = PluginManager()
+    _context(manager).register_approval_transport("phone", lambda request: {"choice": "once"})
+    _configure_manual_guard(monkeypatch, approval, manager)
+    builtin_calls = []
+
+    result = approval.check_all_command_guards(
+        "rm -rf /tmp/example",
+        "local",
+        approval_callback=lambda *args, **kwargs: builtin_calls.append(1) or "once",
+    )
+
+    assert result["approved"] is False
+    assert result["outcome"] == "transport_invalid"
+    assert builtin_calls == []
+
+
+def test_redaction_failure_denies_before_transport_callback(monkeypatch):
+    import agent.redact
+    from tools import approval
+
+    manager = PluginManager()
+    calls = []
+    _context(manager).register_approval_transport(
+        "phone", lambda request: calls.append(request) or request.respond("once")
+    )
+    _configure_manual_guard(monkeypatch, approval, manager)
+
+    def redaction_failed(text):
+        if text in {"rm -rf /tmp/example", "dangerous"}:
+            raise RuntimeError("redactor unavailable")
+        return text
+
+    monkeypatch.setattr(agent.redact, "redact_sensitive_text", redaction_failed)
+
+    result = approval.check_all_command_guards("rm -rf /tmp/example", "local")
+
+    assert result["approved"] is False
+    assert result["outcome"] == "transport_error"
+    assert calls == []
+
+
+def test_explicit_builtin_fallback_uses_existing_surface(monkeypatch):
+    from tools import approval
+
+    manager = PluginManager()
+    _context(manager).register_approval_transport("phone", lambda request: {"choice": "once"})
+    _configure_manual_guard(monkeypatch, approval, manager, fallback="builtin")
+    builtin_calls = []
+
+    result = approval.check_all_command_guards(
+        "rm -rf /tmp/example",
+        "local",
+        approval_callback=lambda *args, **kwargs: builtin_calls.append(1) or "once",
+    )
+
+    assert result["approved"] is True
+    assert builtin_calls == [1]
+
+
+def test_live_temp_home_fixture_plugin_routes_and_hardline_stays_core_owned(
+    tmp_path, monkeypatch
+):
+    """Real discovery + config + guard path under an isolated HERMES_HOME."""
+    import hermes_cli.plugins as plugins_module
+    from tools import approval
+
+    home = tmp_path / "hermes-home"
+    plugin_dir = home / "plugins" / "fixture-approval"
+    bundled = tmp_path / "empty-bundled"
+    plugin_dir.mkdir(parents=True)
+    bundled.mkdir()
+    (plugin_dir / "plugin.yaml").write_text(
+        yaml.safe_dump(
+            {
+                "name": "fixture-approval",
+                "version": "1.0.0",
+                "description": "approval transport fixture",
+            }
+        )
+    )
+    (plugin_dir / "__init__.py").write_text(
+        """import json
+import os
+from pathlib import Path
+
+
+def present(request):
+    output = Path(os.environ["HERMES_HOME"]) / "transport-invocations.jsonl"
+    with output.open("a", encoding="utf-8") as handle:
+        handle.write(json.dumps({
+            "request_id": request.request_id,
+            "digest": request.digest,
+            "command": request.command,
+            "surface": request.surface,
+            "timeout_seconds": request.timeout_seconds,
+        }) + "\\n")
+    return request.respond("once")
+
+
+def register(ctx):
+    ctx.register_approval_transport("fixture", present)
+"""
+    )
+    home.mkdir(exist_ok=True)
+    (home / "config.yaml").write_text(
+        yaml.safe_dump(
+            {
+                "plugins": {"enabled": ["fixture-approval"]},
+                "approvals": {"mode": "manual", "timeout": 2},
+                "security": {
+                    "tirith_enabled": False,
+                    "approval": {"transport": "fixture"},
+                },
+            }
+        )
+    )
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    monkeypatch.setenv("HERMES_BUNDLED_PLUGINS", str(bundled))
+    monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
+    monkeypatch.setattr(approval, "_YOLO_MODE_FROZEN", False)
+    manager = PluginManager()
+    monkeypatch.setattr(plugins_module, "_plugin_manager", manager)
+    manager.discover_and_load()
+    token = approval.set_hermes_interactive_context(True)
+    approval.clear_session("local")
+    approval._permanent_approved.clear()
+    try:
+        routed = approval.check_all_command_guards(
+            "rm -rf /tmp/hermes-approval-transport-fixture", "local"
+        )
+        manager.discover_and_load(force=True)
+        reloaded = approval.check_all_command_guards(
+            "rm -rf /tmp/hermes-approval-transport-fixture-reloaded", "local"
+        )
+        gateway_token = approval.set_hermes_interactive_context(False)
+        monkeypatch.setenv("HERMES_GATEWAY_SESSION", "1")
+        try:
+            gateway_routed = approval.check_all_command_guards(
+                "rm -rf /tmp/hermes-approval-transport-fixture-gateway", "local"
+            )
+        finally:
+            approval.reset_hermes_interactive_context(gateway_token)
+        hardline = approval.check_all_command_guards("rm -rf /", "local")
+    finally:
+        approval.reset_hermes_interactive_context(token)
+
+    records = [
+        json.loads(line)
+        for line in (home / "transport-invocations.jsonl").read_text().splitlines()
+    ]
+    assert routed["approved"] is True
+    assert reloaded["approved"] is True
+    assert gateway_routed["approved"] is True
+    assert records[0]["request_id"]
+    assert records[0]["digest"]
+    assert records[0]["surface"] == "cli"
+    assert records[0]["timeout_seconds"] == 2
+    assert records[2]["surface"] == "gateway"
+    assert hardline["approved"] is False
+    assert len(records) == 3
+
+
+def test_hardline_blocks_before_selected_transport(monkeypatch):
+    from tools import approval
+
+    manager = PluginManager()
+    calls = []
+    _context(manager).register_approval_transport(
+        "phone", lambda request: calls.append(request) or request.respond("once")
+    )
+    _configure_manual_guard(monkeypatch, approval, manager)
+    monkeypatch.setattr(
+        approval,
+        "detect_hardline_command",
+        lambda command: (True, "recursive delete of root filesystem"),
+    )
+
+    result = approval.check_all_command_guards("rm -rf /", "local")
+
+    assert result["approved"] is False
+    assert calls == []

tools/approval.py

@@ -3666,6 +3666,160 @@ def _format_tirith_description(tirith_result: dict) -> str:
     return "Security scan — " + "; ".join(parts)
 
 
+def get_plugin_manager():
+    """Lazy plugin-manager seam used by tests and early tool-only imports."""
+    from hermes_cli.plugins import get_plugin_manager as _get_manager
+
+    return _get_manager()
+
+
+def _get_approval_transport_config() -> tuple[str, str | None]:
+    """Return explicitly selected transport and fail-closed fallback mode."""
+    try:
+        from hermes_cli.config import load_config_readonly
+
+        config = load_config_readonly() or {}
+        approval_config = ((config.get("security") or {}).get("approval") or {})
+        selected = str(approval_config.get("transport") or "builtin").strip().lower()
+        fallback = str(approval_config.get("transport_fallback") or "").strip().lower()
+    except Exception:
+        # An unreadable/malformed selection must not silently materialize a
+        # prompt on a built-in surface the operator may not be watching.
+        return "config-error", None
+    return selected or "builtin", "builtin" if fallback == "builtin" else None
+
+
+def _present_with_selected_transport(
+    *,
+    command: str,
+    description: str,
+    pattern_key: str,
+    pattern_keys: list[str],
+    session_key: str,
+    surface: str,
+    allow_session: bool,
+    allow_permanent: bool,
+) -> dict:
+    """Present through an explicitly selected plugin transport, if any."""
+    name, fallback = _get_approval_transport_config()
+    if name == "builtin":
+        return {"selected": False}
+
+    try:
+        registered = get_plugin_manager().get_approval_transport(name)
+    except Exception:
+        logger.warning("Could not resolve selected approval transport %r", name, exc_info=True)
+        registered = None
+    if registered is None:
+        logger.warning("Selected approval transport %r is unavailable", name)
+        return {
+            "selected": True,
+            "choice": "deny",
+            "failure": "unavailable",
+            "fallback": fallback,
+            "name": name,
+        }
+
+    try:
+        from agent.redact import redact_sensitive_text
+        from hermes_cli.approval_transport import ApprovalRequest, invoke_approval_transport
+
+        timeout_seconds = _get_approval_timeout()
+        request = ApprovalRequest.create(
+            command=redact_sensitive_text(command),
+            description=redact_sensitive_text(description),
+            pattern_key=pattern_key,
+            pattern_keys=tuple(pattern_keys),
+            session_key=session_key,
+            surface=surface,
+            allow_session=allow_session,
+            allow_permanent=allow_permanent,
+            timeout_seconds=timeout_seconds,
+        )
+    except Exception:
+        # Never fall back to raw text if redaction or request construction
+        # fails. The selected boundary exists, so fail closed without calling
+        # the plugin or leaking the unredacted payload to logs/hooks.
+        logger.warning("Could not build redacted plugin approval request")
+        return {
+            "selected": True,
+            "choice": "deny",
+            "failure": "error",
+            "fallback": None,
+            "name": name,
+        }
+    hook_surface = f"transport:{name}"
+    _fire_approval_hook(
+        "pre_approval_request",
+        command=request.command,
+        description=request.description,
+        pattern_key=pattern_key,
+        pattern_keys=list(pattern_keys),
+        session_key=session_key,
+        surface=hook_surface,
+        request_id=request.request_id,
+        request_digest=request.digest,
+    )
+    try:
+        from tools.environments.base import touch_activity_if_due
+    except Exception:  # pragma: no cover - minimal tool-only environments
+        touch_activity_if_due = None
+    now = time.monotonic()
+    activity_state = {"last_touch": now, "start": now}
+
+    def _poll() -> None:
+        if touch_activity_if_due is not None:
+            touch_activity_if_due(activity_state, "waiting for plugin approval transport")
+
+    with human_wait_window(session_key):
+        result = invoke_approval_transport(
+            registered.present,
+            request,
+            timeout_seconds=timeout_seconds,
+            on_poll=_poll,
+            is_interrupted=is_interrupted,
+        )
+    hook_choice = result.choice if result.failure is None else f"transport_{result.failure}"
+    _fire_approval_hook(
+        "post_approval_response",
+        command=request.command,
+        description=request.description,
+        pattern_key=pattern_key,
+        pattern_keys=list(pattern_keys),
+        session_key=session_key,
+        surface=hook_surface,
+        choice=hook_choice,
+        request_id=request.request_id,
+        request_digest=request.digest,
+    )
+    return {
+        "selected": True,
+        "choice": result.choice,
+        "failure": result.failure,
+        "fallback": fallback,
+        "name": name,
+    }
+
+
+def _transport_denied_result(
+    *, pattern_key: str, description: str, failure: str
+) -> dict:
+    breaker_addendum = _denial_breaker_addendum(get_current_session_key())
+    return {
+        "approved": False,
+        "message": (
+            f"BLOCKED: Selected approval transport failed ({failure}); the user "
+            "has NOT consented to this action. Do NOT retry this command or "
+            "attempt the same outcome through another route."
+            f"{breaker_addendum}"
+        ),
+        "pattern_key": pattern_key,
+        "description": description,
+        "outcome": f"transport_{failure}",
+        "user_consent": False,
+    }
+
+
 def _await_gateway_decision(session_key: str, notify_cb, approval_data: dict,
                             *, surface: str = "gateway") -> dict:
     """Enqueue *approval_data*, notify the user, and block the calling agent
@@ -4060,6 +4214,70 @@ def check_all_command_guards(command: str, env_type: str,
     # session — the UI was stricter than the persistence layer.
     has_permanent_capable = any(not is_t for _, _, is_t in warnings)
 
+    # An explicitly selected plugin transport replaces every built-in prompt
+    # surface (CLI/TUI/gateway/ACP). Detection, allowed scopes, persistence,
+    # timeout, and final authorization remain host-owned. A failed transport
+    # reaches a built-in surface only under the explicit fallback opt-in.
+    transport_attempt = _present_with_selected_transport(
+        command=command,
+        description=combined_desc,
+        pattern_key=primary_key,
+        pattern_keys=all_keys,
+        session_key=session_key,
+        surface="gateway" if (is_gateway or is_ask) else "cli",
+        allow_session=not smart_denied_for_owner,
+        allow_permanent=has_permanent_capable and not smart_denied_for_owner,
+    )
+    if transport_attempt.get("selected"):
+        transport_failure = transport_attempt.get("failure")
+        if transport_failure and transport_attempt.get("fallback") == "builtin":
+            logger.warning(
+                "Approval transport %r failed (%s); using explicit builtin fallback",
+                transport_attempt.get("name"),
+                transport_failure,
+            )
+        elif transport_failure:
+            return _transport_denied_result(
+                pattern_key=primary_key,
+                description=combined_desc,
+                failure=transport_failure,
+            )
+        else:
+            transport_choice = transport_attempt.get("choice")
+            if transport_choice == "deny":
+                _record_denial(session_key)
+                breaker_addendum = _denial_breaker_addendum(session_key)
+                return {
+                    "approved": False,
+                    "message": (
+                        "BLOCKED: User denied this command through the selected "
+                        "approval transport. The user has NOT consented to this "
+                        "action. Do NOT retry or attempt the same outcome through "
+                        f"another route.{breaker_addendum}"
+                    ),
+                    "pattern_key": primary_key,
+                    "description": combined_desc,
+                    "outcome": "denied",
+                    "user_consent": False,
+                }
+            if not smart_denied_for_owner:
+                for key, _, is_tirith in warnings:
+                    if transport_choice == "session" or (
+                        transport_choice == "always" and is_tirith
+                    ):
+                        approve_session(session_key, key)
+                    elif transport_choice == "always":
+                        approve_session(session_key, key)
+                        approve_permanent(key)
+                        save_permanent_allowlist(_permanent_approved)
+            _reset_denials(session_key)
+            return {
+                "approved": True,
+                "message": None,
+                "user_approved": True,
+                "description": combined_desc,
+            }
+
     # Gateway/async approval — block the agent thread until the user
     # responds with /approve or /deny, mirroring the CLI's synchronous
     # input() flow.  The agent never sees "approval_required"; it either
@@ -4423,6 +4641,60 @@ def check_execute_code_guard(code: str, env_type: str,
     display_code = redact_sensitive_text(code)
     display_description = redact_sensitive_text(description)
 
+    transport_attempt = _present_with_selected_transport(
+        command=command,
+        description=description,
+        pattern_key=pattern_key,
+        pattern_keys=[pattern_key],
+        session_key=session_key,
+        surface="gateway",
+        allow_session=not smart_denied_for_owner,
+        allow_permanent=not smart_denied_for_owner,
+    )
+    if transport_attempt.get("selected"):
+        transport_failure = transport_attempt.get("failure")
+        if transport_failure and transport_attempt.get("fallback") == "builtin":
+            logger.warning(
+                "Approval transport %r failed (%s); using explicit builtin fallback",
+                transport_attempt.get("name"),
+                transport_failure,
+            )
+        elif transport_failure:
+            return _transport_denied_result(
+                pattern_key=pattern_key,
+                description=description,
+                failure=transport_failure,
+            )
+        else:
+            choice = transport_attempt.get("choice")
+            if choice == "deny":
+                _record_denial(session_key)
+                return {
+                    "approved": False,
+                    "message": (
+                        "BLOCKED: User denied execute_code through the selected "
+                        "approval transport. The user has NOT consented."
+                    ),
+                    "pattern_key": pattern_key,
+                    "description": description,
+                    "outcome": "denied",
+                    "user_consent": False,
+                }
+            if not smart_denied_for_owner:
+                if choice == "session":
+                    approve_session(session_key, pattern_key)
+                elif choice == "always":
+                    approve_session(session_key, pattern_key)
+                    approve_permanent(pattern_key)
+                    save_permanent_allowlist(_permanent_approved)
+            _reset_denials(session_key)
+            return {
+                "approved": True,
+                "message": None,
+                "user_approved": True,
+                "description": description,
+            }
+
     notify_cb = None
     with _lock:
         notify_cb = _gateway_notify_cbs.get(session_key)

website/docs/user-guide/features/plugins.md

@@ -112,6 +112,7 @@ Every `ctx.*` API below is available inside a plugin's `register(ctx)` function.
 | Register an image-generation backend | `ctx.register_image_gen_provider(provider)` — see [Image Generation Provider Plugins](/developer-guide/image-gen-provider-plugin) |
 | Register a video-generation backend | `ctx.register_video_gen_provider(provider)` — see [Video Generation Provider Plugins](/developer-guide/video-gen-provider-plugin) |
 | Register a context-compression engine | `ctx.register_context_engine(engine)` — see [Context Engine Plugins](/developer-guide/context-engine-plugin) |
+| Route human approval prompts | `ctx.register_approval_transport(name, present_fn)` — see [Approval transports](#approval-transports) |
 | Register a memory backend | Subclass `MemoryProvider` in `plugins/memory/<name>/__init__.py` — see [Memory Provider Plugins](/developer-guide/memory-provider-plugin) (uses a separate discovery system) |
 | Run a host-owned LLM call | `ctx.llm.complete(...)` / `ctx.llm.complete_structured(...)` — borrow the user's active model + auth for a one-shot completion with optional JSON schema validation. See [Plugin LLM Access](/developer-guide/plugin-llm-access) |
 | Register an inference backend (LLM provider) | `register_provider(ProviderProfile(...))` in `plugins/model-providers/<name>/__init__.py` — see [Model Provider Plugins](/developer-guide/model-provider-plugin) (uses a separate discovery system) |
@@ -182,6 +183,60 @@ Several categories of plugin bypass `plugins.enabled` — they're part of Hermes
 
 In short: **bundled "always-works" infrastructure loads automatically; third-party general plugins are opt-in.** The `plugins.enabled` allow-list is the gate specifically for arbitrary code a user drops into `~/.hermes/plugins/`.
 
+### Approval transports
+
+An approval transport changes **where a human sees and answers** an existing
+Hermes tool-approval request. It does not decide whether a command needs
+approval and it is not an authorization-policy API.
+
+```python
+def present(request):
+    # Deliver request.command and request.description to your UI, wait for
+    # its authenticated human response, then return a request-bound decision.
+    choice = send_to_my_ui_and_wait(request)  # once/session/always/deny
+    return request.respond(choice)
+
+
+def register(ctx):
+    ctx.register_approval_transport("my-ui", present)
+```
+
+`present` may be synchronous or async. Hermes runs it on a bounded worker and
+enforces the canonical `approvals.timeout` even if the plugin does not. The
+request is immutable and contains redacted display text, its originating
+surface, the host timeout, allowed choices, and an opaque request ID/digest.
+Return the result of
+`request.respond(choice)`; unbound dictionaries and stale or changed request
+IDs/digests are rejected. A plugin cannot return a scope that the host did not
+offer (for example, `always` on a once-only request).
+
+Registration alone does nothing. Enabling the plugin and explicitly selecting
+its transport are separate consent steps:
+
+```yaml
+plugins:
+  enabled: [my-approval-plugin]
+
+security:
+  approval:
+    transport: my-ui
+    transport_fallback: deny     # default
+```
+
+Transport exceptions, timeouts, unavailable registrations, invalid choices,
+and stale responses deny by default. To deliberately show the prompt on the
+ordinary CLI/TUI/gateway/ACP surface when the selected transport fails, set
+`transport_fallback: builtin`. Without that exact opt-in, Hermes never
+materializes the prompt on another surface.
+
+Hermes still owns hardline blocks, sudo-stdin protection, user deny rules,
+request binding, allowed scopes, persistence, hooks, and final authorization.
+Hardline commands are blocked before any transport callback. There is
+intentionally **no plugin approval policy, auto-allow callback, or required
+`pre_tool_call` policy** in this interface. A future approval-policy capability
+may use the plugin capability-consent model, but transport selection does not
+grant it.
+
 ### Migration for existing users
 
 When you upgrade to a version of Hermes that has opt-in plugins (config schema v21+), any user plugins already installed under `~/.hermes/plugins/` that weren't already in `plugins.disabled` are **automatically grandfathered** into `plugins.enabled`. Your existing setup keeps working. Bundled standalone plugins are NOT grandfathered — even existing users have to opt in explicitly. (Bundled platform/backend plugins never needed grandfathering because they were never gated.)