config: spell out the one dynamic config read the census could not see

`_is_dsa_active` asked `getattr(server_args, "_is_dsa_model_arch", False)`, and
that name has never existed on `ServerArgs` -- it arrived as a placeholder with
the CP strategy abstractions (#27313), so the getattr default has always decided
the predicate. A dynamic read of a name nothing sets is the one shape the config
census cannot follow, and it looked like a live decision while being dead.

Spelled as the constant it evaluates to, with the placeholder written down: what
it should ask (whether this process runs a DSA model arch) is the CP path's
call, and its only consumer, `ContextParallelStrategy.per_layer_attn_cp_comm`,
has no readers yet.

That was the sole entry in the read ratchet's `_INERT_DYNAMIC_READS`, so the
exemption list is gone with it -- there is no way to exempt a read from the
baselines any more, which is the invariant worth having. The `counted()`
indirection it existed for goes too (verified the three shapes it guarded still
report: direct, `getattr`, and an attribute-parked alias).

python/sglang/srt/layers/cp/base.py

@@ -230,16 +230,8 @@ class ContextParallelStrategy(ABC):
 
 
 def _is_dsa_active() -> bool:
-    from sglang.srt.runtime_context import get_parallel, get_server_args
-
-    # `_is_dsa_model_arch` is set nowhere in the tree, so this predicate is
-    # inert today (the getattr default makes it False). Kept verbatim rather
-    # than "fixed" here, because deciding what it should name is the CP path's
-    # call; the ratchet exempts it with that reason.
-    return bool(
-        get_parallel().enable_prefill_cp
-        and getattr(get_server_args(), "_is_dsa_model_arch", False)
-    )
+    # Placeholder: a real answer needs the model architecture, not config.
+    return False
 
 
 _STRATEGY: Optional[ContextParallelStrategy] = None

python/sglang/srt/server_args.py

@@ -3527,6 +3527,9 @@ class ServerArgs:
     ] = None
 
     def __post_init__(self):
+        self._run_resolution_pipeline()
+
+    def _run_resolution_pipeline(self):
         """
         Orchestrates the handling of various server arguments, ensuring proper configuration and validation.
 

test/registered/unit/server_args/test_resolution_is_reproducible.py

@@ -402,5 +402,59 @@ class TestResolutionIsReproducible(CustomTestCase):
         self.assertEqual(getattr(first, "_resolved_overrides", None), first_provenance)
 
 
+class TestTheResolutionSeamHasOneCaller(CustomTestCase):
+    """The pipeline is entered from exactly one place.
+
+    Step 12 moves the call from ``__post_init__`` to ``publish`` so the record
+    stays raw; that is a one-line move only while the seam has a single caller.
+    A second entry point would also mean resolution could run twice on one
+    instance, which the strict ``__setattr__`` guard turns into an
+    ``AttributeError`` rather than a silent re-resolve.
+    """
+
+    def test_only_post_init_runs_the_pipeline(self):
+        import ast
+        from pathlib import Path
+
+        import sglang
+
+        package_root = Path(next(iter(sglang.__path__)))
+        callers = []
+        for path in sorted(package_root.rglob("*.py")):
+            try:
+                tree = ast.parse(path.read_text())
+            except SyntaxError:
+                continue
+            # The full (class, function, ...) scope chain, so the assertion can
+            # say "the one caller is ServerArgs.__post_init__" -- not merely
+            # that nothing outside a function named __post_init__ calls it.
+            scopes = {}
+            for node in ast.walk(tree):
+                own = scopes.get(id(node), ())
+                if isinstance(
+                    node, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)
+                ):
+                    own = own + (node.name,)
+                for child in ast.iter_child_nodes(node):
+                    scopes[id(child)] = own
+            for node in ast.walk(tree):
+                if (
+                    isinstance(node, ast.Call)
+                    and isinstance(node.func, ast.Attribute)
+                    and node.func.attr == "_run_resolution_pipeline"
+                ):
+                    rel = path.relative_to(package_root).as_posix()
+                    callers.append((rel, ".".join(scopes.get(id(node), ()))))
+        # Every call, compared whole: a removed call, a duplicate inside
+        # __post_init__, or another class growing a same-named __post_init__
+        # all show up here.
+        self.assertEqual(
+            [("srt/server_args.py", "ServerArgs.__post_init__")],
+            callers,
+            "the resolution pipeline must be entered exactly once, from "
+            f"ServerArgs.__post_init__; found: {callers}",
+        )
+
+
 if __name__ == "__main__":
     unittest.main()

test/registered/unit/test_global_config_read_ratchet.py

@@ -84,12 +84,6 @@ _CONFIGURED_SIZE_CALL_SITES = {
     ),
 }
 
-# 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
 
@@ -105,18 +99,10 @@ def _is_global_call(node) -> bool:
     return isinstance(func, ast.Attribute) and func.attr == "get_server_args"
 
 
-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.
-    """
+def _collect(rel: str, tree: ast.AST):
+    """The (direct, alias) field reads in one module."""
     direct, alias = [], []
 
-    def counted(attr: str) -> bool:
-        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."""
@@ -132,20 +118,15 @@ def _collect(rel: str, tree: ast.AST, inert: frozenset = frozenset()):
         return node.args[1].value
 
     for node in ast.walk(tree):
-        if (
-            isinstance(node, ast.Attribute)
-            and _is_global_call(node.value)
-            and counted(node.attr)
-        ):
+        if isinstance(node, ast.Attribute) and _is_global_call(node.value):
             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):
+        if name is not None and _is_global_call(node.args[0]):
             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):
             # ``sa = get_server_args()`` and its annotated form
@@ -193,7 +174,6 @@ def _collect(rel: str, tree: ast.AST, inert: frozenset = frozenset()):
                 and isinstance(inner.value, ast.Name)
                 and inner.value.id in bound
                 and inner.lineno >= bound[inner.value.id]
-                and counted(inner.attr)
             ):
                 alias.append(
                     f"{rel}:{inner.lineno}: {inner.value.id}.{inner.attr} "
@@ -205,7 +185,6 @@ def _collect(rel: str, tree: ast.AST, inert: frozenset = frozenset()):
                 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}) "
@@ -301,7 +280,7 @@ def _collect(rel: str, tree: ast.AST, inert: frozenset = frozenset()):
                 ):
                     base, attr = node.args[0].id, attr_name
                     shown = f"getattr({base}, {attr!r})"
-            if base and not _shadowed(node, base) and counted(attr):
+            if base and not _shadowed(node, base):
                 alias.append(
                     f"{rel}:{node.lineno}: {shown} "
                     f"(module-level bind from get_server_args() at line "
@@ -349,13 +328,13 @@ def _collect(rel: str, tree: ast.AST, inert: frozenset = frozenset()):
             key = shown = None
             if isinstance(inner, ast.Attribute):
                 key = _bound_attr(inner.value)
-                if key is not None and counted(inner.attr):
+                if key is not None:
                     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):
+                    if key is not None:
                         shown = f"getattr({key[0]}.{key[1]}, {name!r})"
             if shown is not None:
                 alias.append(
@@ -376,8 +355,7 @@ def _field_reads():
             tree = ast.parse(path.read_text())
         except SyntaxError:
             continue
-        inert = frozenset(name for path_, name in _INERT_DYNAMIC_READS if path_ == rel)
-        module_direct, module_alias = _collect(rel, tree, inert)
+        module_direct, module_alias = _collect(rel, tree)
         direct += module_direct
         alias += module_alias
     return direct, alias