refactor(workflow): move agent transfer loop into DynamicNodeScheduler

Migrate the sequential agent transfer (handoff) loop from
_dynamic_node_executor.py into DynamicNodeScheduler.__call__.

- Extract single-node execution logic in DynamicNodeScheduler into
  _execute_step(...), covering run tracking, replay barriers, and
  execution.
- Implement the agent transfer loop in DynamicNodeScheduler.__call__,
  resolving target agents and parent contexts via
  resolve_and_derive_transfer_context. Each hop uses the scheduler that
  owns its own parent context, so run IDs keep coming from that
  context's state.
- Simplify run_node_internal in _dynamic_node_executor.py to delegate to
  ScheduleDynamicNode. When ctx._workflow_scheduler is None the executor
  uses a throwaway DynamicNodeScheduler with enable_replay=False without
  attaching it to ctx: it drives transfers but skips session event
  rehydration and defaults run_id to '1', so a path that previously ran
  without a scheduler stays a direct pass-through to NodeRunner, and
  ctx._workflow_scheduler is not None continues to strictly mean inside a
  workflow.
- Support resume_inputs through ScheduleDynamicNode and
  DynamicNodeScheduler for dynamic node resumption.
- Remove Context._child_run_counters, a run counter mirror that nothing
  read. DynamicNodeState.run_counters is now the only counter.

Behaviour change: when a transfer moves execution to a different parent
context and the target then interrupts, the interrupt IDs now land on
the calling node's Context rather than the transfer target's parent
Context. The calling node's NodeRunner is the only reader of those IDs,
so previously the caller was recorded COMPLETED while the agent was in
fact waiting for user input, and no checkpoint was emitted. Reaching
this required an upward transfer whose target interrupted; sibling
transfers and non-interrupting hops were unaffected.

- Add unit tests for transfer loop execution, interrupt halting, target
  validation, scheduler handover between parent contexts, the replay
  opt-out, and interrupt ID propagation after a transfer.

Co-authored-by: Shangjie Chen <deanchen@google.com>
PiperOrigin-RevId: 981434445

Author: DeanChensj

.agents/skills/adk-architecture/references/interface-workflow.md

@@ -141,6 +141,13 @@ When `ctx.run_node()` is called, the scheduler checks three cases:
    - All resolved → re-execute with `resume_inputs` from the
      resolved function responses.
 
+4. **Agent Transfer** — if the child node is an agent that requests an
+   agent handoff (`child_ctx.actions.transfer_to_agent`), the scheduler
+   drives the sequential transfer loop. It resolves the target agent and
+   parent context, delegates single-step execution to the target context's
+   owning scheduler, and preserves output delegation (`use_as_output`)
+   when execution returns to the invoking context.
+
 State reconstruction is **lazy**: the scheduler scans session events
 only on the first `ctx.run_node()` call for a given path, not
 upfront. This avoids scanning for dynamic nodes that won't be

src/google/adk/agents/context.py

@@ -201,10 +201,6 @@ def __init__(
     self._resume_inputs = resume_inputs or {}
     self._workflow_scheduler = _derive_scheduler(parent_ctx)
     self._node_rerun_on_resume = node.rerun_on_resume if node else True
-    # TODO: Remove. Superseded by DynamicNodeState.run_counters, which is now
-    # the authoritative sequential run_id allocator. Kept until the
-    # transfer-loop refactor lands to avoid churning that change.
-    self._child_run_counters: dict[str, int] = {}
     self._attempt_count = attempt_count
     self._output_delegated = False
     self._output_value: Any = None

src/google/adk/workflow/_dynamic_node_executor.py

@@ -17,7 +17,6 @@
 from __future__ import annotations
 
 from typing import Any
-from typing import cast
 from typing import TYPE_CHECKING
 
 from ..agents.base_agent import BaseAgent
@@ -27,12 +26,10 @@
 from ._graph import NodeLike
 from ._node_runner import NodeRunner
 from ._workflow import Workflow
-from .utils._transfer_utils import resolve_and_derive_transfer_context
 from .utils._workflow_graph_utils import build_node
 
 if TYPE_CHECKING:
   from ..agents.context import Context
-  from ._schedule_dynamic_node import ScheduleDynamicNode
 
 
 async def run_node_internal(
@@ -81,140 +78,82 @@ async def run_node_internal(
         )
       ctx._output_delegated = True
 
-  # Pointers to track the active execution state in the transfer loop.
-  # These will be updated dynamically if an agent transfers execution.
-  curr_parent_ctx = ctx
-  curr_node = built_node
-  curr_run_id = run_id
-  curr_input = node_input
-
-  # Active Execution Loop: Handles both standard execution and sequential Agent Transfers
-  # (e.g. Agent A transferring to Agent B). Instead of recursive execution, we use this
-  # loop to execute the target agent in-place, updating pointers and 'continuing' the loop.
-  while True:
-    curr_use_as_output = use_as_output if (curr_parent_ctx is ctx) else False
-    if ctx._workflow_scheduler:
-      # --- Mode 1: Workflow Execution ---
-      # The node is running as part of a Workflow graph. We must delegate execution
-      # to the workflow scheduler to handle graph dependencies and state.
-
-      # Validate the caller-supplied run_id. A None run_id is passed through
-      # unchanged: the scheduler owns sequential run_id allocation.
-      if curr_run_id and curr_run_id.isdigit() and not skip_run_id_validation:
-        raise ValueError(
-            f'Explicit run_id "{curr_run_id}" for node "{curr_node.name}"'
-            ' must contain non-numeric characters to prevent collision'
-            ' with auto-generated IDs.'
-        )
-
-      scheduler = cast(
-          'ScheduleDynamicNode', curr_parent_ctx._workflow_scheduler
-      )
-      child_ctx = await scheduler(
-          curr_parent_ctx,
-          curr_node,
-          curr_input,
-          node_name=curr_node.name,
-          use_as_output=curr_use_as_output,
-          run_id=curr_run_id,
-          use_sub_branch=use_sub_branch,
-          override_branch=override_branch,
-          override_isolation_scope=override_isolation_scope,
-      )
-    else:
-      # --- Mode 2: Standalone Execution ---
-      # The node is running independently (outside of a workflow).
-      # We run it directly using NodeRunner.
-      child_ctx = await run_node_standalone(
-          curr_parent_ctx,
-          curr_node,
-          curr_input,
-          use_as_output=curr_use_as_output,
-          use_sub_branch=use_sub_branch,
-          override_branch=override_branch,
-          override_isolation_scope=override_isolation_scope,
-          run_id=curr_run_id,
-          resume_inputs=resume_inputs,
+  # Validate the caller-supplied run_id when running inside a workflow.
+  # A None run_id is passed through unchanged: the scheduler owns sequential run_id allocation.
+  # Standalone runs (outside a workflow) allow explicit numeric IDs since there is no auto-allocation collision risk.
+  if ctx._workflow_scheduler is not None:
+    if run_id and run_id.isdigit() and not skip_run_id_validation:
+      raise ValueError(
+          f'Explicit run_id "{run_id}" for node "{built_node.name}"'
+          ' must contain non-numeric characters to prevent collision'
+          ' with auto-generated IDs.'
       )
 
-    # Extract the transfer target if the node requested an agent transfer.
-    transfer_to_agent = (
-        child_ctx.actions.transfer_to_agent if child_ctx else None
+  scheduler = ctx._workflow_scheduler
+  if scheduler is None:
+    from ._dynamic_node_scheduler import DynamicNodeScheduler
+    from ._dynamic_node_scheduler import DynamicNodeState
+
+    # No orchestrator installed one, so this call is not part of a replayable
+    # workflow. Use a transfer-only scheduler: it drives the sequential
+    # agent transfer loop but skips session event rehydration, keeping this
+    # path a direct pass-through to NodeRunner as it was before, without
+    # attaching a scheduler to ctx._workflow_scheduler.
+    #
+    # IMPORTANT: ctx._workflow_scheduler MUST remain None for standalone runs.
+    # Across ADK, `ctx._workflow_scheduler is not None` is the canonical check
+    # for whether execution is inside a workflow graph (e.g. for numeric run_id
+    # validation and replay semantics).
+    scheduler = DynamicNodeScheduler(
+        state=DynamicNodeState(), enable_replay=False
     )
 
-    # Post-Execution Validation: If the caller expects the raw output (not the Context),
-    # we check for errors or interrupts and raise them immediately.
-    if not return_ctx:
-      if child_ctx.error:
-        raise DynamicNodeFailError(
-            message=f'Dynamic node {curr_node.name} failed',
-            error=child_ctx.error,
-            error_node_path=child_ctx.error_node_path,
-        )
-      if child_ctx.interrupt_ids:
-        # Propagate child's interrupt_ids to this node's ctx
-        # so NodeRunner sees them after catching the error.
-        curr_parent_ctx._interrupt_ids.update(child_ctx.interrupt_ids)
-        raise NodeInterruptedError()
-      # When the caller passes raise_on_wait=True, surface a child
-      # execution that's WAITING (wait_for_output, no output, not transferring)
-      # as NodeInterruptedError so the parent's NodeRunner records
-      # the parent as WAITING instead of falsely COMPLETED.
-      if raise_on_wait and child_ctx.output is None and not transfer_to_agent:
-        if isinstance(curr_node, Workflow) or getattr(
-            curr_node, 'wait_for_output', False
-        ):
-          raise NodeInterruptedError()
-
-    # Handle Agent Transfer: If a transfer was requested, we resolve the target agent
-    # and its parent context, update loop pointers, and continue to the next iteration.
-    if isinstance(transfer_to_agent, str):
-      if not isinstance(curr_node, BaseAgent):
-        raise ValueError('Only agents can request an agent transfer.')
-      target_name = transfer_to_agent
-      root_agent = getattr(curr_node, 'root_agent', None)
-      if not root_agent:
-        raise ValueError(f'Cannot find root_agent on node {curr_node.name}')
-
-      target_agent, next_parent_ctx = resolve_and_derive_transfer_context(
-          target_name=target_name,
-          current_agent=curr_node,
-          root_agent=root_agent,
-          curr_ctx=child_ctx,
-          curr_parent_ctx=curr_parent_ctx,
-      )
-      if not target_agent:
-        raise ValueError(f"Transfer target agent '{target_name}' not found.")
-      if not next_parent_ctx:
-        available = []
-        if hasattr(curr_node, '_get_available_agent_names'):
-          available = curr_node._get_available_agent_names()
-        available_str = (
-            f"\nAvailable agents: {', '.join(available)}" if available else ''
-        )
-        raise ValueError(
-            f"Cannot transfer from '{curr_node.name}' to unrelated agent"
-            f" '{target_name}'.{available_str}"
-        )
-      curr_parent_ctx = next_parent_ctx
-
-      # Set up parameters for next iteration (the transfer target).
-      curr_node = target_agent
-      curr_run_id = None
-      curr_input = None  # Input for transfer target is usually empty.
-      resume_inputs = None
-
-      if not curr_parent_ctx:
-        raise AssertionError(
-            'curr_parent_ctx cannot be None during active workflow execution'
-        )
+  child_ctx = await scheduler(
+      ctx,
+      built_node,
+      node_input,
+      node_name=built_node.name,
+      use_as_output=use_as_output,
+      run_id=run_id,
+      use_sub_branch=use_sub_branch,
+      override_branch=override_branch,
+      override_isolation_scope=override_isolation_scope,
+      resume_inputs=resume_inputs,
+  )
 
-      continue
+  transfer_to_agent = child_ctx.actions.transfer_to_agent if child_ctx else None
+
+  # Post-Execution Validation: If the caller expects the raw output (not the Context),
+  # we check for errors or interrupts and raise them immediately.
+  if not return_ctx:
+    if child_ctx.error:
+      executed_name = child_ctx.node.name if child_ctx.node else built_node.name
+      raise DynamicNodeFailError(
+          message=f'Dynamic node {executed_name} failed',
+          error=child_ctx.error,
+          error_node_path=child_ctx.error_node_path,
+      )
+    if child_ctx.interrupt_ids:
+      # Propagate child's interrupt_ids to this node's ctx
+      # so NodeRunner sees them after catching the error.
+      ctx._interrupt_ids.update(child_ctx.interrupt_ids)
+      raise NodeInterruptedError()
+    # When the caller passes raise_on_wait=True, surface a child
+    # execution that's WAITING (wait_for_output, no output, not transferring)
+    # as NodeInterruptedError so the parent's NodeRunner records
+    # the parent as WAITING instead of falsely COMPLETED.
+    if raise_on_wait and child_ctx.output is None and not transfer_to_agent:
+      # After a transfer chain, child_ctx belongs to the last agent that ran,
+      # not to built_node, so the wait decision must follow child_ctx.node.
+      executed_node = child_ctx.node
+      if isinstance(executed_node, Workflow) or getattr(
+          executed_node, 'wait_for_output', False
+      ):
+        raise NodeInterruptedError()
 
-    # If no transfer occurred, execution of the branch is complete.
-    if return_ctx:
-      return child_ctx
-    return child_ctx.output
+  if return_ctx:
+    return child_ctx
+  return child_ctx.output
 
 
 async def run_node_standalone(