feat: bridge ADK Java plugin onRunErrorCallback to the Kotlin engine

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

@@ -48,11 +48,13 @@ import com.google.adk.tools.BaseToolset as JavaBaseToolset
  * control-flow writes reach the engine. Blocking work is fine: calls are dispatched off the thread
  * driving the agent.
  *
+ * A bridged plugin's error callbacks fire: `onRunErrorCallback` is notification-only -- the engine
+ * re-raises the run's error to the caller afterwards regardless, so it cannot recover the run (it
+ * is for logging, telemetry, or cleanup) -- while the `onModelErrorCallback` and
+ * `onToolErrorCallback` recovery hooks fire and can recover.
+ *
  * The interop surfaces a signal the engine cannot honor rather than silently dropping it:
  * - Setting `branch` on a bridged context throws. The branch is the engine's to set.
- * - A bridged plugin that overrides `onRunErrorCallback` is skipped with a warning (not run): the
- *   engine surfaces run-level errors through the returned event stream to the caller, not to
- *   plugins. The `onModelErrorCallback` and `onToolErrorCallback` recovery hooks do fire.
  * - A bridged tool's or plugin's `requestedAuthConfigs` or `deletedArtifactIds` write throws - the
  *   engine's event actions have no equivalent. Its `skipSummarization`,
  *   `requestedToolConfirmations` and `agentState` writes do cross, from a tool and a plugin alike.

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

@@ -16,7 +16,6 @@
 
 package com.google.adk.tokt.adapters
 
-import com.google.adk.agents.InvocationContext as JavaInvocationContext
 import com.google.adk.kt.agents.CallbackContext as KtCallbackContext
 import com.google.adk.kt.agents.InvocationContext as KtInvocationContext
 import com.google.adk.kt.callbacks.CallbackChoice
@@ -40,7 +39,6 @@ import com.google.adk.tokt.context.ktCallbackContextToJava
 import com.google.adk.tokt.context.ktToolContextToJava
 import io.reactivex.rxjava3.core.Completable
 import io.reactivex.rxjava3.core.Maybe
-import java.util.logging.Logger
 import kotlinx.coroutines.rx3.await
 import kotlinx.coroutines.rx3.awaitSingleOrNull
 import kotlinx.coroutines.withContext
@@ -55,30 +53,6 @@ import kotlinx.coroutines.withContext
  */
 internal class JavaPluginToKt(internal val plugin: JavaPlugin) : KtPlugin {
 
-  private companion object {
-    val logger: Logger = Logger.getLogger(JavaPluginToKt::class.java.name)
-  }
-
-  init {
-    // The engine surfaces run-level errors through the event stream, not to plugins, so an
-    // onRunErrorCallback override cannot fire here; warn and skip it rather than failing
-    // adaptation.
-    val onRunError =
-      plugin.javaClass.getMethod(
-        "onRunErrorCallback",
-        JavaInvocationContext::class.java,
-        Throwable::class.java,
-      )
-    // declaringClass is Plugin only when neither the class nor an intermediate interface overrode
-    // it.
-    if (onRunError.declaringClass != JavaPlugin::class.java) {
-      logger.warning(
-        "Bridged Java plugin '${plugin.name}' overrides onRunErrorCallback, which the ADK Kotlin " +
-          "engine does not invoke; the override is skipped (run errors surface via the event stream)."
-      )
-    }
-  }
-
   override val name: String
     get() = plugin.name
 
@@ -128,6 +102,11 @@ internal class JavaPluginToKt(internal val plugin: JavaPlugin) : KtPlugin {
     completeOnIo { plugin.afterRunCallback(javaContext) }
   }
 
+  override suspend fun onRunError(invocationContext: KtInvocationContext, error: Throwable) {
+    val javaContext = KtInvocationContextToJavaView(invocationContext)
+    completeOnIo { plugin.onRunErrorCallback(javaContext, error) }
+  }
+
   // Agent-level callbacks.
 
   override suspend fun beforeAgent(

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

@@ -26,6 +26,8 @@ import com.google.adk.artifacts.InMemoryArtifactService as JavaInMemoryArtifactS
 import com.google.adk.events.Event as JavaEvent
 import com.google.adk.events.EventActions as JavaEventActions
 import com.google.adk.events.EventCompaction as JavaEventCompaction
+import com.google.adk.kt.agents.BaseAgent as KtBaseAgent
+import com.google.adk.kt.agents.InvocationContext as KtInvocationContext
 import com.google.adk.kt.agents.LlmAgent as KtLlmAgent
 import com.google.adk.kt.agents.RunConfig as KtRunConfig
 import com.google.adk.kt.agents.StreamingMode as KtStreamingMode
@@ -149,6 +151,8 @@ import kotlin.test.assertSame
 import kotlin.test.assertTrue
 import kotlin.test.fail
 import kotlin.time.Duration.Companion.seconds
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.runBlocking
 
@@ -451,12 +455,21 @@ class KtRunnerInteropTest {
     }
   }
 
-  /** A Java plugin that overrides onRunErrorCallback, which the engine cannot fire. */
+  /** A Java plugin that records every error its onRunErrorCallback is notified of. */
   private class OnRunErrorJavaPlugin : JavaBasePlugin("on_run_error_plugin") {
+    val errors = CopyOnWriteArrayList<Throwable>()
+
     override fun onRunErrorCallback(
       invocationContext: JavaInvocationContext,
       error: Throwable,
-    ): Completable = Completable.complete()
+    ): Completable = Completable.fromAction { errors.add(error) }
+  }
+
+  /** A native Kotlin agent that always fails, to drive the run-error path. */
+  private class FailingKtAgent : KtBaseAgent(name = "failing") {
+    override fun runAsyncImpl(context: KtInvocationContext): Flow<KtEvent> = flow {
+      throw RuntimeException("boom")
+    }
   }
 
   /** A Java plugin that requests an auth config the engine cannot represent, from before-agent. */
@@ -3444,12 +3457,25 @@ class KtRunnerInteropTest {
   }
 
   @Test
-  fun asKtPlugin_pluginOverridingOnRunError_isSkippedNotRejected() {
-    // The engine never fires onRunErrorCallback, so the override is skipped with a warning rather
-    // than rejected; adaptation still succeeds and the plugin's other callbacks run.
-    val plugin = JavaAdkToKt.asKtPlugin(OnRunErrorJavaPlugin())
+  fun ktRunner_javaPluginOnRunErrorCallback_firesWhenRunFails() = runBlocking {
+    // A bridged Java plugin's onRunErrorCallback fires (notification-only) when the run fails, and
+    // the error still propagates to the caller.
+    val plugin = OnRunErrorJavaPlugin()
+    val runner =
+      KtInMemoryRunner(
+        app =
+          KtApp(
+            appName = "app",
+            rootAgent = FailingKtAgent(),
+            plugins = listOf(JavaAdkToKt.asKtPlugin(plugin)),
+          )
+      )
+
+    val thrown = assertFailsWith<RuntimeException> { runner.turn() }
 
-    assertEquals("on_run_error_plugin", plugin.name)
+    // The plugin was notified exactly once, with the same error instance that failed the run.
+    assertEquals(1, plugin.errors.size)
+    assertSame(thrown, plugin.errors.single())
   }
 
   @Test