fix(agents): route the turn after a paused single_turn sub-agent back to its delegator

src/google/adk/agents/_agent_router.py

@@ -32,6 +32,7 @@
 from ..flows.llm_flows.extensions._agent_transfer import _get_transfer_targets
 from ..flows.llm_flows.tools._functions import _collect_function_call_ids
 from ..flows.llm_flows.tools._functions import find_matching_function_call
+from ..utils._agent_mode import DELEGATED_TASK_MODES
 
 if TYPE_CHECKING:
   from ..agents.base_agent import BaseAgent
@@ -137,7 +138,7 @@ def _resolve_task_delegator(
   delegator = agent
   current = agent
   while current is not root_agent and current.parent_agent is not None:
-    if getattr(current, "mode", None) == "task":
+    if getattr(current, "mode", None) in DELEGATED_TASK_MODES:
       delegator = current.parent_agent
     current = current.parent_agent
   return delegator

tests/unittests/agents/test_agent_router.py

@@ -29,6 +29,7 @@
 from google.adk.events.event_actions import EventActions
 from google.adk.sessions.in_memory_session_service import InMemorySessionService
 from google.adk.sessions.session import Session
+from google.adk.utils._agent_mode import AgentMode
 from google.genai import types
 import pytest
 
@@ -849,6 +850,64 @@ def test_find_agent_to_run_routes_task_agent_to_its_delegator(
   assert _agent_router.find_agent_to_run(session, root) == helper
 
 
+@pytest.mark.parametrize("user_answers_call", [False, True])
+@pytest.mark.parametrize("delegator_name", ["root_agent", "helper"])
+@pytest.mark.parametrize(
+    "mode", ["single_turn", AgentMode.SINGLE_TURN], ids=["str", "enum"]
+)
+def test_find_agent_to_run_routes_single_turn_agent_to_its_delegator(
+    mode: str,
+    delegator_name: str,
+    user_answers_call: bool,
+):
+  """A turn left by a paused single_turn agent routes to its delegator."""
+  root = _MockLlmAgent("root_agent")
+  helper = _MockLlmAgent("helper", parent_agent=root)
+  root.sub_agents = [helper]
+  delegator = root if delegator_name == "root_agent" else helper
+  worker = _MockLlmAgent("worker", parent_agent=delegator)
+  worker.mode = mode
+  delegator.sub_agents = [*delegator.sub_agents, worker]
+  events = [
+      Event(
+          invocation_id="inv1",
+          author="worker",
+          content=types.Content(
+              role="model",
+              parts=[
+                  types.Part(
+                      function_call=types.FunctionCall(
+                          id="fc_1", name="adk_request_confirmation", args={}
+                      )
+                  )
+              ],
+          ),
+      )
+  ]
+  if user_answers_call:
+    events.append(
+        Event(
+            invocation_id="inv2",
+            author="user",
+            content=types.Content(
+                role="user",
+                parts=[
+                    types.Part(
+                        function_response=types.FunctionResponse(
+                            id="fc_1",
+                            name="adk_request_confirmation",
+                            response={"confirmed": True},
+                        )
+                    )
+                ],
+            ),
+        )
+    )
+  session = Session(id="s1", app_name="app", user_id="u1", events=events)
+
+  assert _agent_router.find_agent_to_run(session, root) == delegator
+
+
 def _transfer_events(
     author: str, target: str, *, call_id: str = "transfer_1"
 ) -> list[Event]:

tests/unittests/agents/test_llm_agent_interruptions.py

@@ -403,7 +403,6 @@ def transfer_to_child(tool_context: ToolContext) -> str:
     # Then it should complete successfully
     assert any('Parent final answer' in t for t in text_parts(resume_events))
 
-  @pytest.mark.xfail(reason='Single-turn agent as subagent not supported yet.')
   async def test_single_turn_child_agent_interrupt_and_resume(self):
     """Single-turn child agent yields on LRO and resumes successfully.
 

tests/unittests/test_runners.py

@@ -57,6 +57,7 @@
 from google.adk.sessions.session import Session
 from google.adk.tools.agent_tool import AgentTool
 from google.adk.tools.base_toolset import BaseToolset
+from google.adk.tools.function_tool import FunctionTool
 from google.adk.tools.long_running_tool import LongRunningFunctionTool
 from google.adk.workflow import BaseNode
 from google.adk.workflow import node
@@ -6745,5 +6746,148 @@ async def test_run_async_tool_error_in_root_parallel_agent_keeps_each_author():
   assert await _kept_authors(runner, "fast_tool") == ["a", "b"]
 
 
+@pytest.mark.parametrize("resumable", [False, True])
+@pytest.mark.parametrize("answer", ["new_message", "approve", "reject"])
+@pytest.mark.parametrize("delegator", ["root", "nested_chat"])
+@pytest.mark.asyncio
+async def test_run_async_paused_single_turn_agent_returns_turn_to_delegator(
+    delegator: str, answer: str, resumable: bool
+):
+  """A single_turn agent paused on a confirmation does not take later turns.
+
+  The caller that delegated to it gets the turn back, and its result comes
+  back to the caller as the response to the caller's original call.
+  """
+  paid = []
+
+  def pay(amount: int) -> dict[str, int]:
+    """Pays an amount."""
+    paid.append(amount)
+    return {"paid": amount}
+
+  helper = LlmAgent(
+      name="helper",
+      mode="single_turn",
+      model=testing_utils.MockModel.create(
+          responses=[
+              types.Part.from_function_call(name="pay", args={"amount": 5}),
+              "Helper done",
+              "Helper chatting",
+          ]
+      ),
+      tools=[FunctionTool(pay, require_confirmation=True)],
+  )
+  caller_model = testing_utils.MockModel.create(
+      responses=[
+          types.Part.from_function_call(
+              name="helper", args={"request": "pay 5"}
+          ),
+          "Caller reply",
+          "Caller thanks",
+      ]
+  )
+  if delegator == "root":
+    caller = LlmAgent(
+        name="root_agent", model=caller_model, sub_agents=[helper]
+    )
+    root = caller
+  else:
+    # A chat sub-agent the root transferred to delegates to the helper.
+    caller = LlmAgent(
+        name="coordinator", model=caller_model, sub_agents=[helper]
+    )
+    root = LlmAgent(
+        name="root_agent",
+        model=testing_utils.MockModel.create(
+            responses=[
+                types.Part.from_function_call(
+                    name="transfer_to_agent", args={"agent_name": "coordinator"}
+                ),
+                "Root reply",
+            ]
+        ),
+        sub_agents=[caller],
+    )
+  runner = _abort_runner(root, resumable=resumable)
+
+  first, _ = await _run_turn(runner, "s", "Pay 5")
+  helper_call_id = next(
+      fc.id
+      for e in first
+      if e.author == caller.name
+      for fc in e.get_function_calls()
+      if fc.name == "helper"
+  )
+  confirmation_id = next(
+      fc.id
+      for e in first
+      for fc in e.get_function_calls()
+      if fc.name == "adk_request_confirmation"
+  )
+  if answer == "new_message":
+    message = types.Content(role="user", parts=[types.Part(text="Weather?")])
+  else:
+    message = types.Content(
+        role="user",
+        parts=[
+            types.Part(
+                function_response=types.FunctionResponse(
+                    id=confirmation_id,
+                    name="adk_request_confirmation",
+                    response={"confirmed": answer == "approve"},
+                )
+            )
+        ],
+    )
+  second = [
+      e
+      async for e in runner.run_async(
+          user_id=TEST_USER_ID, session_id="s", new_message=message
+      )
+  ]
+  third, _ = await _run_turn(runner, "s", "Thanks")
+
+  assert paid == ([5] if answer == "approve" else [])
+  assert _texts([e.content for e in second if e.author == caller.name]) == [
+      "Caller reply"
+  ]
+  assert "Helper chatting" not in _texts([e.content for e in second + third])
+  assert _texts([e.content for e in third]) == ["Caller thanks"]
+  assert {e.author for e in third if e.content} == {caller.name}
+
+  # The helper's result comes back to the caller once, under the caller's
+  # original call id, and reaches the caller's next model request.
+  helper_responses = [
+      fr
+      for e in second
+      if e.author == caller.name
+      for fr in e.get_function_responses()
+      if fr.name == "helper"
+  ]
+  next_request = caller_model.requests[1]
+  request_responses = [
+      p.function_response
+      for c in next_request.contents
+      for p in c.parts or []
+      if p.function_response and p.function_response.name == "helper"
+  ]
+  request_calls = [
+      p.function_call
+      for c in next_request.contents
+      for p in c.parts or []
+      if p.function_call
+  ]
+  if answer == "new_message":
+    assert helper_responses == []
+    assert request_responses == []
+    assert request_calls == []  # no dangling call reaches the model
+  else:
+    assert [fr.id for fr in helper_responses] == [helper_call_id]
+    assert helper_responses[0].response == {"result": "Helper done"}
+    assert [fr.response for fr in request_responses] == [
+        {"result": "Helper done"}
+    ]
+
+
 if __name__ == "__main__":
   pytest.main([__file__])