refactor:Refactor Runner to implement a Builder rather than a growing number of constructors

core/src/main/java/com/google/adk/agents/LlmAgent.java

@@ -604,7 +604,16 @@ private void maybeSaveOutputToState(Event event) {
 
   @Override
   protected Flowable<Event> runAsyncImpl(InvocationContext invocationContext) {
-    return llmFlow.run(invocationContext).doOnNext(this::maybeSaveOutputToState);
+    return llmFlow
+        .run(invocationContext)
+        .concatMap(
+            event -> {
+              this.maybeSaveOutputToState(event);
+              if (invocationContext.shouldPauseInvocation(event)) {
+                return Flowable.just(event).concatWith(Flowable.empty());
+              }
+              return Flowable.just(event);
+            });
   }
 
   @Override

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

@@ -156,36 +156,9 @@ protected Flowable<Event> postprocess(
     }
 
     return currentLlmResponse.flatMapPublisher(
-        updatedResponse -> {
-          Flowable<Event> processorEvents = Flowable.fromIterable(Iterables.concat(eventIterables));
-
-          if (updatedResponse.content().isEmpty()
-              && updatedResponse.errorCode().isEmpty()
-              && !updatedResponse.interrupted().orElse(false)
-              && !updatedResponse.turnComplete().orElse(false)) {
-            return processorEvents;
-          }
-
-          Event modelResponseEvent =
-              buildModelResponseEvent(baseEventForLlmResponse, llmRequest, updatedResponse);
-
-          Flowable<Event> modelEventStream = Flowable.just(modelResponseEvent);
-
-          if (modelResponseEvent.functionCalls().isEmpty()) {
-            return processorEvents.concatWith(modelEventStream);
-          }
-
-          Maybe<Event> maybeFunctionCallEvent;
-          if (context.runConfig().streamingMode() == StreamingMode.BIDI) {
-            maybeFunctionCallEvent =
-                Functions.handleFunctionCallsLive(context, modelResponseEvent, llmRequest.tools());
-          } else {
-            maybeFunctionCallEvent =
-                Functions.handleFunctionCalls(context, modelResponseEvent, llmRequest.tools());
-          }
-
-          return processorEvents.concatWith(modelEventStream).concatWith(maybeFunctionCallEvent);
-        });
+        updatedResponse ->
+            buildPostprocessingEvents(
+                updatedResponse, eventIterables, context, baseEventForLlmResponse, llmRequest));
   }
 
   /**
@@ -623,6 +596,45 @@ public void onError(Throwable e) {
    *
    * @return A fully constructed {@link Event} representing the LLM response.
    */
+  private Flowable<Event> buildPostprocessingEvents(
+      LlmResponse updatedResponse,
+      List<Iterable<Event>> eventIterables,
+      InvocationContext context,
+      Event baseEventForLlmResponse,
+      LlmRequest llmRequest) {
+    Flowable<Event> processorEvents = Flowable.fromIterable(Iterables.concat(eventIterables));
+    if (updatedResponse.content().isEmpty()
+        && updatedResponse.errorCode().isEmpty()
+        && !updatedResponse.interrupted().orElse(false)
+        && !updatedResponse.turnComplete().orElse(false)) {
+      return processorEvents;
+    }
+
+    Event modelResponseEvent =
+        buildModelResponseEvent(baseEventForLlmResponse, llmRequest, updatedResponse);
+    if (modelResponseEvent.functionCalls().isEmpty()) {
+      return processorEvents.concatWith(Flowable.just(modelResponseEvent));
+    }
+
+    Maybe<Event> maybeFunctionResponseEvent =
+        context.runConfig().streamingMode() == StreamingMode.BIDI
+            ? Functions.handleFunctionCallsLive(context, modelResponseEvent, llmRequest.tools())
+            : Functions.handleFunctionCalls(context, modelResponseEvent, llmRequest.tools());
+
+    Flowable<Event> functionEvents =
+        maybeFunctionResponseEvent.flatMapPublisher(
+            functionResponseEvent -> {
+              Optional<Event> toolConfirmationEvent =
+                  Functions.generateRequestConfirmationEvent(
+                      context, modelResponseEvent, functionResponseEvent);
+              return toolConfirmationEvent.isPresent()
+                  ? Flowable.just(toolConfirmationEvent.get(), functionResponseEvent)
+                  : Flowable.just(functionResponseEvent);
+            });
+
+    return processorEvents.concatWith(Flowable.just(modelResponseEvent)).concatWith(functionEvents);
+  }
+
   private Event buildModelResponseEvent(
       Event baseEventForLlmResponse, LlmRequest llmRequest, LlmResponse llmResponse) {
     Event.Builder eventBuilder =
@@ -640,10 +652,13 @@ private Event buildModelResponseEvent(
 
     Event event = eventBuilder.build();
 
+    logger.info("event: {} functionCalls: {}", event, event.functionCalls());
+
     if (!event.functionCalls().isEmpty()) {
       Functions.populateClientFunctionCallId(event);
       Set<String> longRunningToolIds =
           Functions.getLongRunningFunctionCalls(event.functionCalls(), llmRequest.tools());
+      logger.info("longRunningToolIds: {}", longRunningToolIds);
       if (!longRunningToolIds.isEmpty()) {
         event.setLongRunningToolIds(Optional.of(longRunningToolIds));
       }

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

@@ -0,0 +1,29 @@
+/*
+ * Copyright 2025 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.google.adk.flows.llmflows;
+
+/**
+ * An app contains Resumability configuration for the agents.
+ *
+ * @param isResumable Whether the app is resumable.
+ */
+public record ResumabilityConfig(boolean isResumable) {
+
+  /** Creates a new {@code ResumabilityConfig} with resumability disabled. */
+  public ResumabilityConfig() {
+    this(false);
+  }
+}