code: def test_caller_task_reconstruction(ray_start_regular):
      """Test a retried task from a dead worker can be correctly processed
      by the receiving actor."""
  
      @ray.remote(max_retries=5)
      def RetryableTask(actor):
          value = ray.get(actor.increase.remote())
          if value > 2:
              return value
          else:
              os._exit(0)
  
      @ray.remote(max_restarts=1)
      class Actor:
          """An actor that will be restarted at most once."""
  
          def __init__(self):
              self.value = 0
  
          def increase(self):
              self.value += 1
              return self.value
  
      remote_actor = Actor.remote()
  
      assert ray.get(RetryableTask.remote(remote_actor)) == 3
  
imports: import atexit
  import asyncio
  import collections
  import numpy
  import os
  import pytest
  import signal
  import sys
  import time
  import ray
  from ray.actor import exit_actor
  import ray.cluster_utils
  from ray._private.test_utils import wait_for_condition, wait_for_pid_to_exit, generate_system_config_map, SignalActor
label_desc: Use image node_id 8 for calling a global function with example usage: test_caller_task_reconstruction(ray_start_regular) and returns: value, self