fix(tools): fail closed on non-boolean confirmation predicates

Author: FOWEPJF255

src/google/adk/tools/function_tool.py

@@ -20,7 +20,6 @@
 import re
 from typing import Any
 from typing import Callable
-from typing import cast
 from typing import Iterable
 from typing import Optional
 from typing import TYPE_CHECKING
@@ -317,10 +316,14 @@ async def check_require_confirmation(
       args_to_call = self._prepare_invocation_args(
           preprocessed_args, tool_context
       )
-      return cast(
-          bool,
-          await self._invoke_callable(self._require_confirmation, args_to_call),
+      result = await self._invoke_callable(
+          self._require_confirmation, args_to_call
       )
+      # typing.cast is erased at runtime. A callable that returns None (e.g. an
+      # unhandled branch) must not skip the confirmation gate.
+      if isinstance(result, bool):
+        return result
+      return True
     return bool(self._require_confirmation)
 
   def _is_invocation_type_error(

src/google/adk/tools/mcp_tool/mcp_tool.py

@@ -546,10 +546,14 @@ async def check_require_confirmation(
       args_to_call = self._prepare_callable_args(
           self._require_confirmation, args, tool_context
       )
-      return cast(
-          bool,
-          await self._invoke_callable(self._require_confirmation, args_to_call),
+      result = await self._invoke_callable(
+          self._require_confirmation, args_to_call
       )
+      # typing.cast is erased at runtime. A callable that returns None (e.g. an
+      # unhandled branch) must not skip the confirmation gate.
+      if isinstance(result, bool):
+        return result
+      return True
     return bool(self._require_confirmation)
 
   @override

src/google/adk/workflow/_function_node.py

@@ -727,13 +727,13 @@ async def _check_require_confirmation(
         args_to_call = {
             k: v for k, v in args_to_call.items() if k in valid_params
         }
-    return bool(
-        await _invoke_callable(
-            self.require_confirmation,
-            args_to_call,
-            unwrapped=pred_spec.unwrapped_func,
-        )
+    result = await _invoke_callable(
+        self.require_confirmation,
+        args_to_call,
+        unwrapped=pred_spec.unwrapped_func,
     )
+    # An omitted return must not silently bypass the confirmation gate.
+    return True if result is None else bool(result)
 
   @override
   async def _run_impl(

tests/unittests/tools/mcp_tool/test_mcp_tool.py

@@ -1708,6 +1708,33 @@ async def _require_confirmation_func(
       }
       tool_context.request_confirmation.assert_called_once()
 
+  @pytest.mark.asyncio
+  async def test_run_async_require_confirmation_callable_none_fail_closed(
+      self,
+  ):
+    """An omitted predicate return must request confirmation."""
+    tool = MCPTool(
+        mcp_tool=self.mock_mcp_tool,
+        mcp_session_manager=self.mock_session_manager,
+        require_confirmation=lambda **kwargs: None,
+    )
+    tool_context = Mock(spec=ToolContext)
+    tool_context.tool_confirmation = None
+    tool_context.request_confirmation = Mock()
+    tool_context.actions = EventActions()
+
+    result = await tool.run_async(
+        args={"param1": "test_value"}, tool_context=tool_context
+    )
+
+    assert result == {
+        "error": (
+            "This tool call requires confirmation, please approve or reject."
+        )
+    }
+    tool_context.request_confirmation.assert_called_once()
+    assert tool_context.actions.skip_summarization is True
+
   @pytest.mark.asyncio
   async def test_run_async_require_confirmation_callable_true_no_confirmation(
       self,

tests/unittests/tools/test_function_tool.py

@@ -488,6 +488,48 @@ def sample_func(arg1: str):
   assert result == {"received_arg": "hello"}
 
 
+@pytest.mark.asyncio
+async def test_check_require_confirmation_callable_none_fail_closed():
+  """Callable returning None must not skip the confirmation gate (#7010)."""
+
+  executed = []
+
+  def sample_func(arg1: str):
+    executed.append(arg1)
+    return {"received_arg": arg1}
+
+  def forgot_a_branch(arg1: str):
+    if arg1 == "force":
+      return True
+    # Falls through and returns None.
+
+  tool = FunctionTool(sample_func, require_confirmation=forgot_a_branch)
+  mock_invocation_context = MagicMock(spec=InvocationContext)
+  mock_invocation_context._state_schema = None
+  mock_invocation_context.session = MagicMock(spec=Session)
+  mock_invocation_context.session.state = MagicMock()
+  mock_invocation_context.agent = MagicMock()
+  mock_invocation_context.agent.name = "test_agent"
+  tool_context_mock = ToolContext(invocation_context=mock_invocation_context)
+  tool_context_mock.function_call_id = "test_function_call_id"
+
+  assert (
+      await tool.check_require_confirmation(
+          {"arg1": "hello"}, tool_context_mock
+      )
+      is True
+  )
+
+  result = await tool.run_async(
+      args={"arg1": "hello"},
+      tool_context=tool_context_mock,
+  )
+  assert result == {
+      "error": "This tool call requires confirmation, please approve or reject."
+  }
+  assert executed == []
+
+
 @pytest.mark.asyncio
 async def test_run_async_parameter_filtering(mock_tool_context):
   """Test that parameter filtering works correctly for functions with explicit parameters."""

tests/unittests/workflow/test_workflow_node.py

@@ -27,11 +27,9 @@
 from google.adk.tools.base_tool import BaseTool
 from google.adk.workflow import FunctionNode
 from google.adk.workflow import START
-from google.adk.workflow._base_node import BaseNode
 from google.adk.workflow._node import node
 from google.adk.workflow._node import Node
 from google.adk.workflow._parallel_worker import _ParallelWorker as ParallelWorker
-from google.adk.workflow._retry_config import RetryConfig
 from google.adk.workflow._tool_node import _ToolNode as ToolNode
 from google.adk.workflow._workflow import Workflow
 from google.genai import types
@@ -577,7 +575,8 @@ def after_node(node_input):
 
 
 @pytest.mark.parametrize(
-    "target_path,should_pause", [("/dev/tmp", False), ("/prod/db", True)]
+    "target_path,should_pause",
+    [("/dev/tmp", False), ("/prod/db", True), ("/home/x", True)],
 )
 @pytest.mark.asyncio
 async def test_node_require_confirmation_callable_predicate(
@@ -596,8 +595,10 @@ async def test_node_require_confirmation_callable_predicate(
   def setup_state(ctx: Context):
     ctx.state["path"] = target_path
 
-  def check_path(path: str) -> str:
+  def check_path(path: str) -> str | None:
     predicate_calls.append(path)
+    if path == "/home/x":
+      return None
     return path if path.startswith("/prod") else ""
 
   @node(require_confirmation=check_path)