From 1f1537dc8c8662b9c4081b5eed40cba5e4851a76 Mon Sep 17 00:00:00 2001 From: Kamil Tomaszek Date: Wed, 4 Nov 2026 05:26:34 -0800 Subject: [PATCH] feat(tokt): send blobs and run a held session through runLive PiperOrigin-RevId: 1013771510 --- tokt/README.md | 9 +- .../com/google/adk/tokt/KtRunnerToJava.kt | 64 ++++++-- .../com/google/adk/tokt/codecs/PartCodec.kt | 15 +- .../google/adk/tokt/KtRunnerInteropTest.kt | 149 ++++++++++++++++-- 4 files changed, 208 insertions(+), 29 deletions(-) diff --git a/tokt/README.md b/tokt/README.md index 612e44245..0c20beacd 100644 --- a/tokt/README.md +++ b/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. diff --git a/tokt/src/main/kotlin/com/google/adk/tokt/KtRunnerToJava.kt b/tokt/src/main/kotlin/com/google/adk/tokt/KtRunnerToJava.kt index 88585d39c..c570fc6a3 100644 --- a/tokt/src/main/kotlin/com/google/adk/tokt/KtRunnerToJava.kt +++ b/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 = Flowable.error(liveUnsupported()) + ): Flowable = 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,13 +195,17 @@ 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, ): Flowable = sessionService() @@ -193,7 +213,7 @@ internal class KtRunnerToJava( .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" } } diff --git a/tokt/src/main/kotlin/com/google/adk/tokt/codecs/PartCodec.kt b/tokt/src/main/kotlin/com/google/adk/tokt/codecs/PartCodec.kt index 7a5f2c9e1..b0a32a185 100644 --- a/tokt/src/main/kotlin/com/google/adk/tokt/codecs/PartCodec.kt +++ b/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) } diff --git a/tokt/src/test/kotlin/com/google/adk/tokt/KtRunnerInteropTest.kt b/tokt/src/test/kotlin/com/google/adk/tokt/KtRunnerInteropTest.kt index 1dc100548..10f5b0eee 100644 --- a/tokt/src/test/kotlin/com/google/adk/tokt/KtRunnerInteropTest.kt +++ b/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() + val partials = CopyOnWriteArrayList() val sendThreads = CopyOnWriteArrayList() @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(