fix: avoid StackOverflowError in PersistBarrier.awaitPersisted for large steps

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);