fix: stop rerunning a waiting wait_for_output node on resume

Merge #7991

On resume, a `wait_for_output` node whose earlier run ended waiting (no output,
route or interrupt) is no longer run again with the input it already handled;
it stays waiting until its next trigger. Agents in task or chat mode and nodes
with `rerun_on_resume=True` still rerun. A custom node that waits on the user
rather than on a trigger now needs `rerun_on_resume=True` to be rerun.

Fixes #7905

PiperOrigin-RevId: 1015845387

Author: AkshayShah03

.agents/skills/adk-architecture/references/interface-workflow.md

@@ -212,9 +212,14 @@ Within a single step (`_execute_step`), when replay is enabled:
      `resume_inputs=recovered.resolved_responses`.
 
 7. **No output, route, or interrupts in recovered events** —
-   - If `node.wait_for_output` or `node.rerun_on_resume`: returns
+   - If `node.wait_for_output` and the node is a `BaseAgent` (task or chat
+     mode), or `node.rerun_on_resume` with `wait_for_output`: returns
      `should_run=True` with `resume_inputs=recovered.resolved_responses`.
-   - Otherwise: returns `should_run = (current_run is not None and not current_run.is_static)`
+   - Else if `node.rerun_on_resume`: fast-forwards (`should_run=False`) when
+     `recovered.finished_after_resume`, otherwise returns `should_run=True`
+     with `resume_inputs`.
+   - Otherwise (including any other `wait_for_output` node, whose next trigger
+     starts a new run): returns `should_run = (current_run is not None and not current_run.is_static)`
      (static nodes that completed with `None` output fast-forward with
      `should_run=False`, while dynamic nodes with no recorded outcome
      re-execute with `should_run=True`).

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. This applies to a waiting run that recorded at least one event, such as a state update; a run that emitted nothing is still rerun today.
+
 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

@@ -21,6 +21,7 @@
 from typing import Any
 from typing import TYPE_CHECKING
 
+from ...agents.base_agent import BaseAgent
 from ...agents.context import Context
 from .._base_node import BaseNode
 from .._node_status import NodeStatus
@@ -172,7 +173,11 @@ 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) or rerun_on_resume node reruns to
+      # take new input. 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 +240,8 @@ 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."""
+  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 ---