From 3c0c51371a7e2d22b62b414386e4ae9cda2a1753 Mon Sep 17 00:00:00 2001 From: chelsealong Date: Sat, 5 Sep 2026 21:02:12 +0000 Subject: [PATCH] fix(workflow): stop a re-emitted output echo from reordering the replay sequence MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A resumable HITL loop (review -> revise -> review -> ...) can deadlock on a later resume with "Replay divergence detected: Timed out waiting for sequence key '@1' to be unblocked." Workflow._maybe_reemit_replayed_output resurfaces a fast-forwarded node's output again on each turn so a resumable stream stays complete. That resurfaced event carries the same node path/run id as the node's original completion. ReplayManager._scan_sequence treated it as a brand-new completion and moved the node past whatever ran in between (e.g. its own downstream child), corrupting the barrier's expected order into a cycle the next resume can never satisfy. _scan_sequence now tracks which child run ids have already reached a genuine terminal completion (output, route, or error) and ignores further terminal events for that same id, since a completed run id cannot complete again — a repeat is always this re-emitted echo. Fixes #7027 --- .../adk/workflow/utils/_replay_manager.py | 16 +++++++ .../workflow/utils/test_replay_manager.py | 46 +++++++++++++++++++ 2 files changed, 62 insertions(+) diff --git a/src/google/adk/workflow/utils/_replay_manager.py b/src/google/adk/workflow/utils/_replay_manager.py index de982f5d7ca..5ba50637f54 100644 --- a/src/google/adk/workflow/utils/_replay_manager.py +++ b/src/google/adk/workflow/utils/_replay_manager.py @@ -241,6 +241,7 @@ def _scan_sequence( """Extract chronological child completion sequence under base_path.""" base_path_builder = _NodePathBuilder.from_string(base_path) sequence: list[str] = [] + completed: set[str] = set() invocation_id = ctx._invocation_context.invocation_id for event in events: @@ -265,9 +266,24 @@ def _scan_sequence( segment: str = child_path.leaf_segment if is_terminal_event(event): + # Once a child has genuinely completed (output, route, or error), its + # run id cannot complete again: a further terminal event for the same + # segment is a resurfaced echo of that completion (see + # Workflow._maybe_reemit_replayed_output), not a new one. Repositioning + # the segment for that echo can shift it past a sibling that + # legitimately completed in between, corrupting the barrier order and + # deadlocking a resumed loop on itself. + if segment in completed: + continue if segment in sequence: sequence.remove(segment) sequence.append(segment) + if ( + event.output is not None + or (event.actions and event.actions.route is not None) + or event.error_code is not None + ): + completed.add(segment) return sequence diff --git a/tests/unittests/workflow/utils/test_replay_manager.py b/tests/unittests/workflow/utils/test_replay_manager.py index 52977fde335..f77a9badfdf 100644 --- a/tests/unittests/workflow/utils/test_replay_manager.py +++ b/tests/unittests/workflow/utils/test_replay_manager.py @@ -19,6 +19,7 @@ from google.adk.events.event import Event from google.adk.events.event import NodeInfo +from google.adk.events.event_actions import EventActions from google.adk.workflow.utils._replay_manager import ReplayManager import pytest @@ -445,6 +446,51 @@ async def test_scan_workflow_events_sequence_empty_when_all_events_are_prior(): await asyncio.wait_for(mgr.sequence_barrier.wait("anything"), timeout=1) +def test_scan_workflow_events_sequence_ignores_reemitted_completion_echo(): + """A fast-forwarded node's resurfaced output must not reorder the sequence. + + A HITL loop (review -> revise -> review -> ...) genuinely completes + `hitl@1` then `revise@1`. On a later turn within the same invocation, + Workflow._maybe_reemit_replayed_output resurfaces hitl@1's output again + (a duplicate echo, since a completed run id cannot complete twice) so a + resumable stream stays complete. Before the fix, `_scan_sequence` treated + that echo as a new completion and moved `hitl@1` after `revise@1`, even + though `revise@1` can only run after `hitl@1` -- corrupting the barrier + into a cycle that deadlocks the next resume (issue #7027). + """ + mgr = ReplayManager() + hitl_1 = Event( + author="node", + node_info=NodeInfo(path="wf@1/hitl@1", run_id="1"), + invocation_id="inv-1", + output="rejected_1", + actions=EventActions(route="rejected"), + ) + revise_1 = Event( + author="node", + node_info=NodeInfo(path="wf@1/revise@1", run_id="1"), + invocation_id="inv-1", + actions=EventActions(route="review"), + ) + hitl_1_echo = Event( + author="node", + node_info=NodeInfo(path="wf@1/hitl@1", run_id="1"), + invocation_id="inv-1", + output="rejected_1", + ) + + ctx = MagicMock() + ctx._invocation_context = MagicMock() + ctx._invocation_context.invocation_id = "inv-1" + ctx._invocation_context.session = MagicMock() + ctx._invocation_context.session.events = [hitl_1, revise_1, hitl_1_echo] + ctx.node_path = "wf@1" + + _, sequence = mgr.scan_workflow_events(ctx) + + assert sequence == ["hitl@1", "revise@1"] + + def _recorded_two_step_ctx(): """A ctx whose session records alpha completing before beta.""" alpha = Event(