refactor: Split Context.run_node to support internal orchestration

This PR extracts the core execution logic from the public `Context.run_node` method into a new internal method `_run_node_internal`. This is the first step in unifying static and dynamic node execution paths.

By splitting these methods, we maintain a clean public API while providing the orchestration engines (such as the workflow engine and dynamic scheduler) with the necessary hooks to drive and inspect node execution.

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

Author: DeanChensj

src/google/adk/agents/context.py

@@ -452,10 +452,47 @@ async def run_node(
       use_sub_branch: If True, the dynamic node will be executed in a sub-branch
         to isolate its state and events from the main branch.
       override_branch: An optional branch to use instead of parent's branch.
+      override_isolation_scope: An optional isolation scope to use instead of
+        the parent's scope.
+      raise_on_wait: If True, raises NodeInterruptedError when the child node
+        is WAITING instead of returning None.
 
     Returns:
       The output of the dynamically executed node, once it finishes executing.
     """
+    return await self._run_node_internal(
+        node,
+        node_input,
+        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,
+        raise_on_wait=raise_on_wait,
+        resume_inputs=None,
+        return_ctx=False,
+    )
+
+  async def _run_node_internal(
+      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,
+      override_isolation_scope: str | None = None,
+      raise_on_wait: bool = False,
+      return_ctx: bool = False,
+      resume_inputs: dict[str, Any] | None = None,
+  ) -> Any:
+    """Executes a node dynamically (Internal Orchestration API).
+
+    See public ``run_node`` for public argument details.
+    Additional internal args:
+      return_ctx: If True, returns the child's Context instead of its output.
+    """
 
     if not self._node_rerun_on_resume:
       raise ValueError(
@@ -538,7 +575,7 @@ async def run_node(
           and not child_ctx.actions.transfer_to_agent
       ):
         raise NodeInterruptedError()
-      return child_ctx.output
+      return child_ctx if return_ctx else child_ctx.output
 
     # Mode 2: Standalone execution (outside of workflow).
     # Run the node directly via NodeRunner.
@@ -550,6 +587,7 @@ async def run_node(
         override_branch=override_branch,
         override_isolation_scope=override_isolation_scope,
         run_id=run_id,
+        resume_inputs=resume_inputs,
     )
     if result.error:
       from ..workflow import _errors
@@ -568,7 +606,7 @@ async def run_node(
       from ..workflow._errors import NodeInterruptedError
 
       raise NodeInterruptedError()
-    return result.output
+    return result if return_ctx else result.output
 
   # ============================================================================
   # Artifact methods

tests/unittests/agents/test_context.py

@@ -673,3 +673,55 @@ def test_get_invocation_context_propagates_isolation_scope(
         }
     )
     assert result is mock_copy
+
+
+class TestContextRunNodeInternal:
+  """Tests for the internal Context._run_node_internal orchestration method."""
+
+  @pytest.mark.asyncio
+  async def test_run_node_internal_returns_ctx_and_handles_resume_inputs(
+      self, mock_invocation_context, mocker
+  ):
+    """Test that _run_node_internal correctly handles return_ctx and resume_inputs."""
+    # Arrange
+    from google.adk.agents.llm_agent import LlmAgent
+    from google.adk.events.event_actions import EventActions
+
+    agent_a = LlmAgent(name="agent_a", rerun_on_resume=True)
+    root = LlmAgent(name="root", sub_agents=[agent_a], rerun_on_resume=True)
+    agent_a.parent_agent = root
+
+    root_ctx = Context(mock_invocation_context, node=root, run_id="1")
+
+    child_ctx_a = Context(
+        mock_invocation_context,
+        parent_ctx=root_ctx,
+        node=agent_a,
+        run_id="1",
+        event_actions=EventActions(),
+    )
+    child_ctx_a.output = "a_output"
+
+    # Mock the standalone execution boundary
+    mock_run_standalone = mocker.patch.object(
+        Context,
+        "_run_node_standalone",
+        return_value=child_ctx_a,
+    )
+
+    # Act 1: Call _run_node_internal with return_ctx=True
+    result_ctx = await root_ctx._run_node_internal(
+        agent_a,
+        node_input="a_input",
+        return_ctx=True,
+        resume_inputs={"some_key": "some_val"},
+    )
+
+    # Assert 1: It returns the child context object itself, not the output!
+    assert result_ctx is child_ctx_a
+    assert result_ctx.output == "a_output"
+
+    # Assert 2: resume_inputs was correctly passed to _run_node_standalone
+    mock_run_standalone.assert_called_once()
+    _, kwargs = mock_run_standalone.call_args
+    assert kwargs.get("resume_inputs") == {"some_key": "some_val"}