diff --git a/README.md b/README.md index 7a19472a..fd8010f3 100644 --- a/README.md +++ b/README.md @@ -8,6 +8,7 @@ It contains the following modules: * [SpringBoot](/springboot): showcases SpringBoot autoconfig integration. * [SpringBoot Basic](/springboot-basic): Minimal sample showing SpringBoot autoconfig integration without any extra external dependencies. * [Spring AI](/springai): demonstrates the Temporal Spring AI integration — durable AI agents with chat models, tools, MCP servers, vector stores, and embeddings. +* [Google ADK](/googleadk): native ADK agents with durable conversations, tools, MCP, human approval, structured output, and asynchronous tool completion. * [Lambda Worker](/lambda-worker): demonstrates running a Temporal Java Worker inside AWS Lambda. ## Learn more about Temporal and Java SDK @@ -228,6 +229,27 @@ To run any of the SpringBoot samples in your Temporal Cloud namespace: 3. Follow the previous section from step 2 +### Running Google ADK samples + +The [Google ADK samples](googleadk) use the Public Preview +[`io.temporal:google-adk`](https://github.com/temporalio/ai-integrations/tree/main/java/google-adk) +integration. They cover durable conversations, Workflow and Activity tools, parallel agents, MCP, +human approval, structured output, and asynchronous tool completion. The module requires Temporal +Java SDK 1.40.0 or later. + +Run the Worker and a sample in separate terminals after starting `temporal server start-dev`: + +```bash +ADK_MODEL_MODE=offline ./gradlew :googleadk:worker +./gradlew :googleadk:starter +``` + +The default starter launches Agent Chat and prints its Workflow ID. Use the +[chat commands](googleadk/src/main/java/io/temporal/samples/googleadk/chat/README.md) to send messages, +receive each Update's reply, and finish the conversation. + +See the [sample README](googleadk/README.md) for the other scenarios and credential-free tests. + ### Running Spring AI Samples The Spring AI samples demonstrate the [Temporal Spring AI integration](https://github.com/temporalio/sdk-java/tree/master/temporal-spring-ai), which makes Spring AI agents durable on Temporal — model calls run as Temporal Activities recorded in Workflow history, and tools are dispatched per their type so they fit Workflow execution. diff --git a/googleadk/README.md b/googleadk/README.md new file mode 100644 index 00000000..d053850e --- /dev/null +++ b/googleadk/README.md @@ -0,0 +1,133 @@ +# Google ADK on Temporal + +Run native [Google ADK Java](https://github.com/google/adk-java) agents durably with +[`io.temporal:google-adk:0.1.1`](https://central.sonatype.com/artifact/io.temporal/google-adk). +The integration is **Public Preview**; APIs and configuration may change before General Availability. + +Model calls and MCP operations run as Temporal Activities. Agent orchestration, deterministic function +tools, and session state stay in the Workflow. Build Runners with `TemporalAdk.runnerBuilder()` and +await agent turns with `TemporalAdk.run()` so ADK uses Workflow time, identifiers, and threads. + +## Requirements + +- Java 17 or later. +- A local Temporal Service: `temporal server start-dev`. +- For live Gemini runs, `GOOGLE_API_KEY` in the Worker's environment, or Vertex AI configuration. + +Dependencies come from Maven Central. This module supplies Temporal SDK/testing in `[1.40.0,2.0)`, +preferring 1.40.0, independently of other samples. The integration brings ADK Java 1.12.0. + +## Run offline or with Gemini + +Run every command from the repository root. Start the Service in one terminal: + +```bash +temporal server start-dev +``` + +Start the Worker in a second terminal, choosing one model mode: + +```bash +ADK_MODEL_MODE=offline ./gradlew :googleadk:worker +``` + +```bash +GOOGLE_API_KEY=your_key ADK_MODEL_MODE=live ./gradlew :googleadk:worker +``` + +`live` is the default. Offline mode uses a deliberately small, stateless scripted `BaseLlm` for these +scenarios, including fixed Chicago weather and travel examples. It makes no provider requests and +needs no key. Live mode constructs Gemini only on the Worker and disables GenAI's HTTP retries so +Temporal Activities own retries. Stop the Worker before switching modes. + +Both modes run the same Worker registration, plugin, Workflows, Activities, messages, and result +path. The MCP scenario launches the real Java stdio server in both modes. Offline mode still needs +Temporal running, and a first Gradle build may download dependencies. + +Use a third terminal for the scenario commands below. Every Worker polls `google-adk-samples`. +The Temporal UI at shows model and tool Activities. Use a new Workflow ID +when repeating a command that names an ID. + +Start with Agent Chat in that third terminal: + +```bash +./gradlew :googleadk:starter +``` + +The starter prints the Workflow ID. Follow the [chat commands](src/main/java/io/temporal/samples/googleadk/chat) +to send Updates, receive their replies, and finish the conversation. + +## Coverage + +| Category | Scenario | Durable behavior | +| --- | --- | --- | +| Agent Chat | [chat](src/main/java/io/temporal/samples/googleadk/chat) | Updates return each turn's reply; accepted turns finish before a session snapshot carries the conversation through Continue-As-New. | +| Tools | [tools](src/main/java/io/temporal/samples/googleadk/tools) | A nondeterministic weather observation runs in an Activity; a pure temperature conversion runs in the Workflow. | +| MCP | [mcp](src/main/java/io/temporal/samples/googleadk/mcp) | List and call a bundled stateless Java stdio server through Activities. | +| Streaming | Unsupported | Plugin 0.1.1 buffers SSE responses until the model Activity completes; it cannot deliver output through Workflow Streams while the Activity is running. See [#257](https://github.com/temporalio/ai-integrations/issues/257). | +| Human in the Loop | [humanintheloop](src/main/java/io/temporal/samples/googleadk/humanintheloop) | A Signal approves or rejects ADK's pending tool confirmation; cancellation ends the durable wait. | +| Structured Output | [structuredoutput](src/main/java/io/temporal/samples/googleadk/structuredoutput) | ADK supplies the model's output schema; the Workflow validates JSON and returns a typed `TravelPlan` or a non-retryable failure. | +| Multi-Agent | [multiagent](src/main/java/io/temporal/samples/googleadk/multiagent) | ADK's `ParallelAgent` starts planner and reviewer model Activities concurrently on Workflow threads. | +| Sandboxes | Unsupported | The plugin does not durably manage an agent-facing code execution environment. ADK code executors that perform host I/O cannot run on the Workflow thread. | +| Observability | Unsupported | Native ADK tracing has no replay suppression; installing an OpenTelemetry SDK can export duplicate agent/model/tool spans during replay. See [#258](https://github.com/temporalio/ai-integrations/issues/258). | + +The plugin's [documented limits](https://github.com/temporalio/ai-integrations/blob/java/google-adk/v0.1.1/java/google-adk/README.md#limitations) +still apply to live token forwarding, agent-facing code execution, and native ADK global tracing. + +Start with [Agent Chat](src/main/java/io/temporal/samples/googleadk/chat); it is the introductory sample. +The suite also includes: + +- [longrunning](src/main/java/io/temporal/samples/googleadk/longrunning) uses the plugin's + `ActivityTools.createLongRunning`, `Hitl.longRunningCalls`, and `Hitl.functionResponse` APIs. + A submitted report remains pending until a correlated callback Signal delivers its result. + This differs from approving an operation before it executes: it resumes an operation already started. + +## Approve or reject a tool call + +See [humanintheloop](src/main/java/io/temporal/samples/googleadk/humanintheloop) for CLI commands to +start, query, decide, and cancel. Query until `pendingConfirmations` contains a request before +sending `decide`. A decision applies to the currently pending batch; premature Signals are ignored, and the first +decision for a pending batch wins. Rejection never schedules `ShipOrder`. + +## Continue a conversation as new + +See [chat](src/main/java/io/temporal/samples/googleadk/chat) for CLI commands. After three turns the +Workflow waits for accepted Updates to finish, exports its ADK session, and carries the session and +latest reply into a new Run. `finish` rejects new messages, drains accepted Updates, and returns the +last reply. Continue-As-New bounds each Run's history; the conversation snapshot itself still grows +and must fit Temporal's payload limits. + +## Test without provider credentials + +```bash +./gradlew :googleadk:test :googleadk:spotlessCheck :googleadk:build +``` + +Tests use the published `ScriptedModel` or the same `OfflineModel` as the runnable Worker, and +`TestWorkflowEnvironment`. They need no provider credentials or external Temporal Service. +The MCP tests launch the bundled local subprocess. + +Each supported scenario has a targeted test command in its README. Tests assert Activity scheduling, +retry behavior, function results, confirmation decisions and cancellation, schema rejection, +correlated callback completion, Update replies and concurrency, and session recovery across +Continue-As-New. Recorded histories replay on a fresh Worker without repeating model or tool calls. +`OfflineSamplesTest` runs every +scenario with `SampleWorker.register` and the actual offline provider. + +The `:googleadk:starter` task starts Agent Chat by default. It also accepts `tools`, `multiagent`, +`mcp`, `approval`, and `chat`, and their message commands. The scenario READMEs use Temporal CLI directly. +External Java clients that exchange ADK `Session` or `Content` payloads must also register +`GoogleAdkPlugin` to use its payload converter; ordinary CLI JSON inputs need no Java client. + +## Boundaries + +- Put host clocks, files, network services, and nondeterministic tools in Activities or MCP. The + weather Activity deliberately reads `Instant.now()` alongside illustrative, fixed conditions; + it does not call a real weather provider. Shipping and report submission also return sample data. +- Do not override agent-level schedulers/executors or use RxJava blocking APIs inside Workflows. +- Session snapshots preserve session-scoped state and events, not app-, user-, or temporary state. +- A cancellation stops the Workflow's wait. It does not undo a shipment or cancel an external report + already submitted; a real backend needs an explicit cancellation Activity. + +See the [integration reference](https://github.com/temporalio/ai-integrations/tree/main/java/google-adk) +for the full API and limitations. diff --git a/googleadk/build.gradle b/googleadk/build.gradle new file mode 100644 index 00000000..99651613 --- /dev/null +++ b/googleadk/build.gradle @@ -0,0 +1,45 @@ +// ADK samples consume the independently versioned integration from Maven Central. +dependencies { + implementation 'io.temporal:google-adk:0.1.1' + implementation 'com.fasterxml.jackson.core:jackson-databind' + implementation('io.temporal:temporal-sdk') { + version { + strictly '[1.40.0,2.0[' + prefer '1.40.0' + } + } + testImplementation('io.temporal:temporal-testing') { + version { + strictly '[1.40.0,2.0[' + prefer '1.40.0' + } + } + implementation 'io.modelcontextprotocol.sdk:mcp:2.0.0' + implementation 'io.modelcontextprotocol.sdk:mcp-json-jackson3:2.0.0' + runtimeOnly 'ch.qos.logback:logback-classic:1.5.18' + + testImplementation platform('org.junit:junit-bom:5.11.4') + testImplementation 'org.junit.jupiter:junit-jupiter' + testRuntimeOnly 'org.junit.platform:junit-platform-launcher' + + errorproneJavac 'com.google.errorprone:javac:9+181-r4173-1' + errorprone 'com.google.errorprone:error_prone_core:2.28.0' +} + +tasks.register('worker', JavaExec) { + classpath = sourceSets.main.runtimeClasspath + mainClass = 'io.temporal.samples.googleadk.SampleWorker' +} + +tasks.register('starter', JavaExec) { + classpath = sourceSets.main.runtimeClasspath + mainClass = 'io.temporal.samples.googleadk.SampleStarter' +} + +test { + // Gradle's test worker uses its own bootstrap classpath. The MCP subprocess + // needs the sample application's classpath, just as it has in the worker task. + doFirst { + systemProperty 'googleadk.echo.classpath', sourceSets.main.runtimeClasspath.asPath + } +} diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/AgentSupport.java b/googleadk/src/main/java/io/temporal/samples/googleadk/AgentSupport.java new file mode 100644 index 00000000..5f8885f9 --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/AgentSupport.java @@ -0,0 +1,25 @@ +package io.temporal.samples.googleadk; + +import com.google.adk.events.Event; +import com.google.genai.types.Content; +import java.util.List; + +/** Shared names and a deterministic helper for reading an agent's final answer. */ +public final class AgentSupport { + public static final String MODEL = "gemini-2.5-flash"; + public static final String TASK_QUEUE = "google-adk-samples"; + public static final String USER = "user-1"; + public static final String SESSION = "session-1"; + + private AgentSupport() {} + + public static String lastText(List events) { + for (int i = events.size() - 1; i >= 0; i--) { + String text = events.get(i).content().map(Content::text).orElse(""); + if (text != null && !text.isEmpty()) { + return text; + } + } + return ""; + } +} diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/OfflineModel.java b/googleadk/src/main/java/io/temporal/samples/googleadk/OfflineModel.java new file mode 100644 index 00000000..35f574a6 --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/OfflineModel.java @@ -0,0 +1,141 @@ +package io.temporal.samples.googleadk; + +import com.google.adk.models.BaseLlm; +import com.google.adk.models.BaseLlmConnection; +import com.google.adk.models.LlmRequest; +import com.google.adk.models.LlmResponse; +import com.google.genai.types.Content; +import com.google.genai.types.FunctionCall; +import com.google.genai.types.FunctionResponse; +import com.google.genai.types.Part; +import io.reactivex.rxjava3.core.Flowable; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +public final class OfflineModel extends BaseLlm { + public OfflineModel() { + super(AgentSupport.MODEL); + } + + @Override + public Flowable generateContent(LlmRequest request, boolean stream) { + return Flowable.just(respond(request)); + } + + private static LlmResponse respond(LlmRequest request) { + if (request.tools().containsKey("getWeather")) { + Optional> weather = functionResponse(request, "getWeather"); + if (weather.isEmpty()) { + return call("getWeather", Map.of("city", "Chicago")); + } + Optional> converted = functionResponse(request, "toFahrenheit"); + if (converted.isEmpty()) { + return call("toFahrenheit", Map.of("celsius", weather.get().get("celsius"))); + } + return text( + "Chicago is sunny: " + + weather.get().get("celsius") + + " C / " + + converted.get().get("fahrenheit") + + " F."); + } + if (request.tools().containsKey("shipOrder")) { + return functionResponse(request, "shipOrder") + .map( + response -> + response.containsKey("error") + ? text( + response.get("error").toString().contains("rejected") + ? "Order rejected." + : "Please approve the shipment.") + : text("Order shipped.")) + .orElseGet(() -> call("shipOrder", Map.of("orderId", "demo-order"))); + } + if (request.tools().containsKey("startReport")) { + return functionResponse(request, "startReport") + .map( + response -> + text( + "done".equals(response.get("status")) + ? "Report complete: " + response.get("result") + : "Report is pending.")) + .orElseGet(() -> call("startReport", Map.of("name", "sales"))); + } + if (request.tools().containsKey("echo")) { + return functionResponse(request, "echo") + .map(OfflineModel::echo) + .orElseGet(() -> call("echo", Map.of("text", lastUserText(request)))); + } + if (request.config().flatMap(config -> config.responseSchema()).isPresent()) { + return text( + "{\"city\":\"Chicago\",\"days\":2,\"activities\":[\"Walk the lakefront\",\"Visit a museum\"]}"); + } + if (request.getSystemInstructions().stream().anyMatch(text -> text.contains("\"planner\""))) { + return text("Visit Chicago's lakefront."); + } + if (request.getSystemInstructions().stream().anyMatch(text -> text.contains("\"reviewer\""))) { + return text("Bring a warm jacket."); + } + long turns = + request.contents().stream() + .filter(content -> content.role().orElse("").equals("user")) + .count(); + return text("turns=" + turns + "; latest: " + lastUserText(request)); + } + + static Optional> functionResponse(LlmRequest request, String name) { + return request.contents().stream() + .flatMap(content -> content.parts().orElse(List.of()).stream()) + .flatMap(part -> part.functionResponse().stream()) + .filter(response -> response.name().orElse("").equals(name)) + .map(FunctionResponse::response) + .flatMap(Optional::stream) + .reduce((first, last) -> last); + } + + private static LlmResponse echo(Map response) { + if (response.get("text_output") instanceof List outputs + && !outputs.isEmpty() + && outputs.get(0) instanceof Map first + && first.get("text") instanceof String value) { + return text(value); + } + throw new IllegalArgumentException("The echo tool returned no text output."); + } + + private static String lastUserText(LlmRequest request) { + return request.contents().stream() + .filter(content -> content.role().orElse("").equals("user")) + .flatMap(content -> content.parts().orElse(List.of()).stream()) + .flatMap(part -> part.text().stream()) + .reduce((first, last) -> last) + .orElse("Hello"); + } + + private static LlmResponse text(String value) { + return LlmResponse.builder() + .content(Content.builder().role("model").parts(List.of(Part.fromText(value))).build()) + .turnComplete(true) + .build(); + } + + private static LlmResponse call(String name, Map arguments) { + return LlmResponse.builder() + .content( + Content.builder() + .role("model") + .parts( + List.of( + Part.builder() + .functionCall(FunctionCall.builder().name(name).args(arguments).build()) + .build())) + .build()) + .build(); + } + + @Override + public BaseLlmConnection connect(LlmRequest request) { + throw new UnsupportedOperationException("The offline model has no live connection."); + } +} diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/SampleStarter.java b/googleadk/src/main/java/io/temporal/samples/googleadk/SampleStarter.java new file mode 100644 index 00000000..f2134339 --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/SampleStarter.java @@ -0,0 +1,119 @@ +package io.temporal.samples.googleadk; + +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.client.WorkflowOptions; +import io.temporal.googleadk.GoogleAdkPlugin; +import io.temporal.samples.googleadk.chat.ChatInput; +import io.temporal.samples.googleadk.chat.ChatWorkflow; +import io.temporal.samples.googleadk.humanintheloop.ApprovalWorkflow; +import io.temporal.samples.googleadk.mcp.McpWorkflow; +import io.temporal.samples.googleadk.multiagent.MultiAgentWorkflow; +import io.temporal.samples.googleadk.tools.ToolsWorkflow; +import io.temporal.serviceclient.WorkflowServiceStubs; +import java.util.Arrays; +import java.util.UUID; + +/** Starts samples or sends messages to their existing Workflow Executions. */ +public final class SampleStarter { + private SampleStarter() {} + + public static void main(String[] args) { + if (args.length == 0) { + args = new String[] {"chat"}; + } + WorkflowServiceStubs service = WorkflowServiceStubs.newLocalServiceStubs(); + try { + WorkflowClient client = + WorkflowClient.newInstance( + service, + WorkflowClientOptions.newBuilder() + .setPlugins(GoogleAdkPlugin.newBuilder().build()) + .build()); + String id = "google-adk-" + UUID.randomUUID(); + WorkflowOptions options = + WorkflowOptions.newBuilder() + .setTaskQueue(AgentSupport.TASK_QUEUE) + .setWorkflowId(id) + .build(); + switch (args[0]) { + case "tools": + System.out.println( + client + .newWorkflowStub(ToolsWorkflow.class, options) + .ask("What's the weather in Chicago?")); + break; + case "multiagent": + client + .newWorkflowStub(MultiAgentWorkflow.class, options) + .plan("Chicago") + .forEach(System.out::println); + break; + case "mcp": + System.out.println( + client.newWorkflowStub(McpWorkflow.class, options).echo("Hello from Temporal")); + break; + case "approval": + ApprovalWorkflow approval = client.newWorkflowStub(ApprovalWorkflow.class, options); + WorkflowClient.start(approval::fulfill, "Ship order demo-order, after confirmation."); + System.out.println("Workflow ID: " + id); + break; + case "chat": + // @@@SNIPSTART java-google-adk-starter + ChatWorkflow chat = client.newWorkflowStub(ChatWorkflow.class, options); + WorkflowClient.start(chat::chat, ChatInput.empty()); + System.out.println("Workflow ID: " + id); + // @@@SNIPEND + break; + case "pending": + requireArgs(args, 2); + client + .newWorkflowStub(ApprovalWorkflow.class, args[1]) + .pendingConfirmations() + .forEach(System.out::println); + break; + case "decide": + requireArgs(args, 3); + if (!args[2].equals("true") && !args[2].equals("false")) { + throw new IllegalArgumentException("Decision must be true or false."); + } + ApprovalWorkflow pending = client.newWorkflowStub(ApprovalWorkflow.class, args[1]); + pending.decide(Boolean.parseBoolean(args[2])); + System.out.println("Decision sent."); + break; + case "result": + requireArgs(args, 2); + System.out.println(client.newUntypedWorkflowStub(args[1]).getResult(String.class)); + break; + case "message": + requireArgs(args, 3); + System.out.println( + client + .newWorkflowStub(ChatWorkflow.class, args[1]) + .message(String.join(" ", Arrays.copyOfRange(args, 2, args.length)))); + break; + case "reply": + requireArgs(args, 2); + System.out.println(client.newWorkflowStub(ChatWorkflow.class, args[1]).lastReply()); + break; + case "finish": + requireArgs(args, 2); + ChatWorkflow conversation = client.newWorkflowStub(ChatWorkflow.class, args[1]); + conversation.finish(); + System.out.println( + io.temporal.client.WorkflowStub.fromTyped(conversation).getResult(String.class)); + break; + default: + throw new IllegalArgumentException("Unknown command: " + args[0]); + } + } finally { + service.shutdown(); + } + } + + private static void requireArgs(String[] args, int count) { + if (args.length < count) { + throw new IllegalArgumentException("Missing arguments; see googleadk/README.md."); + } + } +} diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/SampleWorker.java b/googleadk/src/main/java/io/temporal/samples/googleadk/SampleWorker.java new file mode 100644 index 00000000..7ebfe560 --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/SampleWorker.java @@ -0,0 +1,96 @@ +package io.temporal.samples.googleadk; + +import com.google.adk.models.BaseLlm; +import com.google.adk.models.Gemini; +import com.google.genai.Client; +import com.google.genai.types.HttpOptions; +import com.google.genai.types.HttpRetryOptions; +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.googleadk.GoogleAdkPlugin; +import io.temporal.samples.googleadk.chat.ChatWorkflowImpl; +import io.temporal.samples.googleadk.humanintheloop.ApprovalWorkflowImpl; +import io.temporal.samples.googleadk.humanintheloop.OrderActivitiesImpl; +import io.temporal.samples.googleadk.longrunning.LongRunningWorkflowImpl; +import io.temporal.samples.googleadk.longrunning.ReportActivitiesImpl; +import io.temporal.samples.googleadk.mcp.EchoMcpServer; +import io.temporal.samples.googleadk.mcp.McpWorkflowImpl; +import io.temporal.samples.googleadk.multiagent.MultiAgentWorkflowImpl; +import io.temporal.samples.googleadk.structuredoutput.StructuredOutputWorkflowImpl; +import io.temporal.samples.googleadk.tools.ToolsWorkflowImpl; +import io.temporal.samples.googleadk.tools.WeatherActivitiesImpl; +import io.temporal.serviceclient.WorkflowServiceStubs; +import io.temporal.worker.Worker; +import io.temporal.worker.WorkerFactory; +import java.util.function.Supplier; + +public final class SampleWorker { + private SampleWorker() {} + + public static void main(String[] args) { + String mode = System.getenv().getOrDefault("ADK_MODEL_MODE", "live"); + if (!mode.equals("offline") && !mode.equals("live")) { + throw new IllegalArgumentException("ADK_MODEL_MODE must be offline or live."); + } + // @@@SNIPSTART java-google-adk-worker + Client modelClient = + mode.equals("live") + ? Client.builder() + .httpOptions( + HttpOptions.builder() + .retryOptions(HttpRetryOptions.builder().attempts(1).build()) + .build()) + .build() + : null; + GoogleAdkPlugin plugin = + plugin( + modelClient == null + ? OfflineModel::new + : () -> + Gemini.builder().modelName(AgentSupport.MODEL).apiClient(modelClient).build()); + WorkflowServiceStubs service = WorkflowServiceStubs.newLocalServiceStubs(); + WorkflowClient client = + WorkflowClient.newInstance( + service, WorkflowClientOptions.newBuilder().setPlugins(plugin).build()); + WorkerFactory factory = WorkerFactory.newInstance(client); + Worker worker = factory.newWorker(AgentSupport.TASK_QUEUE); + register(worker); + factory.start(); + // @@@SNIPEND + Runtime.getRuntime() + .addShutdownHook( + new Thread( + () -> { + factory.shutdown(); + factory.awaitTermination(10, java.util.concurrent.TimeUnit.SECONDS); + if (modelClient != null) { + modelClient.close(); + } + service.shutdown(); + })); + System.out.println("Google ADK Worker: " + mode + " model, queue " + AgentSupport.TASK_QUEUE); + } + + // @@@SNIPSTART java-google-adk-worker-plugin + public static GoogleAdkPlugin plugin(Supplier models) { + return GoogleAdkPlugin.newBuilder() + .model(AgentSupport.MODEL, models) + .mcpToolset("echo", EchoMcpServer::toolset) + .build(); + } + + // @@@SNIPEND + + public static void register(Worker worker) { + worker.registerWorkflowImplementationTypes( + ToolsWorkflowImpl.class, + MultiAgentWorkflowImpl.class, + McpWorkflowImpl.class, + ApprovalWorkflowImpl.class, + ChatWorkflowImpl.class, + StructuredOutputWorkflowImpl.class, + LongRunningWorkflowImpl.class); + worker.registerActivitiesImplementations( + new WeatherActivitiesImpl(), new OrderActivitiesImpl(), new ReportActivitiesImpl()); + } +} diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/chat/ChatInput.java b/googleadk/src/main/java/io/temporal/samples/googleadk/chat/ChatInput.java new file mode 100644 index 00000000..f719f750 --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/chat/ChatInput.java @@ -0,0 +1,9 @@ +package io.temporal.samples.googleadk.chat; + +import com.google.adk.sessions.Session; + +public record ChatInput(Session session, String lastReply) { + public static ChatInput empty() { + return new ChatInput(null, ""); + } +} diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/chat/ChatWorkflow.java b/googleadk/src/main/java/io/temporal/samples/googleadk/chat/ChatWorkflow.java new file mode 100644 index 00000000..4288c192 --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/chat/ChatWorkflow.java @@ -0,0 +1,28 @@ +package io.temporal.samples.googleadk.chat; + +import io.temporal.workflow.QueryMethod; +import io.temporal.workflow.SignalMethod; +import io.temporal.workflow.UpdateMethod; +import io.temporal.workflow.UpdateValidatorMethod; +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; + +// @@@SNIPSTART java-google-adk-chat-interface +@WorkflowInterface +public interface ChatWorkflow { + @WorkflowMethod + String chat(ChatInput input); + + @UpdateMethod + String message(String text); + + @UpdateValidatorMethod(updateName = "message") + void validateMessage(String text); + + @SignalMethod + void finish(); + + @QueryMethod + String lastReply(); +} +// @@@SNIPEND diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/chat/ChatWorkflowImpl.java b/googleadk/src/main/java/io/temporal/samples/googleadk/chat/ChatWorkflowImpl.java new file mode 100644 index 00000000..0b1c2cc7 --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/chat/ChatWorkflowImpl.java @@ -0,0 +1,92 @@ +package io.temporal.samples.googleadk.chat; + +import com.google.adk.agents.LlmAgent; +import com.google.adk.runner.Runner; +import com.google.adk.sessions.Session; +import com.google.genai.types.Content; +import com.google.genai.types.Part; +import io.temporal.googleadk.TemporalAdk; +import io.temporal.googleadk.model.TemporalModel; +import io.temporal.googleadk.session.Sessions; +import io.temporal.samples.googleadk.AgentSupport; +import io.temporal.workflow.Workflow; +import io.temporal.workflow.WorkflowInit; +import io.temporal.workflow.WorkflowLock; + +// @@@SNIPSTART java-google-adk-chat-workflow +public class ChatWorkflowImpl implements ChatWorkflow { + private final WorkflowLock turnLock = Workflow.newWorkflowLock(); + private final Runner runner; + private boolean finished; + private String reply; + private int turns; + + @WorkflowInit + public ChatWorkflowImpl(ChatInput input) { + reply = input.lastReply(); + LlmAgent agent = + LlmAgent.builder() + .name("chat_assistant") + .model(TemporalModel.create(AgentSupport.MODEL)) + .instruction("Answer concisely, remembering the earlier turns in this conversation.") + .build(); + runner = TemporalAdk.runnerBuilder().appName("chat").agent(agent).build(); + if (input.session() != null) { + Sessions.importSession(runner.sessionService(), input.session()); + } + } + + @Override + public String chat(ChatInput input) { + Workflow.await( + () -> + finished || turns >= 3 || (turns > 0 && Workflow.getInfo().isContinueAsNewSuggested())); + Workflow.await(Workflow::isEveryHandlerFinished); + if (!finished) { + Session snapshot = + Sessions.exportSession( + runner.sessionService(), "chat", AgentSupport.USER, AgentSupport.SESSION); + Workflow.continueAsNew(new ChatInput(snapshot, reply)); + } + return reply; + } + + @Override + public String message(String text) { + turnLock.lock(); + try { + reply = + AgentSupport.lastText( + TemporalAdk.run( + runner, + AgentSupport.USER, + AgentSupport.SESSION, + Content.fromParts(Part.fromText(text)))); + turns++; + return reply; + } finally { + turnLock.unlock(); + } + } + + @Override + public void validateMessage(String text) { + if (finished) { + throw new IllegalStateException("The conversation is finishing."); + } + if (text == null || text.isBlank()) { + throw new IllegalArgumentException("The message must not be blank."); + } + } + + @Override + public void finish() { + finished = true; + } + + @Override + public String lastReply() { + return reply; + } +} +// @@@SNIPEND diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/chat/README.md b/googleadk/src/main/java/io/temporal/samples/googleadk/chat/README.md new file mode 100644 index 00000000..d82289d1 --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/chat/README.md @@ -0,0 +1,52 @@ +# Agent Chat + +Send each user message as a Workflow Update and receive that turn's reply directly. Concurrent turns +run one at a time on the same ADK session. After three completed turns, the Workflow waits for accepted +Updates to finish, exports the session, and Continues-As-New with the latest reply. + +Run commands from the repository root. Start `temporal server start-dev` in one terminal. +In another, choose the credential-free offline Worker or a live Gemini Worker: + +```bash +ADK_MODEL_MODE=offline ./gradlew :googleadk:worker +``` + +```bash +GOOGLE_API_KEY=your_key ADK_MODEL_MODE=live ./gradlew :googleadk:worker +``` + +Stop the Worker before switching modes. See the [suite README](../../../../../../../../README.md) +for prerequisites and integration limits. + +In a third terminal: + +```bash +temporal workflow start --workflow-id adk-chat --type ChatWorkflow --task-queue google-adk-samples --input '{"session":null,"lastReply":""}' +temporal workflow update execute --workflow-id adk-chat --name message --input '"My name is Ada."' +temporal workflow update execute --workflow-id adk-chat --name message --input '"Plan a trip to Chicago."' +temporal workflow update execute --workflow-id adk-chat --name message --input '"What should I pack?"' +``` + +Each Update waits for its own reply. In offline mode, the third response starts with `turns=3;`. +Continue the conversation across the Run transition, then finish: + +```bash +temporal workflow update execute --workflow-id adk-chat --name message --input '"What is my name?"' +temporal workflow signal --workflow-id adk-chat --name finish +temporal workflow result --workflow-id adk-chat +``` + +The fourth offline reply is `turns=4; latest: What is my name?`. Live Gemini also sees the imported +conversation and can recall Ada's name. `finish` rejects new messages and lets accepted Updates finish +before returning the last reply. Blank messages are rejected before invoking the model. + +The optional `lastReply` Query inspects the latest completed turn; Updates return their replies +without polling. The tests check concurrent Updates at a Run transition, draining accepted Updates +on finish, and replay of both histories. A session snapshot still grows with the conversation, so +bound or compact real conversations before they reach payload limits. + +Test without provider credentials or an external Service: + +```bash +./gradlew :googleadk:test --tests io.temporal.samples.googleadk.ChatWorkflowTest +``` diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/humanintheloop/ApprovalWorkflow.java b/googleadk/src/main/java/io/temporal/samples/googleadk/humanintheloop/ApprovalWorkflow.java new file mode 100644 index 00000000..d48f2ea0 --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/humanintheloop/ApprovalWorkflow.java @@ -0,0 +1,20 @@ +package io.temporal.samples.googleadk.humanintheloop; + +import io.temporal.googleadk.tool.Hitl; +import io.temporal.workflow.QueryMethod; +import io.temporal.workflow.SignalMethod; +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; +import java.util.List; + +@WorkflowInterface +public interface ApprovalWorkflow { + @WorkflowMethod + String fulfill(String question); + + @QueryMethod + List pendingConfirmations(); + + @SignalMethod + void decide(boolean confirmed); +} diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/humanintheloop/ApprovalWorkflowImpl.java b/googleadk/src/main/java/io/temporal/samples/googleadk/humanintheloop/ApprovalWorkflowImpl.java new file mode 100644 index 00000000..3026749f --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/humanintheloop/ApprovalWorkflowImpl.java @@ -0,0 +1,77 @@ +package io.temporal.samples.googleadk.humanintheloop; + +import com.google.adk.agents.LlmAgent; +import com.google.adk.events.Event; +import com.google.adk.runner.Runner; +import com.google.genai.types.Content; +import com.google.genai.types.Part; +import io.temporal.activity.ActivityOptions; +import io.temporal.common.RetryOptions; +import io.temporal.googleadk.TemporalAdk; +import io.temporal.googleadk.model.TemporalModel; +import io.temporal.googleadk.tool.ActivityTools; +import io.temporal.googleadk.tool.Hitl; +import io.temporal.samples.googleadk.AgentSupport; +import io.temporal.workflow.Workflow; +import java.time.Duration; +import java.util.List; + +public class ApprovalWorkflowImpl implements ApprovalWorkflow { + private List pending = List.of(); + private Boolean decision; + + // @@@SNIPSTART java-google-adk-approval-workflow + @Override + public String fulfill(String question) { + OrderActivities orders = + Workflow.newActivityStub( + OrderActivities.class, + ActivityOptions.newBuilder() + .setStartToCloseTimeout(Duration.ofSeconds(30)) + .setRetryOptions(RetryOptions.newBuilder().setMaximumAttempts(3).build()) + .build()); + LlmAgent agent = + LlmAgent.builder() + .name("fulfillment_assistant") + .model(TemporalModel.create(AgentSupport.MODEL)) + .instruction( + "Use shipOrder when asked to ship an order. Respect rejected confirmations.") + .tools(ActivityTools.create(orders, "shipOrder", true)) + .build(); + Runner runner = TemporalAdk.runnerBuilder().appName("approval").agent(agent).build(); + List events = + TemporalAdk.run( + runner, + AgentSupport.USER, + AgentSupport.SESSION, + Content.fromParts(Part.fromText(question))); + pending = Hitl.pendingConfirmations(events); + while (!pending.isEmpty()) { + Workflow.await(() -> decision != null); + Content response = + Hitl.confirmationResponse( + pending.stream() + .map(p -> new Hitl.ConfirmationDecision(p.functionCallId(), decision)) + .toArray(Hitl.ConfirmationDecision[]::new)); + decision = null; + pending = List.of(); + events = TemporalAdk.run(runner, AgentSupport.USER, AgentSupport.SESSION, response); + pending = Hitl.pendingConfirmations(events); + } + return AgentSupport.lastText(events); + } + + // @@@SNIPEND + + @Override + public List pendingConfirmations() { + return List.copyOf(pending); + } + + @Override + public void decide(boolean confirmed) { + if (!pending.isEmpty() && decision == null) { + decision = confirmed; + } + } +} diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/humanintheloop/OrderActivities.java b/googleadk/src/main/java/io/temporal/samples/googleadk/humanintheloop/OrderActivities.java new file mode 100644 index 00000000..90b328e2 --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/humanintheloop/OrderActivities.java @@ -0,0 +1,13 @@ +package io.temporal.samples.googleadk.humanintheloop; + +import com.google.adk.tools.Annotations.Schema; +import io.temporal.activity.ActivityInterface; +import io.temporal.activity.ActivityMethod; +import java.util.Map; + +@ActivityInterface +public interface OrderActivities { + @ActivityMethod(name = "ShipOrder") + Map shipOrder( + @Schema(name = "orderId", description = "The order to ship") String orderId); +} diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/humanintheloop/OrderActivitiesImpl.java b/googleadk/src/main/java/io/temporal/samples/googleadk/humanintheloop/OrderActivitiesImpl.java new file mode 100644 index 00000000..a5a21152 --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/humanintheloop/OrderActivitiesImpl.java @@ -0,0 +1,11 @@ +package io.temporal.samples.googleadk.humanintheloop; + +import java.util.Map; + +public class OrderActivitiesImpl implements OrderActivities { + @Override + public Map shipOrder(String orderId) { + // No order is actually shipped. A real integration would call the fulfillment API here. + return Map.of("orderId", orderId, "status", "shipped"); + } +} diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/humanintheloop/README.md b/googleadk/src/main/java/io/temporal/samples/googleadk/humanintheloop/README.md new file mode 100644 index 00000000..e25609da --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/humanintheloop/README.md @@ -0,0 +1,42 @@ +# Human approval + +ADK's `requireConfirmation` pauses a shipping tool before its Activity runs. The Workflow exposes +pending calls through a Query and waits durably for a decision Signal. Shipping returns sample data. + +Run commands from the repository root. Start `temporal server start-dev` in one terminal. +In another, choose the credential-free offline Worker or a live Gemini Worker: + +```bash +ADK_MODEL_MODE=offline ./gradlew :googleadk:worker +``` + +```bash +GOOGLE_API_KEY=your_key ADK_MODEL_MODE=live ./gradlew :googleadk:worker +``` + +Stop the Worker before switching modes. See the [suite README](../../../../../../../../README.md) +for prerequisites and integration limits. + +In a third terminal: + +```bash +temporal workflow start --workflow-id adk-approval --type ApprovalWorkflow --task-queue google-adk-samples --input '"Ship order demo-order, after confirmation."' +temporal workflow query --workflow-id adk-approval --name pendingConfirmations +temporal workflow signal --workflow-id adk-approval --name decide --input true +temporal workflow result --workflow-id adk-approval +``` + +Query again if the pending list is initially empty. Send `decide` only after a request appears; +use `--input false` on a fresh execution to reject instead. Offline results are `Order shipped.` or +`Order rejected.`. Live Gemini phrases the result. A rejection schedules no `ShipOrder` Activity. +A Signal decides the current pending batch; premature Signals are ignored and the first decision wins. If live +Gemini requests another confirmation, query and decide again. + +To cancel while waiting, run `temporal workflow cancel --workflow-id adk-approval` instead of deciding. +The Workflow cancels without shipping. Tests cover approval, rejection, cancellation, and replay. + +Test without provider credentials or an external Service: + +```bash +./gradlew :googleadk:test --tests io.temporal.samples.googleadk.ApprovalWorkflowTest +``` diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/longrunning/LongRunningWorkflow.java b/googleadk/src/main/java/io/temporal/samples/googleadk/longrunning/LongRunningWorkflow.java new file mode 100644 index 00000000..2c374086 --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/longrunning/LongRunningWorkflow.java @@ -0,0 +1,18 @@ +package io.temporal.samples.googleadk.longrunning; + +import io.temporal.workflow.QueryMethod; +import io.temporal.workflow.SignalMethod; +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; + +@WorkflowInterface +public interface LongRunningWorkflow { + @WorkflowMethod + String prepare(String request); + + @QueryMethod + String pendingCall(); + + @SignalMethod + void complete(String functionCallId, String result); +} diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/longrunning/LongRunningWorkflowImpl.java b/googleadk/src/main/java/io/temporal/samples/googleadk/longrunning/LongRunningWorkflowImpl.java new file mode 100644 index 00000000..50a9dbb2 --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/longrunning/LongRunningWorkflowImpl.java @@ -0,0 +1,77 @@ +package io.temporal.samples.googleadk.longrunning; + +import com.google.adk.agents.LlmAgent; +import com.google.adk.events.Event; +import com.google.adk.runner.Runner; +import com.google.genai.types.Content; +import com.google.genai.types.FunctionCall; +import com.google.genai.types.Part; +import io.temporal.activity.ActivityOptions; +import io.temporal.failure.ApplicationFailure; +import io.temporal.googleadk.TemporalAdk; +import io.temporal.googleadk.model.TemporalModel; +import io.temporal.googleadk.tool.ActivityTools; +import io.temporal.googleadk.tool.Hitl; +import io.temporal.samples.googleadk.AgentSupport; +import io.temporal.workflow.Workflow; +import java.time.Duration; +import java.util.List; +import java.util.Map; + +public class LongRunningWorkflowImpl implements LongRunningWorkflow { + private String callId = ""; + private String result; + + // @@@SNIPSTART java-google-adk-longrunning-workflow + @Override + public String prepare(String request) { + ReportActivities reports = + Workflow.newActivityStub( + ReportActivities.class, + ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(30)).build()); + LlmAgent agent = + LlmAgent.builder() + .name("report_assistant") + .model(TemporalModel.create(AgentSupport.MODEL)) + .instruction("Start exactly one report with startReport. Wait for its final result.") + .tools(ActivityTools.createLongRunning(reports, "startReport")) + .build(); + Runner runner = TemporalAdk.runnerBuilder().appName("reports").agent(agent).build(); + List events = + TemporalAdk.run( + runner, + AgentSupport.USER, + AgentSupport.SESSION, + Content.fromParts(Part.fromText(request))); + List pending = Hitl.longRunningCalls(events); + if (pending.size() != 1) { + throw ApplicationFailure.newNonRetryableFailure( + "Expected exactly one pending report.", "InvalidReportRequest"); + } + FunctionCall call = pending.get(0); + callId = call.id().orElseThrow(); + Workflow.await(() -> result != null); + events = + TemporalAdk.run( + runner, + AgentSupport.USER, + AgentSupport.SESSION, + Hitl.functionResponse(call, Map.of("status", "done", "result", result))); + callId = ""; + return AgentSupport.lastText(events); + } + + // @@@SNIPEND + + @Override + public String pendingCall() { + return callId; + } + + @Override + public void complete(String functionCallId, String value) { + if (!callId.isEmpty() && callId.equals(functionCallId) && result == null && value != null) { + result = value; + } + } +} diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/longrunning/README.md b/googleadk/src/main/java/io/temporal/samples/googleadk/longrunning/README.md new file mode 100644 index 00000000..050fd550 --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/longrunning/README.md @@ -0,0 +1,46 @@ +# Long-running tool completion + +An Activity-backed long-running ADK tool submits a report and returns a pending status. The +Workflow waits for a correlated callback Signal, then uses `Hitl.functionResponse` to resume the +original call. No polling Activity or open model connection is needed while waiting. + +The Activity simulates submission; it does not run a report service. A real service would save the +Workflow/call correlation and send the same Signal when its work finishes. + +Run commands from the repository root. Start `temporal server start-dev` in one terminal. +In another, choose the credential-free offline Worker or a live Gemini Worker: + +```bash +ADK_MODEL_MODE=offline ./gradlew :googleadk:worker +``` + +```bash +GOOGLE_API_KEY=your_key ADK_MODEL_MODE=live ./gradlew :googleadk:worker +``` + +Stop the Worker before switching modes. See the [suite README](../../../../../../../../README.md) +for prerequisites and integration limits. + +In a third terminal: + +```bash +temporal workflow start --workflow-id adk-report --type LongRunningWorkflow --task-queue google-adk-samples --input '"Prepare a sales report."' +temporal workflow query --workflow-id adk-report --name pendingCall +temporal workflow signal --workflow-id adk-report --name complete --input '"CALL_ID"' --input '"Revenue grew 10%."' +temporal workflow result --workflow-id adk-report +``` + +Query until `pendingCall` is nonempty, then replace `CALL_ID` with that value. While waiting, wrong IDs and premature +callbacks are ignored, and the first matching result wins. Offline mode returns `Report complete: Revenue grew 10%.`; +live Gemini summarizes the result. History contains one `StartReport` Activity, with model calls +before submission, after the pending status, and after the final callback. + +To cancel the durable wait, use `temporal workflow cancel --workflow-id adk-report`. This does not +cancel a job already submitted to a real backend; that needs a separate cancellation Activity. +Tests cover wrong-call correlation, cancellation, and replay without another submission. + +Test without provider credentials or an external Service: + +```bash +./gradlew :googleadk:test --tests io.temporal.samples.googleadk.LongRunningWorkflowTest +``` diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/longrunning/ReportActivities.java b/googleadk/src/main/java/io/temporal/samples/googleadk/longrunning/ReportActivities.java new file mode 100644 index 00000000..c95bb182 --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/longrunning/ReportActivities.java @@ -0,0 +1,15 @@ +package io.temporal.samples.googleadk.longrunning; + +import com.google.adk.tools.Annotations.Schema; +import io.temporal.activity.ActivityInterface; +import io.temporal.activity.ActivityMethod; +import java.util.Map; + +// @@@SNIPSTART java-google-adk-longrunning-activity +@ActivityInterface +public interface ReportActivities { + @ActivityMethod(name = "StartReport") + Map startReport( + @Schema(name = "name", description = "The report to prepare") String name); +} +// @@@SNIPEND diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/longrunning/ReportActivitiesImpl.java b/googleadk/src/main/java/io/temporal/samples/googleadk/longrunning/ReportActivitiesImpl.java new file mode 100644 index 00000000..76a64c4e --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/longrunning/ReportActivitiesImpl.java @@ -0,0 +1,10 @@ +package io.temporal.samples.googleadk.longrunning; + +import java.util.Map; + +public class ReportActivitiesImpl implements ReportActivities { + @Override + public Map startReport(String name) { + return Map.of("name", name, "status", "pending"); + } +} diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/mcp/EchoMcpServer.java b/googleadk/src/main/java/io/temporal/samples/googleadk/mcp/EchoMcpServer.java new file mode 100644 index 00000000..d044304e --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/mcp/EchoMcpServer.java @@ -0,0 +1,58 @@ +package io.temporal.samples.googleadk.mcp; + +import com.google.adk.tools.mcp.McpToolset; +import com.google.adk.tools.mcp.StdioServerParameters; +import io.modelcontextprotocol.json.jackson3.JacksonMcpJsonMapper; +import io.modelcontextprotocol.server.McpServer; +import io.modelcontextprotocol.server.McpSyncServer; +import io.modelcontextprotocol.server.transport.StdioServerTransportProvider; +import io.modelcontextprotocol.spec.McpSchema; +import java.nio.file.Path; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import tools.jackson.databind.json.JsonMapper; + +/** A local MCP echo server, launched by the worker without Node or another service. */ +public final class EchoMcpServer { + private EchoMcpServer() {} + + public static McpToolset toolset() { + String java = Path.of(System.getProperty("java.home"), "bin", "java").toString(); + String classpath = + System.getProperty("googleadk.echo.classpath", System.getProperty("java.class.path")); + return new McpToolset( + StdioServerParameters.builder() + .command(java) + .args(List.of("-cp", classpath, EchoMcpServer.class.getName())) + .build() + .toServerParameters()); + } + + public static void main(String[] args) throws InterruptedException { + JacksonMcpJsonMapper mapper = new JacksonMcpJsonMapper(JsonMapper.builder().build()); + McpSchema.Tool echo = + McpSchema.Tool.builder("echo") + .description("Echo the supplied text.") + .inputSchema( + Map.of( + "type", "object", + "properties", Map.of("text", Map.of("type", "string")), + "required", List.of("text"))) + .build(); + McpSyncServer server = + McpServer.sync(new StdioServerTransportProvider(mapper)) + .serverInfo("temporal-sample-echo", "1.0.0") + .capabilities(McpSchema.ServerCapabilities.builder().tools(false).build()) + .toolCall( + echo, + (exchange, request) -> + McpSchema.CallToolResult.builder() + .addTextContent((String) request.arguments().get("text")) + .isError(false) + .build()) + .build(); + Runtime.getRuntime().addShutdownHook(new Thread(server::closeGracefully)); + new CountDownLatch(1).await(); + } +} diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/mcp/McpWorkflow.java b/googleadk/src/main/java/io/temporal/samples/googleadk/mcp/McpWorkflow.java new file mode 100644 index 00000000..fcc2a77d --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/mcp/McpWorkflow.java @@ -0,0 +1,10 @@ +package io.temporal.samples.googleadk.mcp; + +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; + +@WorkflowInterface +public interface McpWorkflow { + @WorkflowMethod + String echo(String message); +} diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/mcp/McpWorkflowImpl.java b/googleadk/src/main/java/io/temporal/samples/googleadk/mcp/McpWorkflowImpl.java new file mode 100644 index 00000000..74daa71f --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/mcp/McpWorkflowImpl.java @@ -0,0 +1,32 @@ +package io.temporal.samples.googleadk.mcp; + +import com.google.adk.agents.LlmAgent; +import com.google.adk.runner.Runner; +import com.google.genai.types.Content; +import com.google.genai.types.Part; +import io.temporal.googleadk.TemporalAdk; +import io.temporal.googleadk.mcp.TemporalMcpToolset; +import io.temporal.googleadk.model.TemporalModel; +import io.temporal.samples.googleadk.AgentSupport; + +public class McpWorkflowImpl implements McpWorkflow { + // @@@SNIPSTART java-google-adk-mcp-workflow + @Override + public String echo(String message) { + LlmAgent agent = + LlmAgent.builder() + .name("echo_assistant") + .model(TemporalModel.create(AgentSupport.MODEL)) + .instruction("Call echo with the user's message, then return the echoed text.") + .tools(TemporalMcpToolset.newBuilder("echo").cacheTools(true).build()) + .build(); + Runner runner = TemporalAdk.runnerBuilder().appName("echo").agent(agent).build(); + return AgentSupport.lastText( + TemporalAdk.run( + runner, + AgentSupport.USER, + AgentSupport.SESSION, + Content.fromParts(Part.fromText(message)))); + } + // @@@SNIPEND +} diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/mcp/README.md b/googleadk/src/main/java/io/temporal/samples/googleadk/mcp/README.md new file mode 100644 index 00000000..3503a280 --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/mcp/README.md @@ -0,0 +1,36 @@ +# Stateless MCP over stdio + +The agent lists and calls `echo` on a bundled Java MCP server. The Worker starts the subprocess on +first use; Workflows only reference its registered toolset name. Both protocol operations are +Activities. No Node installation, remote MCP service, or listening port is required. + +Run commands from the repository root. Start `temporal server start-dev` in one terminal. +In another, choose the credential-free offline Worker or a live Gemini Worker: + +```bash +ADK_MODEL_MODE=offline ./gradlew :googleadk:worker +``` + +```bash +GOOGLE_API_KEY=your_key ADK_MODEL_MODE=live ./gradlew :googleadk:worker +``` + +Stop the Worker before switching modes. See the [suite README](../../../../../../../../README.md) +for prerequisites and integration limits. + +In a third terminal: + +```bash +temporal workflow execute --type McpWorkflow --task-queue google-adk-samples --input '"Hello MCP"' +``` + +Both modes echo `Hello MCP` after calling the real local server. History contains `ListMcpTools`, +`InvokeModel`, `CallMcpTool`, and a final `InvokeModel`, all under the `googleadk.` prefix. +`cacheTools(true)` avoids listing again during the second model turn. The plugin closes the toolset +and subprocess when the Worker factory shuts down. Tests exercise stdio and replay the history. + +Test without provider credentials or an external Service: + +```bash +./gradlew :googleadk:test --tests io.temporal.samples.googleadk.McpWorkflowTest +``` diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/multiagent/MultiAgentWorkflow.java b/googleadk/src/main/java/io/temporal/samples/googleadk/multiagent/MultiAgentWorkflow.java new file mode 100644 index 00000000..ebacd2fa --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/multiagent/MultiAgentWorkflow.java @@ -0,0 +1,11 @@ +package io.temporal.samples.googleadk.multiagent; + +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; +import java.util.List; + +@WorkflowInterface +public interface MultiAgentWorkflow { + @WorkflowMethod + List plan(String question); +} diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/multiagent/MultiAgentWorkflowImpl.java b/googleadk/src/main/java/io/temporal/samples/googleadk/multiagent/MultiAgentWorkflowImpl.java new file mode 100644 index 00000000..033b0251 --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/multiagent/MultiAgentWorkflowImpl.java @@ -0,0 +1,46 @@ +package io.temporal.samples.googleadk.multiagent; + +import com.google.adk.agents.LlmAgent; +import com.google.adk.agents.ParallelAgent; +import com.google.adk.runner.Runner; +import com.google.genai.types.Content; +import com.google.genai.types.Part; +import io.temporal.googleadk.TemporalAdk; +import io.temporal.googleadk.model.TemporalModel; +import io.temporal.samples.googleadk.AgentSupport; +import java.util.List; + +public class MultiAgentWorkflowImpl implements MultiAgentWorkflow { + // @@@SNIPSTART java-google-adk-multiagent-workflow + @Override + public List plan(String question) { + LlmAgent planner = + LlmAgent.builder() + .name("planner") + .model(TemporalModel.create(AgentSupport.MODEL)) + .instruction("Propose a short travel itinerary for the user's destination.") + .build(); + LlmAgent reviewer = + LlmAgent.builder() + .name("reviewer") + .model(TemporalModel.create(AgentSupport.MODEL)) + .instruction("List practical travel considerations for the user's destination.") + .build(); + ParallelAgent root = + ParallelAgent.builder().name("travel_team").subAgents(planner, reviewer).build(); + // The runner supplies workflow threads for ADK's parallel fan-out. + Runner runner = TemporalAdk.runnerBuilder().appName("travel").agent(root).build(); + return TemporalAdk.run( + runner, + AgentSupport.USER, + AgentSupport.SESSION, + Content.fromParts(Part.fromText(question))) + .stream() + .flatMap(event -> event.content().stream()) + .map(Content::text) + .filter(text -> text != null && !text.isEmpty()) + .sorted() + .toList(); + } + // @@@SNIPEND +} diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/multiagent/README.md b/googleadk/src/main/java/io/temporal/samples/googleadk/multiagent/README.md new file mode 100644 index 00000000..071c02de --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/multiagent/README.md @@ -0,0 +1,35 @@ +# Parallel agents + +ADK's `ParallelAgent` runs distinct planner and reviewer agents using the scheduler supplied by +`TemporalAdk.runnerBuilder()`. Their model Activities are scheduled concurrently, and the Workflow +returns their answers in a stable order. Do not set an agent-specific scheduler or executor. + +Run commands from the repository root. Start `temporal server start-dev` in one terminal. +In another, choose the credential-free offline Worker or a live Gemini Worker: + +```bash +ADK_MODEL_MODE=offline ./gradlew :googleadk:worker +``` + +```bash +GOOGLE_API_KEY=your_key ADK_MODEL_MODE=live ./gradlew :googleadk:worker +``` + +Stop the Worker before switching modes. See the [suite README](../../../../../../../../README.md) +for prerequisites and integration limits. + +In a third terminal: + +```bash +temporal workflow execute --type MultiAgentWorkflow --task-queue google-adk-samples --input '"Chicago"' +``` + +Offline mode returns `Bring a warm jacket.` and `Visit Chicago's lakefront.`. Live Gemini generates +both specialists' answers. The tests verify that both model Activities are scheduled before either +completes and that replay makes no further model calls. + +Test without provider credentials or an external Service: + +```bash +./gradlew :googleadk:test --tests io.temporal.samples.googleadk.MultiAgentWorkflowTest +``` diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/structuredoutput/README.md b/googleadk/src/main/java/io/temporal/samples/googleadk/structuredoutput/README.md new file mode 100644 index 00000000..f99dfcd4 --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/structuredoutput/README.md @@ -0,0 +1,36 @@ +# Structured output + +ADK's `outputSchema` supplies a JSON schema to the model. The Workflow parses the reply into a +`TravelPlan` and validates required values before returning the typed result. Malformed JSON, +fractional days, trailing JSON, and invalid fields fail with the non-retryable `InvalidTravelPlan` +failure instead of repeatedly failing a Workflow Task. + +Run commands from the repository root. Start `temporal server start-dev` in one terminal. +In another, choose the credential-free offline Worker or a live Gemini Worker: + +```bash +ADK_MODEL_MODE=offline ./gradlew :googleadk:worker +``` + +```bash +GOOGLE_API_KEY=your_key ADK_MODEL_MODE=live ./gradlew :googleadk:worker +``` + +Stop the Worker before switching modes. See the [suite README](../../../../../../../../README.md) +for prerequisites and integration limits. + +In a third terminal: + +```bash +temporal workflow execute --type StructuredOutputWorkflow --task-queue google-adk-samples --input '"Plan two days in Chicago."' +``` + +Offline mode returns `{"city":"Chicago","days":2,"activities":["Walk the lakefront","Visit a museum"]}`. +Live Gemini supplies its own plan with the same shape. History contains one `googleadk.InvokeModel` +Activity. Tests verify the schema on the worker-side request, typed results, invalid replies, and replay. + +Test without provider credentials or an external Service: + +```bash +./gradlew :googleadk:test --tests io.temporal.samples.googleadk.StructuredOutputWorkflowTest +``` diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/structuredoutput/StructuredOutputWorkflow.java b/googleadk/src/main/java/io/temporal/samples/googleadk/structuredoutput/StructuredOutputWorkflow.java new file mode 100644 index 00000000..0934bf41 --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/structuredoutput/StructuredOutputWorkflow.java @@ -0,0 +1,10 @@ +package io.temporal.samples.googleadk.structuredoutput; + +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; + +@WorkflowInterface +public interface StructuredOutputWorkflow { + @WorkflowMethod + TravelPlan plan(String request); +} diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/structuredoutput/StructuredOutputWorkflowImpl.java b/googleadk/src/main/java/io/temporal/samples/googleadk/structuredoutput/StructuredOutputWorkflowImpl.java new file mode 100644 index 00000000..7f4db0e0 --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/structuredoutput/StructuredOutputWorkflowImpl.java @@ -0,0 +1,74 @@ +package io.temporal.samples.googleadk.structuredoutput; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.DeserializationFeature; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.adk.SchemaUtils; +import com.google.adk.agents.LlmAgent; +import com.google.adk.runner.Runner; +import com.google.genai.types.Content; +import com.google.genai.types.Part; +import com.google.genai.types.Schema; +import io.temporal.failure.ApplicationFailure; +import io.temporal.googleadk.TemporalAdk; +import io.temporal.googleadk.model.TemporalModel; +import io.temporal.samples.googleadk.AgentSupport; +import java.util.List; +import java.util.Map; + +public class StructuredOutputWorkflowImpl implements StructuredOutputWorkflow { + // @@@SNIPSTART java-google-adk-structuredoutput-schema + private static final Schema PLAN_SCHEMA = + Schema.builder() + .type("OBJECT") + .properties( + Map.of( + "city", Schema.builder().type("STRING").build(), + "days", Schema.builder().type("INTEGER").minimum(1.0).maximum(7.0).build(), + "activities", + Schema.builder() + .type("ARRAY") + .items(Schema.builder().type("STRING").build()) + .minItems(1L) + .build())) + .required(List.of("city", "days", "activities")) + .build(); + + // @@@SNIPEND + + // @@@SNIPSTART java-google-adk-structuredoutput-workflow + @Override + public TravelPlan plan(String request) { + LlmAgent agent = + LlmAgent.builder() + .name("structured_planner") + .model(TemporalModel.create(AgentSupport.MODEL)) + .instruction( + "Return a travel plan as JSON with city, days (1 through 7), and activities.") + .outputSchema(PLAN_SCHEMA) + .build(); + Runner runner = TemporalAdk.runnerBuilder().appName("structured_plan").agent(agent).build(); + String json = + AgentSupport.lastText( + TemporalAdk.run( + runner, + AgentSupport.USER, + AgentSupport.SESSION, + Content.fromParts(Part.fromText(request)))); + try { + ObjectMapper mapper = + new ObjectMapper().enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS); + Map output = mapper.readValue(json, new TypeReference<>() {}); + if (output == null) { + throw new IllegalArgumentException("A plan must be a JSON object."); + } + SchemaUtils.validateMapOnSchema(output, PLAN_SCHEMA, false); + return mapper.convertValue(output, TravelPlan.class); + } catch (JsonProcessingException | IllegalArgumentException e) { + throw ApplicationFailure.newNonRetryableFailure( + "The model did not return a valid travel plan.", "InvalidTravelPlan"); + } + } + // @@@SNIPEND +} diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/structuredoutput/TravelPlan.java b/googleadk/src/main/java/io/temporal/samples/googleadk/structuredoutput/TravelPlan.java new file mode 100644 index 00000000..7a351dd1 --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/structuredoutput/TravelPlan.java @@ -0,0 +1,17 @@ +package io.temporal.samples.googleadk.structuredoutput; + +import java.util.List; + +public record TravelPlan(String city, int days, List activities) { + public TravelPlan { + if (city == null || city.isBlank() || days < 1 || days > 7) { + throw new IllegalArgumentException("A plan needs a city and between one and seven days."); + } + if (activities == null + || activities.isEmpty() + || activities.stream().anyMatch(activity -> activity == null || activity.isBlank())) { + throw new IllegalArgumentException("A plan needs at least one nonempty activity."); + } + activities = List.copyOf(activities); + } +} diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/tools/README.md b/googleadk/src/main/java/io/temporal/samples/googleadk/tools/README.md new file mode 100644 index 00000000..452d29a2 --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/tools/README.md @@ -0,0 +1,36 @@ +# Workflow and Activity tools + +The agent calls `getWeather` as a Temporal Activity, then `toFahrenheit` as a deterministic function +on the Workflow thread. The Activity reads a host timestamp and returns illustrative fixed weather; +replace its body with provider I/O for real observations. The conversion needs no Activity. + +Run commands from the repository root. Start `temporal server start-dev` in one terminal. +In another, choose the credential-free offline Worker or a live Gemini Worker: + +```bash +ADK_MODEL_MODE=offline ./gradlew :googleadk:worker +``` + +```bash +GOOGLE_API_KEY=your_key ADK_MODEL_MODE=live ./gradlew :googleadk:worker +``` + +Stop the Worker before switching modes. See the [suite README](../../../../../../../../README.md) +for prerequisites and integration limits. + +In a third terminal: + +```bash +temporal workflow execute --type ToolsWorkflow --task-queue google-adk-samples --input '"What is the weather in Chicago? Give Celsius and Fahrenheit."' +``` + +Offline mode returns `Chicago is sunny: 20 C / 68.0 F.`. Live Gemini uses the same tools. History shows +three `googleadk.InvokeModel` Activities and one `GetWeather`, with no Activity for `toFahrenheit`. +The tests inject a transient weather failure, verify its retry and both tool results, then replay +without repeating the observation. + +Test without provider credentials or an external Service: + +```bash +./gradlew :googleadk:test --tests io.temporal.samples.googleadk.ToolsWorkflowTest +``` diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/tools/TemperatureTools.java b/googleadk/src/main/java/io/temporal/samples/googleadk/tools/TemperatureTools.java new file mode 100644 index 00000000..9e972573 --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/tools/TemperatureTools.java @@ -0,0 +1,15 @@ +package io.temporal.samples.googleadk.tools; + +import com.google.adk.tools.Annotations.Schema; +import java.util.Map; + +public final class TemperatureTools { + private TemperatureTools() {} + + // @@@SNIPSTART java-google-adk-tools-function + public static Map toFahrenheit( + @Schema(name = "celsius", description = "Temperature in degrees Celsius") double celsius) { + return Map.of("fahrenheit", celsius * 9 / 5 + 32); + } + // @@@SNIPEND +} diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/tools/ToolsWorkflow.java b/googleadk/src/main/java/io/temporal/samples/googleadk/tools/ToolsWorkflow.java new file mode 100644 index 00000000..e5a90236 --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/tools/ToolsWorkflow.java @@ -0,0 +1,10 @@ +package io.temporal.samples.googleadk.tools; + +import io.temporal.workflow.WorkflowInterface; +import io.temporal.workflow.WorkflowMethod; + +@WorkflowInterface +public interface ToolsWorkflow { + @WorkflowMethod + String ask(String question); +} diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/tools/ToolsWorkflowImpl.java b/googleadk/src/main/java/io/temporal/samples/googleadk/tools/ToolsWorkflowImpl.java new file mode 100644 index 00000000..9a7e9fae --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/tools/ToolsWorkflowImpl.java @@ -0,0 +1,48 @@ +package io.temporal.samples.googleadk.tools; + +import com.google.adk.agents.LlmAgent; +import com.google.adk.runner.Runner; +import com.google.adk.tools.FunctionTool; +import com.google.genai.types.Content; +import com.google.genai.types.Part; +import io.temporal.activity.ActivityOptions; +import io.temporal.common.RetryOptions; +import io.temporal.googleadk.TemporalAdk; +import io.temporal.googleadk.model.TemporalModel; +import io.temporal.googleadk.tool.ActivityTools; +import io.temporal.samples.googleadk.AgentSupport; +import io.temporal.workflow.Workflow; +import java.time.Duration; + +public class ToolsWorkflowImpl implements ToolsWorkflow { + // @@@SNIPSTART java-google-adk-tools-workflow + @Override + public String ask(String question) { + WeatherActivities weather = + Workflow.newActivityStub( + WeatherActivities.class, + ActivityOptions.newBuilder() + .setStartToCloseTimeout(Duration.ofSeconds(30)) + .setRetryOptions(RetryOptions.newBuilder().setMaximumAttempts(3).build()) + .build()); + LlmAgent agent = + LlmAgent.builder() + .name("weather_assistant") + .model(TemporalModel.create(AgentSupport.MODEL)) + .instruction( + "Use getWeather for conditions and Celsius temperature, then toFahrenheit to" + + " convert the temperature. Report the conditions and both temperatures.") + .tools( + ActivityTools.create(weather, "getWeather"), + FunctionTool.create(TemperatureTools.class, "toFahrenheit")) + .build(); + Runner runner = TemporalAdk.runnerBuilder().appName("tools").agent(agent).build(); + return AgentSupport.lastText( + TemporalAdk.run( + runner, + AgentSupport.USER, + AgentSupport.SESSION, + Content.fromParts(Part.fromText(question)))); + } + // @@@SNIPEND +} diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/tools/WeatherActivities.java b/googleadk/src/main/java/io/temporal/samples/googleadk/tools/WeatherActivities.java new file mode 100644 index 00000000..8716a1df --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/tools/WeatherActivities.java @@ -0,0 +1,15 @@ +package io.temporal.samples.googleadk.tools; + +import com.google.adk.tools.Annotations.Schema; +import io.temporal.activity.ActivityInterface; +import io.temporal.activity.ActivityMethod; +import java.util.Map; + +// @@@SNIPSTART java-google-adk-weather-activity +@ActivityInterface +public interface WeatherActivities { + @ActivityMethod(name = "GetWeather") + Map getWeather( + @Schema(name = "city", description = "The city to look up") String city); +} +// @@@SNIPEND diff --git a/googleadk/src/main/java/io/temporal/samples/googleadk/tools/WeatherActivitiesImpl.java b/googleadk/src/main/java/io/temporal/samples/googleadk/tools/WeatherActivitiesImpl.java new file mode 100644 index 00000000..d556ba41 --- /dev/null +++ b/googleadk/src/main/java/io/temporal/samples/googleadk/tools/WeatherActivitiesImpl.java @@ -0,0 +1,12 @@ +package io.temporal.samples.googleadk.tools; + +import java.time.Instant; +import java.util.Map; + +public class WeatherActivitiesImpl implements WeatherActivities { + @Override + public Map getWeather(String city) { + return Map.of( + "city", city, "conditions", "sunny", "celsius", 20, "observedAt", Instant.now().toString()); + } +} diff --git a/googleadk/src/main/resources/logback.xml b/googleadk/src/main/resources/logback.xml new file mode 100644 index 00000000..825d6ab1 --- /dev/null +++ b/googleadk/src/main/resources/logback.xml @@ -0,0 +1,8 @@ + + + + System.err + %d{HH:mm:ss} %-5level %logger{36} - %msg%n + + + diff --git a/googleadk/src/test/java/io/temporal/samples/googleadk/AdkTestEnvironment.java b/googleadk/src/test/java/io/temporal/samples/googleadk/AdkTestEnvironment.java new file mode 100644 index 00000000..bcaa0c9b --- /dev/null +++ b/googleadk/src/test/java/io/temporal/samples/googleadk/AdkTestEnvironment.java @@ -0,0 +1,102 @@ +package io.temporal.samples.googleadk; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.google.adk.models.BaseLlm; +import com.google.adk.models.LlmRequest; +import io.temporal.api.history.v1.HistoryEvent; +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowClientOptions; +import io.temporal.client.WorkflowOptions; +import io.temporal.client.WorkflowStub; +import io.temporal.common.WorkflowExecutionHistory; +import io.temporal.googleadk.GoogleAdkPlugin; +import io.temporal.googleadk.testing.ScriptedModel; +import io.temporal.testing.TestEnvironmentOptions; +import io.temporal.testing.TestWorkflowEnvironment; +import io.temporal.worker.Worker; +import java.time.Duration; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.UUID; +import java.util.function.BooleanSupplier; + +/** Tests use the published scripted model and an in-process Temporal Service. */ +final class AdkTestEnvironment implements AutoCloseable { + final TestWorkflowEnvironment environment; + final WorkflowClient client; + final Worker worker; + final String taskQueue = "google-adk-test-" + UUID.randomUUID(); + private final BaseLlm model; + + AdkTestEnvironment(BaseLlm model, Class... workflows) { + this.model = model; + GoogleAdkPlugin plugin = SampleWorker.plugin(() -> model); + environment = newEnvironment(plugin); + client = environment.getWorkflowClient(); + worker = environment.newWorker(taskQueue); + if (workflows.length > 0) { + worker.registerWorkflowImplementationTypes(workflows); + } + } + + private static TestWorkflowEnvironment newEnvironment(GoogleAdkPlugin plugin) { + return TestWorkflowEnvironment.newInstance( + TestEnvironmentOptions.newBuilder() + .setWorkflowClientOptions(WorkflowClientOptions.newBuilder().setPlugins(plugin).build()) + .build()); + } + + T workflow(Class type) { + return client.newWorkflowStub( + type, + WorkflowOptions.newBuilder() + .setTaskQueue(taskQueue) + .setWorkflowId("google-adk-test-" + UUID.randomUUID()) + .setWorkflowExecutionTimeout(Duration.ofMinutes(2)) + .build()); + } + + WorkflowExecutionHistory history(Object workflow) { + return client.fetchHistory(WorkflowStub.fromTyped(workflow).getExecution().getWorkflowId()); + } + + void replay(WorkflowExecutionHistory history, Class implementation) throws Exception { + int requests = model instanceof ScriptedModel scripted ? scripted.requests().size() : 0; + // A new Worker proves replay works without cached Workflow or ADK session state. + GoogleAdkPlugin plugin = GoogleAdkPlugin.newBuilder().model(AgentSupport.MODEL, model).build(); + try (TestWorkflowEnvironment replay = newEnvironment(plugin)) { + Worker replayWorker = replay.newWorker("replay"); + replayWorker.registerWorkflowImplementationTypes(implementation); + replayWorker.replayWorkflowExecution(history); + } + if (model instanceof ScriptedModel scripted) { + assertEquals(requests, scripted.requests().size(), "replay must not call the model again"); + } + } + + static List activities(WorkflowExecutionHistory history) { + return history.getEvents().stream() + .filter(HistoryEvent::hasActivityTaskScheduledEventAttributes) + .map(event -> event.getActivityTaskScheduledEventAttributes().getActivityType().getName()) + .toList(); + } + + static Optional> functionResponse(LlmRequest request, String name) { + return OfflineModel.functionResponse(request, name); + } + + static void await(BooleanSupplier ready) throws InterruptedException { + for (int i = 0; i < 1000 && !ready.getAsBoolean(); i++) { + Thread.sleep(10); + } + assertTrue(ready.getAsBoolean(), "the Workflow did not reach the expected state"); + } + + @Override + public void close() { + environment.close(); + } +} diff --git a/googleadk/src/test/java/io/temporal/samples/googleadk/ApprovalWorkflowTest.java b/googleadk/src/test/java/io/temporal/samples/googleadk/ApprovalWorkflowTest.java new file mode 100644 index 00000000..dadaed6b --- /dev/null +++ b/googleadk/src/test/java/io/temporal/samples/googleadk/ApprovalWorkflowTest.java @@ -0,0 +1,137 @@ +package io.temporal.samples.googleadk; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowFailedException; +import io.temporal.client.WorkflowStub; +import io.temporal.common.WorkflowExecutionHistory; +import io.temporal.failure.CanceledFailure; +import io.temporal.googleadk.testing.ScriptedModel; +import io.temporal.samples.googleadk.humanintheloop.ApprovalWorkflow; +import io.temporal.samples.googleadk.humanintheloop.ApprovalWorkflowImpl; +import io.temporal.samples.googleadk.humanintheloop.OrderActivities; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +@Timeout(30) +class ApprovalWorkflowTest { + private static ScriptedModel model() { + return ScriptedModel.responding( + AgentSupport.MODEL, + request -> + AdkTestEnvironment.functionResponse(request, "shipOrder") + .map( + response -> { + if (response.containsKey("error")) { + return ScriptedModel.text( + String.valueOf(response.get("error")).contains("rejected") + ? "Order rejected." + : "Please approve the shipment."); + } + assertEquals("demo-order", response.get("orderId")); + return ScriptedModel.text("Order shipped."); + }) + .orElseGet( + () -> + ScriptedModel.functionCall("shipOrder", Map.of("orderId", "demo-order")))); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void orderShipsOnlyAfterConfirmation(boolean confirmed) throws Exception { + ScriptedModel model = model(); + AtomicInteger shipments = new AtomicInteger(); + OrderActivities orders = + order -> { + shipments.incrementAndGet(); + return Map.of("orderId", order, "status", "shipped"); + }; + try (AdkTestEnvironment test = new AdkTestEnvironment(model, ApprovalWorkflowImpl.class)) { + test.worker.registerActivitiesImplementations(orders); + test.environment.start(); + ApprovalWorkflow workflow = test.workflow(ApprovalWorkflow.class); + WorkflowClient.start(workflow::fulfill, "Ship demo-order"); + AdkTestEnvironment.await(() -> !workflow.pendingConfirmations().isEmpty()); + + assertEquals("shipOrder", workflow.pendingConfirmations().get(0).toolName()); + assertEquals( + "demo-order", + workflow + .pendingConfirmations() + .get(0) + .originalFunctionCall() + .args() + .orElseThrow() + .get("orderId")); + assertEquals(0, shipments.get()); + assertFalse(AdkTestEnvironment.activities(test.history(workflow)).contains("ShipOrder")); + + workflow.decide(confirmed); + assertEquals( + confirmed ? "Order shipped." : "Order rejected.", + WorkflowStub.fromTyped(workflow).getResult(String.class)); + assertEquals(confirmed ? 1 : 0, shipments.get()); + WorkflowExecutionHistory history = test.history(workflow); + assertEquals(confirmed, AdkTestEnvironment.activities(history).contains("ShipOrder")); + test.replay(history, ApprovalWorkflowImpl.class); + assertEquals(confirmed ? 1 : 0, shipments.get()); + } + } + + @Test + void aPrematureDecisionCannotApproveAFutureToolCall() throws Exception { + ScriptedModel model = model(); + AtomicInteger shipments = new AtomicInteger(); + try (AdkTestEnvironment test = new AdkTestEnvironment(model, ApprovalWorkflowImpl.class)) { + test.worker.registerActivitiesImplementations( + (OrderActivities) + order -> { + shipments.incrementAndGet(); + return Map.of("orderId", order, "status", "shipped"); + }); + test.environment.start(); + ApprovalWorkflow workflow = test.workflow(ApprovalWorkflow.class); + WorkflowStub stub = WorkflowStub.fromTyped(workflow); + stub.signalWithStart("decide", new Object[] {true}, new Object[] {"Ship demo-order"}); + AdkTestEnvironment.await(() -> !workflow.pendingConfirmations().isEmpty()); + workflow.decide(false); + assertEquals("Order rejected.", stub.getResult(String.class)); + assertEquals(0, shipments.get()); + test.replay(test.history(workflow), ApprovalWorkflowImpl.class); + } + } + + @Test + void cancellationWhileAwaitingApprovalNeverShipsTheOrder() throws Exception { + ScriptedModel model = model(); + AtomicInteger shipments = new AtomicInteger(); + try (AdkTestEnvironment test = new AdkTestEnvironment(model, ApprovalWorkflowImpl.class)) { + test.worker.registerActivitiesImplementations( + (OrderActivities) + order -> { + shipments.incrementAndGet(); + return Map.of("orderId", order, "status", "shipped"); + }); + test.environment.start(); + ApprovalWorkflow workflow = test.workflow(ApprovalWorkflow.class); + WorkflowClient.start(workflow::fulfill, "Ship demo-order"); + AdkTestEnvironment.await(() -> !workflow.pendingConfirmations().isEmpty()); + WorkflowStub stub = WorkflowStub.fromTyped(workflow); + stub.cancel(); + WorkflowFailedException failure = + assertThrows(WorkflowFailedException.class, () -> stub.getResult(String.class)); + assertInstanceOf(CanceledFailure.class, failure.getCause()); + assertEquals(0, shipments.get()); + assertFalse(AdkTestEnvironment.activities(test.history(workflow)).contains("ShipOrder")); + test.replay(test.history(workflow), ApprovalWorkflowImpl.class); + } + } +} diff --git a/googleadk/src/test/java/io/temporal/samples/googleadk/ChatWorkflowTest.java b/googleadk/src/test/java/io/temporal/samples/googleadk/ChatWorkflowTest.java new file mode 100644 index 00000000..5a5f0214 --- /dev/null +++ b/googleadk/src/test/java/io/temporal/samples/googleadk/ChatWorkflowTest.java @@ -0,0 +1,264 @@ +package io.temporal.samples.googleadk; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.google.adk.models.LlmRequest; +import io.temporal.api.common.v1.WorkflowExecution; +import io.temporal.api.enums.v1.EventType; +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowStub; +import io.temporal.client.WorkflowUpdateException; +import io.temporal.client.WorkflowUpdateHandle; +import io.temporal.client.WorkflowUpdateStage; +import io.temporal.common.WorkflowExecutionHistory; +import io.temporal.failure.ApplicationFailure; +import io.temporal.googleadk.testing.ScriptedModel; +import io.temporal.samples.googleadk.chat.ChatInput; +import io.temporal.samples.googleadk.chat.ChatWorkflow; +import io.temporal.samples.googleadk.chat.ChatWorkflowImpl; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.NullAndEmptySource; +import org.junit.jupiter.params.provider.ValueSource; + +@Timeout(30) +class ChatWorkflowTest { + private static List messages(LlmRequest request) { + return request.contents().stream() + .filter(content -> content.role().orElse("").equals("user")) + .flatMap(content -> content.parts().orElse(List.of()).stream()) + .flatMap(part -> part.text().stream()) + .toList(); + } + + private static WorkflowUpdateHandle message(ChatWorkflow workflow, String text) { + return WorkflowStub.fromTyped(workflow) + .startUpdate("message", WorkflowUpdateStage.ACCEPTED, String.class, text); + } + + private static void blockModel(CountDownLatch entered, CountDownLatch release) { + entered.countDown(); + try { + if (!release.await(10, TimeUnit.SECONDS)) { + throw new IllegalStateException("The test did not release the model call."); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(e); + } + } + + private static boolean continuedAsNew(WorkflowExecutionHistory history) { + return history.getEvents().stream() + .anyMatch( + event -> + event.getEventType() == EventType.EVENT_TYPE_WORKFLOW_EXECUTION_CONTINUED_AS_NEW); + } + + @Test + void updatesReturnTheirOwnRepliesAndReplay() throws Exception { + // @@@SNIPSTART java-google-adk-chat-test + ScriptedModel model = + ScriptedModel.of( + AgentSupport.MODEL, + ScriptedModel.text("First reply"), + ScriptedModel.text("Last reply")); + try (AdkTestEnvironment test = new AdkTestEnvironment(model, ChatWorkflowImpl.class)) { + test.environment.start(); + ChatWorkflow workflow = test.workflow(ChatWorkflow.class); + WorkflowClient.start(workflow::chat, ChatInput.empty()); + + assertEquals("First reply", workflow.message("one")); + assertEquals("Last reply", workflow.message("two")); + workflow.finish(); + assertEquals("Last reply", WorkflowStub.fromTyped(workflow).getResult(String.class)); + assertEquals(2, model.requests().size()); + test.replay(test.history(workflow), ChatWorkflowImpl.class); + } + // @@@SNIPEND + } + + @Test + void concurrentUpdatesFinishBeforeContinueAsNewAndKeepTheConversation() throws Exception { + CountDownLatch thirdEntered = new CountDownLatch(1); + CountDownLatch releaseThird = new CountDownLatch(1); + CountDownLatch fourthEntered = new CountDownLatch(1); + CountDownLatch releaseFourth = new CountDownLatch(1); + ScriptedModel model = + ScriptedModel.responding( + AgentSupport.MODEL, + request -> { + List turns = messages(request); + if (turns.size() == 3) { + blockModel(thirdEntered, releaseThird); + } else if (turns.size() == 4) { + blockModel(fourthEntered, releaseFourth); + } + return ScriptedModel.text(String.join(", ", turns)); + }); + try (AdkTestEnvironment test = new AdkTestEnvironment(model, ChatWorkflowImpl.class)) { + test.environment.start(); + ChatWorkflow starter = test.workflow(ChatWorkflow.class); + WorkflowExecution firstRun = WorkflowClient.start(starter::chat, ChatInput.empty()); + ChatWorkflow workflow = + test.client.newWorkflowStub(ChatWorkflow.class, firstRun.getWorkflowId()); + assertEquals("one", workflow.message("one")); + assertEquals("one, two", workflow.message("two")); + + WorkflowUpdateHandle third = message(workflow, "three"); + assertTrue(thirdEntered.await(10, TimeUnit.SECONDS)); + WorkflowUpdateHandle fourth = message(workflow, "four"); + assertEquals(3, AdkTestEnvironment.activities(test.history(workflow)).size()); + releaseThird.countDown(); + assertEquals("one, two, three", third.getResult()); + assertTrue(fourthEntered.await(10, TimeUnit.SECONDS)); + assertFalse( + continuedAsNew(test.client.fetchHistory(firstRun.getWorkflowId(), firstRun.getRunId()))); + releaseFourth.countDown(); + assertEquals("one, two, three, four", fourth.getResult()); + AdkTestEnvironment.await( + () -> + continuedAsNew( + test.client.fetchHistory(firstRun.getWorkflowId(), firstRun.getRunId()))); + + assertEquals("one, two, three, four", workflow.lastReply()); + assertEquals("one, two, three, four, five", workflow.message("five")); + workflow.finish(); + assertEquals( + "one, two, three, four, five", WorkflowStub.fromTyped(workflow).getResult(String.class)); + assertEquals(5, model.requests().size()); + WorkflowExecutionHistory firstHistory = + test.client.fetchHistory(firstRun.getWorkflowId(), firstRun.getRunId()); + WorkflowExecutionHistory lastHistory = test.history(workflow); + assertEquals( + firstRun.getRunId(), + lastHistory + .getEvents() + .get(0) + .getWorkflowExecutionStartedEventAttributes() + .getContinuedExecutionRunId()); + assertEquals(4, AdkTestEnvironment.activities(firstHistory).size()); + assertEquals(1, AdkTestEnvironment.activities(lastHistory).size()); + test.replay(firstHistory, ChatWorkflowImpl.class); + test.replay(lastHistory, ChatWorkflowImpl.class); + } finally { + releaseThird.countDown(); + releaseFourth.countDown(); + } + } + + @Test + void finishDrainsAcceptedUpdatesAndRejectsNewMessages() throws Exception { + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + ScriptedModel model = + ScriptedModel.responding( + AgentSupport.MODEL, + request -> { + List turns = messages(request); + if (turns.size() == 1) { + blockModel(entered, release); + } + return ScriptedModel.text(String.join(", ", turns)); + }); + try (AdkTestEnvironment test = new AdkTestEnvironment(model, ChatWorkflowImpl.class)) { + test.environment.start(); + ChatWorkflow workflow = test.workflow(ChatWorkflow.class); + WorkflowClient.start(workflow::chat, ChatInput.empty()); + WorkflowUpdateHandle first = message(workflow, "one"); + assertTrue(entered.await(10, TimeUnit.SECONDS)); + WorkflowUpdateHandle second = message(workflow, "two"); + assertEquals(1, AdkTestEnvironment.activities(test.history(workflow)).size()); + + workflow.finish(); + assertEquals("", workflow.lastReply()); + WorkflowUpdateException failure = + assertThrows(WorkflowUpdateException.class, () -> workflow.message("late")); + assertEquals( + IllegalStateException.class.getName(), + assertInstanceOf(ApplicationFailure.class, failure.getCause()).getType()); + release.countDown(); + assertEquals("one", first.getResult()); + assertEquals("one, two", second.getResult()); + assertEquals("one, two", WorkflowStub.fromTyped(workflow).getResult(String.class)); + assertEquals(2, model.requests().size()); + test.replay(test.history(workflow), ChatWorkflowImpl.class); + } finally { + release.countDown(); + } + } + + @ParameterizedTest + @NullAndEmptySource + @ValueSource(strings = {" "}) + void invalidMessagesAreRejectedWithoutCallingTheModel(String text) throws Exception { + ScriptedModel model = ScriptedModel.of(AgentSupport.MODEL, ScriptedModel.text("ok")); + try (AdkTestEnvironment test = new AdkTestEnvironment(model, ChatWorkflowImpl.class)) { + test.environment.start(); + ChatWorkflow workflow = test.workflow(ChatWorkflow.class); + WorkflowClient.start(workflow::chat, ChatInput.empty()); + WorkflowUpdateException failure = + assertThrows(WorkflowUpdateException.class, () -> workflow.message(text)); + assertEquals( + IllegalArgumentException.class.getName(), + assertInstanceOf(ApplicationFailure.class, failure.getCause()).getType()); + assertTrue(model.requests().isEmpty()); + assertEquals("ok", workflow.message("valid")); + workflow.finish(); + assertEquals("ok", WorkflowStub.fromTyped(workflow).getResult(String.class)); + assertEquals(1, model.requests().size()); + test.replay(test.history(workflow), ChatWorkflowImpl.class); + } + } + + @Test + void finishBeforeTheFirstTurnCompletesWithoutCallingTheModel() throws Exception { + ScriptedModel model = ScriptedModel.of(AgentSupport.MODEL, ScriptedModel.text("unused")); + try (AdkTestEnvironment test = new AdkTestEnvironment(model, ChatWorkflowImpl.class)) { + test.environment.start(); + ChatWorkflow workflow = test.workflow(ChatWorkflow.class); + WorkflowClient.start(workflow::chat, ChatInput.empty()); + workflow.finish(); + assertEquals("", WorkflowStub.fromTyped(workflow).getResult(String.class)); + assertTrue(model.requests().isEmpty()); + assertTrue(AdkTestEnvironment.activities(test.history(workflow)).isEmpty()); + test.replay(test.history(workflow), ChatWorkflowImpl.class); + } + } + + @Test + void failedUpdatesReleaseTheTurnLockAndTheConversationCanContinue() throws Exception { + AtomicInteger attempts = new AtomicInteger(); + ScriptedModel model = + ScriptedModel.responding( + AgentSupport.MODEL, + request -> { + if (attempts.incrementAndGet() == 1) { + throw ApplicationFailure.newNonRetryableFailure("Model unavailable", "ModelDenied"); + } + return ScriptedModel.text("Recovered"); + }); + try (AdkTestEnvironment test = new AdkTestEnvironment(model, ChatWorkflowImpl.class)) { + test.environment.start(); + ChatWorkflow workflow = test.workflow(ChatWorkflow.class); + WorkflowClient.start(workflow::chat, ChatInput.empty()); + assertThrows(WorkflowUpdateException.class, () -> workflow.message("one")); + assertEquals("", workflow.lastReply()); + assertEquals("Recovered", workflow.message("two")); + workflow.finish(); + assertEquals("Recovered", WorkflowStub.fromTyped(workflow).getResult(String.class)); + assertEquals(2, attempts.get()); + assertEquals(2, model.requests().size()); + test.replay(test.history(workflow), ChatWorkflowImpl.class); + } + } +} diff --git a/googleadk/src/test/java/io/temporal/samples/googleadk/LongRunningWorkflowTest.java b/googleadk/src/test/java/io/temporal/samples/googleadk/LongRunningWorkflowTest.java new file mode 100644 index 00000000..628f4ab4 --- /dev/null +++ b/googleadk/src/test/java/io/temporal/samples/googleadk/LongRunningWorkflowTest.java @@ -0,0 +1,93 @@ +package io.temporal.samples.googleadk; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowFailedException; +import io.temporal.client.WorkflowStub; +import io.temporal.failure.CanceledFailure; +import io.temporal.googleadk.testing.ScriptedModel; +import io.temporal.samples.googleadk.longrunning.LongRunningWorkflow; +import io.temporal.samples.googleadk.longrunning.LongRunningWorkflowImpl; +import io.temporal.samples.googleadk.longrunning.ReportActivities; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +@Timeout(30) +class LongRunningWorkflowTest { + private static ScriptedModel model() { + return ScriptedModel.responding( + AgentSupport.MODEL, + request -> + AdkTestEnvironment.functionResponse(request, "startReport") + .map( + response -> + ScriptedModel.text( + "done".equals(response.get("status")) + ? "Report complete: " + response.get("result") + : "Report is pending.")) + .orElseGet( + () -> ScriptedModel.functionCall("startReport", Map.of("name", "sales")))); + } + + @Test + void callbackResumesTheOriginalToolCallWithoutSubmittingAnotherJob() throws Exception { + ScriptedModel model = model(); + AtomicInteger submissions = new AtomicInteger(); + ReportActivities reports = + name -> { + submissions.incrementAndGet(); + return Map.of("name", name, "status", "pending"); + }; + try (AdkTestEnvironment test = new AdkTestEnvironment(model, LongRunningWorkflowImpl.class)) { + test.worker.registerActivitiesImplementations(reports); + test.environment.start(); + LongRunningWorkflow workflow = test.workflow(LongRunningWorkflow.class); + WorkflowClient.start(workflow::prepare, "Prepare a sales report."); + AdkTestEnvironment.await(() -> !workflow.pendingCall().isEmpty()); + String callId = workflow.pendingCall(); + test.replay(test.history(workflow), LongRunningWorkflowImpl.class); + workflow.complete("wrong-call", "ignore this"); + assertEquals(callId, workflow.pendingCall()); + workflow.complete(callId, "Revenue grew 10%."); + assertEquals( + "Report complete: Revenue grew 10%.", + WorkflowStub.fromTyped(workflow).getResult(String.class)); + assertEquals(1, submissions.get()); + assertEquals( + List.of( + "googleadk.InvokeModel", + "StartReport", + "googleadk.InvokeModel", + "googleadk.InvokeModel"), + AdkTestEnvironment.activities(test.history(workflow))); + test.replay(test.history(workflow), LongRunningWorkflowImpl.class); + assertEquals(1, submissions.get()); + } + } + + @Test + void cancellationEndsTheDurableWaitWithoutAnotherModelCall() throws Exception { + ScriptedModel model = model(); + try (AdkTestEnvironment test = new AdkTestEnvironment(model, LongRunningWorkflowImpl.class)) { + test.worker.registerActivitiesImplementations( + (ReportActivities) name -> Map.of("name", name, "status", "pending")); + test.environment.start(); + LongRunningWorkflow workflow = test.workflow(LongRunningWorkflow.class); + WorkflowClient.start(workflow::prepare, "Prepare a report."); + AdkTestEnvironment.await(() -> !workflow.pendingCall().isEmpty()); + WorkflowStub stub = WorkflowStub.fromTyped(workflow); + stub.cancel(); + WorkflowFailedException failure = + assertThrows(WorkflowFailedException.class, () -> stub.getResult(String.class)); + assertInstanceOf(CanceledFailure.class, failure.getCause()); + assertEquals(2, model.requests().size()); + test.replay(test.history(workflow), LongRunningWorkflowImpl.class); + } + } +} diff --git a/googleadk/src/test/java/io/temporal/samples/googleadk/McpWorkflowTest.java b/googleadk/src/test/java/io/temporal/samples/googleadk/McpWorkflowTest.java new file mode 100644 index 00000000..9c485e13 --- /dev/null +++ b/googleadk/src/test/java/io/temporal/samples/googleadk/McpWorkflowTest.java @@ -0,0 +1,48 @@ +package io.temporal.samples.googleadk; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.temporal.googleadk.testing.ScriptedModel; +import io.temporal.samples.googleadk.mcp.McpWorkflow; +import io.temporal.samples.googleadk.mcp.McpWorkflowImpl; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +@Timeout(30) +class McpWorkflowTest { + @Test + void realStdioEchoServerIsListedAndCalledThroughActivities() throws Exception { + ScriptedModel model = + ScriptedModel.responding( + AgentSupport.MODEL, + request -> + AdkTestEnvironment.functionResponse(request, "echo") + .map( + response -> { + assertTrue(response.toString().contains("hello MCP")); + return ScriptedModel.text("hello MCP"); + }) + .orElseGet( + () -> { + assertTrue(request.tools().containsKey("echo")); + return ScriptedModel.functionCall("echo", Map.of("text", "hello MCP")); + })); + try (AdkTestEnvironment test = new AdkTestEnvironment(model, McpWorkflowImpl.class)) { + test.environment.start(); + McpWorkflow workflow = test.workflow(McpWorkflow.class); + + assertEquals("hello MCP", workflow.echo("hello MCP")); + assertEquals( + List.of( + "googleadk.ListMcpTools", + "googleadk.InvokeModel", + "googleadk.CallMcpTool", + "googleadk.InvokeModel"), + AdkTestEnvironment.activities(test.history(workflow))); + test.replay(test.history(workflow), McpWorkflowImpl.class); + } + } +} diff --git a/googleadk/src/test/java/io/temporal/samples/googleadk/MultiAgentWorkflowTest.java b/googleadk/src/test/java/io/temporal/samples/googleadk/MultiAgentWorkflowTest.java new file mode 100644 index 00000000..ac2d6410 --- /dev/null +++ b/googleadk/src/test/java/io/temporal/samples/googleadk/MultiAgentWorkflowTest.java @@ -0,0 +1,51 @@ +package io.temporal.samples.googleadk; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import io.temporal.api.enums.v1.EventType; +import io.temporal.api.history.v1.HistoryEvent; +import io.temporal.common.WorkflowExecutionHistory; +import io.temporal.googleadk.testing.ScriptedModel; +import io.temporal.samples.googleadk.multiagent.MultiAgentWorkflow; +import io.temporal.samples.googleadk.multiagent.MultiAgentWorkflowImpl; +import java.util.List; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +@Timeout(30) +class MultiAgentWorkflowTest { + @Test + void bothAgentsScheduleTheirModelActivitiesTogetherAndReplay() throws Exception { + ScriptedModel model = + ScriptedModel.responding( + AgentSupport.MODEL, + request -> + ScriptedModel.text( + request.getSystemInstructions().stream() + .anyMatch(text -> text.contains("\"planner\"")) + ? "Visit the lakefront." + : "Bring a warm jacket.")); + try (AdkTestEnvironment test = new AdkTestEnvironment(model, MultiAgentWorkflowImpl.class)) { + test.environment.start(); + MultiAgentWorkflow workflow = test.workflow(MultiAgentWorkflow.class); + + assertEquals( + List.of("Bring a warm jacket.", "Visit the lakefront."), workflow.plan("Chicago")); + WorkflowExecutionHistory history = test.history(workflow); + List activityOrder = + history.getEvents().stream() + .map(HistoryEvent::getEventType) + .filter( + type -> + type == EventType.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED + || type == EventType.EVENT_TYPE_ACTIVITY_TASK_COMPLETED) + .toList(); + assertEquals( + List.of( + EventType.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED, + EventType.EVENT_TYPE_ACTIVITY_TASK_SCHEDULED), + activityOrder.subList(0, 2)); + test.replay(history, MultiAgentWorkflowImpl.class); + } + } +} diff --git a/googleadk/src/test/java/io/temporal/samples/googleadk/OfflineSamplesTest.java b/googleadk/src/test/java/io/temporal/samples/googleadk/OfflineSamplesTest.java new file mode 100644 index 00000000..dbf1bb01 --- /dev/null +++ b/googleadk/src/test/java/io/temporal/samples/googleadk/OfflineSamplesTest.java @@ -0,0 +1,59 @@ +package io.temporal.samples.googleadk; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import io.temporal.client.WorkflowClient; +import io.temporal.client.WorkflowStub; +import io.temporal.samples.googleadk.chat.ChatInput; +import io.temporal.samples.googleadk.chat.ChatWorkflow; +import io.temporal.samples.googleadk.humanintheloop.ApprovalWorkflow; +import io.temporal.samples.googleadk.longrunning.LongRunningWorkflow; +import io.temporal.samples.googleadk.mcp.McpWorkflow; +import io.temporal.samples.googleadk.multiagent.MultiAgentWorkflow; +import io.temporal.samples.googleadk.structuredoutput.StructuredOutputWorkflow; +import io.temporal.samples.googleadk.structuredoutput.TravelPlan; +import io.temporal.samples.googleadk.tools.ToolsWorkflow; +import java.util.List; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +@Timeout(60) +class OfflineSamplesTest { + @Test + void everyScenarioRunsWithTheProductionWorkerRegistrationAndOfflineProvider() throws Exception { + try (AdkTestEnvironment test = new AdkTestEnvironment(new OfflineModel())) { + SampleWorker.register(test.worker); + test.environment.start(); + assertEquals( + "Chicago is sunny: 20 C / 68.0 F.", + test.workflow(ToolsWorkflow.class).ask("What's the weather in Chicago?")); + assertEquals( + List.of("Bring a warm jacket.", "Visit Chicago's lakefront."), + test.workflow(MultiAgentWorkflow.class).plan("Chicago")); + assertEquals("Hello MCP", test.workflow(McpWorkflow.class).echo("Hello MCP")); + assertEquals( + new TravelPlan("Chicago", 2, List.of("Walk the lakefront", "Visit a museum")), + test.workflow(StructuredOutputWorkflow.class).plan("Plan two days in Chicago.")); + ChatWorkflow chat = test.workflow(ChatWorkflow.class); + WorkflowClient.start(chat::chat, ChatInput.empty()); + assertEquals("turns=1; latest: one", chat.message("one")); + assertEquals("turns=2; latest: two", chat.message("two")); + chat.finish(); + assertEquals("turns=2; latest: two", WorkflowStub.fromTyped(chat).getResult(String.class)); + + ApprovalWorkflow approval = test.workflow(ApprovalWorkflow.class); + WorkflowClient.start(approval::fulfill, "Ship demo-order."); + AdkTestEnvironment.await(() -> !approval.pendingConfirmations().isEmpty()); + approval.decide(true); + assertEquals("Order shipped.", WorkflowStub.fromTyped(approval).getResult(String.class)); + + LongRunningWorkflow report = test.workflow(LongRunningWorkflow.class); + WorkflowClient.start(report::prepare, "Prepare a sales report."); + AdkTestEnvironment.await(() -> !report.pendingCall().isEmpty()); + report.complete(report.pendingCall(), "Revenue grew 10%."); + assertEquals( + "Report complete: Revenue grew 10%.", + WorkflowStub.fromTyped(report).getResult(String.class)); + } + } +} diff --git a/googleadk/src/test/java/io/temporal/samples/googleadk/StructuredOutputWorkflowTest.java b/googleadk/src/test/java/io/temporal/samples/googleadk/StructuredOutputWorkflowTest.java new file mode 100644 index 00000000..4a0d7c16 --- /dev/null +++ b/googleadk/src/test/java/io/temporal/samples/googleadk/StructuredOutputWorkflowTest.java @@ -0,0 +1,82 @@ +package io.temporal.samples.googleadk; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import io.temporal.client.WorkflowFailedException; +import io.temporal.failure.ApplicationFailure; +import io.temporal.googleadk.testing.ScriptedModel; +import io.temporal.samples.googleadk.structuredoutput.StructuredOutputWorkflow; +import io.temporal.samples.googleadk.structuredoutput.StructuredOutputWorkflowImpl; +import io.temporal.samples.googleadk.structuredoutput.TravelPlan; +import java.util.List; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +@Timeout(30) +class StructuredOutputWorkflowTest { + @Test + void schemaReachesTheModelAndTheWorkflowReturnsATypedPlan() throws Exception { + ScriptedModel model = + ScriptedModel.responding( + AgentSupport.MODEL, + request -> { + var config = request.config().orElseThrow(); + assertEquals("application/json", config.responseMimeType().orElseThrow()); + assertEquals( + List.of("city", "days", "activities"), + config.responseSchema().orElseThrow().required().orElseThrow()); + return ScriptedModel.text( + "{\"city\":\"Chicago\",\"days\":2,\"activities\":[\"Walk the lakefront\"]}"); + }); + try (AdkTestEnvironment test = + new AdkTestEnvironment(model, StructuredOutputWorkflowImpl.class)) { + test.environment.start(); + StructuredOutputWorkflow workflow = test.workflow(StructuredOutputWorkflow.class); + assertEquals( + new TravelPlan("Chicago", 2, List.of("Walk the lakefront")), + workflow.plan("Plan two days in Chicago.")); + assertEquals( + List.of("googleadk.InvokeModel"), AdkTestEnvironment.activities(test.history(workflow))); + test.replay(test.history(workflow), StructuredOutputWorkflowImpl.class); + } + } + + @ParameterizedTest + @ValueSource( + strings = { + "not JSON", + "null", + "[]", + "{\"city\":123,\"days\":2,\"activities\":[\"Walk\"]}", + "{\"city\":\"Chicago\",\"days\":\"2\",\"activities\":[\"Walk\"]}", + "{\"city\":\"Chicago\",\"days\":8,\"activities\":[\"Walk\"]}", + "{\"city\":\"Chicago\",\"days\":2,\"activities\":[42]}", + "{\"city\":\"Chicago\",\"days\":2}", + "{\"city\":\"Chicago\",\"days\":0,\"activities\":[\"Walk\"]}", + "{\"city\":\"Chicago\",\"days\":2,\"activities\":[]}", + "{\"city\":\"Chicago\",\"days\":2.5,\"activities\":[\"Walk\"]}", + "{\"city\":\"Chicago\",\"days\":null,\"activities\":[\"Walk\"]}", + "{\"city\":\"Chicago\",\"days\":2,\"activities\":[\"Walk\"]} {}" + }) + void invalidModelOutputFailsTheWorkflowInsteadOfRetryingItsTask(String response) + throws Exception { + ScriptedModel model = ScriptedModel.of(AgentSupport.MODEL, ScriptedModel.text(response)); + try (AdkTestEnvironment test = + new AdkTestEnvironment(model, StructuredOutputWorkflowImpl.class)) { + test.environment.start(); + StructuredOutputWorkflow workflow = test.workflow(StructuredOutputWorkflow.class); + WorkflowFailedException failure = + assertThrows( + WorkflowFailedException.class, () -> workflow.plan("Plan two days in Chicago.")); + assertEquals( + "InvalidTravelPlan", + assertInstanceOf(ApplicationFailure.class, failure.getCause()).getType()); + assertEquals(1, model.requests().size()); + test.replay(test.history(workflow), StructuredOutputWorkflowImpl.class); + } + } +} diff --git a/googleadk/src/test/java/io/temporal/samples/googleadk/ToolsWorkflowTest.java b/googleadk/src/test/java/io/temporal/samples/googleadk/ToolsWorkflowTest.java new file mode 100644 index 00000000..6a74806b --- /dev/null +++ b/googleadk/src/test/java/io/temporal/samples/googleadk/ToolsWorkflowTest.java @@ -0,0 +1,107 @@ +package io.temporal.samples.googleadk; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import io.temporal.client.WorkflowFailedException; +import io.temporal.failure.ActivityFailure; +import io.temporal.failure.ApplicationFailure; +import io.temporal.googleadk.testing.ScriptedModel; +import io.temporal.samples.googleadk.tools.ToolsWorkflow; +import io.temporal.samples.googleadk.tools.ToolsWorkflowImpl; +import io.temporal.samples.googleadk.tools.WeatherActivities; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +@Timeout(30) +class ToolsWorkflowTest { + @Test + void weatherToolRetriesAsAnActivityAndItsResultReachesTheModel() throws Exception { + ScriptedModel model = + ScriptedModel.responding( + AgentSupport.MODEL, + request -> + AdkTestEnvironment.functionResponse(request, "toFahrenheit") + .map( + response -> { + assertEquals(68.0, ((Number) response.get("fahrenheit")).doubleValue()); + return ScriptedModel.text("Chicago is sunny: 20 C / 68 F."); + }) + .orElseGet( + () -> + AdkTestEnvironment.functionResponse(request, "getWeather") + .map( + response -> { + assertEquals("Chicago", response.get("city")); + assertEquals("sunny", response.get("conditions")); + return ScriptedModel.functionCall( + "toFahrenheit", + Map.of("celsius", response.get("celsius"))); + }) + .orElseGet( + () -> + ScriptedModel.functionCall( + "getWeather", Map.of("city", "Chicago"))))); + AtomicInteger attempts = new AtomicInteger(); + WeatherActivities weather = + city -> { + if (attempts.incrementAndGet() == 1) { + throw ApplicationFailure.newFailure("Temporary provider failure", "WeatherUnavailable"); + } + return Map.of("city", city, "conditions", "sunny", "celsius", 20); + }; + try (AdkTestEnvironment test = new AdkTestEnvironment(model, ToolsWorkflowImpl.class)) { + test.worker.registerActivitiesImplementations(weather); + test.environment.start(); + ToolsWorkflow workflow = test.workflow(ToolsWorkflow.class); + + assertEquals( + "Chicago is sunny: 20 C / 68 F.", workflow.ask("What's the weather in Chicago?")); + assertEquals(2, attempts.get()); + assertEquals(3, model.requests().size()); + assertEquals( + List.of( + "googleadk.InvokeModel", + "GetWeather", + "googleadk.InvokeModel", + "googleadk.InvokeModel"), + AdkTestEnvironment.activities(test.history(workflow))); + test.replay(test.history(workflow), ToolsWorkflowImpl.class); + assertEquals(2, attempts.get(), "replay must not repeat the weather call"); + } + } + + @Test + void aNonRetryableToolFailureFailsTheWorkflowWithoutAnotherModelTurn() throws Exception { + ScriptedModel model = + ScriptedModel.of( + AgentSupport.MODEL, + ScriptedModel.functionCall("getWeather", Map.of("city", "Chicago"))); + AtomicInteger attempts = new AtomicInteger(); + try (AdkTestEnvironment test = new AdkTestEnvironment(model, ToolsWorkflowImpl.class)) { + test.worker.registerActivitiesImplementations( + (WeatherActivities) + city -> { + attempts.incrementAndGet(); + throw ApplicationFailure.newNonRetryableFailure( + "Weather access denied", "WeatherDenied"); + }); + test.environment.start(); + ToolsWorkflow workflow = test.workflow(ToolsWorkflow.class); + WorkflowFailedException failure = + assertThrows( + WorkflowFailedException.class, () -> workflow.ask("What's the weather in Chicago?")); + ActivityFailure activityFailure = assertInstanceOf(ActivityFailure.class, failure.getCause()); + assertEquals( + "WeatherDenied", + assertInstanceOf(ApplicationFailure.class, activityFailure.getCause()).getType()); + assertEquals(1, attempts.get()); + assertEquals(1, model.requests().size()); + test.replay(test.history(workflow), ToolsWorkflowImpl.class); + } + } +} diff --git a/settings.gradle b/settings.gradle index e66f00a2..07fa40d9 100644 --- a/settings.gradle +++ b/settings.gradle @@ -1,6 +1,7 @@ rootProject.name = 'temporal-java-samples' include 'core' +include 'googleadk' include 'springai:basic' include 'springai:mcp' include 'springai:multimodel'