docs(workflow): Add and update workflow developer guides

Add comprehensive guides explaining:
- Workflow Graphs (nodes, edges, validation)
- Function Nodes (wrapping functions and generators)
- Join Nodes (aggregating parallel branches)
- Retry Config (resilience policies)
- Parallel Workers (concurrent list processing)
- Dynamic Nodes (runtime execution)

Update the main guides index and the adk-unit-guide skill rules.

Co-authored-by: Bo Yang <ybo@google.com>
PiperOrigin-RevId: 933979245

Author: boyangsvl

.agents/skills/adk-unit-guide/SKILL.md

@@ -53,6 +53,7 @@ Use the following structure and instructions to create the guide for the code un
 - Use unit test code as a starting point for the code example, if available.
 - When writing a sample agent, do not set the `model` attribute.
 - For workflow node samples, prefer using a simple Python function rather than extending `BaseNode` to demonstrate the node's logic, unless class extension is explicitly required for the use case.
+- When wrapping Python functions as workflow nodes, prefer using the `@node` decorator instead of `FunctionNode` directly, whenever possible.
 
 ## How it works
 
@@ -64,6 +65,8 @@ Use the following structure and instructions to create the guide for the code un
 ## Configuration options
 
 - If the code unit has configuration options (e.g., settings, configuration objects), document them in a table detailing parameters, types, default values, and descriptions.
+- **Do NOT** list options inherited from base classes. Focus only on options introduced by the code unit itself.
+- Dive into each option to provide detailed description and usage patterns, rather than just repeating the type and a brief description.
 - **Do NOT** list references of all attributes or methods of the classes. Exhaustive API references belong in auto-generated reference documentation, not in guides. Guides should focus on how to use the code unit.
 
 ## Advanced applications

docs/guides/README.md

@@ -11,3 +11,12 @@ This directory contains specific developer guides for the ADK Python implementat
 ### Events
 * [Event and NodeInfo](events/event/index.md) - Understanding Event and NodeInfo in workflows.
 * [RequestInput](events/request_input/index.md) - How to use RequestInput for human-in-the-loop interactions.
+
+### Workflows
+* [Workflow](workflow/workflow/index.md) - Graph-based orchestration of complex, multi-step agent interactions.
+* [Workflow Graphs](workflow/graph/index.md) - Understanding nodes, edges, and graph structures in workflows.
+* [Function Nodes](workflow/function_node/index.md) - Wrapping Python functions and generators as workflow nodes.
+* [JoinNode](workflow/join_node/index.md) - Synchronizing parallel execution paths in workflows.
+* [RetryConfig](workflow/retry_config/index.md) - Configuring retry policies for resilient workflow nodes.
+* [ParallelWorker](workflow/parallel_worker/index.md) - Processing lists of items concurrently in workflows.
+* [Dynamic Nodes](workflow/dynamic_nodes/index.md) - Scheduling and executing nodes dynamically at runtime.

docs/guides/workflow/dynamic_nodes/index.md

@@ -0,0 +1,199 @@
+# Dynamic Node Scheduling
+
+Dynamic node scheduling allows you to execute workflow nodes dynamically at runtime using `ctx.run_node()`. This enables imperative workflow construction using standard Python control flow instead of static graph edges.
+
+## Introduction
+
+While static graph definitions (`Workflow(edges=[...])`) are suitable for many structured tasks, some scenarios require more flexibility. For example, you might need to:
+- Loop a set of nodes until a condition is met (e.g., generator-evaluator loops).
+- Run a variable number of tasks in parallel based on runtime input (dynamic fan-out).
+- Conditionally execute nodes based on complex logic that is difficult to express in static edges.
+
+`ctx.run_node()` allows a parent node to execute a child node (which can be a function, an Agent, or another Workflow) and await its result.
+
+## Get started
+
+The following example demonstrates how to dynamically execute a child agent from a parent node.
+
+```python
+from google.adk import Agent, Context, Event, Workflow
+from google.adk.workflow import node
+
+# Define a child agent
+generate_headline = Agent(
+    name="generate_headline",
+    instruction="Write a catchy headline about the topic in the user message.",
+)
+
+
+# Define the parent orchestrator node (MUST have rerun_on_resume=True)
+@node(rerun_on_resume=True)
+async def orchestrate(ctx: Context, node_input: str) -> str:
+  # Dynamically execute the child agent and await its output
+  headline = await ctx.run_node(generate_headline, node_input=node_input)
+  
+  yield Event(output=headline)
+
+# Build the workflow
+root_agent = Workflow(
+    name="root_agent",
+    edges=[("START", orchestrate)],
+)
+```
+
+
+## How it works
+
+When `await ctx.run_node(node_like, ...)` is called:
+
+1.  **Orchestrator Registration**: The workflow's `DynamicNodeScheduler` registers the child node execution.
+2.  **State Tracking**: The execution state and events of the child node are tracked under the parent node's path (e.g., `parent_node@1/child_node@1`).
+3.  **Resumption Support**: If the child node interrupts (e.g., waiting for user input), the parent node is also paused. When the workflow resumes, the parent node is re-run from the beginning (`rerun_on_resume=True`), but previous successful `ctx.run_node()` calls are replayed from history (cached outputs are returned) to avoid re-executing completed steps.
+
+### Input Mapping
+
+The `node_input` passed to `ctx.run_node(node, node_input=value)` is delivered differently depending on the type of the child node:
+
+-   **Python Functions / FunctionNodes**: The `value` is passed directly to the function parameter named `node_input`. Other parameters are bound from the session state (default mode).
+-   **Agents (Single-Turn Mode)**: The `value` is converted to a user-role message (`types.Content`) and appended to the session events history. The agent receives it as the incoming user message.
+-   **Agents (Task Mode)**: The `value` is set as `user_content` in the `InvocationContext`, serving as the fallback first user turn for the task agent if it wasn't triggered by a tool call.
+
+## Requirements & Rules
+
+### 1. `rerun_on_resume=True` is Mandatory for Parents
+
+Any node that calls `ctx.run_node()` **must** be configured with `rerun_on_resume=True`.
+If the parent node does not have this setting, calling `ctx.run_node()` will raise a `ValueError` at runtime.
+
+### 2. Function Parameter Mapping (`node_input` vs. Dict Binding)
+
+By default, functions wrapped as nodes look up their arguments in the session state (state binding). However, the `node_input` argument passed to `ctx.run_node(..., node_input=value)` is passed directly to the node.
+
+How you receive this input depends on how you define your function:
+
+#### Pass-through `node_input` (Default)
+To receive the raw `value` directly, the function's parameter must be named exactly `node_input`.
+
+```python
+# Correct: receives the raw value passed to node_input
+def my_worker(node_input: str):
+  return f"Done: {node_input}"
+
+# Incorrect: will fail because it tries to look up 'data' in session state
+def my_worker(data: str):
+  return f"Done: {data}"
+```
+
+#### Binding Dictionary Keys to Parameters (`parameter_binding='node_input'`)
+If you pass a dictionary to `node_input` (e.g., `node_input={'foo': 'bar'}`) and want to bind its keys to individual function parameters (e.g., `def my_worker(foo: str)`), you must configure the node with `parameter_binding='node_input'`.
+
+You can configure this using the `@node` decorator with `parameter_binding='node_input'`:
+
+```python
+from google.adk.workflow import node
+
+# Decorate with parameter_binding='node_input'
+@node(parameter_binding='node_input')
+def my_worker(foo: str):
+  return f"Done: {foo}"
+
+# Call via ctx.run_node
+result = await ctx.run_node(my_worker, node_input={'foo': 'bar'}) # foo gets 'bar'
+```
+
+
+### 3. Nested Dynamic Nodes
+
+If a dynamically scheduled node *itself* calls `ctx.run_node()`, it becomes a parent and must also have `rerun_on_resume=True`.
+You should decorate the nested function with `@node(rerun_on_resume=True)` to ensure it has this property when executed:
+
+```python
+from google.adk.workflow import node
+
+@node(rerun_on_resume=True)
+async def inner_parent(ctx: Context):
+  # Calls another dynamic node internally
+  result = await ctx.run_node(some_child)
+  yield Event(output=result)
+
+# In the outer parent:
+await ctx.run_node(inner_parent)
+```
+
+
+### 4. Generator Returns
+
+In nodes that use `yield` (generators), you cannot use `return value` to produce the final output of the node due to Python syntax constraints. You must yield `Event(output=value)` instead.
+
+## Method Signature
+
+```python
+async def run_node(
+    self,
+    node: NodeLike,
+    node_input: Any = None,
+    *,
+    use_as_output: bool = False,
+    run_id: str | None = None,
+    use_sub_branch: bool = False,
+    override_branch: str | None = None,
+) -> Any:
+```
+
+### Parameters
+
+| Parameter | Type | Default | Description |
+| :--- | :--- | :--- | :--- |
+| `node` | `NodeLike` | *Required* | The node to execute (Function, Agent, or Workflow). |
+| `node_input` | `Any` | `None` | Input data to pass to the dynamic node. |
+| `use_as_output` | `bool` | `False` | If `True`, the child node's output is used as the calling parent node's output. The parent's own output event is suppressed. Can only be set once per parent execution. |
+| `run_id` | `str \| None` | `None` | Optional custom run ID. If provided, **must contain non-numeric characters** (e.g., `"run_a"`) to prevent collision with auto-generated IDs. |
+| `use_sub_branch` | `bool` | `False` | If `True`, executes the node in a sub-branch (appending `node_name@run_id` to the branch path). Essential for parallel runs to isolate events. |
+| `override_branch` | `str \| None` | `None` | Explicitly overrides the branch name for the execution context. |
+
+## Advanced Applications
+
+### Dynamic Fan-Out (Parallel Execution)
+
+You can perform dynamic fan-out by scheduling multiple tasks in parallel using `asyncio.gather`. When doing this, you **must** set `use_sub_branch=True` to isolate the events of each parallel execution.
+
+```python
+import asyncio
+from google.adk import Context, Event, Agent
+from google.adk.workflow import node
+
+worker = Agent(name="worker", instruction="Process {node_input}")
+
+@node(rerun_on_resume=True)
+async def parallel_orchestrator(ctx: Context, node_input: list[str]):
+  tasks = []
+  for topic in node_input:
+    tasks.append(
+        ctx.run_node(
+            worker,
+            node_input=topic,
+            use_sub_branch=True, # Critical for parallel isolation
+        )
+    )
+  
+  # Await all tasks concurrently
+  results = await asyncio.gather(*tasks)
+  yield Event(output=results)
+```
+
+## Best Practices
+
+- **Avoid Unsupervised Tasks**: Always `await` `ctx.run_node()` directly (or via `asyncio.gather`). Do **not** wrap it in `asyncio.create_task()` without awaiting it, as errors will be swallowed, and tasks won't be cancelled if the workflow is interrupted.
+- **Manage Side Effects and Resumption**: Because a parent node with `rerun_on_resume=True` is executed from the beginning on resumption, any code with side effects (e.g., database writes, API calls) in the parent node will run again.
+    - *Best Practice*: Keep the parent orchestrator node's logic as light as possible, containing mostly control flow and `ctx.run_node` calls.
+    - *Best Practice*: Move any logic with side effects into dedicated child nodes and execute them via `ctx.run_node`. Since completed child nodes are cached and replayed, their side effects will *not* be executed again on resumption.
+
+
+## Limitations
+
+- **Replay Overhead**: Because the parent node is re-run from the beginning on resume, long-running parent node logic (outside of `ctx.run_node` calls) will be re-executed. Keep the orchestrator node logic light and delegate heavy lifting to child nodes.
+
+## Related samples
+
+- [Dynamic Nodes Sample](../../../../contributing/samples/workflows/dynamic_nodes/)
+- [Dynamic Fan-Out / Fan-In Sample](../../../../contributing/samples/workflows/dynamic_fan_out_fan_in/)

docs/guides/workflow/function_node/index.md

@@ -0,0 +1,165 @@
+# Function Nodes
+
+In ADK, any standard Python function, coroutine, or generator can be used as a workflow node. The framework automatically wraps these callables under the hood, allowing you to build complex graphs with minimal boilerplate.
+
+## Introduction
+
+Function nodes are the most common and lightweight way to implement logic in ADK workflows. Instead of subclassing `BaseNode` for every step, you can write standard Python functions.
+
+Developer problems solved:
+- **Zero Boilerplate**: Write standard Python code without framework-specific class definitions.
+- **Implicit Wrapping**: Pass functions directly to workflow edges; the framework handles integration automatically.
+- **Declarative Signatures**: Access workflow state, input from predecessor nodes, or the execution context simply by declaring them in the function parameters.
+
+## Get started
+
+The following example demonstrates how to define standard Python functions and use them directly in a workflow chain.
+
+```python
+from google.adk import START, Workflow
+
+# 1. Simple sequential steps
+# The output of step_one is automatically passed as input to step_two
+def step_one(node_input: str) -> str:
+    return f"{node_input} -> step_one"
+
+def step_two(node_input: str) -> str:
+    return f"{node_input} -> step_two"
+
+# 2. Step that accesses workflow state
+# user_name is automatically resolved from ctx.state["user_name"]
+def step_three(node_input: str, user_name: str) -> str:
+    return f"Hello {user_name}! {node_input}"
+
+# Use the functions directly in the workflow edges
+workflow = Workflow(
+    name="my_workflow",
+    edges=[
+        (START, step_one, step_two, step_three),
+    ],
+)
+```
+
+## How it works
+
+When a workflow executes a function node, it performs several operations automatically:
+
+### Parameter Resolution
+The framework inspects the function signature to determine how to populate its arguments:
+*   **`ctx`** (or any parameter type-hinted as `Context`): Injects the workflow `Context` object.
+*   **`node_input`**: Injects the output value from the predecessor node.
+*   **Any other parameter**: Resolved by looking up the parameter name in `ctx.state` (or `node_input` if parameter binding is customized).
+
+### Type Coercion
+Input values are automatically validated and coerced to match the function's type hints using Pydantic:
+*   **Pydantic Models**: If a parameter is type-hinted as a Pydantic `BaseModel` (e.g., `node_input: MyModel`) and the input is a dictionary, it is auto-converted to the model instance.
+*   **Content to String**: If a parameter expects a `str` but receives a `types.Content` object (e.g. the raw user message from `START`), it automatically extracts and concatenates the text parts.
+
+### Event Normalization
+Return and yield values are normalized to `Event` objects:
+*   Returning or yielding `None` does not emit an output event, but execution continues downstream with `None` passed as the input to successor nodes.
+*   Raw values (strings, dicts, etc.) are wrapped in `Event(output=value)`.
+*   Pydantic models are serialized to dictionaries.
+*   State changes made via `ctx.state` during execution are automatically captured and attached to the event to be persisted.
+
+## Configuration & Explicit Wrapping
+
+While implicit wrapping works for most cases, you can wrap functions explicitly using the `FunctionNode` class or the `@node` decorator when you need to configure execution behavior.
+
+Use explicit configuration when you need to define:
+*   `rerun_on_resume`: Control if the node should rerun when the workflow resumes (default is `False` for function nodes, meaning they complete with the resuming input).
+*   `retry_config`: Enable retries on failures.
+*   `timeout`: Set a maximum execution time.
+*   `auth_config`: Gate execution with user authentication.
+
+### Using `@node` Decorator
+
+```python
+from google.adk.workflow import node
+
+@node(rerun_on_resume=True)
+def process_payment(node_input: dict) -> str:
+    # This node will rerun if the workflow is resumed after a pause
+    ...
+```
+
+### Using `FunctionNode` Class
+
+```python
+from google.adk.workflow import FunctionNode, RetryConfig
+
+def my_func(node_input: str) -> str:
+    ...
+
+# Wrap explicitly to configure retries
+custom_node = FunctionNode(
+    my_func,
+    name="payment_step",
+    retry_config=RetryConfig(max_attempts=3),
+)
+```
+
+## Advanced applications
+
+### Emitting Message Events for Web UI
+Only the `Event.message` (user-facing content) is rendered in the Web UI, while `Event.output` is internal and passed downstream. For terminal nodes or nodes producing user-visible intermediate results, yield both a message event and an output event:
+
+```python
+from google.adk.events.event import Event
+
+async def summarize(ctx: Context, node_input: str):
+    result = f"Summary: {node_input}"
+    # Rendered in UI (message accepts a raw string and auto-wraps it)
+    yield Event(message=result)
+    # Passed to downstream nodes
+    yield Event(output=result)
+```
+
+### State Integration
+
+You can update the shared workflow state in two ways: by mutating `ctx.state` directly, or by yielding/returning an `Event(state=...)`.
+
+#### 1. Mutating `ctx.state` directly (Imperative)
+This is the most common way when your function already accesses the context. Mutations are tracked and automatically persisted by the framework when the node finishes execution.
+
+```python
+def update_via_context(ctx: Context, node_input: str) -> str:
+    # State is updated immediately in memory
+    ctx.state["counter"] = ctx.state.get("counter", 0) + 1
+    return node_input
+```
+
+#### 2. Yielding/Returning `Event(state=...)` (Declarative)
+This is useful if you want to declare state changes as events, or if your function does not need the `ctx` object otherwise.
+
+```python
+from google.adk.events.event import Event
+
+def update_via_event(node_input: str):
+    # Returns the state change without needing 'ctx' in the signature
+    return Event(
+        output=node_input,
+        state={"last_processed": node_input}
+    )
+```
+
+#### Key Differences
+
+| Feature | Mutating `ctx.state` | Yielding `Event(state=...)` |
+| :--- | :--- | :--- |
+| **Visibility** | Changes are visible **immediately** to subsequent lines in the same function. | Changes are only visible **after** the event is yielded and processed by the framework. |
+| **Signature** | Requires `ctx: Context` in the function parameters. | Can be used in any function (no `ctx` required). |
+| **Style** | Imperative state modification. | Declarative event-driven state update. |
+
+## Limitations
+
+- **Union Type Hints**: If `node_input` is hinted with a `Union` type (e.g. `str | dict`), the framework skips automatic type validation to avoid false positives. You must perform manual `isinstance` checks in the function body if you need to validate the input type.
+
+## Related samples
+
+The following samples demonstrate function node usage:
+- [Node Output](../../../../contributing/samples/workflows/node_output/agent.py) - Auto type conversion to Pydantic models.
+- [Route](../../../../contributing/samples/workflows/route/agent.py) - Yielding events with routes.
+- [State](../../../../contributing/samples/workflows/state/agent.py) - Interacting with workflow state.
+- [Auth API Key](../../../../contributing/samples/workflows/auth_api_key/agent.py) - Using authentication.
+- [Request Input Advanced](../../../../contributing/samples/workflows/request_input_advanced/agent.py) - Human-in-the-loop with schemas.