[AI-303] google_adk_agents: graph workflows, dynamic workflows, and durable HITL (ADK v2)

CHANGELOG.md

@@ -20,6 +20,14 @@ to include examples, links to docs, or any other relevant information.
 
 ### Added
 
+- **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.
 - Added GCP Cloud Run serverless-worker OpenTelemetry plugin in `temporalio.contrib.opentelemetry`.
 - Added new options to ActivityHandle.describe() to retrieve associated payloads, such as activity input and outcome.
 - New properties and methods in ActivityExecution and ActivityExecutionDescription.
@@ -37,6 +45,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
@@ -52,6 +61,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.
 - **Experimental**: External storage metrics now report the wall-clock time storage was in flight.
   Previously each batch's duration was summed, over-reporting the time whenever storage operations
   ran concurrently.

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.2.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 = [

temporalio/contrib/google_adk_agents/README.md

@@ -248,6 +248,168 @@ 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_as_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_as_tool` with no extra plumbing —
+`FunctionTool(func=activity_as_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.
+
+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
+> 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 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.
+
 ## Telemetry and Workflow Replay
 
 ADK records OpenTelemetry metrics (scope `gcp.vertex.agent`, e.g.

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,
@@ -16,9 +22,13 @@
 
 __all__ = [
     "GoogleAdkPlugin",
+    "HitlRequest",
     "TemporalMcpToolSet",
     "TemporalMcpToolSetProvider",
     "TemporalStatefulMcpToolSet",
     "TemporalStatefulMcpToolSetProvider",
     "TemporalModel",
+    "hitl_confirmation_response",
+    "hitl_input_response",
+    "pending_hitl_requests",
 ]