[refactor] Add a read-through server_args accessor to RuntimeContext (stack 1/15) (#30063)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

python/sglang/srt/runtime_context.py

@@ -11,7 +11,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 # ==============================================================================
-"""A single structured accessor for process-static parallel-topology state.
+"""A single structured accessor for process-static runtime state.
 
 ``get_parallel()`` returns a ``ParallelContext`` whose attributes — tp / pp /
 moe / attn size and rank, plus the process-group handles — each delegate live to
@@ -20,12 +20,20 @@ Returned values are exactly what those getters return; this is a read-through
 wrapper, not a cache. It gives call-sites one import and one naming scheme in
 place of a dozen free functions, plus a test-only ``override()`` hook to force a
 topology without monkeypatching the underlying getters.
+
+``get_server_args()`` returns the process-wide ``ServerArgs`` (the config tier).
+It is a read-through to ``server_args.get_global_server_args()`` — same object,
+same pre-publish error — so new code can adopt the context accessor while the
+legacy getter remains canonical.
 """
 
 from __future__ import annotations
 
 from contextlib import contextmanager
-from typing import Any
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from sglang.srt.server_args import ServerArgs
 
 
 # Imported lazily so this module has no import-time dependencies: any module can
@@ -42,6 +50,12 @@ def _dp():
     return dp_attention
 
 
+def _sa():
+    from sglang.srt import server_args
+
+    return server_args
+
+
 _PARALLEL_FIELDS = frozenset(
     {
         "world_size",
@@ -206,13 +220,19 @@ class ParallelContext:
 
 
 class RuntimeContext:
-    """Container for the structured runtime accessors; currently exposes ``parallel``."""
+    """Container for the structured runtime accessors; exposes ``parallel`` and
+    ``server_args``."""
 
     __slots__ = ("parallel",)
 
     def __init__(self, parallel: ParallelContext):
         self.parallel = parallel
 
+    @property
+    def server_args(self) -> ServerArgs:
+        """The process-wide ``ServerArgs``, read through the global getter."""
+        return _sa().get_global_server_args()
+
 
 _PARALLEL = ParallelContext()
 _CONTEXT = RuntimeContext(parallel=_PARALLEL)
@@ -224,3 +244,7 @@ def get_context() -> RuntimeContext:
 
 def get_parallel() -> ParallelContext:
     return _PARALLEL
+
+
+def get_server_args() -> ServerArgs:
+    return _CONTEXT.server_args

test/registered/unit/test_runtime_context.py

@@ -12,11 +12,13 @@ from sglang.srt.runtime_context import (
     RuntimeContext,
     get_context,
     get_parallel,
+    get_server_args,
 )
 from sglang.test.test_utils import CustomTestCase
 
 _PS = "sglang.srt.distributed.parallel_state"
 _DP = "sglang.srt.layers.dp_attention"
+_SA = "sglang.srt.server_args"
 
 SIZE_RANK_DELEGATIONS = [
     ("world_size", f"{_PS}.get_world_size"),
@@ -142,5 +144,42 @@ class TestParallelOverride(_IsolatedOverrides):
         self.assertEqual(p._overrides, {})
 
 
+class TestServerArgsReadThrough(CustomTestCase):
+    """``server_args`` delegates live to the global getter (read-through, V2a)."""
+
+    def test_delegates_to_global_getter(self):
+        sentinel = object()
+        with patch(f"{_SA}.get_global_server_args", return_value=sentinel):
+            self.assertIs(get_server_args(), sentinel)
+            self.assertIs(get_context().server_args, sentinel)
+
+    def test_identity_with_global_getter(self):
+        import sglang.srt.server_args as server_args_module
+
+        # Identity (not equality) is the contract; publish accepts any object.
+        sentinel = object()
+        saved = server_args_module._global_server_args
+        try:
+            server_args_module.set_global_server_args_for_scheduler(sentinel)
+            self.assertIs(
+                get_server_args(), server_args_module.get_global_server_args()
+            )
+            self.assertIs(get_server_args(), sentinel)
+        finally:
+            server_args_module._global_server_args = saved
+
+    def test_pre_publish_error_passes_through(self):
+        import sglang.srt.server_args as server_args_module
+
+        saved = server_args_module._global_server_args
+        server_args_module._global_server_args = None
+        try:
+            with self.assertRaises(ValueError) as cm:
+                get_server_args()
+            self.assertEqual(str(cm.exception), "Global server args is not set yet!")
+        finally:
+            server_args_module._global_server_args = saved
+
+
 if __name__ == "__main__":
     unittest.main()