feat(tokt): run agents through the ADK Java runner's runLive

tokt/README.md

@@ -52,5 +52,47 @@ List<Event> events =
         .blockingGet();
 ```
 
+## Run a live conversation
+
+Live runs are experimental: they build on the ADK Kotlin live API, which may
+change without notice.
+
+`runLive` runs a live (bidirectional) conversation on the Kotlin engine, fed
+from an ADK Java `LiveRequestQueue`. The agent's model must support live
+connections, as the Kotlin engine's `Gemini` does:
+
+```java
+import com.google.adk.agents.LiveRequestQueue; // ADK Java queue
+
+LlmAgent liveAgent =
+    LlmAgent.builder()
+        .name("voice_assistant")
+        .model(new com.google.adk.kt.models.Gemini("gemini-3.8-live"))
+        .build();
+Runner liveRunner =
+    KotlinAdkToJava.asJavaRunner(
+        InMemoryRunner.builder()
+            .app(App.builder().appName("voice").rootAgent(liveAgent).build())
+            .build());
+
+LiveRequestQueue queue = new LiveRequestQueue();
+queue.content(Content.fromParts(Part.fromText("Hello!")));
+liveRunner
+    .runLive("user", "session", queue, RunConfig.builder().autoCreateSession(true).build())
+    .blockingSubscribe(
+        event -> {
+          event.outputTranscription().flatMap(Transcription::text).ifPresent(System.out::println);
+          if (event.turnComplete().orElse(false)) {
+            queue.close(); // Ends the run and closes the connection.
+          }
+        });
+```
+
+-   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.
+-   `RunConfig` settings the Kotlin engine cannot honor fail the stream rather
+    than being dropped.
+
 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/KotlinAdkToJava.kt

@@ -23,17 +23,19 @@ import kotlinx.coroutines.CoroutineDispatcher
 /**
  * Reverse interop entry point: [asJavaRunner] wraps an ADK Kotlin-engine `Runner` in a real ADK
  * Java `Runner`, so it can be injected into code written against the Java runner. The wrapper's
- * `runAsync` streams `Event`s backed by the Kotlin engine; live mode is not bridged. The forward
- * direction (Java components onto the Kotlin engine) lives in [com.google.adk.tokt.JavaAdkToKt].
+ * `runAsync` and `runLive` stream `Event`s backed by the Kotlin engine; `runLive` is experimental,
+ * as the ADK Kotlin live API it builds on may change without notice. The forward direction (Java
+ * components onto the Kotlin engine) lives in [com.google.adk.tokt.JavaAdkToKt].
  */
 object KotlinAdkToJava {
 
   /**
    * Exposes a Kotlin-engine [runner] as an ADK Java `Runner`. Its reverse service adapters bridge
-   * RxJava calls onto the Kotlin engine via `dispatcher` (default `Dispatchers.IO`). That
-   * dispatcher must be able to run nested bridged calls in parallel: if code on one of its threads
-   * (such as a bridged call's RxJava callback) blocks on another bridged call, it holds that thread
-   * until the second call returns, so a single-threaded or tightly bounded dispatcher can deadlock.
+   * RxJava calls onto the Kotlin engine on `dispatcher` (default `Dispatchers.IO`), which also
+   * emits `runLive`'s events. That dispatcher must be able to run nested bridged calls in parallel:
+   * if code on one of its threads (such as a bridged call's RxJava callback) blocks on another
+   * bridged call, it holds that thread until the second call returns, so a single-threaded or
+   * tightly bounded dispatcher can deadlock.
    */
   @JvmStatic
   @JvmOverloads

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

@@ -14,6 +14,8 @@
  * limitations under the License.
  */
 
+@file:OptIn(ExperimentalLiveApi::class)
+
 package com.google.adk.tokt
 
 import com.google.adk.agents.LiveRequestQueue
@@ -22,6 +24,9 @@ import com.google.adk.annotations.Experimental
 import com.google.adk.artifacts.BaseArtifactService as JavaArtifactService
 import com.google.adk.artifacts.InMemoryArtifactService as JavaInMemoryArtifactService
 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.runners.Runner as KtRunner
 import com.google.adk.kt.types.Content as KtContent
 import com.google.adk.kt.types.Role as KtRole
@@ -41,25 +46,33 @@ import com.google.genai.types.Content as GenaiContent
 import io.reactivex.rxjava3.core.Completable
 import io.reactivex.rxjava3.core.Flowable
 import java.util.Optional
+import kotlin.jvm.optionals.getOrNull
 import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.flow.buffer
+import kotlinx.coroutines.flow.emitAll
+import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.reactive.asFlow
 import kotlinx.coroutines.rx3.asFlowable
 
 /**
  * A [JavaRunner] that delegates to a Kotlin-engine [KtRunner], so a Kotlin runner can be dropped
- * into code written against the ADK Java [JavaRunner]: `runAsync` converts the request in and each
- * event out, running on the Kotlin engine against its own services and plugins. It honors the Java
- * `RunConfig.autoCreateSession` contract (erroring on a missing session unless it is set), but
- * per-session request sequencing follows the Kotlin engine, not the Java runner; live mode is not
- * bridged, so [runLive] returns a failed stream. [agent], [sessionService], [memoryService],
- * [artifactService] and [pluginManager] read the Kotlin runner's own components back through the
- * reverse adapters ([memoryService] / [artifactService] return `null` when absent; [pluginManager]
- * is read-only and throws on registration). The reverse service adapters bridge Java RxJava calls
- * onto the Kotlin engine via `dispatcher`.
+ * into code written against the ADK Java [JavaRunner]; `runAsync` and [runLive] run on the Kotlin
+ * engine against its own services and plugins. It honors the Java `RunConfig.autoCreateSession`
+ * contract, but per-session request sequencing follows the Kotlin engine, not the Java runner.
+ * [agent], [sessionService], [memoryService], [artifactService] and [pluginManager] read the Kotlin
+ * runner's own components back through reverse adapters on `dispatcher` ([memoryService] and
+ * [artifactService] return `null` when absent; [pluginManager] is read-only).
  */
 // Subclassing Runner via its @Deprecated 8-arg super-constructor is intended here.
 @Suppress("DEPRECATION")
-internal class KtRunnerToJava(private val ktRunner: KtRunner, dispatcher: CoroutineDispatcher) :
+internal class KtRunnerToJava(
+  private val ktRunner: KtRunner,
+  private val dispatcher: CoroutineDispatcher,
+) :
   JavaRunner(
     // A Java view of the Kotlin runner's agent so agent() reads back; never run (see runAsync).
     ktAgentAsJava(ktRunner.agent),
@@ -122,37 +135,115 @@ internal class KtRunnerToJava(private val ktRunner: KtRunner, dispatcher: Corout
           )
           .map { EventCodec.toJava(it) }
           .asFlowable()
-      // The Kotlin engine always creates a missing session; the Java Runner errors unless
-      // autoCreateSession is set, so honor that contract before delegating.
-      if (runConfig.autoCreateSession()) return@defer events
-      sessionService()
-        .getSession(appName(), userId, sessionId, Optional.empty())
-        .map { true }
-        .defaultIfEmpty(false)
-        .flatMapPublisher { exists ->
-          if (exists) events
-          else
-            Flowable.error(
-              IllegalArgumentException("Session not found and autoCreateSession=false")
-            )
-        }
+      honoringAutoCreate(userId, sessionId, runConfig, events)
     }
   }
 
-  // Live mode is unsupported; surface it through the stream (like the base Runner) rather than
-  // throwing eagerly, so callers on the reactive path see it via onError.
+  // Not bridged for a Session: fails through the stream, like the base Runner, not eagerly.
   override fun runLive(
     session: JavaSession,
     liveRequestQueue: LiveRequestQueue,
     runConfig: JavaRunConfig,
   ): Flowable<JavaEvent> = Flowable.error(liveUnsupported())
 
+  /**
+   * Runs a live conversation on the Kotlin engine: the Java queue's requests feed it, and its
+   * events arrive on `dispatcher`'s threads, never the sender's, until the queue closes or the
+   * stream is cancelled, which closes the connection.
+   */
   override fun runLive(
     userId: String,
     sessionId: String,
     liveRequestQueue: LiveRequestQueue,
     runConfig: JavaRunConfig,
-  ): Flowable<JavaEvent> = Flowable.error(liveUnsupported())
+  ): Flowable<JavaEvent> = Flowable.defer {
+    honoringAutoCreate(
+      userId,
+      sessionId,
+      runConfig,
+      liveEvents(userId, sessionId, liveRequestQueue, runConfig),
+    )
+  }
+
+  /**
+   * Returns [events], first checking that the session exists unless [runConfig] auto-creates it:
+   * the Kotlin engine always creates a missing session, while the Java Runner errors unless told
+   * otherwise.
+   */
+  private fun honoringAutoCreate(
+    userId: String,
+    sessionId: String,
+    runConfig: JavaRunConfig,
+    events: Flowable<JavaEvent>,
+  ): Flowable<JavaEvent> =
+    if (runConfig.autoCreateSession()) {
+      events
+    } else {
+      whenSessionExists(userId, sessionId, events)
+    }
+
+  /** Returns [events] if the Kotlin runner's service holds the session, else a failed stream. */
+  private fun whenSessionExists(
+    userId: String,
+    sessionId: 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))
+      }
+
+  private fun liveEvents(
+    userId: String,
+    sessionId: String,
+    javaQueue: LiveRequestQueue,
+    runConfig: JavaRunConfig,
+  ): Flowable<JavaEvent> {
+    val ktRunConfig = RunConfigCodec.fromJavaLive(runConfig)
+    val ktQueue = KtLiveRequestQueue()
+    // Called at subscribe time so the Kotlin runner captures the subscriber's trace context.
+    val ktEvents = ktRunner.runLive(userId, sessionId, ktQueue, ktRunConfig)
+    return flow {
+        coroutineScope {
+          // Safe: both queues are thread-safe, and the Kotlin one never blocks the pump.
+          @Suppress("UnsafeCoroutineCrossing")
+          val pump = launch { pumpRequests(javaQueue, ktQueue) }
+          try {
+            emitAll(ktEvents.map { EventCodec.toJava(it) })
+          } finally {
+            pump.cancel()
+            ktQueue.close()
+          }
+        }
+      }
+      // The runner's dispatcher, not Unconfined, so a Java sender never runs the live loop.
+      .asFlowable(dispatcher)
+  }
+
+  /**
+   * Feeds the Java queue's requests to [ktQueue], closing it once the Java queue closes or ends.
+   */
+  private suspend fun pumpRequests(javaQueue: LiveRequestQueue, ktQueue: KtLiveRequestQueue) {
+    try {
+      // Unbounded, so a burst can't overflow the Java queue.
+      javaQueue.get().asFlow().buffer(Channel.UNLIMITED).collect { request ->
+        val content = request.content().getOrNull()
+        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" }
+          else -> ktQueue.close()
+        }
+        if (request.shouldClose()) ktQueue.close()
+      }
+    } finally {
+      ktQueue.close()
+    }
+  }
 
   // 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.
@@ -172,12 +263,19 @@ internal class KtRunnerToJava(private val ktRunner: KtRunner, dispatcher: Corout
     Completable.mergeArrayDelayError(super.close(), Completable.fromAction { ktRunner.close() })
 
   private fun liveUnsupported() =
-    UnsupportedOperationException("Live mode is not supported when running on the Kotlin engine.")
+    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.
    */
   private fun withUserRole(content: KtContent): KtContent =
     if (content.role.isNullOrEmpty()) content.copy(role = KtRole.USER) else content
+
+  companion object {
+    private val logger = LoggerFactory.getLogger(KtRunnerToJava::class)
+    private const val SESSION_MISSING = "Session not found and autoCreateSession=false"
+  }
 }

tokt/src/main/kotlin/com/google/adk/tokt/adapters/JavaPluginToKt.kt

@@ -32,6 +32,8 @@ import com.google.adk.tokt.codecs.ContentCodec
 import com.google.adk.tokt.codecs.EventCodec
 import com.google.adk.tokt.codecs.LlmRequestCodec
 import com.google.adk.tokt.codecs.LlmResponseCodec
+import com.google.adk.tokt.codecs.reapplyJavaEvent
+import com.google.adk.tokt.codecs.reapplyJavaResponse
 import com.google.adk.tokt.context.KtInvocationContextToJavaView
 import com.google.adk.tokt.context.javaAgentView
 import com.google.adk.tokt.context.ktCallbackContextToJava
@@ -97,7 +99,7 @@ internal class JavaPluginToKt(
   override suspend fun onEvent(invocationContext: KtInvocationContext, event: KtEvent): KtEvent {
     val javaContext = KtInvocationContextToJavaView(invocationContext, dispatcher)
     val replacement = onDispatcher { plugin.onEventCallback(javaContext, EventCodec.toJava(event)) }
-    return replacement?.let { EventCodec.fromJava(it) } ?: event
+    return if (replacement != null) reapplyJavaEvent(event, replacement) else event
   }
 
   override suspend fun afterRun(invocationContext: KtInvocationContext) {
@@ -157,7 +159,7 @@ internal class JavaPluginToKt(
       plugin.afterModelCallback(javaContext, LlmResponseCodec.toJava(response))
     }
     reconcileActionsToKt(javaContext.eventActions(), context.eventActions)
-    return if (override != null) LlmResponseCodec.fromJava(override) else response
+    return if (override != null) reapplyJavaResponse(response, override) else response
   }
 
   override suspend fun onModelError(

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

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.adk.tokt.codecs
+
+import com.google.adk.kt.types.AudioTranscriptionConfig as KtAudioTranscriptionConfig
+import com.google.adk.kt.types.AudioTranscriptionConfigMode as KtAudioTranscriptionConfigMode
+import com.google.genai.types.AudioTranscriptionConfig as GenaiAudioTranscriptionConfig
+import com.google.genai.types.AudioTranscriptionConfigMode as GenaiAudioTranscriptionConfigMode
+import kotlin.jvm.optionals.getOrNull
+
+/**
+ * Converts a live transcription setting between the ADK Kotlin [KtAudioTranscriptionConfig] and the
+ * genai [GenaiAudioTranscriptionConfig] that the ADK Java `RunConfig` carries. The genai-only
+ * fields, and a `mode` Kotlin has no constant for, cannot cross, so [unsupportedFields] names the
+ * ones a Java caller set.
+ */
+internal object AudioTranscriptionConfigCodec {
+
+  /** Returns the genai [GenaiAudioTranscriptionConfig] view of the Kotlin [config]. */
+  fun toJava(config: KtAudioTranscriptionConfig): GenaiAudioTranscriptionConfig {
+    val builder = GenaiAudioTranscriptionConfig.builder()
+    config.languageCodes?.let { builder.languageCodes(it) }
+    config.customVocabulary?.let { builder.customVocabulary(it) }
+    config.diarization?.let { builder.diarization(it) }
+    config.wordTimestamp?.let { builder.wordTimestamp(it) }
+    config.mode?.let { builder.mode(GenaiAudioTranscriptionConfigMode(it.name)) }
+    return builder.build()
+  }
+
+  /** Returns the Kotlin [KtAudioTranscriptionConfig] view of the genai [config]. */
+  fun fromJava(config: GenaiAudioTranscriptionConfig): KtAudioTranscriptionConfig =
+    KtAudioTranscriptionConfig(
+      languageCodes = config.languageCodes().getOrNull(),
+      customVocabulary = config.customVocabulary().getOrNull(),
+      diarization = config.diarization().getOrNull(),
+      wordTimestamp = config.wordTimestamp().getOrNull(),
+      mode = config.mode().getOrNull()?.let { modeFromJava(it) },
+    )
+
+  /** Returns the fields set on [config] that [fromJava] cannot carry. */
+  fun unsupportedFields(config: GenaiAudioTranscriptionConfig): List<String> = buildList {
+    if (config.languageAuto().isPresent) add("languageAuto")
+    if (config.languageHints().isPresent) add("languageHints")
+    if (config.adaptationPhrases().isPresent) add("adaptationPhrases")
+    val mode = config.mode().getOrNull()
+    if (mode != null && modeFromJava(mode) == null) add("mode")
+  }
+
+  private fun modeFromJava(
+    mode: GenaiAudioTranscriptionConfigMode
+  ): KtAudioTranscriptionConfigMode? =
+    enumByNameOrNull<KtAudioTranscriptionConfigMode>(mode.knownEnum().name)
+}