code: def test_nested_workflow_no_download(workflow_start_regular):
      """Test that we _only_ load from storage on recovery. For a nested workflow
      task, we should checkpoint the input/output, but continue to reuse the
      in-memory value.
      """
  
      @ray.remote
      def recursive(ref, count):
          if count == 0:
              return ref
          return workflow.continuation(recursive.bind(ref, count - 1))
  
      with tempfile.TemporaryDirectory() as temp_dir:
          debug_store = DebugStorage(temp_dir)
          utils._alter_storage(debug_store)
  
          ref = ray.put("hello")
          result = workflow.run(recursive.bind([ref], 10))
  
          ops = debug_store._logged_storage.get_op_counter()
          get_objects_count = 0
          for key in ops["get"]:
              if "objects" in key:
                  get_objects_count += 1
          assert (
              get_objects_count == 1
          ), "We should only get once when resuming."
          put_objects_count = 0
          for key in ops["put"]:
              if "objects" in key:
                  print(key)
                  put_objects_count += 1
          assert (
              put_objects_count == 1
          ), "We should detect the object exists before uploading"
          assert ray.get(result) == ["hello"]
  
imports: import subprocess
  import tempfile
  import time
  from ray.tests.conftest import *
  import pytest
  from filelock import FileLock
  import ray
  from ray._private.test_utils import run_string_as_driver_nonblocking
  from ray import workflow
  from ray.workflow import workflow_storage
  from ray.workflow.storage.debug import DebugStorage
  from ray.workflow.tests import utils
  from ray.workflow.exceptions import WorkflowNotResumableError
label_desc: Use image node_id 5 for calling a global function with example usage: test_nested_workflow_no_download(workflow_start_regular) and returns: workflow, ref