code: def test_recovery_simple_1(workflow_start_regular):
      utils.unset_global_mark()
      workflow_id = "test_recovery_simple_1"
      with pytest.raises(workflow.WorkflowExecutionError):
          # internally we get WorkerCrashedError
          workflow.run(
              the_failed_task.bind("x"), workflow_id=workflow_id
          )
  
      assert (
          workflow.get_status(workflow_id)
          == workflow.WorkflowStatus.FAILED
      )
  
      utils.set_global_mark()
      assert workflow.resume(workflow_id) == "foo(x)"
      utils.unset_global_mark()
      # resume from workflow output checkpoint
      assert workflow.resume(workflow_id) == "foo(x)"
  
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 7 for calling a global function with example usage: test_recovery_simple_1(workflow_start_regular) without return types