Method _start_node_task

src/google/adk/workflow/_workflow.py:538–627

Create NodeRunner and start asyncio task for a node.

(
      self,
      loop_state: _LoopState,
      ctx: Context,
      node_name: str,
      trigger: Trigger,
  )

    node_state.status = NodeStatus.RUNNING

  def _start_node_task(
      self,
      loop_state: _LoopState,
      ctx: Context,
      node_name: str,
      trigger: Trigger,
  ) -> None:
    """Create NodeRunner and start asyncio task for a node."""

    assert self.graph is not None

    node = self._get_static_node_by_name(node_name)
    is_terminal = node_name in self.graph._terminal_node_names

    node_state = loop_state.nodes[node_name]
    # Reuse run_id on resume; assign a new sequential id for fresh runs.
    run_id = node_state.run_id
    if not run_id:
      run_id = self._next_run_id(node_state)
    node_state.run_id = run_id

    # Intercept execution based on historical session events.
    key = f'{node_name}@{run_id}'
    if key in loop_state.recovered_executions:
      recovered = loop_state.recovered_executions[key]

      result = check_interception(
          node_path=f'{ctx.node_path}/{node_name}@{run_id}',
          node=node,
          recovered=recovered,
          curr_parent_ctx=ctx,
      )

      if not result.should_run:
        is_terminal = node_name in self.graph._terminal_node_names
        ancestor_path = ctx.node_path if is_terminal else None

        if ancestor_path:
          ancestors = [ancestor_path] + list(ctx._output_for_ancestors or [])
        else:
          ancestors = list(ctx._output_for_ancestors or [])

        mock_ctx = create_mock_context(
            parent_ctx=ctx,
            node=node,
            run_id=run_id,
            result=result,
            ancestors=ancestors,
            branch=recovered.branch,
        )

        async def return_ctx():
          if loop_state.sequence_barrier:
            await loop_state.sequence_barrier.wait(key)
          return mock_ctx

        loop_state.pending_tasks[node_name] = asyncio.create_task(return_ctx())
        return