From 1705a109d5450b0283d465c2535355ea5ff15ab4 Mon Sep 17 00:00:00 2001 From: DABH Date: Fri, 24 Jul 2026 15:35:19 -0500 Subject: [PATCH 1/4] Bump google-adk floor to 2.5.0 Graph/dynamic workflow support relies on Workflow-as-Tool (added in ADK 2.4.0) and HITL resumption for standalone nodes and NodeTool plus related resumable-mode hardening (added in ADK 2.5.0). Existing contrib tests pass unchanged against 2.5.0. The exclude-newer-package exemption can be removed once 2.5.0 (published 2026-07-16) passes the two-week cooldown. --- pyproject.toml | 4 +++- uv.lock | 15 +++++++++------ 2 files changed, 12 insertions(+), 7 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index ba0abfc3d..a6c1747fe 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,7 +29,7 @@ grpc = ["grpcio>=1.48.2,<2"] opentelemetry = ["opentelemetry-api>=1.11.1,<2", "opentelemetry-sdk>=1.11.1,<2"] pydantic = ["pydantic>=2.0.0,<3"] openai-agents = ["openai-agents>=0.17.5", "mcp>=1.9.4, <2"] -google-adk = ["google-adk>=2.2.0,<3"] +google-adk = ["google-adk>=2.5.0,<3"] langgraph = ["langgraph>=1.1.0"] langsmith = ["langsmith>=0.7.34,<0.9"] lambda-worker-otel = [ @@ -267,3 +267,5 @@ exclude = ["temporalio/bridge/target/**/*", "temporalio/bridge/sdk-core/.git"] # Prevent uv commands from building the package by default package = false exclude-newer = "2 weeks" +# TODO: remove once google-adk 2.5.0 (published 2026-07-16) passes the cooldown +exclude-newer-package = { google-adk = false } diff --git a/uv.lock b/uv.lock index f3b9f8c9d..291159b75 100644 --- a/uv.lock +++ b/uv.lock @@ -12,6 +12,9 @@ resolution-markers = [ exclude-newer = "0001-01-01T00:00:00Z" # This has no effect and is included for backwards compatibility when using relative exclude-newer values. exclude-newer-span = "P2W" +[options.exclude-newer-package] +google-adk = false + [[package]] name = "aioboto3" version = "15.5.0" @@ -991,7 +994,7 @@ name = "exceptiongroup" version = "1.3.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "typing-extensions", marker = "python_full_version < '3.11'" }, + { name = "typing-extensions" }, ] sdist = { url = "https://files.pythonhosted.org/packages/50/79/66800aadf48771f6b62f7eb014e352e5d06856655206165d775e675a02c9/exceptiongroup-1.3.1.tar.gz", hash = "sha256:8b412432c6055b0b7d14c310000ae93352ed6754f70fa8f7c34141f91c4e3219", size = 30371, upload-time = "2025-11-21T23:01:54.787Z" } wheels = [ @@ -1258,7 +1261,7 @@ wheels = [ [[package]] name = "google-adk" -version = "2.2.0" +version = "2.5.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "aiosqlite" }, @@ -1286,9 +1289,9 @@ dependencies = [ { name = "watchdog" }, { name = "websockets" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/4a/65/3ff3f50b10dac3323ddecd694515e9f9ed345886e0eaf666d0e42c90748b/google_adk-2.2.0.tar.gz", hash = "sha256:04cb6318aba8829fe7c941ee1b456ccb4745253898c13595708c9eb07b4582ff", size = 3391545, upload-time = "2026-06-04T22:15:12.9Z" } +sdist = { url = "https://files.pythonhosted.org/packages/d7/8b/d014c98e987ed3a95ac3740d2b5c8e8e891bfd88c3ac2253fca9547f3b1f/google_adk-2.5.0.tar.gz", hash = "sha256:55b88cac9d5072d511fd3224e5f334e57fb2b0ae567507e531e03fdfb60c82c2", size = 3608134, upload-time = "2026-07-16T20:43:06.464Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/64/f5/44a3b20b17bac130497f2d1dde8b93c90cfc026983cd94f24488d540ea70/google_adk-2.2.0-py3-none-any.whl", hash = "sha256:ebdf3d931dc2b9c5b30d995358fc2ae99d59594c48a4aaf7496869ccd2c5f245", size = 3912613, upload-time = "2026-06-04T22:15:15.411Z" }, + { url = "https://files.pythonhosted.org/packages/52/fe/699d21edebd1305b6d23fd570140cf0cf921f34f66e4611d840684717c3a/google_adk-2.5.0-py3-none-any.whl", hash = "sha256:d247ca3639921a54a86feb797a88d08c1d2c9a60c3f5ff2805e49beb29a9cb8d", size = 4169976, upload-time = "2026-07-16T20:43:04.647Z" }, ] [[package]] @@ -2760,7 +2763,7 @@ wheels = [ [package.optional-dependencies] litellm = [ - { name = "litellm", marker = "python_full_version < '3.14'" }, + { name = "litellm" }, ] [[package]] @@ -4758,7 +4761,7 @@ dev = [ [package.metadata] requires-dist = [ { name = "aioboto3", marker = "extra == 'aioboto3'", specifier = ">=10.4.0" }, - { name = "google-adk", marker = "extra == 'google-adk'", specifier = ">=2.2.0,<3" }, + { name = "google-adk", marker = "extra == 'google-adk'", specifier = ">=2.5.0,<3" }, { name = "google-genai", marker = "extra == 'google-genai'", specifier = ">=2.10.0,<3.0.0" }, { name = "grpcio", marker = "extra == 'grpc'", specifier = ">=1.48.2,<2" }, { name = "langgraph", marker = "extra == 'langgraph'", specifier = ">=1.1.0" }, From 9017412445d6e88f5f615f56bf61d62818eb813d Mon Sep 17 00:00:00 2001 From: DABH Date: Fri, 24 Jul 2026 15:40:52 -0500 Subject: [PATCH 2/4] Add graph workflow, dynamic workflow, and durable HITL support for ADK v2 - activity_node() wraps a Temporal activity as an ADK FunctionNode for use in Workflow graphs and dynamic ctx.run_node() calls. - HITL helpers (HitlRequest, pending_hitl_requests, hitl_input_response, hitl_confirmation_response) cover ADK's pause/resume wire format so workflows can durably wait on human input via signals/updates. - setup_deterministic_runtime() additionally installs a workflow.random()- backed provider on ADK versions exposing the platform random seam. - README sections for graph workflows, dynamic workflows, durable HITL, and determinism notes. --- .../contrib/google_adk_agents/README.md | 150 ++++++++++++++ .../contrib/google_adk_agents/__init__.py | 10 + temporalio/contrib/google_adk_agents/_hitl.py | 193 ++++++++++++++++++ .../contrib/google_adk_agents/_plugin.py | 19 ++ .../contrib/google_adk_agents/workflow.py | 131 +++++++++++- 5 files changed, 502 insertions(+), 1 deletion(-) create mode 100644 temporalio/contrib/google_adk_agents/_hitl.py diff --git a/temporalio/contrib/google_adk_agents/README.md b/temporalio/contrib/google_adk_agents/README.md index 40ebb9aee..2a9222848 100644 --- a/temporalio/contrib/google_adk_agents/README.md +++ b/temporalio/contrib/google_adk_agents/README.md @@ -195,6 +195,156 @@ agent = Agent( ) ``` +## Graph Workflows (ADK v2) + +ADK v2's graph runtime (`google.adk.workflow`) runs inside Temporal workflows: +the scheduler is pure asyncio and executes deterministically on Temporal's +workflow event loop, while LLM calls (`TemporalModel`), MCP tools +(`TemporalMcpToolSet`), and activity-backed nodes leave the workflow as +activities. + +Use `activity_node(...)` to run a graph node as a Temporal activity. The +previous node's output is passed to the activity — directly for a +single-parameter activity, bound by name (from a dict) for multi-parameter +activities: + +```python +from google.adk.workflow import JoinNode, Workflow +from temporalio.contrib.google_adk_agents.workflow import activity_node + +fetch = activity_node(fetch_data, start_to_close_timeout=timedelta(seconds=30)) + +def summarize(node_input): # plain nodes run in-workflow: keep deterministic + return f"{node_input} summarized" + +graph = Workflow(name="pipeline", edges=[("START", fetch, summarize)]) +``` + +Conditional routing (`(router, {"KEY": handler, ...})`, `DEFAULT_ROUTE`), +parallel fan-out with `JoinNode`, and `LlmAgent` nodes (with +`mode="task"`/`"single_turn"`) all work — agent nodes route their model calls +through `TemporalModel` as usual. + +## Dynamic Workflows + +Dynamic nodes (`await ctx.run_node(...)` with loops, branches, and +`asyncio.gather`) work in-workflow; child-run caching reads only the +in-memory session, so re-entry after a HITL resume replays deterministically. + +```python +from google.adk.workflow import node + +@node(rerun_on_resume=True) +async def pipeline(ctx): + data = await ctx.run_node(fetch, "query") # activity_node child + results = await asyncio.gather( + *(ctx.run_node(worker, item) for item in data) # parallel children + ) + return results +``` + +On a HITL resume, a `rerun_on_resume=True` dynamic node re-executes its body +while completed children are skipped from the session cache. Place activity +invocations in child nodes (`activity_node`, `activity_tool`) rather than +inline in the dynamic node body, or make them idempotent — inline calls run +again on re-entry (ADK's documented at-least-once semantics). + +A `Workflow` with an `input_schema` can also be passed in an agent's +`tools=[...]` list (Workflow-as-Tool), letting the model invoke whole graphs +as tools. + +## Durable Human-in-the-Loop + +ADK pauses a run for human input (a node yielding `RequestInput`) or tool +confirmation (`FunctionTool(..., require_confirmation=True)`); in a Temporal +workflow that pause becomes a durable wait. The +`pending_hitl_requests` / `hitl_input_response` / `hitl_confirmation_response` +helpers cover the wire format; the wait itself is ordinary workflow code: + +```python +from temporalio.contrib.google_adk_agents import ( + HitlRequest, + hitl_input_response, + pending_hitl_requests, +) + +@workflow.defn +class ApprovalWorkflow: + def __init__(self) -> None: + self._pending: dict[str, HitlRequest] = {} + self._responses: dict[str, Any] = {} + + @workflow.query + def pending_requests(self) -> list[HitlRequest]: + return list(self._pending.values()) + + @workflow.update + def respond(self, interrupt_id: str, response: Any) -> None: + self._responses[interrupt_id] = response + + @workflow.run + async def run(self, prompt: str) -> str: + runner = Runner( + app_name="app", node=graph, session_service=InMemorySessionService() + ) + session = await runner.session_service.create_session( + app_name="app", user_id="user" + ) + message = types.Content(role="user", parts=[types.Part(text=prompt)]) + result = "" + while True: + async for event in runner.run_async( + user_id="user", session_id=session.id, new_message=message + ): + for request in pending_hitl_requests(event): + self._pending[request.interrupt_id] = request + if event.content and event.content.parts and event.content.parts[0].text: + result = event.content.parts[0].text + if not self._pending: + return result + await workflow.wait_condition( + lambda: any(i in self._responses for i in self._pending) + ) + parts = [ + hitl_input_response(i, self._responses.pop(i)) + for i in list(self._pending) + if i in self._responses + ] + for part in parts: + self._pending.pop(part.function_response.id) + message = types.Content(role="user", parts=parts) +``` + +Tool confirmation composes with `activity_tool` with no extra plumbing — +`FunctionTool(func=activity_tool(risky_activity, ...), require_confirmation=True)` +never schedules the activity until the human approves (answer with +`hitl_confirmation_response(interrupt_id, confirmed=True)`). MCP tools +requesting confirmation via `tool_context.request_confirmation(...)` flow +through the same loop. Partial responses are fine: unanswered requests stay +pending across `run_async` turns. + +## Determinism Notes + +- The plugin patches ADK's `google.adk.platform` time, uuid, and (on ADK + versions that expose it) random providers to `workflow.now()`, + `workflow.uuid4()`, and `workflow.random()` inside workflows. +- ADK node `timeout=`/`RetryConfig` map onto durable timers + (`asyncio.wait_for`/`asyncio.sleep`). For activity-backed nodes, prefer + Temporal activity timeouts and `retry_policy` via `activity_node(...)` + options; an ADK `RetryConfig` on top would retry on top of Temporal's own + activity retries, and an ADK node timeout cancels the in-flight activity. +- Never set `RunConfig.tool_thread_pool_config` inside a workflow — it runs + tools on threads, which breaks workflow determinism. Live/BIDI mode is + likewise unsupported in workflows. +- ADK resume is at-least-once: on a HITL resume, completed nodes fast-forward + from the in-memory session, but `rerun_on_resume=True` node bodies + re-execute. This is deterministic under Temporal replay; schedule side + effects through activities (retried/tracked by Temporal) or make them + idempotent. +- Very long HITL conversations grow the workflow history with each turn; + consider `continue-as-new` boundaries between `run_async` turns for + long-running chats. + ## Integration Points This integration provides comprehensive support for running Google ADK Agents within Temporal workflows while maintaining: diff --git a/temporalio/contrib/google_adk_agents/__init__.py b/temporalio/contrib/google_adk_agents/__init__.py index 3f236516b..f17d1b3c9 100644 --- a/temporalio/contrib/google_adk_agents/__init__.py +++ b/temporalio/contrib/google_adk_agents/__init__.py @@ -3,6 +3,12 @@ This module provides the necessary components to run ADK Agents within Temporal Workflows. """ +from temporalio.contrib.google_adk_agents._hitl import ( + HitlRequest, + hitl_confirmation_response, + hitl_input_response, + pending_hitl_requests, +) from temporalio.contrib.google_adk_agents._mcp import ( TemporalMcpToolSet, TemporalMcpToolSetProvider, @@ -14,7 +20,11 @@ __all__ = [ "GoogleAdkPlugin", + "HitlRequest", "TemporalMcpToolSet", "TemporalMcpToolSetProvider", "TemporalModel", + "hitl_confirmation_response", + "hitl_input_response", + "pending_hitl_requests", ] diff --git a/temporalio/contrib/google_adk_agents/_hitl.py b/temporalio/contrib/google_adk_agents/_hitl.py new file mode 100644 index 000000000..0fb60051e --- /dev/null +++ b/temporalio/contrib/google_adk_agents/_hitl.py @@ -0,0 +1,193 @@ +"""Human-in-the-loop helpers for Google ADK agents running in Temporal workflows. + +ADK pauses a run by emitting an event that carries a special function call +(``adk_request_input`` for human-input nodes, ``adk_request_confirmation`` for +tool confirmation, ``adk_request_credential`` for auth) and resumes when a +later user message answers it with a matching ``FunctionResponse``. Inside a +Temporal workflow the pause maps naturally onto a durable wait: collect the +pending requests from the events yielded by ``runner.run_async``, expose them +via a query, wait for responses via ``workflow.wait_condition`` on a signal or +update handler, then call ``runner.run_async`` again with the response parts. + +These helpers cover the wire format only; the wait topology stays ordinary +Temporal workflow code. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Literal, Mapping, Optional + +from google.adk.events import Event +from google.adk.tools.tool_confirmation import ToolConfirmation +from google.genai import types + +# The function-call names ADK uses on the wire for HITL pauses. +_REQUEST_INPUT_FUNCTION_CALL_NAME = "adk_request_input" +_REQUEST_CONFIRMATION_FUNCTION_CALL_NAME = "adk_request_confirmation" +_REQUEST_CREDENTIAL_FUNCTION_CALL_NAME = "adk_request_credential" + +_KIND_BY_FUNCTION_CALL_NAME: dict[ + str, Literal["input", "tool_confirmation", "credential"] +] = { + _REQUEST_INPUT_FUNCTION_CALL_NAME: "input", + _REQUEST_CONFIRMATION_FUNCTION_CALL_NAME: "tool_confirmation", + _REQUEST_CREDENTIAL_FUNCTION_CALL_NAME: "credential", +} + + +@dataclass(frozen=True) +class HitlRequest: + """A pending human-in-the-loop request extracted from an ADK event. + + .. warning:: + This class is experimental and may change in future versions. + Use with caution in production environments. + + Attributes: + kind: ``"input"`` for a human-input node's ``RequestInput``, + ``"tool_confirmation"`` for a tool confirmation request, + ``"credential"`` for an auth request. + interrupt_id: The id a response must reference. Pass it to + :func:`hitl_input_response` or :func:`hitl_confirmation_response`. + invocation_id: The ADK invocation that is paused on this request. + author: The agent/node that raised the request. + message: Human-readable prompt (``RequestInput.message`` or the + confirmation hint), if any. + payload: Custom payload attached to the request, if any. + response_schema: JSON schema the response must satisfy (input + requests only), if any. + original_function_call: For tool confirmations, the gated tool call + as ``{"name": ..., "args": ..., "id": ...}`` — useful for + displaying what is being approved. + """ + + kind: Literal["input", "tool_confirmation", "credential"] + interrupt_id: str + invocation_id: Optional[str] = None + author: Optional[str] = None + message: Optional[str] = None + payload: Optional[Any] = None + response_schema: Optional[dict[str, Any]] = None + original_function_call: Optional[dict[str, Any]] = None + + +def pending_hitl_requests(event: Event) -> list[HitlRequest]: + """Extracts pending human-in-the-loop requests from an ADK event. + + .. warning:: + This function is experimental and may change in future versions. + Use with caution in production environments. + + Call this on each event yielded by ``runner.run_async``. A non-empty + result means the run is pausing for the returned requests; once + ``run_async`` completes, resume by sending a new user message whose parts + answer them (see :func:`hitl_input_response` and + :func:`hitl_confirmation_response`). + + Args: + event: An event yielded by ``runner.run_async``. + + Returns: + The requests carried by this event; empty for ordinary events. + """ + if not event.long_running_tool_ids: + return [] + if not event.content or not event.content.parts: + return [] + requests: list[HitlRequest] = [] + for part in event.content.parts: + function_call = part.function_call + if not function_call or not function_call.id: + continue + kind = _KIND_BY_FUNCTION_CALL_NAME.get(function_call.name or "") + if kind is None: + continue + args = function_call.args or {} + message: Optional[str] = None + payload: Optional[Any] = None + response_schema: Optional[dict[str, Any]] = None + original_function_call: Optional[dict[str, Any]] = None + if kind == "input": + message = args.get("message") + payload = args.get("payload") + response_schema = args.get("response_schema") + elif kind == "tool_confirmation": + confirmation = args.get("toolConfirmation") or {} + message = confirmation.get("hint") + payload = confirmation.get("payload") + original_function_call = args.get("originalFunctionCall") + else: + payload = args + requests.append( + HitlRequest( + kind=kind, + interrupt_id=function_call.id, + invocation_id=event.invocation_id, + author=event.author, + message=message, + payload=payload, + response_schema=response_schema, + original_function_call=original_function_call, + ) + ) + return requests + + +def hitl_input_response(interrupt_id: str, response: Any) -> types.Part: + """Builds the message part answering a human-input (``RequestInput``) request. + + .. warning:: + This function is experimental and may change in future versions. + Use with caution in production environments. + + Compose one or more parts into ``types.Content(role="user", parts=[...])`` + and pass it as ``new_message`` to ``runner.run_async`` to resume the + paused run. Non-mapping values are wrapped as ``{"result": value}`` per + ADK's convention and unwrapped on delivery to the node. + + Args: + interrupt_id: The :attr:`HitlRequest.interrupt_id` being answered. + response: The human's response value. + """ + if isinstance(response, Mapping): + response_dict = dict(response) + else: + response_dict = {"result": response} + return types.Part( + function_response=types.FunctionResponse( + id=interrupt_id, + name=_REQUEST_INPUT_FUNCTION_CALL_NAME, + response=response_dict, + ) + ) + + +def hitl_confirmation_response( + interrupt_id: str, *, confirmed: bool, payload: Optional[Any] = None +) -> types.Part: + """Builds the message part answering a tool-confirmation request. + + .. warning:: + This function is experimental and may change in future versions. + Use with caution in production environments. + + Compose one or more parts into ``types.Content(role="user", parts=[...])`` + and pass it as ``new_message`` to ``runner.run_async`` to resume the + paused run. If ``confirmed`` is false the gated tool is not executed and + the model receives a rejection response instead. + + Args: + interrupt_id: The :attr:`HitlRequest.interrupt_id` being answered. + confirmed: Whether the human approved running the tool. + payload: Optional custom payload made available to the tool via + ``tool_context.tool_confirmation.payload``. + """ + confirmation = ToolConfirmation(confirmed=confirmed, payload=payload) + return types.Part( + function_response=types.FunctionResponse( + id=interrupt_id, + name=_REQUEST_CONFIRMATION_FUNCTION_CALL_NAME, + response=confirmation.model_dump(mode="json"), + ) + ) diff --git a/temporalio/contrib/google_adk_agents/_plugin.py b/temporalio/contrib/google_adk_agents/_plugin.py index 7344485c8..1a52cacb9 100644 --- a/temporalio/contrib/google_adk_agents/_plugin.py +++ b/temporalio/contrib/google_adk_agents/_plugin.py @@ -1,6 +1,7 @@ from __future__ import annotations import dataclasses +import random import time import uuid from collections.abc import AsyncIterator, Callable @@ -57,6 +58,24 @@ def _deterministic_id_provider() -> str: except Exception as e: print(f"Warning: Failed to set deterministic runtime providers: {e}") + try: + # Available on ADK versions that route retry jitter through the + # platform random seam; a no-op ImportError on older versions. + from google.adk.platform import set_random_provider + + _local_random = random.Random() + + def _deterministic_random_provider() -> random.Random: + if workflow.in_workflow(): + return workflow.random() + return _local_random + + set_random_provider(_deterministic_random_provider) + except ImportError: + pass + except Exception as e: + print(f"Warning: Failed to set deterministic random provider: {e}") + class GoogleAdkPlugin(SimplePlugin): """A Temporal Worker Plugin configured for ADK. diff --git a/temporalio/contrib/google_adk_agents/workflow.py b/temporalio/contrib/google_adk_agents/workflow.py index b1d150391..24fc32c04 100644 --- a/temporalio/contrib/google_adk_agents/workflow.py +++ b/temporalio/contrib/google_adk_agents/workflow.py @@ -2,11 +2,15 @@ import functools import inspect -from typing import Any, Callable +import typing +from typing import TYPE_CHECKING, Any, Callable, Optional import temporalio.workflow from temporalio import workflow +if TYPE_CHECKING: + from google.adk.workflow import FunctionNode + def activity_tool(activity_def: Callable, **kwargs: Any) -> Callable: """Decorator/Wrapper to wrap a Temporal Activity as an ADK Tool. @@ -55,3 +59,128 @@ async def wrapper(*args: Any, **kw: Any): setattr(wrapper, "__signature__", inspect.signature(activity_def)) return wrapper + + +def activity_node( + activity_def: Callable, + *, + name: Optional[str] = None, + rerun_on_resume: bool = False, + **kwargs: Any, +) -> "FunctionNode": + """Wraps a Temporal Activity as a node for an ADK workflow graph. + + .. warning:: + This function is experimental and may change in future versions. + Use with caution in production environments. + + The returned :class:`~google.adk.workflow.FunctionNode` can be placed in a + ``Workflow(edges=[...])`` graph or invoked from a dynamic node via + ``ctx.run_node(...)``. The previous node's output (``node_input``) is + passed to the activity: directly for a single-parameter activity, or bound + by parameter name for a multi-parameter activity (``node_input`` must then + be a dict). Outside a workflow (local ADK runs) the activity function is + invoked directly. + + Args: + activity_def: The ``@activity.defn`` function to run. + name: Node name; defaults to the activity function's name. + rerun_on_resume: Passed through to ``FunctionNode``. Keep the default + ``False`` so that on a human-in-the-loop resume the node is + fast-forwarded from the session instead of re-executing the + activity. + **kwargs: Activity execution options for + ``workflow.execute_activity`` (e.g. ``start_to_close_timeout``). + Prefer configuring retries here via ``retry_policy`` rather than + wrapping the node with an ADK ``RetryConfig``, which would retry + on top of Temporal's own activity retries. + """ + from google.adk.workflow import FunctionNode + + sig = inspect.signature(activity_def) + params = [ + p + for p in sig.parameters.values() + if p.kind + in (inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY) + ] + # Resolve hints against the activity's own module so that string + # annotations (e.g. under `from __future__ import annotations`) still + # resolve once copied onto the wrapper, whose globals differ. + try: + resolved_hints = typing.get_type_hints(activity_def) + except Exception: + resolved_hints = {} + + async def wrapper(node_input: Any = None) -> Any: + options = kwargs.copy() + + if not params: + activity_args: list[Any] = [] + elif len(params) == 1: + activity_args = [node_input] + else: + if not isinstance(node_input, dict): + raise TypeError( + f"Activity node '{activity_def.__name__}' takes" + f" {len(params)} parameters, so its input must be a dict" + f" of parameter names to values, got" + f" {type(node_input).__name__}." + ) + bound = sig.bind(**node_input) + bound.apply_defaults() + activity_args = list(bound.arguments.values()) + + if not temporalio.workflow.in_workflow(): + # Outside a workflow, like local adk runs, use the function directly. + result = activity_def(*activity_args) + if inspect.isawaitable(result): + return await result + return result + + if not activity_args: + return await workflow.execute_activity(activity_def, **options) + if len(activity_args) == 1: + return await workflow.execute_activity( + activity_def, activity_args[0], **options + ) + return await workflow.execute_activity( + activity_def, args=activity_args, **options + ) + + # ADK's FunctionNode binds parameters from the wrapper's signature and + # type hints: a single `node_input` parameter receives the previous + # node's output directly. Set metadata explicitly rather than via + # functools.wraps: copying the activity's multi-parameter signature (or a + # `__wrapped__` link, which ADK's type-hint resolution follows) would make + # ADK bind the activity's own parameters from workflow state instead. + wrapper.__name__ = name or activity_def.__name__ + wrapper.__qualname__ = wrapper.__name__ + wrapper.__doc__ = activity_def.__doc__ + input_annotation = ( + resolved_hints.get(params[0].name, Any) if len(params) == 1 else Any + ) + return_annotation = resolved_hints.get("return", Any) + wrapper.__annotations__ = { + "node_input": input_annotation, + "return": return_annotation, + } + setattr( + wrapper, + "__signature__", + inspect.Signature( + parameters=[ + inspect.Parameter( + "node_input", + inspect.Parameter.POSITIONAL_OR_KEYWORD, + default=None, + annotation=input_annotation, + ) + ], + return_annotation=return_annotation, + ), + ) + + return FunctionNode( + func=wrapper, name=wrapper.__name__, rerun_on_resume=rerun_on_resume + ) From 3e44374b56619948b8af52c548f2fac623e9c4f5 Mon Sep 17 00:00:00 2001 From: DABH Date: Fri, 24 Jul 2026 16:29:54 -0500 Subject: [PATCH 3/4] Install ADK platform providers as process-wide defaults MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ADK holds its platform time/uuid/random providers in ContextVars, and the public set_*_provider helpers only affect the calling context. Temporal executes workflow code on executor threads whose contexts never see the run_context call, so the deterministic providers were previously never active inside workflows: event/function-call ids came from stdlib uuid4 and Event timestamps from wall-clock time. This went unnoticed because Temporal replay compares command sequences, not payloads — but any flow whose control depends on a generated id (HITL resume matching recorded responses by interrupt/function-call id) diverged on replay and hung. Rebind each platform module's ContextVar with the deterministic provider as its default. Context-local set_*_provider calls still override it, and the providers fall back to real primitives outside workflows. --- .../contrib/google_adk_agents/README.md | 7 +++ .../contrib/google_adk_agents/_plugin.py | 44 ++++++++++++++++--- 2 files changed, 45 insertions(+), 6 deletions(-) diff --git a/temporalio/contrib/google_adk_agents/README.md b/temporalio/contrib/google_adk_agents/README.md index 2a9222848..a820af31e 100644 --- a/temporalio/contrib/google_adk_agents/README.md +++ b/temporalio/contrib/google_adk_agents/README.md @@ -323,6 +323,13 @@ requesting confirmation via `tool_context.request_confirmation(...)` flow through the same loop. Partial responses are fine: unanswered requests stay pending across `run_async` turns. +> **Replay-safety note:** HITL resume matches recorded human responses against +> generated interrupt/function-call ids, so those ids must regenerate +> identically on replay. The plugin installs ADK's platform time/uuid/random +> providers as process-wide defaults to guarantee this. On google-adk versions +> where `RequestInput` ids bypass the platform seam, pass an explicit +> `interrupt_id` to `RequestInput(...)` (as the examples here do). + ## Determinism Notes - The plugin patches ADK's `google.adk.platform` time, uuid, and (on ADK diff --git a/temporalio/contrib/google_adk_agents/_plugin.py b/temporalio/contrib/google_adk_agents/_plugin.py index 1a52cacb9..d470f807b 100644 --- a/temporalio/contrib/google_adk_agents/_plugin.py +++ b/temporalio/contrib/google_adk_agents/_plugin.py @@ -26,6 +26,23 @@ from temporalio.worker.workflow_sandbox import SandboxedWorkflowRunner +def _install_provider(module: Any, var_name: str, provider: Callable[[], Any]) -> None: + """Installs a provider as the process-wide default for an ADK platform seam. + + ADK's platform providers are held in ``ContextVar``s. Setting them with the + public ``set_*_provider`` helpers only affects the calling context, and + Temporal executes workflow code on executor threads whose contexts never + see that call — so the provider must be installed at the ContextVar + *default* level to be visible inside workflows. Rebinding the module's + ContextVar with a new default preserves the public setters' semantics + (a context-local ``set_*_provider`` still overrides the default). + """ + from contextvars import ContextVar + + context_var = getattr(module, var_name) + setattr(module, var_name, ContextVar(context_var.name, default=provider)) + + def setup_deterministic_runtime(): """Configures ADK runtime for Temporal determinism. @@ -33,8 +50,11 @@ def setup_deterministic_runtime(): This function is experimental and may change in future versions. Use with caution in production environments. - This should be called at the start of a Temporal Workflow before any ADK components - (like SessionService) are used, if they rely on runtime.get_time() or runtime.new_uuid(). + Installs Temporal-aware time, uuid, and (when the seam exists) random + providers as the process-wide defaults for ADK's ``google.adk.platform`` + seams. Inside a workflow they derive from ``workflow.now()`` / + ``workflow.uuid4()`` / ``workflow.random()`` so replays are + deterministic; outside a workflow they fall back to the real primitives. """ try: import google.adk.platform.time @@ -51,8 +71,16 @@ def _deterministic_id_provider() -> str: return str(workflow.uuid4()) return str(uuid.uuid4()) - google.adk.platform.time.set_time_provider(_deterministic_time_provider) - google.adk.platform.uuid.set_id_provider(_deterministic_id_provider) + _install_provider( + google.adk.platform.time, + "_time_provider_context_var", + _deterministic_time_provider, + ) + _install_provider( + google.adk.platform.uuid, + "_id_provider_context_var", + _deterministic_id_provider, + ) except ImportError: pass except Exception as e: @@ -61,7 +89,7 @@ def _deterministic_id_provider() -> str: try: # Available on ADK versions that route retry jitter through the # platform random seam; a no-op ImportError on older versions. - from google.adk.platform import set_random_provider + import google.adk.platform._random # type: ignore _local_random = random.Random() @@ -70,7 +98,11 @@ def _deterministic_random_provider() -> random.Random: return workflow.random() return _local_random - set_random_provider(_deterministic_random_provider) + _install_provider( + google.adk.platform._random, + "_random_provider_context_var", + _deterministic_random_provider, + ) except ImportError: pass except Exception as e: From 697e407f8c54a2f1fa647e9e70bae4e2da8ac5fa Mon Sep 17 00:00:00 2001 From: DABH Date: Fri, 24 Jul 2026 16:30:08 -0500 Subject: [PATCH 4/4] Add graph, dynamic workflow, and durable HITL integration tests - Graph: sequential/conditional-routing/parallel-join graphs with activity-backed nodes, LlmAgent node through the invoke_model activity, node timeout (durable timer), ADK RetryConfig retry, and a seam-gated jittered-retry replay test. All run with max_cached_workflows=0 so every workflow task fully replays. - Dynamic: ctx.run_node loops and asyncio.gather fan-out, Workflow-as-Tool, and HITL resume proving completed activity children are served from the session cache (exactly one real execution). - HITL: human-input node resume via query + update, activity_tool behind FunctionTool(require_confirmation=True) with the gated activity running exactly once on approval and never on rejection (replay-proven via forced eviction), multiple pending requests with partial responses, and a seam-gated default-interrupt-id replay test. - Replay: recorded graph_workflow.json and hitl_workflow.json histories added to the replay regression test. Tests marked seam-gated skip on google-adk releases that predate the upstream platform-seam routing (google/adk-python PR) and run against a build that includes it. --- .../histories/graph_workflow.json | 233 +++++++++ .../histories/hitl_workflow.json | 218 ++++++++ .../test_adk_dynamic_workflows.py | 394 +++++++++++++++ .../test_adk_graph_workflows.py | 447 ++++++++++++++++ .../google_adk_agents/test_adk_hitl.py | 477 ++++++++++++++++++ .../test_google_adk_agents_replay.py | 13 +- 6 files changed, 1781 insertions(+), 1 deletion(-) create mode 100644 tests/contrib/google_adk_agents/histories/graph_workflow.json create mode 100644 tests/contrib/google_adk_agents/histories/hitl_workflow.json create mode 100644 tests/contrib/google_adk_agents/test_adk_dynamic_workflows.py create mode 100644 tests/contrib/google_adk_agents/test_adk_graph_workflows.py create mode 100644 tests/contrib/google_adk_agents/test_adk_hitl.py diff --git a/tests/contrib/google_adk_agents/histories/graph_workflow.json b/tests/contrib/google_adk_agents/histories/graph_workflow.json new file mode 100644 index 000000000..a7f2d628d --- /dev/null +++ b/tests/contrib/google_adk_agents/histories/graph_workflow.json @@ -0,0 +1,233 @@ +{ + "events": [ + { + "eventId": "1", + "eventTime": "2026-07-24T21:26:01.955058Z", + "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_STARTED", + "taskId": "1048587", + "workflowExecutionStartedEventAttributes": { + "workflowType": { + "name": "SequentialGraphWorkflow" + }, + "taskQueue": { + "name": "record-q", + "kind": "TASK_QUEUE_KIND_NORMAL" + }, + "input": { + "payloads": [ + { + "metadata": { + "encoding": "anNvbi9wbGFpbg==" + }, + "data": "ImhlbGxvIg==" + } + ] + }, + "workflowExecutionTimeout": "60s", + "workflowRunTimeout": "60s", + "workflowTaskTimeout": "10s", + "originalExecutionRunId": "019f9605-5de3-70de-8269-7ae6eac6b0f1", + "identity": "91090@Davids-MacBook-Pro.local", + "firstExecutionRunId": "019f9605-5de3-70de-8269-7ae6eac6b0f1", + "attempt": 1, + "workflowExecutionExpirationTime": "2026-07-24T21:27:01.955Z", + "firstWorkflowTaskBackoff": "0s", + "workflowId": "record-graph", + "priority": {} + } + }, + { + "eventId": "2", + "eventTime": "2026-07-24T21:26:01.955114Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_SCHEDULED", + "taskId": "1048588", + "workflowTaskScheduledEventAttributes": { + "taskQueue": { + "name": "record-q", + "kind": "TASK_QUEUE_KIND_NORMAL" + }, + "startToCloseTimeout": "10s", + "attempt": 1 + } + }, + { + "eventId": "3", + "eventTime": "2026-07-24T21:26:01.957572Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_STARTED", + "taskId": "1048594", + "workflowTaskStartedEventAttributes": { + "scheduledEventId": "2", + "identity": "91090@Davids-MacBook-Pro.local", + "requestId": "1e5adcfe-3447-4757-8edf-6a601343f8d8", + "historySizeBytes": "312", + "workerVersion": { + "buildId": "2cfe1ab4865039ddf543acece575fd66" + } + } + }, + { + "eventId": "4", + "eventTime": "2026-07-24T21:26:02.011785Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_COMPLETED", + "taskId": "1048598", + "workflowTaskCompletedEventAttributes": { + "scheduledEventId": "2", + "startedEventId": "3", + "identity": "91090@Davids-MacBook-Pro.local", + "workerVersion": { + "buildId": "2cfe1ab4865039ddf543acece575fd66" + }, + "sdkMetadata": { + "coreUsedFlags": [ + 2, + 1, + 3 + ], + "sdkName": "temporal-python", + "sdkVersion": "1.30.0" + }, + "meteringMetadata": {} + } + }, + { + "eventId": "5", + "eventTime": "2026-07-24T21:26:02.011847Z", + "eventType": "EVENT_TYPE_ACTIVITY_TASK_SCHEDULED", + "taskId": "1048599", + "activityTaskScheduledEventAttributes": { + "activityId": "1", + "activityType": { + "name": "fetch_data" + }, + "taskQueue": { + "name": "record-q", + "kind": "TASK_QUEUE_KIND_NORMAL" + }, + "header": {}, + "input": { + "payloads": [ + { + "metadata": { + "encoding": "anNvbi9wbGFpbg==" + }, + "data": "ImhlbGxvIg==" + } + ] + }, + "scheduleToCloseTimeout": "60s", + "scheduleToStartTimeout": "60s", + "startToCloseTimeout": "30s", + "heartbeatTimeout": "0s", + "workflowTaskCompletedEventId": "4", + "retryPolicy": { + "initialInterval": "1s", + "backoffCoefficient": 2.0, + "maximumInterval": "100s" + }, + "useWorkflowBuildId": true, + "priority": {} + } + }, + { + "eventId": "6", + "eventTime": "2026-07-24T21:26:02.012954Z", + "eventType": "EVENT_TYPE_ACTIVITY_TASK_STARTED", + "taskId": "1048605", + "activityTaskStartedEventAttributes": { + "scheduledEventId": "5", + "identity": "91090@Davids-MacBook-Pro.local", + "requestId": "3557f515-0e9b-4f1b-9e47-1eac9d49978b", + "attempt": 1, + "workerVersion": { + "buildId": "2cfe1ab4865039ddf543acece575fd66" + } + } + }, + { + "eventId": "7", + "eventTime": "2026-07-24T21:26:02.014913Z", + "eventType": "EVENT_TYPE_ACTIVITY_TASK_COMPLETED", + "taskId": "1048606", + "activityTaskCompletedEventAttributes": { + "result": { + "payloads": [ + { + "metadata": { + "encoding": "anNvbi9wbGFpbg==" + }, + "data": "ImRhdGEtZm9yLWhlbGxvIg==" + } + ] + }, + "scheduledEventId": "5", + "startedEventId": "6", + "identity": "91090@Davids-MacBook-Pro.local" + } + }, + { + "eventId": "8", + "eventTime": "2026-07-24T21:26:02.014918Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_SCHEDULED", + "taskId": "1048607", + "workflowTaskScheduledEventAttributes": { + "taskQueue": { + "name": "91090@Davids-MacBook-Pro.local-574ad978ab364497ba3cb256de40fa91", + "kind": "TASK_QUEUE_KIND_STICKY", + "normalName": "record-q" + }, + "startToCloseTimeout": "10s", + "attempt": 1 + } + }, + { + "eventId": "9", + "eventTime": "2026-07-24T21:26:02.015483Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_STARTED", + "taskId": "1048611", + "workflowTaskStartedEventAttributes": { + "scheduledEventId": "8", + "identity": "91090@Davids-MacBook-Pro.local", + "requestId": "3b7f0d35-e6e0-4a2f-ad5b-26507d4ccaa2", + "historySizeBytes": "1072", + "workerVersion": { + "buildId": "2cfe1ab4865039ddf543acece575fd66" + } + } + }, + { + "eventId": "10", + "eventTime": "2026-07-24T21:26:02.018330Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_COMPLETED", + "taskId": "1048615", + "workflowTaskCompletedEventAttributes": { + "scheduledEventId": "8", + "startedEventId": "9", + "identity": "91090@Davids-MacBook-Pro.local", + "workerVersion": { + "buildId": "2cfe1ab4865039ddf543acece575fd66" + }, + "sdkMetadata": {}, + "meteringMetadata": {} + } + }, + { + "eventId": "11", + "eventTime": "2026-07-24T21:26:02.018361Z", + "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED", + "taskId": "1048616", + "workflowExecutionCompletedEventAttributes": { + "result": { + "payloads": [ + { + "metadata": { + "encoding": "anNvbi9wbGFpbg==" + }, + "data": "ImRhdGEtZm9yLWhlbGxvIHN1bW1hcml6ZWQi" + } + ] + }, + "workflowTaskCompletedEventId": "10" + } + } + ] +} \ No newline at end of file diff --git a/tests/contrib/google_adk_agents/histories/hitl_workflow.json b/tests/contrib/google_adk_agents/histories/hitl_workflow.json new file mode 100644 index 000000000..08ff34537 --- /dev/null +++ b/tests/contrib/google_adk_agents/histories/hitl_workflow.json @@ -0,0 +1,218 @@ +{ + "events": [ + { + "eventId": "1", + "eventTime": "2026-07-24T21:26:02.023110Z", + "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_STARTED", + "taskId": "1048621", + "workflowExecutionStartedEventAttributes": { + "workflowType": { + "name": "HumanInputGraphWorkflow" + }, + "taskQueue": { + "name": "record-q", + "kind": "TASK_QUEUE_KIND_NORMAL" + }, + "input": { + "payloads": [ + { + "metadata": { + "encoding": "anNvbi9wbGFpbg==" + }, + "data": "Im1ha2UgYSBwbGFuIg==" + } + ] + }, + "workflowExecutionTimeout": "60s", + "workflowRunTimeout": "60s", + "workflowTaskTimeout": "10s", + "originalExecutionRunId": "019f9605-5e27-71a9-a94c-bf0fad88f962", + "identity": "91090@Davids-MacBook-Pro.local", + "firstExecutionRunId": "019f9605-5e27-71a9-a94c-bf0fad88f962", + "attempt": 1, + "workflowExecutionExpirationTime": "2026-07-24T21:27:02.023Z", + "firstWorkflowTaskBackoff": "0s", + "workflowId": "record-hitl", + "priority": {} + } + }, + { + "eventId": "2", + "eventTime": "2026-07-24T21:26:02.023137Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_SCHEDULED", + "taskId": "1048622", + "workflowTaskScheduledEventAttributes": { + "taskQueue": { + "name": "record-q", + "kind": "TASK_QUEUE_KIND_NORMAL" + }, + "startToCloseTimeout": "10s", + "attempt": 1 + } + }, + { + "eventId": "3", + "eventTime": "2026-07-24T21:26:02.024065Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_STARTED", + "taskId": "1048628", + "workflowTaskStartedEventAttributes": { + "scheduledEventId": "2", + "identity": "91090@Davids-MacBook-Pro.local", + "requestId": "ca438925-9669-4d91-a35a-3ab2785e5a02", + "historySizeBytes": "314", + "workerVersion": { + "buildId": "2cfe1ab4865039ddf543acece575fd66" + } + } + }, + { + "eventId": "4", + "eventTime": "2026-07-24T21:26:02.064390Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_COMPLETED", + "taskId": "1048632", + "workflowTaskCompletedEventAttributes": { + "scheduledEventId": "2", + "startedEventId": "3", + "identity": "91090@Davids-MacBook-Pro.local", + "workerVersion": { + "buildId": "2cfe1ab4865039ddf543acece575fd66" + }, + "sdkMetadata": { + "coreUsedFlags": [ + 2, + 1, + 3 + ], + "sdkName": "temporal-python", + "sdkVersion": "1.30.0" + }, + "meteringMetadata": {} + } + }, + { + "eventId": "5", + "eventTime": "2026-07-24T21:26:02.068995Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_SCHEDULED", + "taskId": "1048639", + "workflowTaskScheduledEventAttributes": { + "taskQueue": { + "name": "91090@Davids-MacBook-Pro.local-574ad978ab364497ba3cb256de40fa91", + "kind": "TASK_QUEUE_KIND_STICKY", + "normalName": "record-q" + }, + "startToCloseTimeout": "10s", + "attempt": 1 + } + }, + { + "eventId": "6", + "eventTime": "2026-07-24T21:26:02.069226Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_STARTED", + "taskId": "1048640", + "workflowTaskStartedEventAttributes": { + "scheduledEventId": "5", + "identity": "91090@Davids-MacBook-Pro.local", + "requestId": "07b39d6d-c065-4f79-9d3d-4c1843a9182c", + "historySizeBytes": "583", + "workerVersion": { + "buildId": "2cfe1ab4865039ddf543acece575fd66" + } + } + }, + { + "eventId": "7", + "eventTime": "2026-07-24T21:26:02.072464Z", + "eventType": "EVENT_TYPE_WORKFLOW_TASK_COMPLETED", + "taskId": "1048641", + "workflowTaskCompletedEventAttributes": { + "scheduledEventId": "5", + "startedEventId": "6", + "identity": "91090@Davids-MacBook-Pro.local", + "workerVersion": { + "buildId": "2cfe1ab4865039ddf543acece575fd66" + }, + "sdkMetadata": {}, + "meteringMetadata": {} + } + }, + { + "eventId": "8", + "eventTime": "2026-07-24T21:26:02.072516Z", + "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_UPDATE_ACCEPTED", + "taskId": "1048642", + "workflowExecutionUpdateAcceptedEventAttributes": { + "protocolInstanceId": "cdbc4873-f79c-4bbd-9613-b3140b1b8ed2", + "acceptedRequestMessageId": "cdbc4873-f79c-4bbd-9613-b3140b1b8ed2/request", + "acceptedRequestSequencingEventId": "5", + "acceptedRequest": { + "meta": { + "updateId": "cdbc4873-f79c-4bbd-9613-b3140b1b8ed2", + "identity": "91090@Davids-MacBook-Pro.local" + }, + "input": { + "name": "respond", + "args": { + "payloads": [ + { + "metadata": { + "encoding": "anNvbi9wbGFpbg==" + }, + "data": "ImFwcHJvdmFsIg==" + }, + { + "metadata": { + "encoding": "anNvbi9wbGFpbg==" + }, + "data": "InNoaXAtaXQi" + } + ] + } + } + } + } + }, + { + "eventId": "9", + "eventTime": "2026-07-24T21:26:02.072567Z", + "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_UPDATE_COMPLETED", + "taskId": "1048643", + "workflowExecutionUpdateCompletedEventAttributes": { + "meta": { + "updateId": "cdbc4873-f79c-4bbd-9613-b3140b1b8ed2", + "identity": "91090@Davids-MacBook-Pro.local" + }, + "outcome": { + "success": { + "payloads": [ + { + "metadata": { + "encoding": "YmluYXJ5L251bGw=" + } + } + ] + } + }, + "acceptedEventId": "8" + } + }, + { + "eventId": "10", + "eventTime": "2026-07-24T21:26:02.072587Z", + "eventType": "EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED", + "taskId": "1048644", + "workflowExecutionCompletedEventAttributes": { + "result": { + "payloads": [ + { + "metadata": { + "encoding": "anNvbi9wbGFpbg==" + }, + "data": "ImFwcHJvdmVkOnNoaXAtaXQi" + } + ] + }, + "workflowTaskCompletedEventId": "7" + } + } + ] +} \ No newline at end of file diff --git a/tests/contrib/google_adk_agents/test_adk_dynamic_workflows.py b/tests/contrib/google_adk_agents/test_adk_dynamic_workflows.py new file mode 100644 index 000000000..624db5190 --- /dev/null +++ b/tests/contrib/google_adk_agents/test_adk_dynamic_workflows.py @@ -0,0 +1,394 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Integration tests for ADK v2 dynamic workflows running in Temporal workflows.""" + +import asyncio +import uuid +from collections.abc import AsyncGenerator +from datetime import timedelta +from typing import Any, Optional + +import pytest +from google.adk.agents import LlmAgent +from google.adk.agents.context import Context +from google.adk.events import RequestInput +from google.adk.models import BaseLlm, LLMRegistry +from google.adk.models.llm_request import LlmRequest +from google.adk.models.llm_response import LlmResponse +from google.adk.runners import Runner +from google.adk.sessions import InMemorySessionService +from google.adk.utils.context_utils import Aclosing +from google.adk.workflow import START, Workflow, node +from google.genai import types +from google.genai.types import Content, FunctionCall, Part +from pydantic import BaseModel + +from temporalio import activity, workflow +from temporalio.client import Client +from temporalio.contrib.google_adk_agents import ( + GoogleAdkPlugin, + HitlRequest, + TemporalModel, + hitl_input_response, + pending_hitl_requests, +) +from temporalio.contrib.google_adk_agents.workflow import activity_node +from temporalio.worker import Worker + +TASK_QUEUE = "adk-dynamic-task-queue" + +# Worker-side record of real activity executions, keyed by workflow id. +# Replayed workflow tasks do not re-execute activities, so this counts +# actual executions only. +_ACTIVITY_EXECUTIONS: dict[str, list[Any]] = {} + + +@activity.defn +async def enrich_number(n: int) -> str: + """Activity that enriches a number.""" + return f"enriched-{n}" + + +@activity.defn +async def counted_fetch(tag: str) -> str: + """Activity that records each real execution.""" + _ACTIVITY_EXECUTIONS.setdefault(str(activity.info().workflow_id), []).append(tag) + return f"fetched-{tag}" + + +async def drive_graph(graph: Workflow, prompt: str) -> Any: + """Runs an ADK graph to completion in-workflow, returning the last output.""" + runner = Runner( + app_name="test_app", node=graph, session_service=InMemorySessionService() + ) + session = await runner.session_service.create_session( + app_name="test_app", user_id="test" + ) + last_output: Any = None + async with Aclosing( + runner.run_async( + user_id="test", + session_id=session.id, + new_message=types.Content(role="user", parts=[types.Part(text=prompt)]), + ) + ) as agen: + async for event in agen: + if getattr(event, "output", None) is not None: + last_output = event.output + return last_output + + +@workflow.defn +class DynamicLoopWorkflow: + """A dynamic node drives activity-backed children in a plain Python loop.""" + + @workflow.run + async def run(self, count: int) -> list[str]: + child = activity_node( + enrich_number, start_to_close_timeout=timedelta(seconds=30) + ) + + @node(rerun_on_resume=True) + async def driver(ctx: Context) -> list[str]: + results = [] + for i in range(count): + results.append(await ctx.run_node(child, node_input=i)) + return results + + graph = Workflow(name="dynamic_loop", edges=[(START, driver)]) + return await drive_graph(graph, "go") + + +@workflow.defn +class DynamicGatherWorkflow: + """A dynamic node fans out children concurrently with asyncio.gather.""" + + @workflow.run + async def run(self, count: int) -> list[str]: + child = activity_node( + enrich_number, start_to_close_timeout=timedelta(seconds=30) + ) + + @node(rerun_on_resume=True) + async def driver(ctx: Context) -> list[str]: + return list( + await asyncio.gather( + *(ctx.run_node(child, node_input=i) for i in range(count)) + ) + ) + + graph = Workflow(name="dynamic_gather", edges=[(START, driver)]) + return await drive_graph(graph, "go") + + +class EnrichInput(BaseModel): + value: int + + +def _make_enrich_flow() -> Workflow: + """An inner graph workflow usable as an agent tool (Workflow-as-Tool).""" + + def pick(node_input: EnrichInput) -> int: + return node_input.value + + child = activity_node(enrich_number, start_to_close_timeout=timedelta(seconds=30)) + return Workflow( + name="enrich_flow", + description="Enriches a number and returns the enriched text.", + input_schema=EnrichInput, + edges=[(START, pick, child)], + ) + + +class WorkflowToolModel(BaseLlm): + """Scripted model: call the enrich_flow tool once, then answer with its result.""" + + async def generate_content_async( + self, llm_request: LlmRequest, stream: bool = False + ) -> AsyncGenerator[LlmResponse, None]: + tool_response: Optional[types.FunctionResponse] = None + for content in llm_request.contents: + for part in content.parts or []: + if part.function_response is not None: + tool_response = part.function_response + if tool_response is None: + yield LlmResponse( + content=Content( + role="model", + parts=[ + Part( + function_call=FunctionCall( + name="enrich_flow", args={"value": 7} + ) + ) + ], + ) + ) + else: + yield LlmResponse( + content=Content( + role="model", + parts=[Part(text=f"tool-said:{tool_response.response}")], + ) + ) + + @classmethod + def supported_models(cls) -> list[str]: + return ["workflow_tool_model"] + + +@workflow.defn +class WorkflowAsToolWorkflow: + """An agent invokes a whole graph workflow as a tool.""" + + @workflow.run + async def run(self, model_name: str) -> str: + agent = LlmAgent( + name="root", + model=TemporalModel(model_name), + instruction="Use the enrich_flow tool.", + tools=[_make_enrich_flow()], # type: ignore + ) + runner = Runner( + app_name="test_app", + agent=agent, + session_service=InMemorySessionService(), + ) + session = await runner.session_service.create_session( + app_name="test_app", user_id="test" + ) + final_text = "" + async with Aclosing( + runner.run_async( + user_id="test", + session_id=session.id, + new_message=types.Content( + role="user", parts=[types.Part(text="enrich 7")] + ), + ) + ) as agen: + async for event in agen: + if event.content and event.content.parts: + text = event.content.parts[0].text + if text: + final_text = text + return final_text + + +@workflow.defn +class DynamicResumeWorkflow: + """HITL resume re-runs the dynamic driver but skips completed children.""" + + def __init__(self) -> None: + self._pending: dict[str, HitlRequest] = {} + self._responses: dict[str, Any] = {} + + @workflow.query + def pending_requests(self) -> list[HitlRequest]: + return list(self._pending.values()) + + @workflow.update + async def respond(self, interrupt_id: str, response: Any) -> None: + self._responses[interrupt_id] = response + + @workflow.run + async def run(self, prompt: str) -> str: + child = activity_node( + counted_fetch, start_to_close_timeout=timedelta(seconds=30) + ) + + def approval_gate(): + yield RequestInput(interrupt_id="approval", message="Approve?") # type: ignore + + @node(rerun_on_resume=True) + async def driver(ctx: Context) -> str: + fetched = await ctx.run_node(child, node_input="step1") + approval = await ctx.run_node(approval_gate) + return f"{fetched}|{approval}" + + graph = Workflow(name="dynamic_resume", edges=[(START, driver)]) + runner = Runner( + app_name="test_app", node=graph, session_service=InMemorySessionService() + ) + session = await runner.session_service.create_session( + app_name="test_app", user_id="test" + ) + message = types.Content(role="user", parts=[types.Part(text=prompt)]) + result = "" + while True: + async with Aclosing( + runner.run_async( + user_id="test", session_id=session.id, new_message=message + ) + ) as agen: + async for event in agen: + for request in pending_hitl_requests(event): + self._pending[request.interrupt_id] = request + if getattr(event, "output", None) is not None: + result = str(event.output) + if not self._pending: + return result + await workflow.wait_condition( + lambda: any(i in self._responses for i in self._pending) + ) + parts = [ + hitl_input_response(i, self._responses.pop(i)) + for i in list(self._pending) + if i in self._responses + ] + for part in parts: + assert part.function_response and part.function_response.id + self._pending.pop(part.function_response.id) + message = types.Content(role="user", parts=parts) + + +def _adk_client(client: Client) -> Client: + new_config = client.config() + new_config["plugins"] = [GoogleAdkPlugin()] + return Client(**new_config) + + +def _worker(client: Client) -> Worker: + return Worker( + client, + task_queue=TASK_QUEUE, + activities=[enrich_number, counted_fetch], + workflows=[ + DynamicLoopWorkflow, + DynamicGatherWorkflow, + WorkflowAsToolWorkflow, + DynamicResumeWorkflow, + ], + max_cached_workflows=0, + ) + + +@pytest.mark.asyncio +async def test_dynamic_loop(client: Client): + client = _adk_client(client) + async with _worker(client): + result = await client.execute_workflow( + DynamicLoopWorkflow.run, + 3, + id=f"dynamic-loop-{uuid.uuid4()}", + task_queue=TASK_QUEUE, + execution_timeout=timedelta(seconds=30), + ) + assert result == ["enriched-0", "enriched-1", "enriched-2"] + + +@pytest.mark.asyncio +async def test_dynamic_gather(client: Client): + client = _adk_client(client) + async with _worker(client): + result = await client.execute_workflow( + DynamicGatherWorkflow.run, + 4, + id=f"dynamic-gather-{uuid.uuid4()}", + task_queue=TASK_QUEUE, + execution_timeout=timedelta(seconds=30), + ) + assert result == ["enriched-0", "enriched-1", "enriched-2", "enriched-3"] + + +@pytest.mark.asyncio +async def test_workflow_as_tool(client: Client): + client = _adk_client(client) + async with _worker(client): + LLMRegistry.register(WorkflowToolModel) + result = await client.execute_workflow( + WorkflowAsToolWorkflow.run, + "workflow_tool_model", + id=f"workflow-as-tool-{uuid.uuid4()}", + task_queue=TASK_QUEUE, + execution_timeout=timedelta(seconds=30), + ) + assert "enriched-7" in result + + +@pytest.mark.asyncio +async def test_dynamic_resume_skips_completed_children(client: Client): + client = _adk_client(client) + workflow_id = f"dynamic-resume-{uuid.uuid4()}" + async with _worker(client): + handle = await client.start_workflow( + DynamicResumeWorkflow.run, + "go", + id=workflow_id, + task_queue=TASK_QUEUE, + execution_timeout=timedelta(seconds=60), + ) + + async def _pending() -> list[HitlRequest]: + while True: + pending = await handle.query(DynamicResumeWorkflow.pending_requests) + if pending: + return pending + await asyncio.sleep(0.1) + + pending = await asyncio.wait_for(_pending(), timeout=20) + assert pending[0].kind == "input" + assert pending[0].interrupt_id == "approval" + assert pending[0].message == "Approve?" + + await handle.execute_update( + DynamicResumeWorkflow.respond, args=["approval", "yes"] + ) + result = await handle.result() + + assert result == "fetched-step1|yes" + # The dynamic driver body re-ran on resume, but the completed activity + # child was served from the session cache: exactly one real execution. + assert _ACTIVITY_EXECUTIONS.get(workflow_id) == ["step1"] diff --git a/tests/contrib/google_adk_agents/test_adk_graph_workflows.py b/tests/contrib/google_adk_agents/test_adk_graph_workflows.py new file mode 100644 index 000000000..f7b8617ff --- /dev/null +++ b/tests/contrib/google_adk_agents/test_adk_graph_workflows.py @@ -0,0 +1,447 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Integration tests for ADK v2 graph workflows running in Temporal workflows.""" + +import asyncio +import uuid +from collections.abc import AsyncGenerator +from datetime import timedelta +from typing import Any + +import pytest +from google.adk.agents import LlmAgent +from google.adk.models import BaseLlm, LLMRegistry +from google.adk.models.llm_request import LlmRequest +from google.adk.models.llm_response import LlmResponse +from google.adk.runners import Runner +from google.adk.sessions import InMemorySessionService +from google.adk.utils.context_utils import Aclosing +from google.adk.workflow import ( + DEFAULT_ROUTE, + START, + FunctionNode, + JoinNode, + NodeTimeoutError, + RetryConfig, + Workflow, +) +from google.genai import types +from google.genai.types import Content, Part + +from temporalio import activity, workflow +from temporalio.client import Client +from temporalio.contrib.google_adk_agents import GoogleAdkPlugin, TemporalModel +from temporalio.contrib.google_adk_agents.workflow import activity_node +from temporalio.worker import Replayer, Worker + +TASK_QUEUE = "adk-graph-task-queue" + + +def _adk_has_random_seam() -> bool: + """Whether ADK exposes the platform random provider seam.""" + import importlib.util + + return importlib.util.find_spec("google.adk.platform._random") is not None + + +@activity.defn +async def fetch_data(query: str) -> str: + """Activity that fetches data for a query.""" + return f"data-for-{query}" + + +@activity.defn +async def enrich_item(item: str) -> str: + """Activity that enriches a single item.""" + return f"enriched-{item}" + + +@activity.defn +async def combine_parts(left: str, right: str) -> str: + """Activity that combines two named parts.""" + return f"{left}+{right}" + + +async def drive_graph(graph: Workflow, prompt: str) -> Any: + """Runs an ADK graph to completion in-workflow, returning the last output.""" + runner = Runner( + app_name="test_app", node=graph, session_service=InMemorySessionService() + ) + session = await runner.session_service.create_session( + app_name="test_app", user_id="test" + ) + last_output: Any = None + async with Aclosing( + runner.run_async( + user_id="test", + session_id=session.id, + new_message=types.Content(role="user", parts=[types.Part(text=prompt)]), + ) + ) as agen: + async for event in agen: + if getattr(event, "output", None) is not None: + last_output = event.output + return last_output + + +@workflow.defn +class SequentialGraphWorkflow: + """START -> activity node -> plain in-workflow node.""" + + @workflow.run + async def run(self, query: str) -> str: + fetch = activity_node(fetch_data, start_to_close_timeout=timedelta(seconds=30)) + + def summarize(node_input: str) -> str: + return f"{node_input} summarized" + + graph = Workflow(name="pipeline", edges=[(START, fetch, summarize)]) + return await drive_graph(graph, query) + + +@workflow.defn +class RoutingGraphWorkflow: + """Conditional routing through a dict edge with a DEFAULT_ROUTE fallback.""" + + @workflow.run + async def run(self, ticket: str) -> str: + from google.adk.events import Event + + def route_ticket(node_input: Any) -> Event: + text = str(node_input) + return Event(route="bug" if "bug" in text else "other", output=text) # type: ignore + + def handle_bug(node_input: str) -> str: + return "routed-to-bug" + + def handle_other(node_input: str) -> str: + return "routed-to-other" + + graph = Workflow( + name="router", + edges=[ + (START, route_ticket), + (route_ticket, {"bug": handle_bug, DEFAULT_ROUTE: handle_other}), + ], + ) + return await drive_graph(graph, ticket) + + +@workflow.defn +class ParallelJoinGraphWorkflow: + """Parallel fan-out of two activity-backed branches joined by a JoinNode.""" + + @workflow.run + async def run(self, prompt: str) -> dict[str, Any]: + def make_a(node_input: Any) -> str: + return "alpha" + + def make_b(node_input: Any) -> str: + return "beta" + + enrich_a = activity_node( + enrich_item, name="enrich_a", start_to_close_timeout=timedelta(seconds=30) + ) + enrich_b = activity_node( + enrich_item, name="enrich_b", start_to_close_timeout=timedelta(seconds=30) + ) + + join = JoinNode(name="join") + graph = Workflow( + name="fanout", + edges=[ + (START, make_a, enrich_a, join), + (START, make_b, enrich_b, join), + ], + ) + result = await drive_graph(graph, prompt) + assert isinstance(result, dict) + return result + + +@workflow.defn +class MultiParamActivityNodeWorkflow: + """A multi-parameter activity node bound from a dict node_input.""" + + @workflow.run + async def run(self, prompt: str) -> str: + def prepare(node_input: Any) -> dict[str, str]: + return {"left": "L", "right": "R"} + + combine = activity_node( + combine_parts, start_to_close_timeout=timedelta(seconds=30) + ) + graph = Workflow(name="multi", edges=[(START, prepare, combine)]) + return await drive_graph(graph, prompt) + + +@workflow.defn +class AgentNodeGraphWorkflow: + """An LlmAgent node inside a graph, calling the model via an activity.""" + + @workflow.run + async def run(self, model_name: str) -> str: + greeter = LlmAgent( + name="greeter", + model=TemporalModel(model_name), + instruction="You are a greeter", + mode="single_turn", + ) + + def finalize(node_input: Any) -> str: + text = node_input + if isinstance(node_input, types.Content) and node_input.parts: + text = node_input.parts[0].text + return f"final:{text}" + + graph = Workflow(name="agent_graph", edges=[(START, greeter, finalize)]) + return await drive_graph(graph, "greet the user") + + +@workflow.defn +class TimeoutGraphWorkflow: + """A node timeout surfaces as NodeTimeoutError via a durable timer.""" + + @workflow.run + async def run(self, prompt: str) -> str: + async def slow(node_input: Any) -> str: + await asyncio.sleep(5) + return "never" + + slow_node = FunctionNode(func=slow, timeout=0.2) + graph = Workflow(name="slowpoke", edges=[(START, slow_node)]) + try: + await drive_graph(graph, prompt) + return "no-timeout" + except NodeTimeoutError: + return "timed-out" + + +@workflow.defn +class RetryGraphWorkflow: + """An ADK RetryConfig retries a failing in-workflow node deterministically.""" + + @workflow.run + async def run(self, prompt: str) -> str: + attempts: list[int] = [] + + def flaky(node_input: Any) -> str: + attempts.append(1) + if len(attempts) < 2: + raise RuntimeError("transient failure") + return f"ok-after-{len(attempts)}" + + flaky_node = FunctionNode( + func=flaky, + retry_config=RetryConfig(max_attempts=3, initial_delay=0.01, jitter=0.0), + ) + graph = Workflow(name="retrier", edges=[(START, flaky_node)]) + return await drive_graph(graph, prompt) + + +@workflow.defn +class JitteredRetryGraphWorkflow: + """A retried node with default-style jitter must replay deterministically. + + Retry jitter feeds asyncio.sleep, i.e. a durable timer; unless the delay is + drawn from workflow.random() (via ADK's platform random seam), replays + compute a different timer duration and diverge. + """ + + @workflow.run + async def run(self, prompt: str) -> str: + attempts: list[int] = [] + + def flaky(node_input: Any) -> str: + attempts.append(1) + if len(attempts) < 2: + raise RuntimeError("transient failure") + return f"ok-after-{len(attempts)}" + + flaky_node = FunctionNode( + func=flaky, + retry_config=RetryConfig(max_attempts=3, initial_delay=0.05, jitter=0.5), + ) + graph = Workflow(name="jittery", edges=[(START, flaky_node)]) + return await drive_graph(graph, prompt) + + +class GraphAgentModel(BaseLlm): + """Scripted model for the agent-node graph test.""" + + async def generate_content_async( + self, llm_request: LlmRequest, stream: bool = False + ) -> AsyncGenerator[LlmResponse, None]: + yield LlmResponse( + content=Content(role="model", parts=[Part(text="agent-says-hi")]) + ) + + @classmethod + def supported_models(cls) -> list[str]: + return ["graph_model"] + + +def _adk_client(client: Client) -> Client: + new_config = client.config() + new_config["plugins"] = [GoogleAdkPlugin()] + return Client(**new_config) + + +def _worker(client: Client) -> Worker: + return Worker( + client, + task_queue=TASK_QUEUE, + activities=[fetch_data, enrich_item, combine_parts], + workflows=[ + SequentialGraphWorkflow, + RoutingGraphWorkflow, + ParallelJoinGraphWorkflow, + MultiParamActivityNodeWorkflow, + AgentNodeGraphWorkflow, + TimeoutGraphWorkflow, + RetryGraphWorkflow, + JitteredRetryGraphWorkflow, + ], + max_cached_workflows=0, + ) + + +@pytest.mark.asyncio +async def test_graph_sequential_with_activity_node(client: Client): + client = _adk_client(client) + async with _worker(client): + result = await client.execute_workflow( + SequentialGraphWorkflow.run, + "hello", + id=f"graph-sequential-{uuid.uuid4()}", + task_queue=TASK_QUEUE, + execution_timeout=timedelta(seconds=30), + ) + assert result == "data-for-hello summarized" + + +@pytest.mark.parametrize( + "ticket,expected", + [("bug: crash", "routed-to-bug"), ("question", "routed-to-other")], +) +@pytest.mark.asyncio +async def test_graph_conditional_routing(client: Client, ticket: str, expected: str): + client = _adk_client(client) + async with _worker(client): + result = await client.execute_workflow( + RoutingGraphWorkflow.run, + ticket, + id=f"graph-routing-{uuid.uuid4()}", + task_queue=TASK_QUEUE, + execution_timeout=timedelta(seconds=30), + ) + assert result == expected + + +@pytest.mark.asyncio +async def test_graph_parallel_fanout_join(client: Client): + client = _adk_client(client) + async with _worker(client): + result = await client.execute_workflow( + ParallelJoinGraphWorkflow.run, + "go", + id=f"graph-join-{uuid.uuid4()}", + task_queue=TASK_QUEUE, + execution_timeout=timedelta(seconds=30), + ) + # JoinNode aggregates branch outputs keyed by predecessor node name. + assert result == {"enrich_a": "enriched-alpha", "enrich_b": "enriched-beta"} + + +@pytest.mark.asyncio +async def test_graph_multi_param_activity_node(client: Client): + client = _adk_client(client) + async with _worker(client): + result = await client.execute_workflow( + MultiParamActivityNodeWorkflow.run, + "go", + id=f"graph-multiparam-{uuid.uuid4()}", + task_queue=TASK_QUEUE, + execution_timeout=timedelta(seconds=30), + ) + assert result == "L+R" + + +@pytest.mark.asyncio +async def test_graph_llm_agent_node(client: Client): + client = _adk_client(client) + async with _worker(client): + LLMRegistry.register(GraphAgentModel) + result = await client.execute_workflow( + AgentNodeGraphWorkflow.run, + "graph_model", + id=f"graph-agent-{uuid.uuid4()}", + task_queue=TASK_QUEUE, + execution_timeout=timedelta(seconds=30), + ) + assert result == "final:agent-says-hi" + + +@pytest.mark.asyncio +async def test_graph_node_timeout(client: Client): + client = _adk_client(client) + async with _worker(client): + result = await client.execute_workflow( + TimeoutGraphWorkflow.run, + "go", + id=f"graph-timeout-{uuid.uuid4()}", + task_queue=TASK_QUEUE, + execution_timeout=timedelta(seconds=30), + ) + assert result == "timed-out" + + +@pytest.mark.asyncio +async def test_graph_node_retry(client: Client): + client = _adk_client(client) + async with _worker(client): + result = await client.execute_workflow( + RetryGraphWorkflow.run, + "go", + id=f"graph-retry-{uuid.uuid4()}", + task_queue=TASK_QUEUE, + execution_timeout=timedelta(seconds=30), + ) + assert result == "ok-after-2" + + +@pytest.mark.asyncio +async def test_graph_node_retry_jitter_replay_safe(client: Client): + if not _adk_has_random_seam(): + pytest.skip( + "requires google-adk with the platform random seam (upstream PR pending)" + ) + client = _adk_client(client) + async with _worker(client): + handle = await client.start_workflow( + JitteredRetryGraphWorkflow.run, + "go", + id=f"graph-retry-jitter-{uuid.uuid4()}", + task_queue=TASK_QUEUE, + execution_timeout=timedelta(seconds=30), + ) + result = await handle.result() + assert result == "ok-after-2" + history = await handle.fetch_history() + # The jittered retry delay is a durable timer; replay must recompute the + # exact same duration from workflow.random(). + await Replayer( + workflows=[JitteredRetryGraphWorkflow], plugins=[GoogleAdkPlugin()] + ).replay_workflow(history) diff --git a/tests/contrib/google_adk_agents/test_adk_hitl.py b/tests/contrib/google_adk_agents/test_adk_hitl.py new file mode 100644 index 000000000..a0894bf2f --- /dev/null +++ b/tests/contrib/google_adk_agents/test_adk_hitl.py @@ -0,0 +1,477 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Integration tests for durable human-in-the-loop with ADK in Temporal workflows.""" + +import asyncio +import uuid +from collections.abc import AsyncGenerator +from datetime import timedelta +from typing import Any, Optional + +import pytest +from google.adk.agents import LlmAgent +from google.adk.events import RequestInput +from google.adk.models import BaseLlm, LLMRegistry +from google.adk.models.llm_request import LlmRequest +from google.adk.models.llm_response import LlmResponse +from google.adk.runners import Runner +from google.adk.sessions import InMemorySessionService +from google.adk.tools.function_tool import FunctionTool +from google.adk.utils.context_utils import Aclosing +from google.adk.workflow import START, JoinNode, Workflow +from google.genai import types +from google.genai.types import Content, FunctionCall, Part + +import temporalio.contrib.google_adk_agents.workflow +from temporalio import activity, workflow +from temporalio.client import Client, WorkflowHandle +from temporalio.contrib.google_adk_agents import ( + GoogleAdkPlugin, + HitlRequest, + TemporalModel, + hitl_confirmation_response, + hitl_input_response, + pending_hitl_requests, +) +from temporalio.worker import Replayer, Worker + +TASK_QUEUE = "adk-hitl-task-queue" + +# Worker-side record of real activity executions, keyed by workflow id. +_ACTIVITY_EXECUTIONS: dict[str, list[Any]] = {} + + +def _adk_routes_interrupt_ids_through_platform() -> bool: + """Whether ADK mints default RequestInput ids via the platform uuid seam.""" + import google.adk.platform.uuid as platform_uuid + + platform_uuid.set_id_provider(lambda: "probe-id") + try: + return RequestInput().interrupt_id == "probe-id" # type: ignore + finally: + platform_uuid.reset_id_provider() + + +@activity.defn +async def danger_activity(target: str) -> str: + """Activity gated behind human confirmation.""" + _ACTIVITY_EXECUTIONS.setdefault(str(activity.info().workflow_id), []).append(target) + return f"deleted-{target}" + + +class _HitlLoopMixin: + """Shared pending/response bookkeeping for HITL workflows.""" + + def __init__(self) -> None: + self._pending: dict[str, HitlRequest] = {} + self._responses: dict[str, Any] = {} + + async def _drive(self, runner: Runner, first_message: types.Content) -> str: + session = await runner.session_service.create_session( + app_name="test_app", user_id="test" + ) + message = first_message + result = "" + while True: + async with Aclosing( + runner.run_async( + user_id="test", session_id=session.id, new_message=message + ) + ) as agen: + async for event in agen: + for request in pending_hitl_requests(event): + self._pending[request.interrupt_id] = request + output = getattr(event, "output", None) + if output is not None: + result = str(output) + elif ( + event.content + and event.content.parts + and event.content.parts[0].text + ): + result = event.content.parts[0].text + if not self._pending: + return result + await workflow.wait_condition( + lambda: any(i in self._responses for i in self._pending) + ) + parts = [] + for interrupt_id in list(self._pending): + if interrupt_id not in self._responses: + continue + request = self._pending.pop(interrupt_id) + response = self._responses.pop(interrupt_id) + if request.kind == "tool_confirmation": + parts.append( + hitl_confirmation_response( + interrupt_id, confirmed=bool(response) + ) + ) + else: + parts.append(hitl_input_response(interrupt_id, response)) + message = types.Content(role="user", parts=parts) + + +@workflow.defn +class HumanInputGraphWorkflow(_HitlLoopMixin): + """A graph human-input node maps onto a durable Temporal wait.""" + + @workflow.query + def pending_requests(self) -> list[HitlRequest]: + return list(self._pending.values()) + + @workflow.update + async def respond(self, interrupt_id: str, response: Any) -> None: + self._responses[interrupt_id] = response + + @workflow.run + async def run(self, prompt: str) -> str: + def approval_gate(): + yield RequestInput( # type: ignore + interrupt_id="approval", message="Approve the plan?" + ) + + def formatter(node_input: Any) -> str: + return f"approved:{node_input}" + + graph = Workflow(name="hitl_graph", edges=[(START, approval_gate, formatter)]) + runner = Runner( + app_name="test_app", node=graph, session_service=InMemorySessionService() + ) + return await self._drive( + runner, types.Content(role="user", parts=[types.Part(text=prompt)]) + ) + + +class ConfirmationModel(BaseLlm): + """Scripted model: call danger_activity once, then acknowledge its response.""" + + async def generate_content_async( + self, llm_request: LlmRequest, stream: bool = False + ) -> AsyncGenerator[LlmResponse, None]: + saw_tool_response = any( + part.function_response is not None + and part.function_response.name == "danger_activity" + for content in llm_request.contents + for part in content.parts or [] + ) + saw_confirmation_request = any( + part.function_call is not None + and part.function_call.name == "adk_request_confirmation" + for content in llm_request.contents + for part in content.parts or [] + ) + if saw_tool_response: + yield LlmResponse( + content=Content(role="model", parts=[Part(text="all-done")]) + ) + elif saw_confirmation_request: + # The confirmation is pending; end the turn while waiting. + yield LlmResponse( + content=Content(role="model", parts=[Part(text="waiting-for-approval")]) + ) + else: + yield LlmResponse( + content=Content( + role="model", + parts=[ + Part( + function_call=FunctionCall( + name="danger_activity", args={"target": "prod"} + ) + ) + ], + ) + ) + + @classmethod + def supported_models(cls) -> list[str]: + return ["confirmation_model"] + + +@workflow.defn +class ConfirmationAgentWorkflow(_HitlLoopMixin): + """Tool confirmation gates an activity_tool: the activity only runs on approval.""" + + @workflow.query + def pending_requests(self) -> list[HitlRequest]: + return list(self._pending.values()) + + @workflow.update + async def respond(self, interrupt_id: str, response: Any) -> None: + self._responses[interrupt_id] = response + + @workflow.run + async def run(self, model_name: str) -> str: + danger_tool = temporalio.contrib.google_adk_agents.workflow.activity_tool( + danger_activity, start_to_close_timeout=timedelta(seconds=30) + ) + agent = LlmAgent( + name="ops_agent", + model=TemporalModel(model_name), + instruction="You are an ops agent", + tools=[FunctionTool(func=danger_tool, require_confirmation=True)], + ) + runner = Runner( + app_name="test_app", agent=agent, session_service=InMemorySessionService() + ) + return await self._drive( + runner, + types.Content(role="user", parts=[types.Part(text="delete prod")]), + ) + + +@workflow.defn +class MultiPendingGraphWorkflow(_HitlLoopMixin): + """Two parallel human-input branches; partial responses keep the rest pending.""" + + @workflow.query + def pending_requests(self) -> list[HitlRequest]: + return list(self._pending.values()) + + @workflow.update + async def respond(self, interrupt_id: str, response: Any) -> None: + self._responses[interrupt_id] = response + + @workflow.run + async def run(self, prompt: str) -> str: + def gate_a(): + yield RequestInput(interrupt_id="a", message="Approve A?") # type: ignore + + def gate_b(): + yield RequestInput(interrupt_id="b", message="Approve B?") # type: ignore + + def combine(node_input: dict[str, Any]) -> str: + return f"{node_input['gate_a']}&{node_input['gate_b']}" + + join = JoinNode(name="join") + graph = Workflow( + name="multi_hitl", + edges=[ + (START, gate_a, join), + (START, gate_b, join), + (join, combine), + ], + ) + runner = Runner( + app_name="test_app", node=graph, session_service=InMemorySessionService() + ) + return await self._drive( + runner, types.Content(role="user", parts=[types.Part(text=prompt)]) + ) + + +@workflow.defn +class DefaultInterruptIdWorkflow(_HitlLoopMixin): + """A RequestInput with no explicit id relies on the platform uuid seam.""" + + @workflow.query + def pending_requests(self) -> list[HitlRequest]: + return list(self._pending.values()) + + @workflow.update + async def respond(self, interrupt_id: str, response: Any) -> None: + self._responses[interrupt_id] = response + + @workflow.run + async def run(self, prompt: str) -> str: + def unnamed_gate(): + yield RequestInput(message="Approve?") # type: ignore + + def formatter(node_input: Any) -> str: + return f"got:{node_input}" + + graph = Workflow(name="default_id", edges=[(START, unnamed_gate, formatter)]) + runner = Runner( + app_name="test_app", node=graph, session_service=InMemorySessionService() + ) + return await self._drive( + runner, types.Content(role="user", parts=[types.Part(text=prompt)]) + ) + + +def _adk_client(client: Client) -> Client: + new_config = client.config() + new_config["plugins"] = [GoogleAdkPlugin()] + return Client(**new_config) + + +def _worker(client: Client) -> Worker: + return Worker( + client, + task_queue=TASK_QUEUE, + activities=[danger_activity], + workflows=[ + HumanInputGraphWorkflow, + ConfirmationAgentWorkflow, + MultiPendingGraphWorkflow, + DefaultInterruptIdWorkflow, + ], + max_cached_workflows=0, + ) + + +async def _wait_for_pending( + handle: WorkflowHandle, + query: Any, + count: int = 1, + expected_ids: Optional[set[str]] = None, +) -> list[HitlRequest]: + async def _poll() -> list[HitlRequest]: + while True: + pending = await handle.query(query) + if expected_ids is not None: + if {p.interrupt_id for p in pending} == expected_ids: + return pending + elif len(pending) >= count: + return pending + await asyncio.sleep(0.1) + + return await asyncio.wait_for(_poll(), timeout=20) + + +@pytest.mark.asyncio +async def test_human_input_node_update_resume(client: Client): + client = _adk_client(client) + async with _worker(client): + handle = await client.start_workflow( + HumanInputGraphWorkflow.run, + "make a plan", + id=f"hitl-input-{uuid.uuid4()}", + task_queue=TASK_QUEUE, + execution_timeout=timedelta(seconds=60), + ) + pending = await _wait_for_pending( + handle, HumanInputGraphWorkflow.pending_requests + ) + assert pending[0].kind == "input" + assert pending[0].interrupt_id == "approval" + assert pending[0].message == "Approve the plan?" + assert pending[0].invocation_id + + await handle.execute_update( + HumanInputGraphWorkflow.respond, args=["approval", "ship-it"] + ) + result = await handle.result() + assert result == "approved:ship-it" + + +@pytest.mark.parametrize("confirmed", [True, False]) +@pytest.mark.asyncio +async def test_tool_confirmation_activity_tool(client: Client, confirmed: bool): + client = _adk_client(client) + workflow_id = f"hitl-confirm-{confirmed}-{uuid.uuid4()}" + # max_cached_workflows=0 forces a full history replay on every workflow + # task, proving the confirmation resume is replay-safe: the recorded human + # response references the confirmation function-call id, which must + # regenerate identically on replay (it derives from workflow.uuid4() via + # the platform uuid seam the plugin installs). + async with _worker(client): + LLMRegistry.register(ConfirmationModel) + handle = await client.start_workflow( + ConfirmationAgentWorkflow.run, + "confirmation_model", + id=workflow_id, + task_queue=TASK_QUEUE, + execution_timeout=timedelta(seconds=60), + ) + pending = await _wait_for_pending( + handle, ConfirmationAgentWorkflow.pending_requests + ) + assert pending[0].kind == "tool_confirmation" + assert pending[0].original_function_call is not None + assert pending[0].original_function_call["name"] == "danger_activity" + assert pending[0].original_function_call["args"] == {"target": "prod"} + + await handle.execute_update( + ConfirmationAgentWorkflow.respond, + args=[pending[0].interrupt_id, confirmed], + ) + result = await handle.result() + + assert result == "all-done" + executions = _ACTIVITY_EXECUTIONS.get(workflow_id, []) + if confirmed: + # The gated activity ran exactly once, only after approval. + assert executions == ["prod"] + else: + # Rejected: the activity was never scheduled. + assert executions == [] + + +@pytest.mark.asyncio +async def test_hitl_multiple_pending_partial_response(client: Client): + client = _adk_client(client) + async with _worker(client): + handle = await client.start_workflow( + MultiPendingGraphWorkflow.run, + "go", + id=f"hitl-multi-{uuid.uuid4()}", + task_queue=TASK_QUEUE, + execution_timeout=timedelta(seconds=60), + ) + pending = await _wait_for_pending( + handle, MultiPendingGraphWorkflow.pending_requests, expected_ids={"a", "b"} + ) + + # Answer only one; the other must stay pending. + await handle.execute_update( + MultiPendingGraphWorkflow.respond, args=["a", "yes-a"] + ) + pending = await _wait_for_pending( + handle, MultiPendingGraphWorkflow.pending_requests, expected_ids={"b"} + ) + + await handle.execute_update( + MultiPendingGraphWorkflow.respond, args=["b", "yes-b"] + ) + result = await handle.result() + assert result == "yes-a&yes-b" + + +@pytest.mark.asyncio +async def test_default_interrupt_id_replay_safe(client: Client): + if not _adk_routes_interrupt_ids_through_platform(): + pytest.skip( + "requires google-adk with RequestInput ids routed through the" + " platform uuid seam (upstream PR pending)" + ) + client = _adk_client(client) + async with _worker(client): + handle = await client.start_workflow( + DefaultInterruptIdWorkflow.run, + "go", + id=f"hitl-default-id-{uuid.uuid4()}", + task_queue=TASK_QUEUE, + execution_timeout=timedelta(seconds=60), + ) + pending = await _wait_for_pending( + handle, DefaultInterruptIdWorkflow.pending_requests + ) + # The generated id must be a workflow-deterministic uuid. + interrupt_id = pending[0].interrupt_id + uuid.UUID(interrupt_id) + + await handle.execute_update( + DefaultInterruptIdWorkflow.respond, args=[interrupt_id, "fine"] + ) + result = await handle.result() + assert result == "got:fine" + + # Replaying the full history must regenerate the same interrupt id. + history = await handle.fetch_history() + await Replayer( + workflows=[DefaultInterruptIdWorkflow], plugins=[GoogleAdkPlugin()] + ).replay_workflow(history) diff --git a/tests/contrib/google_adk_agents/test_google_adk_agents_replay.py b/tests/contrib/google_adk_agents/test_google_adk_agents_replay.py index 810ca95f0..fff10d384 100644 --- a/tests/contrib/google_adk_agents/test_google_adk_agents_replay.py +++ b/tests/contrib/google_adk_agents/test_google_adk_agents_replay.py @@ -6,6 +6,10 @@ from temporalio.client import WorkflowHistory from temporalio.contrib.google_adk_agents import GoogleAdkPlugin from temporalio.worker import Replayer +from tests.contrib.google_adk_agents.test_adk_graph_workflows import ( + SequentialGraphWorkflow, +) +from tests.contrib.google_adk_agents.test_adk_hitl import HumanInputGraphWorkflow from tests.contrib.google_adk_agents.test_google_adk_agents import ( MultiAgentWorkflow, ResearchModel, @@ -19,6 +23,8 @@ [ "multi_agent.json", "single_agent.json", + "graph_workflow.json", + "hitl_workflow.json", ], ) async def test_replay(file_name: str) -> None: @@ -28,6 +34,11 @@ async def test_replay(file_name: str) -> None: LLMRegistry.register(ResearchModel) LLMRegistry.register(WeatherModel) await Replayer( - workflows=[MultiAgentWorkflow, WeatherAgent], + workflows=[ + MultiAgentWorkflow, + WeatherAgent, + SequentialGraphWorkflow, + HumanInputGraphWorkflow, + ], plugins=[GoogleAdkPlugin()], ).replay_workflow(WorkflowHistory.from_json("fake", history_json))