code: def test_get_output_4(workflow_start_regular, tmp_path):
      """Test getting output of a workflow tasks that are dynamically generated."""
      lock_path = str(tmp_path / "lock")
      lock = FileLock(lock_path)
  
      @ray.remote
      def recursive(n):
          if n <= 0:
              with FileLock(lock_path):
                  return 42
          return workflow.continuation(
              recursive.options(
                  **workflow.options(task_id=str(n - 1))
              ).bind(n - 1)
          )
  
      workflow_id = "test_get_output_4"
      lock.acquire()
      obj = workflow.run_async(
          recursive.options(**workflow.options(task_id="10")).bind(10),
          workflow_id=workflow_id,
      )
  
      outputs = [
          workflow.get_output_async(workflow_id, task_id=str(i))
          for i in range(11)
      ]
      outputs.append(obj)
  
      import time
  
      # wait so that 'get_output' is scheduled before executing the workflow
      time.sleep(3)
      lock.release()
      assert ray.get(outputs) == [42] * len(outputs)
  
file_path: ray/python/ray/workflow/tests/test_basic_workflows_2.py
imports: import pytest
  import ray
  from filelock import FileLock
  from ray._private.test_utils import SignalActor
  from ray import workflow
  from ray.tests.conftest import *
  from ray._private.test_utils import skip_flaky_core_test_premerge
label_desc: Use image node_id 5 for calling a global function with example usage: test_get_output_4(workflow_start_regular, tmp_path) and returns: workflow, int