code: def test_checkpoints_saved_after_resume(
      ray_start_2_cpus, tmp_path, trainable_type
  ):
      """Checkpoints saved after experiment restore should pick up at the correct
      iteration and should not overwrite the checkpoints from the original run.
      Old checkpoints should still be deleted if the total number of checkpoints
      (old + new) exceeds `num_to_keep`.
  
      In this test, `num_to_keep=4`:
      - Initial run saves checkpoint_000000 and checkpoint_000001
      - Restored run saves checkpoint_000002, checkpoint_000003, and checkpoint_000004
      - Checkpoint 000000 should be deleted.
      """
  
      def get_checkpoints(experiment_dir):
          checkpoint_dirs = [
              path
              for path in os.listdir(experiment_dir)
              if path.startswith("checkpoint_")
          ]
          sorted_checkpoint_dirs = sorted(checkpoint_dirs)
          checkpoints = [
              Checkpoint.from_directory(os.path.join(experiment_dir, d))
              for d in sorted_checkpoint_dirs
          ]
          return sorted_checkpoint_dirs, checkpoints
  
      fail_marker = tmp_path / "fail_marker"
      fail_marker.write_text("", encoding="utf-8")
  
      num_to_keep = 4
      checkpoint_config = CheckpointConfig(num_to_keep=num_to_keep)
      param_space = {
          "failing_hanging": (fail_marker, None),
          "num_epochs": 2,
      }
  
      if trainable_type == "function":
          trainable = _train_fn_sometimes_failing
      elif trainable_type == "class":
          trainable = _ClassTrainableSometimesFailing
          checkpoint_config.checkpoint_frequency = 1
          param_space["num_epochs"] = 4
          param_space["fail_epochs"] = 2
      elif trainable_type == "data_parallel":
          trainable = DataParallelTrainer(
              _train_fn_sometimes_failing,
              scaling_config=ScalingConfig(num_workers=1),
          )
          param_space = {"train_loop_config": param_space}
      else:
          raise ValueError(f"Invalid trainable type: {trainable_type}")
  
      tuner = Tuner(
          trainable,
          tune_config=TuneConfig(num_samples=1),
          run_config=RunConfig(
              name="exp_name",
              storage_path=str(tmp_path),
              checkpoint_config=checkpoint_config,
          ),
          param_space=param_space,
      )
      results = tuner.fit()
      training_iteration = results[0].metrics["training_iteration"]
      assert (
          training_iteration == 2
      ), f"Should be at 2 iters before erroring, got {training_iteration}"
  
      # Initial run saves the first 2 checkpoints
      checkpoint_dirs, checkpoints = get_checkpoints(results[0].path)
      assert checkpoint_dirs == [
          "checkpoint_000000",
          "checkpoint_000001",
      ]
      assert [
          load_dict_checkpoint(ckpt)["it"] for ckpt in checkpoints
      ] == [1, 2]
  
      fail_marker.unlink()
      tuner = Tuner.restore(
          str(tmp_path / "exp_name"),
          trainable=trainable,
          resume_errored=True,
      )
      results = tuner.fit()
  
      assert len(results.errors) == 0
      training_iteration = results[0].metrics["training_iteration"]
      # Restored at it=2, reported 3 more times -> should have it=5
      assert training_iteration == 5
  
      # Restored run saves the 3 more checkpoints, and first checkpoint should be deleted
      checkpoint_dirs, checkpoints = get_checkpoints(results[0].path)
      assert checkpoint_dirs == [
          f"checkpoint_00000{i}" for i in range(1, 5)
      ]
      assert [
          load_dict_checkpoint(ckpt)["it"] for ckpt in checkpoints
      ] == [2, 3, 4, 5]
  
imports: import json
  import logging
  import os
  import shutil
  import time
  import unittest
  import pytest
  import ray
  from ray import train, tune
  import ray.cloudpickle
  from ray.train import Checkpoint, CheckpointConfig, FailureConfig, RunConfig, ScalingConfig
  from ray.air._internal.uri_utils import URI
  from ray.train.data_parallel_trainer import DataParallelTrainer
  from ray.train._internal.storage import get_fs_and_path, _download_from_fs_path, _upload_to_fs_path
  from ray.tune import Callback, Trainable
  from ray.tune.analysis import ExperimentAnalysis
  from ray.tune.execution.experiment_state import _find_newest_experiment_checkpoint
  from ray.tune.experiment import Trial
  from ray.tune.result_grid import ResultGrid
  from ray.tune.schedulers.async_hyperband import ASHAScheduler
  from ray.tune.search.optuna import OptunaSearch
  from ray.tune.tune_config import TuneConfig
  from ray.tune.tuner import Tuner
  from ray.train.tests.util import create_dict_checkpoint, load_dict_checkpoint
label_desc: Use image node_id 23 for calling a global function with example usage: test_checkpoints_saved_after_resume(ray_start_2_cpus, tmp_path, trainable_type) and returns: sorted_checkpoint_dirs,  checkpoints