[Bug] ADK 2.8.0: repeated HITL deadlocks on replay within one invocation (no LLM)

[Breaknus]

Describe the Bug

On google-adk==2.8.0, a resumable Workflow using the documented review/revise HITL pattern deadlocks when approving after two rejected reviews. The error is a replay sequence barrier timeout on HITL@1.

This reproduces with only public FunctionNodes and InMemorySessionService: no LLM, network request, database, private API override, or SDK patch. The same App and Runner are used throughout. Interrupt IDs are unique, the review counter lives in native session state, and Runner infers the same invocation from the function responses.

The review function follows Feeding the answer back into a loop in the 2.8.0 HITL reference, including Event(output=response, route=..., state=...).

Steps to Reproduce

1. Use Python 3.12 with google-adk==2.8.0 installed. No API key or model configuration is needed.
2. Save the standalone code below as repro.py and run python repro.py.
3. The script answers the three successive native requests with approved=False, False, then True.
4. The third resume fails after the native 15-second replay barrier timeout.

Expected Behavior

Three unique review requests are followed by exactly one execution of Publish. The final native review_count should be 3, all within one invocation.

Observed Behavior

Requests review_0, review_1, and review_2 are emitted. On approval, Publish is never executed, and review_count remains 2:

RuntimeError: Replay divergence detected: Timed out waiting for sequence key 'HITL@1' to be unblocked.

The relevant stack is _workflow.py:_run_loop -> _workflow.py:return_ctx -> _replay_sequence_barrier.py:wait. The retained node paths before failure, with the common DocumentedHitl@1/ prefix omitted, are:

HITL@1
HITL@1
Revise@1
HITL@2
HITL@1
HITL@2
Revise@2
HITL@3

All recorded events with an invocation ID belong to a single invocation. The script prints the IDs and paths so this can be checked independently.

Environment Details

• ADK: 2.8.0
• Python: 3.12.12
• OS: macOS, arm64
• google-genai: 2.22.0
• pydantic: 2.13.4
• Session service: InMemorySessionService
• ResumabilityConfig(is_resumable=True)
• HITL node: rerun_on_resume=True

Model Information

• LiteLLM: No, not used by this reproduction.
• Model: N/A; no LLM calls.

Minimal Reproduction Code

"""Check the documented HITL loop with unchanged public ADK."""

import asyncio
import json
from pydantic import BaseModel
from google.adk import Context, Event, Workflow
from google.adk.apps import App, ResumabilityConfig
from google.adk.events import RequestInput
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.workflow import START, Edge, FunctionNode
from google.genai import types


class ApprovalSchema(BaseModel):
    """Describe the documented approval response."""

    approved: bool


async def main():
    """Run the documented loop using one persistent App and Runner."""
    published = []

    async def review(ctx: Context, node_input: object):
        """Run the review body from the pinned official HITL reference."""
        review_count = ctx.state.get("review_count", 0)
        interrupt_id = f"review_{review_count}"
        response = ctx.resume_inputs.get(interrupt_id)
        if response:
            yield Event(
                output=response,
                route="approved" if response.get("approved") else "rejected",
                state={"review_count": review_count + 1},
            )
            return
        yield RequestInput(
            interrupt_id=interrupt_id,
            message="Approve this plan?",
            response_schema=ApprovalSchema,
        )

    async def revise(ctx: Context, node_input: object):
        """Send the documented rejected route back to review."""
        return Event(route="review")

    async def publish(ctx: Context, node_input: object):
        """Record reaching the documented approved successor."""
        published.append(True)
        return "PUBLISHED"

    hitl = FunctionNode(name="HITL", func=review, rerun_on_resume=True)
    revision = FunctionNode(name="Revise", func=revise)
    publication = FunctionNode(name="Publish", func=publish)
    app = App(
        name="documented-hitl",
        root_agent=Workflow(
            name="DocumentedHitl",
            edges=[
                Edge(from_node=START, to_node=hitl),
                Edge(from_node=hitl, to_node=revision, route="rejected"),
                Edge(from_node=revision, to_node=hitl, route="review"),
                Edge(from_node=hitl, to_node=publication, route="approved"),
            ],
        ),
        resumability_config=ResumabilityConfig(is_resumable=True),
    )
    sessions = InMemorySessionService()
    await sessions.create_session(
        app_name=app.name, user_id="user", session_id="session"
    )
    runner = Runner(app=app, session_service=sessions)
    message = types.Content(role="user", parts=[types.Part(text="Review this plan")])
    request_ids = []
    try:
        for phase in range(4):
            events = [
                event
                async for event in runner.run_async(
                    user_id="user", session_id="session", new_message=message
                )
            ]
            calls = [
                part.function_call
                for event in events
                for part in (event.content.parts or [] if event.content else [])
                if part.function_call and part.function_call.name == "adk_request_input"
            ]
            print(
                json.dumps({"phase": phase, "requests": [c.id for c in calls]}),
                flush=True,
            )
            if phase < 3:
                assert len(calls) == 1
                call = calls[0]
                assert call.id not in request_ids
                request_ids.append(call.id)
                message = types.Content(
                    role="user",
                    parts=[
                        types.Part(
                            function_response=types.FunctionResponse(
                                id=call.id,
                                name="adk_request_input",
                                response={"approved": phase == 2},
                            )
                        )
                    ],
                )
        assert published == [True]
    except Exception as error:
        print(
            json.dumps(
                {
                    "exception_type": type(error).__name__,
                    "exception": str(error),
                    "published": len(published),
                }
            ),
            flush=True,
        )
        raise
    finally:
        session = await sessions.get_session(
            app_name=app.name, user_id="user", session_id="session"
        )
        print(
            json.dumps(
                {
                    "review_count": session.state.get("review_count"),
                    "invocations": sorted(
                        {e.invocation_id for e in session.events if e.invocation_id}
                    ),
                    "trace": [
                        e.node_info.path
                        for e in session.events
                        if e.node_info.path
                        and (
                            "/HITL@" in e.node_info.path
                            or "/Revise@" in e.node_info.path
                        )
                    ],
                }
            ),
            flush=True,
        )


asyncio.run(main())

Control experiment / limited workaround

Removing only output=response, from the review function's Event(...), while retaining route, the state update, is_resumable=True, and every other line, makes this exact script pass: exit 0, review_count=3, Publish exactly once. The original script exits 1 with the barrier timeout.

This is a narrow control and a possible public-API workaround when downstream code needs only routing/state. It is not a general workaround for nodes whose downstream consumers need their output, and it does not establish recovery of histories already containing the duplicate events.

Possible Cause

Source inspection suggests an interaction between two native behaviors:

• Workflow._maybe_reemit_replayed_output emits another output event for a fast-forwarded node in a resumable run.
• ReplayManager._scan_sequence removes an earlier occurrence of a terminal child key and appends the later one. On the next resume, this can place Revise@1 before its prerequisite HITL@1 in the barrier sequence.

The graph starts by fast-forwarding HITL@1, but its barrier is waiting for Revise@1, which cannot be scheduled until HITL completes. This matches the observed trace. I am reporting the behavior rather than proposing a general SDK patch: nested/dynamic ordering would need to be considered when choosing a fix.

Related issue / Regression

This appears distinct from #6497: that issue involved events from an earlier invocation. This reproduction uses one invocation, one Runner, unique interrupt IDs and no LLM. The current invocation filter is present in 2.8.0.

Earlier ADK versions have not been tested, so I am not claiming a version regression. The failure was reproduced in both the original small split-node graph and this documented loop; the exact standalone script above was rerun before filing.

[chelsealong]

Picking this one up now — opening a PR shortly. Flagging it here so nobody duplicates the work; if someone is already on it, say so and I will drop mine.