[CI vehicle, do not merge] config: raw-input ServerArgs follow-up series (#36618-#36622)

python/sglang/benchmark/one_batch.py

@@ -548,7 +548,7 @@ def _maybe_prepare_mlp_sync_batch(batch: ScheduleBatch, model_runner):
         prepare_mlp_sync_batch_raw(
             batch,
             model_runner=model_runner,
-            dp_size=get_parallel().config.dp_size,
+            dp_size=get_parallel().dp_size,
             attn_tp_size=get_parallel().attn_tp_size,
             attn_cp_size=model_runner.ps.attn_cp_size,
             tp_group=model_runner.tp_group,
@@ -899,9 +899,12 @@ def latency_test(
     initialize_fp4_gemm_config()
 
     if get_bool_env_var("SGLANG_SET_CPU_AFFINITY"):
-        parallel = get_parallel().config
+        parallel = get_parallel()
         set_gpu_proc_affinity(
-            parallel.pp_size, parallel.tp_size, parallel.nnodes, tp_rank
+            parallel.pp_size,
+            parallel.tp_size,
+            parallel.nnodes,
+            tp_rank,
         )
 
     # Configure the logger

python/sglang/compile_deep_gemm.py

@@ -65,7 +65,7 @@ async def warm_up_compile(
     disaggregation_mode: str, tokenizer_manager: TokenizerManager
 ):
     print("\nGenerate warm up request for compiling DeepGEMM...\n")
-    dp_size = get_parallel().config.dp_size
+    dp_size = get_parallel().dp_size
     base_ids = [0, 1, 2, 3]
     sampling_params = {
         "temperature": 0.0,
@@ -81,7 +81,7 @@ async def warm_up_compile(
         )
         generate_req_input.bootstrap_host = [FAKE_BOOTSTRAP_HOST] * dp_size
         generate_req_input.bootstrap_room = [
-            i * (2**63 // dp_size) + (i % get_parallel().config.tp_size)
+            i * (2**63 // dp_size) + (i % get_parallel().tp_size)
             for i in range(dp_size)
         ]
     else:

python/sglang/multimodal_gen/runtime/distributed/parallel_state.py

@@ -701,7 +701,16 @@ def model_parallel_is_initialized() -> bool:
 
 @contextmanager
 def use_tensor_parallel_group(tp_group: GroupCoordinator):
-    """Use one TP group consistently across diffusion and reused SRT modules."""
+    """Use one TP group consistently across diffusion and reused SRT modules.
+
+    The scope replaces the module globals that ``get_tp_group()`` and srt's
+    ``get_tp_group()`` / ``get_attention_tp_group()`` read, and — like srt's
+    ``patch_tensor_parallel_group`` — the three members the runtime context
+    answers with, so that a size read from the published bag cannot disagree
+    with a rank read from the swapped group.
+    """
+    from sglang.srt.runtime_context import get_parallel
+
     old_tp_group = get_tp_group()
     import sglang.srt.distributed.parallel_state as srt_parallel_state
 
@@ -712,7 +721,12 @@ def use_tensor_parallel_group(tp_group: GroupCoordinator):
     srt_parallel_state._TP = tp_group
     srt_parallel_state._ATTN_TP = tp_group
     try:
-        yield
+        with get_parallel().override(
+            tp_size=tp_group.world_size,
+            tp_rank=tp_group.rank_in_group,
+            tp_group=tp_group,
+        ):
+            yield
     finally:
         _TP = old_tp_group
         srt_parallel_state._TP = old_srt_tp_group

python/sglang/multimodal_gen/runtime/managers/gpu_worker.py

@@ -291,7 +291,12 @@ def init_device_and_model(self) -> None:
         from sglang.srt.server_args import ServerArgs as SrtServerArgs
 
         if get_context()._server_args is None:
-            publish(SrtServerArgs(model_path="dummy"), role="diffusion_gpu_worker")
+            # srt reads the size from the configuration and the rank from the
+            # live group, so the dummy carries the width just installed.
+            publish(
+                SrtServerArgs(model_path="dummy", tp_size=self.server_args.tp_size),
+                role="diffusion_gpu_worker",
+            )
 
         # set proc title
         if model_parallel_is_initialized():

python/sglang/multimodal_gen/test/unit/test_component_accuracy_parallel_runtime.py

@@ -17,10 +17,19 @@
     initialize_parallel_runtime,
 )
 from sglang.srt.distributed import parallel_state as srt_parallel_state
+from sglang.srt.runtime_context import get_parallel
 
 _UTILS = "sglang.multimodal_gen.test.single_test_file.component_accuracy.utils"
 
 
+def _tp_group(world_size: int = 1, rank_in_group: int = 0) -> SimpleNamespace:
+    """A TP group handle carrying the two members the scope declares to the
+    runtime context (`use_tensor_parallel_group` overrides `tp_size` /
+    `tp_rank` / `tp_group` for its duration); the scope otherwise only stores
+    the handle and compares it by identity."""
+    return SimpleNamespace(world_size=world_size, rank_in_group=rank_in_group)
+
+
 def _server_args(*, ulysses_degree: int, ring_degree: int) -> SimpleNamespace:
     return SimpleNamespace(
         tp_size=1,
@@ -162,7 +171,7 @@ def test_srt_tp_groups_follow_encoder_folding_context():
     original_diffusion_tp_group = object()
     original_srt_tp_group = object()
     original_srt_attention_tp_group = object()
-    folding_tp_group = object()
+    folding_tp_group = _tp_group(world_size=2, rank_in_group=1)
 
     with (
         patch.object(parallel_state, "_TP", original_diffusion_tp_group),
@@ -177,6 +186,9 @@ def test_srt_tp_groups_follow_encoder_folding_context():
             assert parallel_state._TP is folding_tp_group
             assert srt_parallel_state._TP is folding_tp_group
             assert srt_parallel_state._ATTN_TP is folding_tp_group
+            assert get_parallel().tp_size == 2
+            assert get_parallel().tp_rank == 1
+            assert get_parallel().tp_group is folding_tp_group
 
         assert parallel_state._TP is original_diffusion_tp_group
         assert srt_parallel_state._TP is original_srt_tp_group
@@ -185,23 +197,28 @@ def test_srt_tp_groups_follow_encoder_folding_context():
 
 def test_encoder_folding_context_is_nested_and_restores_each_group():
     original_tp_group = object()
-    outer_tp_group = object()
-    inner_tp_group = object()
+    outer_tp_group = _tp_group(world_size=4, rank_in_group=3)
+    inner_tp_group = _tp_group(world_size=2, rank_in_group=1)
 
     with (
         patch.object(parallel_state, "_TP", original_tp_group),
         patch.object(srt_parallel_state, "_TP", original_tp_group),
         patch.object(srt_parallel_state, "_ATTN_TP", original_tp_group),
     ):
         with parallel_state.use_tensor_parallel_group(outer_tp_group):
+            assert get_parallel().tp_size == 4
             with parallel_state.use_tensor_parallel_group(inner_tp_group):
                 assert parallel_state._TP is inner_tp_group
                 assert srt_parallel_state._TP is inner_tp_group
                 assert srt_parallel_state._ATTN_TP is inner_tp_group
+                assert get_parallel().tp_size == 2
+                assert get_parallel().tp_rank == 1
 
             assert parallel_state._TP is outer_tp_group
             assert srt_parallel_state._TP is outer_tp_group
             assert srt_parallel_state._ATTN_TP is outer_tp_group
+            assert get_parallel().tp_size == 4
+            assert get_parallel().tp_rank == 3
 
         assert parallel_state._TP is original_tp_group
         assert srt_parallel_state._TP is original_tp_group

python/sglang/srt/arg_groups/overrides.py

@@ -190,10 +190,8 @@ def resolving_view(server_args: Any) -> ResolvingConfig:
     return ResolvingConfig(server_args)
 
 
-# Ordered post-process passes (the normalization stage). List order is the
-# end-state execution order and mirrors today's handler call sequence in
-# __post_init__; during the transition each pass is invoked from its legacy
-# slot via run_post_process_pass, so ordering is preserved byte-for-byte.
+# Registered post-process passes. This is a registry, not an execution order:
+# each pass is invoked from its own slot via run_post_process_pass.
 POST_PROCESS_PASSES: List[Callable[..., dict]] = []
 
 
@@ -223,11 +221,29 @@ def run_post_process_pass(server_args: Any, fn: Callable[..., dict]) -> None:
 
     Evaluates the pass on the resolving state (a read-only view with the
     accumulated declarations overlaid from the stash) and appends its
-    declaration to the stash. During ``__post_init__`` the fields stay
-    untouched: the stash is what the config bags are projected from. A pass
-    invoked after resolution finished (a post-init slot) writes through
-    immediately, because there is no later projection to pick it up.
+    declaration to the stash, which is what the config bags are projected from.
+    The fields stay untouched.
+
+    A slot that runs after resolution -- ``check_server_args`` hosts one -- lands
+    in the same stash, which publish projects from later, so it needs no field
+    write either. After *publish* there is no such later projection: the stash
+    would grow an entry nothing reads. So, like ``declare_late_resolution``,
+    this refuses the published record -- post-publish changes go to the bags
+    through ``get_context().override(...)``.
     """
+    from sglang.srt.runtime_context import get_context
+
+    try:
+        published = get_context().server_args
+    except ValueError:
+        published = None
+    if published is server_args:
+        raise ValueError(
+            f"run_post_process_pass({fn.__qualname__!r}) called on the published "
+            "config; the stash is projected at publish and never again, so a "
+            "declaration made here would be a silent no-op -- post-publish "
+            "changes go to the bags via get_context().override(...)"
+        )
     declared = fn(ResolvedView(server_args, overlay=_declaration_overlay(server_args)))
     if not isinstance(declared, dict):
         raise TypeError(
@@ -246,8 +262,6 @@ def run_post_process_pass(server_args: Any, fn: Callable[..., dict]) -> None:
             stash = server_args._resolved_overrides = []
         stash.append(entry)
         validate_declarations(server_args, [entry])
-        if getattr(server_args, "_resolution_finished", False):
-            _apply_fields(server_args, declared)
 
 
 def _apply_fields(server_args: Any, fields: Dict[str, Any]) -> None:
@@ -281,7 +295,7 @@ def declare_resolution(server_args: Any, source: str, **fields: Any) -> None:
     stash = getattr(server_args, "_resolved_overrides", None)
     if stash is None:
         stash = []
-        object.__setattr__(server_args, "_resolved_overrides", stash)
+        server_args._resolved_overrides = stash
     stash.append((source, dict(fields)))
 
 
@@ -315,12 +329,12 @@ def declare_late_resolution(server_args: Any, source: str, **fields: Any) -> Non
     log = getattr(server_args, "_runtime_mutations", None)
     if log is None:
         log = []
-        object.__setattr__(server_args, "_runtime_mutations", log)
+        server_args._runtime_mutations = log
     log.append((source, dict(fields)))
     stash = getattr(server_args, "_resolved_overrides", None)
     if stash is None:
         stash = []
-        object.__setattr__(server_args, "_resolved_overrides", stash)
+        server_args._resolved_overrides = stash
     stash.append((source, dict(fields)))
 
 
@@ -358,7 +372,7 @@ def declare_direct_writes(
     stash = getattr(server_args, "_resolved_overrides", None)
     if stash is None:
         stash = []
-        object.__setattr__(server_args, "_resolved_overrides", stash)
+        server_args._resolved_overrides = stash
     # A resolver reached this way can also declare properly -- the in-tree
     # implementations of these hooks do. Those fields are already explained, and
     # recording them again would attribute them to the wrapper and bury an
@@ -1745,9 +1759,7 @@ def _step3p_overrides(server_args: Any, hf_config: Any) -> dict:
 
 
 # ---------------------------------------------------------------------------
-# Post-process passes (normalization stage), in end-state execution order.
-# Faithful ports of the legacy __post_init__ handlers; each is invoked from
-# its legacy slot via run_post_process_pass during the transition.
+# Post-process passes (normalization stage).
 # ---------------------------------------------------------------------------
 
 
@@ -2803,6 +2815,7 @@ def _moe_runner_fusion_disable(view: Any) -> dict:
     return {}
 
 
+@register_post_process
 def _a2a_fusion_adjustments(view: Any) -> dict:
     """A2A-backend-driven shared-experts fusion adjustments, declared at the
     legacy write slots in _handle_a2a_moe: Waterfill requires the
@@ -2982,6 +2995,7 @@ def validate_declarations(
             )
 
 
+@register_post_process
 def _hrm_text_attention_force(view: Any) -> dict:
     """HRM-Text's bidirectional prefix attention only works on the Triton
     backend. Invoked as the last attention declaration of the resolution

python/sglang/srt/batch_overlap/two_batch_overlap.py

@@ -787,7 +787,7 @@ def filter_batch(
 
         # TODO improve, e.g. unify w/ `init_raw`
         if (
-            get_parallel().config.moe_dense_tp_size == 1
+            get_parallel().moe_dense_tp_size == 1
             and batch.global_dp_buffer_len is not None
         ):
             sum_len = end_token_index - start_token_index

python/sglang/srt/configs/zaya.py

@@ -268,7 +268,7 @@ def mamba2_cache_params(self) -> Optional[Mamba2CacheParams]:
         try:
 
             tp_size = get_parallel().tp_size
-        except (AssertionError, RuntimeError):
+        except (AssertionError, RuntimeError, ValueError):
             tp_size = 1
 
         in_out_ch_full = (

python/sglang/srt/debug_utils/dumper.py

@@ -1735,8 +1735,8 @@ def collect_parallel_info(self) -> dict:
             info["moe_tp_rank"] = parallel.moe_tp_rank
             info["moe_tp_size"] = parallel.moe_tp_size
             info["moe_dp_rank"] = parallel.moe_dp_rank
-            info["moe_dp_size"] = parallel.moe_dp_size
-        except (AttributeError, AssertionError):
+            info["moe_dp_size"] = self._dp_attn.get_moe_cp_size()
+        except (AttributeError, AssertionError, ValueError):
             info["distributed_error"] = True
 
         try:
@@ -1748,7 +1748,7 @@ def collect_parallel_info(self) -> dict:
             info["attn_dp_size"] = self._dp_attn.get_attention_dp_size()
             info["attn_cp_rank"] = parallel.attn_cp_rank
             info["attn_cp_size"] = parallel.attn_cp_size
-        except (AttributeError, AssertionError):
+        except (AttributeError, AssertionError, ValueError):
             info["dp_attention_error"] = True
 
         return info