feat(tokt): send blobs and run a held session through runLive

tokt/README.md

@@ -90,9 +90,16 @@ liveRunner
 
 -   Events arrive on the runner's dispatcher (`Dispatchers.IO` unless you pass
     one to `asJavaRunner`), never on the thread that sends to the queue.
--   Blobs sent to the Java `LiveRequestQueue` are dropped with a warning.
+-   A blob with an `audio/*` MIME type goes to the model as audio, and one with
+    `image/*` as a video frame. Any other blob, or one with no MIME type, is
+    dropped with a warning, because the Kotlin engine has no untyped media
+    input.
 -   `RunConfig` settings the Kotlin engine cannot honor fail the stream rather
     than being dropped.
+-   `runLive(session, ...)` runs the copy of that session the Kotlin runner's
+    session service stores under the runner's app, and does not update the
+    passed `Session` object. A session of another app, or one that service
+    lacks, fails the stream even with `autoCreateSession`.
 
 See the `JavaAdkToKt` KDoc for the full set of adapters (including session,
 artifact, and memory services) and the documented interop limits.

tokt/src/main/kotlin/com/google/adk/tokt/KtRunnerToJava.kt

@@ -27,6 +27,7 @@ import com.google.adk.events.Event as JavaEvent
 import com.google.adk.kt.agents.LiveRequestQueue as KtLiveRequestQueue
 import com.google.adk.kt.annotations.ExperimentalLiveApi
 import com.google.adk.kt.logging.LoggerFactory
+import com.google.adk.kt.models.RealtimeInput
 import com.google.adk.kt.runners.Runner as KtRunner
 import com.google.adk.kt.types.Content as KtContent
 import com.google.adk.kt.types.Role as KtRole
@@ -37,11 +38,13 @@ import com.google.adk.tokt.adapters.ktAgentAsJava
 import com.google.adk.tokt.adapters.ktPluginManagerAsJava
 import com.google.adk.tokt.codecs.ContentCodec
 import com.google.adk.tokt.codecs.EventCodec
+import com.google.adk.tokt.codecs.PartCodec
 import com.google.adk.tokt.codecs.RunConfigCodec
 import com.google.adk.tokt.codecs.stateDeltaFromJava
 import com.google.adk.tokt.services.ktArtifactServiceAsJava
 import com.google.adk.tokt.services.ktMemoryServiceAsJava
 import com.google.adk.tokt.services.ktSessionServiceAsJava
+import com.google.genai.types.Blob as GenaiBlob
 import com.google.genai.types.Content as GenaiContent
 import io.reactivex.rxjava3.core.Completable
 import io.reactivex.rxjava3.core.Flowable
@@ -139,12 +142,25 @@ internal class KtRunnerToJava(
     }
   }
 
-  // Not bridged for a Session: fails through the stream, like the base Runner, not eagerly.
+  /**
+   * Runs [session]'s conversation live on the Kotlin engine, which reads the session from its own
+   * service by app, user and id, and writes events to that stored copy rather than [session]. A
+   * session of another app, or one that service lacks, fails the stream even when [runConfig] sets
+   * `autoCreateSession`.
+   */
   override fun runLive(
     session: JavaSession,
     liveRequestQueue: LiveRequestQueue,
     runConfig: JavaRunConfig,
-  ): Flowable<JavaEvent> = Flowable.error(liveUnsupported())
+  ): Flowable<JavaEvent> = Flowable.defer {
+    require(session.appName() == appName()) { SESSION_NOT_HELD }
+    whenSessionExists(
+      session.userId(),
+      session.id(),
+      SESSION_NOT_HELD,
+      liveEvents(session.userId(), session.id(), liveRequestQueue, runConfig),
+    )
+  }
 
   /**
    * Runs a live conversation on the Kotlin engine: the Java queue's requests feed it, and its
@@ -179,21 +195,25 @@ internal class KtRunnerToJava(
     if (runConfig.autoCreateSession()) {
       events
     } else {
-      whenSessionExists(userId, sessionId, events)
+      whenSessionExists(userId, sessionId, SESSION_MISSING, events)
     }
 
-  /** Returns [events] if the Kotlin runner's service holds the session, else a failed stream. */
+  /**
+   * Returns [events] if the Kotlin runner's service holds the session, else a stream failing with
+   * [notFoundMessage].
+   */
   private fun whenSessionExists(
     userId: String,
     sessionId: String,
+    notFoundMessage: String,
     events: Flowable<JavaEvent>,
   ): Flowable<JavaEvent> =
     sessionService()
       .getSession(appName(), userId, sessionId, Optional.empty())
       .map { true }
       .defaultIfEmpty(false)
       .flatMapPublisher { exists ->
-        if (exists) events else Flowable.error(IllegalArgumentException(SESSION_MISSING))
+        if (exists) events else Flowable.error(IllegalArgumentException(notFoundMessage))
       }
 
   private fun liveEvents(
@@ -234,8 +254,14 @@ internal class KtRunnerToJava(
         val blob = request.blob().getOrNull()
         // Content wins over a blob, and an empty request closes the run.
         when {
-          content != null -> ktQueue.sendContent(withUserRole(ContentCodec.fromJava(content)))
-          blob != null -> logger.warn { "Live input blob not sent: blobs are not supported" }
+          content != null -> {
+            val ktContent = withUserRole(ContentCodec.fromJava(content))
+            // A function response goes out as a tool response, which has no partial form.
+            val partial =
+              request.isPartial() && ktContent.parts.none { it.functionResponse != null }
+            ktQueue.sendContent(ktContent, partial)
+          }
+          blob != null -> blob.toRealtimeInput()?.let { ktQueue.sendRealtime(it) }
           else -> ktQueue.close()
         }
         if (request.shouldClose()) ktQueue.close()
@@ -245,6 +271,23 @@ internal class KtRunnerToJava(
     }
   }
 
+  /**
+   * Maps a blob with an `audio/` MIME type to audio input and one with `image/` to a video frame,
+   * or drops any other blob with a warning.
+   */
+  private fun GenaiBlob.toRealtimeInput(): RealtimeInput? {
+    val ktBlob = PartCodec.blobFromJava(this)
+    val mimeType = ktBlob.mimeType.orEmpty()
+    return when {
+      mimeType.startsWith("audio/") -> RealtimeInput.Audio(ktBlob)
+      mimeType.startsWith("image/") -> RealtimeInput.Video(ktBlob)
+      else -> {
+        logger.warn { "Live input blob not sent, MIME type not supported: \"$mimeType\"" }
+        null
+      }
+    }
+  }
+
   // A read-only Java view of the Kotlin runner's plugins (adapted Java plugins unwrapped); the
   // engine's plugins are fixed at construction, so the returned manager throws on registration.
   override fun pluginManager(): JavaPluginManager = ktPluginManagerAsJava(ktRunner.pluginManager)
@@ -262,11 +305,6 @@ internal class KtRunnerToJava(
   override fun close(): Completable =
     Completable.mergeArrayDelayError(super.close(), Completable.fromAction { ktRunner.close() })
 
-  private fun liveUnsupported() =
-    UnsupportedOperationException(
-      "runLive(Session) is not supported on the Kotlin engine; use runLive(userId, sessionId, ...)"
-    )
-
   /**
    * Defaults a role-less message to the user role, as Python does, since model requests omit
    * role-less content.
@@ -277,5 +315,7 @@ internal class KtRunnerToJava(
   companion object {
     private val logger = LoggerFactory.getLogger(KtRunnerToJava::class)
     private const val SESSION_MISSING = "Session not found and autoCreateSession=false"
+    private const val SESSION_NOT_HELD =
+      "Session not found under this runner's app in the Kotlin runner's session service"
   }
 }

tokt/src/main/kotlin/com/google/adk/tokt/codecs/PartCodec.kt

@@ -102,6 +102,14 @@ internal object PartCodec {
   fun toJavaOrThrow(part: KtPart): GenaiPart =
     requireNotNull(toJava(part)) { "Artifact part is empty or unmapped" }
 
+  /** Returns the Kotlin [KtBlob] view of the genai [blob]. */
+  fun blobFromJava(blob: GenaiBlob): KtBlob =
+    KtBlob(
+      mimeType = blob.mimeType().getOrNull(),
+      displayName = blob.displayName().getOrNull(),
+      data = blob.data().getOrNull(),
+    )
+
   /** Returns the Kotlin [KtPart] view of the genai [part], or null if it carries nothing mapped. */
   fun fromJava(part: GenaiPart): KtPart? {
     val functionCall = part.functionCall().getOrNull()
@@ -229,13 +237,6 @@ internal object PartCodec {
     return builder.build()
   }
 
-  private fun blobFromJava(blob: GenaiBlob): KtBlob =
-    KtBlob(
-      mimeType = blob.mimeType().getOrNull(),
-      displayName = blob.displayName().getOrNull(),
-      data = blob.data().getOrNull(),
-    )
-
   private fun blobToJava(blob: KtBlob): GenaiBlob {
     val builder = GenaiBlob.builder()
     blob.data?.let { builder.data(it) }

tokt/src/test/kotlin/com/google/adk/tokt/KtRunnerInteropTest.kt

@@ -2080,12 +2080,41 @@ class KtRunnerInteropTest {
   }
 
   @Test
-  fun asJavaRunner_runLive_blob_isDroppedAndTheRunGoesOn() {
+  fun asJavaRunner_runLive_mapsEachBlobByItsMimeType() {
     val model = EchoLiveModel()
     val queue = LiveRequestQueue()
     val events = liveJavaRunner(model).runLive("u", "s", queue, autoCreateSessionConfig()).test()
 
-    queue.realtime(blob("audio/pcm"))
+    queue.realtime(blob("audio/pcm;rate=16000"))
+    queue.realtime(blob("image/jpeg").toBuilder().displayName("frame").build())
+    // A request carrying both content and a blob sends only the content.
+    queue.send(JavaLiveRequest.builder().content(userText("both")).blob(blob("audio/pcm")).build())
+    try {
+      awaitTrue("the echo should arrive") { events.values().size >= 2 }
+    } finally {
+      queue.close()
+    }
+
+    events.awaitDone(10, TimeUnit.SECONDS).assertComplete().assertNoErrors()
+    assertThat(model.connections.single().sent)
+      .containsExactly(
+        KtRealtimeInput.Audio(blobKt("audio/pcm;rate=16000")),
+        KtRealtimeInput.Video(blobKt("image/jpeg").copy(displayName = "frame")),
+        ContentCodec.fromJava(userText("both")),
+      )
+      .inOrder()
+  }
+
+  @Test
+  fun asJavaRunner_runLive_blobOfAnotherMimeType_isDroppedAndTheRunGoesOn() {
+    // Only `audio/` and `image/` blobs are sent; `video/`, `text/` and untyped blobs are dropped.
+    val model = EchoLiveModel()
+    val queue = LiveRequestQueue()
+    val events = liveJavaRunner(model).runLive("u", "s", queue, autoCreateSessionConfig()).test()
+
+    queue.realtime(blob("video/mp4"))
+    queue.realtime(blob("text/plain"))
+    queue.realtime(GenaiBlob.builder().data(byteArrayOf(1)).build())
     queue.content(userText("hi"))
     try {
       awaitTrue("the echo should arrive") { events.values().size >= 2 }
@@ -2098,6 +2127,57 @@ class KtRunnerInteropTest {
       .containsExactly(ContentCodec.fromJava(userText("hi")))
   }
 
+  @Test
+  fun asJavaRunner_runLive_partialContent_reachesTheModelAsPartial() {
+    val model = EchoLiveModel()
+    val queue = LiveRequestQueue()
+    val events = liveJavaRunner(model).runLive("u", "s", queue, autoCreateSessionConfig()).test()
+
+    queue.content(userText("hel"), /* partial= */ true)
+    queue.content(userText("lo"))
+    try {
+      awaitTrue("both sends should arrive") {
+        model.connections.singleOrNull()?.partials?.size == 2
+      }
+    } finally {
+      queue.close()
+    }
+
+    events.awaitDone(10, TimeUnit.SECONDS).assertComplete().assertNoErrors()
+    assertThat(model.connections.single().partials).containsExactly(true, false).inOrder()
+  }
+
+  @Test
+  fun asJavaRunner_runLive_partialFunctionResponse_reachesTheModelAsComplete() {
+    // A function response has no partial form, so the flag is dropped instead of failing the run.
+    val model = EchoLiveModel()
+    val queue = LiveRequestQueue()
+    val events = liveJavaRunner(model).runLive("u", "s", queue, autoCreateSessionConfig()).test()
+    val response =
+      GenaiContent.builder()
+        .role("user")
+        .parts(
+          GenaiPart.builder()
+            .functionResponse(
+              GenaiFunctionResponse.builder().name("lookup").response(mapOf("ok" to true)).build()
+            )
+            .build()
+        )
+        .build()
+
+    queue.content(response, /* partial= */ true)
+    try {
+      awaitTrue("the response should arrive") {
+        model.connections.singleOrNull()?.partials?.size == 1
+      }
+    } finally {
+      queue.close()
+    }
+
+    events.awaitDone(10, TimeUnit.SECONDS).assertComplete().assertNoErrors()
+    assertThat(model.connections.single().partials).containsExactly(false)
+  }
+
   @Test
   fun asJavaRunner_runLive_emptyRoleContent_reachesTheModelAsUser() {
     // The engine defaults only a missing role, so an empty one is the adapter's to fix.
@@ -2339,20 +2419,41 @@ class KtRunnerInteropTest {
   }
 
   @Test
-  fun asJavaRunner_runLiveOnASession_isUnsupported() {
+  fun asJavaRunner_runLiveOnASessionTheKotlinServiceLacks_failsEvenWithAutoCreate() {
     val javaRunner = liveJavaRunner(EchoLiveModel())
-    val session =
-      javaRunner.sessionService().createSession(JavaSessionKey("app", "u", "s")).blockingGet()
+    val stranger =
+      JavaInMemorySessionService().createSession(JavaSessionKey("app", "u", "s")).blockingGet()
+
+    val events = javaRunner.runLive(stranger, LiveRequestQueue(), autoCreateSessionConfig()).test()
+
+    events.awaitDone(10, TimeUnit.SECONDS)
+    events.assertError {
+      it is IllegalArgumentException &&
+        it.message ==
+          "Session not found under this runner's app in the Kotlin runner's session service"
+    }
+  }
+
+  @Test
+  fun asJavaRunner_runLiveOnAnotherAppsSession_failsTheStream() {
+    // The Kotlin service holds (u, s) under "app", which must not stand in for "other"'s session.
+    val javaRunner = liveJavaRunner(EchoLiveModel())
+    javaRunner
+      .sessionService()
+      .createSession(JavaSessionKey("app", "u", "s"))
+      .ignoreElement()
+      .blockingAwait()
+    val otherApp =
+      JavaInMemorySessionService().createSession(JavaSessionKey("other", "u", "s")).blockingGet()
 
     val events =
-      javaRunner.runLive(session, LiveRequestQueue(), JavaRunConfig.builder().build()).test()
+      javaRunner.runLive(otherApp, LiveRequestQueue(), JavaRunConfig.builder().build()).test()
 
     events.awaitDone(10, TimeUnit.SECONDS)
     events.assertError {
-      it is UnsupportedOperationException &&
+      it is IllegalArgumentException &&
         it.message ==
-          "runLive(Session) is not supported on the Kotlin engine; " +
-            "use runLive(userId, sessionId, ...)"
+          "Session not found under this runner's app in the Kotlin runner's session service"
     }
   }
 
@@ -2377,6 +2478,32 @@ class KtRunnerInteropTest {
       .containsExactly(ContentCodec.fromJava(userText("hi")))
   }
 
+  @Test
+  fun asJavaRunner_runLiveOnAHeldSession_runs() {
+    val model = EchoLiveModel()
+    val javaRunner = liveJavaRunner(model)
+    val session =
+      javaRunner.sessionService().createSession(JavaSessionKey("app", "u", "s")).blockingGet()
+    val queue = LiveRequestQueue()
+
+    val events = javaRunner.runLive(session, queue, JavaRunConfig.builder().build()).test()
+    try {
+      queue.content(userText("hi"))
+      awaitTrue("the echo should arrive") { events.values().size >= 2 }
+    } finally {
+      queue.close()
+    }
+
+    events.awaitDone(10, TimeUnit.SECONDS).assertComplete().assertNoErrors()
+    assertThat(model.connections.single().sent)
+      .containsExactly(ContentCodec.fromJava(userText("hi")))
+    // The turn lands on the stored copy, not on the caller's object.
+    val stored =
+      javaRunner.sessionService().getSession("app", "u", "s", Optional.empty()).blockingGet()
+    assertThat(assertNotNull(stored).immutableEvents()).isNotEmpty()
+    assertThat(session.immutableEvents()).isEmpty()
+  }
+
   @Test
   fun asJavaRunner_runLive_runsOnTheRunnersDispatcherNotTheSendersThread() {
     // Model sends and Java signals run on the runner's dispatcher, never on the sender's thread.
@@ -6717,6 +6844,7 @@ class KtRunnerInteropTest {
     private val receiveFailure: Throwable?,
   ) : KtLiveConnection {
     val sent = CopyOnWriteArrayList<Any>()
+    val partials = CopyOnWriteArrayList<Boolean>()
     val sendThreads = CopyOnWriteArrayList<String>()
     @Volatile
     var closed = false
@@ -6728,6 +6856,7 @@ class KtRunnerInteropTest {
 
     override suspend fun sendContent(content: KtContent, partial: Boolean) {
       sent += content
+      partials += partial
       sendThreads += Thread.currentThread().name
       val text = "echo: " + content.parts.mapNotNull { it.text }.joinToString("")
       replies.send(
@@ -6845,6 +6974,8 @@ class KtRunnerInteropTest {
     fun blob(mimeType: String): GenaiBlob =
       GenaiBlob.builder().mimeType(mimeType).data(byteArrayOf(1, 2)).build()
 
+    fun blobKt(mimeType: String) = KtBlob(mimeType = mimeType, data = byteArrayOf(1, 2))
+
     /** A speech setting with every field set, including a replicated voice and two speakers. */
     fun ktSpeechConfig() =
       KtSpeechConfig(