Answering parallel RequestInput pauses in one message deadlocks replay: ReplaySequenceBarrier drops out-of-order completions

🔴 Required Information

Describe the Bug:
A resumable Workflow with two parallel RequestInput pauses fails when both are answered in one message. The resume raises WorkflowInvariantError: Replay divergence detected: Timed out waiting for sequence key 'step2@1' to be unblocked. after the 15 s barrier timeout.

The workflow forks into three branches that meet at a JoinNode: a chain of three nodes, a pause after one node, and a pause right after the fork. Pauses are FunctionNodes with rerun_on_resume=True.

Steps to Reproduce:

pip install google-adk==2.11.0
Run the script below. It answers both pending requests in one message.

Expected Behavior:

pending: ['confirm', 'approve']
pending: []
completed

Observed Behavior:

pending: ['confirm', 'approve']
google.adk.workflow._errors.WorkflowInvariantError: Replay divergence detected: Timed out waiting for sequence key 'step2@1' to be unblocked.

It does not fail with a single pause, or with a chain of one or two nodes. Whether it fails depends on the order of the fan-out branches.

Environment Details:

ADK Library Version: 2.11.0 (also reproduced on main at bca4a42)
Desktop OS: macOS 15.7.3
Python Version: 3.13.9

Model Information:

Are you using LiteLLM: No
Which model is being used: N/A (function nodes only, no LLM)

🟡 Optional Information

Minimal Reproduction Code:

import asyncio

from google.adk.events.event import Event
from google.adk.events.request_input import RequestInput
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.workflow import START, FunctionNode, JoinNode, Workflow
from google.genai import types


def node(name):
    def run(node_input: dict):
        return node_input
    return FunctionNode(func=run, name=name)


def pause(name):
    async def ask(ctx, node_input: dict):
        if name in ctx.resume_inputs:
            yield Event(output=node_input)
            return
        yield RequestInput(interrupt_id=name, message=f"{name}?")
    return FunctionNode(func=ask, name=name, rerun_on_resume=True)


fork, prepare, join = node("fork"), node("prepare"), JoinNode(name="join")
approve, confirm = pause("approve"), pause("confirm")
step0, step1, step2 = node("step0"), node("step1"), node("step2")
workflow = Workflow(name="root", edges=[
    (START, fork), (fork, (step0, prepare, confirm)),
    (step0, step1), (step1, step2), (step2, join),   # A chain of three nodes.
    (prepare, approve), (approve, join),             # A pause after one node.
    (confirm, join),                                 # A pause right after the fork.
])


async def main():
    sessions = InMemorySessionService()
    runner = Runner(app_name="repro", node=workflow, session_service=sessions)
    session = await sessions.create_session(app_name="repro", user_id="user")
    message = types.Content(role="user", parts=[types.Part(text='{"x": 0}')])
    while True:
        pending = []
        async for event in runner.run_async(user_id="user", session_id=session.id, new_message=message):
            if event.content:
                pending += [part.function_call.id for part in event.content.parts or [] if part.function_call]
        print("pending:", pending)
        if not pending:
            print("completed")
            return
        # Answer every pending request in one message.
        message = types.Content(role="user", parts=[types.Part(function_response=types.FunctionResponse(
            id=interrupt_id, name="adk_request_input", response={"ok": True})) for interrupt_id in pending])


asyncio.run(main())

Possible cause:
On resume, ReplayManager._scan_sequence records the completion order of the root's children from the first run:

['fork@1', 'step0@1', 'prepare@1', 'confirm@1', 'step1@1', 'approve@1', 'step2@1']

During replay, children are handled in two different ways (workflow/_dynamic_node_scheduler.py):

A fast-forwarded child goes through _check_existing_run, which calls wait_sequence before returning, so it waits for its turn.
A child that must run again (here the two answered pauses, via the "Rerun!" branch) does not wait. It runs live, and when it finishes, the scheduler calls advance_sequence.

ReplaySequenceBarrier.check_and_advance only advances when the key is the one currently expected, and otherwise ignores it. Instrumenting the barrier shows:

advance confirm@1   ok
advance approve@1   ignored (expected step1@1)
advance step1@1     ok
wait    step2@1     -> times out; approve@1 never advances again

So a rerun child that finishes ahead of its recorded position loses its completion, and the barrier never reaches the next key. Nested Workflow nodes take the same rerun path, so the same timeout also appears when such a node finishes out of order.

Possible fix:
Keep completions that arrive early, and apply them when their turn comes. With this patch applied, the script above completes:

def check_and_advance(self, key: str) -> None:
  early = self.__dict__.setdefault("early", set())
  if key in self.sequence[self.current_index:]:
    early.add(key)
  while self.current_index < len(self.sequence) and self.sequence[self.current_index] in early:
    early.discard(self.sequence[self.current_index])
    self.current_index += 1
    if self.current_index < len(self.sequence):
      self.events[self.sequence[self.current_index]].set()

An alternative is to have rerun children wait on the barrier before they advance it.

Follow-up: a single pause is enough to trigger this. The original report answers two parallel pauses in one message, but the same deadlock happens with one RequestInput when the parallel branches are sub-workflows.

Below, one branch is a two-node sub-workflow followed by a node, and the other is a two-node sub-workflow whose first node pauses once. Answering that single request fails the same way:

pending: ['approve']
google.adk.workflow._errors.WorkflowInvariantError: Replay divergence detected: Timed out waiting for sequence key 'report@1' to be unblocked.

Expected:

pending: ['approve']
pending: []
completed

Reproduced on 2.11.0 and on main at bca4a42 (InMemorySessionService, no LLM). Replacing either sub-workflow with a plain node makes it pass. The check_and_advance patch from the original report also makes it pass.

So batching answers is not required. Nested Workflow nodes also take the rerun path during replay, so they advance the barrier without waiting. When one finishes ahead of its recorded position, its completion is dropped, as in the original trace.

import asyncio

from google.adk.events.event import Event
from google.adk.events.request_input import RequestInput
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.workflow import START, FunctionNode, JoinNode, Workflow
from google.genai import types


def node(name):
    def run(node_input: dict):
        return node_input
    return FunctionNode(func=run, name=name)


async def ask(ctx, node_input: dict):
    if "approve" in ctx.resume_inputs:
        yield Event(output=node_input)
        return
    yield RequestInput(interrupt_id="approve", message="approve?")


# Two parallel sub-workflows of two nodes each; one pauses once, the other is followed by a node.
lint, test = node("lint"), node("test")
checks = Workflow(name="checks", edges=[(START, lint), (lint, test)])
gate = FunctionNode(func=ask, name="gate", rerun_on_resume=True)
review = Workflow(name="review", edges=[(START, gate), (gate, node("record"))])
fork, report, join = node("fork"), node("report"), JoinNode(name="join")
workflow = Workflow(name="root", edges=[
    (START, fork), (fork, (checks, review)),
    (checks, report), (report, join),  # One more node after the sub-workflow.
    (review, join),
])


async def main():
    sessions = InMemorySessionService()
    runner = Runner(app_name="repro", node=workflow, session_service=sessions)
    session = await sessions.create_session(app_name="repro", user_id="user")
    message = types.Content(role="user", parts=[types.Part(text='{"x": 0}')])
    while True:
        pending = []
        async for event in runner.run_async(user_id="user", session_id=session.id, new_message=message):
            if event.content:
                pending += [part.function_call.id for part in event.content.parts or [] if part.function_call]
        print("pending:", pending)
        if not pending:
            print("completed")
            return
        message = types.Content(role="user", parts=[types.Part(function_response=types.FunctionResponse(
            id=pending[0], name="adk_request_input", response={"ok": True}))])


asyncio.run(main())

I reproduced the replay deadlock and can take this issue. My proposed fix is to buffer child completions that arrive before their recorded turn, then advance the barrier through any buffered completions once they reach the head of the sequence. I would add focused barrier tests and a workflow regression test for the single-pause nested-workflow case. Please confirm this approach is acceptable before I start the patch.

@yifanxiong272

Thanks @yifanxiong272 for the detailed report and follow-up. I've reproduced the single-pause nested-workflow case on google-adk==2.11.0 (timeout on report@1), so this is confirmed as a bug in ReplaySequenceBarrier.

@pei711 your approach looks right, buffer early completions and advance through them once they reach the head of the sequence. Please go ahead and open a PR with tests covering both repros.

Thanks.

fix(workflow): buffer out-of-order completions in ReplaySequenceBarrier

Navigation Menu

Repository navigation

Answering parallel RequestInput pauses in one message deadlocks replay: ReplaySequenceBarrier drops out-of-order completions #7502

🔴 Required Information

🟡 Optional Information

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Answering parallel RequestInput pauses in one message deadlocks replay: ReplaySequenceBarrier drops out-of-order completions #7502

Description

🔴 Required Information

🟡 Optional Information

Activity

yifanxiong272 commented on Oct 11, 2026

pei711 commented on Oct 12, 2026

sanketpatil06 commented on Oct 13, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions