fix(workflow): Prevent replay divergence hang in sequence barrier

The wait method in ReplaySequenceBarrier silently blocked forever when
the workflow replayed with divergent history. Added a 15-second timeout
to loudly raise a RuntimeError instead.

Co-authored-by: Shangjie Chen <deanchen@google.com>
PiperOrigin-RevId: 930765155

Author: DeanChensj

src/google/adk/workflow/utils/_replay_sequence_barrier.py

@@ -22,8 +22,9 @@
 class ReplaySequenceBarrier:
   """Unified chronological sequence barrier to ensure deterministic replay ordering."""
 
-  def __init__(self, sequence: list[str]) -> None:
+  def __init__(self, sequence: list[str], timeout_sec: float = 15.0) -> None:
     self.sequence = sequence
+    self.timeout_sec = timeout_sec
     self.current_index = 0
     self.events = {key: asyncio.Event() for key in sequence}
     if sequence:
@@ -37,7 +38,15 @@ async def wait(self, key: str) -> None:
     output are not in the sequence barrier, so they fast-forward immediately.
     """
     if key in self.events:
-      await self.events[key].wait()
+      try:
+        await asyncio.wait_for(
+            self.events[key].wait(), timeout=self.timeout_sec
+        )
+      except asyncio.TimeoutError:
+        raise RuntimeError(
+            "Replay divergence detected: Timed out waiting for sequence key"
+            f" '{key}' to be unblocked."
+        )
 
   def check_and_advance(self, key: str) -> None:
     """Advance the sequence if the key matches the current expected execution."""

tests/unittests/workflow/utils/test_replay_sequence_barrier.py

@@ -94,3 +94,15 @@ async def test_barrier_wait_non_existent_key():
 
   # No blocks, successfully completes!
   assert True
+
+
+@pytest.mark.asyncio
+async def test_barrier_wait_timeout_on_divergence():
+  """Verifies that waiting on a blocked key raises RuntimeError due to timeout."""
+  # We use a short sequence where NodeB is never unblocked
+  sequence = ['NodeA@1', 'NodeB@1']
+  # Use a fast timeout to keep the test rapid without mocking standard library functions
+  barrier = ReplaySequenceBarrier(sequence, timeout_sec=0.01)
+
+  with pytest.raises(RuntimeError, match='Replay divergence detected'):
+    await barrier.wait('NodeB@1')