fix: use asyncio.sleep to avoid blocking event loop

Co-authored-by: George Weale <gweale@google.com>
PiperOrigin-RevId: 910839657

Author: GWeale

src/google/adk/tools/environment_simulation/environment_simulation_engine.py

@@ -18,7 +18,6 @@
 import concurrent.futures
 import logging
 import random
-import time
 from typing import Any
 from typing import Dict
 from typing import Optional
@@ -107,7 +106,7 @@ async def simulate(
           self._random_generator.random()
           < injection_config.injection_probability
       ):
-        time.sleep(injection_config.injected_latency_seconds)
+        await asyncio.sleep(injection_config.injected_latency_seconds)
         if injection_config.injected_error:
           return {
               "error_code": (

tests/unittests/tools/environment_simulation/test_environment_simulation_engine.py

@@ -221,3 +221,37 @@ async def test_injection_with_random_seed_is_deterministic(
     result2 = await engine_injected.simulate(mock_tool, {}, MagicMock())
     assert result2 == {"injected": True}
     mock_create_strategy.assert_not_called()
+
+  async def test_injected_latency_awaits_asyncio_sleep(
+      self, mock_create_strategy, mock_analyzer
+  ):
+    """Regression guard against blocking time.sleep in the async path."""
+    del mock_create_strategy, mock_analyzer
+    latency = 0.2
+    config = EnvironmentSimulationConfig(
+        tool_simulation_configs=[
+            ToolSimulationConfig(
+                tool_name="test_tool",
+                injection_configs=[
+                    InjectionConfig(
+                        injected_latency_seconds=latency,
+                        injected_response={"injected": True},
+                    )
+                ],
+            )
+        ],
+        simulation_model="test-model",
+        simulation_model_configuration=genai_types.GenerateContentConfig(),
+    )
+    engine = EnvironmentSimulationEngine(config)
+    mock_tool = MagicMock()
+    mock_tool.name = "test_tool"
+
+    with patch(
+        "google.adk.tools.environment_simulation.environment_simulation_engine.asyncio.sleep",
+        new_callable=AsyncMock,
+    ) as mock_sleep:
+      result = await engine.simulate(mock_tool, {}, MagicMock())
+
+    mock_sleep.assert_awaited_once_with(latency)
+    assert result == {"injected": True}