[feat] Add Delta Weight Sync Support in SkyRL

[SumanthRH]

What does this PR do?

Adds delta weight sync to SkyRL following #1903

Implementation

Component	Category	Responsibility
DeltaWeightSyncConfig	Config	User-facing configuration for checkpoint-delta weight sync
DeltaTransferStrategy	Weight Sync	Creates the sender, receiver init payload and vLLM transfer engine
DeltaWeightTransferSender	Trainer	Owns the trainer-side weight send logic
DeltaPublisher	Trainer	Maintains the CPU byte snapshot, computes XOR deltas, compresses payloads and publishes manifests
DeltaManifest / DeltaTensorRecord	Data Model	Version and per-tensor metadata for a published delta
RemoteInferenceClient.fetch_weights	Control Plane	Runs the pre-pause receiver-side fetch+process phase on every inference worker
DeltaWeightTransferEngine	Inference	vLLM transfer engine for delta weight sync
LocalCheckpointStore	Inference	Maintains the local checkpoint and applies deltas into it

Test Plan

# CPU unit tests 
uv run --isolated --extra dev --extra fsdp pytest tests/backends/skyrl_train/weight_sync/
uv run --isolated --extra dev --extra fsdp pytest tests/backends/skyrl_train/inference_servers/test_remote_inference_client.py

# GPU end-to-end
uv run --isolated --extra dev --extra fsdp pytest -s -vvv \
  tests/backends/skyrl_train/gpu/gpu_ci/test_delta_weight_sync_e2e.py -m "not megatron"
uv run --isolated --extra dev --extra megatron pytest -s -vvv \
  tests/backends/skyrl_train/gpu/gpu_ci/test_delta_weight_sync_e2e.py -m megatron

Tests by component

All CPU tests live in tests/backends/skyrl_train/weight_sync/ unless noted otherwise.

Component	Category	Tests
DeltaWeightSyncConfig	Config	Covered indirectly through DeltaTransferStrategy below — test_delta_create_init_info_requires_sync_dir (required sync_dir) and test_delta_create_init_info (checkpoint_load_format validation). No direct unit test; see coverage gaps.
DeltaTransferStrategy	Weight Sync	1.test_transfer_strategies.py::test_delta_create_init_info: config → DeltaInitInfo field mapping, including override_existing_receiver for remote engines 2. test_transfer_strategies.py::test_delta_create_init_info_requires_sync_dir: raises ValueError when the delta sub-config is absent or sync_dir is unset
DeltaWeightTransferSender	Trainer	1. test_delta_sender_seed_sync_skips_chunk_iteration: the first sync treats base_model_path as version 0 and skips chunk extraction/publish entirely 2. test_delta_checkpoint_non_source_rank_drains_without_publishing: non-source ranks drain the chunk stream to drive the extractor's collectives without publishing
DeltaPublisher	Trainer	1. test_delta_checkpoint_payload_stores_xor_patch: payload is the XOR patch, and base ^ patch == updated byte-for-byte 2.test_delta_checkpoint_publisher_converts_to_base_checkpoint_dtype: runtime bf16 tensors are normalized back to the base checkpoint's fp32 3. test_delta_checkpoint_splits_payload_files_by_size: max_file_size_in_gb starts a new safetensors file 4. test_delta_checkpoint_skips_missing_lm_head_when_checkpoint_ties_embeddings: tied-embedding models skip lm_head.weight instead of failing 5. test_delta_checkpoint_unchanged_publish_advances_version: an unchanged publish is not a no-op — it advances the version and writes an empty delta
DeltaManifest / DeltaTensorRecord	Data Model	1. test_delta_checkpoint_publisher_converts_to_base_checkpoint_dtype: asserts dtype, payload_key, checksum_algorithm == "xxh3-128" and uncompressed_num_bytes 2. test_delta_checkpoint_unchanged_publish_advances_version: asserts empty tensors / payload_files on a no-change version
RemoteInferenceClient.fetch_weights	Control Plane	inference_servers/test_remote_inference_client.py::test_fetch_weights: fan-out of the pre-pause fetch to every server
DeltaWeightTransferEngine	Inference	No CPU unit test — the engine runs inside a vLLM worker process. Exercised by the GPU e2e test below.
LocalCheckpointStore	Inference	1. test_delta_checkpoint_publish_fetch_and_reload_roundtrip: publish → fetch → reload; changed and unchanged tensors both land correctly, state advances to v1 2. test_delta_checkpoint_replays_multiple_versions_for_late_join: a receiver joining at v0 replays v1 then v2 3. test_local_checkpoint_store_fetch_is_single_writer_with_concurrent_ray_actors: concurrent Ray actors sharing a cache dir serialize on the file lock and download once 4. test_delta_checkpoint_checksum_failure_marks_write_in_progress: a corrupt manifest leaves write_in_progress=True, and a subsequent valid delta recovers 5. test_delta_checkpoint_vllm_multi_thread_safetensors_iterator_roundtrip: iter_tensors round-trip under the multi-thread safetensors loader 6. test_safe_path_name_disambiguates_long_sibling_uris: per-version cache keys stay distinct when the sync_dir URI is long enough to be truncated
Cloud transport	Storage	1. test_delta_checkpoint_gcs_cli_publish_fetch_roundtrip: full publish/fetch round-trip for GCS with mocked transfer 2. test_delta_checkpoint_s3_cli_publish_fetch_roundtrip: similar test for S3

Integration Tests

tests/backends/skyrl_train/gpu/gpu_ci/test_delta_weight_sync_e2e.py::test_delta_weight_sync_sparse_update_e2e, parametrized over fsdp and megatron (the latter behind the megatron marker).

Runs Qwen/Qwen3-0.6B non-colocated (1 trainer GPU, 1 vLLM engine at TP=1), applies a sparse weight perturbation between syncs to simulate a weight update on the trainer side.

Other tests

• test_prefix_cache_reset.py: Tests who resets the inference engines' prefix cache (PolicyWorker or WeightTransferSender)
• test_worker_dispatch.py::TestSaveWeights : Tests the new dispatcher branch in save_weights_for_sampler for delta weight sync.

E2E runs

I've tested Delta weight sync with disk, GCS and S3 based weight transfer with a small model (Qwen 1.5B Instruct) on the GSM8K dataset on 4 GPUs.

1. gsm8k-qwen1p5b-nccl: Baseline, NCCL based weight sync
2. gsm8k-qwen1p5b-delta-disk: Delta based weight sync via shared disk
3. gsm8k-qwen1p5b-delta-gcs: Delta based weight sync via GCS
4. gsm8k-qwen1p5b-delta-s3: Delta based weight sync via S3
5. gsm8k-qwen1p5b-delta-s3-disagg: Delta based weight sync via S3 in a disaggregated setup: trainer and inference nodes are in separate clusters.

Performance

Here are some performance numbers for delta weight sync for training Qwen/Qwen3.5-35B-A3B on the DAPO recipe with delta weight sync via GCS on 2 8xB200 nodes (1 trainer, 1 inference node in a non-colocated setup) :

Metric	Value
Total sync_weights	75.49 s
E2E publish (incl. upload)	47.05 s
— GCS upload	6.41 s
— non-upload publish (all-gather, CPU copy, delta, compression)	40.64 s
Compressed delta size	2.66 GiB
Uncompressed changed bytes	64.56 GiB
/fetch_weights	15.78 s
/update_weights	9.5 s

There are some known low hanging fruits (ex: fetch and publish are both handled by a single rank right now, weight loading is currently disk -> CPU -> GPU instead of disk -> GPU because of a vllm limitation)

[gemini-code-assist]

Code Review

This pull request introduces a new checkpoint-delta weight synchronization strategy ('Delta strategy') for non-colocated training and inference setups, allowing compressed XOR deltas to be published and fetched via shared filesystems or cloud storage (GCS/S3). The review feedback highlights critical bugs in both FSDP and Megatron workers where the persistent inference client is prematurely closed during non-delta syncs. Additionally, the reviewer pointed out several resource leaks (file descriptors and handles) during lock exceptions or mmap failures, high memory overhead when writing local tensors, and robustness issues concerning unhandled memory errors, uncleaned temporary files, and dataclass schema evolution.

[gemini-code-assist]

Resource Leak: File Descriptor Leaked on flock Exception

If fcntl.flock(fd, flags) raises an exception other than BlockingIOError (for example, OSError with ENOLCK or EOPNOTSUPP on NFS filesystems that do not support locking), the file descriptor fd is leaked because it is never closed. Since NFS is explicitly supported as a target filesystem for delta weight sync, this is a highly probable failure mode.

    def acquire(self, blocking: bool = True) -> bool:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        fd = os.open(self.path, os.O_CREAT | os.O_RDWR, 0o644)
        flags = fcntl.LOCK_EX
        if not blocking:
            flags |= fcntl.LOCK_NB
        try:
            fcntl.flock(fd, flags)
        except BlockingIOError:
            os.close(fd)
            return False
        except Exception:
            os.close(fd)
            raise
        self._fd = fd
        return True

[gemini-code-assist]

Robustness: Clean Up Temporary Files on Copy Failure

If _run_cloud_cp or shutil.copy2 fails, the temporary file tmp_path is left behind in the checkpoint directory. Wrapping the copy and replace operations in a try...except block ensures that any stale temporary files are cleaned up on failure.

Suggested change

	def _copy_from_uri(uri: str, local_path: Path) -> None:
	local_path.parent.mkdir(parents=True, exist_ok=True)
	tmp_path = local_path.with_name(f".{local_path.name}.{os.getpid()}.tmp")
	if tmp_path.exists():
	tmp_path.unlink()
	if _is_cloud_uri(uri):
	_run_cloud_cp(uri, str(tmp_path))
	else:
	shutil.copy2(Path(uri), tmp_path)
	os.replace(tmp_path, local_path)
	def _copy_from_uri(uri: str, local_path: Path) -> None:
	local_path.parent.mkdir(parents=True, exist_ok=True)
	tmp_path = local_path.with_name(f".{local_path.name}.{os.getpid()}.tmp")
	if tmp_path.exists():
	tmp_path.unlink()
	try:
	if _is_cloud_uri(uri):
	_run_cloud_cp(uri, str(tmp_path))
	else:
	shutil.copy2(Path(uri), tmp_path)
	os.replace(tmp_path, local_path)
	except Exception:
	tmp_path.unlink(missing_ok=True)
	raise

[SumanthRH]

Interface change: Added base_model_path because this is needed for the delta weight sync strategy

Debatable if we should upgrade this to the full TrainerConfig and pass that - but IMO we should keep the args for our abstractions as minimal as possible so I've just used base_model_path for now.

[SumanthRH]

Interface change: Added a class attribute handles_prefix_cache_reset to indicate whether the sender will handle resetting prefix cache internally.

[SumanthRH]

Important change: Delta weight sync runs pause_generation internally - so we need to skip it in the controller/ WorkerDispatch method