feat(ascend): support full and delta weight sync over disk

docker/npu_patch/vllm-ascend.patch

@@ -51,3 +51,41 @@ index a35d9af8d..66a179cd6 100644
 +        # stream. Wait before the packing stream reads them in torch.cat.
 +        streams[buffer_idx].wait_stream(source_stream)
          # Start tasks for the new buffer in a new stream
+diff --git a/vllm_ascend/worker/worker.py b/vllm_ascend/worker/worker.py
+index 062b3ecd..f0d1e4a1 100644
+--- a/vllm_ascend/worker/worker.py
++++ b/vllm_ascend/worker/worker.py
+@@ -333,1 +333,33 @@ class NPUWorker(WorkerBase):
++    def pull_weights(
++        self,
++        local_checkpoint_dir: str,
++        source_dir: str,
++        target_version: int,
++        pre_read_hook: str | None = None,
++    ) -> dict:
++        """Materialize a full or delta checkpoint on this rollout host.
++
++        ``collective_rpc`` invokes this on all NPU workers. The checkpoint
++        helper serializes same-host ranks with a filesystem lock, so each host
++        applies a version exactly once before reload_weights is called.
++        """
++        from vllm.utils.local_checkpoint import pull_checkpoint
++
++        # reload_weights updates model_config.model to the materialized local
++        # checkpoint. Preserve the original model path as the immutable
++        # version-zero seed for retries and restarted training runs.
++        base_dir = getattr(self, "_local_checkpoint_base_dir", None)
++        if base_dir is None:
++            base_dir = self.model_config.model
++            self._local_checkpoint_base_dir = base_dir
++
++        pull_checkpoint(
++            local_checkpoint_dir=local_checkpoint_dir,
++            base_dir=base_dir,
++            source_dir=source_dir,
++            target_version=target_version,
++            pre_read_hook=pre_read_hook,
++        )
++        return {"success": True, "weight_version": str(target_version)}
++
+     def shutdown(self) -> None: