diff --git a/pom.xml b/pom.xml index 26440989b..afb900088 100644 --- a/pom.xml +++ b/pom.xml @@ -72,10 +72,10 @@ 1.0.0 3.1.12 - 2.1.20 + 2.3.21 2.2.0 1.11.0 - 1.3.0 + 1.4.0 3.7.0 2.35.1 3.27.7 diff --git a/tokt/src/main/kotlin/com/google/adk/tokt/JavaAdkToKt.kt b/tokt/src/main/kotlin/com/google/adk/tokt/JavaAdkToKt.kt index 6c0171f40..24af4a6ab 100644 --- a/tokt/src/main/kotlin/com/google/adk/tokt/JavaAdkToKt.kt +++ b/tokt/src/main/kotlin/com/google/adk/tokt/JavaAdkToKt.kt @@ -65,8 +65,8 @@ import kotlinx.coroutines.CoroutineDispatcher * `EventActions`, the adapter throws once the Java call returns - the engine's event actions have * no equivalent. * - * Writes to the context's `skipSummarization`, `requestedToolConfirmations`, `agentState`, and - * `rewindBeforeInvocationId` do cross to the engine. + * Writes to the context's `skipSummarization`, `requestedToolConfirmations`, `agentState`, + * `rewindBeforeInvocationId`, and `route` do cross to the engine. */ object JavaAdkToKt { @@ -139,9 +139,9 @@ object JavaAdkToKt { /** * Adapts an ADK Java session service for the Kotlin engine, running its calls on `dispatcher`. A * Java view of a Kotlin service is unwrapped to that Kotlin service, with no `dispatcher` hop. - * Resumption works across the adapter (`EventActions.agentState` crosses); - * `rewindBeforeInvocationId` crosses too, but a rewind drops the rewound turns only if the Java - * service stores that field. + * Resumption and rewinds work across the adapter if the Java service, like + * `InMemorySessionService`, returns the fields they read unchanged: `EventActions.agentState`, + * `rewindBeforeInvocationId`, and a workflow node's output, node info, and routes. */ @JvmStatic @JvmOverloads diff --git a/tokt/src/main/kotlin/com/google/adk/tokt/adapters/ActionReconciler.kt b/tokt/src/main/kotlin/com/google/adk/tokt/adapters/ActionReconciler.kt index 340a99529..e23001888 100644 --- a/tokt/src/main/kotlin/com/google/adk/tokt/adapters/ActionReconciler.kt +++ b/tokt/src/main/kotlin/com/google/adk/tokt/adapters/ActionReconciler.kt @@ -20,15 +20,16 @@ import com.google.adk.events.EventActions as JavaEventActions import com.google.adk.kt.events.EventActions as KtEventActions import com.google.adk.tokt.codecs.EventCompactionCodec import com.google.adk.tokt.codecs.ToolConfirmationCodec +import com.google.adk.tokt.codecs.WorkflowCodec import com.google.adk.tokt.codecs.agentStateFromJava import com.google.adk.tokt.codecs.reconcileRemovedSentinels /** * Copies what a bridged tool or plugin callback wrote on its [JavaEventActions] onto the live - * [KtEventActions] the engine reads. Fields the live view already writes through are re-applied - * idempotently; skip-summarization, tool confirmations, agentState, compaction and - * rewindBeforeInvocationId are carried here. A signal the engine cannot represent - * (requestedAuthConfigs, deletedArtifactIds) throws. + * [KtEventActions] the engine reads. Fields the live view writes through, such as the workflow + * route, are re-applied idempotently, which also carries them after a wholesale `setActions(...)`; + * skip-summarization, tool confirmations, and agentState are carried only here. A signal the engine + * cannot represent (requestedAuthConfigs, deletedArtifactIds) throws. */ internal fun reconcileActionsToKt(java: JavaEventActions, kt: KtEventActions) { require(java.requestedAuthConfigs().isNullOrEmpty()) { @@ -43,6 +44,7 @@ internal fun reconcileActionsToKt(java: JavaEventActions, kt: KtEventActions) { java.agentState().ifPresent { kt.agentState = agentStateFromJava(it) } java.compaction().ifPresent { kt.compaction = EventCompactionCodec.fromJava(it) } java.rewindBeforeInvocationId().ifPresent { kt.rewindBeforeInvocationId = it } + java.route().ifPresent { routes -> kt.route = routes.map { WorkflowCodec.routeFromJava(it) } } kt.endOfAgent = kt.endOfAgent || java.endOfAgent() // Merge deltas only when the tool replaced the actions (else they are the same live Kotlin map). if (java.stateDelta() !== kt.stateDelta) kt.stateDelta.putAll(java.stateDelta()) diff --git a/tokt/src/main/kotlin/com/google/adk/tokt/codecs/EventCodec.kt b/tokt/src/main/kotlin/com/google/adk/tokt/codecs/EventCodec.kt index 0e75fae53..f05f264ab 100644 --- a/tokt/src/main/kotlin/com/google/adk/tokt/codecs/EventCodec.kt +++ b/tokt/src/main/kotlin/com/google/adk/tokt/codecs/EventCodec.kt @@ -31,10 +31,11 @@ import kotlin.jvm.optionals.getOrNull * Carries the identity (`id`, `invocationId`, `author`, `branch`), the [content][ContentCodec], the * [actions][KtEventActions] (state/artifact deltas, the control-flow flags - transfer, escalate, * skipSummarization, endOfAgent - the requested tool confirmations, the context-compaction summary, - * the resumable-run `agentState`, and the rewind marker `rewindBeforeInvocationId`), the - * streaming/error signals (`partial`, `turnComplete`, `interrupted`, `errorMessage`, `errorCode`, - * `finishReason`), the token `usageMetadata` ([UsageMetadataCodec]) and `groundingMetadata` - * ([GroundingMetadataCodec]), long-running tool ids, and the timestamp. + * the resumable-run `agentState`, the rewind marker `rewindBeforeInvocationId`, and the workflow + * `route`), the streaming/error signals (`partial`, `turnComplete`, `interrupted`, `errorMessage`, + * `errorCode`, `finishReason`), the token `usageMetadata` ([UsageMetadataCodec]) and + * `groundingMetadata` ([GroundingMetadataCodec]), a workflow node's `output` and `nodeInfo` + * ([WorkflowCodec]), long-running tool ids, and the timestamp. */ internal object EventCodec { @@ -62,6 +63,8 @@ internal object EventCodec { event.groundingMetadata?.let { builder.groundingMetadata(GroundingMetadataCodec.toJava(it)) } event.modelVersion?.let { builder.modelVersion(it) } event.customMetadata?.let { builder.customMetadata(CustomMetadataCodec.toJava(it)) } + event.output?.let { builder.output(WorkflowCodec.outputToJava(it)) } + event.nodeInfo?.let { builder.nodeInfo(WorkflowCodec.nodeInfoToJava(it)) } builder.timestamp(event.timestamp) return builder.build() } @@ -93,6 +96,7 @@ internal object EventCodec { javaActions.compaction().getOrNull()?.let { EventCompactionCodec.fromJava(it) }, agentState = agentStateFromJava(javaActions.agentState().getOrNull()), rewindBeforeInvocationId = javaActions.rewindBeforeInvocationId().getOrNull(), + route = javaActions.route().getOrNull()?.map { WorkflowCodec.routeFromJava(it) }, ), branch = event.branch().getOrNull(), partial = event.partial().getOrNull() ?: false, @@ -109,6 +113,8 @@ internal object EventCodec { event.groundingMetadata().getOrNull()?.let { GroundingMetadataCodec.fromJava(it) }, modelVersion = event.modelVersion().getOrNull(), customMetadata = event.customMetadata().getOrNull()?.let { CustomMetadataCodec.fromJava(it) }, + output = event.output().getOrNull()?.let { WorkflowCodec.outputFromJava(it) }, + nodeInfo = event.nodeInfo().getOrNull()?.let { WorkflowCodec.nodeInfoFromJava(it) }, timestamp = event.timestamp(), ) } diff --git a/tokt/src/main/kotlin/com/google/adk/tokt/codecs/KtToJavaCodecs.kt b/tokt/src/main/kotlin/com/google/adk/tokt/codecs/KtToJavaCodecs.kt index 7b89ec409..33cf0561b 100644 --- a/tokt/src/main/kotlin/com/google/adk/tokt/codecs/KtToJavaCodecs.kt +++ b/tokt/src/main/kotlin/com/google/adk/tokt/codecs/KtToJavaCodecs.kt @@ -25,6 +25,8 @@ import com.google.adk.kt.sessions.SessionKey as KtSessionKey import com.google.adk.kt.sessions.State as KtState import com.google.adk.sessions.Session as JavaSession import com.google.adk.sessions.State as JavaState +import com.google.adk.workflow.Route as JavaRoute +import com.google.common.collect.ImmutableList import java.util.AbstractList import java.util.AbstractMap import java.util.Optional @@ -81,6 +83,17 @@ internal class KtEventActionsToJavaView(private val actions: KtEventActions) : J actions.rewindBeforeInvocationId = rewindBeforeInvocationId } + override fun route(): Optional> = + Optional.ofNullable( + actions.route?.let { routes -> + ImmutableList.copyOf(routes.map { WorkflowCodec.routeToJava(it) }) + } + ) + + override fun setRoute(route: List?) { + actions.route = route?.map { WorkflowCodec.routeFromJava(it) } + } + // Overridden so the write reaches the Kotlin side; the base setters write private fields the // overridden getters never read. @@ -113,9 +126,9 @@ internal class KtEventActionsToJavaView(private val actions: KtEventActions) : J * Builds a read-only Java [JavaEventActions] snapshot from a Kotlin [KtEventActions], carrying the * deltas, the control-flow signals (skip-summarization, transfer, escalate, end-of-agent), the * requested tool confirmations, the compaction summary, the resumable-run `agentState` - * ([agentStateToJava]), and the rewind marker (`rewindBeforeInvocationId`). Used when exposing an - * existing Kotlin event to a Java runner ([EventCodec.toJava]); unlike [KtEventActionsToJavaView] - * (a live write sink), nothing writes back through it. + * ([agentStateToJava]), the rewind marker (`rewindBeforeInvocationId`), and the workflow `route`. + * Used when exposing an existing Kotlin event to a Java runner ([EventCodec.toJava]); unlike + * [KtEventActionsToJavaView] (a live write sink), nothing writes back through it. */ internal fun eventActionsToJava(actions: KtEventActions): JavaEventActions = JavaEventActions.builder() @@ -131,6 +144,7 @@ internal fun eventActionsToJava(actions: KtEventActions): JavaEventActions = .compaction(actions.compaction?.let { EventCompactionCodec.toJava(it) }) .agentState(agentStateToJava(actions.agentState)) .rewindBeforeInvocationId(actions.rewindBeforeInvocationId) + .route(actions.route?.map { WorkflowCodec.routeToJava(it) }) .build() /** Copies a Kotlin state delta to Java, mapping the Kotlin [KtState.REMOVED] deletion sentinel. */ diff --git a/tokt/src/main/kotlin/com/google/adk/tokt/codecs/WorkflowCodec.kt b/tokt/src/main/kotlin/com/google/adk/tokt/codecs/WorkflowCodec.kt new file mode 100644 index 000000000..7088f057a --- /dev/null +++ b/tokt/src/main/kotlin/com/google/adk/tokt/codecs/WorkflowCodec.kt @@ -0,0 +1,125 @@ +/* + * 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.JsonBaseModel +import com.google.adk.kt.annotations.FrameworkInternalApi +import com.google.adk.kt.serialization.anyToJsonElement +import com.google.adk.kt.serialization.jsonElementToAny +import com.google.adk.kt.types.Content as KtContent +import com.google.adk.kt.workflow.NodeInfo as KtNodeInfo +import com.google.adk.kt.workflow.Route as KtRoute +import com.google.adk.workflow.NodeInfo as JavaNodeInfo +import com.google.adk.workflow.Route as JavaRoute +import com.google.genai.types.Content as GenaiContent + +/** + * Converts a workflow node's output, its [NodeInfo][KtNodeInfo], and the routes on `actions.route` + * between ADK Kotlin and the ADK Java facade. + */ +internal object WorkflowCodec { + + /** + * Returns the Java form of a node's [output]: a Kotlin `Content` becomes a genai `Content`, and a + * kotlinx `JsonElement` becomes the plain maps, lists, and values of an event read from JSON. + * Maps and lists are copied with each entry converted; a value with no JSON form throws. + */ + @OptIn(FrameworkInternalApi::class) + fun outputToJava(output: Any?): Any? = + when (output) { + null, + is String, + is Boolean, + is Number -> output + is KtContent -> ContentCodec.toJava(output) + // A JsonObject or JsonArray is also a Map or List, so it converts entry by entry here. + is Map<*, *> -> + output.entries.associate { (key, value) -> key.toString() to outputToJava(value) } + is List<*> -> output.map { outputToJava(it) } + // Unwraps a JSON primitive or JSON null; throws if a value has no JSON form. + else -> jsonElementToAny(anyToJsonElement(output)) + } + + /** + * Returns the Kotlin form of a node's [output]: a genai `Content` becomes a Kotlin `Content`, and + * any other non-JSON-native object becomes its JSON form, as in an event read from JSON. Maps and + * lists are copied with each entry converted. + */ + fun outputFromJava(output: Any?): Any? = + when (output) { + null, + is String, + is Boolean, + is Int, + is Long, + is Float, + is Double, + is Short, + is Byte -> output + // ADK Kotlin has no JSON form for BigDecimal or BigInteger, so convert to Long or Double. + is Number -> output.toString().toLongOrNull() ?: output.toDouble() + is GenaiContent -> ContentCodec.fromJava(output) + is Map<*, *> -> + output.entries.associate { (key, value) -> key.toString() to outputFromJava(value) } + is List<*> -> output.map { outputFromJava(it) } + is Array<*> -> output.map { outputFromJava(it) } + // Write, then read: convertValue can return an embedded value, such as a byte[], unchanged. + else -> + outputFromJava( + JsonBaseModel.getMapper().let { + it.readValue(it.writeValueAsString(output), Any::class.java) + } + ) + } + + /** Returns the Java [JavaNodeInfo] copy of the Kotlin [info]. */ + fun nodeInfoToJava(info: KtNodeInfo): JavaNodeInfo = + JavaNodeInfo.builder() + .path(info.path) + .outputFor(info.outputFor.orEmpty()) + .messageAsOutput(info.messageAsOutput) + .build() + + /** + * Returns the Kotlin [KtNodeInfo] copy of the Java [info]. An empty `outputFor` reads as null, + * the Kotlin default, since the Java list is never null. + */ + fun nodeInfoFromJava(info: JavaNodeInfo): KtNodeInfo = + KtNodeInfo( + path = info.path(), + outputFor = info.outputFor().ifEmpty { null }, + messageAsOutput = info.messageAsOutput(), + ) + + /** Returns the Java [JavaRoute] copy of the Kotlin [route]. */ + fun routeToJava(route: KtRoute): JavaRoute = + when (route) { + is KtRoute.Tag -> JavaRoute.Tag(route.value) + is KtRoute.Num -> JavaRoute.Num(route.value) + is KtRoute.Flag -> JavaRoute.Flag(route.value) + KtRoute.Default -> JavaRoute.Default.INSTANCE + } + + /** Returns the Kotlin [KtRoute] copy of the Java [route]. */ + fun routeFromJava(route: JavaRoute): KtRoute = + when (route) { + is JavaRoute.Tag -> KtRoute.Tag(route.value()) + is JavaRoute.Num -> KtRoute.Num(route.value()) + is JavaRoute.Flag -> KtRoute.Flag(route.value()) + is JavaRoute.Default -> KtRoute.Default + } +} 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 b69dd387d..54323121d 100644 --- a/tokt/src/test/kotlin/com/google/adk/tokt/KtRunnerInteropTest.kt +++ b/tokt/src/test/kotlin/com/google/adk/tokt/KtRunnerInteropTest.kt @@ -27,12 +27,15 @@ 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.Context as KtContext import com.google.adk.kt.agents.InvocationContext as KtInvocationContext import com.google.adk.kt.agents.LlmAgent as KtLlmAgent import com.google.adk.kt.agents.ResumabilityConfig as KtResumabilityConfig import com.google.adk.kt.agents.RunConfig as KtRunConfig import com.google.adk.kt.agents.StreamingMode as KtStreamingMode import com.google.adk.kt.agents.TypedData +import com.google.adk.kt.annotations.ExperimentalWorkflowApi +import com.google.adk.kt.annotations.FrameworkInternalApi import com.google.adk.kt.apps.App as KtApp import com.google.adk.kt.callbacks.AfterModelCallback as KtAfterModelCallback import com.google.adk.kt.events.Event as KtEvent @@ -40,6 +43,7 @@ import com.google.adk.kt.events.EventActions as KtEventActions import com.google.adk.kt.models.LlmResponse as KtLlmResponse import com.google.adk.kt.runners.InMemoryRunner as KtInMemoryRunner import com.google.adk.kt.runners.Runner as KtRunner +import com.google.adk.kt.serialization.anyToJsonElement import com.google.adk.kt.sessions.GetSessionConfig as KtGetSessionConfig import com.google.adk.kt.sessions.InMemorySessionService as KtInMemorySessionService import com.google.adk.kt.sessions.Session as KtSession @@ -80,6 +84,13 @@ import com.google.adk.kt.types.VertexAISearch as KtVertexAISearch import com.google.adk.kt.types.VertexAISearchDataStoreSpec as KtVertexAISearchDataStoreSpec import com.google.adk.kt.types.VertexRagStore as KtVertexRagStore import com.google.adk.kt.types.VertexRagStoreRagResource as KtVertexRagStoreRagResource +import com.google.adk.kt.workflow.Edge as KtEdge +import com.google.adk.kt.workflow.Node as KtNode +import com.google.adk.kt.workflow.NodeInfo as KtNodeInfo +import com.google.adk.kt.workflow.RequestInput as KtRequestInput +import com.google.adk.kt.workflow.Route as KtRoute +import com.google.adk.kt.workflow.Start as KtStart +import com.google.adk.kt.workflow.Workflow as KtWorkflow import com.google.adk.memory.InMemoryMemoryService as JavaInMemoryMemoryService import com.google.adk.models.BaseLlm as JavaBaseLlm import com.google.adk.models.BaseLlmConnection as JavaBaseLlmConnection @@ -96,6 +107,7 @@ import com.google.adk.sessions.Session as JavaSession import com.google.adk.sessions.SessionKey as JavaSessionKey import com.google.adk.sessions.State as JavaState import com.google.adk.tokt.adapters.reconcileActionsToKt +import com.google.adk.tokt.codecs.ContentCodec import com.google.adk.tokt.codecs.EventCodec import com.google.adk.tokt.codecs.FunctionDeclarationCodec import com.google.adk.tokt.codecs.GenerateContentConfigCodec @@ -114,6 +126,8 @@ import com.google.adk.tokt.codecs.typedDataFromJava import com.google.adk.tools.BaseTool as JavaBaseTool import com.google.adk.tools.BaseToolset as JavaBaseToolset import com.google.adk.tools.ToolContext as JavaToolContext +import com.google.adk.workflow.NodeInfo as JavaNodeInfo +import com.google.adk.workflow.Route as JavaRoute import com.google.errorprone.annotations.CanIgnoreReturnValue import com.google.genai.types.CodeExecutionResult as GenaiCodeExecutionResult import com.google.genai.types.Content as GenaiContent @@ -150,6 +164,8 @@ import io.reactivex.rxjava3.core.Completable import io.reactivex.rxjava3.core.Flowable import io.reactivex.rxjava3.core.Maybe import io.reactivex.rxjava3.core.Single +import java.math.BigDecimal +import java.math.BigInteger import java.time.Instant import java.util.Optional import java.util.concurrent.ConcurrentHashMap @@ -506,6 +522,72 @@ class KtRunnerInteropTest { } } + /** A workflow node that emits [output] as its output and counts its runs. */ + @OptIn(ExperimentalWorkflowApi::class) + private class EmittingKtNode(override val name: String, private val output: Any) : KtNode { + var runs = 0 + private set + + override fun runNode(context: KtContext, nodeInput: Any?): Flow = flow { + runs++ + emit(output) + } + } + + /** A workflow node that asks the user for input, which pauses the workflow. */ + @OptIn(ExperimentalWorkflowApi::class) + private class AskingKtNode(override val name: String, private val interruptId: String) : KtNode { + override fun runNode(context: KtContext, nodeInput: Any?): Flow = flow { + emit(KtRequestInput(interruptId = interruptId, message = "Approve?")) + } + } + + /** A workflow node that selects [route] and emits no output. */ + @OptIn(ExperimentalWorkflowApi::class) + private class RoutingKtNode(override val name: String, private val route: KtRoute) : KtNode { + override fun runNode(context: KtContext, nodeInput: Any?): Flow = flow { + context.routes = listOf(route) + } + } + + /** + * A Java tool that selects the route "yes", either in place or by replacing its actions + * wholesale. + */ + private class JavaRoutingTool(private val replaceActions: Boolean) : + JavaBaseTool("java_route", "selects a workflow route") { + override fun declaration(): Optional = + Optional.of(GenaiFunctionDeclaration.builder().name("java_route").build()) + + @JvmSuppressWildcards + override fun runAsync( + args: Map, + toolContext: JavaToolContext, + ): Single> { + val route = listOf(JavaRoute.Tag("yes")) + if (replaceActions) { + toolContext.setActions(JavaEventActions.builder().route(route).build()) + } else { + toolContext.actions().setRoute(route) + } + return Single.just(mapOf("ok" to true)) + } + } + + /** A Java plugin whose afterTool callback selects the route "yes". */ + private class RoutingJavaPlugin : JavaBasePlugin("routing_plugin") { + @JvmSuppressWildcards + override fun afterToolCallback( + tool: JavaBaseTool, + toolArgs: Map, + toolContext: JavaToolContext, + result: Map, + ): Maybe> { + toolContext.actions().setRoute(listOf(JavaRoute.Tag("yes"))) + return Maybe.empty() + } + } + /** A Java plugin that requests an auth config the engine cannot represent, from before-agent. */ private class AuthConfigRequestingJavaPlugin : JavaBasePlugin("auth_config_plugin") { override fun beforeAgentCallback( @@ -969,6 +1051,192 @@ class KtRunnerInteropTest { assertFalse("forget me" in lastPrompt, "the rewound turn must not reach the model") } + @Test + @OptIn(ExperimentalWorkflowApi::class) + fun ktRunner_workflowThroughAnAdaptedJavaSessionService_reloadsNodeInfoOutputAndRoute() = + runBlocking { + // A resumed workflow reads these fields back, so the Java store must keep each one. + val sessions = JavaAdkToKt.asKtSessionService(JavaInMemorySessionService()) + val draft = EmittingKtNode("draft", mapOf("text" to "hello", "tags" to listOf("a", "b"))) + val router = RoutingKtNode("router", KtRoute.Tag("yes")) + val done = EmittingKtNode("done", KtContent.fromText("model", "finished")) + val workflow = + KtWorkflow( + name = "wf", + edges = + listOf( + KtEdge(KtStart, draft), + KtEdge(draft, router), + KtEdge(router, done, listOf(KtRoute.Tag("yes"))), + ), + ) + val runner = + KtInMemoryRunner( + app = KtApp(appName = "app", rootNode = workflow), + sessionService = sessions, + ) + + val emitted = runner.turn().filter { it.nodeInfo != null } + + assertTrue(emitted.any { it.output is Map<*, *> }, "a node should emit a value output") + assertTrue(emitted.any { it.output is KtContent }, "a node should emit a message output") + assertTrue(emitted.any { it.actions.route != null }, "a node should select a route") + val reloaded = + sessions.getSession(KtSessionKey("app", "u", "s"))?.events?.associateBy { it.id } + ?: fail("the adapted Java service should hold the session") + fun KtEvent.workflowFields() = Triple(nodeInfo, output, actions.route) + assertEquals( + emitted.map { it.workflowFields() }, + emitted.map { reloaded[it.id]?.workflowFields() }, + "each node event should reload with its node info, output and route", + ) + } + + @Test + @OptIn(ExperimentalWorkflowApi::class) + fun ktRunner_workflowResumedThroughAnAdaptedJavaSessionService_replaysFinishedNodes() = + runBlocking { + // The answering turn reloads the session from the Java store, so prep must replay from it. + val prep = EmittingKtNode("prep", "prepared") + val ask = AskingKtNode("ask", "ask_1") + val done = EmittingKtNode("done", "finished") + val workflow = + KtWorkflow( + name = "wf", + edges = listOf(KtEdge(KtStart, prep), KtEdge(prep, ask), KtEdge(ask, done)), + ) + val runner = + KtInMemoryRunner( + app = + KtApp( + appName = "app", + rootNode = workflow, + resumabilityConfig = KtResumabilityConfig(isResumable = true), + ), + sessionService = JavaAdkToKt.asKtSessionService(JavaInMemorySessionService()), + ) + runner.turn() + + val answer = + KtFunctionResponse( + name = KtRequestInput.FUNCTION_CALL_NAME, + id = "ask_1", + response = mapOf("approved" to true), + ) + runner + .runAsync( + userId = "u", + sessionId = "s", + newMessage = KtContent(role = "user", parts = listOf(KtPart(functionResponse = answer))), + ) + .toList() + + assertEquals(1, prep.runs, "prep finished before the pause, so it should replay") + assertEquals(1, done.runs, "the answer should let the workflow continue past the pause") + } + + @Test + @OptIn(ExperimentalWorkflowApi::class, FrameworkInternalApi::class) + fun asJavaRunner_workflowOutputs_reachJavaCodeAsJavaTypes() { + // Java callers and Java session services read these events, so outputs must use Java types. + val sessions = JavaInMemorySessionService() + val message = EmittingKtNode("message", KtContent.fromText("model", "drafted")) + val json = EmittingKtNode("json", anyToJsonElement(mapOf("k" to 1))) + val workflow = + KtWorkflow(name = "wf", edges = listOf(KtEdge(KtStart, message), KtEdge(message, json))) + val javaRunner = + KotlinAdkToJava.asJavaRunner( + KtInMemoryRunner( + app = KtApp(appName = "app", rootNode = workflow), + sessionService = JavaAdkToKt.asKtSessionService(sessions), + ) + ) + + val streamed = + javaRunner + .runAsync( + "u", + "s", + GenaiContent.builder().role("user").parts(GenaiPart.fromText("go")).build(), + JavaRunConfig.builder().autoCreateSession(true).build(), + ) + .toList() + .blockingGet() + val stored = + sessions.getSession("app", "u", "s", Optional.empty()).blockingGet()?.events().orEmpty() + + for ((source, events) in mapOf("runAsync" to streamed, "the session service" to stored)) { + val outputs = events.mapNotNull { it.output().getOrNull() } + assertEquals( + "drafted", + outputs.filterIsInstance().firstOrNull()?.text(), + "$source should hold the message output as a genai Content", + ) + assertContains(outputs, mapOf("k" to 1L), "$source should hold the JSON output as a map") + } + } + + @Test + fun ktRunner_routeSetByAJavaToolOrPlugin_routesTheWorkflow() = runBlocking { + // The engine routes on the agent node's own events, so the Java write must reach Kotlin. + assertEquals( + listOf("yes"), + routedNodesAfter(JavaAdkToKt.asKtTool(JavaRoutingTool(replaceActions = false))), + "a Java tool setting its route in place should route the workflow", + ) + assertEquals( + listOf("yes"), + routedNodesAfter(JavaAdkToKt.asKtTool(JavaRoutingTool(replaceActions = true))), + "a Java tool replacing its actions should route the workflow", + ) + assertEquals( + listOf("yes"), + routedNodesAfter(NativeKtEchoTool(), RoutingJavaPlugin()), + "a Java plugin setting the route after a tool call should route the workflow", + ) + } + + /** + * Runs a workflow whose agent node calls [tool] once, with [plugin] installed, and returns which + * of the nodes behind the routes "yes" and "no" ran. + */ + @OptIn(ExperimentalWorkflowApi::class) + private suspend fun routedNodesAfter( + tool: KtBaseTool, + plugin: JavaBasePlugin? = null, + ): List { + val agent = + KtLlmAgent( + name = "a", + model = + JavaAdkToKt.asKtModel( + SequentialJavaModel(listOf(modelFunctionCall(tool.name, emptyMap()), modelText("done"))) + ), + tools = listOf(tool), + ) + val yes = EmittingKtNode("yes", "took yes") + val no = EmittingKtNode("no", "took no") + val workflow = + KtWorkflow( + name = "wf", + edges = + listOf( + KtEdge(KtStart, agent), + KtEdge(agent, yes, listOf(KtRoute.Tag("yes"))), + KtEdge(agent, no, listOf(KtRoute.Tag("no"))), + ), + ) + KtInMemoryRunner( + KtApp( + appName = "app", + rootNode = workflow, + plugins = listOfNotNull(plugin?.let { JavaAdkToKt.asKtPlugin(it) }), + ) + ) + .turn() + return listOf(yes, no).filter { it.runs > 0 }.map { it.name } + } + @Test fun ktRunner_pluginBeforeToolCallback_firesForNativeKotlinTool_andCanDenyIt() = runBlocking { // A native Kotlin tool has no Java form, yet a Java plugin's beforeToolCallback must still @@ -4041,6 +4309,123 @@ class KtRunnerInteropTest { ) } + @Test + fun eventCodec_workflowFields_roundTripThroughTheJavaEvent() { + // Java code reads these fields too, so a Kotlin Content in the output crosses as a genai one. + val message = KtContent.fromText("model", "draft") + val output = mapOf("message" to message, "scores" to listOf(1, 2.5, null)) + val ktEvent = + KtEvent( + author = "wf", + output = output, + nodeInfo = + KtNodeInfo( + path = "wf@1/a@1", + outputFor = listOf("wf@1/a@1", "wf@1"), + messageAsOutput = true, + ), + actions = + KtEventActions( + route = listOf(KtRoute.Tag("yes"), KtRoute.Num(2), KtRoute.Flag(false), KtRoute.Default) + ), + ) + + val javaEvent = EventCodec.toJava(ktEvent) + val roundTripped = EventCodec.fromJava(javaEvent) + + assertEquals( + mapOf("message" to ContentCodec.toJava(message), "scores" to listOf(1, 2.5, null)), + javaEvent.output().getOrNull(), + ) + assertEquals(output, roundTripped.output) + assertEquals( + JavaNodeInfo.builder() + .path("wf@1/a@1") + .outputFor(listOf("wf@1/a@1", "wf@1")) + .messageAsOutput(true) + .build(), + javaEvent.nodeInfo().getOrNull(), + ) + assertEquals?>( + listOf( + JavaRoute.Tag("yes"), + JavaRoute.Num(2), + JavaRoute.Flag(false), + JavaRoute.Default.INSTANCE, + ), + javaEvent.actions().route().getOrNull(), + ) + assertEquals(ktEvent.nodeInfo, roundTripped.nodeInfo) + assertEquals(ktEvent.actions.route, roundTripped.actions.route) + } + + @Test + fun eventCodec_emptyRouteList_staysARoutingDecision() { + // An empty route list selects no edge, while no list makes no routing decision. + val selectsNoRoute = KtEvent(author = "a", actions = KtEventActions(route = emptyList())) + + assertEquals( + emptyList(), + EventCodec.fromJava(EventCodec.toJava(selectsNoRoute)).actions.route, + ) + assertNull(EventCodec.fromJava(EventCodec.toJava(KtEvent(author = "a"))).actions.route) + } + + @Test + @OptIn(FrameworkInternalApi::class) + fun eventCodec_jsonElementOutput_crossesAsPlainJavaValues() { + // Java code can't use a kotlinx JsonElement, and Jackson would write its internals. + val json: Any = anyToJsonElement(mapOf("k" to 1, "tags" to listOf("a", true))) + + val javaEvent = EventCodec.toJava(KtEvent(author = "wf", output = json)) + + val plain = mapOf("k" to 1L, "tags" to listOf("a", true)) + assertEquals(plain, javaEvent.output().getOrNull()) + assertContains(javaEvent.toJson(), "\"output\":{\"k\":1,\"tags\":[\"a\",true]}") + assertEquals(plain, EventCodec.fromJava(javaEvent).output) + } + + @Test + fun eventCodec_kotlinOutputWithNoJsonForm_throws() { + // An output with no JSON form fails here, just as when a Kotlin session service saves it. + assertFailsWith { + EventCodec.toJava(KtEvent(author = "wf", output = setOf("a"))) + } + } + + /** An output object that Jackson serializes through its getters, as ADK Java allows. */ + private data class Draft(val title: String, val tags: Set, val price: BigDecimal) + + @Test + fun eventCodec_javaObjectOutput_crossesAsItsJsonForm() { + // The Kotlin engine holds only JSON-native outputs, so a Java object must not reach it as is. + val draft = Draft(title = "t", tags = setOf("a"), price = BigDecimal("1.50")) + val javaEvent = JavaEvent.builder().author("wf").output(draft).build() + + val ktEvent = EventCodec.fromJava(javaEvent) + + val json = mapOf("title" to "t", "tags" to listOf("a"), "price" to 1.5) + assertEquals(json, ktEvent.output) + assertEquals(json, EventCodec.toJava(ktEvent).output().getOrNull()) + val bigInteger = JavaEvent.builder().author("wf").output(BigInteger.valueOf(7)).build() + assertEquals(7L, EventCodec.fromJava(bigInteger).output) + val array = JavaEvent.builder().author("wf").output(arrayOf("a", BigDecimal("2"))).build() + assertEquals(listOf("a", 2L), EventCodec.fromJava(array).output) + } + + @Test + fun ktEventActionsView_route_readsAndWritesTheKotlinActions() { + // A Java tool or plugin sees the current route, and its write reaches the engine at once. + val actions = KtEventActions(route = listOf(KtRoute.Num(1))) + val view = KtEventActionsToJavaView(actions) + + assertEquals?>(listOf(JavaRoute.Num(1)), view.route().getOrNull()) + view.setRoute(listOf(JavaRoute.Tag("yes"))) + assertEquals(listOf(KtRoute.Tag("yes")), actions.route) + view.setRoute(null) + assertNull(actions.route) + } + @Test fun schemaCodec_carriesEveryFacet_roundTrips() { // Every facet the Kotlin Schema models must survive Java -> Kotlin -> Java, or structured