code: def test_policies_are_restored_on_recovered_worker(self):
      # Counter that will survive restarts.
      COUNTER_NAME = "test_policies_are_restored_on_recovered_worker"
      counter = Counter.options(name=COUNTER_NAME).remote()
  
      config = (
          PGConfig()
          .rollouts(
              env_runner_cls=ForwardHealthCheckToEnvWorker,
              num_rollout_workers=2,
              rollout_fragment_length=16,
          )
          .training(
              train_batch_size=32,
              model={"fcnet_hiddens": [4]},
          )
          .environment(
              env="multi_agent_fault_env",
              env_config={
                  # Make both worker idx=1 and 2 fail.
                  "bad_indices": [1, 2],
                  "failure_start_count": 3,
                  "failure_stop_count": 4,
                  "counter": COUNTER_NAME,
              },
          )
          .evaluation(
              evaluation_num_workers=1,
              evaluation_interval=1,
              evaluation_config=PGConfig.overrides(
                  recreate_failed_workers=True,
                  # Restart the entire eval worker.
                  restart_failed_sub_environments=False,
                  env_config={
                      "evaluation": True,
                      # Make eval worker (index 1) fail.
                      "bad_indices": [1],
                      "failure_start_count": 3,
                      "failure_stop_count": 4,
                      "counter": COUNTER_NAME,
                  },
              ),
          )
          .callbacks(AddPolicyCallback)
          .fault_tolerance(
              recreate_failed_workers=True,  # But recover.
              # Throwing error in constructor is a bad idea.
              # 0 delay for testing purposes.
              delay_between_worker_restarts_s=0,
          )
      )
  
      for _ in framework_iterator(config, frameworks=("tf2", "torch")):
          # Reset interaction counter.
          ray.wait([counter.reset.remote()])
  
          a = config.build()
  
          # Should have the custom policy.
          self.assertIsNotNone(a.get_policy("test_policy"))
  
          # Before train loop, workers are fresh and not recreated.
          self.assertEqual(a.workers.num_healthy_remote_workers(), 2)
          self.assertEqual(a.workers.num_remote_worker_restarts(), 0)
          self.assertEqual(
              a.evaluation_workers.num_healthy_remote_workers(), 1
          )
          self.assertEqual(
              a.evaluation_workers.num_remote_worker_restarts(), 0
          )
  
          a.train()
          wait_for_restore()
          # One more iteration. Workers will be recovered during this round.
          a.train()
  
          # Everything still healthy. And all workers are restarted.
          self.assertEqual(a.workers.num_healthy_remote_workers(), 2)
          self.assertEqual(a.workers.num_remote_worker_restarts(), 2)
          self.assertEqual(
              a.evaluation_workers.num_healthy_remote_workers(), 1
          )
          self.assertEqual(
              a.evaluation_workers.num_remote_worker_restarts(), 1
          )
  
          # Let's verify that our custom policy exists on both recovered workers.
          def has_test_policy(w):
              return "test_policy" in w.policy_map
  
          # Rollout worker has test policy.
          self.assertTrue(
              all(
                  a.workers.foreach_worker(
                      has_test_policy, local_worker=False
                  )
              )
          )
          # Eval worker has test policy.
          self.assertTrue(
              all(
                  a.evaluation_workers.foreach_worker(
                      has_test_policy, local_worker=False
                  )
              )
          )
  
imports: from collections import defaultdict
  import gymnasium
  import numpy
  import time
  import unittest
  import ray
  from ray.util.state import list_actors
  from ray.rllib.algorithms.algorithm_config import AlgorithmConfig
  from ray.rllib.algorithms.callbacks import DefaultCallbacks
  from ray.rllib.algorithms.dqn.dqn import DQNConfig
  from ray.rllib.algorithms.impala import ImpalaConfig
  from ray.rllib.algorithms.pg import PGConfig
  from ray.rllib.algorithms.pg.pg_tf_policy import PGTF2Policy
  from ray.rllib.algorithms.pg.pg_torch_policy import PGTorchPolicy
  from ray.rllib.algorithms.ppo import PPOConfig
  from ray.rllib.env.multi_agent_env import make_multi_agent
  from ray.rllib.evaluation.rollout_worker import RolloutWorker
  from ray.rllib.examples.env.random_env import RandomEnv
  from ray.rllib.utils.test_utils import framework_iterator
  from ray.tune.registry import register_env
label_desc: Use image node_id 16 for calling the TestWorkerFailures obj's underlying member method code with example usage: obj.test_policies_are_restored_on_recovered_worker() and returns: unknown