code: def result_collection_reload(
      default_root_dir, accelerator="auto", devices=1, **kwargs
  ):
      class CustomException(Exception):
          pass
  
      batches = 5
  
      class ExtendedBoringModel(BoringModel):
          def __init__(self):
              super().__init__()
              self.breaking_batch_idx = 3
              self.has_validated_sum = False
              self.dummy_metric = DummyMeanMetric()
  
          @property
          def results(self):
              return self.trainer.fit_loop._results
  
          def training_step(self, batch, batch_idx):
              # We run 5 batches, meaning batch_idx from [0..4]
              # Without failure, we expect to get `total=sum(range(5))` and `num_batches=5`
              # When not restarting, it simulates a failure on `batch_idx=3` and test the state after reload
              # Compute `on_epoch_end` would be `10/5=2` if the metric state had been serialized and reloaded
              if self.trainer.fit_loop.restarting:
                  self.log(
                      "tracking", batch_idx, on_step=True, on_epoch=True
                  )
                  self.log(
                      "tracking_2",
                      batch_idx,
                      on_step=True,
                      on_epoch=True,
                      sync_dist=True,
                  )
  
                  self.dummy_metric(batch_idx)
                  self.log(
                      "tracking_metric",
                      self.dummy_metric,
                      on_step=True,
                      on_epoch=True,
                  )
  
                  value = self.results["training_step.tracking_metric"]
                  value_2 = self.results["training_step.tracking"]
  
                  # On failure, the Metric states are being accumulated on rank 0 and zeroed-out on other ranks.
                  # The shift indicates we failed while the state was `shift=sign(is_global_zero > 0) * [0..3]`
                  shift = 0
                  if devices == 2:
                      shift = 3 if self.trainer.is_global_zero else -3
                  expected = sum(range(batch_idx + 1)) + shift
                  assert expected == value == value_2
              else:
                  if batch_idx == self.breaking_batch_idx:
                      # simulate failure mid epoch
                      raise CustomException
  
                  self.log(
                      "tracking", batch_idx, on_step=True, on_epoch=True
                  )
                  self.log(
                      "tracking_2",
                      batch_idx,
                      on_step=True,
                      on_epoch=True,
                      sync_dist=True,
                  )
  
                  self.dummy_metric(batch_idx)
                  self.log(
                      "tracking_metric",
                      self.dummy_metric,
                      on_step=True,
                      on_epoch=True,
                  )
  
                  value = self.results["training_step.tracking"]
                  assert value == sum(range(batch_idx + 1))
  
                  value = self.results["training_step.tracking_2"]
                  assert value == sum(range(batch_idx + 1))
  
              return super().training_step(batch, batch_idx)
  
          def on_train_epoch_end(self) -> None:
              if self.trainer.fit_loop.restarting:
                  # the state of the results before the exception is not saved and restored, so the total starts after
                  # the breaking_batch_idx
                  total = sum(range(self.breaking_batch_idx, batches))
                  metrics = self.results.metrics(on_step=False)
                  computed_value = self.dummy_metric.compute()
  
                  assert (
                      self.results["training_step.tracking"].value
                      == total
                  )
                  expected = total / (batches - self.breaking_batch_idx)
                  assert metrics["callback"]["tracking"] == expected
                  assert computed_value == 2
  
                  assert (
                      self.results["training_step.tracking_2"].value
                      == total
                  )
                  assert metrics["callback"]["tracking_2"] == expected
                  assert computed_value == 2
                  self.has_validated_sum = True
  
      model = ExtendedBoringModel()
      trainer_kwargs = {
          "max_epochs": 1,
          "limit_train_batches": batches,
          "limit_val_batches": 0,
          "accelerator": accelerator,
          "devices": devices,
          "enable_progress_bar": False,
          "enable_model_summary": False,
          "default_root_dir": default_root_dir,
          "callbacks": OnExceptionCheckpoint(default_root_dir),
      }
      trainer_kwargs.update(kwargs)
      trainer = Trainer(**trainer_kwargs)
  
      with suppress(CustomException):
          trainer.fit(model)
      assert not model.has_validated_sum
  
      tmpdir = (
          trainer.strategy.broadcast(
              trainer_kwargs["default_root_dir"], 0
          )
          if devices >= 2
          else trainer_kwargs["default_root_dir"]
      )
      ckpt_path = os.path.join(tmpdir, "on_exception.ckpt")
  
      trainer = Trainer(**trainer_kwargs)
      trainer.fit(model, ckpt_path=ckpt_path)
      assert model.has_validated_sum
  
imports: import os
  import pickle
  from contextlib import nullcontext, suppress
  from unittest import mock
  import lightning.pytorch
  import pytest
  import torch
  from lightning.fabric.utilities.warnings import PossibleUserWarning
  from lightning.pytorch import Trainer
  from lightning.pytorch.callbacks import OnExceptionCheckpoint
  from lightning.pytorch.demos.boring_classes import BoringModel
  from lightning.pytorch.trainer.connectors.logger_connector.result import _Metadata, _ResultCollection, _ResultMetric, _Sync
  from lightning.pytorch.utilities.imports import _TORCHMETRICS_GREATER_EQUAL_0_11
  from lightning_utilities.test.warning import no_warning_call
  from torch import Tensor, tensor
  from torch.nn import ModuleDict, ModuleList
  from torchmetrics import Metric, MetricCollection
  from torchmetrics.classification import Accuracy
  from tests_pytorch.core.test_results import spawn_launch
  from tests_pytorch.helpers.runif import RunIf
label_desc: Use image node_id 7 for calling a global function with example usage: result_collection_reload(default_root_dir, accelerator, devices) and returns: self, super