From 5bff2c28eea0d9fe900cc232bbbc33723ec9f12e Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Sun, 12 Jul 2026 20:50:20 -0700
Subject: [PATCH] feat(approvals): add profile-local per-tool policies

---
 agent/agent_init.py                  |   6 +
 agent/agent_runtime_helpers.py       |  19 +
 agent/memory_manager.py              |   7 +-
 agent/tool_executor.py               |   6 +
 hermes_cli/config.py                 |   8 +
 hermes_cli/middleware.py             | 195 +++++-
 model_tools.py                       |  32 +-
 tests/run_agent/test_run_agent.py    |   3 +
 tests/test_tool_approval_policies.py | 942 +++++++++++++++++++++++++++
 tools/approval.py                    |  43 +-
 tools/mcp_tool.py                    |  25 +-
 tools/registry.py                    |  33 +-
 website/docs/user-guide/security.md  |  27 +
 13 files changed, 1307 insertions(+), 39 deletions(-)
 create mode 100644 tests/test_tool_approval_policies.py

diff --git a/agent/agent_init.py b/agent/agent_init.py
index d9207b0b3d9..4c2447bdf99 100644
--- a/agent/agent_init.py
+++ b/agent/agent_init.py
@@ -1906,6 +1906,8 @@ def init_agent(
     # would still leak lcm_* tools into the tool surface and incur the
     # same local-model latency penalty.
     agent._context_engine_tool_names: set = set()
+    if not hasattr(agent, "_dynamic_tool_entries"):
+        agent._dynamic_tool_entries = {}
     if (
         hasattr(agent, "context_compressor")
         and agent.context_compressor
@@ -1940,6 +1942,10 @@ def init_agent(
             agent.tools.append(_wrapped)
             agent.valid_tool_names.add(_tname)
             agent._context_engine_tool_names.add(_tname)
+            from tools.registry import DynamicToolEntry as _DynamicToolEntry
+            agent._dynamic_tool_entries[_tname] = _DynamicToolEntry(
+                _tname, "context_engine"
+            )
             _existing_tool_names.add(_tname)
 
     # Notify context engine of session start
diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py
index 1cde73419a4..f0812664a2f 100644
--- a/agent/agent_runtime_helpers.py
+++ b/agent/agent_runtime_helpers.py
@@ -2304,6 +2304,15 @@ def invoke_tool(agent, function_name: str, function_args: dict, effective_task_i
                     ),
                 )
             return _finish_agent_tool(result, next_args)
+    elif (
+        getattr(agent, "_context_engine_tool_names", None)
+        and function_name in agent._context_engine_tool_names
+    ):
+        def _execute(next_args: dict) -> Any:
+            return _finish_agent_tool(
+                agent.context_compressor.handle_tool_call(function_name, next_args),
+                next_args,
+            )
     elif agent._memory_manager and agent._memory_manager.has_tool(function_name):
         def _execute(next_args: dict) -> Any:
             return _finish_agent_tool(agent._memory_manager.handle_tool_call(function_name, next_args), next_args)
@@ -2333,6 +2342,8 @@ def invoke_tool(agent, function_name: str, function_args: dict, effective_task_i
         def _execute(next_args: dict) -> Any:
             return _finish_agent_tool(agent._dispatch_delegate_task(next_args), next_args)
     else:
+        from model_tools import registry
+
         def _execute(next_args: dict) -> Any:
             return _ra().handle_function_call(
                 function_name, next_args, effective_task_id,
@@ -2346,9 +2357,15 @@ def invoke_tool(agent, function_name: str, function_args: dict, effective_task_i
                 enabled_toolsets=getattr(agent, "enabled_toolsets", None),
                 disabled_toolsets=getattr(agent, "disabled_toolsets", None),
                 tool_request_middleware_trace=list(_tool_middleware_trace),
+                dispatch_registry=registry,
             )
 
     from hermes_cli.middleware import run_tool_execution_middleware
+    from model_tools import registry
+
+    registry_entry = registry.get_entry(function_name)
+    if registry_entry is None:
+        registry_entry = getattr(agent, "_dynamic_tool_entries", {}).get(function_name)
 
     return run_tool_execution_middleware(
         function_name,
@@ -2360,6 +2377,8 @@ def invoke_tool(agent, function_name: str, function_args: dict, effective_task_i
         tool_call_id=tool_call_id or "",
         turn_id=getattr(agent, "_current_turn_id", "") or "",
         api_request_id=getattr(agent, "_current_api_request_id", "") or "",
+        registry_entry=registry_entry,
+        dispatch_registry=registry if registry_entry is registry.get_entry(function_name) else None,
     )
 
 
diff --git a/agent/memory_manager.py b/agent/memory_manager.py
index 32dd8e28884..9cea9e03e74 100644
--- a/agent/memory_manager.py
+++ b/agent/memory_manager.py
@@ -35,7 +35,7 @@ from typing import Any, Callable, Dict, List, Optional
 
 from agent.memory_provider import MemoryProvider
 from agent.skill_commands import extract_user_instruction_from_skill_message
-from tools.registry import tool_error
+from tools.registry import DynamicToolEntry, tool_error
 
 logger = logging.getLogger(__name__)
 
@@ -123,6 +123,10 @@ def inject_memory_provider_tools(agent: Any) -> int:
     if valid_tool_names is None:
         valid_tool_names = set()
         agent.valid_tool_names = valid_tool_names
+    dynamic_entries = getattr(agent, "_dynamic_tool_entries", None)
+    if dynamic_entries is None:
+        dynamic_entries = {}
+        agent._dynamic_tool_entries = dynamic_entries
 
     added = 0
     for raw_schema in get_schemas():
@@ -139,6 +143,7 @@ def inject_memory_provider_tools(agent: Any) -> int:
             continue
         tools.append({"type": "function", "function": schema})
         valid_tool_names.add(tool_name)
+        dynamic_entries[tool_name] = DynamicToolEntry(tool_name, "memory")
         existing_tool_names.add(tool_name)
         added += 1
 
diff --git a/agent/tool_executor.py b/agent/tool_executor.py
index ac505c6d829..e158789470d 100644
--- a/agent/tool_executor.py
+++ b/agent/tool_executor.py
@@ -307,6 +307,11 @@ def _run_agent_tool_execution_middleware(
         return execute(observed_args)
 
     from hermes_cli.middleware import run_tool_execution_middleware
+    from model_tools import registry
+
+    registry_entry = registry.get_entry(function_name)
+    if registry_entry is None:
+        registry_entry = getattr(agent, "_dynamic_tool_entries", {}).get(function_name)
 
     result = run_tool_execution_middleware(
         function_name,
@@ -318,6 +323,7 @@ def _run_agent_tool_execution_middleware(
         tool_call_id=tool_call_id or "",
         turn_id=getattr(agent, "_current_turn_id", "") or "",
         api_request_id=getattr(agent, "_current_api_request_id", "") or "",
+        registry_entry=registry_entry,
     )
     return result, observed_args
 
diff --git a/hermes_cli/config.py b/hermes_cli/config.py
index 03570bceba4..306624daf52 100644
--- a/hermes_cli/config.py
+++ b/hermes_cli/config.py
@@ -2534,6 +2534,14 @@ DEFAULT_CONFIG = {
         "mode": "smart",
         "timeout": 60,
         "cron_mode": "deny",
+        # Declarative per-tool approval policy. Keys are case-insensitive
+        # fnmatch globs matched against both the tool name and its registered
+        # toolset name; values are deny, ask, or allow. Across all matching
+        # tool-name and toolset rules, the safest decision wins
+        # (deny > ask > allow).
+        # Explicit allow never bypasses command hardlines/user denies, session
+        # tool scope, cron policy, plugin escalation, or path protections.
+        "tool_policies": {},
         # User-defined deny rules: fnmatch globs matched against terminal
         # commands. A match blocks the command unconditionally — BEFORE the
         # --yolo / /yolo / mode=off bypass — making this the user-editable
diff --git a/hermes_cli/middleware.py b/hermes_cli/middleware.py
index 8795952a2b7..bf4f446f169 100644
--- a/hermes_cli/middleware.py
+++ b/hermes_cli/middleware.py
@@ -7,13 +7,50 @@ contract helpers here so agent-loop call sites and plugins share one vocabulary.
 
 from __future__ import annotations
 
+import contextvars
+import fnmatch
+import json
 import logging
+import threading
 from copy import deepcopy
 from dataclasses import dataclass, field
 from typing import Any, Callable, Dict, List, Optional
 
 logger = logging.getLogger(__name__)
 
+_TOOL_POLICY_PRIORITY = {"allow": 0, "ask": 1, "deny": 2}
+
+
+class _RegistryDispatchCapability:
+    """Opaque, single-use permission for one wrapper→registry handoff.
+
+    Context copies share this object's consumed state, so only one dispatch can
+    redeem it even if a caller copies the Context or reuses the same args.
+    """
+
+    __slots__ = ("_registry", "_tool_name", "_args", "_consumed", "_lock")
+
+    def __init__(self, registry: Any, tool_name: str, args: Dict[str, Any]) -> None:
+        self._registry = registry
+        self._tool_name = tool_name
+        self._args = args
+        self._consumed = False
+        self._lock = threading.Lock()
+
+    def consume(self, registry: Any, tool_name: str, args: Dict[str, Any]) -> bool:
+        if registry is not self._registry or tool_name != self._tool_name or args is not self._args:
+            return False
+        with self._lock:
+            if self._consumed:
+                return False
+            self._consumed = True
+            return True
+
+
+_pending_registry_dispatch: contextvars.ContextVar[Optional[_RegistryDispatchCapability]] = (
+    contextvars.ContextVar("pending_registry_dispatch", default=None)
+)
+
 OBSERVER_SCHEMA_VERSION = "hermes.observer.v1"
 MIDDLEWARE_SCHEMA_VERSION = "hermes.middleware.v1"
 
@@ -195,19 +232,153 @@ def run_tool_execution_middleware(
     next_call: Callable[[Dict[str, Any]], Any],
     **context: Any,
 ) -> Any:
-    """Run tool execution through registered tool execution middleware."""
-    callbacks = _get_middleware_callbacks(TOOL_EXECUTION_MIDDLEWARE)
-    if not callbacks:
-        return next_call(args)
-    return _run_execution_chain(
-        TOOL_EXECUTION_MIDDLEWARE,
-        callbacks,
-        next_call,
-        tool_name=tool_name,
-        args=args,
-        original_args=context.pop("original_args", args),
-        **context,
+    """Run a tool through the shared policy gate and plugin middleware chain.
+
+    ``dispatch_registry`` marks a wrapper whose immediate ``next_call`` enters
+    that registry. Only that exact wrapper→registry handoff is deduplicated;
+    nested dispatches from inside the handler remain genuine executions.
+    """
+    dispatch_registry = context.pop("dispatch_registry", None)
+    registry_entry = context.pop("registry_entry", None)
+    registry_dispatch = context.pop("registry_dispatch", None)
+    expected_dispatch = _pending_registry_dispatch.get()
+    is_wrapped_registry_dispatch = bool(
+        registry_dispatch
+        and expected_dispatch is not None
+        and expected_dispatch.consume(registry_dispatch, tool_name, args)
     )
+    if is_wrapped_registry_dispatch:
+        _pending_registry_dispatch.set(None)
+        return next_call(args)
+
+    policy = (
+        resolve_tool_approval_policy(tool_name, entry=registry_entry)
+        if registry_entry is not None
+        else resolve_tool_approval_policy(tool_name)
+    )
+    blocked = _apply_tool_approval_policy(tool_name, policy)
+    if blocked is not None:
+        return blocked
+
+    terminal_allow_token = None
+    if tool_name == "terminal" and policy == "allow":
+        from tools.approval import set_tool_policy_terminal_allow
+
+        terminal_allow_token = set_tool_policy_terminal_allow()
+
+    def call_next(next_args: Dict[str, Any]) -> Any:
+        if dispatch_registry is None:
+            return next_call(next_args)
+        token = _pending_registry_dispatch.set(
+            _RegistryDispatchCapability(dispatch_registry, tool_name, next_args)
+        )
+        try:
+            return next_call(next_args)
+        finally:
+            _pending_registry_dispatch.reset(token)
+
+    try:
+        callbacks = _get_middleware_callbacks(TOOL_EXECUTION_MIDDLEWARE)
+        if not callbacks:
+            return call_next(args)
+        return _run_execution_chain(
+            TOOL_EXECUTION_MIDDLEWARE,
+            callbacks,
+            call_next,
+            tool_name=tool_name,
+            args=args,
+            original_args=context.pop("original_args", args),
+            **context,
+        )
+    finally:
+        if terminal_allow_token is not None:
+            from tools.approval import reset_tool_policy_terminal_allow
+
+            reset_tool_policy_terminal_allow(terminal_allow_token)
+
+
+def resolve_tool_approval_policy(
+    tool_name: str, *, entry: Any = None
+) -> Optional[str]:
+    """Resolve glob policies for a tool name and its actual dispatch entry."""
+    try:
+        from hermes_cli.config import load_config
+
+        config = load_config() or {}
+        approvals = config.get("approvals", {}) or {}
+        patterns = approvals.get("tool_policies", {}) or {}
+    except Exception as exc:
+        logger.warning("Failed to load tool approval policies: %s", exc)
+        return None
+    if not isinstance(patterns, dict):
+        return None
+
+    identifiers = [str(tool_name or "").strip().lower()]
+    toolset = str(getattr(entry, "toolset", "") or "").strip().lower()
+    if toolset:
+        identifiers.append(toolset)
+
+    matches: list[str] = []
+    for pattern, raw_policy in patterns.items():
+        if not isinstance(pattern, str) or not isinstance(raw_policy, str):
+            continue
+        normalized_pattern = pattern.strip().lower()
+        policy = raw_policy.strip().lower()
+        if not normalized_pattern or policy not in _TOOL_POLICY_PRIORITY:
+            continue
+        if any(
+            fnmatch.fnmatchcase(identifier, normalized_pattern)
+            for identifier in identifiers
+        ):
+            matches.append(policy)
+    if not matches:
+        return None
+    return max(matches, key=_TOOL_POLICY_PRIORITY.__getitem__)
+
+
+def _apply_tool_approval_policy(
+    tool_name: str, policy: Optional[str]
+) -> Optional[str]:
+    """Return a synthetic blocked result, or None to continue execution."""
+    if policy == "deny":
+        return json.dumps(
+            {
+                "error": (
+                    f"BLOCKED: Tool '{tool_name}' is denied by "
+                    "approvals.tool_policies in config.yaml."
+                ),
+                "policy": "deny",
+                "tool": tool_name,
+            },
+            ensure_ascii=False,
+        )
+    if policy != "ask":
+        return None
+
+    try:
+        from tools.approval import request_tool_approval
+
+        decision = request_tool_approval(
+            tool_name,
+            f"config.yaml requires approval for tool '{tool_name}'",
+            rule_key=f"tool_policy:{tool_name}",
+            enforce_under_yolo=True,
+        )
+    except Exception as exc:
+        logger.warning("Tool approval policy check failed for %s: %s", tool_name, exc)
+        decision = {
+            "approved": False,
+            "message": "BLOCKED: tool approval policy could not be evaluated.",
+        }
+    if decision.get("approved"):
+        return None
+    payload = dict(decision)
+    payload["error"] = payload.pop("message", None) or (
+        f"BLOCKED: Tool '{tool_name}' was not approved."
+    )
+    payload.setdefault("policy", "ask")
+    payload.setdefault("tool", tool_name)
+    return json.dumps(payload, ensure_ascii=False)
 
 
 def run_api_execution_middleware(
diff --git a/model_tools.py b/model_tools.py
index c59c189e36d..215e572c6b2 100644
--- a/model_tools.py
+++ b/model_tools.py
@@ -1037,6 +1037,7 @@ def handle_function_call(
     tool_request_middleware_trace: Optional[List[Dict[str, Any]]] = None,
     enabled_toolsets: Optional[List[str]] = None,
     disabled_toolsets: Optional[List[str]] = None,
+    dispatch_registry: Any = None,
 ) -> str:
     """
     Main function call dispatcher that routes calls to the tool registry.
@@ -1276,19 +1277,26 @@ def handle_function_call(
                         session_id=session_id,
                         user_task=user_task,
                     )
-            from hermes_cli.middleware import run_tool_execution_middleware
+            if dispatch_registry is not None:
+                # An outer runtime wrapper already ran policy/middleware and
+                # armed the exact wrapper→registry handoff capability.
+                result = _dispatch(function_args)
+            else:
+                from hermes_cli.middleware import run_tool_execution_middleware
 
-            result = run_tool_execution_middleware(
-                function_name,
-                function_args,
-                _dispatch,
-                original_args=_tool_original_args,
-                task_id=task_id or "",
-                session_id=session_id or "",
-                tool_call_id=tool_call_id or "",
-                turn_id=turn_id or "",
-                api_request_id=api_request_id or "",
-            )
+                result = run_tool_execution_middleware(
+                    function_name,
+                    function_args,
+                    _dispatch,
+                    original_args=_tool_original_args,
+                    task_id=task_id or "",
+                    session_id=session_id or "",
+                    tool_call_id=tool_call_id or "",
+                    turn_id=turn_id or "",
+                    api_request_id=api_request_id or "",
+                    registry_entry=registry.get_entry(function_name),
+                    dispatch_registry=registry,
+                )
         finally:
             if _approval_tokens is not None and reset_current_observability_context is not None:
                 try:
diff --git a/tests/run_agent/test_run_agent.py b/tests/run_agent/test_run_agent.py
index 396657ea4e8..00a48700686 100644
--- a/tests/run_agent/test_run_agent.py
+++ b/tests/run_agent/test_run_agent.py
@@ -2854,6 +2854,8 @@ class TestConcurrentToolExecution:
 
     def test_invoke_tool_dispatches_to_handle_function_call(self, agent):
         """_invoke_tool should route regular tools through handle_function_call."""
+        import model_tools
+
         with patch("run_agent.handle_function_call", return_value="result") as mock_hfc:
             result = agent._invoke_tool("web_search", {"q": "test"}, "task-1")
             mock_hfc.assert_called_once_with(
@@ -2868,6 +2870,7 @@ class TestConcurrentToolExecution:
                 enabled_toolsets=agent.enabled_toolsets,
                 disabled_toolsets=agent.disabled_toolsets,
                 tool_request_middleware_trace=[],
+                dispatch_registry=model_tools.registry,
             )
             assert result == "result"
 
diff --git a/tests/test_tool_approval_policies.py b/tests/test_tool_approval_policies.py
new file mode 100644
index 00000000000..e3ee74e343b
--- /dev/null
+++ b/tests/test_tool_approval_policies.py
@@ -0,0 +1,942 @@
+from __future__ import annotations
+
+import contextvars
+import json
+from concurrent.futures import ThreadPoolExecutor
+from threading import Barrier, Lock
+from types import SimpleNamespace
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def _no_plugin_execution_middleware(monkeypatch):
+    monkeypatch.setattr(
+        "hermes_cli.middleware._get_middleware_callbacks", lambda _kind: []
+    )
+
+
+def test_policy_resolves_tool_name_before_toolset_and_uses_most_restrictive(monkeypatch):
+    from hermes_cli import middleware
+
+    monkeypatch.setattr(
+        "hermes_cli.config.load_config",
+        lambda: {
+            "approvals": {
+                "tool_policies": {
+                    "file*": "allow",
+                    "write_*": "ask",
+                    "write_file": "deny",
+                }
+            }
+        },
+    )
+    entry = SimpleNamespace(toolset="files")
+
+    assert middleware.resolve_tool_approval_policy("write_file", entry=entry) == "deny"
+
+
+def test_broad_deny_outweighs_exact_allow(monkeypatch):
+    from hermes_cli import middleware
+
+    monkeypatch.setattr(
+        "hermes_cli.config.load_config",
+        lambda: {
+            "approvals": {
+                "tool_policies": {
+                    "*": "deny",
+                    "read_file": "allow",
+                }
+            }
+        },
+    )
+
+    assert middleware.resolve_tool_approval_policy("read_file") == "deny"
+
+
+def test_toolset_deny_outweighs_tool_name_allow(monkeypatch):
+    from hermes_cli import middleware
+
+    monkeypatch.setattr(
+        "hermes_cli.config.load_config",
+        lambda: {
+            "approvals": {
+                "tool_policies": {
+                    "read_file": "allow",
+                    "file": "deny",
+                }
+            }
+        },
+    )
+    entry = SimpleNamespace(toolset="file")
+
+    assert middleware.resolve_tool_approval_policy("read_file", entry=entry) == "deny"
+
+
+def test_policy_can_match_registered_toolset(monkeypatch):
+    from hermes_cli import middleware
+
+    monkeypatch.setattr(
+        "hermes_cli.config.load_config",
+        lambda: {"approvals": {"tool_policies": {"browser*": "ask"}}},
+    )
+    entry = SimpleNamespace(toolset="browser")
+
+    assert middleware.resolve_tool_approval_policy("browser_click", entry=entry) == "ask"
+
+
+def test_malformed_policy_entries_are_ignored(monkeypatch):
+    from hermes_cli import middleware
+
+    monkeypatch.setattr(
+        "hermes_cli.config.load_config",
+        lambda: {
+            "approvals": {
+                "tool_policies": {
+                    "terminal": "maybe",
+                    42: "deny",
+                    "*": None,
+                }
+            }
+        },
+    )
+
+    assert middleware.resolve_tool_approval_policy("terminal") is None
+
+
+def test_deny_policy_blocks_without_calling_tool(monkeypatch):
+    from hermes_cli import middleware
+
+    monkeypatch.setattr(
+        middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "deny"
+    )
+    called = []
+
+    result = middleware.run_tool_execution_middleware(
+        "write_file", {"path": "notes.txt"}, lambda args: called.append(args)
+    )
+
+    assert called == []
+    assert json.loads(result)["error"].startswith("BLOCKED: Tool 'write_file'")
+
+
+def test_ask_policy_uses_shared_fail_closed_approval_gate(monkeypatch):
+    from hermes_cli import middleware
+
+    monkeypatch.setattr(
+        middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "ask"
+    )
+    requested = []
+    monkeypatch.setattr(
+        "tools.approval.request_tool_approval",
+        lambda tool_name, reason, **kwargs: requested.append(
+            (tool_name, reason, kwargs)
+        )
+        or {"approved": False, "message": "BLOCKED by cron policy"},
+    )
+    called = []
+
+    result = middleware.run_tool_execution_middleware(
+        "terminal", {"command": "printf ok"}, lambda args: called.append(args)
+    )
+
+    assert json.loads(result)["error"] == "BLOCKED by cron policy"
+    assert called == []
+    assert requested[0][0] == "terminal"
+    assert requested[0][2]["rule_key"] == "tool_policy:terminal"
+
+
+def test_allow_policy_does_not_skip_downstream_execution_middleware(monkeypatch):
+    from hermes_cli import middleware
+
+    monkeypatch.setattr(
+        middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "allow"
+    )
+    events = []
+
+    def plugin_middleware(**kwargs):
+        events.append("plugin")
+        return kwargs["next_call"](kwargs["args"])
+
+    monkeypatch.setattr(
+        middleware, "_get_middleware_callbacks", lambda _kind: [plugin_middleware]
+    )
+
+    result = middleware.run_tool_execution_middleware(
+        "read_file", {"path": "README.md"}, lambda _args: events.append("tool") or "ok"
+    )
+
+    assert result == "ok"
+    assert events == ["plugin", "tool"]
+
+
+def test_direct_registry_dispatch_is_policy_gated(monkeypatch):
+    import model_tools
+    from hermes_cli import middleware
+
+    monkeypatch.setattr(
+        middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "deny"
+    )
+    dispatched = []
+    monkeypatch.setattr(
+        model_tools.registry,
+        "dispatch",
+        lambda name, args, **kwargs: dispatched.append((name, args)) or "ran",
+    )
+
+    result = model_tools.handle_function_call("read_file", {"path": "README.md"})
+
+    assert dispatched == []
+    assert "Tool 'read_file' is denied" in json.loads(result)["error"]
+
+
+def test_registry_dispatch_itself_is_policy_gated(monkeypatch):
+    from hermes_cli import middleware
+    from tools.registry import ToolRegistry
+
+    monkeypatch.setattr(
+        middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "deny"
+    )
+    called = []
+    registry = ToolRegistry()
+    registry.register(
+        name="plugin_tool",
+        toolset="plugin",
+        schema={},
+        handler=lambda args, **kwargs: called.append(args) or '{"ran": true}',
+    )
+
+    result = registry.dispatch("plugin_tool", {"value": 1})
+
+    assert called == []
+    assert "Tool 'plugin_tool' is denied" in json.loads(result)["error"]
+
+
+def test_custom_registry_dispatch_resolves_policy_from_its_own_toolset(monkeypatch):
+    from tools.registry import ToolRegistry
+
+    monkeypatch.setattr(
+        "hermes_cli.config.load_config",
+        lambda: {"approvals": {"tool_policies": {"custom-toolset": "deny"}}},
+    )
+    called = []
+    custom_registry = ToolRegistry()
+    custom_registry.register(
+        name="shared_name",
+        toolset="custom-toolset",
+        schema={},
+        handler=lambda args, **kwargs: called.append(args) or "ran",
+    )
+
+    result = custom_registry.dispatch("shared_name", {})
+
+    assert called == []
+    assert json.loads(result)["policy"] == "deny"
+
+
+def test_genuine_recursive_same_name_dispatch_rechecks_policy_and_middleware(monkeypatch):
+    from hermes_cli import middleware
+    from tools.registry import ToolRegistry
+
+    monkeypatch.setattr(
+        middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "ask"
+    )
+    approvals = []
+    monkeypatch.setattr(
+        "tools.approval.request_tool_approval",
+        lambda *_args, **_kwargs: approvals.append(True)
+        or {"approved": True, "message": None},
+    )
+    executions = []
+
+    def execution_middleware(**kwargs):
+        executions.append(kwargs["args"]["depth"])
+        return kwargs["next_call"](kwargs["args"])
+
+    monkeypatch.setattr(
+        middleware, "_get_middleware_callbacks", lambda _kind: [execution_middleware]
+    )
+    custom_registry = ToolRegistry()
+
+    def recursive_handler(args, **_kwargs):
+        if args["depth"]:
+            return custom_registry.dispatch("recursive", {"depth": args["depth"] - 1})
+        return '{"done": true}'
+
+    custom_registry.register("recursive", "custom", {}, recursive_handler)
+
+    assert json.loads(custom_registry.dispatch("recursive", {"depth": 1})) == {"done": True}
+    assert approvals == [True, True]
+    assert executions == [1, 0]
+
+
+def test_wrapper_to_registry_dispatch_is_deduped_exactly_once(monkeypatch):
+    from hermes_cli import middleware
+    from tools.registry import ToolRegistry
+
+    monkeypatch.setattr(
+        middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "ask"
+    )
+    approvals = []
+    monkeypatch.setattr(
+        "tools.approval.request_tool_approval",
+        lambda *_args, **_kwargs: approvals.append(True)
+        or {"approved": True, "message": None},
+    )
+    executions = []
+
+    def execution_middleware(**kwargs):
+        executions.append(kwargs["tool_name"])
+        return kwargs["next_call"](kwargs["args"])
+
+    monkeypatch.setattr(
+        middleware, "_get_middleware_callbacks", lambda _kind: [execution_middleware]
+    )
+    custom_registry = ToolRegistry()
+    custom_registry.register("wrapped", "custom", {}, lambda _args, **_kwargs: "ok")
+    args = {"value": 1}
+
+    result = middleware.run_tool_execution_middleware(
+        "wrapped",
+        args,
+        lambda next_args: custom_registry.dispatch("wrapped", next_args),
+        dispatch_registry=custom_registry,
+    )
+
+    assert result == "ok"
+    assert approvals == [True]
+    assert executions == ["wrapped"]
+
+
+def test_copied_wrapper_context_cannot_reuse_consumed_registry_handoff(monkeypatch):
+    from hermes_cli import middleware
+    from tools.registry import ToolRegistry
+
+    monkeypatch.setattr(middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "ask")
+    approvals = []
+    monkeypatch.setattr(
+        "tools.approval.request_tool_approval",
+        lambda *_args, **_kwargs: approvals.append(True) or {"approved": True, "message": None},
+    )
+    custom_registry = ToolRegistry()
+    custom_registry.register("wrapped", "custom", {}, lambda _args, **_kwargs: "ok")
+    args = {"value": 1}
+    copied = None
+
+    def dispatch_and_copy(next_args):
+        nonlocal copied
+        copied = contextvars.copy_context()
+        return custom_registry.dispatch("wrapped", next_args)
+
+    assert middleware.run_tool_execution_middleware(
+        "wrapped", args, dispatch_and_copy, dispatch_registry=custom_registry
+    ) == "ok"
+    assert copied is not None
+    assert copied.run(custom_registry.dispatch, "wrapped", args) == "ok"
+    assert approvals == [True, True]
+
+
+def test_one_wrapper_handoff_cannot_suppress_two_same_argument_dispatches(monkeypatch):
+    from hermes_cli import middleware
+    from tools.registry import ToolRegistry
+
+    monkeypatch.setattr(middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "ask")
+    approvals = []
+    monkeypatch.setattr(
+        "tools.approval.request_tool_approval",
+        lambda *_args, **_kwargs: approvals.append(True) or {"approved": True, "message": None},
+    )
+    custom_registry = ToolRegistry()
+    custom_registry.register("wrapped", "custom", {}, lambda _args, **_kwargs: "ok")
+    args = {"value": 1}
+
+    def dispatch_twice(next_args):
+        custom_registry.dispatch("wrapped", next_args)
+        return custom_registry.dispatch("wrapped", next_args)
+
+    assert middleware.run_tool_execution_middleware(
+        "wrapped", args, dispatch_twice, dispatch_registry=custom_registry
+    ) == "ok"
+    assert approvals == [True, True]
+
+
+def test_wrapper_dedupe_does_not_cover_a_different_tool_name(monkeypatch):
+    from hermes_cli import middleware
+    from tools.registry import ToolRegistry
+
+    monkeypatch.setattr(
+        middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "ask"
+    )
+    approvals = []
+    monkeypatch.setattr(
+        "tools.approval.request_tool_approval",
+        lambda *_args, **_kwargs: approvals.append(True)
+        or {"approved": True, "message": None},
+    )
+    custom_registry = ToolRegistry()
+    custom_registry.register("outer", "custom", {}, lambda _args, **_kwargs: "unused")
+    custom_registry.register("inner", "custom", {}, lambda _args, **_kwargs: "ok")
+    args = {"value": 1}
+
+    result = middleware.run_tool_execution_middleware(
+        "outer",
+        args,
+        lambda next_args: custom_registry.dispatch("inner", next_args),
+        dispatch_registry=custom_registry,
+    )
+
+    assert result == "ok"
+    assert approvals == [True, True]
+
+
+def test_wrapper_registry_dedupe_tokens_are_isolated_across_concurrent_contexts(monkeypatch):
+    from hermes_cli import middleware
+    from tools.registry import ToolRegistry
+
+    monkeypatch.setattr(
+        middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "ask"
+    )
+    approvals = []
+    lock = Lock()
+
+    def approve(*_args, **_kwargs):
+        with lock:
+            approvals.append(True)
+        return {"approved": True, "message": None}
+
+    monkeypatch.setattr("tools.approval.request_tool_approval", approve)
+    executions = []
+
+    def execution_middleware(**kwargs):
+        with lock:
+            executions.append(kwargs["args"]["call"])
+        return kwargs["next_call"](kwargs["args"])
+
+    monkeypatch.setattr(
+        middleware, "_get_middleware_callbacks", lambda _kind: [execution_middleware]
+    )
+    barrier = Barrier(2)
+    custom_registry = ToolRegistry()
+
+    def handler(args, **_kwargs):
+        barrier.wait(timeout=5)
+        return str(args["call"])
+
+    custom_registry.register("concurrent", "custom", {}, handler)
+
+    def invoke(call):
+        args = {"call": call}
+        return middleware.run_tool_execution_middleware(
+            "concurrent",
+            args,
+            lambda next_args: custom_registry.dispatch("concurrent", next_args),
+            dispatch_registry=custom_registry,
+        )
+
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        results = list(pool.map(invoke, (1, 2)))
+
+    assert results == ["1", "2"]
+    assert len(approvals) == 2
+    assert sorted(executions) == [1, 2]
+
+
+def test_plugin_context_dispatch_cannot_bypass_policy(monkeypatch):
+    from hermes_cli import middleware
+    from hermes_cli.plugins import PluginContext, PluginManager, PluginManifest
+    from tools.registry import registry
+
+    monkeypatch.setattr(
+        middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "deny"
+    )
+    called = []
+    registry.register(
+        name="plugin_context_policy_test",
+        toolset="plugin-test",
+        schema={},
+        handler=lambda args, **kwargs: called.append(args) or '{"ran": true}',
+    )
+    try:
+        context = PluginContext(
+            PluginManifest(name="policy-test", source="user"), PluginManager()
+        )
+        result = context.dispatch_tool("plugin_context_policy_test", {})
+    finally:
+        registry.deregister("plugin_context_policy_test")
+
+    assert called == []
+    assert "is denied" in json.loads(result)["error"]
+
+
+def test_nested_model_and_registry_dispatch_ask_only_once(monkeypatch):
+    import model_tools
+    from hermes_cli import middleware
+
+    monkeypatch.setattr(
+        middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "ask"
+    )
+    approvals = []
+    monkeypatch.setattr(
+        "tools.approval.request_tool_approval",
+        lambda *_args, **_kwargs: approvals.append(True)
+        or {"approved": True, "message": None},
+    )
+
+    result = model_tools.handle_function_call(
+        "read_file", {"path": "README.md"}, skip_pre_tool_call_hook=True
+    )
+
+    assert json.loads(result).get("error") is None
+    assert approvals == [True]
+
+
+def test_nested_model_and_registry_dispatch_runs_execution_middleware_once(monkeypatch):
+    import model_tools
+    from hermes_cli import middleware
+
+    monkeypatch.setattr(
+        middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: None
+    )
+    executions = []
+
+    def execution_middleware(**kwargs):
+        executions.append(kwargs["tool_name"])
+        return kwargs["next_call"](kwargs["args"])
+
+    monkeypatch.setattr(
+        middleware,
+        "_get_middleware_callbacks",
+        lambda _kind: [execution_middleware],
+    )
+
+    result = model_tools.handle_function_call(
+        "read_file", {"path": "README.md"}, skip_pre_tool_call_hook=True
+    )
+
+    assert json.loads(result).get("error") is None
+    assert executions == ["read_file"]
+
+
+def test_sequential_agent_level_dispatch_is_policy_gated(monkeypatch):
+    from agent import tool_executor
+    from hermes_cli import middleware
+
+    monkeypatch.setattr(
+        middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "deny"
+    )
+    called = []
+    agent = SimpleNamespace(
+        session_id="s",
+        _current_turn_id="t",
+        _current_api_request_id="r",
+    )
+
+    result, _ = tool_executor._run_agent_tool_execution_middleware(
+        agent,
+        function_name="todo",
+        function_args={"todos": []},
+        effective_task_id="task",
+        tool_call_id="call",
+        execute=lambda args: called.append(args) or "ran",
+    )
+
+    assert called == []
+    assert "Tool 'todo' is denied" in json.loads(result)["error"]
+
+
+def test_concurrent_agent_level_dispatch_is_policy_gated(monkeypatch):
+    from agent import agent_runtime_helpers
+    from hermes_cli import middleware
+
+    monkeypatch.setattr(
+        middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "deny"
+    )
+    called = []
+    agent = SimpleNamespace(
+        session_id="s",
+        _current_turn_id="t",
+        _current_api_request_id="r",
+        _todo_store=object(),
+        _memory_manager=None,
+        valid_tool_names=set(),
+        enabled_toolsets=None,
+        disabled_toolsets=None,
+    )
+    monkeypatch.setattr(
+        "tools.todo_tool.todo_tool", lambda **kwargs: called.append(kwargs) or "ran"
+    )
+
+    result = agent_runtime_helpers.invoke_tool(
+        agent, "todo", {"todos": []}, "task", tool_call_id="call",
+        pre_tool_block_checked=True,
+    )
+
+    assert called == []
+    assert "Tool 'todo' is denied" in json.loads(result)["error"]
+
+
+def test_runtime_fallback_registry_runs_policy_and_execution_middleware_once(monkeypatch):
+    """The real invoke_tool fallback must hand off directly to the registry once."""
+    import model_tools
+    from agent import agent_runtime_helpers
+    from hermes_cli import middleware
+
+    approvals = []
+    executions = []
+    monkeypatch.setattr(
+        middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "ask"
+    )
+    monkeypatch.setattr(
+        "tools.approval.request_tool_approval",
+        lambda *_args, **_kwargs: approvals.append(True)
+        or {"approved": True, "message": None},
+    )
+
+    def execution_middleware(**kwargs):
+        executions.append(kwargs["tool_name"])
+        return kwargs["next_call"](kwargs["args"])
+
+    monkeypatch.setattr(
+        middleware, "_get_middleware_callbacks", lambda _kind: [execution_middleware]
+    )
+    tool_name = "runtime_fallback_policy_probe"
+    model_tools.registry.register(
+        tool_name, "probe", {}, lambda _args, **_kwargs: '{"success": true}'
+    )
+    agent = SimpleNamespace(
+        session_id="s",
+        _current_turn_id="t",
+        _current_api_request_id="r",
+        _memory_manager=None,
+        _context_engine_tool_names=set(),
+        valid_tool_names={tool_name},
+        enabled_toolsets=None,
+        disabled_toolsets=None,
+    )
+    try:
+        result = agent_runtime_helpers.invoke_tool(
+            agent, tool_name, {}, "task", tool_call_id="call", pre_tool_block_checked=True
+        )
+    finally:
+        model_tools.registry.deregister(tool_name)
+
+    assert json.loads(result) == {"success": True}
+    assert approvals == [True]
+    assert executions == [tool_name]
+
+
+@pytest.mark.parametrize(
+    ("family", "tool_name", "toolset"),
+    [
+        ("context", "context_policy_probe", "context_engine"),
+        ("memory", "memory_policy_probe", "memory"),
+    ],
+)
+def test_dynamic_tool_family_deny_overrides_exact_allow_on_runtime_path(
+    monkeypatch, family, tool_name, toolset
+):
+    """Dynamic schemas carry logical toolset identity without global registration."""
+    from agent import agent_runtime_helpers
+
+    monkeypatch.setattr(
+        "hermes_cli.config.load_config",
+        lambda: {
+            "approvals": {
+                "tool_policies": {tool_name: "allow", toolset: "deny"}
+            }
+        },
+    )
+    from tools.mcp_tool import _reinject_post_build_tools
+
+    calls = []
+    schema = {"name": tool_name, "description": "probe", "parameters": {}}
+    memory_manager = SimpleNamespace(
+        has_tool=lambda name: family == "memory" and name == tool_name,
+        handle_tool_call=lambda name, args: calls.append((name, args)) or "ran",
+        get_all_tool_schemas=lambda: [schema] if family == "memory" else [],
+    )
+    context_engine = SimpleNamespace(
+        handle_tool_call=lambda name, args, **kwargs: calls.append((name, args)) or "ran",
+        get_tool_schemas=lambda: [schema] if family == "context" else [],
+    )
+    dynamic_entries = {}
+    agent = SimpleNamespace(
+        session_id="s",
+        _current_turn_id="t",
+        _current_api_request_id="r",
+        _memory_manager=memory_manager,
+        context_compressor=context_engine,
+        enabled_toolsets=None,
+        disabled_toolsets=None,
+    )
+    tools = []
+    names = set()
+    engine_names = _reinject_post_build_tools(
+        agent, tools, names, dynamic_entries=dynamic_entries
+    )
+    agent._context_engine_tool_names = engine_names
+    agent._dynamic_tool_entries = dynamic_entries
+    agent.valid_tool_names = names
+
+    assert dynamic_entries[tool_name].toolset == toolset
+    result = agent_runtime_helpers.invoke_tool(
+        agent, tool_name, {}, "task", tool_call_id="call", pre_tool_block_checked=True
+    )
+
+    assert calls == []
+    assert json.loads(result)["policy"] == "deny"
+
+
+@pytest.mark.parametrize(
+    "tool_name,toolset", [("read_terminal", "terminal"), ("delegate_task", "delegation")]
+)
+def test_sequential_fast_paths_apply_registered_toolset_policy(monkeypatch, tool_name, toolset):
+    import model_tools
+    from agent import tool_executor
+
+    entry = model_tools.registry.get_entry(tool_name)
+    assert entry is not None and entry.toolset == toolset
+    monkeypatch.setattr(
+        "hermes_cli.config.load_config",
+        lambda: {"approvals": {"tool_policies": {tool_name: "allow", toolset: "deny"}}},
+    )
+    called = []
+    agent = SimpleNamespace(session_id="s", _current_turn_id="t", _current_api_request_id="r")
+
+    result, _ = tool_executor._run_agent_tool_execution_middleware(
+        agent,
+        function_name=tool_name,
+        function_args={},
+        effective_task_id="task",
+        tool_call_id="call",
+        execute=lambda args: called.append(args) or "ran",
+    )
+
+    assert called == []
+    assert json.loads(result)["policy"] == "deny"
+
+
+@pytest.mark.parametrize(
+    "tool_name,toolset", [("read_terminal", "terminal"), ("delegate_task", "delegation")]
+)
+def test_concurrent_fast_paths_apply_registered_toolset_policy(monkeypatch, tool_name, toolset):
+    import model_tools
+    from agent import agent_runtime_helpers
+
+    entry = model_tools.registry.get_entry(tool_name)
+    assert entry is not None and entry.toolset == toolset
+    monkeypatch.setattr(
+        "hermes_cli.config.load_config",
+        lambda: {"approvals": {"tool_policies": {tool_name: "allow", toolset: "deny"}}},
+    )
+    called = []
+    agent = SimpleNamespace(
+        session_id="s",
+        _current_turn_id="t",
+        _current_api_request_id="r",
+        _memory_manager=None,
+        valid_tool_names=set(),
+        enabled_toolsets=None,
+        disabled_toolsets=None,
+        read_terminal_callback=lambda *_a: called.append("read"),
+        _dispatch_delegate_task=lambda args: called.append(args) or "ran",
+    )
+
+    result = agent_runtime_helpers.invoke_tool(
+        agent, tool_name, {}, "task", tool_call_id="call", pre_tool_block_checked=True
+    )
+
+    assert called == []
+    assert json.loads(result)["policy"] == "deny"
+
+
+def test_explicit_allow_does_not_bypass_plugin_escalation(monkeypatch):
+    import model_tools
+    from hermes_cli import middleware
+
+    monkeypatch.setattr(
+        middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "allow"
+    )
+    monkeypatch.setattr(
+        "hermes_cli.plugins.resolve_pre_tool_block",
+        lambda *_args, **_kwargs: "plugin requires approval",
+    )
+    dispatched = []
+    monkeypatch.setattr(
+        model_tools.registry,
+        "dispatch",
+        lambda name, args, **kwargs: dispatched.append((name, args)) or "ran",
+    )
+
+    result = model_tools.handle_function_call("read_file", {"path": "README.md"})
+
+    assert dispatched == []
+    assert json.loads(result)["error"] == "plugin requires approval"
+
+
+def test_explicit_allow_does_not_bypass_terminal_hardline(monkeypatch):
+    from hermes_cli import middleware
+    from tools.approval import check_all_command_guards
+
+    monkeypatch.setattr(
+        middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "allow"
+    )
+    called = []
+
+    result = middleware.run_tool_execution_middleware(
+        "terminal",
+        {"command": "rm -rf /"},
+        lambda args: called.append(args)
+        or check_all_command_guards(args["command"], "local"),
+    )
+
+    assert called == [{"command": "rm -rf /"}]
+    assert result["approved"] is False
+    assert "hardline" in result["message"].lower()
+
+
+def test_explicit_allow_bypasses_only_ordinary_terminal_prompt(monkeypatch):
+    from hermes_cli import middleware
+    from tools import approval
+
+    monkeypatch.setattr(
+        middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "allow"
+    )
+    prompted = []
+
+    result = middleware.run_tool_execution_middleware(
+        "terminal",
+        {"command": "rm -rf /tmp/hermes-policy-test"},
+        lambda args: approval.check_all_command_guards(
+            args["command"],
+            "local",
+            approval_callback=lambda *_args, **_kwargs: prompted.append(True) or "deny",
+        ),
+    )
+
+    assert result["approved"] is True
+    assert prompted == []
+
+
+def test_explicit_allow_does_not_bypass_terminal_user_deny(monkeypatch):
+    from hermes_cli import middleware
+    from tools import approval
+
+    monkeypatch.setattr(
+        middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "allow"
+    )
+    monkeypatch.setattr(approval, "_match_user_deny_rule", lambda _command: "git push *")
+
+    result = middleware.run_tool_execution_middleware(
+        "terminal",
+        {"command": "git push origin main"},
+        lambda args: approval.check_all_command_guards(args["command"], "local"),
+    )
+
+    assert result["approved"] is False
+    assert "user-defined deny rule" in result["message"].lower()
+
+
+def test_explicit_allow_does_not_bypass_credential_path_guard(monkeypatch, tmp_path):
+    from hermes_cli import middleware
+    from tools.file_tools import write_file_tool
+
+    monkeypatch.setattr(
+        middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "allow"
+    )
+    profile = tmp_path / ".hermes"
+    profile.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(profile))
+
+    result = middleware.run_tool_execution_middleware(
+        "write_file",
+        {"path": str(profile / ".env"), "content": "SECRET=value"},
+        lambda args: write_file_tool(**args),
+    )
+
+    assert not (profile / ".env").exists()
+    assert "write denied" in json.loads(result)["error"].lower()
+
+
+def test_ask_policy_honors_cron_deny(monkeypatch):
+    from hermes_cli import middleware
+
+    monkeypatch.setattr(
+        middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "ask"
+    )
+    monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+    monkeypatch.setattr("tools.approval._get_cron_approval_mode", lambda: "deny")
+    called = []
+
+    result = middleware.run_tool_execution_middleware(
+        "read_file", {"path": "README.md"}, lambda args: called.append(args)
+    )
+
+    assert called == []
+    assert "cron jobs run without a user present" in json.loads(result)["error"]
+
+
+def test_ask_policy_is_not_bypassed_by_session_yolo(monkeypatch):
+    from hermes_cli import middleware
+    from tools import approval
+
+    monkeypatch.setattr(
+        middleware, "resolve_tool_approval_policy", lambda _name, **_kwargs: "ask"
+    )
+    monkeypatch.setattr(approval, "is_current_session_yolo_enabled", lambda: True)
+    monkeypatch.setattr(approval, "_is_interactive_cli", lambda: True)
+    monkeypatch.setattr(approval, "prompt_dangerous_approval", lambda *_a, **_kw: "deny")
+    called = []
+
+    result = middleware.run_tool_execution_middleware(
+        "write_file", {"path": "notes.txt"}, lambda args: called.append(args)
+    )
+
+    assert called == []
+    assert "User denied" in json.loads(result)["error"]
+
+
+def test_default_config_exposes_empty_profile_local_policy_mapping():
+    from hermes_cli.config import DEFAULT_CONFIG
+
+    assert DEFAULT_CONFIG["approvals"]["tool_policies"] == {}
+
+
+def test_reloading_allowlist_replaces_previous_profiles_state(monkeypatch):
+    from tools import approval
+
+    configs = iter(
+        [
+            {"command_allowlist": ["default-only"]},
+            {"command_allowlist": ["work-only"]},
+        ]
+    )
+    monkeypatch.setattr("hermes_cli.config.load_config", lambda: next(configs))
+    monkeypatch.setattr(approval, "_permanent_approved", set())
+
+    approval.load_permanent_allowlist()
+    assert approval.is_approved("session", "default-only") is True
+
+    approval.load_permanent_allowlist()
+    assert approval.is_approved("session", "default-only") is False
+    assert approval.is_approved("session", "work-only") is True
+
+
+def test_failed_profile_allowlist_load_clears_previous_profile_state(monkeypatch):
+    from tools import approval
+
+    configs = iter([{"command_allowlist": ["default-only"]}, ValueError("malformed profile")])
+
+    def load_config():
+        value = next(configs)
+        if isinstance(value, Exception):
+            raise value
+        return value
+
+    monkeypatch.setattr("hermes_cli.config.load_config", load_config)
+    monkeypatch.setattr(approval, "_permanent_approved", set())
+    approval.load_permanent_allowlist()
+    assert approval.is_approved("session", "default-only") is True
+
+    assert approval.load_permanent_allowlist() == set()
+    assert approval.is_approved("session", "default-only") is False
diff --git a/tools/approval.py b/tools/approval.py
index 836066eda98..2bba8d160db 100644
--- a/tools/approval.py
+++ b/tools/approval.py
@@ -50,6 +50,19 @@ _approval_tool_call_id: contextvars.ContextVar[str] = contextvars.ContextVar(
     "approval_tool_call_id",
     default="",
 )
+_tool_policy_terminal_allow: contextvars.ContextVar[bool] = contextvars.ContextVar(
+    "tool_policy_terminal_allow",
+    default=False,
+)
+
+
+def set_tool_policy_terminal_allow() -> contextvars.Token[bool]:
+    """Allow ordinary terminal prompts for the current tool execution only."""
+    return _tool_policy_terminal_allow.set(True)
+
+
+def reset_tool_policy_terminal_allow(token: contextvars.Token[bool]) -> None:
+    _tool_policy_terminal_allow.reset(token)
 
 # Interactive-CLI flag. Concurrent ACP sessions run on a shared
 # ThreadPoolExecutor (acp_adapter/server.py), so mutating the process-global
@@ -1605,8 +1618,9 @@ def approve_permanent(pattern_key: str):
 
 
 def load_permanent(patterns: set):
-    """Bulk-load permanent allowlist entries from config."""
+    """Replace permanent entries with the active profile's allowlist."""
     with _lock:
+        _permanent_approved.clear()
         _permanent_approved.update(patterns)
 
 
@@ -1662,10 +1676,14 @@ def load_permanent_allowlist() -> set:
         from hermes_cli.config import load_config
         config = load_config()
         patterns = set(config.get("command_allowlist", []) or [])
-        if patterns:
-            load_permanent(patterns)
+        # Replace rather than merge: long-lived processes can reload under a
+        # different profile, whose permanent approvals must remain isolated.
+        load_permanent(patterns)
         return patterns
     except Exception as e:
+        # Fail closed across malformed profile transitions: stale grants from
+        # the previously active profile must not remain live in this process.
+        load_permanent(set())
         logger.warning("Failed to load permanent allowlist: %s", e)
         return set()
 
@@ -2031,6 +2049,7 @@ def _run_approval_gate(
     autoapprove_log_prefix: str,
     fail_closed_when_no_human: bool = False,
     no_human_block_message: str = "",
+    enforce_under_yolo: bool = False,
 ) -> dict:
     """Shared human-approval gate for a flagged action (command or tool).
 
@@ -2075,7 +2094,9 @@ def _run_approval_gate(
     # --yolo bypasses all approval prompts (session- or process-scoped).
     # Hardline blocks are handled by the caller BEFORE this gate, so yolo
     # here only skips the recoverable approval layer.
-    if _YOLO_MODE_FROZEN or is_current_session_yolo_enabled():
+    if not enforce_under_yolo and (
+        _YOLO_MODE_FROZEN or is_current_session_yolo_enabled()
+    ):
         return {"approved": True, "message": None}
 
     session_key = get_current_session_key()
@@ -2298,6 +2319,12 @@ def check_dangerous_command(command: str, env_type: str,
     if _YOLO_MODE_FROZEN or is_current_session_yolo_enabled():
         return {"approved": True, "message": None}
 
+    # A declarative terminal allow skips only the ordinary dangerous-command
+    # approval layer. Hardline and user-deny checks above still win, and cron
+    # never honors this shortcut because its unattended policy outranks allow.
+    if _tool_policy_terminal_allow.get() and not env_var_enabled("HERMES_CRON_SESSION"):
+        return {"approved": True, "message": None}
+
     if _command_matches_permanent_allowlist(command):
         return {"approved": True, "message": None}
 
@@ -2329,6 +2356,7 @@ def request_tool_approval(
     *,
     rule_key: str = "",
     approval_callback=None,
+    enforce_under_yolo: bool = False,
 ) -> dict:
     """Escalate an arbitrary tool call to the human-approval gate.
 
@@ -2407,6 +2435,7 @@ def request_tool_approval(
             "but no interactive user or gateway is present to approve it. "
             "A plugin flagged this action for human confirmation."
         ),
+        enforce_under_yolo=enforce_under_yolo,
     )
 
 
@@ -2613,6 +2642,12 @@ def check_all_command_guards(command: str, env_type: str,
     if _YOLO_MODE_FROZEN or is_current_session_yolo_enabled() or approval_mode == "off":
         return {"approved": True, "message": None}
 
+    # Per-tool allow is intentionally narrower than yolo/mode=off: it skips
+    # routine terminal prompts only after hardline/user-deny checks, and never
+    # overrides cron's unattended policy.
+    if _tool_policy_terminal_allow.get() and not env_var_enabled("HERMES_CRON_SESSION"):
+        return {"approved": True, "message": None}
+
     if _command_matches_permanent_allowlist(command):
         return {"approved": True, "message": None}
 
diff --git a/tools/mcp_tool.py b/tools/mcp_tool.py
index cd999f2712f..c90d664c320 100644
--- a/tools/mcp_tool.py
+++ b/tools/mcp_tool.py
@@ -5315,7 +5315,10 @@ def refresh_agent_mcp_tools(
     # (``build_api_kwargs``) can't see a partial rebuild or a cross-attribute
     # half-swap. ``staged_engine_names`` are the context-engine routing names
     # this rebuild actually appended (matching agent_init's dedup-aware add).
-    staged_engine_names = _reinject_post_build_tools(agent, new_defs, new_names)
+    staged_dynamic_entries: dict = {}
+    staged_engine_names = _reinject_post_build_tools(
+        agent, new_defs, new_names, dynamic_entries=staged_dynamic_entries
+    )
 
     # Single atomic read-diff-publish so the returned ``added`` is consistent
     # with what was actually published, even under concurrent callers, and a
@@ -5335,12 +5338,14 @@ def refresh_agent_mcp_tools(
             for t in (getattr(agent, "tools", None) or [])
         }
         if new_names == current:
-            # No change → leave the live snapshot untouched (no churn), but
-            # record the generation so an in-flight older caller can't clobber.
+            # No schema change, but refresh dynamic policy identities in case
+            # this agent predates metadata propagation.
+            agent._dynamic_tool_entries = staged_dynamic_entries
             agent._tool_snapshot_generation = max(published_gen, snapshot_generation)
             return set()
         agent.tools = new_defs
         agent.valid_tool_names = new_names
+        agent._dynamic_tool_entries = staged_dynamic_entries
         # Publish context-engine routing names atomically with the snapshot.
         engine_names = getattr(agent, "_context_engine_tool_names", None)
         if isinstance(engine_names, set):
@@ -5350,7 +5355,9 @@ def refresh_agent_mcp_tools(
         return new_names - current
 
 
-def _reinject_post_build_tools(agent, tools_list: list, name_set: set) -> set:
+def _reinject_post_build_tools(
+    agent, tools_list: list, name_set: set, *, dynamic_entries: Optional[dict] = None
+) -> set:
     """Append memory-provider and context-engine tools onto staged locals.
 
     Mirrors the post-``get_tool_definitions`` injection in ``agent_init`` so a
@@ -5365,12 +5372,16 @@ def _reinject_post_build_tools(agent, tools_list: list, name_set: set) -> set:
     caller publishes this into ``agent._context_engine_tool_names`` atomically
     with the snapshot.
     """
-    def _add(schema: dict) -> bool:
+    def _add(schema: dict, toolset: str) -> bool:
         name = schema.get("name", "")
         if not name or name in name_set:
             return False
         tools_list.append({"type": "function", "function": schema})
         name_set.add(name)
+        if dynamic_entries is not None:
+            from tools.registry import DynamicToolEntry
+
+            dynamic_entries[name] = DynamicToolEntry(name, toolset)
         return True
 
     # Memory-provider tools (mem0/honcho/byterover/supermemory/…).
@@ -5383,7 +5394,7 @@ def _reinject_post_build_tools(agent, tools_list: list, name_set: set) -> set:
             if "memory" in name_set or memory_provider_tools_enabled(getattr(agent, "enabled_toolsets", None)):
                 for schema in get_mem_schemas():
                     if isinstance(schema, dict):
-                        _add(schema)
+                        _add(schema, "memory")
     except Exception:
         logger.debug("Memory-provider tool re-injection skipped", exc_info=True)
 
@@ -5407,7 +5418,7 @@ def _reinject_post_build_tools(agent, tools_list: list, name_set: set) -> set:
                 # Only claim the routing name when WE appended the schema, so a
                 # name already owned by a registry/plugin tool keeps its own
                 # dispatch (matches agent_init.py's `continue`-before-claim).
-                if _add(schema) and name:
+                if _add(schema, "context_engine") and name:
                     staged_engine_names.add(name)
     except Exception:
         logger.debug("Context-engine tool re-injection skipped", exc_info=True)
diff --git a/tools/registry.py b/tools/registry.py
index 9b6611fb407..f7d1ea85c0e 100644
--- a/tools/registry.py
+++ b/tools/registry.py
@@ -21,6 +21,7 @@ import logging
 import sys
 import threading
 import time
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Callable, Dict, List, Optional, Set
 
@@ -107,6 +108,14 @@ class ToolEntry:
         self.dynamic_schema_overrides = dynamic_schema_overrides
 
 
+@dataclass(frozen=True)
+class DynamicToolEntry:
+    """Policy identity for a tool advertised outside the global registry."""
+
+    name: str
+    toolset: str
+
+
 # ---------------------------------------------------------------------------
 # check_fn TTL cache
 #
@@ -614,13 +623,31 @@ class ToolRegistry:
         entry = self.get_entry(name)
         if not entry:
             return json.dumps({"error": f"Unknown tool: {name}"})
-        try:
+
+        def execute(next_args: dict) -> str | dict:
             if entry.is_async:
                 from model_tools import _run_async
-                result = _run_async(entry.handler(args, **kwargs))
+                result = _run_async(entry.handler(next_args, **kwargs))
             else:
-                result = entry.handler(args, **kwargs)
+                result = entry.handler(next_args, **kwargs)
             return self._normalize_handler_result(name, result)
+
+        try:
+            # This is the narrowest shared execution boundary: model calls,
+            # PluginContext, MCP, and direct framework callers converge here.
+            # The actual entry is passed through so custom registries resolve
+            # their own toolset. Only an explicitly marked, immediate wrapper
+            # handoff is deduplicated; recursive dispatch remains a new call.
+            from hermes_cli.middleware import run_tool_execution_middleware
+
+            return run_tool_execution_middleware(
+                name,
+                args,
+                execute,
+                registry_entry=entry,
+                registry_dispatch=self,
+                **kwargs,
+            )
         except Exception as e:
             logger.exception("Tool %s dispatch error: %s", name, e)
             # Route through the sanitizer so framing tokens / CDATA / fences
diff --git a/website/docs/user-guide/security.md b/website/docs/user-guide/security.md
index 71d1b131d2e..98eafc9933b 100644
--- a/website/docs/user-guide/security.md
+++ b/website/docs/user-guide/security.md
@@ -33,6 +33,10 @@ approvals:
   mode: smart                     # smart | manual | off
   timeout: 60                     # seconds to wait for user response (default: 60)
   cron_mode: deny                 # deny | approve — what cron jobs do when they hit a dangerous command
+  tool_policies:                  # optional per-tool / per-toolset glob rules
+    "*": ask                     # supervise every tool by default
+    read_file: allow              # still asks because the broader rule also matches
+    terminal: deny                # block this tool without prompting
   mcp_reload_confirm: true        # /reload-mcp asks before invalidating the MCP tool cache
   destructive_slash_confirm: true # /clear, /new, /reset, /undo prompt before discarding state
 ```
@@ -44,6 +48,7 @@ The full set of keys:
 | `mode` | `smart` | Approval policy for dangerous shell commands — see the table below. |
 | `timeout` | `60` | Seconds Hermes waits for an approval reply before timing out. |
 | `cron_mode` | `deny` | How [cron jobs](./features/cron.md) behave headlessly when they trigger a dangerous-command prompt. `deny` blocks the command (the agent must find another path); `approve` auto-approves everything in cron context. |
+| `tool_policies` | `{}` | Profile-local, case-insensitive glob mapping for tools and registered toolsets. Values are `allow`, `ask`, or `deny`. |
 | `mcp_reload_confirm` | `true` | When true, `/reload-mcp` asks before rebuilding the MCP tool set. Rebuilding invalidates the provider prompt cache (tool schemas live in the system prompt), so the next message re-sends full input tokens. Users who click **Always Approve** flip this key to `false`. |
 | `destructive_slash_confirm` | `true` | When true, destructive session slash commands (`/clear`, `/new`, `/reset`, `/undo`) prompt before discarding conversation state. Three-option dialog (Approve Once / Always Approve / Cancel) routed through native yes/no buttons on Telegram, Discord, and Slack; text fallback elsewhere. Users who click **Always Approve** flip this key to `false`. TUI uses its own modal overlay (set `HERMES_TUI_NO_CONFIRM=1` to opt out there). |
 
@@ -53,6 +58,28 @@ The full set of keys:
 | **manual** | Always prompt the user for approval on dangerous commands. |
 | **off** | Disable all approval checks — equivalent to running with `--yolo`. All commands execute without prompts. |
 
+### Per-tool policies
+
+`approvals.tool_policies` applies at the shared execution boundary used by the
+CLI, gateway platforms, MCP tools, plugins, and both sequential and concurrent
+agent dispatch. It does not alter the advertised tool schemas or system prompt,
+so changing policy does not invalidate the conversation prompt cache.
+
+- `deny` returns a blocked tool result without executing the tool.
+- `ask` uses the normal CLI or gateway approval UI and fails closed when no
+  human approval surface is available. Explicit `ask` rules remain active in
+  YOLO / `approvals.mode: off`; cron still follows `approvals.cron_mode`.
+- `allow` skips this policy prompt. For `terminal`, it also skips the ordinary
+  dangerous-command prompt, but never bypasses hardline blocks, user-defined
+  `approvals.deny` rules, cron policy, plugin escalation, tool scope, or
+  sensitive-path protections.
+
+Patterns are case-insensitive Python `fnmatch` globs matched against both the
+tool name and its registered toolset. Across every overlapping match, the most
+restrictive result always wins (`deny` > `ask` > `allow`), regardless of pattern
+specificity or whether the match came from the tool name or toolset. Always
+quote wildcard keys in YAML.
+
 :::warning
 Setting `approvals.mode: off` disables all safety prompts. Use only in trusted environments (CI/CD, containers, etc.).
 :::