Repository navigation
Resuming a workflow re-runs the earlier no-output runs of a wait_for_output node, so its side effects happen twice #7905
Description
Describe the bug
A node with wait_for_output=True that accumulates inputs, the pattern the wait_for_output section of the base node guide describes, runs again when the user answers a later RequestInput. On resume, every earlier run of that node that finished without an output is executed a second time with its original input, even though a later run of the same node already completed. Whatever the node does runs twice: here it appends to ctx.state twice.
In a loop it gets worse. The duplicate inputs can make the node emit again, which re-triggers everything after it: the loop runs an extra lap, the user is asked for the same approval a second time, and the turn after that hangs. With a random workflow fuzzer, I also saw WorkflowInvariantError: Replay divergence detected: Timed out waiting for sequence key ... on resumable apps, from graphs that combine this with a loop.
To Reproduce
import asyncio
from google.adk import Event, Workflow
from google.adk.events.request_input import RequestInput
from google.adk.runners import InMemoryRunner
from google.adk.workflow import START, BaseNode
from google.genai import types
calls = []
class Gather(BaseNode):
"""Emits once it has received two inputs (the pattern from the wait_for_output docs)."""
async def _run_impl(self, *, ctx, node_input):
calls.append(f"gather({node_input})")
seen = list(ctx.state.get("seen") or []) + [node_input]
ctx.state["seen"] = seen
if len(seen) == 2:
yield Event(output=seen)
def b(node_input):
return "B"
def c(node_input):
return "C"
def approve(node_input):
return RequestInput(interrupt_id="ok", message=f"Approve {node_input}?")
def done(node_input):
calls.append(f"done({node_input})")
async def main():
gather = Gather(name="gather", wait_for_output=True)
wf = Workflow(
name="wf",
edges=[(START, b, gather), (START, c, gather), (gather, approve, done)],
)
runner = InMemoryRunner(node=wf)
session = await runner.session_service.create_session(
app_name=runner.app_name, user_id="u"
)
async def run(message):
async for _ in runner.run_async(
user_id="u", session_id=session.id, new_message=message
):
pass
await run(types.UserContent("go"))
print("turn 1:", calls)
calls.clear()
await run(
types.Content(
role="user",
parts=[
types.Part(
function_response=types.FunctionResponse(
id="ok", name="adk_request_input", response={"yes": True}
)
)
],
)
)
print("turn 2:", calls)
session = await runner.session_service.get_session(
app_name=runner.app_name, user_id="u", session_id=session.id
)
print("seen:", session.state["seen"])
asyncio.run(main())Output on main (8cd3b12):
turn 1: ['gather(B)', 'gather(C)']
turn 2: ['gather(B)', "done({'yes': True})"]
seen: ['B', 'C', 'B']
Expected behavior
Turn 2 runs only done. gather already ran for both inputs in turn 1 and emitted its output, so seen stays ['B', 'C'].
The same happens with ResumabilityConfig(is_resumable=True).
In a loop
Same Gather node (emitting on every second input), inside a loop that runs three laps and then asks for approval: START -> b -> gather, b -> c -> gather, gather -> again, again -> {"loop": b, "stop": approve}, approve -> done. After the approval is answered:
turn 1: b, gather(B), c, gather(C), again#1, ... again#3, approve (asks once)
turn 2: gather(B), gather(B), gather(B), again#4, done(...), approve (asks again)
turn 3: gather(B), then the run hangs
Each lap's no-output run of gather is re-executed. The extra B makes gather emit again, so again runs a 4th time and routes to approve once more.
Root cause
check_interception in src/google/adk/workflow/utils/_replay_interceptor.py, Case 6 (no output, route or interrupts recorded): a wait_for_output node always gets should_run = True, so a run that finished in the WAITING state is executed again on replay. That covers a run that is still waiting for more input, but it also covers a run that a later run of the same node (gather@2 above) has already superseded.
Simply fast-forwarding wait_for_output runs in Case 6 fixes the repro, but breaks test_workflow_task_mode_plain_text_resume_auto_routing. A task-mode agent is a wait_for_output node that really does need to run again to receive the next user message. So the check seems to need to tell apart:
- a
WAITINGrun that is still the node's latest run: rerun it, as today; - a
WAITINGrun followed by a later run of the same node in the recorded events: fast-forward it asWAITING, without executing it.
Questions
- Does that distinction match how you'd like this to work, or is there a better signal for "this waiting run was already superseded"?
- This sits next to Workflow: a node that returns None runs again when the workflow resumes after RequestInput #7666 / fix(workflow): record runs that finish without a result so resume replays them #7771, which records runs that finish without a result. fix(workflow): record runs that finish without a result so resume replays them #7771 deliberately leaves
wait_for_outputnodes out. Would you prefer this as a separate PR after that one, or folded into it?
I'm happy to open a PR with a regression test for both the simple and the loop case.
Desktop (please complete the following information):
- OS: macOS
- Python version: 3.12
- ADK version: main (8cd3b12, 2.13.0)
Model Information:
- Not model-related: function nodes only.
Activity
surajksharma07 commentedon Nov 2, 2026
@AkshayShah03 I could reproduce both the simple and the loop case on 2.13.0 and current main (extra lap, second approval, then the hang). It works fine on 2.4.0 and breaks from 2.5.0, so it looks like it came in with 9d306f5 when the wait_for_output rerun was widened beyond Workflow nodes for task mode.
Your “superseded run” idea looks right. As a workaround I marked recovered runs that have a later run of the same node (gather@1 vs gather@2) and fast-forwarded only those in Case 6 when they have no output/route/interrupts. That fixed both cases with and without resumability and the task-mode resume test still passed.
Could you check if that works on your side, including the fuzzer graphs that hit the replay divergence? Please test it well before raising the PR (simple + loop regression tests, task mode, resumable apps). It’s probably worth syncing with #7771 since it touches the same replay path.
AkshayShah03 commentedon Nov 2, 2026
Thanks for confirming and for tracking it to 9d306f5. I tried the superseded-run check, and it fixes both cases from the issue, but while testing I found a case it doesn't cover: a waiting run that is still the node's latest run when the user answers.
START -> A -> Gather,START -> Ask -> B -> Gather,Gather -> Done: Gather waits after A's output while Ask pauses. On resume, Gather@1 is still the latest run, so it reruns with A's output again, emits['a', 'a']to Done, and only then gets B's output. That fails on main and with the superseded check.What fixes all three for me is narrowing Case 6 back toward its behaviour before 9d306f5: rerun a
wait_for_outputnode with no recorded outcome only if it is an agent (task or chat mode, which waits for the user's next message) or hasrerun_on_resume. Any otherwait_for_outputnode falls through to the default Case 6 handling. Its next trigger starts a new run anyway, so a static run is fast-forwarded and a dynamic run still reruns. Branch: fix/wait-for-output-superseded-rerun (one file insrc, +11/−1).Testing:
rerun_on_resumenodes still rerun pass on both.test_workflow_task_mode_plain_text_resume_auto_routingand the rest of the task/chat-mode tests pass. The full unit suite shows the same 18 environment-only failures as main.wait_for_outputand dynamic nodes, with and without resumability): main hits 24 replay divergences, 0 with this change plus fix(workflow): record runs that finish without a result so resume replays them #7771, and no new failures.On syncing with #7771: the fuzzer's divergences also need one change there. The fuzzer's waiting node emits no event at all when it waits, so the run leaves nothing to replay and is treated as fresh. #7771 currently skips its completion marker for every
wait_for_outputnode. Narrowing that skip to agents andrerun_on_resumenodes (the ones that rerun anyway) closes it. Would you prefer that as a follow-up commit on #7771, or should the two go in as one PR?surajksharma07 commentedon Nov 4, 2026
Thanks @AkshayShah03, good catch on the still-waiting case. It shows the superseded check alone isn’t enough. Narrowing Case 6 to agents + rerun_on_resume looks right: it’s basically the pre-9d306f5 rule, and nested Workflows still rerun since they default to rerun_on_resume=True. Main is unchanged so this is still needed.
I couldn’t find fix/wait-for-output-superseded-rerun on your fork so haven’t been able to look at the diff yet. Could you push it or open it as a draft PR linked here?
For the #7771 side a follow-up commit there seems cleaner since it’s that PR’s marker logic. One thing to verify: a dynamic ctx.run_node gather run that gets the marker should still replay as WAITING so the caller isn’t handed a finished None.
Also a custom non-agent node that really waits on the user would now need rerun_on_resume=True so a line in the wait_for_output guide would help. Please run the tests and the fuzzer once more with both changes before review.