feat(acp): honor --toolsets and add per-session tool scoping

acp_adapter/entry.py

@@ -38,7 +38,12 @@
 class _BenignProbeMethodFilter(logging.Filter):
     """Suppress acp 'Background task failed' tracebacks caused by unknown liveness-probe methods
     (e.g. ``ping``); every other background-task error, incl. method_not_found for non-probe
-    methods, stays visible."""
+    methods, stays visible.
+
+    An ``invalid_params`` rejection is also not a crash — it is the adapter answering a malformed
+    request exactly as designed (e.g. a bad ``_meta.hermes.toolsets``). The traceback made a normal
+    protocol rejection look like a failure in the log, so it is collapsed to one WARNING line that
+    still names the reason."""
 
     def filter(self, record: logging.LogRecord) -> bool:
         if record.getMessage() != "Background task failed" or not record.exc_info:
@@ -49,9 +54,17 @@ def filter(self, record: logging.LogRecord) -> bool:
         except ImportError:
             return True
         exc = record.exc_info[1]
-        if not isinstance(exc, RequestError) or getattr(exc, "code", None) != -32601:
+        if not isinstance(exc, RequestError):
+            return True
+        code, data = getattr(exc, "code", None), getattr(exc, "data", None)
+        if code == -32602:
+            details = data.get("details") if isinstance(data, dict) else None
+            record.msg = "Rejected request: %s (%s)" % (details or exc, code)
+            record.args, record.exc_info, record.exc_text = (), None, None
+            record.levelno, record.levelname = logging.WARNING, "WARNING"
+            return True
+        if code != -32601:
             return True
-        data = getattr(exc, "data", None)
         return not (isinstance(data, dict) and data.get("method") in _BENIGN_PROBE_METHODS)
 
 
@@ -96,6 +109,10 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     parser.add_argument("--yes", "-y", action="store_true", dest="assume_yes",
                         help="Accept all prompts (currently used by --setup-browser to skip the "
                              "~400 MB Chromium download confirmation).")
+    parser.add_argument("-t", "--toolsets", default=None,
+                        help="Comma-separated toolsets enabled for every session this process serves "
+                             "(default: the ACP platform toolsets). A client can scope a single session "
+                             "further via the session/new `_meta.hermes.toolsets` extension.")
     return parser.parse_args(argv)
 
 
@@ -207,13 +224,18 @@ def main(argv: list[str] | None = None) -> None:
     # model_tools.py module scope to avoid freezing the gateway's loop on lazy import (#16856).
     if os.environ.get("HERMES_ACP_SKIP_CONFIGURED_MCP", "").strip() != "1":
         try:
-            from hermes_cli.mcp_startup import start_background_mcp_discovery
+            from hermes_cli.mcp_startup import set_mcp_server_filter, start_background_mcp_discovery
 
+            # ``hermes acp`` arrives here through hermes_cli.main, which has already narrowed the
+            # spawn set by ``-t``; ``hermes-acp`` / ``python -m acp_adapter.entry`` does not go
+            # through it, so apply the same filter here — otherwise one flag means two different
+            # things depending on which launcher the editor was pointed at.
+            set_mcp_server_filter(args.toolsets)
             start_background_mcp_discovery(logger=logger, thread_name="acp-mcp-discovery")
         except Exception:
             logger.debug("MCP tool discovery failed at ACP startup", exc_info=True)
 
-    agent = HermesACPAgent()
+    agent = HermesACPAgent(default_toolsets=args.toolsets)
     try:
         asyncio.run(acp.run_agent(agent, use_unstable_protocol=True))
     except KeyboardInterrupt:

acp_adapter/server.py

@@ -34,7 +34,9 @@
 from acp_adapter.model_catalog import build_model_state, encode_model_choice
 from acp_adapter.permissions import make_approval_callback
 from acp_adapter.provenance import session_provenance_meta
-from acp_adapter.session import SessionManager, SessionState, _expand_acp_enabled_toolsets
+from acp_adapter.session import (
+    SessionManager, SessionState, _expand_acp_enabled_toolsets, _normalize_acp_toolsets, _unknown_acp_toolsets,
+)
 from acp_adapter.tools import build_tool_complete, build_tool_start, coerce_tool_args
 from agent.context_compressor import (COMPRESSED_SUMMARY_METADATA_KEY, ContextCompressor)
 from agent.interrupt_compat import request_hard_interrupt
@@ -251,9 +253,11 @@ class HermesACPAgent(SlashCommandsMixin, acp.Agent):
     _MODE_TO_EDIT_APPROVAL_POLICY = {mode: spec[0] for mode, spec in _MODES.items()}
     _EDIT_APPROVAL_POLICY_TO_MODE = {spec[0]: mode for mode, spec in _MODES.items()}
 
-    def __init__(self, session_manager: SessionManager | None = None):
+    def __init__(self, session_manager: SessionManager | None = None, default_toolsets: Any = None):
+        """``default_toolsets``: process-wide selection from ``hermes acp --toolsets``, applied to
+        every session that does not scope itself (ignored when a ``session_manager`` is injected)."""
         super().__init__()
-        self.session_manager = session_manager or SessionManager()
+        self.session_manager = session_manager or SessionManager(default_toolsets=default_toolsets)
         self._conn: Optional[acp.Client] = None
 
     # ---- Connection lifecycle -----------------------------------------------
@@ -606,17 +610,49 @@ async def _attach_session_mcp(self, state: SessionState, mcp_servers: list | Non
         self._schedule_mcp_late_refresh(state)
         logger.info(log, *log_args)
 
+    @staticmethod
+    def _requested_toolsets(kwargs: dict[str, Any]) -> list[str] | None:
+        """Per-session tool scope from ``_meta.hermes.toolsets`` — a JSON array of toolset names
+        or a comma-separated string, under the ``_meta.hermes`` namespace this adapter already
+        uses for its own ``_meta`` (see ``provenance.py``). The acp library splats ``_meta`` into
+        handler kwargs, so ``kwargs["hermes"]`` is the spec-conformant read; a top-level
+        ``toolsets`` field is dropped by the request model.
+
+        Every other ``_meta`` key is ignored, a bare ``_meta.toolsets`` included: ``_meta`` is a
+        shared extension map and another client may well use that name for its own purposes.
+
+        Absent (or JSON ``null``) means "no selection" and the process default applies. Present
+        but unusable — ``[]``, ``7``, ``{}``, an unknown name — is an ``invalid_params`` error: a
+        client that meant to narrow must never be silently handed the full default toolset.
+
+        Blocking (the registry import, and plugin discovery for an unknown name), so callers run
+        it off the event loop."""
+        hermes_meta = kwargs.get("hermes")
+        value = hermes_meta.get("toolsets") if isinstance(hermes_meta, dict) else None
+        if value is None:
+            return None
+        from acp.exceptions import RequestError
+        names = _normalize_acp_toolsets(value)
+        if not names:
+            raise RequestError.invalid_params(
+                {"details": "_meta.hermes.toolsets must be a non-empty array of toolset names"})
+        if unknown := _unknown_acp_toolsets(names):
+            raise RequestError.invalid_params({"details": f"Unknown toolset(s): {', '.join(unknown)}"})
+        return names
+
     async def new_session(self, cwd: str, mcp_servers: list | None = None, **kwargs: Any) -> NewSessionResponse:
+        toolsets = await asyncio.to_thread(self._requested_toolsets, kwargs)
         # Agent construction (config, memory-provider import, SessionDB) is slow and fully
         # blocking; inline it froze the loop serving every JSON-RPC request (#58083).
-        state = await asyncio.to_thread(self.session_manager.create_session, cwd=cwd)
+        state = await asyncio.to_thread(self.session_manager.create_session, cwd=cwd, toolsets=toolsets)
         await self._attach_session_mcp(state, mcp_servers, "New session %s (cwd=%s)", state.session_id, cwd)
         return NewSessionResponse(session_id=state.session_id, **await self._session_response_fields(state))
 
     async def load_session(
         self, cwd: str, session_id: str, mcp_servers: list | None = None, **kwargs: Any
     ) -> LoadSessionResponse | None:
-        state = await asyncio.to_thread(self.session_manager.update_cwd, session_id, cwd)
+        toolsets = await asyncio.to_thread(self._requested_toolsets, kwargs)
+        state = await asyncio.to_thread(self.session_manager.update_cwd, session_id, cwd, toolsets)
         if state is None:
             logger.warning("load_session: session %s not found", session_id)
             return None
@@ -626,10 +662,11 @@ async def load_session(
     async def resume_session(
         self, cwd: str, session_id: str, mcp_servers: list | None = None, **kwargs: Any
     ) -> ResumeSessionResponse:
-        state = await asyncio.to_thread(self.session_manager.update_cwd, session_id, cwd)
+        toolsets = await asyncio.to_thread(self._requested_toolsets, kwargs)
+        state = await asyncio.to_thread(self.session_manager.update_cwd, session_id, cwd, toolsets)
         if state is None:
             logger.warning("resume_session: session %s not found, creating new", session_id)
-            state = await asyncio.to_thread(self.session_manager.create_session, cwd=cwd)
+            state = await asyncio.to_thread(self.session_manager.create_session, cwd=cwd, toolsets=toolsets)
         await self._attach_session_mcp(state, mcp_servers, "Resumed session %s", state.session_id)
         return ResumeSessionResponse(**await self._session_response_fields(state, "resume"))