fix(workflow): stop rerunning a waiting wait_for_output node on resume

docs/guides/workflow/base_node/index.md

@@ -118,6 +118,8 @@ With the default `False`, a node completes as soon as its logic finishes, whethe
 
 With `True`, the node completes only if it yields an output or a route. Finishing without yielding either leaves it in the `WAITING` state, and its successors are not triggered. A `WAITING` node still accepts new triggers, so a predecessor firing again re-runs it. That behavior is what you want for a node that accumulates across several triggers and only emits once it has enough, such as one keeping a running total in state and producing an output when that total crosses a threshold.
 
+When an interrupted workflow resumes, a `WAITING` node in the graph is not run again with the input it already handled: it stays `WAITING` until its next trigger, which starts a new run. Agents in task or chat mode are the exception, since they wait for the user's next message rather than for a trigger, and rerun on resume. A custom node that waits on the user in the same way, rather than on another trigger, needs `rerun_on_resume=True` as well.
+
 The failure mode is worth knowing before you set this. A node that is *never* triggered again stays `WAITING` to the end of the run, and the run does not wait for it. The workflow finishes straight away, having emitted nothing, and the whole downstream half of your graph is silently skipped. Nothing raises and nothing is logged about it, at any level. The framework treats this as a configuration mistake rather than a fault to report.
 
 One node type ignores this flag entirely. `JoinNode` does not use it, because it synchronizes by waiting for all of its predecessors, which is a different mechanism.

src/google/adk/workflow/utils/_replay_interceptor.py

@@ -172,7 +172,10 @@ def check_interception(
 
   else:
     # Case 6: Cross-turn no events, or events contain no output, route, or interrupts.
-    if node.wait_for_output:
+    if node.wait_for_output and _reruns_while_waiting(node):
+      # A waiting agent (task or chat mode) reruns to take the user's next
+      # message. Other wait_for_output nodes fall through: their next trigger
+      # starts a new run, so rerunning this one would repeat its side effects.
       should_run = True
       resume_inputs = recovered.resolved_responses
     elif node.rerun_on_resume:
@@ -235,3 +238,10 @@ def create_mock_context(
   mock_ctx._interrupt_ids = result.interrupts  # pylint: disable=protected-access
 
   return mock_ctx
+
+
+def _reruns_while_waiting(node: BaseNode) -> bool:
+  """Whether a wait_for_output node with no recorded outcome runs again."""
+  from ...agents.base_agent import BaseAgent  # pylint: disable=g-import-not-at-top
+
+  return node.rerun_on_resume or isinstance(node, BaseAgent)

tests/unittests/workflow/test_workflow_hitl.py

@@ -2704,3 +2704,189 @@ def book() -> dict[str, str]:
 
   assert booked == ['trip']
   assert {e.branch for e in events1 + events2 if e.author == 'booker'} == {None}
+
+
+class _GatherNode(BaseNode):
+  """Collects inputs in state and emits every `batch` inputs.
+
+  This is the accumulate-then-emit pattern `wait_for_output` is meant for.
+  """
+
+  batch: int = 2
+  calls: list[Any] = Field(default_factory=list)
+
+  @override
+  async def _run_impl(
+      self, *, ctx: Context, node_input: Any
+  ) -> AsyncGenerator[Any, None]:
+    self.calls.append(node_input)
+    seen = list(ctx.state.get('seen') or []) + [node_input]
+    ctx.state['seen'] = seen
+    if len(seen) % self.batch == 0:
+      yield Event(output=seen[-self.batch :])
+
+
+def _gather_app(
+    request: pytest.FixtureRequest, agent: Workflow, resumable: bool
+) -> App:
+  return App(
+      name=request.function.__name__,
+      root_agent=agent,
+      resumability_config=(
+          ResumabilityConfig(is_resumable=True) if resumable else None
+      ),
+  )
+
+
+@pytest.mark.parametrize('resumable', [False, True])
+@pytest.mark.asyncio
+async def test_resume_does_not_rerun_superseded_waiting_run(
+    request: pytest.FixtureRequest, resumable: bool
+):
+  """START -> A -> Gather, START -> B -> Gather, Gather -> Ask -> Done.
+
+  Gather waits after A's output (Gather@1) and emits after B's (Gather@2).
+  Answering Ask must not run Gather@1 again: Gather@2 already superseded it.
+  """
+  gather = _GatherNode(name='Gather', wait_for_output=True)
+  ask = RequestInputNode(name='Ask')
+  done = InputCapturingNode(name='Done')
+  agent = Workflow(
+      name='test_agent',
+      edges=[
+          (START, _TestingNode(name='A', message='a'), gather),
+          (START, _TestingNode(name='B', message='b'), gather),
+          (gather, ask, done),
+      ],
+  )
+  runner = testing_utils.InMemoryRunner(
+      app=_gather_app(request, agent, resumable)
+  )
+
+  events1 = await runner.run_async(testing_utils.get_user_content('go'))
+  [interrupt_id] = extract_event_interrupt_ids(
+      workflow_testing_utils.get_request_input_events(events1)[0]
+  )
+  assert sorted(gather.calls) == ['a', 'b']
+
+  await runner.run_async(
+      new_message=testing_utils.UserContent(
+          create_request_input_response(interrupt_id, {'yes': True})
+      ),
+      invocation_id=events1[0].invocation_id,
+  )
+
+  assert sorted(gather.calls) == ['a', 'b']
+  assert done.received_inputs == [{'yes': True}]
+  assert sorted(runner.session.state['seen']) == ['a', 'b']
+
+
+@pytest.mark.parametrize('resumable', [False, True])
+@pytest.mark.asyncio
+async def test_resume_does_not_rerun_waiting_run_before_its_next_trigger(
+    request: pytest.FixtureRequest, resumable: bool
+):
+  """START -> A -> Gather, START -> Ask -> B -> Gather, Gather -> Done.
+
+  Gather waits after A's output while Ask pauses for the user. Answering Ask
+  must not run Gather again with A's output: it should stay waiting until B's
+  output arrives, then emit both.
+  """
+  gather = _GatherNode(name='Gather', wait_for_output=True)
+  ask = RequestInputNode(name='Ask')
+  done = InputCapturingNode(name='Done')
+  agent = Workflow(
+      name='test_agent',
+      edges=[
+          (START, _TestingNode(name='A', message='a'), gather),
+          (START, ask, _TestingNode(name='B', message='b'), gather),
+          (gather, done),
+      ],
+  )
+  runner = testing_utils.InMemoryRunner(
+      app=_gather_app(request, agent, resumable)
+  )
+
+  events1 = await runner.run_async(testing_utils.get_user_content('go'))
+  [interrupt_id] = extract_event_interrupt_ids(
+      workflow_testing_utils.get_request_input_events(events1)[0]
+  )
+  assert gather.calls == ['a']
+
+  await runner.run_async(
+      new_message=testing_utils.UserContent(
+          create_request_input_response(interrupt_id, {'yes': True})
+      ),
+      invocation_id=events1[0].invocation_id,
+  )
+
+  assert gather.calls == ['a', 'b']
+  assert done.received_inputs == [['a', 'b']]
+  assert runner.session.state['seen'] == ['a', 'b']
+
+
+class _LapRouterNode(BaseNode):
+  """Routes 'loop' until it has run `laps` times, then 'stop'."""
+
+  laps: int = 3
+
+  @override
+  async def _run_impl(
+      self, *, ctx: Context, node_input: Any
+  ) -> AsyncGenerator[Any, None]:
+    lap = (ctx.state.get('lap') or 0) + 1
+    ctx.state['lap'] = lap
+    yield Event(output=lap, route='loop' if lap < self.laps else 'stop')
+
+
+@pytest.mark.parametrize('resumable', [False, True])
+@pytest.mark.asyncio
+async def test_resume_does_not_rerun_waiting_runs_from_earlier_loop_laps(
+    request: pytest.FixtureRequest, resumable: bool
+):
+  """START -> A -> Gather, A -> B -> Gather, Gather -> Router.
+
+  Router loops back to A twice, then routes to Ask -> Done. Each lap leaves
+  one waiting run of Gather. Answering Ask must not rerun them, which would
+  feed Gather extra inputs, run the loop again and ask a second time.
+  """
+  gather = _GatherNode(name='Gather', wait_for_output=True)
+  node_a = _TestingNode(name='A', message='a')
+  node_b = _TestingNode(name='B', message='b')
+  router = _LapRouterNode(name='Router')
+  ask = RequestInputNode(name='Ask')
+  done = InputCapturingNode(name='Done')
+  agent = Workflow(
+      name='test_agent',
+      edges=[
+          (START, node_a, gather),
+          (node_a, node_b, gather),
+          (gather, router),
+          (router, {'loop': node_a, 'stop': ask}),
+          (ask, done),
+      ],
+  )
+  runner = testing_utils.InMemoryRunner(
+      app=_gather_app(request, agent, resumable)
+  )
+
+  events1 = await runner.run_async(testing_utils.get_user_content('go'))
+  [interrupt_id] = extract_event_interrupt_ids(
+      workflow_testing_utils.get_request_input_events(events1)[0]
+  )
+  assert gather.calls == ['a', 'b'] * 3
+
+  events2 = await asyncio.wait_for(
+      runner.run_async(
+          new_message=testing_utils.UserContent(
+              create_request_input_response(interrupt_id, {'yes': True})
+          ),
+          invocation_id=events1[0].invocation_id,
+      ),
+      timeout=10,
+  )
+
+  assert gather.calls == ['a', 'b'] * 3
+  assert done.received_inputs == [{'yes': True}]
+  assert not workflow_testing_utils.get_request_input_events(events2)
+  assert runner.session.state['lap'] == 3

tests/unittests/workflow/utils/test_replay_interceptor.py

@@ -345,6 +345,57 @@ def test_cross_turn_no_outcome_static_vs_dynamic_run(
   assert dynamic_result.should_run
 
 
+class _AgentNode(BaseAgent):
+  """A minimal agent, standing in for a task- or chat-mode LlmAgent."""
+
+  async def _run_async_impl(self, ctx):
+    yield  # pragma: no cover
+
+
+@pytest.mark.parametrize('is_static', [True, False])
+def test_cross_turn_no_outcome_wait_for_output_node_like_plain_node(
+    is_static: bool,
+):
+  """Case 6 treats a plain wait_for_output node like any other node.
+
+  Its next trigger starts a new run, so a static run is fast-forwarded rather
+  than rerun with the input it already handled. A dynamic run still reruns.
+  """
+  recovered = _ChildScanState(run_id='1')
+  node = BaseNode(name='node', wait_for_output=True)
+  run = DynamicNodeRun(
+      state=NodeState(run_id='1'),
+      recovered_state=recovered,
+      is_static=is_static,
+  )
+
+  result = check_interception(node=node, recovered=recovered, current_run=run)
+
+  assert result.should_run == (not is_static)
+
+
+@pytest.mark.parametrize(
+    'node',
+    [
+        BaseNode(name='node', wait_for_output=True, rerun_on_resume=True),
+        _AgentNode(name='agent', wait_for_output=True),
+    ],
+    ids=['rerun_on_resume', 'agent'],
+)
+def test_cross_turn_no_outcome_waiting_agent_or_rerun_node_reruns(
+    node: BaseNode,
+):
+  """Case 6 reruns a waiting agent or rerun_on_resume node to take new input."""
+  recovered = _ChildScanState(run_id='1')
+  run = DynamicNodeRun(
+      state=NodeState(run_id='1'), recovered_state=recovered, is_static=True
+  )
+
+  result = check_interception(node=node, recovered=recovered, current_run=run)
+
+  assert result.should_run
+
+
 # --- create_mock_context ---