feat: Add graph pattern nodes for dynamic dispatch and composition

Add DynamicNode (runtime agent selection), NestedGraphNode (hierarchical
workflow composition), and DynamicParallelGroup (variable-count concurrent
execution). Extends CLI visualization with pattern-aware rendering
(diamond, parallelogram, sub-cluster shapes). Includes pattern samples,
node type reference, and design documentation.

Author: drahnreb

contributing/docs/advanced_graph_patterns.md

@@ -0,0 +1,250 @@
+# GraphAgent Node Types
+
+A GraphAgent is made of **nodes** connected by **edges**.  Each node
+represents one unit of work.  This document explains every node type, when
+to use each one, and why the pattern-API nodes exist alongside the older
+function-node approach.
+
+---
+
+## GraphNode — the fundamental building block
+
+`GraphNode` is the only node type you strictly need.  Everything else is
+built on top of it.
+
+A `GraphNode` wraps either an **agent** (any ADK `BaseAgent` subclass) or a
+**function** (any `async def` that takes `state` and `ctx`).
+
+```python
+# Pattern 1 (Explicit): Wrapping an agent
+graph.add_node(GraphNode(name="summarise", agent=summariser_agent))
+
+# Pattern 2 (Convenience): Wrapping an agent
+graph.add_node("summarise", agent=summariser_agent)
+
+# Wrapping a function
+async def clean(state, ctx):
+    return state.data.get("raw", "").strip()
+
+# Pattern 1 (Explicit)
+graph.add_node(GraphNode(name="clean", function=clean))
+
+# Pattern 2 (Convenience)
+graph.add_node("clean", function=clean)
+```
+
+**What it handles for you**
+- Feeds the right part of state to the agent as input (`input_mapper`)
+- Stores the agent's output back into state under the node name (`output_mapper`)
+- Supports four merge strategies: overwrite, append, sum, or custom
+
+**When to use it**: always — for individual agents or short custom logic.
+
+---
+
+## ParallelNodeGroup — static concurrency
+
+`ParallelNodeGroup` runs a **fixed set of nodes at the same time**.  The
+nodes to run and their names are declared when you build the graph.
+
+```python
+from google.adk.agents.graph import ParallelNodeGroup, JoinStrategy
+
+graph.add_group(ParallelNodeGroup(
+    name="fan_out",
+    nodes=["search_web", "search_db", "search_docs"],
+    join_strategy=JoinStrategy.WAIT_ALL,
+))
+```
+
+**Three join strategies**
+
+| Strategy | Meaning |
+|---|---|
+| `WAIT_ALL` | Wait for every node to finish before continuing |
+| `WAIT_ANY` | Continue as soon as the first node finishes |
+| `WAIT_N` | Continue after N nodes finish (you pick N) |
+
+**When to use it**: you know the exact set of agents to run in parallel at
+the time you write the code, and that set never changes at runtime.
+
+---
+
+## DynamicNode — choose the agent at runtime
+
+`DynamicNode` is a `GraphNode` that decides **which agent to call** by
+running a selector function against the current graph state.
+
+```python
+def pick_agent(state):
+    if state.data.get("task_type") == "complex":
+        return thorough_agent
+    return quick_agent
+
+# Pattern 1 (Explicit)
+graph.add_node(DynamicNode(
+    name="respond",
+    agent_selector=pick_agent,
+    fallback_agent=quick_agent,   # used if selector returns None
+))
+
+# Pattern 2 (Convenience)
+graph.add_node("respond", agent_selector=pick_agent, fallback_agent=quick_agent)
+```
+
+**Do you need it?**  Not strictly.  You can achieve the same result with a
+plain `GraphNode(function=...)`:
+
+```python
+async def dispatch(state, ctx):
+    agent = thorough_agent if state.data.get("task_type") == "complex" else quick_agent
+    node_ctx = ctx.model_copy(update={"user_content": ...})
+    output = ""
+    async for event in agent.run_async(node_ctx):
+        if event.content and event.content.parts:
+            output = event.content.parts[0].text or ""
+    return output
+```
+
+**Why DynamicNode exists**
+
+1. It removes ten lines of boilerplate that you'd copy-paste every time you
+   add a new routing node.
+2. It encapsulates agent selection logic with built-in fallback support.
+3. It has a built-in `fallback_agent` parameter — with a function node you
+   must remember to add that guard yourself.
+
+**When to use it**: any time you want to route to different agents based on
+what's in the state, especially if you have more than one such node or want
+the selection to show up in traces.
+
+---
+
+## NestedGraphNode — run a whole sub-graph as one step
+
+`NestedGraphNode` runs a complete `GraphAgent` as a single step inside a
+parent graph.  The parent sees only the sub-graph's final answer.
+
+```python
+# Pattern 1 (Explicit)
+graph.add_node(NestedGraphNode(
+    name="research",
+    graph_agent=research_pipeline,   # a full GraphAgent with its own nodes
+    inherit_session=True,            # True = share state; False = clean slate
+))
+
+# Pattern 2 (Convenience)
+graph.add_node("research", graph_agent=research_pipeline, inherit_session=True)
+```
+
+**Do you need it?**  No.  A function node can call a sub-graph directly:
+
+```python
+async def run_research(state, ctx):
+    sub_ctx = ctx.model_copy(update={"user_content": ...})
+    result = ""
+    async for event in research_graph._run_async_impl(sub_ctx):
+        if event.content and event.content.parts:
+            text = event.content.parts[0].text or ""
+            if text and not text.startswith("[GraphMetadata]"):  # must filter manually
+                result = text
+    return result
+```
+
+The function-node version **works**, and it worked before `NestedGraphNode`
+was added.  `NestedGraphNode` is purely a convenience that:
+
+1. Does the event-filtering for you (the inner graph emits empty sentinel
+   events at the end that would otherwise overwrite your result — a subtle
+   bug that's easy to introduce yourself).
+2. Gives you `inherit_session` as a first-class parameter instead of you
+   having to remember whether to copy the context or create a new session.
+3. Records the inner graph's iteration count and execution path in metadata
+   automatically.
+
+**When to use it**: when you want to embed a reusable multi-step sub-workflow
+as a single node and don't want to write the filtering and context-management
+boilerplate by hand.  If you've already written a working function-node
+version, there is no reason to switch.
+
+---
+
+## DynamicParallelGroup — variable-count concurrency
+
+`DynamicParallelGroup` is like `ParallelNodeGroup` except the **number of
+agents is decided at runtime** by reading the graph state.
+
+```python
+def create_agents(state):
+    n = int(state.data.get("num_branches", "3"))
+    return [ThoughtAgent(f"branch_{i}") for i in range(n)]
+
+def combine(results, state):
+    return "\n---\n".join(results)
+
+# Pattern 1 (Explicit)
+graph.add_node(DynamicParallelGroup(
+    name="explore",
+    agent_generator=create_agents,
+    aggregator=combine,
+    max_parallelism=5,    # never run more than 5 at once
+))
+
+# Pattern 2 (Convenience)
+graph.add_node("explore", agent_generator=create_agents, aggregator=combine, max_parallelism=5)
+```
+
+**How it differs from ParallelNodeGroup**
+
+| | `ParallelNodeGroup` | `DynamicParallelGroup` |
+|---|---|---|
+| Number of agents | Fixed at build time | Computed from state at runtime |
+| Custom aggregation | Not built-in | Required (`aggregator` param) |
+| Concurrency cap | None | `max_parallelism` |
+
+**Do you need it?**  If `N` is always fixed (say, always 3 branches), use
+`ParallelNodeGroup`.  If `N` changes per request — e.g. "run as many
+reviewers as there are items in the queue" — `DynamicParallelGroup` is the
+right tool and there is no equally clean alternative with existing APIs.
+
+**When to use it**: Tree of Thoughts, self-consistency sampling, fan-out
+over a variable-length list of items, or any other case where the number of
+parallel workers is not known when you write the code.
+
+---
+
+## Summary
+
+| Node type | Fixed at build time? | Needs agent upfront? | Custom aggregation? | Use when… |
+|---|---|---|---|---|
+| `GraphNode` | — | yes or function | — | Single agent or custom logic |
+| `ParallelNodeGroup` | yes | yes | no | Known fixed set of parallel agents |
+| `DynamicNode` | — | no (picked at runtime) | — | Different agents for different inputs |
+| `NestedGraphNode` | — | no (a full sub-graph) | — | Reusable sub-workflow as one step |
+| `DynamicParallelGroup` | — | no (generated at runtime) | yes | Variable number of parallel agents |
+
+---
+
+## Coverage (feature branch)
+
+Measured with `pytest --cov` across 1197 tests:
+
+| Module | Coverage |
+|---|---|
+| `graph/patterns.py` | 100% |
+| `graph/__init__.py` | 100% |
+| `graph/callbacks.py` | 100% |
+| `graph/graph_agent_config.py` | 100% |
+| `graph/graph_state.py` | 100% |
+| `graph/graph_tracing.py` | 100% |
+| `graph/graph_node.py` | 97% |
+| `graph/interrupt_service.py` | 97% |
+| `graph/parallel.py` | 97% |
+| `graph/interrupt.py` | 95% |
+| `checkpoints/checkpoint_service.py` | 98% |
+| `graph/graph_agent.py` | 71% |
+| **Branch total** | **89%** |
+
+`graph_agent.py` pulls the average down; the uncovered lines are primarily
+in the OTel baggage propagation paths and some telemetry-sampling branches
+that require a live tracing exporter to exercise.