feat(executor): narrow tool set to declared capabilities (#1425) (#1433)

* feat(executor): narrow subagent tools to declared filesystem set

* test(executor): pin daemon toolset declaration parity

* test(executor): pin prompt declaration to registered toolset

* refactor(executor): share declared toolset between prompt and registry

* test(executor): keep isolated prompt extraction compatible

* fix(executor): keep bridge prompt independent of manager test doubles

Author: ozand

nanobot/agent/loop.py

@@ -96,6 +96,7 @@ def __init__(
             model=self.model,
             web_search_config=self.web_search_config,
             web_proxy=web_proxy,
+            web_tools_enabled=True,
             exec_config=self.exec_config,
             restrict_to_workspace=restrict_to_workspace,
             max_iterations=self.max_iterations,

nanobot/agent/subagent.py

@@ -15,7 +15,7 @@
 from nanobot.agent.tools.filesystem import EditFileTool, ListDirTool, ReadFileTool, WriteFileTool
 from nanobot.agent.tools.registry import ToolRegistry
 from nanobot.agent.tools.shell import ExecTool
-from nanobot.agent.tools.web import WebFetchTool, WebSearchTool
+from nanobot.agent.tools.toolsets import EXECUTOR_TOOL_NAMES as EXECUTOR_TOOLSET
 from nanobot.bus.events import InboundMessage
 from nanobot.bus.queue import MessageBus
 from nanobot.config.schema import ExecToolConfig
@@ -126,6 +126,7 @@ def __init__(
         # Optional: names to exclude from the loop skills summary (Part E).
         excluded_skill_names: "list[str] | None" = None,
         telemetry_component: str = "",
+        web_tools_enabled: bool = False,
     ):
         from nanobot.config.schema import ExecToolConfig, WebSearchConfig
 
@@ -163,6 +164,7 @@ def __init__(
         # #939 Part E: excluded skill names for the loop summary
         self._excluded_skill_names: list[str] = list(excluded_skill_names or [])
         self._telemetry_component = str(telemetry_component or "").strip()
+        self.web_tools_enabled = bool(web_tools_enabled)
 
     async def spawn(
         self,
@@ -271,9 +273,15 @@ def _on_skill_read(skill_path: Path) -> None:  # noqa: E301
                 restrict_to_workspace=self.restrict_to_workspace,
                 path_append=self.exec_config.path_append,
             ))
-            tools.register(WebSearchTool(config=self.web_search_config, proxy=self.web_proxy))
-            tools.register(WebFetchTool(proxy=self.web_proxy))
+            if self.web_tools_enabled:
+                from nanobot.agent.tools.web import WebFetchTool, WebSearchTool
 
+                tools.register(WebSearchTool(config=self.web_search_config, proxy=self.web_proxy))
+                tools.register(WebFetchTool(proxy=self.web_proxy))
+
+            assert tuple(tools.tool_names) == self.registered_tool_names(), (
+                "registered tool set must match the configured declaration"
+            )
             system_prompt = self._build_subagent_prompt()
             messages: list[dict[str, Any]] = [
                 {"role": "system", "content": system_prompt},
@@ -660,6 +668,19 @@ def _write_subagent_telemetry(self, task_id: str, payload: dict[str, Any]) -> No
         # telemetry with `glob("*.json")` over this directory and started
         # picking up the sidecar instead. Never add a second `.json` here.
 
+    EXECUTOR_TOOL_NAMES = EXECUTOR_TOOLSET
+
+    @classmethod
+    def declared_tool_names(cls) -> tuple[str, ...]:
+        """Names declared in the loop executor prompt and registered by it."""
+        return cls.EXECUTOR_TOOL_NAMES
+
+    def registered_tool_names(self) -> tuple[str, ...]:
+        """Names registered for this manager's configured execution role."""
+        if self.web_tools_enabled:
+            return self.EXECUTOR_TOOL_NAMES + ("web_search", "web_fetch")
+        return self.EXECUTOR_TOOL_NAMES
+
     def _build_subagent_prompt(self) -> str:
         """Build the system prompt for the subagent.
 

nanobot/agent/tools/toolsets.py

@@ -0,0 +1,7 @@
+"""Shared toolset declarations for prompt and registration parity."""
+
+from __future__ import annotations
+
+EXECUTOR_TOOL_NAMES = ("read_file", "write_file", "edit_file", "list_dir", "exec")
+INTERACTIVE_WEB_TOOL_NAMES = ("web_search", "web_fetch")
+INTERACTIVE_TOOL_NAMES = EXECUTOR_TOOL_NAMES + INTERACTIVE_WEB_TOOL_NAMES

nanobot/runtime/bridge.py

@@ -50,6 +50,7 @@
 install_promoted_overlay()
 
 from nanobot.runtime import llm_proposer, demand  # noqa: E402
+from nanobot.agent.tools.toolsets import EXECUTOR_TOOL_NAMES  # noqa: E402
 from nanobot.runtime.cycle_ledger import (  # noqa: E402
     VALID_OUTCOMES,
     append_event,
@@ -1606,7 +1607,8 @@ def build_task(req: dict, goal_text: str, report_source: str,
                repair_context: 'str | None' = None,
                selfevo_repo_root: 'Path | None' = None,
                max_iterations: int = 15,
-               charter_in_system: bool = False) -> str:
+               charter_in_system: bool = False,
+               declared_tool_names: tuple[str, ...] | None = None) -> str:
     """Build a concrete task prompt for the subagent from the request payload.
 
     Args:
@@ -1832,6 +1834,7 @@ def build_task(req: dict, goal_text: str, report_source: str,
     _verification_note = (
         '' if _pytest_available else '     (pytest is not installed — use python3 -c imports as smoke tests)'
     )
+    declared_tool_names = declared_tool_names or ("read_file", "write_file", "edit_file", "list_dir", "exec")
     lines += [
         '## Your instructions',
         'You MUST take a concrete action in this session. Do not return a review only.',
@@ -1857,7 +1860,7 @@ def build_task(req: dict, goal_text: str, report_source: str,
         '  "findings": ["<observation1>", "<observation2>"]',
         '}',
         '',
-        'Use your tools: read_file, write_file, edit_file, list_dir, exec.',
+        'Use your tools: ' + ', '.join(declared_tool_names) + '.',
         f'You have up to {max_iterations} tool iterations. Use them deliberately.',
     ]
 
@@ -2513,6 +2516,7 @@ async def _main_impl_body():
             selfevo_repo_root=_selfevo_repo_check,
             max_iterations=resolved_iterations,
             charter_in_system=bool(_charter),
+            declared_tool_names=EXECUTOR_TOOL_NAMES,
         )
 
         # Extract backlog title for MEMORY.md safety-net update after execution
@@ -2849,6 +2853,7 @@ async def _evaluate_candidate(cand_cycle_id: str, do_integration: bool, meas_met
             model=config.agents.defaults.model,
             web_search_config=config.tools.web.search,
             web_proxy=config.tools.web.proxy,
+            web_tools_enabled=False,
             exec_config=config.tools.exec,
             subagent_config=config.tools.subagent,
             restrict_to_workspace=False,
@@ -3127,6 +3132,7 @@ async def _evaluate_candidate(cand_cycle_id: str, do_integration: bool, meas_met
                         req, goal_text, report_source,
                         state_dir=STATE_DIR,
                         repair_context=_smoke_output,
+                        declared_tool_names=EXECUTOR_TOOL_NAMES,
                     )
                     # Spawn repair subagent
                     from nanobot.agent.subagent import SubagentManager as _SM2
@@ -3139,6 +3145,7 @@ async def _evaluate_candidate(cand_cycle_id: str, do_integration: bool, meas_met
                         model=_repair_cfg.agents.defaults.model,
                         web_search_config=_repair_cfg.tools.web.search,
                         web_proxy=_repair_cfg.tools.web.proxy,
+                        web_tools_enabled=False,
                         exec_config=_repair_cfg.tools.exec,
                         subagent_config=_repair_cfg.tools.subagent,
                         restrict_to_workspace=False,

tests/test_subagent_manager.py

@@ -1,6 +1,32 @@
 import pytest
 
 
+def test_executor_declared_and_registered_tool_names_match():
+    from nanobot.agent.subagent import SubagentManager
+
+    assert SubagentManager.declared_tool_names() == (
+        "read_file", "write_file", "edit_file", "list_dir", "exec"
+    )
+    manager = object.__new__(SubagentManager)
+    manager.web_tools_enabled = False
+    assert manager.registered_tool_names() == manager.declared_tool_names()
+    assert "web_search" not in manager.registered_tool_names()
+    assert "web_fetch" not in manager.registered_tool_names()
+
+    from nanobot.runtime import bridge
+    assert "Use your tools: " + ", ".join(SubagentManager.declared_tool_names()) + "." in bridge.build_task({}, "goal", "")
+
+
+def test_interactive_subagent_role_keeps_web_tools_available():
+    from nanobot.agent.subagent import SubagentManager
+
+    manager = object.__new__(SubagentManager)
+    manager.web_tools_enabled = True
+    assert manager.registered_tool_names() == (
+        "read_file", "write_file", "edit_file", "list_dir", "exec", "web_search", "web_fetch"
+    )
+
+
 def test_subagent_manager_accepts_deployed_bridge_compat_kwargs(tmp_path):
     from nanobot.agent.subagent import SubagentManager
     from nanobot.bus.queue import MessageBus