fix: avoid StackOverflowError in PersistBarrier.awaitPersisted for large steps

awaitPersisted folded a step's per-event barriers into a left-nested
Completable.andThen(...) chain. A single step can carry many events (an agent
transfer folds the sub-agent's events into the parent step), and the nested
chain recurses once per element on both subscription and completion, throwing
StackOverflowError for long sessions.

Await the barriers with Completable.concat instead, which drains its sources
iteratively and stays stack-safe while preserving the same await semantics
(order is irrelevant, as each subject retains its terminal state).

Adds a regression test that awaits 50k events in a single step.

Author: zhangxiangnan

core/src/main/java/com/google/adk/flows/llmflows/PersistBarrier.java

@@ -21,6 +21,7 @@
 import com.google.common.annotations.VisibleForTesting;
 import io.reactivex.rxjava3.core.Completable;
 import io.reactivex.rxjava3.subjects.CompletableSubject;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -72,14 +73,21 @@ public static Completable awaitPersisted(InvocationContext context, List<Event>
     if (enabled == null || !enabled) {
       return Completable.complete();
     }
-    Completable result = Completable.complete();
+    // Await the per-event barriers via Completable.concat instead of folding them into a
+    // left-nested chain of Completable.andThen(...). A single step's event list can be large (an
+    // agent transfer folds the sub-agent's events into the parent step), and a nested andThen chain
+    // recurses once per element on both subscription and completion, overflowing the stack for
+    // long sessions. Completable.concat drains its sources iteratively, so it stays stack-safe
+    // while keeping the same await semantics (order is irrelevant, as each subject retains its
+    // terminal state).
+    List<Completable> barriers = new ArrayList<>(events.size());
     for (Event event : events) {
       String eventId = event.id();
       if (eventId != null) {
-        result = result.andThen(barrier(context, eventId));
+        barriers.add(barrier(context, eventId));
       }
     }
-    return result;
+    return Completable.concat(barriers);
   }
 
   /** Signals that the {@code Runner} persisted the event with the given id. */

core/src/test/java/com/google/adk/flows/llmflows/PersistBarrierTest.java

@@ -112,6 +112,29 @@ public void multiEventStep_completesOnlyAfterAllMarked() {
     assertThat(PersistBarrier.pendingCount(context)).isEqualTo(0);
   }
 
+  @Test
+  public void largeStep_awaitsAllWithoutStackOverflow() {
+    // A single step's event list can be large (e.g. an agent transfer folds the sub-agent's events
+    // into the parent step). Awaiting them must not build a deeply nested chain that overflows the
+    // stack when the returned Completable is subscribed or completed.
+    PersistBarrier.enable(context);
+    int eventCount = 50_000;
+    List<Event> events = new ArrayList<>(eventCount);
+    for (int i = 0; i < eventCount; i++) {
+      events.add(event("e" + i));
+    }
+
+    TestObserver<Void> observer = PersistBarrier.awaitPersisted(context, events).test();
+    observer.assertNotComplete();
+
+    for (Event event : events) {
+      PersistBarrier.markPersisted(context, event.id());
+    }
+
+    observer.assertComplete();
+    assertThat(PersistBarrier.pendingCount(context)).isEqualTo(0);
+  }
+
   @Test
   public void markFailedBeforeAwait_awaitFails() {
     PersistBarrier.enable(context);