fix(security): enforce disabled toolsets after dynamic injection (#49386)

acp_adapter/server.py

@@ -2092,7 +2092,12 @@ def _cmd_tools(self, args: str, state: SessionState) -> str:
             toolsets = _expand_acp_enabled_toolsets(
                 getattr(state.agent, "enabled_toolsets", None) or ["hermes-acp"]
             )
-            tools = get_tool_definitions(enabled_toolsets=toolsets, quiet_mode=True)
+            disabled_toolsets = getattr(state.agent, "disabled_toolsets", None)
+            tools = get_tool_definitions(
+                enabled_toolsets=toolsets,
+                disabled_toolsets=disabled_toolsets,
+                quiet_mode=True,
+            )
             tool_view = SimpleNamespace(
                 tools=list(tools or []),
                 valid_tool_names={
@@ -2101,6 +2106,7 @@ def _cmd_tools(self, args: str, state: SessionState) -> str:
                     if isinstance(tool, dict)
                 },
                 enabled_toolsets=toolsets,
+                disabled_toolsets=disabled_toolsets,
                 _memory_manager=getattr(state.agent, "_memory_manager", None),
             )
             inject_memory_provider_tools(tool_view)

acp_adapter/session.py

@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 from hermes_constants import get_hermes_home
+from hermes_cli.toolset_validation import normalize_toolset_names
 
 import copy
 import json
@@ -620,12 +621,20 @@ def _make_agent(
             if not isinstance(cfg, dict) or cfg.get("enabled", True) is not False
         ]
 
+        agent_cfg = config.get("agent")
+        disabled_toolsets = normalize_toolset_names(
+            agent_cfg.get("disabled_toolsets")
+            if isinstance(agent_cfg, dict)
+            else None
+        )
+
         kwargs = {
             "platform": "acp",
             "enabled_toolsets": _expand_acp_enabled_toolsets(
                 ["hermes-acp"],
                 mcp_server_names=configured_mcp_servers,
             ),
+            "disabled_toolsets": disabled_toolsets,
             "quiet_mode": True,
             "session_id": session_id,
             "session_db": self._get_db(),

agent/agent_init.py

@@ -870,6 +870,10 @@ def init_agent(
     # late/concurrent refresh reject a stale (older-generation) rebuild instead
     # of clobbering a newer one. Set adjacent to the tool snapshot below.
     agent._tool_snapshot_generation = 0
+    # Policy publication has its own ordering because registry generation does
+    # not change when an explicit reload alters only enabled/disabled toolsets.
+    agent._tool_policy_epoch = 0
+    agent._tool_published_policy_epoch = 0
     # Rate limit tracking — updated from x-ratelimit-* response headers
     # after each API call.  Accessed by /usage slash command.
     agent._rate_limit_state: Optional["RateLimitState"] = None

agent/agent_runtime_helpers.py

@@ -63,14 +63,21 @@ def _ra():
 )
 
 
+def memory_provider_owns_tool(agent: Any, function_name: str) -> bool:
+    """Return whether the published agent surface assigns provider routing."""
+    owned_names = getattr(agent, "_memory_provider_tool_names", None)
+    return isinstance(owned_names, set) and function_name in owned_names
+
+
 def agent_runtime_owns_post_tool_hook(agent: Any, function_name: str) -> bool:
     """Return True when an agent-level tool path emits its own post hook."""
     if function_name in AGENT_RUNTIME_POST_HOOK_TOOL_NAMES:
         return True
     if getattr(agent, "_context_engine_tool_names", None) and function_name in agent._context_engine_tool_names:
         return True
-    memory_manager = getattr(agent, "_memory_manager", None)
-    return bool(memory_manager and memory_manager.has_tool(function_name))
+    return bool(getattr(agent, "_memory_manager", None)) and memory_provider_owns_tool(
+        agent, function_name
+    )
 
 
 def convert_to_trajectory_format(agent, messages: List[Dict[str, Any]], user_query: str, completed: bool) -> List[Dict[str, Any]]:
@@ -2604,7 +2611,7 @@ def _execute(next_args: dict) -> Any:
                     ),
                 )
             return _finish_agent_tool(result, next_args)
-    elif agent._memory_manager and agent._memory_manager.has_tool(function_name):
+    elif agent._memory_manager and memory_provider_owns_tool(agent, function_name):
         def _execute(next_args: dict) -> Any:
             return _finish_agent_tool(agent._memory_manager.handle_tool_call(function_name, next_args), next_args)
     elif function_name == "clarify":

agent/memory_manager.py

@@ -107,25 +107,91 @@ def memory_provider_tools_enabled(
         return False
 
 
+def memory_provider_denied_tool_names(
+    disabled_toolsets: Optional[List[str]],
+) -> Optional[set[str]]:
+    """Resolve provider-tool subtraction.
+
+    Returns the exact denied tool names, or ``None`` when all provider tools
+    must be denied. Unknown toolsets are ignored, matching the registry tool
+    filtering path. Resolver failures fail closed.
+    """
+    if not disabled_toolsets:
+        return set()
+    if any(name in {"memory", "all", "*"} for name in disabled_toolsets):
+        return None
+
+    try:
+        from toolsets import bundle_non_core_tools, resolve_toolset, validate_toolset
+
+        denied: set[str] = set()
+        for name in disabled_toolsets:
+            if not validate_toolset(name):
+                continue
+            resolved = set(
+                bundle_non_core_tools(name)
+                if name.startswith("hermes-")
+                else resolve_toolset(name)
+            )
+            if "memory" in resolved:
+                return None
+            denied.update(resolved)
+    except Exception:
+        logger.debug("Failed to resolve disabled toolsets for memory-provider tools", exc_info=True)
+        return None
+    return denied
+
+
+def memory_provider_tools_disabled(disabled_toolsets: Optional[List[str]]) -> bool:
+    """Return whether subtraction denies the complete provider tool family."""
+    return memory_provider_denied_tool_names(disabled_toolsets) is None
+
+
+def effective_memory_provider_tool_schemas(
+    raw_schemas,
+    *,
+    enabled_toolsets: Optional[List[str]],
+    disabled_toolsets: Optional[List[str]],
+    memory_selected: bool = False,
+) -> List[Dict[str, Any]]:
+    """Return normalized provider schemas surviving selection and subtraction."""
+    if not memory_selected and not memory_provider_tools_enabled(enabled_toolsets):
+        return []
+
+    denied_names = memory_provider_denied_tool_names(disabled_toolsets)
+    if denied_names is None:
+        return []
+
+    effective = []
+    for raw_schema in raw_schemas:
+        schema = normalize_tool_schema(raw_schema)
+        if schema is None:
+            logger.warning(
+                "Memory provider returned a tool schema with no resolvable "
+                "name; skipping to avoid poisoning the request (%r)",
+                raw_schema,
+            )
+            continue
+        if schema["name"] not in denied_names:
+            effective.append(schema)
+    return effective
+
+
 def inject_memory_provider_tools(agent: Any) -> int:
     """Append external memory-provider tool schemas to an agent tool surface."""
+    # This is routing authorization, not merely diagnostic metadata.  Only
+    # schemas this injector actually appends are owned by the provider; a
+    # registry/MCP/plugin collision keeps generic dispatch precedence.
+    agent._memory_provider_tool_names = set()
     memory_manager = getattr(agent, "_memory_manager", None)
     tools = getattr(agent, "tools", None)
     if not memory_manager or tools is None:
         return 0
-
     existing_tool_names = {
         tool.get("function", {}).get("name")
         for tool in tools
         if isinstance(tool, dict)
     }
-    if not memory_provider_tools_enabled(
-        getattr(agent, "enabled_toolsets", None),
-        getattr(agent, "disabled_toolsets", None),
-        memory_tool_present="memory" in existing_tool_names,
-    ):
-        return 0
-
     get_schemas = getattr(memory_manager, "get_all_tool_schemas", None)
     if not callable(get_schemas):
         return 0
@@ -136,21 +202,20 @@ def inject_memory_provider_tools(agent: Any) -> int:
         agent.valid_tool_names = valid_tool_names
 
     added = 0
-    for raw_schema in get_schemas():
-        schema = normalize_tool_schema(raw_schema)
-        if schema is None:
-            logger.warning(
-                "Memory provider returned a tool schema with no resolvable "
-                "name; skipping to avoid poisoning the request (%r)",
-                raw_schema,
-            )
-            continue
+    schemas = effective_memory_provider_tool_schemas(
+        get_schemas(),
+        enabled_toolsets=getattr(agent, "enabled_toolsets", None),
+        disabled_toolsets=getattr(agent, "disabled_toolsets", None),
+        memory_selected="memory" in existing_tool_names,
+    )
+    for schema in schemas:
         tool_name = schema["name"]
         if tool_name in existing_tool_names:
             continue
         tools.append({"type": "function", "function": schema})
         valid_tool_names.add(tool_name)
         existing_tool_names.add(tool_name)
+        agent._memory_provider_tool_names.add(tool_name)
         added += 1
 
     return added
@@ -483,15 +548,33 @@ def get_provider(self, name: str) -> Optional[MemoryProvider]:
 
     # -- System prompt -------------------------------------------------------
 
-    def build_system_prompt(self) -> str:
+    def build_system_prompt(
+        self, *, available_tool_names: Optional[set[str]] = None
+    ) -> str:
         """Collect system prompt blocks from all providers.
 
         Returns combined text, or empty string if no providers contribute.
         Each non-empty block is labeled with the provider name.
+
+        When ``available_tool_names`` is provided, providers whose schemas are
+        all absent are omitted so their tool-use instructions cannot advertise
+        unavailable capabilities. Providers without schemas remain visible
+        because their prompt describes passive context injection.
         """
         blocks = []
         for provider in self._providers:
             try:
+                provider_schemas = provider.get_tool_schemas()
+                if available_tool_names is not None and provider_schemas:
+                    schema_names = {
+                        schema["name"]
+                        for raw_schema in provider_schemas
+                        if (schema := normalize_tool_schema(raw_schema)) is not None
+                    }
+                    # Provider blocks are opaque: retaining one when even one
+                    # callable is unavailable can advertise the denied tool.
+                    if not schema_names.issubset(available_tool_names):
+                        continue
                 block = provider.system_prompt_block()
                 if block and block.strip():
                     blocks.append(block)
@@ -781,7 +864,7 @@ def flush_pending(self, timeout: Optional[float] = None) -> bool:
 
     # -- Tools ---------------------------------------------------------------
 
-    def get_all_tool_schemas(self) -> List[Dict[str, Any]]:
+    def _collect_all_tool_schemas(self, *, strict: bool) -> List[Dict[str, Any]]:
         """Collect tool schemas from all providers.
 
         Reserved core tool names (``clarify``, ``delegate_task``, etc.) are
@@ -812,12 +895,22 @@ def get_all_tool_schemas(self) -> List[Dict[str, Any]]:
                         schemas.append(schema)
                         seen.add(name)
             except Exception as e:
+                if strict:
+                    raise
                 logger.warning(
                     "Memory provider '%s' get_tool_schemas() failed: %s",
                     provider.name, e,
                 )
         return schemas
 
+    def get_all_tool_schemas(self) -> List[Dict[str, Any]]:
+        """Collect provider schemas fail-soft for optional runtime callers."""
+        return self._collect_all_tool_schemas(strict=False)
+
+    def get_all_tool_schemas_strict(self) -> List[Dict[str, Any]]:
+        """Collect every provider schema or fail without a partial result."""
+        return self._collect_all_tool_schemas(strict=True)
+
     def get_all_tool_names(self) -> set:
         """Return set of all tool names across all providers."""
         return set(self._tool_to_provider.keys())

agent/system_prompt.py

@@ -511,7 +511,30 @@ def build_system_prompt_parts(agent: Any, system_message: Optional[str] = None)
     # External memory provider system prompt block (additive to built-in)
     if agent._memory_manager:
         try:
-            _ext_mem_block = agent._memory_manager.build_system_prompt()
+            from agent.memory_manager import (
+                effective_memory_provider_tool_schemas,
+            )
+
+            _provider_schemas = agent._memory_manager.get_all_tool_schemas()
+            _effective_provider_schemas = effective_memory_provider_tool_schemas(
+                _provider_schemas,
+                enabled_toolsets=getattr(agent, "enabled_toolsets", None),
+                disabled_toolsets=getattr(agent, "disabled_toolsets", None),
+                memory_selected="memory" in getattr(agent, "valid_tool_names", set()),
+            )
+            _effective_provider_names = {
+                schema["name"] for schema in _effective_provider_schemas
+            }
+            _published_provider_names = getattr(
+                agent, "_memory_provider_tool_names", None
+            )
+            if isinstance(_published_provider_names, set):
+                _effective_provider_names.intersection_update(
+                    _published_provider_names
+                )
+            _ext_mem_block = agent._memory_manager.build_system_prompt(
+                available_tool_names=_effective_provider_names
+            )
             if _ext_mem_block:
                 volatile_parts.append(_ext_mem_block)
         except Exception:

agent/tool_executor.py

@@ -32,6 +32,7 @@
     _detect_tool_failure,
 )
 from agent.tool_guardrails import ToolGuardrailDecision
+from agent.agent_runtime_helpers import memory_provider_owns_tool
 from agent.tool_dispatch_helpers import (
     _is_destructive_command,
     _is_multimodal_tool_result,
@@ -1463,7 +1464,7 @@ def _execute(next_args: dict) -> Any:
                     spinner.stop(cute_msg)
                 elif agent._should_emit_quiet_tool_messages():
                     agent._vprint(f"  {cute_msg}")
-        elif agent._memory_manager and agent._memory_manager.has_tool(function_name):
+        elif agent._memory_manager and memory_provider_owns_tool(agent, function_name):
             # Memory provider tools (hindsight_retain, honcho_search, etc.)
             # These are not in the tool registry — route through MemoryManager.
             spinner = None

cli.py

@@ -53,6 +53,7 @@
 import yaml
 
 from hermes_cli.fallback_config import get_fallback_chain
+from hermes_cli.toolset_validation import normalize_toolset_names
 from hermes_cli.cli_agent_setup_mixin import CLIAgentSetupMixin
 from hermes_cli.cli_commands_mixin import CLICommandsMixin
 from hermes_cli.cli_billing_mixin import CLIBillingMixin
@@ -4118,7 +4119,9 @@ def __init__(
 
         # Parse and validate toolsets
         self.enabled_toolsets = toolsets
-        self.disabled_toolsets = CLI_CONFIG["agent"].get("disabled_toolsets") or []
+        self.disabled_toolsets = normalize_toolset_names(
+            CLI_CONFIG["agent"].get("disabled_toolsets")
+        ) or []
 
         if toolsets and "all" not in toolsets and "*" not in toolsets:
             # Validate each toolset — MCP server names are resolved via

cron/scheduler.py

@@ -43,6 +43,7 @@
 from hermes_cli._subprocess_compat import windows_hide_flags
 from hermes_cli.config import load_config, _expand_env_vars
 from hermes_cli.fallback_config import get_fallback_chain
+from hermes_cli.toolset_validation import normalize_toolset_names
 from hermes_time import now as _hermes_now
 
 logger = logging.getLogger(__name__)
@@ -168,7 +169,9 @@ def _resolve_cron_disabled_toolsets(cfg: dict) -> list[str]:
     """
     disabled = ["cronjob", "messaging", "clarify"]
     agent_cfg = (cfg or {}).get("agent") or {}
-    user_disabled = agent_cfg.get("disabled_toolsets") or []
+    user_disabled = normalize_toolset_names(
+        agent_cfg.get("disabled_toolsets")
+    ) or []
     for name in user_disabled:
         name = str(name).strip()
         if name and name not in disabled: