feat(delegation): enforce per-child tool allowlists

agent/tool_executor.py

@@ -596,6 +596,17 @@ def _advance_start_order(callback=None) -> None:
 
         block_message = scope_block
         block_error_type = "tool_scope_block"
+        tool_allowlist = getattr(agent, "_tool_allowlist", None)
+        if (
+            block_message is None
+            and tool_allowlist is not None
+            and function_name not in tool_allowlist
+        ):
+            block_message = (
+                f"Tool '{function_name}' is not permitted by this delegated "
+                "agent's runtime tool_allowlist."
+            )
+            block_error_type = "tool_allowlist_block"
         if block_message is None:
             block_error_type = "plugin_block"
 
@@ -640,7 +651,12 @@ def _resolve_pre_tool_block():
             _advance_start_order()
             state["blocked"] = True
             if block_message is not None:
-                result = json.dumps({"error": block_message}, ensure_ascii=False)
+                error_payload = {"error": block_message}
+                if block_error_type == "tool_allowlist_block":
+                    error_payload.update(
+                        {"error_type": block_error_type, "tool": function_name}
+                    )
+                result = json.dumps(error_payload, ensure_ascii=False)
                 error_type = block_error_type
                 error_message = block_message
             else:

run_agent.py

@@ -8238,6 +8238,7 @@ def _dispatch_delegate_task(self, function_args: dict) -> str:
             max_iterations=function_args.get("max_iterations"),
             role=function_args.get("role"),
             background=(not _is_subagent),
+            tool_allowlist=function_args.get("tool_allowlist"),
             action=function_args.get("action"),
             subagent_id=function_args.get("subagent_id"),
             message=function_args.get("message"),

tests/run_agent/test_tool_call_guardrail_runtime.py

@@ -136,6 +136,53 @@ def test_config_enabled_hard_stop_blocks_repeated_exact_failure_before_execution
     assert "repeated_exact_failure_block" in messages[0]["content"]
 
 
+def test_tool_allowlist_blocks_disallowed_sequential_call_before_dispatch():
+    agent = _make_agent("read_file", "write_file", "patch", "terminal")
+    agent._tool_allowlist = frozenset({"read_file", "search_files", "grep"})
+    tc = _mock_tool_call(
+        "write_file",
+        json.dumps({"path": "/tmp/must-not-exist", "content": "blocked"}),
+        "c-allowlist-sequential",
+    )
+    msg = SimpleNamespace(content="", tool_calls=[tc])
+    messages = []
+
+    with patch("run_agent.handle_function_call", return_value="SHOULD_NOT_RUN") as mock_hfc:
+        agent._execute_tool_calls_sequential(msg, messages, "task-allowlist")
+
+    mock_hfc.assert_not_called()
+    result = json.loads(messages[0]["content"])
+    assert result["error_type"] == "tool_allowlist_block"
+    assert result["tool"] == "write_file"
+
+
+def test_tool_allowlist_blocks_disallowed_concurrent_calls_before_dispatch():
+    agent = _make_agent(
+        "read_file", "write_file", "patch", "terminal", "mcp__roshhome__update_request"
+    )
+    agent._tool_allowlist = frozenset({"read_file", "search_files", "grep"})
+    calls = [
+        _mock_tool_call("patch", "{}", "c-allowlist-patch"),
+        _mock_tool_call("terminal", "{}", "c-allowlist-terminal"),
+        _mock_tool_call("mcp__roshhome__update_request", "{}", "c-allowlist-mcp"),
+    ]
+    msg = SimpleNamespace(content="", tool_calls=calls)
+    messages = []
+
+    with patch("run_agent.handle_function_call", return_value="SHOULD_NOT_RUN") as mock_hfc:
+        agent._execute_tool_calls_concurrent(msg, messages, "task-allowlist")
+
+    mock_hfc.assert_not_called()
+    contents = [message["content"] for message in messages]
+    assert all('"error_type": "tool_allowlist_block"' in item for item in contents)
+    assert all(
+        f'"tool": "{tool_name}"' in content
+        for tool_name, content in zip(
+            ["patch", "terminal", "mcp__roshhome__update_request"], contents
+        )
+    )
+
+
 def test_sequential_after_call_appends_guidance_to_tool_result_without_extra_messages():
     agent = _make_agent("web_search")
     args = {"query": "same"}

tests/tools/test_delegate.py

@@ -70,6 +70,10 @@ def test_schema_valid(self):
         # capability-selection surface the model should not control.
         self.assertNotIn("toolsets", props)
         self.assertNotIn("toolsets", props["tasks"]["items"]["properties"])
+        self.assertIn("tool_allowlist", props)
+        self.assertIn(
+            "tool_allowlist", props["tasks"]["items"]["properties"]
+        )
         # max_iterations is intentionally NOT exposed to the model — it's
         # config-authoritative via delegation.max_iterations so users get
         # predictable budgets.
@@ -251,6 +255,80 @@ def test_orchestrator_composite_regains_only_delegate_task(self):
         )
 
 
+class TestChildToolAllowlist(unittest.TestCase):
+    @staticmethod
+    def _tool(name):
+        return {
+            "type": "function",
+            "function": {"name": name, "description": "", "parameters": {}},
+        }
+
+    def _build(self, tool_allowlist):
+        parent = _make_mock_parent()
+        parent.enabled_toolsets = ["file", "terminal", "mcp-roshhome"]
+        child = MagicMock()
+        child.tools = [
+            self._tool("read_file"),
+            self._tool("write_file"),
+            self._tool("terminal"),
+            self._tool("mcp__roshhome__update_request"),
+        ]
+        child.valid_tool_names = {
+            "read_file",
+            "write_file",
+            "terminal",
+            "mcp__roshhome__update_request",
+        }
+        child._context_engine_tool_names = set()
+
+        with patch("run_agent.AIAgent", return_value=child):
+            result = _build_child_agent(
+                task_index=0,
+                goal="Inspect safely",
+                context=None,
+                toolsets=None,
+                model=None,
+                max_iterations=10,
+                task_count=1,
+                parent_agent=parent,
+                tool_allowlist=tool_allowlist,
+            )
+        return result
+
+    def test_allowlist_intersects_final_builtin_and_mcp_snapshot(self):
+        child = self._build(["read_file", "search_files", "grep"])
+
+        self.assertEqual(child.valid_tool_names, {"read_file"})
+        self.assertEqual(
+            [tool["function"]["name"] for tool in child.tools], ["read_file"]
+        )
+        self.assertEqual(
+            child._tool_allowlist,
+            frozenset({"read_file", "search_files", "grep"}),
+        )
+
+    def test_empty_allowlist_is_deny_all(self):
+        child = self._build([])
+
+        self.assertEqual(child.tools, [])
+        self.assertEqual(child.valid_tool_names, set())
+        self.assertEqual(child._tool_allowlist, frozenset())
+
+    def test_absent_allowlist_preserves_current_snapshot(self):
+        child = self._build(None)
+
+        self.assertEqual(
+            child.valid_tool_names,
+            {
+                "read_file",
+                "write_file",
+                "terminal",
+                "mcp__roshhome__update_request",
+            },
+        )
+        self.assertIsNone(child._tool_allowlist)
+
+
 class TestDelegateTask(unittest.TestCase):
     def test_no_parent_agent(self):
         result = json.loads(delegate_task(goal="test"))
@@ -1366,11 +1444,13 @@ def fake_delegate_task(**kwargs):
                 parent,
                 {
                     "goal": "test",
+                    "tool_allowlist": ["read_file", "search_files"],
                     "acp_command": "claude",
                     "acp_args": ["--acp", "--stdio"],
                     "tasks": [
                         {
                             "goal": "nested",
+                            "tool_allowlist": [],
                             "acp_command": "codex",
                             "acp_args": ["--acp"],
                         },
@@ -1381,6 +1461,8 @@ def fake_delegate_task(**kwargs):
         self.assertNotIn("acp_command", captured)
         self.assertNotIn("acp_args", captured)
         self.assertEqual(captured["goal"], "test")
+        self.assertEqual(captured["tool_allowlist"], ["read_file", "search_files"])
+        self.assertEqual(captured["tasks"][0]["tool_allowlist"], [])
         self.assertNotIn("acp_command", captured["tasks"][0])
         self.assertNotIn("acp_args", captured["tasks"][0])
 
@@ -1620,6 +1702,77 @@ def test_orchestrator_role_keeps_delegation_at_depth_1(
             self.assertIn("delegation", kwargs["enabled_toolsets"])
             self.assertEqual(mock_child._delegate_role, "orchestrator")
 
+    @patch("tools.delegate_tool._resolve_delegation_credentials")
+    @patch("tools.delegate_tool._load_config", return_value={"max_spawn_depth": 2})
+    def test_top_level_tool_allowlist_reaches_child_runtime(
+        self, mock_cfg, mock_creds
+    ):
+        mock_creds.return_value = {
+            "provider": None, "base_url": None,
+            "api_key": None, "api_mode": None, "model": None,
+        }
+        parent = _make_mock_parent(depth=0)
+        parent.enabled_toolsets = ["terminal", "file"]
+        with patch("run_agent.AIAgent") as MockAgent:
+            mock_child = _make_role_mock_child()
+            mock_child.tools = []
+            mock_child.valid_tool_names = set()
+            MockAgent.return_value = mock_child
+
+            delegate_task(
+                goal="Inspect safely",
+                tool_allowlist=["read_file", "search_files", "grep"],
+                parent_agent=parent,
+            )
+
+        self.assertEqual(
+            mock_child._tool_allowlist,
+            frozenset({"read_file", "search_files", "grep"}),
+        )
+
+    @patch("tools.delegate_tool._resolve_delegation_credentials")
+    @patch("tools.delegate_tool._load_config", return_value={"max_spawn_depth": 2})
+    def test_per_task_tool_allowlist_overrides_top_level(
+        self, mock_cfg, mock_creds
+    ):
+        mock_creds.return_value = {
+            "provider": None, "base_url": None,
+            "api_key": None, "api_mode": None, "model": None,
+        }
+        parent = _make_mock_parent(depth=0)
+        parent.enabled_toolsets = ["terminal", "file"]
+        children = [_make_role_mock_child(), _make_role_mock_child()]
+        for child in children:
+            child.tools = []
+            child.valid_tool_names = set()
+
+        with patch("run_agent.AIAgent", side_effect=children):
+            delegate_task(
+                tasks=[
+                    {"goal": "Inspect the first component thoroughly"},
+                    {
+                        "goal": "Inspect the second component thoroughly",
+                        "tool_allowlist": [],
+                    },
+                ],
+                tool_allowlist=["read_file"],
+                parent_agent=parent,
+            )
+
+        self.assertEqual(children[0]._tool_allowlist, frozenset({"read_file"}))
+        self.assertEqual(children[1]._tool_allowlist, frozenset())
+
+    def test_invalid_tool_allowlist_fails_before_spawn(self):
+        parent = _make_mock_parent(depth=0)
+        result = json.loads(
+            delegate_task(
+                goal="Inspect safely",
+                tool_allowlist="read_file",
+                parent_agent=parent,
+            )
+        )
+        self.assertIn("tool_allowlist must be an array", result["error"])
+
     @patch("tools.delegate_tool._resolve_delegation_credentials")
     @patch("tools.delegate_tool._load_config",
            return_value={"max_spawn_depth": 2})

tests/tools/test_refresh_agent_mcp_tools.py

@@ -44,6 +44,28 @@ def test_refresh_adds_late_landing_tools(monkeypatch):
     assert len(agent.tools) == 3
 
 
+def test_refresh_intersects_late_mcp_tools_with_agent_allowlist(monkeypatch):
+    """A late MCP refresh must not reopen a delegated child's hard boundary."""
+    agent = _agent(["read_file"])
+    agent._tool_allowlist = frozenset({"read_file", "search_files", "grep"})
+
+    import model_tools
+    monkeypatch.setattr(
+        model_tools,
+        "get_tool_definitions",
+        lambda **kw: [
+            _tool("read_file"),
+            _tool("mcp__roshhome__update_request"),
+        ],
+    )
+
+    added = mcp_tool.refresh_agent_mcp_tools(agent)
+
+    assert added == set()
+    assert agent.valid_tool_names == {"read_file"}
+    assert [tool["function"]["name"] for tool in agent.tools] == ["read_file"]
+
+
 def test_refresh_preserves_memory_provider_and_context_engine_tools(monkeypatch):
     """B1 regression: a rebuild must NOT drop post-build-injected tools.