diff --git a/tests/test_empty_colocated_weight_bucket.py b/tests/test_empty_colocated_weight_bucket.py
index c900d846..d2051fe0 100644
--- a/tests/test_empty_colocated_weight_bucket.py
+++ b/tests/test_empty_colocated_weight_bucket.py
@@ -150,42 +150,44 @@ def _load_update_weight_module(monkeypatch):
     return module, dist_state
 
 
-def test_empty_colocated_bucket_does_not_hide_remote_weights(monkeypatch):
+def test_packed_colocated_bucket_rejects_mismatched_rank_metadata(monkeypatch):
     module, _ = _load_update_weight_module(monkeypatch)
-    empty = {"names": [], "dtype_names": [], "shapes": [], "ipc_handles": []}
+    empty = {
+        "names": [],
+        "dtype_names": [],
+        "shapes": [],
+        "tensor_sizes": [],
+        "ipc_handles": {"gpu-0": ("empty",)},
+    }
     remote = {
         "names": ["expert.weight"],
         "dtype_names": ["bfloat16"],
         "shapes": [[4, 8]],
-        "ipc_handles": [{"gpu-1": ("remote",)}],
+        "tensor_sizes": [64],
+        "ipc_handles": {"gpu-1": ("remote",)},
     }
 
-    assert module._merge_ipc_update_infos([empty, remote]) == remote
+    with pytest.raises(ValueError, match="packed IPC metadata must match"):
+        module._merge_ipc_update_infos([empty, remote])
 
 
-def test_colocated_bucket_merges_handles_by_parameter_name(monkeypatch):
+def test_packed_colocated_bucket_merges_rank_handles(monkeypatch):
     module, _ = _load_update_weight_module(monkeypatch)
     first = {
-        "names": ["shared.weight"],
-        "dtype_names": ["float16"],
-        "shapes": [[2, 2]],
-        "ipc_handles": [{"gpu-0": ("first",)}],
+        "names": ["shared.weight", "expert.weight"],
+        "dtype_names": ["float16", "bfloat16"],
+        "shapes": [[2, 2], [4, 8]],
+        "tensor_sizes": [8, 64],
+        "ipc_handles": {"gpu-0": ("first",)},
     }
     second = {
-        "names": ["expert.weight", "shared.weight"],
-        "dtype_names": ["bfloat16", "float16"],
-        "shapes": [[4, 8], [2, 2]],
-        "ipc_handles": [{"gpu-1": ("expert",)}, {"gpu-1": ("second",)}],
+        **first,
+        "ipc_handles": {"gpu-1": ("second",)},
     }
 
     assert module._merge_ipc_update_infos([first, second]) == {
-        "names": ["shared.weight", "expert.weight"],
-        "dtype_names": ["float16", "bfloat16"],
-        "shapes": [[2, 2], [4, 8]],
-        "ipc_handles": [
-            {"gpu-0": ("first",), "gpu-1": ("second",)},
-            {"gpu-1": ("expert",)},
-        ],
+        **first,
+        "ipc_handles": {"gpu-0": ("first",), "gpu-1": ("second",)},
     }
 
 
diff --git a/tests/utils/test_update_weight_from_distributed.py b/tests/utils/test_update_weight_from_distributed.py
index faab2b65..ccf9c323 100644
--- a/tests/utils/test_update_weight_from_distributed.py
+++ b/tests/utils/test_update_weight_from_distributed.py
@@ -17,8 +17,12 @@
 import _unit_stubs
 import pytest
 import torch
+from vime.utils.types import ParamInfo
 
 MODULE_PATH = "vime.backends.megatron_utils.update_weight.update_weight_from_distributed"
+COMMON_MODULE = "vime.backends.megatron_utils.update_weight.common"
+DIRECT_MODULE = "vime.backends.megatron_utils.update_weight.hf_weight_iterator_direct"
+CONVERTER_MODULE = "vime.backends.megatron_utils.megatron_to_hf"
 
 NUM_GPUS = 0
 
@@ -195,7 +199,7 @@ def _patch_trainer_send(monkeypatch, upw, seen: list[dict]) -> None:
 
 def _make_instance(upw):
     obj = object.__new__(upw.UpdateWeightFromDistributed)
-    obj.args = type("Args", (), {"update_weight_buffer_size": 1 << 30, "vllm_weight_sync_packed": True})()
+    obj.args = type("Args", (), {"update_weight_buffer_size": 1 << 30})()
     obj.model = []
     obj.weights_getter = lambda: {}
     obj.model_name = "test"
@@ -215,7 +219,8 @@ def test_signature_no_use_vllm(upw):
     sig = inspect.signature(upw.update_weights_from_distributed)
     params = sig.parameters
     assert "use_vllm" not in params
-    for p in ("group_name", "group", "weight_version", "rollout_engines", "converted_named_tensors", "packed"):
+    assert "packed" not in params
+    for p in ("group", "weight_version", "rollout_engines", "converted_named_tensors"):
         assert p in params
 
 
@@ -223,25 +228,23 @@ def test_signature_no_use_vllm(upw):
 def test_signature_rejects_legacy_use_vllm_call(upw):
     with pytest.raises(TypeError, match="use_vllm"):
         upw.update_weights_from_distributed(
-            "g",
             DummyGroup(),
             1,
             [RecordingEngine()],
             _real_tensors(),
             use_vllm=True,
-            packed=False,
         )
 
 
 @pytest.mark.unit
-def test_packed_true_uses_vllm_trainer_send_weights(upw, monkeypatch):
+def test_uses_packed_vllm_trainer_send_weights(upw, monkeypatch):
     group = DummyGroup()
     engine = RecordingEngine()
     tensors = _real_tensors()
     seen = []
     _patch_trainer_send(monkeypatch, upw, seen)
 
-    refs = upw.update_weights_from_distributed("groupA", group, 7, [engine], tensors, packed=True)
+    refs = upw.update_weights_from_distributed(group, 7, [engine], tensors)
 
     assert len(seen) == 1
     sent = seen[0]["items"]
@@ -251,35 +254,6 @@ def test_packed_true_uses_vllm_trainer_send_weights(upw, monkeypatch):
     assert refs == ["ref"]
 
 
-@pytest.mark.unit
-def test_packed_false_still_uses_vllm_trainer_send_weights(upw, monkeypatch):
-    group = DummyGroup()
-    engine = RecordingEngine()
-    tensors = _real_tensors()
-    seen = []
-    _patch_trainer_send(monkeypatch, upw, seen)
-
-    refs = upw.update_weights_from_distributed("groupB", group, 7, [engine], tensors, packed=False)
-
-    assert len(seen) == 1
-    assert len(seen[0]["items"]) == len(tensors)
-    assert seen[0]["packed"] is False
-    assert refs == ["ref"]
-
-
-@pytest.mark.unit
-def test_default_packed_is_false(upw, monkeypatch):
-    group = DummyGroup()
-    engine = RecordingEngine()
-    seen = []
-    _patch_trainer_send(monkeypatch, upw, seen)
-
-    upw.update_weights_from_distributed("g", group, 1, [engine], _real_tensors())
-
-    assert len(seen) == 1
-    assert seen[0]["packed"] is False
-
-
 @pytest.mark.unit
 def test_no_dist_broadcast_fallback(upw, monkeypatch):
     import torch.distributed as dist
@@ -295,52 +269,34 @@ def fake_broadcast(*a, **k):
 
     group = DummyGroup()
     engine = RecordingEngine()
-    upw.update_weights_from_distributed("g", group, 1, [engine], _real_tensors(), packed=False)
+    upw.update_weights_from_distributed(group, 1, [engine], _real_tensors())
 
     assert seen_broadcast == []
     assert len(seen_send) == 1
 
 
 @pytest.mark.unit
-def test_remote_kwargs_include_packed_true(upw, monkeypatch):
+def test_remote_kwargs_are_always_packed(upw, monkeypatch):
     group = DummyGroup()
     engine = RecordingEngine()
     tensors = _real_tensors(n=1)
     seen_send = []
     _patch_trainer_send(monkeypatch, upw, seen_send)
 
-    upw.update_weights_from_distributed("myg", group, 42, [engine], tensors, packed=True)
+    upw.update_weights_from_distributed(group, 42, [engine], tensors)
 
     assert len(seen_send) == 1
     assert seen_send[0]["packed"] is True
     assert len(engine.update_weights_from_distributed.calls) == 1
     kw = engine.update_weights_from_distributed.calls[0].kwargs
-    assert kw["packed"] is True
-    assert kw["group_name"] == "myg"
+    assert "packed" not in kw
+    assert "group_name" not in kw
     assert kw["weight_version"] == "42"
     assert kw["names"] == ["layer.0.weight"]
     assert kw["shapes"] == [torch.Size([2, 2])]
     assert kw["dtypes"] == [torch.float32]
 
 
-@pytest.mark.unit
-def test_remote_kwargs_include_packed_false(upw, monkeypatch):
-    group = DummyGroup()
-    engine = RecordingEngine()
-    tensors = _real_tensors(n=2)
-    seen_send = []
-    _patch_trainer_send(monkeypatch, upw, seen_send)
-
-    upw.update_weights_from_distributed("g", group, 99, [engine], tensors, packed=False)
-
-    assert len(seen_send) == 1
-    assert seen_send[0]["packed"] is False
-    kw = engine.update_weights_from_distributed.calls[0].kwargs
-    assert kw["packed"] is False
-    assert kw["weight_version"] == "99"
-    assert kw["names"] == ["layer.0.weight", "layer.1.weight"]
-
-
 @pytest.mark.unit
 def test_remote_kwargs_no_use_vllm(upw, monkeypatch):
     group = DummyGroup()
@@ -348,7 +304,7 @@ def test_remote_kwargs_no_use_vllm(upw, monkeypatch):
     seen_send = []
     _patch_trainer_send(monkeypatch, upw, seen_send)
 
-    upw.update_weights_from_distributed("g", group, 1, [engine], _real_tensors(), packed=False)
+    upw.update_weights_from_distributed(group, 1, [engine], _real_tensors())
 
     assert len(seen_send) == 1
     kw = engine.update_weights_from_distributed.calls[0].kwargs
@@ -362,7 +318,7 @@ def test_multiple_engines_each_get_call(upw, monkeypatch):
     seen_send = []
     _patch_trainer_send(monkeypatch, upw, seen_send)
 
-    upw.update_weights_from_distributed("g", group, 1, engines, _real_tensors(), packed=True)
+    upw.update_weights_from_distributed(group, 1, engines, _real_tensors())
     assert len(seen_send) == 1
     assert seen_send[0]["packed"] is True
     for e in engines:
@@ -376,7 +332,7 @@ def test_empty_tensor_list_still_dispatches(upw, monkeypatch):
     seen_send = []
     _patch_trainer_send(monkeypatch, upw, seen_send)
 
-    refs = upw.update_weights_from_distributed("g", group, 1, [engine], [], packed=False)
+    refs = upw.update_weights_from_distributed(group, 1, [engine], [])
 
     assert refs == ["ref"]
     kw = engine.update_weights_from_distributed.calls[0].kwargs
@@ -384,24 +340,24 @@ def test_empty_tensor_list_still_dispatches(upw, monkeypatch):
     assert kw["shapes"] == []
     assert len(seen_send) == 1
     assert seen_send[0]["items"] == []
+    assert seen_send[0]["packed"] is True
 
 
 @pytest.mark.unit
-def test_raw_packed_path_sends_dense_chunks_only(upw, monkeypatch):
+def test_raw_path_sends_dense_then_expert(upw, monkeypatch):
     obj = _make_instance(upw)
     obj._is_pp_src_rank = True
     obj._group_name = "g"
     obj._hf_weight_iterator = None
-    obj._use_vllm_packed = lambda: True
     obj._iter_non_expert_chunks = lambda: iter([[("dense.0", torch.zeros(1))], [("dense.1", torch.zeros(1))]])
-    obj._iter_expert_chunks = lambda: (_ for _ in ()).throw(AssertionError("expert pass should be skipped"))
+    obj._iter_expert_chunks = lambda: iter([[("expert.0", torch.zeros(1))]])
 
-    seen: list[tuple[list[str], bool, str]] = []
+    seen: list[tuple[list[str], str]] = []
     monkeypatch.setattr(
         upw.UpdateWeightFromDistributed,
         "_update_bucket_weights_from_distributed",
-        lambda self, converted_named_tensors, pbar=None, packed=False: seen.append(
-            ([name for name, _ in converted_named_tensors], packed, pbar)
+        lambda self, converted_named_tensors, pbar=None: seen.append(
+            ([name for name, _ in converted_named_tensors], pbar)
         ),
     )
     monkeypatch.setattr(upw.dist, "barrier", lambda *args, **kwargs: None)
@@ -409,43 +365,76 @@ def test_raw_packed_path_sends_dense_chunks_only(upw, monkeypatch):
 
     upw.UpdateWeightFromDistributed._send_weights(obj, pbar="pbar")
 
-    assert seen == [(["dense.0"], True, "pbar"), (["dense.1"], True, "pbar")]
+    assert seen == [
+        (["dense.0"], "pbar"),
+        (["dense.1"], "pbar"),
+        (["expert.0"], "pbar"),
+    ]
+
+
+@pytest.mark.unit
+def test_source_has_no_packed_mode_switch(upw):
+    src = inspect.getsource(upw)
+    assert "vllm_weight_sync_packed" not in src
+    assert "packed=False" not in src
 
 
 @pytest.mark.unit
-def test_raw_nonpacked_path_runs_dense_then_expert(upw, monkeypatch):
+def test_single_ep_converts_without_collective(upw, monkeypatch):
     obj = _make_instance(upw)
-    obj._is_pp_src_rank = True
-    obj._group_name = "g"
-    obj._hf_weight_iterator = None
-    obj._use_vllm_packed = lambda: False
-    obj._iter_non_expert_chunks = lambda: iter([[("dense.0", torch.zeros(1))], [("dense.1", torch.zeros(1))]])
-    obj._iter_expert_chunks = lambda: iter([[("expert.0", torch.zeros(1))]])
+    tensors = [("expert.0", torch.ones(2)), ("expert.1", torch.ones(3))]
+    collectives = []
 
-    seen: list[tuple[list[str], bool, str]] = []
+    monkeypatch.setattr(upw.mpu, "get_expert_model_parallel_world_size", lambda: 1)
+    monkeypatch.setattr(upw.dist, "all_gather", lambda *args, **kwargs: collectives.append(args))
     monkeypatch.setattr(
-        upw.UpdateWeightFromDistributed,
-        "_update_bucket_weights_from_distributed",
-        lambda self, converted_named_tensors, pbar=None, packed=False: seen.append(
-            ([name for name, _ in converted_named_tensors], packed, pbar)
-        ),
+        upw,
+        "convert_to_hf",
+        lambda args, model_name, name, tensor, quantization_config: [(f"hf.{name}", tensor)],
     )
-    barriers: list[str] = []
-    monkeypatch.setattr(upw.dist, "barrier", lambda *args, **kwargs: barriers.append(kwargs.get("group")))
-    monkeypatch.setattr(upw, "get_gloo_group", lambda: "gloo")
 
-    upw.UpdateWeightFromDistributed._send_weights(obj, pbar="pbar")
+    converted = upw.UpdateWeightFromDistributed._ep_gather_and_convert(obj, tensors)
 
-    assert seen == [
-        (["dense.0"], False, "pbar"),
-        (["dense.1"], False, "pbar"),
-        (["expert.0"], False, "pbar"),
+    assert [name for name, _ in converted] == ["hf.expert.0", "hf.expert.1"]
+    assert tensors == []
+    assert collectives == []
+
+
+@pytest.mark.unit
+def test_expert_chunks_keep_each_layer_together(upw, monkeypatch):
+    obj = _make_instance(upw)
+    obj.args.update_weight_buffer_size = 24
+    params = [
+        ("decoder.layers.0.mlp.experts.linear_fc1.weight0", torch.ones(2)),
+        ("decoder.layers.1.mlp.experts.linear_fc1.weight0", torch.ones(2)),
+        ("decoder.layers.0.mlp.experts.linear_fc2.weight0", torch.ones(2)),
+        ("decoder.layers.1.mlp.experts.linear_fc2.weight0", torch.ones(2)),
+    ]
+
+    monkeypatch.setattr(upw, "all_gather_param", lambda name, param: param)
+    monkeypatch.setattr(upw.mpu, "get_expert_model_parallel_world_size", lambda: 1)
+    monkeypatch.setattr(
+        upw,
+        "convert_to_hf",
+        lambda args, model_name, name, tensor, quantization_config: [(name, tensor)],
+    )
+
+    chunks = list(upw.UpdateWeightFromDistributed._iter_expert_chunks(obj, iter(params)))
+
+    assert [[name for name, _ in chunk] for chunk in chunks] == [
+        [
+            "decoder.layers.0.mlp.experts.linear_fc1.weight0",
+            "decoder.layers.0.mlp.experts.linear_fc2.weight0",
+        ],
+        [
+            "decoder.layers.1.mlp.experts.linear_fc1.weight0",
+            "decoder.layers.1.mlp.experts.linear_fc2.weight0",
+        ],
     ]
-    assert barriers == ["gloo", "gloo"]
 
 
 @pytest.mark.unit
-def test_bridge_path_forwards_packed_flag_and_listifies_chunks(upw, monkeypatch):
+def test_bridge_path_listifies_chunks(upw, monkeypatch):
     obj = _make_instance(upw)
     obj._is_pp_src_rank = True
     obj._group_name = "g"
@@ -455,29 +444,29 @@ def test_bridge_path_forwards_packed_flag_and_listifies_chunks(upw, monkeypatch)
         ((("bridge.0", torch.zeros(1)),), (("bridge.1", torch.zeros(1)),))
     )
 
-    seen: list[tuple[list[str], bool, str]] = []
+    seen: list[tuple[list[str], str]] = []
     monkeypatch.setattr(
         upw.UpdateWeightFromDistributed,
         "_update_bucket_weights_from_distributed",
-        lambda self, converted_named_tensors, pbar=None, packed=False: seen.append(
-            ([name for name, _ in converted_named_tensors], packed, pbar)
+        lambda self, converted_named_tensors, pbar=None: seen.append(
+            ([name for name, _ in converted_named_tensors], pbar)
         ),
     )
     monkeypatch.setattr(upw.dist, "barrier", lambda *args, **kwargs: None)
     monkeypatch.setattr(upw, "get_gloo_group", lambda: "gloo")
 
-    upw.UpdateWeightFromDistributed._sync_bridge_weights_to_rollout_engines(obj, pbar="pbar", use_vllm_packed=True)
+    upw.UpdateWeightFromDistributed._sync_bridge_weights_to_rollout_engines(obj, pbar="pbar")
 
     assert seen == [
-        (["bridge.0"], True, "pbar"),
-        (["bridge.1"], True, "pbar"),
+        (["bridge.0"], "pbar"),
+        (["bridge.1"], "pbar"),
     ]
 
 
 @pytest.mark.unit
 def test_source_no_standalone_use_vllm_param(upw):
     src = inspect.getsource(upw)
-    lines = [line.strip() for line in src.splitlines() if "use_vllm=" in line and "use_vllm_packed" not in line]
+    lines = [line.strip() for line in src.splitlines() if "use_vllm=" in line]
     assert lines == []
 
 
@@ -670,7 +659,6 @@ def test_bridge_export_runs_on_non_source_pp_stage(upw, monkeypatch):
     obj._hf_weight_iterator.get_hf_weight_chunks.return_value = []
     barriers: list[object] = []
 
-    monkeypatch.setattr(upw.UpdateWeightFromDistributed, "_use_vllm_packed", lambda self: True)
     monkeypatch.setattr(upw, "get_gloo_group", lambda: "gloo")
     monkeypatch.setattr(upw.dist, "barrier", lambda *args, **kwargs: barriers.append(kwargs.get("group")))
 
@@ -731,5 +719,149 @@ def test_cuda_sync_once_after_all_buckets_not_per_bucket(upw):
     assert "torch.cuda.synchronize" in sync_src
 
 
+@pytest.fixture
+def weight_modules():
+    module_names = (
+        "megatron",
+        "megatron.core",
+        "megatron.core.parallel_state",
+        "megatron.core.transformer",
+        "megatron.core.transformer.transformer_layer",
+        CONVERTER_MODULE,
+        COMMON_MODULE,
+        DIRECT_MODULE,
+    )
+    saved = _unit_stubs.save_sys_modules(module_names)
+    for name in module_names:
+        sys.modules.pop(name, None)
+    _unit_stubs.install_megatron_mpu_stub()
+    converter = types.ModuleType(CONVERTER_MODULE)
+    converter.convert_to_hf = lambda *args, **kwargs: []
+    sys.modules[CONVERTER_MODULE] = converter
+    try:
+        yield importlib.import_module(COMMON_MODULE), importlib.import_module(DIRECT_MODULE)
+    finally:
+        _unit_stubs.restore_sys_modules(saved)
+
+
+class _Handle:
+    def wait(self) -> None:
+        pass
+
+
+def _param_info(name: str, param: torch.Tensor, src_rank: int = 0) -> ParamInfo:
+    return ParamInfo(name, param.dtype, param.shape, {}, param.nbytes, src_rank)
+
+
+def _tp_param(values, partition_dim: int) -> torch.nn.Parameter:
+    param = torch.nn.Parameter(torch.tensor(values, dtype=torch.float32))
+    param.tensor_model_parallel = True
+    param.partition_dim = partition_dim
+    param.partition_stride = 1
+    return param
+
+
+@pytest.mark.unit
+def test_single_tp_returns_parameter_without_collective(monkeypatch, weight_modules):
+    common, _ = weight_modules
+    parameter = _tp_param([[1.0, 1.0]], partition_dim=0)
+    calls = []
+    monkeypatch.setattr(common.mpu, "get_tensor_model_parallel_world_size", lambda: 1)
+    monkeypatch.setattr(common.mpu, "get_tensor_model_parallel_group", lambda: "tp")
+    monkeypatch.setattr(common.dist, "all_gather", lambda *args, **kwargs: calls.append(args))
+
+    gathered = common.all_gather_param("decoder.weight", parameter)
+
+    assert gathered.data_ptr() == parameter.data_ptr()
+    assert calls == []
+
+
+@pytest.mark.unit
+def test_all_gather_params_coalesces_and_restores_layouts(monkeypatch, weight_modules):
+    common, _ = weight_modules
+    direct = torch.nn.Parameter(torch.tensor([99.0]))
+    direct.tensor_model_parallel = False
+    column = _tp_param([[1.0, 2.0], [3.0, 4.0]], partition_dim=0)
+    glu = _tp_param([[1.0], [2.0], [10.0], [20.0]], partition_dim=0)
+    glu.partition_stride = 2
+    row = _tp_param([[1.0, 2.0], [3.0, 4.0]], partition_dim=0)
+    entries = [
+        (_param_info("dense", direct), direct),
+        (_param_info("linear.weight", column), column),
+        (_param_info("linear_fc1.weight", glu), glu),
+        (_param_info("linear_fc2.weight", row), row),
+    ]
+    remote_flat = torch.cat(
+        [
+            torch.tensor([[5.0, 6.0], [7.0, 8.0]]).flatten(),
+            torch.tensor([[3.0], [4.0], [30.0], [40.0]]).flatten(),
+            torch.tensor([[5.0, 6.0], [7.0, 8.0]]).flatten(),
+        ]
+    )
+    calls = []
+
+    def all_gather_into_tensor(output, local, group, async_op):
+        calls.append((group, async_op))
+        output[: local.numel()].copy_(local)
+        output[local.numel() :].copy_(remote_flat)
+        return _Handle()
+
+    monkeypatch.setattr(common.mpu, "get_tensor_model_parallel_world_size", lambda: 2)
+    monkeypatch.setattr(common.mpu, "get_tensor_model_parallel_group", lambda: "tp")
+    monkeypatch.setattr(common.dist, "all_gather_into_tensor", all_gather_into_tensor)
+
+    gathered = common.all_gather_params_async(entries)
+
+    assert calls == [("tp", True)]
+    assert gathered[0].data_ptr() == direct.data_ptr()
+    assert torch.equal(gathered[1], torch.tensor([[1.0, 2.0], [3.0, 4.0], [5.0, 6.0], [7.0, 8.0]]))
+    assert torch.equal(gathered[2], torch.tensor([[1.0], [2.0], [3.0], [4.0], [10.0], [20.0], [30.0], [40.0]]))
+    assert torch.equal(gathered[3], torch.tensor([[1.0, 2.0, 5.0, 6.0], [3.0, 4.0, 7.0, 8.0]]))
+
+
+@pytest.mark.unit
+def test_broadcast_expert_params_coalesces_by_source(monkeypatch, weight_modules):
+    _, direct = weight_modules
+    params = [torch.tensor([float(index)]) for index in range(4)]
+    infos = [
+        _param_info("layers.0.experts.0.weight", params[0], src_rank=4),
+        _param_info("layers.0.experts.1.weight", params[1], src_rank=5),
+        _param_info("layers.0.dense.weight", params[2], src_rank=4),
+        _param_info("layers.0.experts.2.weight", params[3], src_rank=9),
+    ]
+    calls = []
+    monkeypatch.setattr(direct.mpu, "get_expert_model_parallel_group", lambda: "ep")
+    monkeypatch.setattr(
+        direct.dist,
+        "_broadcast_coalesced",
+        lambda group, tensors, buffer_size, src: calls.append((group, tensors, buffer_size, src)),
+    )
+
+    direct._broadcast_expert_params(infos, params, 1024, {4: 0, 5: 1, 9: 0})
+
+    assert calls == [
+        ("ep", [params[0], params[3]], 1024, 0),
+        ("ep", [params[1]], 1024, 1),
+    ]
+
+
+@pytest.mark.unit
+def test_ep_broadcast_source_map_tracks_pp_groups(monkeypatch, weight_modules):
+    _, direct = weight_modules
+    monkeypatch.setattr(direct.mpu, "get_expert_model_parallel_group", lambda: "ep")
+    monkeypatch.setattr(direct.mpu, "get_expert_model_parallel_world_size", lambda: 2)
+    monkeypatch.setattr(direct.mpu, "get_pipeline_model_parallel_group", lambda: "pp")
+    monkeypatch.setattr(direct.dist, "get_process_group_ranks", lambda group: [0, 2] if group == "pp" else [0, 1])
+
+    def all_gather_object(output, local_pp_group, group):
+        assert local_pp_group == [0, 2]
+        assert group == "ep"
+        output[:] = [[0, 2], [1, 3]]
+
+    monkeypatch.setattr(direct.dist, "all_gather_object", all_gather_object)
+
+    assert direct._get_ep_broadcast_src_rank_map() == {0: 0, 2: 0, 1: 1, 3: 1}
+
+
 if __name__ == "__main__":
     raise SystemExit(pytest.main([__file__]))
diff --git a/tests/utils/test_update_weight_from_tensor.py b/tests/utils/test_update_weight_from_tensor.py
index 83c67a9b..58833c1b 100644
--- a/tests/utils/test_update_weight_from_tensor.py
+++ b/tests/utils/test_update_weight_from_tensor.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import importlib
+import inspect
 import sys
 import types
 from argparse import Namespace
@@ -217,8 +218,14 @@ def test_colocated_lifecycle_uses_pause_flush_and_weight_transfer_apis(upw_vllm)
     engine = RecordingVLLMEngine()
     _bind_single_slot(obj, engine, src=0)
 
-    dummy_info = {"names": ["w"], "dtype_names": ["bfloat16"], "shapes": [[2, 2]], "ipc_handles": [{"u": ("f", ())}]}
-    with patch(f"{MODULE_PATH}._build_ipc_update_info_from_named_tensors", return_value=(dummy_info, [])):
+    dummy_info = {
+        "names": ["w"],
+        "dtype_names": ["bfloat16"],
+        "shapes": [[2, 2]],
+        "tensor_sizes": [8],
+        "ipc_handles": {"u": ("f", ())},
+    }
+    with patch(f"{MODULE_PATH}._build_packed_ipc_update_info", return_value=(dummy_info, [])):
         counters = _run_update(obj, chunks=_chunks(2))
 
     # Colocate quiesce: pause_generation + flush_cache only, no /sleep round-trip;
@@ -233,8 +240,8 @@ def test_colocated_lifecycle_uses_pause_flush_and_weight_transfer_apis(upw_vllm)
     assert engine.start_weight_update.calls[0].kwargs.get("is_checkpoint_format") is True
     assert len(engine.finish_weight_update.calls) == 1
     assert len(engine.continue_generation.calls) == 1
-    # ipc_collect: one per HF chunk + one after the loop.
-    assert counters["ipc_collect"] == 2 + 1
+    # Both chunks are kept alive until the bounded in-flight batch drains.
+    assert counters["ipc_collect"] == 2
     # lifecycle barriers (no per-chunk barrier).
     assert counters["barrier"] >= 4
 
@@ -251,8 +258,14 @@ def test_colocated_mtp_updates_target_then_draft_from_fresh_weight_stream(upw_vl
     engine = RecordingVLLMEngine()
     _bind_single_slot(obj, engine, src=0)
 
-    dummy_info = {"names": ["w"], "dtype_names": ["bfloat16"], "shapes": [[2, 2]], "ipc_handles": []}
-    with patch(f"{MODULE_PATH}._build_ipc_update_info_from_named_tensors", return_value=(dummy_info, [])):
+    dummy_info = {
+        "names": ["w"],
+        "dtype_names": ["bfloat16"],
+        "shapes": [[2, 2]],
+        "tensor_sizes": [8],
+        "ipc_handles": {},
+    }
+    with patch(f"{MODULE_PATH}._build_packed_ipc_update_info", return_value=(dummy_info, [])):
         _run_update(obj, chunks=_chunks(2))
 
     assert len(engine.start_weight_update.calls) == 1
@@ -272,9 +285,15 @@ def test_send_via_ipc_dispatches_update_weights_from_tensor_with_version(upw_vll
     engine = RecordingVLLMEngine()
     _bind_single_slot(obj, engine, src=0)
 
-    dummy_info = {"names": ["w"], "dtype_names": ["bfloat16"], "shapes": [[2, 2]], "ipc_handles": [{"u": ("f", ())}]}
+    dummy_info = {
+        "names": ["w"],
+        "dtype_names": ["bfloat16"],
+        "shapes": [[2, 2]],
+        "tensor_sizes": [8],
+        "ipc_handles": {"u": ("f", ())},
+    }
     with patch(
-        f"{MODULE_PATH}._build_ipc_update_info_from_named_tensors",
+        f"{MODULE_PATH}._build_packed_ipc_update_info",
         return_value=(dummy_info, []),
     ):
         _run_update(obj, chunks=_chunks(2))
@@ -306,13 +325,15 @@ def test_send_via_ipc_dispatches_update_weights_from_tensor_coordinator_multi_gp
         "names": ["w"],
         "dtype_names": ["bfloat16"],
         "shapes": [[2, 2]],
-        "ipc_handles": [{"uuid-gpu0": ("f", ())}],
+        "tensor_sizes": [8],
+        "ipc_handles": {"uuid-gpu0": ("f", ())},
     }
     dummy_info_1 = {
         "names": ["w"],
         "dtype_names": ["bfloat16"],
         "shapes": [[2, 2]],
-        "ipc_handles": [{"uuid-gpu1": ("f", ())}],
+        "tensor_sizes": [8],
+        "ipc_handles": {"uuid-gpu1": ("f", ())},
     }
 
     def fake_gather_object(payload, object_gather_list=None, dst=None, group=None):
@@ -322,7 +343,7 @@ def fake_gather_object(payload, object_gather_list=None, dst=None, group=None):
         gathered_payloads[1] = "payload1"
 
     with patch(
-        f"{MODULE_PATH}._build_ipc_update_info_from_named_tensors",
+        f"{MODULE_PATH}._build_packed_ipc_update_info",
         return_value=(dummy_info_0, []),
     ), patch(
         f"{MODULE_PATH}._serialize_ipc_update_info", return_value="payload0"
@@ -336,27 +357,88 @@ def fake_gather_object(payload, object_gather_list=None, dst=None, group=None):
     assert kwargs["names"] == dummy_info_0["names"]
     assert kwargs["dtype_names"] == dummy_info_0["dtype_names"]
     assert kwargs["shapes"] == dummy_info_0["shapes"]
-    assert len(kwargs["ipc_handles"]) == 1
-    assert set(kwargs["ipc_handles"][0].keys()) == {"uuid-gpu0", "uuid-gpu1"}
+    assert set(kwargs["ipc_handles"]) == {"uuid-gpu0", "uuid-gpu1"}
     assert kwargs["weight_version"] == "1"
 
 
 @pytest.mark.unit
-def test_merge_ipc_update_infos_combines_gpu_uuids(upw_vllm):
-    info0 = {
+def test_colocated_update_waits_in_bounded_batches(upw_vllm):
+    obj = _make_instance(upw_vllm)
+    engine = RecordingVLLMEngine()
+    _bind_single_slot(obj, engine, src=0)
+    next_ref = iter(range(5))
+
+    def fake_send(_hf_named_tensors):
+        index = next(next_ref)
+        return [f"update-{index}"], [torch.zeros(1)]
+
+    obj._send_hf_params = fake_send
+    update_batches = []
+
+    def record_get(refs):
+        if isinstance(refs, list) and refs and all(str(ref).startswith("update-") for ref in refs):
+            update_batches.append(refs)
+
+    with patch(f"{MODULE_PATH}._MAX_COLOCATED_UPDATES_INFLIGHT", 2), patch(
+        f"{MODULE_PATH}.ray.get", side_effect=record_get
+    ):
+        counters = _run_update(obj, chunks=_chunks(5))
+
+    assert [len(batch) for batch in update_batches] == [2, 2, 1]
+    assert counters["ipc_collect"] == 4
+
+
+@pytest.mark.unit
+def test_merge_packed_ipc_update_infos_combines_gpu_uuids(upw_vllm):
+    base = {
         "names": ["w"],
         "dtype_names": ["bfloat16"],
         "shapes": [[2, 2]],
-        "ipc_handles": [{"uuid-gpu0": ("f0", ())}],
+        "tensor_sizes": [8],
     }
-    info1 = {
-        "names": ["w"],
+    info0 = {**base, "ipc_handles": {"uuid-gpu0": ("f0", ())}}
+    info1 = {**base, "ipc_handles": {"uuid-gpu1": ("f1", ())}}
+
+    merged = upw_vllm._merge_ipc_update_infos([info0, info1])
+
+    assert set(merged["ipc_handles"]) == {"uuid-gpu0", "uuid-gpu1"}
+
+
+@pytest.mark.unit
+def test_merge_packed_ipc_update_infos_rejects_mismatched_metadata(upw_vllm):
+    info0 = {
+        "names": ["a"],
         "dtype_names": ["bfloat16"],
-        "shapes": [[2, 2]],
-        "ipc_handles": [{"uuid-gpu1": ("f1", ())}],
+        "shapes": [[2]],
+        "tensor_sizes": [4],
+        "ipc_handles": {"uuid-gpu0": ("f0", ())},
     }
-    merged = upw_vllm._merge_ipc_update_infos([info0, info1])
-    assert set(merged["ipc_handles"][0].keys()) == {"uuid-gpu0", "uuid-gpu1"}
+    info1 = {**info0, "names": ["b"], "ipc_handles": {"uuid-gpu1": ("f1", ())}}
+
+    with pytest.raises(ValueError, match="packed IPC metadata must match"):
+        upw_vllm._merge_ipc_update_infos([info0, info1])
+
+
+@pytest.mark.unit
+def test_build_packed_ipc_update_info_preserves_metadata_and_bytes(upw_vllm):
+    tensors = [("a", torch.tensor([1, 2], dtype=torch.int16)), ("b", torch.tensor([3.0]))]
+
+    with patch("torch.multiprocessing.reductions.reduce_tensor", return_value=(None, ("rebuild", ()))), patch(
+        "torch.cuda.current_device", return_value=0
+    ), patch("torch.cuda.get_device_properties", return_value=MagicMock(uuid="uuid-gpu0")):
+        update_info, packed = upw_vllm._build_packed_ipc_update_info(tensors)
+
+    assert update_info["names"] == ["a", "b"]
+    assert update_info["tensor_sizes"] == [4, 4]
+    assert update_info["ipc_handles"] == {"uuid-gpu0": ("rebuild", ())}
+    assert torch.equal(packed, torch.cat([tensor.view(torch.uint8) for _, tensor in tensors]))
+
+
+@pytest.mark.unit
+def test_colocated_source_has_no_nonpacked_path(upw_vllm):
+    source = inspect.getsource(upw_vllm)
+    assert "vllm_weight_sync_packed" not in source
+    assert "_build_ipc_update_info_from_named_tensors" not in source
 
 
 @pytest.mark.unit
@@ -402,9 +484,9 @@ def test_non_leader_skips_start_finish_and_merged_rpc(upw_vllm):
     # slot leader is rank 0; we drive update_weights as rank 1 (non-leader).
     _bind_single_slot(obj, engine, src=0)
 
-    dummy_info = {"names": [], "dtype_names": [], "shapes": [], "ipc_handles": []}
+    dummy_info = {"names": [], "dtype_names": [], "shapes": [], "tensor_sizes": [], "ipc_handles": {}}
     with patch(
-        f"{MODULE_PATH}._build_ipc_update_info_from_named_tensors",
+        f"{MODULE_PATH}._build_packed_ipc_update_info",
         return_value=(dummy_info, []),
     ), patch(
         f"{MODULE_PATH}._serialize_ipc_update_info", return_value="payload"
diff --git a/tests/utils/test_vllm_arguments.py b/tests/utils/test_vllm_arguments.py
index 5c861382..314bbc4b 100644
--- a/tests/utils/test_vllm_arguments.py
+++ b/tests/utils/test_vllm_arguments.py
@@ -176,7 +176,8 @@ def test_add_vllm_arguments_prefixes_regular_engine_flags(args_mod, monkeypatch)
     flags = {s for a in parser._actions for s in a.option_strings}
     assert "--vllm-server-concurrency" in flags
     assert "--vllm-tool-call-parser" in flags
-    assert "--vllm-weight-sync-packed" in flags
+    assert "--vllm-weight-sync-packed" not in flags
+    assert "--no-vllm-weight-sync-packed" not in flags
 
 
 @pytest.mark.unit
diff --git a/tests/utils/test_vllm_engine.py b/tests/utils/test_vllm_engine.py
index 9260ad71..98f0dde4 100644
--- a/tests/utils/test_vllm_engine.py
+++ b/tests/utils/test_vllm_engine.py
@@ -381,10 +381,11 @@ def fake_post(endpoint: str, payload: dict):
     assert vllm_engine._weight_version is None
 
     vllm_engine.update_weights_from_tensor(
-        names=["layer.0.weight"],
-        dtype_names=["float32"],
-        shapes=[[2, 2]],
-        ipc_handles=[{"uuid-gpu0": ("rebuild_fn", (1, 2, 3))}],
+        names=["a", "b"],
+        dtype_names=["bfloat16", "float32"],
+        shapes=[[2], [1]],
+        ipc_handles={"uuid-gpu0": ("rebuild_fn", (1, 2, 3))},
+        tensor_sizes=[4, 4],
         weight_version="42",
     )
 
@@ -393,8 +394,10 @@ def fake_post(endpoint: str, payload: dict):
     # ipc_handles got cloudpickle'd into ipc_handles_pickled
     assert "ipc_handles" not in sent
     assert isinstance(sent["ipc_handles_pickled"], str)
-    assert sent["names"] == ["layer.0.weight"]
-    assert sent["shapes"] == [[2, 2]]
+    assert sent["names"] == ["a", "b"]
+    assert sent["shapes"] == [[2], [1]]
+    assert sent["tensor_sizes"] == [4, 4]
+    assert sent["packed"] is True
     # version recorded after POST success
     assert vllm_engine._weight_version == "42"
 
@@ -411,7 +414,7 @@ def fake_post_fail(endpoint: str, payload: dict) -> dict:
     vllm_engine._weight_version = "old"
     with pytest.raises(RuntimeError, match="simulated POST failure"):
         vllm_engine.update_weights_from_tensor(
-            names=[], dtype_names=[], shapes=[], ipc_handles=[], weight_version="new"
+            names=[], dtype_names=[], shapes=[], ipc_handles={}, tensor_sizes=[], weight_version="new"
         )
     assert vllm_engine._weight_version == "old"
 
@@ -456,9 +459,7 @@ def fake_make_request(endpoint: str, payload: dict) -> dict:
         names,
         dtypes,
         shapes,
-        group_name="vime-pp_0",
         weight_version="7",
-        packed=True,
     )
 
     assert len(calls) == 1
diff --git a/vime/backends/megatron_utils/update_weight/common.py b/vime/backends/megatron_utils/update_weight/common.py
index 89555aca..6c9f341d 100644
--- a/vime/backends/megatron_utils/update_weight/common.py
+++ b/vime/backends/megatron_utils/update_weight/common.py
@@ -31,6 +31,9 @@ def all_gather_param(name: str, param: torch.nn.Parameter) -> torch.Tensor:
         tp_size = mpu.get_tensor_model_parallel_world_size()
         tp_group = mpu.get_tensor_model_parallel_group()
 
+    if tp_size == 1:
+        return param.data
+
     param_partitions = [torch.empty_like(param.data) for _ in range(tp_size)]
     dist.all_gather(param_partitions, param.data, group=tp_group)
     partition_dim = param.partition_dim
@@ -54,65 +57,62 @@ def all_gather_params_async(
     param_infos_and_params: list[tuple[ParamInfo, torch.Tensor]],
 ) -> list[torch.Tensor]:
     """
-    Parallel TP all-gather for multiple params. Loop 1: for each TP param, allocate buffers +
-    dist.all_gather(async_op=True) on expert-TP/regular-TP group (skip expert_bias/non-TP/duplicated).
-    Loop 2: wait all NCCL handles (enables overlap). Loop 3: concat partitions + apply GLU rechunk/MoE dim fix.
+    Coalesce TP-sharded params by process group and dtype, then reconstruct
+    their original layouts after one all-gather per bucket.
     """
-    # Phase 1: Start all async all_gather operations
-    gather_tasks = []
-    handles = []
+    gathered_params: list[torch.Tensor | None] = [None] * len(param_infos_and_params)
+    grouped_params: dict[tuple[bool, torch.dtype], list[tuple[int, ParamInfo, torch.Tensor]]] = {}
 
-    for info, param in param_infos_and_params:
-        # Prepare async all_gather
+    for index, (info, param) in enumerate(param_infos_and_params):
         if "expert_bias" in info.name:
-            gather_tasks.append((info, param, None, None, None))
-            handles.append(None)
+            gathered_params[index] = param
         elif not param.tensor_model_parallel or getattr(param, "parallel_mode", None) == "duplicated":
-            gather_tasks.append((info, param.data, None, None, None))
-            handles.append(None)
+            gathered_params[index] = param.data
         else:
-            # Start async all_gather
-            if ".experts." in info.name:
-                tp_size = mpu.get_expert_tensor_parallel_world_size()
-                tp_group = mpu.get_expert_tensor_parallel_group()
+            is_expert = ".experts." in info.name
+            tp_size = (
+                mpu.get_expert_tensor_parallel_world_size()
+                if is_expert
+                else mpu.get_tensor_model_parallel_world_size()
+            )
+            if tp_size == 1:
+                gathered_params[index] = param.data
             else:
-                tp_size = mpu.get_tensor_model_parallel_world_size()
-                tp_group = mpu.get_tensor_model_parallel_group()
-
-            param_partitions = [torch.empty_like(param.data) for _ in range(tp_size)]
-            handle = dist.all_gather(param_partitions, param.data, group=tp_group, async_op=True)
-            gather_tasks.append((info, None, handle, param_partitions, param.partition_dim))
-            handles.append(handle)
-
-    # Phase 2: Wait for ALL async operations to complete at once
-    # This ensures maximum parallelism by not blocking on individual operations
-    for handle in handles:
-        if handle is not None:
-            handle.wait()
-
-    # Phase 3: Process all results after all communications are done
-    gathered_params = []
-    for info, direct_param, handle, param_partitions, partition_dim in gather_tasks:
-        if handle is None:
-            # No all_gather needed
-            param = direct_param
-        else:
-            # Process the gathered partitions (same logic as original all_gather_param)
-            assert partition_dim is not None, "partition_stride != 1 is not supported"
-            # TODO: here we did an extra copy during concat, maybe merge this with convert_to_hf is better?
-            # TODO: check only GLU is used.
+                grouped_params.setdefault((is_expert, param.dtype), []).append((index, info, param))
+
+    gather_tasks = []
+    for (is_expert, _dtype), entries in grouped_params.items():
+        tp_size = (
+            mpu.get_expert_tensor_parallel_world_size() if is_expert else mpu.get_tensor_model_parallel_world_size()
+        )
+        tp_group = mpu.get_expert_tensor_parallel_group() if is_expert else mpu.get_tensor_model_parallel_group()
+        local_flat = torch.cat([param.data.reshape(-1) for _, _, param in entries])
+        gathered_flat = torch.empty(tp_size * local_flat.numel(), dtype=local_flat.dtype, device=local_flat.device)
+        handle = dist.all_gather_into_tensor(gathered_flat, local_flat, group=tp_group, async_op=True)
+        gather_tasks.append((handle, gathered_flat, local_flat.numel(), entries, tp_size))
+
+    for handle, gathered_flat, rank_stride, entries, tp_size in gather_tasks:
+        handle.wait()
+        offset = 0
+        for index, info, param in entries:
+            numel = param.numel()
+            param_partitions = [
+                gathered_flat.narrow(0, rank * rank_stride + offset, numel).view_as(param) for rank in range(tp_size)
+            ]
+            partition_dim = param.partition_dim
+            assert param.partition_stride == 1 or (
+                param.partition_stride == 2 and "linear_fc1" in info.name
+            ), "partition_stride != 1 is not supported"
             if "linear_fc1.weight" in info.name or "linear_fc1.bias" in info.name:
                 param_partitions = [p.chunk(2, dim=0) for p in param_partitions]
                 param_partitions = [p[0] for p in param_partitions] + [p[1] for p in param_partitions]
-            # this is bug in megatron's grouped moe.
-            if "linear_fc2.weight" in info.name:
-                if partition_dim == 0:
-                    partition_dim = 1
-            param = torch.cat(param_partitions, dim=partition_dim)
-
-        gathered_params.append(param)
+            if "linear_fc2.weight" in info.name and partition_dim == 0:
+                partition_dim = 1
+            gathered_params[index] = torch.cat(param_partitions, dim=partition_dim)
+            offset += numel
 
-    return gathered_params
+    assert all(param is not None for param in gathered_params)
+    return [param for param in gathered_params if param is not None]
 
 
 def named_params_and_buffers(
diff --git a/vime/backends/megatron_utils/update_weight/hf_weight_iterator_direct.py b/vime/backends/megatron_utils/update_weight/hf_weight_iterator_direct.py
index d345adde..bed16a71 100644
--- a/vime/backends/megatron_utils/update_weight/hf_weight_iterator_direct.py
+++ b/vime/backends/megatron_utils/update_weight/hf_weight_iterator_direct.py
@@ -19,6 +19,7 @@ class HfWeightIteratorDirect(HfWeightIteratorBase):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.megatron_local_param_info_buckets = _get_megatron_local_param_info_buckets(self.args, self.model)
+        self.ep_broadcast_src_rank_map = _get_ep_broadcast_src_rank_map()
 
     def get_hf_weight_chunks(self, megatron_local_weights, progress_desc: str = "Update weights"):
         rank = dist.get_rank()
@@ -26,7 +27,12 @@ def get_hf_weight_chunks(self, megatron_local_weights, progress_desc: str = "Upd
         for megatron_local_param_infos in tqdm(
             self.megatron_local_param_info_buckets, disable=rank != 0, desc=progress_desc
         ):
-            megatron_full_params = _get_megatron_full_params(megatron_local_param_infos, megatron_local_weights)
+            megatron_full_params = _get_megatron_full_params(
+                megatron_local_param_infos,
+                megatron_local_weights,
+                self.args.update_weight_buffer_size,
+                self.ep_broadcast_src_rank_map,
+            )
             hf_named_tensors = self._convert_to_hf_named_tensors(megatron_full_params, megatron_local_param_infos)
             yield hf_named_tensors
             del megatron_full_params
@@ -43,10 +49,11 @@ def _convert_to_hf_named_tensors(self, megatron_full_params: Sequence[torch.Tens
 def _get_megatron_full_params(
     megatron_local_param_infos: Sequence[ParamInfo],
     megatron_local_weights,
+    broadcast_buffer_size: int,
+    ep_broadcast_src_rank_map: dict[int, int],
 ) -> Sequence[torch.Tensor]:
     pp_size = mpu.get_pipeline_model_parallel_world_size()
     ep_size = mpu.get_expert_model_parallel_world_size()
-    rank = dist.get_rank()
     # init params:
     params = []
     for info in megatron_local_param_infos:
@@ -59,8 +66,6 @@ def _get_megatron_full_params(
             )
         else:
             params.append(torch.empty(info.shape, dtype=info.dtype, device=torch.cuda.current_device()))
-    torch.cuda.synchronize()
-
     # broadcast params across pp ranks
     if pp_size > 1:
         handles = []
@@ -76,21 +81,7 @@ def _get_megatron_full_params(
 
     # broadcast params across ep ranks
     if ep_size > 1:
-        handles = []
-        for info, param in zip(megatron_local_param_infos, params, strict=False):
-            if ".experts." in info.name:
-                src_rank = (
-                    info.src_rank
-                    if info.src_rank in dist.get_process_group_ranks(mpu.get_expert_model_parallel_group())
-                    else rank
-                )
-                handles.append(
-                    torch.distributed.broadcast(
-                        param, src=src_rank, group=mpu.get_expert_model_parallel_group(), async_op=True
-                    )
-                )
-        for handle in handles:
-            handle.wait()
+        _broadcast_expert_params(megatron_local_param_infos, params, broadcast_buffer_size, ep_broadcast_src_rank_map)
 
     # Set tp attrs for all params
     for info, param in zip(megatron_local_param_infos, params, strict=False):
@@ -103,6 +94,41 @@ def _get_megatron_full_params(
     return gathered_params
 
 
+def _broadcast_expert_params(
+    param_infos: Sequence[ParamInfo],
+    params: Sequence[torch.Tensor],
+    buffer_size: int,
+    src_rank_map: dict[int, int],
+) -> None:
+    ep_group = mpu.get_expert_model_parallel_group()
+    params_by_src: dict[int, list[torch.Tensor]] = {}
+    for info, param in zip(param_infos, params, strict=False):
+        if ".experts." not in info.name:
+            continue
+        params_by_src.setdefault(src_rank_map[info.src_rank], []).append(param)
+
+    for src_rank, expert_params in params_by_src.items():
+        dist._broadcast_coalesced(ep_group, expert_params, buffer_size, src=src_rank)
+
+
+def _get_ep_broadcast_src_rank_map() -> dict[int, int]:
+    ep_group = mpu.get_expert_model_parallel_group()
+    ep_size = mpu.get_expert_model_parallel_world_size()
+    if ep_size == 1:
+        return {dist.get_rank(): 0}
+
+    pp_group_ranks = dist.get_process_group_ranks(mpu.get_pipeline_model_parallel_group())
+    pp_groups: list[list[int] | None] = [None] * ep_size
+    dist.all_gather_object(pp_groups, pp_group_ranks, group=ep_group)
+
+    src_rank_map = {}
+    for ep_rank, pp_group in enumerate(pp_groups):
+        assert pp_group is not None
+        for global_rank in pp_group:
+            src_rank_map[global_rank] = ep_rank
+    return src_rank_map
+
+
 def _get_megatron_local_param_info_buckets(args: Namespace, model: Sequence[torch.nn.Module]) -> list[list[ParamInfo]]:
     """
     Partition params into buckets ≤ update_weight_buffer_size (with TP replication).
diff --git a/vime/backends/megatron_utils/update_weight/update_weight_from_distributed.py b/vime/backends/megatron_utils/update_weight/update_weight_from_distributed.py
index 93af6ba5..67da8937 100644
--- a/vime/backends/megatron_utils/update_weight/update_weight_from_distributed.py
+++ b/vime/backends/megatron_utils/update_weight/update_weight_from_distributed.py
@@ -231,27 +231,24 @@ def _send_weights(self, pbar: tqdm | None) -> None:
         yields broadcast-ready chunks (bucketing happens internally).
         """
         if self._hf_weight_iterator is not None:
-            use_vllm_packed = self._use_vllm_packed()
-            self._sync_bridge_weights_to_rollout_engines(pbar, use_vllm_packed=use_vllm_packed)
+            self._sync_bridge_weights_to_rollout_engines(pbar)
             return
 
         is_active_stage = self._is_active_weight_sync_pp_stage()
-        use_vllm_packed = False
         if is_active_stage:
-            use_vllm_packed = self._use_vllm_packed()
-            if use_vllm_packed and self._is_pp_src_rank:
+            if self._is_pp_src_rank:
                 logger.info("Using vLLM packed weight sync (bucketed; metadata + trainer_send_weights per bucket)")
 
             for hf_chunk in self._iter_non_expert_chunks():
-                self._update_bucket_weights_from_distributed(hf_chunk, pbar=pbar, packed=use_vllm_packed)
+                self._update_bucket_weights_from_distributed(hf_chunk, pbar=pbar)
         dist.barrier(group=get_gloo_group())
 
-        if is_active_stage and not use_vllm_packed:
+        if is_active_stage:
             for hf_chunk in self._iter_expert_chunks():
-                self._update_bucket_weights_from_distributed(hf_chunk, pbar=pbar, packed=False)
+                self._update_bucket_weights_from_distributed(hf_chunk, pbar=pbar)
         dist.barrier(group=get_gloo_group())
 
-    def _sync_bridge_weights_to_rollout_engines(self, pbar: tqdm | None, *, use_vllm_packed: bool) -> None:
+    def _sync_bridge_weights_to_rollout_engines(self, pbar: tqdm | None) -> None:
         """
         Export HF weights through Megatron-Bridge, then send each exported chunk
         over the same NCCL non-colocate path used by the raw converter.
@@ -263,20 +260,10 @@ def _sync_bridge_weights_to_rollout_engines(self, pbar: tqdm | None, *, use_vllm
         for hf_named_tensors in self._hf_weight_iterator.get_hf_weight_chunks(megatron_local_weights):
             if self._is_pp_src_rank:
                 hf_named_tensors = list(hf_named_tensors)
-                self._update_bucket_weights_from_distributed(hf_named_tensors, pbar=pbar, packed=use_vllm_packed)
+                self._update_bucket_weights_from_distributed(hf_named_tensors, pbar=pbar)
 
         dist.barrier(group=get_gloo_group())
 
-    def _use_vllm_packed(self) -> bool:
-        """Use vLLM packed weight transfer (one-shot metadata + trainer_send_weights)."""
-        if not getattr(self.args, "vllm_weight_sync_packed", True):
-            return False
-        if any(".experts." in name for name, _ in named_params_and_buffers(self.args, self.model)):
-            return False
-        if self.quantization_config and self.quantization_config.get("quant_method") == "compressed-tensors":
-            return False
-        return True
-
     def _iter_non_expert_chunks(self) -> Iterator[list[tuple[str, torch.Tensor]]]:
         """
         Yield broadcast-sized HF chunks of non-expert params: TP all-gather +
@@ -307,28 +294,32 @@ def _iter_expert_chunks(
         params: Iterator[tuple[str, torch.Tensor]] | None = None,
     ) -> Iterator[list[tuple[str, torch.Tensor]]]:
         """
-        Yield one HF chunk per EP-weighted batch of expert params: TP gather +
-        buffer until threshold, then EP gather + HF convert.
+        Keep each expert layer together, then bucket complete layers before
+        EP gather and HF conversion.
         """
         if params is None:
             params = ((n, p) for n, p in named_params_and_buffers(self.args, self.model) if ".experts." in n)
 
+        expert_groups: dict[str, list[tuple[str, torch.Tensor]]] = {}
+        for name, param in params:
+            layer_name = name.split(".experts.", 1)[0]
+            expert_groups.setdefault(layer_name, []).append((name, param))
+
         buffer_size = 0
         batch: list[tuple[str, torch.Tensor]] = []
-        for name, param in params:
-            param = all_gather_param(name, param)
-            param_size = param.numel() * param.element_size()
-            if (
-                buffer_size + param_size
-            ) * mpu.get_expert_model_parallel_world_size() > self.args.update_weight_buffer_size:
+        ep_size = mpu.get_expert_model_parallel_world_size()
+        for expert_params in expert_groups.values():
+            gathered_params = [(name, all_gather_param(name, param)) for name, param in expert_params]
+            group_size = sum(param.numel() * param.element_size() for _, param in gathered_params)
+            if batch and (buffer_size + group_size) * ep_size > self.args.update_weight_buffer_size:
                 hf_chunk = self._ep_gather_and_convert(batch)
                 if hf_chunk:
                     yield hf_chunk
                 batch = []
                 buffer_size = 0
 
-            batch.append((name, param))
-            buffer_size += param_size
+            batch.extend(gathered_params)
+            buffer_size += group_size
 
         if batch:
             hf_chunk = self._ep_gather_and_convert(batch)
@@ -340,6 +331,14 @@ def _ep_gather_and_convert(self, named_tensors: list[tuple[str, torch.Tensor]])
         EP all-gather a buffered batch + HF convert on PP source. Returns HF tensors on
         PP source, [] elsewhere. Clears ``named_tensors``.
         """
+        if mpu.get_expert_model_parallel_world_size() == 1:
+            converted = []
+            if self._is_pp_src_rank:
+                for name, param in named_tensors:
+                    converted.extend(convert_to_hf(self.args, self.model_name, name, param, self.quantization_config))
+            named_tensors.clear()
+            return converted
+
         names = [name for name, _ in named_tensors]
         all_names = [None] * mpu.get_expert_model_parallel_world_size()
         dist.all_gather_object(all_names, names, group=mpu.get_expert_model_parallel_group())
@@ -376,8 +375,6 @@ def _update_bucket_weights_from_distributed(
         self,
         converted_named_tensors: list[tuple[str, torch.Tensor]],
         pbar: tqdm | None = None,
-        *,
-        packed: bool = False,
     ) -> None:
         """
         Lock → broadcast → clear → unlock → pbar++. Lock prevents NCCL deadlock.
@@ -387,12 +384,10 @@ def _update_bucket_weights_from_distributed(
             time.sleep(0.1)
 
         refs = update_weights_from_distributed(
-            self._group_name,
             self._model_update_groups,
             self.weight_version,
             self.rollout_engines,
             converted_named_tensors,
-            packed=packed,
         )
 
         ray.get(refs)
@@ -493,13 +488,10 @@ def disconnect_rollout_engines_from_distributed(
 
 
 def update_weights_from_distributed(
-    group_name: str,
     group: Any,
     weight_version: int,
     rollout_engines: Sequence[ActorHandle],
     converted_named_tensors: Sequence[tuple[str, torch.Tensor]],
-    *,
-    packed: bool = False,
 ) -> list[ObjectRef]:
     """
     Send metadata (Ray), broadcast tensors (NCCL rank 0 → engines).
@@ -512,9 +504,7 @@ def update_weights_from_distributed(
             names=[name for name, _ in converted_named_tensors],
             dtypes=[param.dtype for _, param in converted_named_tensors],
             shapes=[param.shape for _, param in converted_named_tensors],
-            group_name=group_name,
             weight_version=str(weight_version),
-            packed=packed,
         )
         for engine in rollout_engines
     ]
@@ -525,7 +515,7 @@ def update_weights_from_distributed(
     )
     NCCLWeightTransferEngine.trainer_send_weights(
         named_gpu_iter,
-        NCCLTrainerSendWeightsArgs(group=group, packed=packed),
+        NCCLTrainerSendWeightsArgs(group=group, packed=True),
     )
 
     return refs
diff --git a/vime/backends/megatron_utils/update_weight/update_weight_from_tensor.py b/vime/backends/megatron_utils/update_weight/update_weight_from_tensor.py
index 25575e5a..5e352025 100644
--- a/vime/backends/megatron_utils/update_weight/update_weight_from_tensor.py
+++ b/vime/backends/megatron_utils/update_weight/update_weight_from_tensor.py
@@ -35,60 +35,41 @@
     update_weights_from_distributed,
 )
 
+_MAX_COLOCATED_UPDATES_INFLIGHT = 4
 
-def _current_gpu_uuid() -> str:
-    device_index = torch.cuda.current_device()
-    props = torch.cuda.get_device_properties(device_index)
-    return str(props.uuid)
 
-
-def _build_ipc_update_info_from_named_tensors(
+def _build_packed_ipc_update_info(
     named_tensors: Iterable[tuple[str, torch.Tensor]],
-) -> tuple[dict[str, list], list[torch.Tensor]]:
-    """Build vLLM IPC ``update_info`` payload from tensors on this rank's GPU.
-
-    Each handle is keyed by the physical GPU UUID of the producing rank rather
-    than by a local device index. The coordinator gathers all ranks' dicts and
-    merges them; the receiver looks up its own UUID to pick the matching handle,
-    then vLLM unconditionally overwrites ``args[6]`` (device_index) with its own
-    local index before ``rebuild_cuda_tensor``. This UUID-keyed routing makes
-    the path correct under any ``CUDA_VISIBLE_DEVICES`` ordering without
-    relying on a torch reductions monkey-patch.
-
-    Return the contiguous tensor refs alongside the payload. ``reduce_tensor``
-    only exports CUDA IPC metadata, so the producer storage must stay alive
-    until the receiver opens the handle.
-    """
+) -> tuple[dict[str, Any], torch.Tensor]:
     from torch.multiprocessing.reductions import reduce_tensor
 
-    names: list[str] = []
-    dtype_names: list[str] = []
-    shapes: list[list[int]] = []
-    ipc_handles: list[dict[str, tuple]] = []
-    weight_refs: list[torch.Tensor] = []
-    gpu_uuid = _current_gpu_uuid()
-
+    names, dtype_names, shapes, tensor_sizes, byte_tensors = [], [], [], [], []
     for name, tensor in named_tensors:
         names.append(name)
         dtype_names.append(str(tensor.dtype).split(".")[-1])
         shapes.append(list(tensor.shape))
-        weight = tensor.detach().contiguous()
-        weight_refs.append(weight)
-        _, ipc_args = reduce_tensor(weight)
-        ipc_handles.append({gpu_uuid: ipc_args})
-
+        byte_tensor = tensor.detach().contiguous().view(torch.uint8).flatten()
+        tensor_sizes.append(byte_tensor.numel())
+        byte_tensors.append(byte_tensor)
+    if not byte_tensors:
+        raise ValueError("cannot build an empty packed IPC update")
+
+    packed_tensor = torch.cat(byte_tensors)
+    _, ipc_args = reduce_tensor(packed_tensor)
+    gpu_uuid = str(torch.cuda.get_device_properties(torch.cuda.current_device()).uuid)
     return (
         {
             "names": names,
             "dtype_names": dtype_names,
             "shapes": shapes,
-            "ipc_handles": ipc_handles,
+            "tensor_sizes": tensor_sizes,
+            "ipc_handles": {gpu_uuid: ipc_args},
         },
-        weight_refs,
+        packed_tensor,
     )
 
 
-def _serialize_ipc_update_info(info: dict[str, list]) -> str:
+def _serialize_ipc_update_info(info: dict[str, Any]) -> str:
     """Pickle IPC handles for cross-rank gather (Gloo ``all_gather_object`` cannot carry them)."""
     import base64
 
@@ -97,7 +78,7 @@ def _serialize_ipc_update_info(info: dict[str, list]) -> str:
     return base64.b64encode(cloudpickle.dumps(info)).decode("ascii")
 
 
-def _deserialize_ipc_update_info(payload: str) -> dict[str, list]:
+def _deserialize_ipc_update_info(payload: str) -> dict[str, Any]:
     import base64
 
     import cloudpickle
@@ -105,33 +86,21 @@ def _deserialize_ipc_update_info(payload: str) -> dict[str, list]:
     return cloudpickle.loads(base64.b64decode(payload.encode("ascii")))
 
 
-def _merge_ipc_update_infos(infos: Sequence[dict[str, list]]) -> dict[str, list]:
-    """Merge per-rank IPC payloads, including empty or uneven expert buckets."""
+def _merge_ipc_update_infos(infos: Sequence[dict[str, Any]]) -> dict[str, Any]:
+    """Merge the per-rank handles for one packed IPC update."""
     if not infos:
         raise ValueError("no IPC update_info payloads to merge")
 
-    merged: dict[str, tuple[str, list[int], dict[str, tuple]]] = {}
+    metadata_keys = ("names", "dtype_names", "shapes", "tensor_sizes")
+    base = infos[0]
+    if "tensor_sizes" not in base or any(
+        "tensor_sizes" not in info or any(info[key] != base[key] for key in metadata_keys) for info in infos[1:]
+    ):
+        raise ValueError("packed IPC metadata must match across all ranks in a slot")
+    handles = {}
     for info in infos:
-        for name, dtype_name, shape, handles in zip(
-            info["names"], info["dtype_names"], info["shapes"], info["ipc_handles"], strict=True
-        ):
-            if name not in merged:
-                merged[name] = (dtype_name, shape, dict(handles))
-                continue
-            merged_dtype, merged_shape, merged_handles = merged[name]
-            if dtype_name != merged_dtype or shape != merged_shape:
-                raise ValueError(
-                    f"inconsistent IPC metadata for {name}: "
-                    f"{(merged_dtype, merged_shape)} != {(dtype_name, shape)}"
-                )
-            merged_handles.update(handles)
-
-    return {
-        "names": list(merged),
-        "dtype_names": [metadata[0] for metadata in merged.values()],
-        "shapes": [metadata[1] for metadata in merged.values()],
-        "ipc_handles": [metadata[2] for metadata in merged.values()],
-    }
+        handles.update(info["ipc_handles"])
+    return {**base, "ipc_handles": handles}
 
 
 class UpdateWeightFromTensor:
@@ -297,15 +266,7 @@ def update_weights(self) -> None:
         dist.barrier(group=get_gloo_group())
 
         megatron_local_weights = self.weights_getter()
-
-        for hf_named_tensors in self._hf_weight_iterator.get_hf_weight_chunks(megatron_local_weights):
-            refs, long_lived_tensors = self._send_hf_params(hf_named_tensors)
-            ray.get(refs)
-            # Free GPU tensors so the caching allocator can reuse the blocks,
-            # then release CUDA IPC cache entries whose consumers (vLLM engines)
-            # have already closed their IPC handles.
-            del long_lived_tensors, hf_named_tensors
-            torch.cuda.ipc_collect()
+        self._send_weight_chunks(megatron_local_weights)
 
         dist.barrier(group=get_gloo_group())
         # After the barrier all engines have returned, so every rank's last-chunk
@@ -326,11 +287,7 @@ def update_weights(self) -> None:
                 ray.get(self._ipc_engine.start_draft_weight_update.remote())
             dist.barrier(group=get_gloo_group())
 
-            for hf_named_tensors in self._hf_weight_iterator.get_hf_weight_chunks(megatron_local_weights):
-                refs, long_lived_tensors = self._send_hf_params(hf_named_tensors)
-                ray.get(refs)
-                del long_lived_tensors, hf_named_tensors
-                torch.cuda.ipc_collect()
+            self._send_weight_chunks(megatron_local_weights)
 
             dist.barrier(group=get_gloo_group())
             torch.cuda.ipc_collect()
@@ -349,6 +306,25 @@ def update_weights(self) -> None:
             ray.get([engine.continue_generation.remote() for engine in self.rollout_engines])
         dist.barrier(group=get_gloo_group())
 
+    def _send_weight_chunks(self, megatron_local_weights) -> None:
+        max_inflight = 1 if self.use_distribute else _MAX_COLOCATED_UPDATES_INFLIGHT
+        pending = []
+        for hf_named_tensors in self._hf_weight_iterator.get_hf_weight_chunks(megatron_local_weights):
+            refs, weight_refs = self._send_hf_params(hf_named_tensors)
+            pending.append((refs, weight_refs))
+            if len(pending) >= max_inflight:
+                self._drain_ipc_updates(pending)
+        self._drain_ipc_updates(pending)
+
+    def _drain_ipc_updates(self, pending) -> None:
+        if not pending:
+            return
+        ray.get([ref for refs, _ in pending for ref in refs])
+        if self._ipc_gather_group is not None:
+            dist.barrier(group=self._ipc_gather_group)
+        pending.clear()
+        torch.cuda.ipc_collect()
+
     def _send_hf_params(self, hf_named_tensors) -> tuple[list[ObjectRef], Any]:
         all_refs = []
 
@@ -363,12 +339,10 @@ def _send_hf_params(self, hf_named_tensors) -> tuple[list[ObjectRef], Any]:
 
         if self.use_distribute and self._is_distributed_src_rank:
             refs_distributed = update_weights_from_distributed(
-                self._group_name,
                 self._model_update_groups,
                 self.weight_version,
                 self.distributed_rollout_engines,
                 hf_named_tensors,
-                packed=False,
             )
             if refs_distributed:
                 all_refs.extend(refs_distributed)
@@ -389,13 +363,13 @@ def _send_to_colocated_engine(
     if ipc_gather_group is None:
         return [], None
 
+    local_info, weight_ref = _build_packed_ipc_update_info(hf_named_tensors)
+
     slot_size = dist.get_world_size(ipc_gather_group)
     if slot_size <= 1:
-        local_info, weight_refs = _build_ipc_update_info_from_named_tensors(hf_named_tensors)
         ref = ipc_engine.update_weights_from_tensor.remote(**local_info, weight_version=str(weight_version))
-        return [ref], weight_refs
+        return [ref], weight_ref
 
-    local_info, weight_refs = _build_ipc_update_info_from_named_tensors(hf_named_tensors)
     payload = _serialize_ipc_update_info(local_info)
 
     gathered_payloads = [None] * slot_size if dist.get_rank() == ipc_gather_src else None
@@ -409,4 +383,4 @@ def _send_to_colocated_engine(
         merged = _merge_ipc_update_infos(slot_infos)
         refs.append(ipc_engine.update_weights_from_tensor.remote(**merged, weight_version=str(weight_version)))
 
-    return refs, weight_refs
+    return refs, weight_ref
diff --git a/vime/backends/vllm_utils/arguments.py b/vime/backends/vllm_utils/arguments.py
index cab62386..f23a8b94 100644
--- a/vime/backends/vllm_utils/arguments.py
+++ b/vime/backends/vllm_utils/arguments.py
@@ -43,19 +43,6 @@ def add_vllm_arguments(parser):
             "AND exports ``VLLM_BATCH_INVARIANT=1`` to the vLLM subprocess."
         ),
     )
-    _vllm_packed = parser.add_mutually_exclusive_group()
-    _vllm_packed.add_argument(
-        "--vllm-weight-sync-packed",
-        dest="vllm_weight_sync_packed",
-        action="store_true",
-    )
-    _vllm_packed.add_argument(
-        "--no-vllm-weight-sync-packed",
-        dest="vllm_weight_sync_packed",
-        action="store_false",
-    )
-    parser.set_defaults(vllm_weight_sync_packed=True)
-
     # Monkey-patch parser to prefix all engine flags with --vllm- / vllm_
     old_add_argument = parser.add_argument
     old_add_argument_group = parser.add_argument_group
diff --git a/vime/backends/vllm_utils/vllm_engine.py b/vime/backends/vllm_utils/vllm_engine.py
index 1c3fdba0..4dc2bb1d 100644
--- a/vime/backends/vllm_utils/vllm_engine.py
+++ b/vime/backends/vllm_utils/vllm_engine.py
@@ -275,13 +275,19 @@ def update_weights_from_tensor(
         names: list[str],
         dtype_names: list[str],
         shapes: list[list[int]],
-        ipc_handles: list[dict] | None = None,
+        ipc_handles: dict[str, tuple],
+        tensor_sizes: list[int],
         weight_version: str,
         flush_cache: bool = False,
     ):
-        payload: dict = {"names": names, "dtype_names": dtype_names, "shapes": shapes}
-        if ipc_handles is not None:
-            payload["ipc_handles_pickled"] = base64.b64encode(cloudpickle.dumps(ipc_handles)).decode("utf-8")
+        payload: dict = {
+            "names": names,
+            "dtype_names": dtype_names,
+            "shapes": shapes,
+            "ipc_handles_pickled": base64.b64encode(cloudpickle.dumps(ipc_handles)).decode("utf-8"),
+            "tensor_sizes": tensor_sizes,
+            "packed": True,
+        }
         if flush_cache:
             self.flush_cache()
         result = self._make_request("update_weights", {"update_info": payload})
@@ -424,13 +430,10 @@ def update_weights_from_distributed(
         names,
         dtypes,
         shapes,
-        group_name,
         *,
         flush_cache=False,
         weight_version: str,
-        packed: bool = True,
     ):
-        del group_name
         if flush_cache:
             self.flush_cache()
         dtype_names = [str(d).replace("torch.", "") for d in dtypes]
@@ -438,7 +441,7 @@ def update_weights_from_distributed(
             "names": names,
             "dtype_names": dtype_names,
             "shapes": [list(s) for s in shapes],
-            "packed": bool(packed),
+            "packed": True,
         }
         result = self._make_request("update_weights", {"update_info": update_info})
         self._weight_version = str(weight_version)