# Update worker group state with the new workers.
        new_workers_by_rank = {
            w.distributed_context.world_rank: w for w in new_workers
        }
        updated_workers = [
            new_workers_by_rank.get(w.distributed_context.world_rank, w)
            for w in workers
        ]
        self._worker_group_state = WorkerGroupState(
            start_time=self._worker_group_state.start_time,
            placement_group=pg,
            workers=updated_workers,
            sync_actor=sync_actor,
        )

        # Re-initialize backend (per-group TCPStore + init_process_group)
        # via BackendSetupCallback
        from ray.train.v2._internal.callbacks.backend_setup import BackendSetupCallback

        for callback in self._callbacks:
            if isinstance(callback, BackendSetupCallback):
                callback.reinitialize_workers(self, target_group.world_ranks)
                break

        # Get train context args from callbacks
        train_context_args = {}
        for cb in self._callbacks:
            args = cb.before_init_train_context(new_workers)
            for arg, arg_values in args.items():
                assert len(arg_values) == len(new_workers), (
                    f"Callback {cb} returned {arg} with "
                    f"{len(arg_values)} values, expected {len(new_workers)}."
                )
                assert (
                    arg not in train_context_args
                ), f"Callback {cb} returned {arg} which is already set."
                train_context_args[arg] = arg_values

        # Initialize train context on new workers
        try:
            self._init_train_context_on_workers(
                new_workers, sync_actor, train_context_args
            )
        except RayActorError as actor_error:
            for worker in new_workers:
                ray.kill(worker.actor)
            error_msg = (
                "Replacement workers failed during train context initialization."
            )
            raise WorkerGroupStartupFailedError(error_msg) from actor_error

        # Launch training function on new workers
        ray_get_safe(
            [
                worker.actor.run_train_fn.remote(
                    self._worker_group_context.train_fn_ref
                )
                for worker in new_workers
            ]
        )