Workflow(max_concurrency=1) fails on resume with "Replay divergence detected: Timed out waiting for sequence key": a replayed node waits for its turn while holding the only concurrency slot #7618

"""A Workflow with max_concurrency=1 deadlocks on its second resume: a loop asks twice for input beside a
three-node sibling chain. On replay, a completed node waits on the ReplaySequenceBarrier for its recorded
turn while holding the only concurrency slot, so the node recorded before it is never scheduled and the
barrier times out after 15 s. With max_concurrency=None or 2 both requests are answered and the run completes.

    python adk-max-concurrency-replay-repro.py 1|2|none
"""
import asyncio
import sys
import time
from importlib.metadata import version

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

ROUNDS = 2


def start(node_input: dict):
    return {"round": 0}


async def ask(ctx, node_input: dict):
    ctx.state["round"] = node_input["round"]
    yield RequestInput(interrupt_id=f"ask-{node_input['round']}", message="approve?")


def answered(ctx, node_input):
    return {"round": ctx.state["round"]}


def again(node_input: dict):
    state = {"round": node_input["round"] + 1}
    return Event(route=state["round"] < ROUNDS, output=state)


def other(node_input: dict):
    return node_input


SIBLING = 3  # Two plain nodes in a row are not enough.
nodes = {name: FunctionNode(func=func, name=name) for name, func in
         (("start", start), ("answered", answered), ("again", again))}
chain = [FunctionNode(func=other, name=f"other{i}") for i in range(SIBLING)]
asking = FunctionNode(func=ask, name="ask", rerun_on_resume=False)
join = JoinNode(name="join")
workflow = Workflow(
    name="root",
    edges=[(START, (nodes["start"], chain[0])),  # A loop that asks twice, beside a chain of three nodes.
           (nodes["start"], asking), (asking, nodes["answered"]), (nodes["answered"], nodes["again"]),
           (nodes["again"], {True: asking, False: join}), *zip(chain, chain[1:]), (chain[-1], join)],
    max_concurrency=None if sys.argv[1] == "none" else int(sys.argv[1]),
)


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="{}")])
    began = time.monotonic()
    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]
        if not pending:
            print(f"completed in {time.monotonic() - began:.1f} s")
            return
        print("answering", pending)
        message = types.Content(role="user", parts=[types.Part(function_response=types.FunctionResponse(
            id=call, name="adk_request_input", response={"ok": True})) for call in pending])


print(f"google-adk {version('google-adk')}, max_concurrency={sys.argv[1]}")
asyncio.run(main())