From 6d336394ebcdd15212b08c0c9786096c19719150 Mon Sep 17 00:00:00 2001 From: DABH Date: Fri, 24 Jul 2026 15:35:19 -0500 Subject: [PATCH 01/14] 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 9ebda28ea715104a1d8feb080dbe5aabe0fdb705 Mon Sep 17 00:00:00 2001 From: DABH Date: Fri, 24 Jul 2026 15:40:52 -0500 Subject: [PATCH 02/14] 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 5a52006f918e94832c8a5ca8ba74555cebaac826 Mon Sep 17 00:00:00 2001 From: DABH Date: Fri, 24 Jul 2026 16:29:54 -0500 Subject: [PATCH 03/14] 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 82dad30628ee2dfbb7d5edb07f8588d03b6effb8 Mon Sep 17 00:00:00 2001 From: DABH Date: Fri, 24 Jul 2026 16:30:08 -0500 Subject: [PATCH 04/14] 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)) From 51dc7a5713b8d7f8fa165ac16f8539dd180d34f1 Mon Sep 17 00:00:00 2001 From: DABH Date: Mon, 27 Jul 2026 15:42:22 -0500 Subject: [PATCH 05/14] Fix basedpyright warnings (deprecated Optional/Mapping aliases, unused bindings) basedpyright fails CI on warnings: replace typing.Optional/typing.Mapping with PEP 604/collections.abc forms, drop two unused query-result bindings, and mark the intentionally-unused node_input parameters (the name is load-bearing for ADK FunctionNode binding) with targeted ignores. --- temporalio/contrib/google_adk_agents/_hitl.py | 25 ++++++++++--------- .../contrib/google_adk_agents/workflow.py | 4 +-- .../test_adk_dynamic_workflows.py | 4 +-- .../test_adk_graph_workflows.py | 16 ++++++------ .../google_adk_agents/test_adk_hitl.py | 8 +++--- 5 files changed, 29 insertions(+), 28 deletions(-) diff --git a/temporalio/contrib/google_adk_agents/_hitl.py b/temporalio/contrib/google_adk_agents/_hitl.py index 0fb60051e..83baf2054 100644 --- a/temporalio/contrib/google_adk_agents/_hitl.py +++ b/temporalio/contrib/google_adk_agents/_hitl.py @@ -15,8 +15,9 @@ from __future__ import annotations +from collections.abc import Mapping from dataclasses import dataclass -from typing import Any, Literal, Mapping, Optional +from typing import Any, Literal from google.adk.events import Event from google.adk.tools.tool_confirmation import ToolConfirmation @@ -64,12 +65,12 @@ class HitlRequest: 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 + invocation_id: str | None = None + author: str | None = None + message: str | None = None + payload: Any | None = None + response_schema: dict[str, Any] | None = None + original_function_call: dict[str, Any] | None = None def pending_hitl_requests(event: Event) -> list[HitlRequest]: @@ -104,10 +105,10 @@ def pending_hitl_requests(event: Event) -> list[HitlRequest]: 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 + message: str | None = None + payload: Any | None = None + response_schema: dict[str, Any] | None = None + original_function_call: dict[str, Any] | None = None if kind == "input": message = args.get("message") payload = args.get("payload") @@ -164,7 +165,7 @@ def hitl_input_response(interrupt_id: str, response: Any) -> types.Part: def hitl_confirmation_response( - interrupt_id: str, *, confirmed: bool, payload: Optional[Any] = None + interrupt_id: str, *, confirmed: bool, payload: Any | None = None ) -> types.Part: """Builds the message part answering a tool-confirmation request. diff --git a/temporalio/contrib/google_adk_agents/workflow.py b/temporalio/contrib/google_adk_agents/workflow.py index 24fc32c04..0652dcc39 100644 --- a/temporalio/contrib/google_adk_agents/workflow.py +++ b/temporalio/contrib/google_adk_agents/workflow.py @@ -3,7 +3,7 @@ import functools import inspect import typing -from typing import TYPE_CHECKING, Any, Callable, Optional +from typing import TYPE_CHECKING, Any, Callable import temporalio.workflow from temporalio import workflow @@ -64,7 +64,7 @@ async def wrapper(*args: Any, **kw: Any): def activity_node( activity_def: Callable, *, - name: Optional[str] = None, + name: str | None = None, rerun_on_resume: bool = False, **kwargs: Any, ) -> "FunctionNode": diff --git a/tests/contrib/google_adk_agents/test_adk_dynamic_workflows.py b/tests/contrib/google_adk_agents/test_adk_dynamic_workflows.py index 624db5190..849cc4564 100644 --- a/tests/contrib/google_adk_agents/test_adk_dynamic_workflows.py +++ b/tests/contrib/google_adk_agents/test_adk_dynamic_workflows.py @@ -18,7 +18,7 @@ import uuid from collections.abc import AsyncGenerator from datetime import timedelta -from typing import Any, Optional +from typing import Any import pytest from google.adk.agents import LlmAgent @@ -158,7 +158,7 @@ class WorkflowToolModel(BaseLlm): async def generate_content_async( self, llm_request: LlmRequest, stream: bool = False ) -> AsyncGenerator[LlmResponse, None]: - tool_response: Optional[types.FunctionResponse] = None + tool_response: types.FunctionResponse | None = None for content in llm_request.contents: for part in content.parts or []: if part.function_response is not None: diff --git a/tests/contrib/google_adk_agents/test_adk_graph_workflows.py b/tests/contrib/google_adk_agents/test_adk_graph_workflows.py index f7b8617ff..ba17bdd20 100644 --- a/tests/contrib/google_adk_agents/test_adk_graph_workflows.py +++ b/tests/contrib/google_adk_agents/test_adk_graph_workflows.py @@ -123,10 +123,10 @@ 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: + def handle_bug(node_input: str) -> str: # pyright: ignore[reportUnusedParameter] return "routed-to-bug" - def handle_other(node_input: str) -> str: + def handle_other(node_input: str) -> str: # pyright: ignore[reportUnusedParameter] return "routed-to-other" graph = Workflow( @@ -145,10 +145,10 @@ class ParallelJoinGraphWorkflow: @workflow.run async def run(self, prompt: str) -> dict[str, Any]: - def make_a(node_input: Any) -> str: + def make_a(node_input: Any) -> str: # pyright: ignore[reportUnusedParameter] return "alpha" - def make_b(node_input: Any) -> str: + def make_b(node_input: Any) -> str: # pyright: ignore[reportUnusedParameter] return "beta" enrich_a = activity_node( @@ -177,7 +177,7 @@ class MultiParamActivityNodeWorkflow: @workflow.run async def run(self, prompt: str) -> str: - def prepare(node_input: Any) -> dict[str, str]: + def prepare(node_input: Any) -> dict[str, str]: # pyright: ignore[reportUnusedParameter] return {"left": "L", "right": "R"} combine = activity_node( @@ -216,7 +216,7 @@ class TimeoutGraphWorkflow: @workflow.run async def run(self, prompt: str) -> str: - async def slow(node_input: Any) -> str: + async def slow(node_input: Any) -> str: # pyright: ignore[reportUnusedParameter] await asyncio.sleep(5) return "never" @@ -237,7 +237,7 @@ class RetryGraphWorkflow: async def run(self, prompt: str) -> str: attempts: list[int] = [] - def flaky(node_input: Any) -> str: + def flaky(node_input: Any) -> str: # pyright: ignore[reportUnusedParameter] attempts.append(1) if len(attempts) < 2: raise RuntimeError("transient failure") @@ -264,7 +264,7 @@ class JitteredRetryGraphWorkflow: async def run(self, prompt: str) -> str: attempts: list[int] = [] - def flaky(node_input: Any) -> str: + def flaky(node_input: Any) -> str: # pyright: ignore[reportUnusedParameter] attempts.append(1) if len(attempts) < 2: raise RuntimeError("transient failure") diff --git a/tests/contrib/google_adk_agents/test_adk_hitl.py b/tests/contrib/google_adk_agents/test_adk_hitl.py index a0894bf2f..006644b43 100644 --- a/tests/contrib/google_adk_agents/test_adk_hitl.py +++ b/tests/contrib/google_adk_agents/test_adk_hitl.py @@ -18,7 +18,7 @@ import uuid from collections.abc import AsyncGenerator from datetime import timedelta -from typing import Any, Optional +from typing import Any import pytest from google.adk.agents import LlmAgent @@ -327,7 +327,7 @@ async def _wait_for_pending( handle: WorkflowHandle, query: Any, count: int = 1, - expected_ids: Optional[set[str]] = None, + expected_ids: set[str] | None = None, ) -> list[HitlRequest]: async def _poll() -> list[HitlRequest]: while True: @@ -422,7 +422,7 @@ async def test_hitl_multiple_pending_partial_response(client: Client): task_queue=TASK_QUEUE, execution_timeout=timedelta(seconds=60), ) - pending = await _wait_for_pending( + await _wait_for_pending( handle, MultiPendingGraphWorkflow.pending_requests, expected_ids={"a", "b"} ) @@ -430,7 +430,7 @@ async def test_hitl_multiple_pending_partial_response(client: Client): await handle.execute_update( MultiPendingGraphWorkflow.respond, args=["a", "yes-a"] ) - pending = await _wait_for_pending( + await _wait_for_pending( handle, MultiPendingGraphWorkflow.pending_requests, expected_ids={"b"} ) From fe9ff1a137ac210561bba98a4ba8b5b06eda548e Mon Sep 17 00:00:00 2001 From: DABH Date: Mon, 17 Aug 2026 19:43:03 -0500 Subject: [PATCH 06/14] Pin google-adk to upstream main via uv source --- pyproject.toml | 10 ++++++++-- tests/contrib/google_adk_agents/test_adk_streaming.py | 6 +++++- uv.lock | 11 ++--------- 3 files changed, 15 insertions(+), 12 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index b46f8f85a..c32169265 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -279,5 +279,11 @@ 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 } + +# Dev-only resolution override: the deterministic-runtime seams this plugin +# installs (platform random provider; RequestInput/function-call ids routed +# through the platform uuid provider) are on google-adk main but not yet in a +# release. TODO: remove once a google-adk release contains +# google/adk-python@8f85107c and resolve from PyPI again. +[tool.uv.sources] +google-adk = { git = "https://github.com/google/adk-python", rev = "8f85107cca7fa9d88eea5ca60e32b85173b4ec7c" } diff --git a/tests/contrib/google_adk_agents/test_adk_streaming.py b/tests/contrib/google_adk_agents/test_adk_streaming.py index 30aecd9f4..b15e75cdf 100644 --- a/tests/contrib/google_adk_agents/test_adk_streaming.py +++ b/tests/contrib/google_adk_agents/test_adk_streaming.py @@ -12,7 +12,11 @@ import pytest from google.adk import Agent -from google.adk.agents.run_config import RunConfig, StreamingMode + +# google-adk does not re-export StreamingMode publicly; import it from its +# defining module, as ADK's own code does. +from google.adk.agents._streaming_mode import StreamingMode +from google.adk.agents.run_config import RunConfig from google.adk.models import BaseLlm, LLMRegistry from google.adk.models.llm_request import LlmRequest from google.adk.models.llm_response import LlmResponse diff --git a/uv.lock b/uv.lock index 50a599f87..6fa29ecdd 100644 --- a/uv.lock +++ b/uv.lock @@ -12,9 +12,6 @@ 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" @@ -1302,7 +1299,7 @@ wheels = [ [[package]] name = "google-adk" version = "2.7.1" -source = { registry = "https://pypi.org/simple" } +source = { git = "https://github.com/google/adk-python?rev=8f85107cca7fa9d88eea5ca60e32b85173b4ec7c#8f85107cca7fa9d88eea5ca60e32b85173b4ec7c" } dependencies = [ { name = "aiohttp" }, { name = "aiosqlite" }, @@ -1330,10 +1327,6 @@ dependencies = [ { name = "watchdog" }, { name = "websockets" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/02/55/63336b5cd9679a32b36cb14971a7461811db20126bb63cb092fe653b8d54/google_adk-2.7.1.tar.gz", hash = "sha256:0485bfba1a04960a3784eb67531491475be1a3f8acefcc5c880a906857ecb3e2", size = 3806947, upload-time = "2026-08-17T18:33:11.841Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/a4/ed/6b3da8b43525d9b28a95836abb1042abc26fe1e8be9f63eb01c18b70d4d4/google_adk-2.7.1-py3-none-any.whl", hash = "sha256:cd21e37c9846a80086fd880924aa5548e625ef85178122b40cda81ef5316129f", size = 4396591, upload-time = "2026-08-17T18:33:09.596Z" }, -] [[package]] name = "google-auth" @@ -4810,7 +4803,7 @@ dev = [ requires-dist = [ { name = "aioboto3", marker = "extra == 'aioboto3'", specifier = ">=10.4.0" }, { name = "deepagents", marker = "python_full_version >= '3.11' and extra == 'deepagents'", specifier = ">=0.6.12,<0.7" }, - { name = "google-adk", marker = "extra == 'google-adk'", specifier = ">=2.5.0,<3" }, + { name = "google-adk", marker = "extra == 'google-adk'", git = "https://github.com/google/adk-python?rev=8f85107cca7fa9d88eea5ca60e32b85173b4ec7c" }, { 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 = "langchain", marker = "python_full_version >= '3.11' and extra == 'deepagents'", specifier = ">=1.3.11,<2" }, From 69f033e2e116483476c25fe7765bc5bcbea65b53 Mon Sep 17 00:00:00 2001 From: DABH Date: Mon, 17 Aug 2026 19:43:03 -0500 Subject: [PATCH 07/14] Install the ADK random provider unconditionally --- CHANGELOG.md | 8 ++++ .../contrib/google_adk_agents/README.md | 12 +++--- .../contrib/google_adk_agents/_plugin.py | 37 +++++++------------ .../test_adk_graph_workflows.py | 11 ------ .../google_adk_agents/test_adk_hitl.py | 16 -------- 5 files changed, 28 insertions(+), 56 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e2ce1f736..6fd3fb798 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -35,6 +35,14 @@ to include examples, links to docs, or any other relevant information. `uuid.uuid1()`/`uuid.uuid4()` restrictions. - **Experimental**: `TemporalOperationHandler` can now use Standalone Activities as asynchronous Nexus Operation backing executions through `TemporalNexusClient.start_activity`. +- **Experimental**: `temporalio.contrib.google_adk_agents` now supports ADK v2 + graph workflows (including `activity_node(...)` for running Temporal + activities as graph nodes), dynamic `@node` workflows, and durable + human-in-the-loop via the `HitlRequest` / `pending_hitl_requests` / + `hitl_input_response` / `hitl_confirmation_response` helpers. The plugin + installs ADK's platform time, uuid, and random providers as process-wide + defaults so ADK-generated timestamps, ids (including default `RequestInput` + interrupt ids), and retry jitter replay deterministically. ### Changed diff --git a/temporalio/contrib/google_adk_agents/README.md b/temporalio/contrib/google_adk_agents/README.md index 53da640ae..ab564f2f9 100644 --- a/temporalio/contrib/google_adk_agents/README.md +++ b/temporalio/contrib/google_adk_agents/README.md @@ -379,15 +379,15 @@ 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). +> providers as process-wide defaults, so the ids ADK generates (including +> default `RequestInput` interrupt ids) derive from `workflow.uuid4()` and +> replay identically. ## 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. +- The plugin patches ADK's `google.adk.platform` time, uuid, and 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(...)` diff --git a/temporalio/contrib/google_adk_agents/_plugin.py b/temporalio/contrib/google_adk_agents/_plugin.py index b7095b28e..ac6ab96c2 100644 --- a/temporalio/contrib/google_adk_agents/_plugin.py +++ b/temporalio/contrib/google_adk_agents/_plugin.py @@ -53,13 +53,14 @@ def setup_deterministic_runtime(): This function is experimental and may change in future versions. Use with caution in production environments. - 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. + Installs Temporal-aware time, uuid, and 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._random import google.adk.platform.time import google.adk.platform.uuid @@ -74,6 +75,13 @@ def _deterministic_id_provider() -> str: return str(workflow.uuid4()) return str(uuid.uuid4()) + _local_random = random.Random() + + def _deterministic_random_provider() -> random.Random: + if workflow.in_workflow(): + return workflow.random() + return _local_random + _install_provider( google.adk.platform.time, "_time_provider_context_var", @@ -84,23 +92,6 @@ def _deterministic_id_provider() -> str: "_id_provider_context_var", _deterministic_id_provider, ) - except ImportError: - pass - 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. - import google.adk.platform._random # type: ignore - - _local_random = random.Random() - - def _deterministic_random_provider() -> random.Random: - if workflow.in_workflow(): - return workflow.random() - return _local_random - _install_provider( google.adk.platform._random, "_random_provider_context_var", @@ -109,7 +100,7 @@ def _deterministic_random_provider() -> random.Random: except ImportError: pass except Exception as e: - print(f"Warning: Failed to set deterministic random provider: {e}") + print(f"Warning: Failed to set deterministic runtime providers: {e}") class GoogleAdkPlugin(SimplePlugin): diff --git a/tests/contrib/google_adk_agents/test_adk_graph_workflows.py b/tests/contrib/google_adk_agents/test_adk_graph_workflows.py index ba17bdd20..dd92f7fe8 100644 --- a/tests/contrib/google_adk_agents/test_adk_graph_workflows.py +++ b/tests/contrib/google_adk_agents/test_adk_graph_workflows.py @@ -49,13 +49,6 @@ 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.""" @@ -424,10 +417,6 @@ async def test_graph_node_retry(client: Client): @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( diff --git a/tests/contrib/google_adk_agents/test_adk_hitl.py b/tests/contrib/google_adk_agents/test_adk_hitl.py index d6f30cfe7..e1495e8f8 100644 --- a/tests/contrib/google_adk_agents/test_adk_hitl.py +++ b/tests/contrib/google_adk_agents/test_adk_hitl.py @@ -53,17 +53,6 @@ _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.""" @@ -443,11 +432,6 @@ async def test_hitl_multiple_pending_partial_response(client: Client): @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( From 831bf886a51ce0cb2701f694788e60f7121f769e Mon Sep 17 00:00:00 2001 From: DABH Date: Mon, 17 Aug 2026 20:17:06 -0500 Subject: [PATCH 08/14] Fix unresolvable doc link in activity_node docstring --- temporalio/contrib/google_adk_agents/workflow.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/temporalio/contrib/google_adk_agents/workflow.py b/temporalio/contrib/google_adk_agents/workflow.py index 7c09a688b..a68b9919d 100644 --- a/temporalio/contrib/google_adk_agents/workflow.py +++ b/temporalio/contrib/google_adk_agents/workflow.py @@ -267,7 +267,7 @@ def activity_node( 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 + The returned ``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 From 2b7c969d1529aa8a4953363c2a4ecb0902e051a5 Mon Sep 17 00:00:00 2001 From: DABH Date: Fri, 4 Sep 2026 01:49:10 -0500 Subject: [PATCH 09/14] Require google-adk>=2.8.0 and drop the upstream git pin google-adk 2.8.0 (released 2026-08-26) is the first release containing the platform seams this plugin relies on: the random provider and the routing of RequestInput interrupt ids, _ToolNode function-call ids, and retry jitter through google.adk.platform (google/adk-python@8f85107c), plus the deterministic ParallelWorker failure selection (google/adk-python@d31b5e7d). Resolve google-adk from PyPI again and raise the extra's floor accordingly. The repo's exclude-newer window is two weeks, so a per-package exclude-newer-package override admits 2.8.0 until it ages into the window (2026-09-10); it can be removed after that. --- CHANGELOG.md | 1 + pyproject.toml | 12 ++------ uv.lock | 75 +++++++++++++++++++++++++++----------------------- 3 files changed, 45 insertions(+), 43 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f22db64f9..b9ad1b278 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -42,6 +42,7 @@ to include examples, links to docs, or any other relevant information. ### :boom: Breaking Changes +- The `google-adk` extra now requires `google-adk>=2.8.0,<3`, up from `>=2.2.0`. - Experimental external storage: `ExternalStorage.driver_selector` is now called with a `StorageDriverSelectContext` instead of a `StorageDriverStoreContext`. Update the annotation; the new type carries the same `target` field. Since selectors are plain callables, a stale diff --git a/pyproject.toml b/pyproject.toml index 6fb21a049..7ebb6201e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,7 +29,7 @@ grpc = ["grpcio>=1.48.2,<2"] opentelemetry = ["opentelemetry-api>=1.26,<2", "opentelemetry-sdk>=1.26,<2"] pydantic = ["pydantic>=2.0.0,<3"] openai-agents = ["openai-agents>=0.19.2,<0.20", "mcp>=1.9.4, <2"] -google-adk = ["google-adk>=2.5.0,<3", "mcp>=1.24,<2"] +google-adk = ["google-adk>=2.8.0,<3", "mcp>=1.24,<2"] langgraph = ["langgraph>=1.1.0"] langsmith = ["langsmith>=0.7.34,<0.9"] deepagents = [ @@ -284,11 +284,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" - -# Dev-only resolution override: the deterministic-runtime seams this plugin -# installs (platform random provider; RequestInput/function-call ids routed -# through the platform uuid provider) are on google-adk main but not yet in a -# release. TODO: remove once a google-adk release contains -# google/adk-python@8f85107c and resolve from PyPI again. -[tool.uv.sources] -google-adk = { git = "https://github.com/google/adk-python", rev = "8f85107cca7fa9d88eea5ca60e32b85173b4ec7c" } +# Admit google-adk 2.8.0 (released 2026-08-26) inside the 2-week window; drop after 2026-09-10. +exclude-newer-package = { google-adk = "2026-08-27T00:00:00Z" } diff --git a/uv.lock b/uv.lock index de14c689c..a956b91f4 100644 --- a/uv.lock +++ b/uv.lock @@ -9,9 +9,12 @@ resolution-markers = [ ] [options] -exclude-newer = "2026-08-10T18:40:15.391197Z" +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 = "2026-08-27T00:00:00Z" + [[package]] name = "aioboto3" version = "15.5.0" @@ -257,14 +260,14 @@ name = "anthropic" version = "0.117.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "anyio", marker = "python_full_version >= '3.11'" }, - { name = "distro", marker = "python_full_version >= '3.11'" }, - { name = "docstring-parser", marker = "python_full_version >= '3.11'" }, - { name = "httpx", marker = "python_full_version >= '3.11'" }, - { name = "jiter", marker = "python_full_version >= '3.11'" }, - { name = "pydantic", marker = "python_full_version >= '3.11'" }, - { name = "sniffio", marker = "python_full_version >= '3.11'" }, - { name = "typing-extensions", marker = "python_full_version >= '3.11'" }, + { name = "anyio" }, + { name = "distro" }, + { name = "docstring-parser" }, + { name = "httpx" }, + { name = "jiter" }, + { name = "pydantic" }, + { name = "sniffio" }, + { name = "typing-extensions" }, ] sdist = { url = "https://files.pythonhosted.org/packages/41/0d/8f71d535edb0d438f023bd825fb65f67c14fa88a2bd6b75f292a58a63de4/anthropic-0.117.0.tar.gz", hash = "sha256:98107f2b76439641e0ae2a1754087534b8f178dbab99d6eb1bc4b7bc8c744496", size = 989933, upload-time = "2026-07-16T19:36:13.07Z" } wheels = [ @@ -942,12 +945,12 @@ name = "deepagents" version = "0.6.12" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "langchain", marker = "python_full_version >= '3.11'" }, - { name = "langchain-anthropic", marker = "python_full_version >= '3.11'" }, - { name = "langchain-core", marker = "python_full_version >= '3.11'" }, - { name = "langchain-google-genai", marker = "python_full_version >= '3.11'" }, - { name = "langsmith", marker = "python_full_version >= '3.11'" }, - { name = "wcmatch", marker = "python_full_version >= '3.11'" }, + { name = "langchain" }, + { name = "langchain-anthropic" }, + { name = "langchain-core" }, + { name = "langchain-google-genai" }, + { name = "langsmith" }, + { name = "wcmatch" }, ] sdist = { url = "https://files.pythonhosted.org/packages/e0/db/a6acdc72a9e90c3f07ed10de35c951734a02d4facb693bb59684ad368801/deepagents-0.6.12.tar.gz", hash = "sha256:1f281c0bc5a63132f62e2ee345c1dc593b23188da6e23016401f6879fbe54b5f", size = 211364, upload-time = "2026-06-25T17:26:52.775Z" } wheels = [ @@ -1022,7 +1025,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 = [ @@ -1298,8 +1301,8 @@ wheels = [ [[package]] name = "google-adk" -version = "2.7.1" -source = { git = "https://github.com/google/adk-python?rev=8f85107cca7fa9d88eea5ca60e32b85173b4ec7c#8f85107cca7fa9d88eea5ca60e32b85173b4ec7c" } +version = "2.8.0" +source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "aiohttp" }, { name = "aiosqlite" }, @@ -1327,6 +1330,10 @@ dependencies = [ { name = "watchdog" }, { name = "websockets" }, ] +sdist = { url = "https://files.pythonhosted.org/packages/c7/29/db8042eb489515ef64fc16d24f1a621523451bfc2cbbae0103527b4066f4/google_adk-2.8.0.tar.gz", hash = "sha256:f51524e18cf0a0cdeb4fdd6f0fa16f31bc5e53021647b3e6c72a90f40f583915", size = 3902380, upload-time = "2026-08-26T23:26:20.501Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/0c/9d/8447a0912dcba1fa5dae83697e1af0a7ca9afa4701f93c82f3a1caf20561/google_adk-2.8.0-py3-none-any.whl", hash = "sha256:616bfa21959ae2726432670cb0b1c549e4908d9bf6908bff6fbc08382b075429", size = 4502706, upload-time = "2026-08-26T23:26:17.53Z" }, +] [[package]] name = "google-auth" @@ -1351,7 +1358,7 @@ requests = [ [[package]] name = "google-genai" -version = "2.16.0" +version = "2.19.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "anyio" }, @@ -1365,9 +1372,9 @@ dependencies = [ { name = "typing-extensions" }, { name = "websockets" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/81/e6/ff83088427072cc9d5d21036788cf0ed08cc4906e4a5810e469553a43185/google_genai-2.16.0.tar.gz", hash = "sha256:c4c2524926001b18073db927a5d75bb7c8be7b5fd13ab507d599f51fff2284c5", size = 647939, upload-time = "2026-07-30T14:34:37.366Z" } +sdist = { url = "https://files.pythonhosted.org/packages/37/1a/a834dfed90cf32dba900b533a1d14dcdefbda398bda5661d2a5a60fdc9fc/google_genai-2.19.0.tar.gz", hash = "sha256:d8f4126643793a7de230c396bcd142d21c948c8bb57507580e152549a7a41d9d", size = 659496, upload-time = "2026-08-19T23:05:43.276Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/3b/c6/f111056110030b1a5fb949687d7f93c2b4e8996f6494ae32efb049482796/google_genai-2.16.0-py3-none-any.whl", hash = "sha256:f9eda6a7a3dd4491a0d2253c4bdd4536462d63838ed3f1b0e4fb9a0eb8f43331", size = 1050096, upload-time = "2026-07-30T14:34:35.578Z" }, + { url = "https://files.pythonhosted.org/packages/01/e8/de0accd8cd004cf11252ca53fc5c3dda59bcf1856abfc7609842ceba7363/google_genai-2.19.0-py3-none-any.whl", hash = "sha256:36e0326dd886b52ef765be4c46042732b46b21f637abbe060e3db7c3de23974c", size = 1051056, upload-time = "2026-08-19T23:05:41.462Z" }, ] [[package]] @@ -1968,9 +1975,9 @@ name = "langchain" version = "1.3.14" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "langchain-core", marker = "python_full_version >= '3.11'" }, - { name = "langgraph", marker = "python_full_version >= '3.11'" }, - { name = "pydantic", marker = "python_full_version >= '3.11'" }, + { name = "langchain-core" }, + { name = "langgraph" }, + { name = "pydantic" }, ] sdist = { url = "https://files.pythonhosted.org/packages/29/68/a6dbad9c22df4087a0f9e79ddd46226c442b30128bfeee538d5889492a73/langchain-1.3.14.tar.gz", hash = "sha256:1b6696c72ba3bbbce54d745e0180742c9f6ece8bbc59ed5a46c3e20b9a435929", size = 645181, upload-time = "2026-07-16T13:28:18.29Z" } wheels = [ @@ -1982,9 +1989,9 @@ name = "langchain-anthropic" version = "1.4.8" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "anthropic", marker = "python_full_version >= '3.11'" }, - { name = "langchain-core", marker = "python_full_version >= '3.11'" }, - { name = "pydantic", marker = "python_full_version >= '3.11'" }, + { name = "anthropic" }, + { name = "langchain-core" }, + { name = "pydantic" }, ] sdist = { url = "https://files.pythonhosted.org/packages/98/22/40ab129b08329ca295b391aa1d48267692b42594757084c6918e22b655ac/langchain_anthropic-1.4.8.tar.gz", hash = "sha256:c76891b2044d56105ff13c106ed12650637b53bd598a4bdf15b4796eefa2a4ec", size = 708524, upload-time = "2026-06-26T21:28:46.916Z" } wheels = [ @@ -2016,10 +2023,10 @@ name = "langchain-google-genai" version = "4.2.7" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "filetype", marker = "python_full_version >= '3.11'" }, - { name = "google-genai", marker = "python_full_version >= '3.11'" }, - { name = "langchain-core", marker = "python_full_version >= '3.11'" }, - { name = "pydantic", marker = "python_full_version >= '3.11'" }, + { name = "filetype" }, + { name = "google-genai" }, + { name = "langchain-core" }, + { name = "pydantic" }, ] sdist = { url = "https://files.pythonhosted.org/packages/1d/0c/bc60dabc362ca7c6ffe8c4bcc2f724c7e566b43eb230cee51419f88f784c/langchain_google_genai-4.2.7.tar.gz", hash = "sha256:03b1463ffe4d42435f43c7870467f2215f684bb46400d2543435d10157c80ac7", size = 281605, upload-time = "2026-07-06T13:51:58.724Z" } wheels = [ @@ -2835,7 +2842,7 @@ wheels = [ [package.optional-dependencies] litellm = [ - { name = "litellm", marker = "python_full_version < '3.14'" }, + { name = "litellm" }, ] [[package]] @@ -4803,7 +4810,7 @@ dev = [ requires-dist = [ { name = "aioboto3", marker = "extra == 'aioboto3'", specifier = ">=10.4.0" }, { name = "deepagents", marker = "python_full_version >= '3.11' and extra == 'deepagents'", specifier = ">=0.6.12,<0.7" }, - { name = "google-adk", marker = "extra == 'google-adk'", git = "https://github.com/google/adk-python?rev=8f85107cca7fa9d88eea5ca60e32b85173b4ec7c" }, + { name = "google-adk", marker = "extra == 'google-adk'", specifier = ">=2.8.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 = "langchain", marker = "python_full_version >= '3.11' and extra == 'deepagents'", specifier = ">=1.3.11,<2" }, @@ -5364,7 +5371,7 @@ name = "wcmatch" version = "11.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "bracex", marker = "python_full_version >= '3.11'" }, + { name = "bracex" }, ] sdist = { url = "https://files.pythonhosted.org/packages/16/25/1da725838132221e33568973da484ff43813662ccc06ebf7f6e3abddfcd5/wcmatch-11.0.tar.gz", hash = "sha256:55d95c2447789712774b198ceec72939e88b5618f1f8f0a9b605bf7740b63b96", size = 141360, upload-time = "2026-07-10T05:50:24.183Z" } wheels = [ From 91f157dab4acb048f04f5535c7e947c1cd3c1f6a Mon Sep 17 00:00:00 2001 From: DABH Date: Fri, 4 Sep 2026 02:18:22 -0500 Subject: [PATCH 10/14] Trim _install_provider docstring to one line --- temporalio/contrib/google_adk_agents/_plugin.py | 11 +---------- 1 file changed, 1 insertion(+), 10 deletions(-) diff --git a/temporalio/contrib/google_adk_agents/_plugin.py b/temporalio/contrib/google_adk_agents/_plugin.py index fd352b9cb..ca9d067ed 100644 --- a/temporalio/contrib/google_adk_agents/_plugin.py +++ b/temporalio/contrib/google_adk_agents/_plugin.py @@ -38,16 +38,7 @@ 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). - """ + """Rebinds an ADK platform ContextVar so ``provider`` is its default in every context.""" from contextvars import ContextVar context_var = getattr(module, var_name) From e4b35b99affaa7730557701cea63221394eaef6d Mon Sep 17 00:00:00 2001 From: DABH Date: Tue, 8 Sep 2026 11:49:19 -0500 Subject: [PATCH 11/14] Bind positional-only activity parameters in activity_node activity.defn only rejects keyword-only parameters, so positional-only ones are valid activity signatures. activity_node filtered them out of its parameter list, so a single positional-only parameter fell into the zero-argument branch and the activity was scheduled without its input, and mixed signatures were miscounted. Include positional-only parameters and bind dict inputs to positional arguments by name, with clear errors for missing or unexpected keys, instead of Signature.bind, which cannot bind positional-only parameters from keywords. --- .../contrib/google_adk_agents/workflow.py | 30 +++++++-- .../test_adk_graph_workflows.py | 67 ++++++++++++++++++- 2 files changed, 92 insertions(+), 5 deletions(-) diff --git a/temporalio/contrib/google_adk_agents/workflow.py b/temporalio/contrib/google_adk_agents/workflow.py index a68b9919d..66666e6e4 100644 --- a/temporalio/contrib/google_adk_agents/workflow.py +++ b/temporalio/contrib/google_adk_agents/workflow.py @@ -254,6 +254,29 @@ async def wrapper(*args: Any, **kw: Any): return wrapper +def _bind_node_input( + activity_def: Callable, params: list[inspect.Parameter], node_input: dict[str, Any] +) -> list[Any]: + """Maps a dict node input onto the activity's positional parameters by name.""" + unknown = sorted(set(node_input) - {p.name for p in params}) + if unknown: + raise TypeError( + f"Activity node '{activity_def.__name__}' got unexpected input" + f" key(s) {unknown}." + ) + missing = [ + p.name + for p in params + if p.name not in node_input and p.default is inspect.Parameter.empty + ] + if missing: + raise TypeError( + f"Activity node '{activity_def.__name__}' is missing required input" + f" key(s) {missing}." + ) + return [node_input.get(p.name, p.default) for p in params] + + def activity_node( activity_def: Callable, *, @@ -291,11 +314,12 @@ def activity_node( from google.adk.workflow import FunctionNode sig = inspect.signature(activity_def) + # Activities are invoked positionally; activity.defn rejects keyword-only parameters. params = [ p for p in sig.parameters.values() if p.kind - in (inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY) + in (inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD) ] # Resolve hints against the activity's own module so that string # annotations (e.g. under `from __future__ import annotations`) still @@ -320,9 +344,7 @@ async def wrapper(node_input: Any = None) -> Any: 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()) + activity_args = _bind_node_input(activity_def, params, node_input) if not temporalio.workflow.in_workflow(): # Outside a workflow, like local adk runs, use the function directly. diff --git a/tests/contrib/google_adk_agents/test_adk_graph_workflows.py b/tests/contrib/google_adk_agents/test_adk_graph_workflows.py index dd92f7fe8..bbc7d4fad 100644 --- a/tests/contrib/google_adk_agents/test_adk_graph_workflows.py +++ b/tests/contrib/google_adk_agents/test_adk_graph_workflows.py @@ -67,6 +67,18 @@ async def combine_parts(left: str, right: str) -> str: return f"{left}+{right}" +@activity.defn +async def shout(text: str, /) -> str: + """Activity with a positional-only parameter.""" + return text.upper() + + +@activity.defn +async def join_parts(left: str, right: str, /, sep: str = "-") -> str: + """Activity with positional-only parameters and a defaulted one.""" + return f"{left}{sep}{right}" + + async def drive_graph(graph: Workflow, prompt: str) -> Any: """Runs an ADK graph to completion in-workflow, returning the last output.""" runner = Runner( @@ -180,6 +192,24 @@ def prepare(node_input: Any) -> dict[str, str]: # pyright: ignore[reportUnusedP return await drive_graph(graph, prompt) +@workflow.defn +class PositionalOnlyGraphWorkflow: + """Positional-only activity parameters bind from node_input like any other.""" + + @workflow.run + async def run(self, prompt: str) -> str: + shout_node = activity_node(shout, start_to_close_timeout=timedelta(seconds=30)) + + def prepare(node_input: str) -> dict[str, str]: + return {"left": node_input, "right": "done"} + + join = activity_node(join_parts, start_to_close_timeout=timedelta(seconds=30)) + graph = Workflow( + name="positional_only", edges=[(START, shout_node, prepare, join)] + ) + return await drive_graph(graph, prompt) + + @workflow.defn class AgentNodeGraphWorkflow: """An LlmAgent node inside a graph, calling the model via an activity.""" @@ -296,12 +326,13 @@ def _worker(client: Client) -> Worker: return Worker( client, task_queue=TASK_QUEUE, - activities=[fetch_data, enrich_item, combine_parts], + activities=[fetch_data, enrich_item, combine_parts, shout, join_parts], workflows=[ SequentialGraphWorkflow, RoutingGraphWorkflow, ParallelJoinGraphWorkflow, MultiParamActivityNodeWorkflow, + PositionalOnlyGraphWorkflow, AgentNodeGraphWorkflow, TimeoutGraphWorkflow, RetryGraphWorkflow, @@ -372,6 +403,40 @@ async def test_graph_multi_param_activity_node(client: Client): assert result == "L+R" +@pytest.mark.asyncio +async def test_graph_positional_only_activity_node(client: Client): + client = _adk_client(client) + async with _worker(client): + result = await client.execute_workflow( + PositionalOnlyGraphWorkflow.run, + "go", + id=f"graph-positional-only-{uuid.uuid4()}", + task_queue=TASK_QUEUE, + execution_timeout=timedelta(seconds=30), + ) + assert result == "GO-done" + + +async def test_activity_node_rejects_bad_dict_input(): + join = activity_node(join_parts) + + def missing_right(node_input: str) -> dict[str, str]: # pyright: ignore[reportUnusedParameter] + return {"left": "a"} + + def bogus_key(node_input: str) -> dict[str, Any]: # pyright: ignore[reportUnusedParameter] + return {"left": "a", "right": "b", "bogus": 1} + + # Outside a workflow the node calls the activity function directly. + with pytest.raises(TypeError, match=r"missing required input key\(s\) \['right'\]"): + await drive_graph( + Workflow(name="missing", edges=[(START, missing_right, join)]), "go" + ) + with pytest.raises(TypeError, match=r"unexpected input key\(s\) \['bogus'\]"): + await drive_graph( + Workflow(name="unexpected", edges=[(START, bogus_key, join)]), "go" + ) + + @pytest.mark.asyncio async def test_graph_llm_agent_node(client: Client): client = _adk_client(client) From 8a98a4d165d0aa2cc15b50ef454648a1b30ee51f Mon Sep 17 00:00:00 2001 From: DABH Date: Tue, 8 Sep 2026 11:49:19 -0500 Subject: [PATCH 12/14] Drop the unanswerable credential kind from the HITL helpers pending_hitl_requests surfaced adk_request_credential calls as kind="credential", but no helper could build the adk_request_credential response ADK's auth path consumes, so an auth flow discovered through this API could never be resumed. Supporting it properly is more than a response helper: ADK exchanges the credential with network I/O inside the flow, and the exchanged secret would be recorded in workflow history. Leave auth requests out of the helpers and document that credentials should be resolved worker-side. --- .../contrib/google_adk_agents/README.md | 5 ++++ temporalio/contrib/google_adk_agents/_hitl.py | 27 +++++++++---------- .../google_adk_agents/test_adk_hitl.py | 21 +++++++++++++++ 3 files changed, 38 insertions(+), 15 deletions(-) diff --git a/temporalio/contrib/google_adk_agents/README.md b/temporalio/contrib/google_adk_agents/README.md index 2127b7805..64ae28fd2 100644 --- a/temporalio/contrib/google_adk_agents/README.md +++ b/temporalio/contrib/google_adk_agents/README.md @@ -376,6 +376,11 @@ requesting confirmation via `tool_context.request_confirmation(...)` flow through the same loop. Partial responses are fine: unanswered requests stay pending across `run_async` turns. +Auth requests (`adk_request_credential`) are not covered by these helpers: ADK +exchanges the credential with network I/O inside the flow, and the exchanged +secret would be recorded in workflow history. Resolve credentials worker-side +instead (for example inside an activity or an MCP toolset factory). + > **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 diff --git a/temporalio/contrib/google_adk_agents/_hitl.py b/temporalio/contrib/google_adk_agents/_hitl.py index 83baf2054..a571b43c0 100644 --- a/temporalio/contrib/google_adk_agents/_hitl.py +++ b/temporalio/contrib/google_adk_agents/_hitl.py @@ -2,15 +2,18 @@ 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 +tool confirmation) 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. +Temporal workflow code. Auth requests (``adk_request_credential``) are not +covered: ADK exchanges the credential with network I/O inside the flow, and the +exchanged secret would be recorded in workflow history. Resolve credentials +worker-side instead. """ from __future__ import annotations @@ -26,14 +29,10 @@ # 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"] -] = { +_KIND_BY_FUNCTION_CALL_NAME: dict[str, Literal["input", "tool_confirmation"]] = { _REQUEST_INPUT_FUNCTION_CALL_NAME: "input", _REQUEST_CONFIRMATION_FUNCTION_CALL_NAME: "tool_confirmation", - _REQUEST_CREDENTIAL_FUNCTION_CALL_NAME: "credential", } @@ -47,8 +46,7 @@ class HitlRequest: Attributes: kind: ``"input"`` for a human-input node's ``RequestInput``, - ``"tool_confirmation"`` for a tool confirmation request, - ``"credential"`` for an auth request. + ``"tool_confirmation"`` for a tool confirmation 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. @@ -63,7 +61,7 @@ class HitlRequest: displaying what is being approved. """ - kind: Literal["input", "tool_confirmation", "credential"] + kind: Literal["input", "tool_confirmation"] interrupt_id: str invocation_id: str | None = None author: str | None = None @@ -84,7 +82,8 @@ def pending_hitl_requests(event: Event) -> list[HitlRequest]: 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`). + :func:`hitl_confirmation_response`). Auth requests + (``adk_request_credential``) are not surfaced; see the module docstring. Args: event: An event yielded by ``runner.run_async``. @@ -113,13 +112,11 @@ def pending_hitl_requests(event: Event) -> list[HitlRequest]: message = args.get("message") payload = args.get("payload") response_schema = args.get("response_schema") - elif kind == "tool_confirmation": + else: 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, diff --git a/tests/contrib/google_adk_agents/test_adk_hitl.py b/tests/contrib/google_adk_agents/test_adk_hitl.py index e1495e8f8..c4e26d9b8 100644 --- a/tests/contrib/google_adk_agents/test_adk_hitl.py +++ b/tests/contrib/google_adk_agents/test_adk_hitl.py @@ -331,6 +331,27 @@ async def _poll() -> list[HitlRequest]: return await asyncio.wait_for(_poll(), timeout=20) +def test_pending_hitl_requests_ignores_credential_requests(): + from google.adk.events import Event + + event = Event( + author="agent", + content=types.Content( + role="model", + parts=[ + types.Part( + function_call=types.FunctionCall( + name="adk_request_credential", id="auth-1", args={} + ) + ) + ], + ), + long_running_tool_ids={"auth-1"}, + ) + # Auth requests are deliberately not surfaced: no helper can answer them. + assert pending_hitl_requests(event) == [] + + @pytest.mark.asyncio async def test_human_input_node_update_resume(client: Client): client = _adk_client(client) From 9473bdb0f312587421b857160fdd18500d925105 Mon Sep 17 00:00:00 2001 From: DABH Date: Tue, 8 Sep 2026 11:49:19 -0500 Subject: [PATCH 13/14] Pass optional model SDKs through the ADK workflow sandbox ADK's content processor probes for the anthropic, litellm, and openai model classes on every LLM turn. Any of those SDKs that is installed but not yet imported gets imported inside each workflow sandbox on the first turn, which is slow enough to trip the workflow deadlock detector and fail the workflow task; the retried task re-executes live, which is how the duplicate-span failure in test_single_agent_telemetry showed up in CI. Pass them through like the OpenAI Agents plugin does for openai. --- CHANGELOG.md | 3 +++ temporalio/contrib/google_adk_agents/_plugin.py | 8 +++++++- 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b9ad1b278..5510902a4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -50,6 +50,9 @@ to include examples, links to docs, or any other relevant information. ### Fixed +- `GoogleAdkPlugin` now passes the optional `anthropic`, `litellm`, and `openai` SDKs through + the workflow sandbox. ADK probes them lazily on each LLM turn, and importing an installed one + inside every workflow sandbox was slow enough to trip the workflow deadlock detector. - `StrandsPlugin` now disables Botocore retries for its default Bedrock model so model request retries are handled exclusively by Temporal. - `temporalio.contrib.openai_agents` now honors the `retry-after-ms` and diff --git a/temporalio/contrib/google_adk_agents/_plugin.py b/temporalio/contrib/google_adk_agents/_plugin.py index ca9d067ed..e770d6d11 100644 --- a/temporalio/contrib/google_adk_agents/_plugin.py +++ b/temporalio/contrib/google_adk_agents/_plugin.py @@ -208,7 +208,13 @@ def workflow_runner(runner: WorkflowRunner | None) -> WorkflowRunner: return dataclasses.replace( runner, restrictions=runner.restrictions.with_passthrough_modules( - "google.adk", "google.genai", "mcp" + "google.adk", + "google.genai", + "mcp", + # ADK probes these optional model SDKs lazily on each LLM turn. + "anthropic", + "litellm", + "openai", ), ) return runner From fcdb53387fd7efecf0ea5d76a6d7cb2da5c6d43f Mon Sep 17 00:00:00 2001 From: DABH Date: Wed, 9 Sep 2026 18:42:53 -0500 Subject: [PATCH 14/14] Drop the temporary google-adk exclude-newer override google-adk 2.8.0 is now older than the repository's two-week exclude-newer window, so it resolves from PyPI under the normal cutoff and the per-package override (and the lock entry it generated) is no longer needed. --- pyproject.toml | 2 -- uv.lock | 3 --- 2 files changed, 5 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 7ebb6201e..a677fd120 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -284,5 +284,3 @@ exclude = ["temporalio/bridge/target/**/*", "temporalio/bridge/sdk-core/.git"] # Prevent uv commands from building the package by default package = false exclude-newer = "2 weeks" -# Admit google-adk 2.8.0 (released 2026-08-26) inside the 2-week window; drop after 2026-09-10. -exclude-newer-package = { google-adk = "2026-08-27T00:00:00Z" } diff --git a/uv.lock b/uv.lock index a956b91f4..cb5f1e7b6 100644 --- a/uv.lock +++ b/uv.lock @@ -12,9 +12,6 @@ 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 = "2026-08-27T00:00:00Z" - [[package]] name = "aioboto3" version = "15.5.0"