fix: subscribe to appendEvent Single in runLive

PiperOrigin-RevId: 943748488

Author: google-genai-bot

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

@@ -791,7 +791,8 @@ protected Flowable<Event> runLiveImpl(
                       updatedInvocationContext
                           .agent()
                           .runLive(updatedInvocationContext)
-                          .doOnNext(event -> this.sessionService.appendEvent(session, event)))
+                          .concatMapSingle(
+                              event -> this.sessionService.appendEvent(session, event)))
               .doOnError(
                   throwable -> {
                     Span span = Span.current();

core/src/test/java/com/google/adk/runner/RunnerTest.java

@@ -1620,6 +1620,44 @@ public void runLive_success() throws Exception {
     assertThat(simplifyEvents(testSubscriber.values())).containsExactly("test agent: from llm");
   }
 
+  @Test
+  public void runLive_asyncSessionService_persistsEvents() throws Exception {
+    BaseSessionService asyncSessionService =
+        new AppendDelayingSessionService(new InMemorySessionService(), 10);
+    Runner runnerWithAsyncSession =
+        Runner.builder()
+            .app(App.builder().name("test").rootAgent(agent).build())
+            .sessionService(asyncSessionService)
+            .build();
+    Session asyncSession =
+        runnerWithAsyncSession.sessionService().createSession("test", "user").blockingGet();
+
+    LiveRequestQueue liveRequestQueue = new LiveRequestQueue();
+    TestSubscriber<Event> testSubscriber =
+        runnerWithAsyncSession
+            .runLive(asyncSession, liveRequestQueue, RunConfig.builder().build())
+            .test();
+
+    liveRequestQueue.content(createContent("from user"));
+    liveRequestQueue.close();
+
+    testSubscriber.await();
+    testSubscriber.assertComplete();
+    assertThat(simplifyEvents(testSubscriber.values())).containsExactly("test agent: from llm");
+
+    // Verify that the events are successfully persisted to session history.
+    ImmutableList<Event> history =
+        runnerWithAsyncSession
+            .sessionService()
+            .listEvents("test", "user", asyncSession.id())
+            .blockingGet()
+            .events();
+    // The history should contain only the agent response event (user messages in liveQueue are not
+    // persisted).
+    assertThat(history).hasSize(1);
+    assertThat(history.get(0).author()).isEqualTo("test agent");
+  }
+
   @Test
   public void runLive_withSessionKey_success() throws Exception {
     LiveRequestQueue liveRequestQueue = new LiveRequestQueue();