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

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
 

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", "mcp>=1.24,<2"]
+google-adk = ["google-adk>=2.5.0,<3", "mcp>=1.24,<2"]
 langgraph = ["langgraph>=1.1.0"]
 langsmith = ["langsmith>=0.7.34,<0.9"]
 deepagents = [
@@ -279,3 +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"
+
+# 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" }

temporalio/contrib/google_adk_agents/README.md

@@ -248,6 +248,163 @@ 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.
+
+> **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.
+
 ## Integration Points
 
 This integration provides comprehensive support for running Google ADK Agents within Temporal workflows while maintaining:

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",
 ]