fix: Runner now includes Singles from appendEvent in the Rx graph

Previously, the return values from sessionManager.appendEvent were disarded, leading to issues if that Single did not represent an immediate value.  For example, if a SessionManager implementation performed async work such as persisting the new event to an external datastore that work would never occur.

PiperOrigin-RevId: 782076407

Author: google-genai-bot

core/src/main/java/com/google/adk/runner/Runner.java

@@ -38,6 +38,7 @@
 import io.reactivex.rxjava3.core.Flowable;
 import io.reactivex.rxjava3.core.Maybe;
 import io.reactivex.rxjava3.core.Single;
+import io.reactivex.rxjava3.core.Completable;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -83,7 +84,7 @@ public BaseSessionService sessionService() {
    *
    * @throws IllegalArgumentException if message has no parts.
    */
-  private void appendNewMessageToSession(
+  private Completable appendNewMessageToSession(
       Session session,
       Content newMessage,
       InvocationContext invocationContext,
@@ -123,7 +124,7 @@ private void appendNewMessageToSession(
             .author("user")
             .content(Optional.of(newMessage))
             .build();
-    this.sessionService.appendEvent(session, event);
+    return this.sessionService.appendEvent(session, event).ignoreElement();
   }
 
   /**
@@ -190,14 +191,20 @@ public Flowable<Event> runAsync(Session session, Content newMessage, RunConfig r
                         newMessage,
                         runConfig);
 
+                Completable newMessageCompletable = Completable.complete();
                 if (newMessage != null) {
-                  appendNewMessageToSession(
-                      sess, newMessage, invocationContext, runConfig.saveInputBlobsAsArtifacts());
+                  newMessageCompletable =
+                      appendNewMessageToSession(
+                          sess,
+                          newMessage,
+                          invocationContext,
+                          runConfig.saveInputBlobsAsArtifacts());
                 }
 
                 invocationContext.agent(this.findAgentToRun(sess, rootAgent));
                 Flowable<Event> events = invocationContext.agent().runAsync(invocationContext);
-                return events.doOnNext(event -> this.sessionService.appendEvent(sess, event));
+                return newMessageCompletable.andThen(
+                    events.concatMapSingle(event -> this.sessionService.appendEvent(sess, event)));
               })
           .doOnError(
               throwable -> {