Remove blocking calls from Runner and GcsArtifactService

PiperOrigin-RevId: 776723027

Author: google-genai-bot

core/src/main/java/com/google/adk/artifacts/GcsArtifactService.java

@@ -75,62 +75,63 @@ private String getBlobName(
   @Override
   public Single<Integer> saveArtifact(
       String appName, String userId, String sessionId, String filename, Part artifact) {
-    ImmutableList<Integer> versions =
-        listVersions(appName, userId, sessionId, filename).blockingGet();
-    int nextVersion = versions.isEmpty() ? 0 : max(versions) + 1;
-
-    String blobName = getBlobName(appName, userId, sessionId, filename, nextVersion);
-    BlobId blobId = BlobId.of(bucketName, blobName);
-
-    BlobInfo blobInfo =
-        BlobInfo.newBuilder(blobId)
-            .setContentType(artifact.inlineData().get().mimeType().orElse(null))
-            .build();
-
-    try {
-      byte[] dataToSave =
-          artifact
-              .inlineData()
-              .get()
-              .data()
-              .orElseThrow(
-                  () -> new IllegalArgumentException("Saveable artifact data must be non-empty."));
-      storageClient.create(blobInfo, dataToSave);
-      return Single.just(nextVersion);
-    } catch (StorageException e) {
-      throw new VerifyException("Failed to save artifact to GCS", e);
-    }
+    return listVersions(appName, userId, sessionId, filename)
+        .map(versions -> versions.isEmpty() ? 0 : max(versions) + 1)
+        .map(
+            nextVersion -> {
+              String blobName = getBlobName(appName, userId, sessionId, filename, nextVersion);
+              BlobId blobId = BlobId.of(bucketName, blobName);
+
+              BlobInfo blobInfo =
+                  BlobInfo.newBuilder(blobId)
+                      .setContentType(artifact.inlineData().get().mimeType().orElse(null))
+                      .build();
+
+              try {
+                byte[] dataToSave =
+                    artifact
+                        .inlineData()
+                        .get()
+                        .data()
+                        .orElseThrow(
+                            () ->
+                                new IllegalArgumentException(
+                                    "Saveable artifact data must be non-empty."));
+                storageClient.create(blobInfo, dataToSave);
+                return nextVersion;
+              } catch (StorageException e) {
+                throw new VerifyException("Failed to save artifact to GCS", e);
+              }
+            });
   }
 
   @Override
   public Maybe<Part> loadArtifact(
       String appName, String userId, String sessionId, String filename, Optional<Integer> version) {
-    int versionToLoad;
-    if (version.isPresent()) {
-      versionToLoad = version.get();
-    } else {
-      ImmutableList<Integer> versions =
-          listVersions(appName, userId, sessionId, filename).blockingGet();
-      if (versions.isEmpty()) {
-        return Maybe.empty();
-      }
-      versionToLoad = max(versions);
-    }
-
-    String blobName = getBlobName(appName, userId, sessionId, filename, versionToLoad);
-    BlobId blobId = BlobId.of(bucketName, blobName);
-
-    try {
-      Blob blob = storageClient.get(blobId);
-      if (blob == null || !blob.exists()) {
-        return Maybe.empty();
-      }
-      byte[] data = blob.getContent();
-      String mimeType = blob.getContentType();
-      return Maybe.just(Part.fromBytes(data, mimeType));
-    } catch (StorageException e) {
-      return Maybe.empty();
-    }
+    return version
+        .map(Maybe::just)
+        .orElseGet(
+            () ->
+                listVersions(appName, userId, sessionId, filename)
+                    .flatMapMaybe(
+                        versions -> versions.isEmpty() ? Maybe.empty() : Maybe.just(max(versions))))
+        .flatMap(
+            versionToLoad -> {
+              String blobName = getBlobName(appName, userId, sessionId, filename, versionToLoad);
+              BlobId blobId = BlobId.of(bucketName, blobName);
+
+              try {
+                Blob blob = storageClient.get(blobId);
+                if (blob == null || !blob.exists()) {
+                  return Maybe.empty();
+                }
+                byte[] data = blob.getContent();
+                String mimeType = blob.getContentType();
+                return Maybe.just(Part.fromBytes(data, mimeType));
+              } catch (StorageException e) {
+                return Maybe.empty();
+              }
+            });
   }
 
   @Override

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

@@ -269,14 +269,17 @@ public Flowable<Event> runLive(
 
   public Flowable<Event> runLive(
       String userId, String sessionId, LiveRequestQueue liveRequestQueue, RunConfig runConfig) {
-    Session session =
-        this.sessionService.getSession(appName, userId, sessionId, Optional.empty()).blockingGet();
-    if (session == null) {
-      return Flowable.error(
-          new IllegalArgumentException(
-              String.format("Session not found: %s for user %s", sessionId, userId)));
-    }
-    return this.runLive(session, liveRequestQueue, runConfig);
+    return this.sessionService
+        .getSession(appName, userId, sessionId, Optional.empty())
+        .flatMapPublisher(
+            session -> {
+              if (session == null) {
+                return Flowable.error(
+                    new IllegalArgumentException(
+                        String.format("Session not found: %s for user %s", sessionId, userId)));
+              }
+              return this.runLive(session, liveRequestQueue, runConfig);
+            });
   }
 
   public Flowable<Event> runWithSessionId(