config: business code no longer reads the published ServerArgs (#34081)

test/registered/unit/layers/attention/test_gdn_prefill_backend_policy.py

@@ -18,6 +18,7 @@ from sglang.srt.layers.attention.linear.kernels.gdn_flashinfer import (
 )
 from sglang.srt.layers.attention.linear.kernels.gdn_triton import TritonGDNKernel
 from sglang.srt.layers.attention.linear.utils import LinearAttnKernelBackend
+from sglang.srt.runtime_context import get_context
 from sglang.test.ci.ci_register import register_cpu_ci
 
 register_cpu_ci(est_time=5, suite="base-a-test-cpu")
@@ -151,12 +152,12 @@ class TestFlashInferGDNPrefillBackendPolicy(unittest.TestCase):
             track_ssm_h_dst=torch.empty(4),
         )
 
-        with patch(
-            "sglang.srt.layers.attention.linear.kernels.gdn_flashinfer."
-            "get_server_args",
-            return_value=SimpleNamespace(mamba_cache_chunk_size=64),
-        ):
-            maybe_build_flashinfer_checkpoint_plan(forward_batch, metadata, "cpu")
+        # The chunk size is a derived config member; seed its private cache on a
+        # published config rather than patching an import binding.
+        override = get_context().override_server_args(_mamba_cache_chunk_size=64)
+        override.install()
+        self.addCleanup(override.restore)
+        maybe_build_flashinfer_checkpoint_plan(forward_batch, metadata, "cpu")
 
         torch.testing.assert_close(
             metadata.state_checkpoint_cu_starts,

test/registered/unit/model_loader/test_presharded_loader.py

@@ -786,13 +786,6 @@ class TestShardConfig(unittest.TestCase):
         # `_collect_shard_config` is the exact failure mode that left
         # moe_dense_tp_size / LM-head flags out of the cache key before.
         loader = object.__new__(PreshardedModelLoader)
-        server_args = SimpleNamespace(
-            moe_dp_size=2,
-            enable_fp32_lm_head=True,
-            ep_num_redundant_experts=4,
-            enable_eplb=True,
-            init_expert_location="trivial",
-        )
         model_config = SimpleNamespace(quantization="fp8", dtype=torch.bfloat16)
         required = {
             "tp",
@@ -819,8 +812,8 @@ class TestShardConfig(unittest.TestCase):
             enable_dp_lm_head=True,
         )
         with mock.patch(
-            "sglang.srt.model_loader.loader.get_server_args",
-            return_value=server_args,
+            "sglang.srt.model_loader.loader.configured_moe_dp_size",
+            return_value=2,
         ), mock.patch(
             "sglang.srt.model_loader.loader.get_parallel",
             return_value=parallel,

test/registered/unit/models/test_kimi_k25.py

@@ -39,7 +39,7 @@ from sglang.srt.multimodal.processors.kimi_k25 import (
     _resize_bicubic_if_needed,
     _resize_images_by_source_shape,
 )
-from sglang.srt.runtime_context import get_parallel
+from sglang.srt.runtime_context import get_context, get_parallel
 from sglang.srt.utils.cuda_ipc_transport_utils import (
     DEFER_CUDA_IPC_FEATURE_RECONSTRUCTION_KEY,
     CudaIpcTensorTransportProxy,
@@ -317,7 +317,11 @@ def test_dp_helper_supports_moonvit3d_packed_embeddings_on_tp1():
     tower = _MoonViT3dTower()
     pixel_values = torch.randn(4, 2)
 
-    with get_parallel().override(tp_size=1, tp_rank=0, attn_tp_size=1, attn_tp_rank=0):
+    # The IPC consumer count asks for the *configured* TP size (matching
+    # MmItemMemoryPool.try_to_recycle), so the double publishes one too.
+    with get_context().override_server_args(tp_size=1), get_parallel().override(
+        tp_size=1, tp_rank=0, attn_tp_size=1, attn_tp_rank=0
+    ):
         output = run_dp_sharded_mrope_vision_model(
             tower, pixel_values, [[1, 2, 2]], rope_type="rope_2d_packed"
         )
@@ -331,7 +335,11 @@ def test_dp_helper_can_lazily_load_kimi_features_on_tp1():
     pixel_values = torch.randn(4, 2)
     loader = Mock(return_value=pixel_values)
 
-    with get_parallel().override(tp_size=1, tp_rank=0, attn_tp_size=1, attn_tp_rank=0):
+    # The IPC consumer count asks for the *configured* TP size (matching
+    # MmItemMemoryPool.try_to_recycle), so the double publishes one too.
+    with get_context().override_server_args(tp_size=1), get_parallel().override(
+        tp_size=1, tp_rank=0, attn_tp_size=1, attn_tp_rank=0
+    ):
         output = run_dp_sharded_mrope_vision_model(
             tower,
             None,
@@ -479,11 +487,10 @@ def test_kimi_non_dp_keeps_grid_thws_on_the_host():
     model.mm_projector = _IdentityProjector()
     items = [_image_item(torch.randn(4, 2), [[1, 2, 2]])]
 
-    with get_parallel().override(
+    # The IPC consumer count asks for the *configured* TP size (matching
+    # MmItemMemoryPool.try_to_recycle), so the double publishes one too.
+    with get_context().override_server_args(tp_size=1), get_parallel().override(
         tp_size=1, tp_rank=0, attn_tp_size=1, attn_tp_rank=0
-    ), patch(
-        "sglang.srt.models.kimi_k25.get_server_args",
-        return_value=SimpleNamespace(tp_size=1),
     ):
         model.get_image_feature(items)
 

test/registered/unit/models/test_kimi_k3_vision.py

@@ -464,15 +464,17 @@ def test_kimi_k3_encoder_dp_defers_feature_materialization(monkeypatch):
     ]
     sharded_embeddings = torch.randn(2, 2)
 
+    # The IPC consumer count asks for the *configured* TP size (matching
+    # MmItemMemoryPool.try_to_recycle), so publish it; the live topology the
+    # sharding helper reads is forced through the context's own override.
     with mock_patch(
         "sglang.srt.multimodal.mm_utils.run_dp_sharded_mrope_vision_model",
         return_value=sharded_embeddings,
-    ) as run_dp, mock_patch(
-        "sglang.srt.models.kimi_k3.get_parallel",
-        return_value=SimpleNamespace(tp_size=1, attn_tp_size=1),
+    ) as run_dp, get_context().override_server_args(tp_size=1), get_parallel().override(
+        tp_size=1, attn_tp_size=1
     ):
         output = model.get_image_feature(items)
-        # Exercise the loader while the runtime topology is patched.
+        # Exercise the loader while the runtime topology is forced.
         loader_in_scope = run_dp.call_args.kwargs["load_local_pixel_values"]
         local = loader_in_scope([1])
         both = loader_in_scope([0, 1])
@@ -545,12 +547,13 @@ def test_kimi_k3_preprocesses_only_dp_owner_images(monkeypatch):
         calls.append([int(image[0, 0, 0]) for image in images])
         return torch.tensor([[float(calls[-1][0]), 0.0]]), torch.tensor([[1, 1, 1]])
 
+    # Configured TP size (the IPC consumer count) comes from the published
+    # bags; the live topology is forced through the context's own override.
     with mock_patch(
         "sglang.srt.multimodal.mm_utils.run_dp_sharded_mrope_vision_model",
         return_value=torch.zeros(1, 2),
-    ) as run_dp, mock_patch(
-        "sglang.srt.models.kimi_k3.get_parallel",
-        return_value=SimpleNamespace(tp_size=1, attn_tp_size=1),
+    ) as run_dp, get_context().override_server_args(tp_size=1), get_parallel().override(
+        tp_size=1, attn_tp_size=1
     ), mock_patch(
         "sglang.srt.multimodal.processors.kimi_k25._gpu_preprocess_images",
         side_effect=fake_preprocess,

test/registered/unit/spec/test_ngram_mamba_verify_update.py

@@ -197,9 +197,6 @@ class TestNgramMambaVerifyUpdate(CustomTestCase):
         with patch(
             "sglang.srt.speculative.spec_utils.mambaish_config",
             return_value={"some": "config"},
-        ), patch(
-            "sglang.srt.speculative.spec_utils.get_server_args",
-            return_value=MagicMock(mamba_track_interval=256),
         ), patch(
             "sglang.srt.speculative.spec_utils.get_exec",
             return_value=MagicMock(mamba=MagicMock(mamba_track_interval=256)),

test/registered/unit/test_context_accessor_shadowing.py

@@ -0,0 +1,191 @@
+"""No local may shadow a ``runtime_context`` accessor it also calls.
+
+A mechanical sweep that rewrites ``self.server_args.mamba_cache_chunk_size``
+into ``mamba_cache_chunk_size()`` turns
+
+    mamba_cache_chunk_size = self.server_args.mamba_cache_chunk_size
+
+into ``mamba_cache_chunk_size = mamba_cache_chunk_size()``, which is a
+self-referential local: the name is local for the whole function, so the call
+raises ``UnboundLocalError`` the first time that line runs. Five of these
+shipped in one sweep and only one had unit coverage — a mamba model on the
+radix-cache-v2 path found it at request time.
+
+This scans for the shape directly: a function-scope assignment whose target
+name is an imported accessor.
+"""
+
+import ast
+import unittest
+from pathlib import Path
+
+import sglang
+from sglang.test.ci.ci_register import register_cpu_ci
+from sglang.test.test_utils import CustomTestCase
+
+register_cpu_ci(est_time=5, suite="base-a-test-cpu")
+
+_PACKAGE_ROOT = Path(next(iter(sglang.__path__)))
+_CONTEXT_MODULE = "sglang.srt.runtime_context"
+
+
+def _module_level_accessor_imports(tree: ast.AST) -> set[str]:
+    """Accessors imported at module scope — visible in every function.
+
+    A *function-local* import is visible only inside its own scope, so it is
+    collected per function in the scan below: charging it file-wide would flag
+    an unrelated sibling function that binds the same name, where no shadowing
+    can occur.
+    """
+    names: set[str] = set()
+    stack = list(tree.body)
+    while stack:
+        stmt = stack.pop()
+        if isinstance(
+            stmt, (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda, ast.ClassDef)
+        ):
+            continue
+        if isinstance(stmt, ast.ImportFrom) and stmt.module == _CONTEXT_MODULE:
+            for alias in stmt.names:
+                names.add(alias.asname or alias.name)
+        stack.extend(ast.iter_child_nodes(stmt))
+    return names
+
+
+def _bound_names(target):
+    """Every name a binding target introduces, unpacking included.
+
+    ``a, (b, c) = ...`` and ``for x, y in ...`` bind through Tuple/List/Starred
+    nodes, so a check that only accepts a bare ``ast.Name`` misses them.
+    """
+    if isinstance(target, ast.Name):
+        yield target.id
+    elif isinstance(target, ast.Starred):
+        yield from _bound_names(target.value)
+    elif isinstance(target, (ast.Tuple, ast.List)):
+        for element in target.elts:
+            yield from _bound_names(element)
+
+
+def _own_scope_statements(node) -> tuple:
+    """This function's OWN scope: its statements, plus the (name, lineno) of
+    each nested ``def``/``class`` — the definition's *name* is a binding in
+    this scope (an earlier accessor call raises UnboundLocalError just like an
+    assignment), while its *body* is the nested scope's own and descending into
+    it would misattribute bindings."""
+    own_scope = []
+    nested_def_bindings = []
+    pending = list(node.body)
+    while pending:
+        stmt = pending.pop()
+        if isinstance(stmt, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
+            nested_def_bindings.append((stmt.name, stmt.lineno))
+            continue
+        if isinstance(stmt, ast.Lambda):
+            continue
+        own_scope.append(stmt)
+        pending.extend(ast.iter_child_nodes(stmt))
+    return own_scope, nested_def_bindings
+
+
+def _child_functions(body) -> list:
+    """Function defs directly beneath this scope — descending through plain
+    statements and class bodies (a method closes over the enclosing function's
+    names, not the class's), but never into another function."""
+    funcs = []
+    pending = list(body)
+    while pending:
+        stmt = pending.pop()
+        if isinstance(stmt, (ast.FunctionDef, ast.AsyncFunctionDef)):
+            funcs.append(stmt)
+            continue
+        if isinstance(stmt, ast.Lambda):
+            continue
+        pending.extend(ast.iter_child_nodes(stmt))
+    return funcs
+
+
+def _shadowing_assignments(tree: ast.AST, module_accessors: set[str]):
+    """Function-local bindings whose name shadows an accessor visible in that
+    scope -- every statement form that binds a local, not just ``=``.
+
+    Python decides a name is local from *any* binding in the function, so a
+    loop variable, a ``with ... as``, a walrus, a comprehension target, or an
+    ``except ... as`` all shadow the accessor for the whole function body,
+    exactly like an assignment does.
+
+    Visibility follows lexical scope: module-level imports reach every
+    function; a function-local import reaches its own scope and nested
+    functions (closure), but NOT an unrelated sibling — charging it file-wide
+    would flag bindings where no shadowing occurs. A function-scope *re-import*
+    of the accessor is itself fine: it binds the name to the same callable, so
+    calls after it behave identically (and the module is full of deliberate
+    local imports).
+    """
+    stack = [(fn, module_accessors) for fn in _child_functions(tree.body)]
+    while stack:
+        node, inherited = stack.pop()
+        own_scope, nested_def_bindings = _own_scope_statements(node)
+        local_imports = {
+            alias.asname or alias.name
+            for stmt in own_scope
+            if isinstance(stmt, ast.ImportFrom) and stmt.module == _CONTEXT_MODULE
+            for alias in stmt.names
+        }
+        visible = inherited | local_imports
+        # ``def get_exec(): ...`` nested in the function binds the name in
+        # THIS scope, exactly like an assignment would.
+        for name, lineno in nested_def_bindings:
+            if name in visible:
+                yield node.name, name, lineno
+        for inner in own_scope:
+            targets = []
+            if isinstance(inner, ast.Assign):
+                targets = inner.targets
+            elif isinstance(inner, (ast.AnnAssign, ast.AugAssign)):
+                targets = [inner.target]
+            elif isinstance(inner, (ast.For, ast.AsyncFor, ast.comprehension)):
+                targets = [inner.target]
+            elif isinstance(inner, ast.NamedExpr):
+                targets = [inner.target]
+            elif isinstance(inner, (ast.With, ast.AsyncWith)):
+                targets = [i.optional_vars for i in inner.items if i.optional_vars]
+            elif isinstance(inner, ast.ExceptHandler) and inner.name:
+                targets = [ast.Name(id=inner.name, ctx=ast.Store())]
+            for target in targets:
+                for name in _bound_names(target):
+                    if name in visible:
+                        yield node.name, name, getattr(inner, "lineno", node.lineno)
+        for nested in _child_functions(node.body):
+            stack.append((nested, visible))
+
+
+class TestNoAccessorShadowing(CustomTestCase):
+    def test_no_local_shadows_a_context_accessor(self):
+        offenders = []
+        for path in sorted(_PACKAGE_ROOT.rglob("*.py")):
+            rel = path.relative_to(_PACKAGE_ROOT).as_posix()
+            if rel.startswith("srt/runtime_context.py"):
+                continue
+            source = path.read_text()
+            # A file that never names the module cannot import an accessor
+            # from it, at module scope or inside any function.
+            if _CONTEXT_MODULE not in source:
+                continue
+            try:
+                tree = ast.parse(source)
+            except SyntaxError:
+                continue
+            module_accessors = _module_level_accessor_imports(tree)
+            for func, name, lineno in _shadowing_assignments(tree, module_accessors):
+                offenders.append(f"{rel}:{lineno}: {func}() binds {name!r}")
+        self.assertFalse(
+            offenders,
+            "locals shadow a runtime_context accessor imported in the same "
+            "module; the name is local for the whole function, so any call to "
+            "the accessor there raises UnboundLocalError:\n" + "\n".join(offenders),
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

test/registered/unit/test_global_config_read_ratchet.py

@@ -6,22 +6,24 @@ startup record. Config decisions read the namespace accessors instead
 including post-publish overrides, and per-runner values come from the runner
 that owns them.
 
-Two shapes count as a read: the direct ``get_server_args().field``, and the
-alias ``sa = get_server_args()`` followed by ``sa.field`` in the same function.
-A whole-object pass (``def f(server_args)``) is not a global read and is not
-counted — there the caller decided which instance to hand over.
+Business code no longer reads the published record for a config value at all:
+the baselines are zero for both shapes, over the whole package minus the two
+modules that own the slot.
 
-What legitimately remains:
+Where the remaining reads live (``runtime_context.py``, exempt by module):
 
-- **Derived APIs.** ``@property`` and method members of ``ServerArgs``
-  (``mamba_cache_chunk_size``, ``get_model_config()``,
-  ``enable_mamba_extra_buffer()``, …) are computed from several fields plus the
-  HF config, so they are not namespace leaves and ``ServerArgs`` is their only
-  home. Exempt by name below.
+- **Derived members.** ``@property`` / method members of ``ServerArgs``
+  (``mamba_cache_chunk_size``, ``max_speculative_num_draft_tokens``,
+  ``use_mla_backend()``, ``get_attention_backends()``, ``get_model_config()``,
+  ``cutedsl_moe_max_num_tokens()``) are computed from several fields plus the HF
+  config, so they are not namespace leaves and ``ServerArgs`` is their only
+  home. ``runtime_context`` exposes each one as a named accessor
+  (``mamba_cache_chunk_size()`` …) and is the only module that reads the slot
+  for them.
 - **Config-intent reads of live-shadowed sizes.** ``get_parallel()`` shadows
-  ``tp/pp/dcp/attn_cp/moe_dp_size`` with the live topology, so a config-intent
-  read of one has nowhere else to go. Each exempt site needs an answer the live
-  property cannot give:
+  ``tp/pp/dcp/attn_cp/moe_dp_size`` with the live topology, and a few call sites
+  need what was *configured*: the ``configured_*_size()`` accessors. Their
+  reasons, per call site:
 
   - ``dsa_indexer.pp_size`` gates ``pp_size > 1 and not get_pp_group()...``, and
     the short circuit is the point: with PP off the group is never touched, which
@@ -35,8 +37,23 @@ What legitimately remains:
     sizes are equal there and a live comparison is always false.
   - ``model_loader/loader.py`` reports both: the same dict carries the live
     ``moe_dp_size`` under ``"dp"``, so this entry is the configured intent.
-- The alias-form baseline is not zero yet. Lowering it is the next slice; the
-  failure message lists the sites whenever the count moves.
+
+What the ratchet sees, syntactically: ``get_server_args().field``,
+``sa = get_server_args()`` followed by ``sa.field`` (function-local, module-level,
+or parked on an instance attribute -- ``self._sa = get_server_args()`` read from
+another method of the same class), function-local copies of an alias to a
+fixpoint (``cfg = sa`` then ``cfg.field``), and the dynamic form of each --
+``getattr(<either>, "field")`` -- since a string-named read reaches the same
+slot. What it cannot see is a name computed at runtime (``getattr(sa, name)``)
+or indirection deeper than a local name copy (through a container, an
+attribute of another object, a cross-scope copy); the census tool in the
+context repo is what audits those.
+
+A whole-object pass (``def f(server_args)``) is not a global read and is not
+counted -- there the caller decided which instance to hand over. An optional
+parameter that falls back to the global (``f(server_args=None)``) hides one,
+so those fallbacks were removed; the ratchet cannot see them and the census
+tool in the context repo is what audits that shape.
 """
 
 from sglang.test.ci.ci_register import register_cpu_ci
@@ -54,48 +71,92 @@ from sglang.test.test_utils import CustomTestCase
 # scanned so a new one cannot appear there unnoticed.
 _PACKAGE_ROOT = Path(next(iter(sglang.__path__)))
 
-_DERIVED_MEMBERS = frozenset(
-    {
-        "cutedsl_moe_max_num_tokens",
-        "enable_mamba_extra_buffer",
-        "enable_mamba_extra_buffer_lazy",
-        "get_attention_backends",
-        "get_model_config",
-        "mamba_cache_chunk_size",
-        "max_speculative_num_draft_tokens",
-        "model_config",
-        "use_mla_backend",
-    }
-)
+# The modules that own the slot: runtime_context publishes it and exposes the
+# named accessors for the derived members, server_args/arg_groups ARE the
+# resolution pipeline.
+_SLOT_OWNERS = ("srt/runtime_context.py", "srt/server_args.py", "srt/arg_groups/")
 
-_CONFIG_INTENT_SIZES = frozenset(
-    {
-        ("srt/layers/attention/dsa/dsa_indexer.py", "pp_size"),
-        ("srt/layers/dp_attention.py", "attn_cp_size"),
-        ("srt/layers/dp_attention.py", "moe_dp_size"),
-        ("srt/model_loader/loader.py", "moe_dp_size"),
-        ("srt/utils/cuda_ipc_transport_utils.py", "tp_size"),
-    }
-)
+# Every call site of a ``configured_*_size()`` accessor, with the reason the
+# live topology cannot answer there. The test below asserts this map is exactly
+# the set of call sites, so the reasons cannot drift away from the code.
+_CONFIGURED_SIZE_CALL_SITES = {
+    ("srt/layers/attention/dsa/dsa_indexer.py", "configured_pp_size"): (
+        "gates `pp_size > 1 and not get_pp_group()...`; the short circuit is the "
+        "point, since with PP off the group is never touched, which is what lets "
+        "the Indexer be constructed before distributed init"
+    ),
+    ("srt/layers/dp_attention.py", "configured_attn_cp_size"): (
+        "compared against the configured moe_dp_size below"
+    ),
+    ("srt/layers/dp_attention.py", "configured_moe_dp_size"): (
+        "the configuration this predicate detects (attn_cp_size > moe_dp_size) is "
+        "the one where initialize_model_parallel aliases _MOE_DP to _ATTN_CP, so "
+        "the live sizes are equal there and a live comparison is always false"
+    ),
+    ("srt/model_loader/loader.py", "configured_moe_dp_size"): (
+        "the same dict already carries the live moe_dp_size under 'dp'; this entry "
+        "is the configured intent"
+    ),
+    ("srt/utils/cuda_ipc_transport_utils.py", "configured_tp_size"): (
+        "runs in the tokenizer process, which has no parallel groups at all"
+    ),
+    ("srt/models/kimi_k25.py", "configured_tp_size"): (
+        "the IPC refcount has to name the same number the recycler waits on, and "
+        "that waiter (MmItemMemoryPool.try_to_recycle) reads configured_tp_size() "
+        "because it runs in the tokenizer process; a refcount taken from the live "
+        "attention subgroup would strand items in the bounded pool"
+    ),
+    ("srt/models/kimi_k3.py", "configured_tp_size"): (
+        "same as kimi_k25: the IPC refcount must agree with the recycler's waiter"
+    ),
+}
+
+# A dynamic read whose name is set nowhere in the tree, so the predicate it
+# feeds is inert (the ``getattr`` default decides it). Converting it would mean
+# choosing what it should have named, which is the CP path's call, not this
+# sweep's -- so it is listed here rather than silently counted or "fixed".
+_INERT_DYNAMIC_READS = frozenset({("srt/layers/cp/base.py", "_is_dsa_model_arch")})
 
 _DIRECT_BASELINE = 0
 _ALIAS_BASELINE = 0
 
 
 def _is_global_call(node) -> bool:
-    return (
-        isinstance(node, ast.Call)
-        and isinstance(node.func, ast.Name)
-        and node.func.id == "get_server_args"
-    )
+    """``get_server_args()`` however it is spelled: bare, or module-qualified
+    (``ctx.get_server_args()``), which an ast.Name check alone would miss."""
+    if not isinstance(node, ast.Call):
+        return False
+    func = node.func
+    if isinstance(func, ast.Name):
+        return func.id == "get_server_args"
+    return isinstance(func, ast.Attribute) and func.attr == "get_server_args"
 
 
-def _collect(rel: str, tree: ast.AST):
-    """The (direct, alias) field reads in one module."""
+def _collect(rel: str, tree: ast.AST, inert: frozenset = frozenset()):
+    """The (direct, alias) field reads in one module.
+
+    ``inert`` names the fields listed in ``_INERT_DYNAMIC_READS`` for this file;
+    they are dropped here, at the point the read is recognized, so the filter
+    matches on the field name rather than on the rendered message.
+    """
     direct, alias = [], []
 
     def counted(attr: str) -> bool:
-        return attr not in _DERIVED_MEMBERS and (rel, attr) not in _CONFIG_INTENT_SIZES
+        return attr not in inert
+
+    def _getattr_name(node):
+        """``getattr(<record>, "field")`` names a field just as ``.field`` does;
+        matching only ast.Attribute would let a dynamic read walk past."""
+        if not (
+            isinstance(node, ast.Call)
+            and isinstance(node.func, ast.Name)
+            and node.func.id == "getattr"
+            and len(node.args) >= 2
+            and isinstance(node.args[1], ast.Constant)
+            and isinstance(node.args[1].value, str)
+        ):
+            return None
+        return node.args[1].value
 
     for node in ast.walk(tree):
         if (
@@ -105,17 +166,54 @@ def _collect(rel: str, tree: ast.AST):
         ):
             direct.append(f"{rel}:{node.lineno}: get_server_args().{node.attr}")
 
+        name = _getattr_name(node)
+        if name is not None and _is_global_call(node.args[0]) and counted(name):
+            direct.append(f"{rel}:{node.lineno}: getattr(get_server_args(), {name!r})")
+
         if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
             continue
         params = {a.arg for a in list(node.args.args) + list(node.args.kwonlyargs)}
         bound = {}
         for inner in ast.walk(node):
-            if isinstance(inner, ast.Assign) and _is_global_call(inner.value):
-                for target in inner.targets:
-                    if isinstance(target, ast.Name) and target.id not in params:
-                        bound.setdefault(target.id, inner.lineno)
+            # ``sa = get_server_args()`` and its annotated form
+            # ``sa: ServerArgs = get_server_args()``.
+            if isinstance(inner, (ast.Assign, ast.AnnAssign)) and _is_global_call(
+                getattr(inner, "value", None)
+            ):
+                targets = (
+                    inner.targets if isinstance(inner, ast.Assign) else [inner.target]
+                )
+                for target in targets:
+                    if not isinstance(target, ast.Name):
+                        continue
+                    # A parameter reassigned from the global is the
+                    # optional-injection shape (``f(server_args=None)`` then
+                    # ``server_args = get_server_args()``): the reads that
+                    # follow are global reads wearing a parameter's name, so
+                    # they count from the bind on.
+                    bound.setdefault(target.id, inner.lineno)
         if not bound:
             continue
+        # A copy of an alias reaches the same record (``cfg = sa`` after
+        # ``sa = get_server_args()``), so follow Name-to-Name assignments to a
+        # fixpoint. Deeper indirection (through containers, attributes of
+        # other objects, cross-scope copies) stays census-tool territory.
+        changed = True
+        while changed:
+            changed = False
+            for inner in ast.walk(node):
+                if not isinstance(inner, (ast.Assign, ast.AnnAssign)):
+                    continue
+                value = getattr(inner, "value", None)
+                if not (isinstance(value, ast.Name) and value.id in bound):
+                    continue
+                targets = (
+                    inner.targets if isinstance(inner, ast.Assign) else [inner.target]
+                )
+                for target in targets:
+                    if isinstance(target, ast.Name) and target.id not in bound:
+                        bound[target.id] = inner.lineno
+                        changed = True
         for inner in ast.walk(node):
             if (
                 isinstance(inner, ast.Attribute)
@@ -128,6 +226,170 @@ def _collect(rel: str, tree: ast.AST):
                     f"{rel}:{inner.lineno}: {inner.value.id}.{inner.attr} "
                     f"(bound from get_server_args() at line {bound[inner.value.id]})"
                 )
+            name = _getattr_name(inner)
+            if (
+                name is not None
+                and isinstance(inner.args[0], ast.Name)
+                and inner.args[0].id in bound
+                and inner.lineno >= bound[inner.args[0].id]
+                and counted(name)
+            ):
+                alias.append(
+                    f"{rel}:{inner.lineno}: getattr({inner.args[0].id}, {name!r}) "
+                    f"(bound from get_server_args() at line {bound[inner.args[0].id]})"
+                )
+    # A module-level alias is visible to every function in the file, so it needs
+    # its own pass -- the per-function scan above deliberately does not reach
+    # across scopes.
+    module_bound = {}
+    module_stack = list(tree.body)
+    while module_stack:
+        stmt = module_stack.pop()
+        # A module-level bind can sit inside an `if` / `try` / `with`, so the
+        # walk descends into those bodies -- but not into a nested function or
+        # class, whose binds are that scope's own.
+        if isinstance(
+            stmt, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef, ast.Lambda)
+        ):
+            continue
+        module_stack.extend(ast.iter_child_nodes(stmt))
+        if isinstance(stmt, (ast.Assign, ast.AnnAssign)) and _is_global_call(
+            getattr(stmt, "value", None)
+        ):
+            targets = stmt.targets if isinstance(stmt, ast.Assign) else [stmt.target]
+            for target in targets:
+                if isinstance(target, ast.Name):
+                    module_bound.setdefault(target.id, stmt.lineno)
+    if module_bound:
+        # Shadowing is per lexical scope: a function with its own `sa` hides the
+        # module alias *inside that function only*. Aggregating the names
+        # file-wide would suppress every read in the module, including the
+        # top-level ones and the ones in functions that do resolve to the alias.
+        parents = {}
+        scope_binds = {}
+        stack = [tree]
+        while stack:
+            node = stack.pop()
+            enclosing = parents.get(id(node))
+            for child in ast.iter_child_nodes(node):
+                parents[id(child)] = (
+                    node
+                    if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
+                    else enclosing
+                )
+                stack.append(child)
+            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
+                names = {
+                    a.arg for a in list(node.args.args) + list(node.args.kwonlyargs)
+                }
+                # Only this scope's own stores: a nested function's local `sa`
+                # shadows the alias inside *that* function, not in its parent.
+                pending = list(node.body)
+                while pending:
+                    inner = pending.pop()
+                    if isinstance(
+                        inner,
+                        (
+                            ast.FunctionDef,
+                            ast.AsyncFunctionDef,
+                            ast.Lambda,
+                            ast.ClassDef,
+                        ),
+                    ):
+                        continue
+                    if isinstance(inner, ast.Name) and isinstance(inner.ctx, ast.Store):
+                        names.add(inner.id)
+                    pending.extend(ast.iter_child_nodes(inner))
+                scope_binds[id(node)] = names
+
+        def _shadowed(node, name):
+            scope = parents.get(id(node))
+            while scope is not None:
+                if name in scope_binds.get(id(scope), ()):
+                    return True
+                scope = parents.get(id(scope))
+            return False
+
+        for node in ast.walk(tree):
+            base = attr = None
+            if (
+                isinstance(node, ast.Attribute)
+                and isinstance(node.value, ast.Name)
+                and node.value.id in module_bound
+            ):
+                base, attr = node.value.id, node.attr
+                shown = f"{base}.{attr}"
+            else:
+                attr_name = _getattr_name(node)
+                if (
+                    attr_name is not None
+                    and isinstance(node.args[0], ast.Name)
+                    and node.args[0].id in module_bound
+                ):
+                    base, attr = node.args[0].id, attr_name
+                    shown = f"getattr({base}, {attr!r})"
+            if base and not _shadowed(node, base) and counted(attr):
+                alias.append(
+                    f"{rel}:{node.lineno}: {shown} "
+                    f"(module-level bind from get_server_args() at line "
+                    f"{module_bound[base]})"
+                )
+
+    # An alias parked on an instance attribute (``self._sa = get_server_args()``
+    # in one method, ``self._sa.field`` in another) reaches the same slot and
+    # crosses function scopes, so it is collected per class rather than per
+    # function.
+    for node in ast.walk(tree):
+        if not isinstance(node, ast.ClassDef):
+            continue
+        attr_bound = {}
+        for inner in ast.walk(node):
+            if isinstance(inner, (ast.Assign, ast.AnnAssign)) and _is_global_call(
+                getattr(inner, "value", None)
+            ):
+                targets = (
+                    inner.targets if isinstance(inner, ast.Assign) else [inner.target]
+                )
+                for target in targets:
+                    if (
+                        isinstance(target, ast.Attribute)
+                        and isinstance(target.value, ast.Name)
+                        and target.value.id in ("self", "cls")
+                    ):
+                        attr_bound.setdefault(
+                            (target.value.id, target.attr), inner.lineno
+                        )
+        if not attr_bound:
+            continue
+
+        def _bound_attr(value):
+            """``self._sa`` when that attribute was bound from the global."""
+            if (
+                isinstance(value, ast.Attribute)
+                and isinstance(value.value, ast.Name)
+                and (value.value.id, value.attr) in attr_bound
+            ):
+                return (value.value.id, value.attr)
+            return None
+
+        for inner in ast.walk(node):
+            key = shown = None
+            if isinstance(inner, ast.Attribute):
+                key = _bound_attr(inner.value)
+                if key is not None and counted(inner.attr):
+                    shown = f"{key[0]}.{key[1]}.{inner.attr}"
+            else:
+                name = _getattr_name(inner)
+                if name is not None:
+                    key = _bound_attr(inner.args[0])
+                    if key is not None and counted(name):
+                        shown = f"getattr({key[0]}.{key[1]}, {name!r})"
+            if shown is not None:
+                alias.append(
+                    f"{rel}:{inner.lineno}: {shown} "
+                    f"(attribute bind from get_server_args() at line "
+                    f"{attr_bound[key]})"
+                )
     return direct, alias
 
 
@@ -135,11 +397,14 @@ def _field_reads():
     direct, alias = [], []
     for path in sorted(_PACKAGE_ROOT.rglob("*.py")):
         rel = path.relative_to(_PACKAGE_ROOT).as_posix()
+        if rel.startswith(_SLOT_OWNERS):
+            continue
         try:
             tree = ast.parse(path.read_text())
         except SyntaxError:
             continue
-        module_direct, module_alias = _collect(rel, tree)
+        inert = frozenset(name for path_, name in _INERT_DYNAMIC_READS if path_ == rel)
+        module_direct, module_alias = _collect(rel, tree, inert)
         direct += module_direct
         alias += module_alias
     return direct, alias
@@ -167,5 +432,90 @@ class TestGlobalConfigReadRatchet(CustomTestCase):
         self._check("alias-form", alias, _ALIAS_BASELINE)
 
 
+class TestConfiguredSizeCallSites(CustomTestCase):
+    """The configured-vs-live exceptions are enumerated, with reasons.
+
+    ``configured_*_size()`` answers what the user asked for where
+    ``get_parallel()`` would answer what the process ended up with. Each such
+    exception is listed above with why the live property cannot serve it, and
+    this case fails if the code and that list disagree.
+
+    The unit is **(file, accessor)**, not the individual call: a second
+    `configured_pp_size()` in a file already registered for it collapses into
+    the same entry, so the reason has to cover the file's use of that accessor
+    rather than one line. A new file, or a new accessor in a listed file, is
+    what this catches -- in either call form (bare or module-qualified).
+    """
+
+    def test_the_call_sites_match_the_documented_set(self):
+        found = set()
+        for path in sorted(_PACKAGE_ROOT.rglob("*.py")):
+            rel = path.relative_to(_PACKAGE_ROOT).as_posix()
+            if rel.startswith(_SLOT_OWNERS):
+                continue
+            try:
+                tree = ast.parse(path.read_text())
+            except SyntaxError:
+                continue
+            for node in ast.walk(tree):
+                if not isinstance(node, ast.Call):
+                    continue
+                func = node.func
+                name = (
+                    func.id
+                    if isinstance(func, ast.Name)
+                    else (func.attr if isinstance(func, ast.Attribute) else None)
+                )
+                if name and name.startswith("configured_") and name.endswith("_size"):
+                    found.add((rel, name))
+        documented = set(_CONFIGURED_SIZE_CALL_SITES)
+        self.assertEqual(
+            documented,
+            found,
+            "configured-size call sites drifted from their documented reasons.\n"
+            f"  undocumented: {sorted(found - documented)}\n"
+            f"  stale entries: {sorted(documented - found)}",
+        )
+
+
+class TestNoRenamedAccessorImports(CustomTestCase):
+    """The scanners above match ``get_server_args`` and ``configured_*_size``
+    by their literal names, so an ``import ... as`` rename would walk a read
+    straight past both the zero baseline and the call-site registry. Renaming
+    these accessors buys nothing (the names are already short and unambiguous),
+    so it is banned outright — which is exactly what makes literal-name
+    matching sound."""
+
+    def test_the_scanned_accessors_are_never_import_renamed(self):
+        offenders = []
+        for path in sorted(_PACKAGE_ROOT.rglob("*.py")):
+            rel = path.relative_to(_PACKAGE_ROOT).as_posix()
+            try:
+                tree = ast.parse(path.read_text())
+            except SyntaxError:
+                continue
+            for node in ast.walk(tree):
+                if not isinstance(node, (ast.ImportFrom, ast.Import)):
+                    continue
+                for imported in node.names:
+                    if imported.asname is None or imported.asname == imported.name:
+                        continue
+                    base = imported.name.rsplit(".", 1)[-1]
+                    if base == "get_server_args" or (
+                        base.startswith("configured_") and base.endswith("_size")
+                    ):
+                        offenders.append(
+                            f"{rel}:{node.lineno}: {imported.name} as "
+                            f"{imported.asname}"
+                        )
+        self.assertFalse(
+            offenders,
+            "get_server_args / configured_*_size imported under another name; "
+            "the read ratchet and the configured-size registry match these "
+            "accessors by their literal names, so a rename silently escapes "
+            "both:\n" + "\n".join(offenders),
+        )
+
+
 if __name__ == "__main__":
     unittest.main()

test/registered/unit/test_runtime_context.py

@@ -5,7 +5,10 @@ from sglang.test.ci.ci_register import register_cpu_ci
 register_cpu_ci(est_time=5, suite="base-a-test-cpu")
 
 import dataclasses
+import json
 import os
+import shutil
+import tempfile
 import unittest
 from unittest.mock import patch
 
@@ -20,8 +23,10 @@ from sglang.srt.runtime_context import (
     get_flags,
     get_parallel,
     get_server_args,
+    max_speculative_num_draft_tokens,
     reset_context,
 )
+from sglang.srt.server_args import ServerArgs
 from sglang.test.test_utils import CustomTestCase
 
 _PS = "sglang.srt.distributed.parallel_state"
@@ -378,6 +383,14 @@ class _FakeResolvedArgs:
     sampling_backend: A[
         str | None, Arg(help="s", resolvable=True), NS("exec.kernel")
     ] = None
+    attention_backend: A[str | None, Arg(help="ab"), NS("exec.kernel")] = None
+    prefill_attention_backend: A[str | None, Arg(help="pab"), NS("exec.kernel")] = None
+    decode_attention_backend: A[str | None, Arg(help="dab"), NS("exec.kernel")] = None
+    disable_radix_cache: A[bool, Arg(help="drc"), NS("memory")] = False
+    mamba_radix_cache_strategy: A[str, Arg(help="mrcs"), NS("exec.mamba")] = "auto"
+    speculative_num_draft_tokens: A[int | None, Arg(help="d"), NS("spec")] = None
+    speculative_adaptive: A[bool, Arg(help="a"), NS("spec")] = False
+    speculative_adaptive_config: A[str | None, Arg(help="c"), NS("spec")] = None
     _resolved_overrides: list = dataclasses.field(default_factory=list)
 
 
@@ -965,5 +978,180 @@ class TestPublishLifecycle(_IsolatedServerArgs):
         self.assertFalse(get_flags().capture.enable_torch_compile)
 
 
+class TestDerivedPredicatesAgreeAcrossTiers(_IsolatedServerArgs):
+    """One definition per predicate, checked rather than asserted in prose.
+
+    Each of these exists twice by construction -- once over a config-shaped
+    object (the resolution pipeline's `*_of` helper, which `ServerArgs`
+    delegates to) and once over the published bags. The pair must agree on
+    every input, or a decision made before publish differs from the same
+    decision made after it.
+    """
+
+    _STRATEGIES = ("auto", "no_buffer", "extra_buffer", "extra_buffer_lazy")
+
+    def test_mamba_extra_buffer_matches_the_member(self):
+        from sglang.srt.runtime_context import (
+            mamba_extra_buffer_enabled,
+            mamba_extra_buffer_lazy_enabled,
+        )
+
+        for disable_radix_cache in (False, True):
+            for strategy in self._STRATEGIES:
+                with self.subTest(radix=disable_radix_cache, strategy=strategy):
+                    args = _FakeResolvedArgs(
+                        disable_radix_cache=disable_radix_cache,
+                        mamba_radix_cache_strategy=strategy,
+                    )
+                    get_context().set_server_args(args)
+                    self.assertEqual(
+                        ServerArgs.enable_mamba_extra_buffer(args),
+                        mamba_extra_buffer_enabled(),
+                    )
+                    self.assertEqual(
+                        ServerArgs.enable_mamba_extra_buffer_lazy(args),
+                        mamba_extra_buffer_lazy_enabled(),
+                    )
+
+    def test_attention_backends_match_the_member(self):
+        from sglang.srt.runtime_context import attention_backends
+
+        backends = (None, "fa3", "triton")
+        for base in backends:
+            for prefill in backends:
+                for decode in backends:
+                    with self.subTest(base=base, prefill=prefill, decode=decode):
+                        args = _FakeResolvedArgs(
+                            attention_backend=base,
+                            prefill_attention_backend=prefill,
+                            decode_attention_backend=decode,
+                        )
+                        get_context().set_server_args(args)
+                        self.assertEqual(
+                            ServerArgs.get_attention_backends(args),
+                            attention_backends(),
+                        )
+
+
+class TestAdaptiveDraftBoundLifecycle(_IsolatedServerArgs):
+    """The adaptive draft-token bound is memoized on the config path, so the
+    memo has to end with the publication it was computed under.
+
+    Without that, a process that republishes with the same adaptive-config path
+    -- the file having been rewritten in between -- keeps the previous bound and
+    under-allocates the draft-token buffers sized from it.
+    """
+
+    def _write_config(self, steps):
+        path = os.path.join(tempfile.mkdtemp(prefix="adaptive_cfg_"), "adaptive.json")
+        self.addCleanup(shutil.rmtree, os.path.dirname(path), ignore_errors=True)
+        with open(path, "w") as handle:
+            json.dump({"1": {"candidate_steps": steps}}, handle)
+        return path
+
+    def test_republishing_recomputes_the_bound(self):
+        path = self._write_config([2])
+        get_context().set_server_args(
+            _FakeResolvedArgs(
+                speculative_num_draft_tokens=3,
+                speculative_adaptive=True,
+                speculative_adaptive_config=path,
+            )
+        )
+        self.assertEqual(max_speculative_num_draft_tokens(), 3)
+
+        with open(path, "w") as handle:
+            json.dump({"1": {"candidate_steps": [4]}}, handle)
+        # Same path, new contents: the memo must not survive the republish.
+        get_context().set_server_args(
+            _FakeResolvedArgs(
+                speculative_num_draft_tokens=3,
+                speculative_adaptive=True,
+                speculative_adaptive_config=path,
+            )
+        )
+        self.assertEqual(max_speculative_num_draft_tokens(), 5)
+
+    def test_reset_clears_the_bound(self):
+        path = self._write_config([2])
+        get_context().set_server_args(
+            _FakeResolvedArgs(
+                speculative_num_draft_tokens=3,
+                speculative_adaptive=True,
+                speculative_adaptive_config=path,
+            )
+        )
+        self.assertEqual(max_speculative_num_draft_tokens(), 3)
+        reset_context()
+        with open(path, "w") as handle:
+            json.dump({"1": {"candidate_steps": [6]}}, handle)
+        get_context().set_server_args(
+            _FakeResolvedArgs(
+                speculative_num_draft_tokens=3,
+                speculative_adaptive=True,
+                speculative_adaptive_config=path,
+            )
+        )
+        self.assertEqual(max_speculative_num_draft_tokens(), 7)
+
+
+class TestNamedAccessorsCallWhatTheyWrap(CustomTestCase):
+    """A named accessor must *call* a member that is a method.
+
+    `return get_server_args().x` hands back a bound method when `x` is defined
+    with `def`; the failure then lands far away, in whatever arithmetic the
+    caller does with it. Checked statically so accessors that need a real model
+    config are covered too.
+    """
+
+    def test_accessors_that_wrap_methods_call_them(self):
+        import ast
+        import functools
+        import inspect
+
+        import sglang.srt.runtime_context as rc
+        from sglang.srt.server_args import ServerArgs
+
+        tree = ast.parse(inspect.getsource(rc))
+        wrong = []
+        for node in tree.body:
+            if not isinstance(node, ast.FunctionDef):
+                continue
+            for inner in ast.walk(node):
+                if not (isinstance(inner, ast.Return) and inner.value is not None):
+                    continue
+                value = inner.value
+                called = isinstance(value, ast.Call)
+                target = value.func if called else value
+                if not (
+                    isinstance(target, ast.Attribute)
+                    and isinstance(target.value, ast.Call)
+                    and isinstance(target.value.func, ast.Name)
+                    and target.value.func.id == "get_server_args"
+                ):
+                    continue
+                member = getattr(ServerArgs, target.attr, None)
+                # A `property` / `functools.cached_property` member is already
+                # evaluated by the attribute access, so it is named here to keep
+                # the failure message from calling it "not a method" -- the fix
+                # for those is the opposite one.
+                kind = (
+                    "a property"
+                    if isinstance(member, (property, functools.cached_property))
+                    else "not a method"
+                )
+                if inspect.isfunction(member) and not called:
+                    wrong.append(
+                        f"{node.name}(): returns ServerArgs.{target.attr} without "
+                        "calling it, so callers get a bound method"
+                    )
+                if not inspect.isfunction(member) and called:
+                    wrong.append(
+                        f"{node.name}(): calls ServerArgs.{target.attr}, which is "
+                        f"{kind} -- the attribute access already produced the value"
+                    )
+        self.assertEqual([], wrong, "\n".join(wrong))
+
+
 if __name__ == "__main__":
     unittest.main()