Workflow tool with RequestInput inside a single_turn sub-agent never resumes on 2.13.0; the answer goes to the delegator #7974

Required Information

Describe the Bug:

A RequestInput inside a Workflow used as a tool of a mode='single_turn' sub-agent does not pause the sub-agent's delegation. Since 2.13.0 the user's answer never reaches the workflow either. Nothing is raised.

• Turn 1, on every version I tried (2.11.0 to main): the workflow pauses and the question is shown, but the root's helper call is answered with {'result': None} straight away, and the root model replies in the same turn.
• Turn 2: on 2.12.0 the answer reached helper, the workflow resumed and after ran, although the root only ever saw {'result': None}. Since 2.13.0 the answer goes to the root, which sees the same stale {'result': None}. The workflow never resumes.

The turn 2 change comes from 6727f8f (#7605, for #7602). It added SINGLE_TURN to DELEGATED_TASK_MODES (_agent_mode.py#L62-L65), so find_agent_to_run resolves the answer to helper and then returns _resolve_task_delegator(helper, root), which is root (_agent_router.py#L218-L223, #L134-L144). That routing is right when the delegation really is paused, and it works for a confirmation pause in the same helper (see Additional Context). Here it exposes the turn 1 gap:

1. NodeTool runs the workflow with raise_on_wait=True (_node_tool.py#L366-L373). The scheduler copies the interrupt id onto the per-call tool context and raises NodeInterruptedError (_dynamic_node_scheduler.py#L851-L853).
2. The helper's own node context collects interrupt ids only from events its loop yields (_node_runner.py#L466-L467), so the helper's run ends with no interrupt ids and no output. Instrumenting DynamicNodeScheduler.__call__ on main shows the classify run ending with interrupt_ids={'q1'} and the helper run with interrupt_ids=set(), output=None.
3. _SingleTurnAgentTool calls run_node without raise_on_wait (agent_tool.py#L597-L602), and single_turn agents never get wait_for_output (_workflow_graph_utils.py#L170-L171). So run_node returns None (_dynamic_node_scheduler.py#L854-L863) and the root's call is answered {'result': None}.

Steps to Reproduce:

1. pip install google-adk==2.13.0
2. Save the script under Minimal Reproduction Code as repro.py. It uses scripted models, so no API key is needed.
3. python repro.py

It runs a chat root that delegates to a single_turn sub-agent helper. helper calls the classify workflow tool, which asks one question (RequestInput) and then runs after. The script answers the question and prints which models were called with what, and whether after ran. It does this with and without ResumabilityConfig.

Expected Behavior:

As with a confirmation pause in the same helper: in turn 1 the root's helper call stays open and the root does not reply. After the answer, the workflow resumes, after runs, helper gets the workflow output, and the root gets helper's result.

Observed Behavior:

google-adk 2.13.0, resumable=True
  turn 1: root   model called, helper response = -
  turn 1: helper model called, classify response = -
  turn 1: root   model called, helper response = {'result': None}
  turn 2: root   model called, helper response = {'result': None}
  after ran = False

resumable=False prints the same. 2.12.0 prints this:

google-adk 2.12.0, resumable=True
  turn 1: root   model called, helper response = -
  turn 1: helper model called, classify response = -
  turn 1: root   model called, helper response = {'result': None}
  turn 2: helper model called, classify response = {'result': "after got {'text': 'alpha'}"}
  after ran = True

Environment Details:

• ADK Library Version (pip show google-adk): 2.13.0 (latest on PyPI, released 2026-10-23), and main at d91dd34 (2026-11-03); 2.12.0 for comparison
• Desktop OS: macOS
• Python Version (python -V): 3.11.14

Model Information:

• Are you using LiteLLM: No
• Which model is being used: N/A, scripted BaseLlm models

Optional Information

Regression:

Yes, for the resume. It works at 8909689 and fails at 6727f8f, its child, in both modes, so the first failing release is 2.13.0. The turn 1 {'result': None} shows up on 2.11.0, 2.12.0 and 2.13.0.

Additional Context:

• Control on main: a require_confirmation tool in the same single_turn helper works in both modes. The root stays silent in turn 1, and after the approval helper runs again and the root gets helper's result. With the same instrumentation, the helper's run there ends with the confirmation's interrupt id, so the delegation pauses.
• The same workflow tool resumes on 2.12.0 and later, with and without resumability, when it sits on the root chat agent, on a chat sub-agent reached by transfer (with or without disallow_transfer_to_parent=True), or on a task sub-agent. Task mode pauses its delegation through raise_on_wait=True (_llm_agent_wrapper.py#L276-L283).
• A possible direction: carry a NodeTool's interrupt into the run of the agent that called the tool, as a confirmation already does, so the _SingleTurnAgentTool call pauses and fix(agents): route the turn after a paused single_turn sub-agent back to its delegator #7605's routing to the delegator holds. I have not written a fix.

Minimal Reproduction Code:

import asyncio

import google.adk
from google.adk import Agent, Event, Workflow
from google.adk.apps import App, ResumabilityConfig
from google.adk.events import RequestInput
from google.adk.models.base_llm import BaseLlm
from google.adk.models.llm_response import LlmResponse
from google.adk.runners import InMemoryRunner
from google.genai import types
from pydantic import BaseModel

TURN = {'n': 1}
CALLED = []  # (turn, agent, function responses the model was sent)


class Args(BaseModel):
  item: str


def ask(node_input):
  yield RequestInput(interrupt_id='q1', message='Which heading?')


def after(node_input):
  yield Event(output=f'after got {node_input}', state={'after_ran': True})


class ScriptedLlm(BaseLlm):
  """Calls `tool` once, then answers with text. Never calls a tool in turn 2."""

  model: str = 'scripted'
  tool: str

  async def generate_content_async(self, llm_request, stream=False):
    seen = {
        p.function_response.name: p.function_response.response
        for c in llm_request.contents
        for p in c.parts or []
        if p.function_response
    }
    CALLED.append((TURN['n'], self.tool, seen.get(self.tool, '-')))
    if self.tool in seen or TURN['n'] > 1:
      part = types.Part(text=f'done, {self.tool} returned {seen.get(self.tool)}')
    else:
      call = types.FunctionCall(name=self.tool, args={'item': 'w'})
      if self.tool == 'helper':
        call.args = {'request': 'classify a widget'}
      part = types.Part(function_call=call)
    yield LlmResponse(content=types.Content(role='model', parts=[part]))


async def run(resumable: bool) -> None:
  CALLED.clear()
  TURN['n'] = 1
  classify = Workflow(
      name='classify',
      description='Asks one question, then classifies.',
      input_schema=Args,
      edges=[('START', ask, after)],
  )
  helper = Agent(
      name='helper',
      description='Classifies items.',
      mode='single_turn',
      model=ScriptedLlm(tool='classify'),
      tools=[classify],
  )
  root = Agent(name='root', model=ScriptedLlm(tool='helper'), sub_agents=[helper])
  app = App(
      name='repro',
      root_agent=root,
      resumability_config=(
          ResumabilityConfig(is_resumable=True) if resumable else None
      ),
  )
  runner = InMemoryRunner(app=app)
  session = await runner.session_service.create_session(
      app_name='repro', user_id='u'
  )

  async def send(*parts: types.Part) -> None:
    message = types.Content(role='user', parts=list(parts))
    async for _ in runner.run_async(
        user_id='u', session_id=session.id, new_message=message
    ):
      pass

  await send(types.Part(text='classify a widget'))  # pauses on q1
  TURN['n'] = 2
  await send(
      types.Part(
          function_response=types.FunctionResponse(
              id='q1', name='adk_request_input', response={'text': 'alpha'}
          )
      )
  )
  state = (
      await runner.session_service.get_session(
          app_name='repro', user_id='u', session_id=session.id
      )
  ).state
  print(f'google-adk {google.adk.__version__}, resumable={resumable}')
  for turn, tool, response in CALLED:
    agent = 'root' if tool == 'helper' else 'helper'
    print(f'  turn {turn}: {agent:6} model called, {tool} response = {response}')
  print(f"  after ran = {bool(state.get('after_ran'))}")


async def main() -> None:
  for resumable in (True, False):
    await run(resumable)


asyncio.run(main())

How often has this issue occurred?:

• Always (100%)