fix(flows): prevent double-execution of FunctionTools returning None in thread pool

When RunConfig.tool_thread_pool_config is enabled, _call_tool_in_thread_pool
used None as a sentinel to distinguish "FunctionTool ran in thread pool" from
"non-FunctionTool sync tool, needs async fallback". Because None is also a
valid return value from any FunctionTool whose underlying function has no
explicit return statement (implicit None), the sentinel check failed and
execution fell through to tool.run_async(), invoking the function a second
time silently.

Replace the None sentinel with a dedicated _SYNC_TOOL_RESULT_UNSET object so
that a legitimate None result from a FunctionTool is correctly returned on the
first execution, without triggering the async fallback path.

Fixes #5284

Author: Koushik-Salammagari

src/google/adk/flows/llm_flows/functions.py

@@ -68,6 +68,13 @@
 _TOOL_THREAD_POOLS: dict[int, ThreadPoolExecutor] = {}
 _TOOL_THREAD_POOL_LOCK = threading.Lock()
 
+# Sentinel object used to distinguish a FunctionTool that legitimately returns
+# None from a non-FunctionTool sync tool that skips thread-pool execution.
+# Using None as a sentinel would cause tools whose underlying function has no
+# explicit return statement (implicit None) to fall through to the async
+# fallback path and execute a second time.
+_SYNC_TOOL_RESULT_UNSET = object()
+
 
 def _is_live_request_queue_annotation(param: inspect.Parameter) -> bool:
   """Check whether a parameter is annotated as LiveRequestQueue.
@@ -159,13 +166,14 @@ def run_sync_tool():
         }
         return tool.func(**args_to_call)
       else:
-        # For other sync tool types, we can't easily run them in thread pool
-        return None
+        # For other sync tool types, we can't easily run them in thread pool.
+        # Return the sentinel so the caller knows to fall back to run_async.
+        return _SYNC_TOOL_RESULT_UNSET
 
     result = await loop.run_in_executor(
         executor, lambda: ctx.run(run_sync_tool)
     )
-    if result is not None:
+    if result is not _SYNC_TOOL_RESULT_UNSET:
       return result
   else:
     # For async tools, run them in a new event loop in a background thread.

tests/unittests/flows/llm_flows/test_functions_thread_pool.py

@@ -277,6 +277,70 @@ def blocking_sleep() -> dict:
         event_loop_ticks >= 5
     ), f'Event loop should have ticked at least 5 times, got {event_loop_ticks}'
 
+  @pytest.mark.asyncio
+  async def test_sync_tool_returning_none_executes_exactly_once(self):
+    """Test that a sync FunctionTool returning None is not double-executed.
+
+    Regression test for https://github.com/google/adk-python/issues/5284.
+    A FunctionTool whose underlying function returns None (e.g. side-effect-
+    only tools with no explicit return statement) must execute exactly once.
+    Previously the None return was mistaken for the internal sentinel used to
+    signal 'non-FunctionTool, fall back to run_async', causing a second
+    invocation via tool.run_async().
+    """
+    call_count = 0
+
+    def side_effect_only() -> None:
+      nonlocal call_count
+      call_count += 1
+      # No return statement — implicit None.
+
+    tool = FunctionTool(side_effect_only)
+    model = testing_utils.MockModel.create(responses=[])
+    agent = Agent(name='test_agent', model=model, tools=[tool])
+    invocation_context = await testing_utils.create_invocation_context(
+        agent=agent, user_content=''
+    )
+    tool_context = ToolContext(
+        invocation_context=invocation_context,
+        function_call_id='test_id',
+    )
+
+    result = await _call_tool_in_thread_pool(tool, {}, tool_context)
+
+    assert result is None
+    assert call_count == 1, (
+        f'Tool function executed {call_count} time(s); expected exactly 1.'
+    )
+
+  @pytest.mark.asyncio
+  async def test_sync_tool_returning_explicit_none_executes_exactly_once(self):
+    """Test that explicit None return from FunctionTool is not double-executed."""
+    call_count = 0
+
+    def explicit_none_return() -> None:
+      nonlocal call_count
+      call_count += 1
+      return None  # Explicit None return.
+
+    tool = FunctionTool(explicit_none_return)
+    model = testing_utils.MockModel.create(responses=[])
+    agent = Agent(name='test_agent', model=model, tools=[tool])
+    invocation_context = await testing_utils.create_invocation_context(
+        agent=agent, user_content=''
+    )
+    tool_context = ToolContext(
+        invocation_context=invocation_context,
+        function_call_id='test_id',
+    )
+
+    result = await _call_tool_in_thread_pool(tool, {}, tool_context)
+
+    assert result is None
+    assert call_count == 1, (
+        f'Tool function executed {call_count} time(s); expected exactly 1.'
+    )
+
   @pytest.mark.asyncio
   async def test_sync_tool_exception_propagates(self):
     """Test that exceptions from sync tools propagate correctly."""