fix(flows): avoid StackOverflowError after many steps in BaseLlmFlow

BaseLlmFlow.run subscribed to the next step from inside the previous
step's completion, through concatWith, toList, flatMapPublisher and
PersistBarrier.awaitPersisted(...).andThen(run(...)). When a step
completed on the subscribing thread, as it does with the core models
and the session services, each step ran 23 stack frames deeper than
the one before. With a 1 MB thread stack, the default on x86-64, an
agent that kept calling tools failed with StackOverflowError after
roughly 250-300 LLM calls, before the default maxLlmCalls of 500
could end the run, and the error never reached onError.

Drive the steps with repeatUntil instead, as LoopAgent does on its
resumable path. repeatUntil resubscribes from a loop when the source
completes synchronously, so every step starts at the same stack
depth, like the while loop in adk-python's BaseLlmFlow.run_async.
The end-of-flow checks, the pause on a pending long-running call,
the PersistBarrier wait between steps and maxSteps are unchanged.

Author: innoprej

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

@@ -60,6 +60,8 @@
 import java.util.List;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiFunction;
 import org.slf4j.Logger;
@@ -511,11 +513,37 @@ private Flowable<Event> runOneStep(Context spanContext, InvocationContext contex
    */
   @Override
   public Flowable<Event> run(InvocationContext invocationContext) {
-    return run(Context.current(), invocationContext, 0);
+    Context spanContext = Context.current();
+    return Flowable.defer(
+        () -> {
+          AtomicInteger stepsCompleted = new AtomicInteger(0);
+          AtomicBoolean continueFlow = new AtomicBoolean(false);
+          // Run one step per subscription and let repeatUntil subscribe again until the flow ends.
+          // Subscribing to the next step from the previous step's completion would nest the steps
+          // on the stack whenever a step completes synchronously, as it does with the built-in
+          // models.
+          return Flowable.defer(
+                  () -> {
+                    continueFlow.set(false);
+                    return runStep(
+                        spanContext,
+                        invocationContext,
+                        stepsCompleted.getAndIncrement(),
+                        continueFlow);
+                  })
+              .repeatUntil(() -> !continueFlow.get());
+        });
   }
 
-  private Flowable<Event> run(
-      Context spanContext, InvocationContext invocationContext, int stepsCompleted) {
+  /**
+   * Runs {@link #runOneStep} once. If the flow should run another step, sets {@code continueFlow}
+   * and waits until the Runner has persisted this step's events.
+   */
+  private Flowable<Event> runStep(
+      Context spanContext,
+      InvocationContext invocationContext,
+      int stepsCompleted,
+      AtomicBoolean continueFlow) {
     Flowable<Event> currentStepEvents = runOneStep(spanContext, invocationContext).cache();
     if (stepsCompleted + 1 >= maxSteps) {
       logger.debug("Ending flow execution because max steps reached.");
@@ -543,10 +571,10 @@ private Flowable<Event> run(
                     return Flowable.empty();
                   } else {
                     logger.debug("Continuing to next step of the flow.");
+                    continueFlow.set(true);
                     // Wait until the Runner has persisted this step's events so the next step's
                     // request is not built from a stale session (see PersistBarrier).
-                    return PersistBarrier.awaitPersisted(invocationContext, eventList)
-                        .andThen(run(spanContext, invocationContext, stepsCompleted + 1));
+                    return PersistBarrier.awaitPersisted(invocationContext, eventList).toFlowable();
                   }
                 }));
   }

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

@@ -35,6 +35,7 @@
 import com.google.adk.events.Event;
 import com.google.adk.flows.llmflows.RequestProcessor.RequestProcessingResult;
 import com.google.adk.flows.llmflows.ResponseProcessor.ResponseProcessingResult;
+import com.google.adk.models.LlmCallsLimitExceededException;
 import com.google.adk.models.LlmRequest;
 import com.google.adk.models.LlmResponse;
 import com.google.adk.testing.TestLlm;
@@ -59,6 +60,7 @@
 import io.reactivex.rxjava3.core.Maybe;
 import io.reactivex.rxjava3.core.Single;
 import io.reactivex.rxjava3.schedulers.Schedulers;
+import io.reactivex.rxjava3.subscribers.TestSubscriber;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -191,6 +193,32 @@ public void run_withFunctionCallsAndMaxSteps_stopsAfterMaxSteps() {
         Content.fromParts(Part.fromFunctionResponse("my_function", testResponse)));
   }
 
+  @Test
+  public void run_modelRespondingOnSubscribingThread_reachesMaxLlmCallsWithoutStackOverflow() {
+    // The built-in models emit a non-streaming response on the subscribing thread, as
+    // Flowable.just does here. Running the steps must not nest one step's subscription inside the
+    // previous step's completion, or the stack overflows long before maxLlmCalls is reached.
+    int maxLlmCalls = 2_000;
+    Content contentWithFunctionCall =
+        Content.fromParts(Part.fromFunctionCall("my_function", ImmutableMap.of("arg1", "value1")));
+    TestLlm testLlm =
+        createTestLlm(() -> Flowable.just(createLlmResponse(contentWithFunctionCall)));
+    ImmutableMap<String, Object> testResponse =
+        ImmutableMap.<String, Object>of("response", "response for my_function");
+    InvocationContext invocationContext =
+        createInvocationContext(
+            createTestAgentBuilder(testLlm)
+                .tools(ImmutableList.of(new TestTool("my_function", testResponse)))
+                .build(),
+            RunConfig.builder().maxLlmCalls(maxLlmCalls).build());
+    BaseLlmFlow baseLlmFlow = createBaseLlmFlowWithoutProcessors();
+
+    TestSubscriber<Event> subscriber = baseLlmFlow.run(invocationContext).test();
+
+    subscriber.assertError(LlmCallsLimitExceededException.class);
+    assertThat(testLlm.getRequests()).hasSize(maxLlmCalls);
+  }
+
   @Test
   public void run_withLongRunningFunctionCall_returnsCorrectEventsWithLongRunningToolIds() {
     Content firstContent =