code: def test_dynamic_output(workflow_start_regular_shared):
      @ray.remote
      def exponential_fail(k, n):
          if n > 0:
              if n < 3:
                  raise Exception("Failed intentionally")
              return workflow.continuation(
                  exponential_fail.options(
                      **workflow.options(task_id=f"task_{n}")
                  ).bind(k * 2, n - 1)
              )
          return k
  
      # When workflow fails, the dynamic output should points to the
      # latest successful task.
      try:
          workflow.run(
              exponential_fail.options(
                  **workflow.options(task_id="task_0")
              ).bind(3, 10),
              workflow_id="dynamic_output",
          )
      except Exception:
          pass
      from ray.workflow.workflow_storage import get_workflow_storage
  
      from ray._private.client_mode_hook import client_mode_wrap
  
      @client_mode_wrap
      def _check_storage():
          wf_storage = get_workflow_storage(
              workflow_id="dynamic_output"
          )
          result = wf_storage.inspect_task("task_0")
          return result.output_task_id
  
      assert _check_storage() == "task_3"
  
file_path: ray/python/ray/workflow/tests/test_basic_workflows.py
imports: import time
  from ray.tests.conftest import *
  from filelock import FileLock
  import pytest
  import ray
  from ray import workflow
label_desc: Use image node_id 5 for calling a global function with example usage: test_dynamic_output(workflow_start_regular_shared) and returns: k, result, workflow