[CI] Move the static ratchets back to CPU unit tests (#34913)

.claude/rules/unit-test-admission.md

@@ -26,9 +26,8 @@ A new unit test case must fall into one of these categories:
 3. **Critical-path bookkeeping.** Defends conventions that are easy to break by
    forgetting to sync -- registry completeness, field lifecycle, serialization
    compatibility. Enumerating assertions are fine here; the guarded failure
-   mode is "someone extended X without updating Y". Example: the namespace
+   mode is "someone extended X without updating Y". Example: the ratchet tests
-   coverage tests (`test/registered/unit/test_server_args_namespaces.py`). Static
+   (`test/registered/unit/test_module_state_ratchet.py`).
-   source ratchets belong in `scripts/lint/` as checkers, not unit tests.
 
 Not admissible:
 

.claude/skills/sglang-runtime-context/SKILL.md

@@ -136,7 +136,7 @@ requires today). A process-global seed field-read of one of these
 sizes (`get_server_args().tp_size`, or an alias of it) is a read-ratchet failure; the
 sites that legitimately go around the live property are the `configured_*_size()`
 readers, and those are what the ratchet registers, each with its reason
-(`_CONFIGURED_SIZE_CALL_SITES` in `scripts/lint/check_global_config_read_ratchet.py`). A
+(`_CONFIGURED_SIZE_CALL_SITES` in `test_global_config_read_ratchet.py`). A
 `server_args` the object was *handed* is a different thing and not a ratchet
 matter — see "Reads that legitimately stay on a ServerArgs instance".
 Fail-loud is narrower: before dist init, a live size/group read raises — except
@@ -189,11 +189,11 @@ this).
   topology (`1` / `False` when no group is installed). A site
   that must know the *requested* DCP size before dist init needs its own
   `configured_dcp_size()` (and an entry in `_CONFIGURED_SIZE_CALL_SITES`, which lives
-  in the ratchet checker, not in this skill); note the live pair does not *need*
+  in the ratchet test, not in this skill); note the live pair does not *need*
   dist init — with no group it answers `1` / `False` — it just cannot answer
   with the requested size. Every (file, accessor) pair is registered
-  with its reason in `scripts/lint/check_global_config_read_ratchet.py`
+  with its reason in `test_global_config_read_ratchet.py`
-  (`_CONFIGURED_SIZE_CALL_SITES`), and that checker fails if the code and the list
+  (`_CONFIGURED_SIZE_CALL_SITES`), and that test fails if the code and the list
   disagree — a new file, or a new accessor in a listed file, has to be added — so a new site needs both an answer the live property cannot give and
   an entry saying what it is.
 - **this runner's resolved value** → the runner
@@ -378,31 +378,31 @@ ONE thread — do not design for TBO threads that don't exist.
    resolved config with `get_context().override`; hand a per-runner value to its
    runner as a constructor argument. Projected bags are sealed the same way (leaf
    assignment raises).
-2. **Mutation guard** (`scripts/lint/check_server_args_mutation_ratchet.py`, pinned at 0 over the whole
+2. **Mutation ratchet** (`test_server_args_mutation_ratchet.py`, exact pin 0 over the whole
    package minus the pipeline / multimodal_gen): textual scan for assignment forms. Never
    raise the baseline.
 3. **No-copy contract** (`test_server_args_no_instance_mutation_entry.py`): neither
    `ServerArgs.override` nor `ServerArgs.derive` exists, and nothing in the package
    calls either form. Rerouting a writer to the bags means flipping **all its readers
    in the same commit** (no transitional dual-write).
-4. **Legacy-accessor ratchet** (`scripts/lint/check_legacy_global_ratchet.py`): `get_global_server_args`
+4. **Legacy-accessor ratchet** (`test_legacy_global_ratchet.py`): `get_global_server_args`
    call sites must not grow. The replacement for a *decision* is a bag leaf, a named
    accessor, or the owning runner's stamp — not `get_server_args().field`, which the
    read ratchet below pins at zero. `runtime_context.get_server_args()` is only for the
    whole-object shapes (dumps, provenance, a hand-off to a callee that takes a config).
-5. **Global config read ratchet** (`scripts/lint/check_global_config_read_ratchet.py`): baselines are
+5. **Global config read ratchet** (`test_global_config_read_ratchet.py`): baselines are
    **0** for both the direct `get_server_args().field` and the alias form (function-local
    — including local copies of an alias, `cfg = sa` — module-level, or parked on an
    instance attribute, plus the `getattr(..., "field")` spelling of each; a name
    computed at runtime or indirection deeper than a local name copy is census-tool
-   territory, per the checker's module docstring). The scanners match `get_server_args` and
+   territory, per the test's docstring). The scanners match `get_server_args` and
    `configured_*_size` by their literal names, and the same file *bans*
    `import ... as` renames of them so that matching stays sound. Exempt by owner
    module only (`runtime_context.py`, `server_args.py`, `arg_groups/`). The same file
    carries `_CONFIGURED_SIZE_CALL_SITES`, the (file, accessor) map of every
    `configured_*_size()` reader with the reason the live property cannot serve it — a new
    file or a new accessor in a listed file must be added there.
-6. **Module-state ratchet** (`scripts/lint/check_module_state_ratchet.py`): `global` statements in the
+6. **Module-state ratchet** (`test_module_state_ratchet.py`): `global` statements in the
    flag-owning layers are pinned by name. A new module-level runtime global belongs on a
    flags group / resources slot instead; migrating a pinned survivor must shrink the pin.
 7. **Namespace coverage** (`test_server_args_namespaces.py`,
@@ -456,9 +456,9 @@ Key source files: `python/sglang/srt/runtime_context.py` (the container, every t
 `publish`, `_ConfigBag`, `preserve_config`, `override_server_args`),
 `python/sglang/srt/arg_groups/overrides.py` (override registry, passes,
 `declare_late_resolution`), `python/sglang/srt/server_args.py` (`NS` metadata,
-`Arg(..., resolvable=True)`, `__setattr__` strict guard), the static guardrails under
+`Arg(..., resolvable=True)`, `__setattr__` strict guard), and the guardrail tests under
-`scripts/lint/` (`check_server_args_mutation_ratchet.py`,
+`test/registered/unit/` (`test_server_args_mutation_ratchet.py`,
-`check_global_config_read_ratchet.py`, `check_legacy_global_ratchet.py`,
+`test_global_config_read_ratchet.py`, `test_legacy_global_ratchet.py`,
-`check_module_state_ratchet.py`), and the runtime guardrail tests under
+`test_module_state_ratchet.py`, `test_server_args_namespaces.py`,
-`test/registered/unit/` (`test_server_args_namespaces.py`, `test_runtime_context.py` — the latter doubles
+`test_runtime_context.py` — the last one doubles
 as executable documentation of every tier's semantics).

.pre-commit-config.yaml

@@ -114,12 +114,6 @@ repos:
         entry: python3 scripts/lint/check_no_bare_pytest_main.py
         language: system
         files: ^(python|test)/.*\.py$
-      - id: check-static-ratchets
-        name: validate static runtime ratchets
-        entry: python3 scripts/lint/check_static_ratchets.py
-        language: system
-        files: ^(python/sglang/.*\.py|scripts/lint/check_.*\.py)$
-        pass_filenames: false
       - id: check-lint-script-tests
         name: unit tests for lint checkers
         entry: python3 -m unittest discover -s scripts/lint -p 'test_check_*.py'

python/sglang/srt/server_args.py

@@ -9599,7 +9599,7 @@ def m3_fp8_attn_gemm_enabled(args) -> bool:
 # (sglang.srt.runtime_context). The two functions below are LEGACY shims kept
 # for the existing call-sites; they publish/read the same live object by
 # reference. Do not add new call-sites — the counts are ratcheted
-# (decrease-only) by scripts/lint/check_legacy_global_ratchet.py.
+# (decrease-only) by test/registered/unit/test_legacy_global_ratchet.py.
 # Imports are in-function so the two modules stay cycle-free at import time.
 def set_global_server_args_for_scheduler(server_args: ServerArgs):
     """Legacy publish shim (role=scheduler) — prefer

scripts/lint/check_legacy_global_ratchet.py

@@ -1,52 +0,0 @@
-"""Ratchet guard: legacy global-accessor call-sites may only decrease.
-
-The process-wide ``ServerArgs`` is owned by the runtime context; the legacy
-``get_global_server_args`` / ``set_global_server_args_for_*`` names survive as
-thin shims for the existing call-sites. New code should use the
-``sglang.srt.runtime_context`` accessors (``get_server_args()`` /
-``get_context().set_server_args()``), so the shim call-site counts below must
-never grow. When your change removes call-sites, lower the matching baseline
-to the new count.
-"""
-
-import re
-from pathlib import Path
-
-_SRT_ROOT = Path(__file__).resolve().parents[2] / "python" / "sglang" / "srt"
-
-# Baselines counted over python/sglang/srt/**/*.py, including each function's
-# own def line. Ratchet: decrease-only.
-_RATCHETS = [
-    # Down to the shim definition itself; every call-site now goes through
-    # runtime_context.get_server_args().
-    ("get_global_server_args", r"\bget_global_server_args\s*\(", 1),
-    (
-        "set_global_server_args_for_*",
-        r"\bset_global_server_args_for_(?:scheduler|tokenizer)\s*\(",
-        4,
-    ),
-]
-
-
-def check_legacy_global_ratchet():
-    sources = [
-        path.read_text(encoding="utf-8", errors="replace")
-        for path in sorted(_SRT_ROOT.rglob("*.py"))
-    ]
-    for name, pattern, baseline in _RATCHETS:
-        count = sum(len(re.findall(pattern, source)) for source in sources)
-        if count > baseline:
-            raise AssertionError(
-                f"{name} call-sites grew: {count} > baseline {baseline}. "
-                "New code must use the sglang.srt.runtime_context accessors "
-                "(get_server_args() / get_context().set_server_args())."
-            )
-        if count < baseline:
-            raise AssertionError(
-                f"{name} call-sites shrank: {count} < baseline {baseline}. "
-                "Lower the baseline in this file to lock in the progress."
-            )
-
-
-if __name__ == "__main__":
-    check_legacy_global_ratchet()

scripts/lint/check_module_state_ratchet.py

@@ -1,52 +0,0 @@
-"""Ratchet guard: module-level runtime state in the flag-owning layers may
-only shrink.
-
-Runtime flags belong on ``get_flags()`` groups, which have lifecycle reset and
-a scoped test-override primitive; a module-level ``global`` has neither and
-leaks across test teardowns. The pin below names the survivors -- migrating one
-must shrink it.
-"""
-
-import ast
-from pathlib import Path
-
-_SRT_ROOT = Path(__file__).resolve().parents[2] / "python" / "sglang" / "srt"
-
-_PINNED_GLOBALS = {
-    "layers/moe/utils.py": frozenset(),
-    "layers/dp_attention.py": frozenset(
-        {
-            # DP-attention topology (parallel vertical scope).
-            "_ATTN_DP_RANK",
-            "_ATTN_DP_SIZE",
-        }
-    ),
-}
-
-
-def check_module_state_ratchet():
-    for rel, pinned in _PINNED_GLOBALS.items():
-        tree = ast.parse((_SRT_ROOT / rel).read_text())
-        declared = {
-            name
-            for node in ast.walk(tree)
-            if isinstance(node, ast.Global)
-            for name in node.names
-        }
-        grown = declared - pinned
-        if grown:
-            raise AssertionError(
-                f"{rel} declares new module-level runtime state {sorted(grown)}; "
-                "put runtime flags on a get_flags() group instead "
-                "(see runtime_context.MoeFlags / DpFlags).",
-            )
-        shrunk = pinned - declared
-        if shrunk:
-            raise AssertionError(
-                f"{rel} no longer declares {sorted(shrunk)}; "
-                "shrink the pin in this file to lock in the progress.",
-            )
-
-
-if __name__ == "__main__":
-    check_module_state_ratchet()

scripts/lint/check_server_args_mutation_ratchet.py

@@ -1,57 +0,0 @@
-"""Guard: no server_args mutation outside the resolution pipeline, pinned at 0.
-
-``ServerArgs.__setattr__`` already raises on a bare assignment after
-resolution; this static scan is what reaches the sites tests never execute.
-"""
-
-import re
-from pathlib import Path
-
-_SGLANG_ROOT = Path(__file__).resolve().parents[2] / "python" / "sglang"
-
-# Assignments to a server_args attribute (``server_args.x = ...``,
-# ``self.server_args.x = ...``, and the ``sa`` alias used by a few helpers).
-# ``==`` comparisons are excluded by the negative lookahead.
-_MUTATION_PATTERNS = [
-    # (?![=}]) skips ``==`` comparisons and f-string ``{x=}`` debug specs.
-    re.compile(r"\bserver_args\.[a-z0-9_]+\s*=(?![=}])"),
-    re.compile(r"\bsa\.[a-z0-9_]+\s*=(?![=}])"),
-    re.compile(r"get_(?:global_)?server_args\(\)\.[a-z0-9_]+\s*=(?![=}])"),
-    # setattr is the same write with the attribute name behind a variable.
-    re.compile(
-        r"setattr\(\s*(?:[\w.]+\.)?(?:server_args|sa|get_(?:global_)?server_args\(\))\s*,"
-    ),
-]
-
-# The resolution pipeline itself (mutation is its job) and multimodal_gen,
-# whose ServerArgs is a different class outside this contract.
-_EXCLUDED = (
-    "srt/server_args.py",
-    "srt/arg_groups",
-    "multimodal_gen",
-)
-
-_BASELINE = 0
-
-
-def check_server_args_mutation_ratchet():
-    count = 0
-    for path in sorted(_SGLANG_ROOT.rglob("*.py")):
-        rel = path.relative_to(_SGLANG_ROOT).as_posix()
-        if rel.startswith(_EXCLUDED):
-            continue
-        source = path.read_text()
-        count += sum(len(pattern.findall(source)) for pattern in _MUTATION_PATTERNS)
-    if count > _BASELINE:
-        raise AssertionError(
-            f"server_args mutations outside the resolution pipeline grew: "
-            f"{count} > baseline {_BASELINE}. Configuration is resolved in "
-            "ServerArgs.__post_init__; declare through the pipeline "
-            "(passes / declare_late_resolution), change resolved config "
-            "with get_context().override(source, ...), or hand the value "
-            "to its runner as a constructor argument — do not assign fields."
-        )
-
-
-if __name__ == "__main__":
-    check_server_args_mutation_ratchet()

scripts/lint/check_static_ratchets.py

@@ -1,48 +0,0 @@
-#!/usr/bin/env python3
-"""Runs every static ratchet in one process, so the package is parsed once."""
-
-import sys
-
-from check_decode_bookkeeping_ownership import (
-    check_bookkeeping_sites_match_owner_allowlist,
-    check_spec_v2_draft_workers_do_no_scheduler_bookkeeping,
-)
-from check_global_config_read_ratchet import (
-    check_configured_size_call_sites,
-    check_global_config_read_ratchet,
-    check_no_renamed_accessor_imports,
-)
-from check_legacy_global_ratchet import check_legacy_global_ratchet
-from check_module_state_ratchet import check_module_state_ratchet
-from check_parallel_adoption_ratchet import check_parallel_adoption_ratchet
-from check_server_args_mutation_ratchet import check_server_args_mutation_ratchet
-
-
-def main():
-    checks = (
-        check_bookkeeping_sites_match_owner_allowlist,
-        check_spec_v2_draft_workers_do_no_scheduler_bookkeeping,
-        check_global_config_read_ratchet,
-        check_configured_size_call_sites,
-        check_no_renamed_accessor_imports,
-        check_legacy_global_ratchet,
-        check_module_state_ratchet,
-        check_parallel_adoption_ratchet,
-        check_server_args_mutation_ratchet,
-    )
-    # They guard independent invariants, so one failure must not hide the rest.
-    failures = []
-    for check in checks:
-        try:
-            check()
-        except AssertionError as exc:
-            failures.append(f"[{check.__name__}] {exc}")
-
-    for failure in failures:
-        print(failure, file=sys.stderr)
-        print(file=sys.stderr)
-    return 1 if failures else 0
-
-
-if __name__ == "__main__":
-    sys.exit(main())

test/registered/unit/spec/test_decode_bookkeeping_ownership.py

@@ -12,11 +12,17 @@ leak checker, hence this AST-level guard.
 """
 
 import ast
+import unittest
 import warnings
 from collections import Counter
 from pathlib import Path
 
-_REPO_ROOT = Path(__file__).resolve().parents[2]
+from sglang.test.ci.ci_register import register_cpu_ci
+from sglang.test.test_utils import CustomTestCase
+
+register_cpu_ci(est_time=8, suite="base-a-test-cpu")
+
+_REPO_ROOT = Path(__file__).resolve().parents[4]
 _SRT_DIR = _REPO_ROOT / "python" / "sglang" / "srt"
 _SPECULATIVE_DIR = _SRT_DIR / "speculative"
 assert _SRT_DIR.is_dir(), f"srt dir not found: {_SRT_DIR}"
@@ -187,41 +193,40 @@ def _scan_class_subtree(class_node):
     return sites
 
 
-def check_bookkeeping_sites_match_owner_allowlist():
+class TestDecodeBookkeepingOwnership(CustomTestCase):
-    found = _scan_srt()
+    def test_bookkeeping_sites_match_owner_allowlist(self):
-    allow = Counter(_OWNER_SITES)
+        found = _scan_srt()
-    unexpected = found - allow
+        allow = Counter(_OWNER_SITES)
-    missing = allow - found
+        unexpected = found - allow
-    messages = []
+        missing = allow - found
-    if unexpected:
+        msg = []
-        messages.append(
+        if unexpected:
-            "New bookkeeping mutation(s) beyond the recorded counts:\n  "
+            msg.append(
-            + "\n  ".join(f"{site} x{n}" for site, n in sorted(unexpected.items()))
+                "New bookkeeping mutation(s) beyond the recorded counts:\n  "
-            + "\nThese are owned by the sites in _OWNER_SITES -- do not "
+                + "\n  ".join(f"{site} x{n}" for site, n in sorted(unexpected.items()))
-            "repeat them; a genuinely new owner must be recorded there."
+                + "\nThese are owned by the sites in _OWNER_SITES -- do not "
-        )
+                "repeat them; a genuinely new owner must be recorded there."
-    if missing:
+            )
-        messages.append(
+        if missing:
-            "Recorded site(s) no longer exist (update _OWNER_SITES):\n  "
+            msg.append(
-            + "\n  ".join(f"{site} x{n}" for site, n in sorted(missing.items()))
+                "Recorded site(s) no longer exist (update _OWNER_SITES):\n  "
-        )
+                + "\n  ".join(f"{site} x{n}" for site, n in sorted(missing.items()))
-    if messages:
+            )
-        raise AssertionError("\n\n".join(messages))
+        self.assertFalse(msg, "\n\n".join(msg))
 
+    def test_spec_v2_draft_workers_do_no_scheduler_bookkeeping(self):
+        classes = _draft_worker_classes()
+        names = {node.name for _, node in classes}
+        # Discovery sanity: fail loudly instead of silently guarding nothing.
+        self.assertIn("EagleDraftWorker", names)
+        self.assertIn("FrozenKVMTPDraftWorker", names)
 
-def check_spec_v2_draft_workers_do_no_scheduler_bookkeeping():
+        violations = []
-    classes = _draft_worker_classes()
+        for rel, node in classes:
-    names = {node.name for _, node in classes}
+            for scope, kind in _scan_class_subtree(node):
-    for expected in ("EagleDraftWorker", "FrozenKVMTPDraftWorker"):
+                violations.append((rel, f"{node.name}.{scope}", kind))
-        if expected not in names:
+        self.assertFalse(
-            raise AssertionError(f"draft worker discovery missed {expected}")
+            violations,
-
-    violations = []
-    for rel, node in classes:
-        for scope, kind in _scan_class_subtree(node):
-            violations.append((rel, f"{node.name}.{scope}", kind))
-    if violations:
-        raise AssertionError(
             "Spec-v2 draft worker(s) repeat scheduler-owned bookkeeping:\n  "
             + "\n  ".join(map(str, sorted(violations)))
             + "\nUnder spec v2 the iter-clock ticks, `maybe_evict_swa`, and "
@@ -231,5 +236,4 @@ def check_spec_v2_draft_workers_do_no_scheduler_bookkeeping():
 
 
 if __name__ == "__main__":
-    check_bookkeeping_sites_match_owner_allowlist()
+    unittest.main(verbosity=3)
-    check_spec_v2_draft_workers_do_no_scheduler_bookkeeping()

test/registered/unit/test_global_config_read_ratchet.py

@@ -1,24 +1,47 @@
-"""Guard: business code never reads a config field off the process-global record.
+"""Ratchet guard: process-global config reads may only decrease.
 
-``get_server_args()`` returns the published ``ServerArgs`` -- one process's
+``get_server_args()`` returns the published ``ServerArgs`` — one process's
 startup record. Config decisions read the namespace accessors instead
-(``get_exec()`` / ``get_memory()`` / ...); per-runner values come from the
+(``get_exec()`` / ``get_memory()`` / …), which carry the resolved value
-runner that owns them. Both baselines are zero, over the whole package minus
+including post-publish overrides, and per-runner values come from the runner
-the modules that own the slot.
+that owns them.
 
-The scanners match ``get_server_args`` and ``configured_*_size`` by their
+Business code no longer reads the published record for a config value at all:
-literal names, which is why import-renaming them is banned below. A name
+both baselines are zero, over the whole package minus the modules that own the
+slot.
+
+The reads that remain live in ``runtime_context.py`` (exempt by module): the
+``@property`` / method members computed from several fields plus the HF config,
+which are not namespace leaves and have no home but ``ServerArgs``, and the
+``configured_*_size()`` accessors for the sizes ``get_parallel()`` shadows with
+the live topology. ``_CONFIGURED_SIZE_CALL_SITES`` registers every one of the
+latter with the reason the live property cannot serve it.
+
+What the scan sees: ``get_server_args().field``, an alias (``sa =
+get_server_args()`` then ``sa.field`` -- function-local, module-level, or parked
+on an instance attribute), a local copy of an alias (``cfg = sa``), and the
+``getattr(<either>, "field")`` spelling of each. It matches the accessors by
+their literal names, which is why import-renaming them is banned below. A name
 computed at runtime, or indirection deeper than a local name copy, is invisible
-here -- the census tool in the context repo audits that shape.
+here -- the census tool in the context repo audits that shape. 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.
 """
 
+from sglang.test.ci.ci_register import register_cpu_ci
+
+register_cpu_ci(est_time=5, suite="base-a-test-cpu")
+
 import ast
-from functools import cache
+import unittest
 from pathlib import Path
 
+import sglang
+from sglang.test.test_utils import CustomTestCase
+
 # srt is the migrated surface; the rest of the package has no reads today and is
 # scanned so a new one cannot appear there unnoticed.
-_PACKAGE_ROOT = Path(__file__).resolve().parents[2] / "python" / "sglang"
+_PACKAGE_ROOT = Path(next(iter(sglang.__path__)))
 
 # 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
@@ -26,7 +49,7 @@ _PACKAGE_ROOT = Path(__file__).resolve().parents[2] / "python" / "sglang"
 _SLOT_OWNERS = ("srt/runtime_context.py", "srt/server_args.py", "srt/arg_groups/")
 
 # Every call site of a ``configured_*_size()`` accessor, with the reason the
-# live topology cannot answer there. The checker below asserts this map is exactly
+# 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"): (
@@ -343,24 +366,16 @@ def _collect(rel: str, tree: ast.AST, inert: frozenset = frozenset()):
     return direct, alias
 
 
-@cache
+def _field_reads():
-def _parsed_modules():
+    direct, alias = [], []
-    """(rel, tree) per parseable module; the three scanners below share it."""
-    modules = []
     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
-        modules.append((path.relative_to(_PACKAGE_ROOT).as_posix(), tree))
-    return modules
-
-
-def _field_reads():
-    direct, alias = [], []
-    for rel, tree in _parsed_modules():
-        if rel.startswith(_SLOT_OWNERS):
-            continue
         inert = frozenset(name for path_, name in _INERT_DYNAMIC_READS if path_ == rel)
         module_direct, module_alias = _collect(rel, tree, inert)
         direct += module_direct
@@ -368,61 +383,29 @@ def _field_reads():
     return direct, alias
 
 
-def _configured_size_call_sites():
+class TestGlobalConfigReadRatchet(CustomTestCase):
-    found = set()
+    def _check(self, kind, reads, baseline):
-    for rel, tree in _parsed_modules():
+        if len(reads) > baseline:
-        if rel.startswith(_SLOT_OWNERS):
+            self.fail(
-            continue
+                f"{kind} process-global config field reads grew: {len(reads)} > "
-        for node in ast.walk(tree):
+                f"baseline {baseline}. Read the namespace accessor for the "
-            if not isinstance(node, ast.Call):
+                "field's namespace, or the owning runner for a per-runner "
-                continue
+                "field:\n" + "\n".join(reads)
-            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"):
+        if len(reads) < baseline:
-                found.add((rel, name))
+            self.fail(
-    return found
+                f"{kind} process-global config field reads shrank: {len(reads)} < "
+                f"baseline {baseline}. Lower the baseline in this file to lock "
+                "in the progress."
+            )
+
+    def test_global_field_reads_match_the_baseline(self):
+        direct, alias = _field_reads()
+        self._check("direct", direct, _DIRECT_BASELINE)
+        self._check("alias-form", alias, _ALIAS_BASELINE)
 
 
-def _renamed_accessor_imports():
+class TestConfiguredSizeCallSites(CustomTestCase):
-    offenders = []
-    for rel, tree in _parsed_modules():
-        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 {imported.asname}"
-                    )
-    return offenders
-
-
-def _check_count(kind, reads, baseline):
-    if len(reads) > baseline:
-        raise AssertionError(
-            f"{kind} process-global config field reads grew: {len(reads)} > "
-            f"baseline {baseline}. Read the namespace accessor for the "
-            "field's namespace, or the owning runner for a per-runner "
-            "field:\n" + "\n".join(reads)
-        )
-
-
-def check_global_config_read_ratchet():
-    direct, alias = _field_reads()
-    _check_count("direct", direct, _DIRECT_BASELINE)
-    _check_count("alias-form", alias, _ALIAS_BASELINE)
-
-
-def check_configured_size_call_sites():
     """The configured-vs-live exceptions are enumerated, with reasons.
 
     ``configured_*_size()`` answers what the user asked for where
@@ -437,17 +420,38 @@ def check_configured_size_call_sites():
     what this catches -- in either call form (bare or module-qualified).
     """
 
-    found = _configured_size_call_sites()
+    def test_the_call_sites_match_the_documented_set(self):
-    documented = set(_CONFIGURED_SIZE_CALL_SITES)
+        found = set()
-    if documented != found:
+        for path in sorted(_PACKAGE_ROOT.rglob("*.py")):
-        raise AssertionError(
+            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)}",
         )
 
 
-def check_no_renamed_accessor_imports():
+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
@@ -455,9 +459,30 @@ def check_no_renamed_accessor_imports():
     so it is banned outright — which is exactly what makes literal-name
     matching sound."""
 
-    offenders = _renamed_accessor_imports()
+    def test_the_scanned_accessors_are_never_import_renamed(self):
-    if offenders:
+        offenders = []
-        raise AssertionError(
+        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 "
@@ -466,6 +491,4 @@ def check_no_renamed_accessor_imports():
 
 
 if __name__ == "__main__":
-    check_global_config_read_ratchet()
+    unittest.main()
-    check_configured_size_call_sites()
-    check_no_renamed_accessor_imports()

test/registered/unit/test_legacy_global_ratchet.py

@@ -0,0 +1,65 @@
+"""Ratchet guard: legacy global-accessor call-sites may only decrease.
+
+The process-wide ``ServerArgs`` is owned by the runtime context; the legacy
+``get_global_server_args`` / ``set_global_server_args_for_*`` names survive as
+thin shims for the existing call-sites. New code should use the
+``sglang.srt.runtime_context`` accessors (``get_server_args()`` /
+``get_context().set_server_args()``), so the shim call-site counts below must
+never grow. When your change removes call-sites, lower the matching baseline
+to the new count.
+"""
+
+from sglang.test.ci.ci_register import register_cpu_ci
+
+register_cpu_ci(est_time=5, suite="base-a-test-cpu")
+
+import re
+import unittest
+from pathlib import Path
+
+import sglang.srt
+from sglang.test.test_utils import CustomTestCase
+
+_SRT_ROOT = Path(next(iter(sglang.srt.__path__)))
+
+# Baselines counted over python/sglang/srt/**/*.py, including each function's
+# own def line. Ratchet: decrease-only.
+_RATCHETS = [
+    # Down to the shim definition itself; every call-site now goes through
+    # runtime_context.get_server_args().
+    ("get_global_server_args", r"\bget_global_server_args\s*\(", 1),
+    (
+        "set_global_server_args_for_*",
+        r"\bset_global_server_args_for_(?:scheduler|tokenizer)\s*\(",
+        4,
+    ),
+]
+
+
+class TestLegacyGlobalRatchet(CustomTestCase):
+    def test_legacy_accessor_call_sites_match_the_baselines(self):
+        # Exact pin, failing in BOTH directions: a grown count means new code
+        # bypassed the runtime_context accessors; a shrunk count means a
+        # removal forgot to lower the baseline, which would let later changes
+        # silently re-add call-sites up to the stale ceiling.
+        sources = [
+            path.read_text(encoding="utf-8", errors="replace")
+            for path in sorted(_SRT_ROOT.rglob("*.py"))
+        ]
+        for name, pattern, baseline in _RATCHETS:
+            count = sum(len(re.findall(pattern, source)) for source in sources)
+            if count > baseline:
+                self.fail(
+                    f"{name} call-sites grew: {count} > baseline {baseline}. "
+                    "New code must use the sglang.srt.runtime_context accessors "
+                    "(get_server_args() / get_context().set_server_args())."
+                )
+            if count < baseline:
+                self.fail(
+                    f"{name} call-sites shrank: {count} < baseline {baseline}. "
+                    "Lower the baseline in this file to lock in the progress."
+                )
+
+
+if __name__ == "__main__":
+    unittest.main()

test/registered/unit/test_module_state_ratchet.py

@@ -0,0 +1,61 @@
+"""Ratchet guard: module-level runtime state in the flag-owning layers may
+only shrink.
+
+Runtime flags belong on ``get_flags()`` groups, which have lifecycle reset and
+a scoped test-override primitive; a module-level ``global`` has neither and
+leaks across test teardowns. ``_PINNED_GLOBALS`` names the survivors --
+migrating one must shrink it, adding one fails the ratchet.
+"""
+
+from sglang.test.ci.ci_register import register_cpu_ci
+
+register_cpu_ci(est_time=5, suite="base-a-test-cpu")
+
+import ast
+import unittest
+from pathlib import Path
+
+import sglang.srt
+from sglang.test.test_utils import CustomTestCase
+
+_SRT_ROOT = Path(next(iter(sglang.srt.__path__)))
+
+_PINNED_GLOBALS = {
+    "layers/moe/utils.py": frozenset(),
+    "layers/dp_attention.py": frozenset(
+        {
+            # DP-attention topology (parallel vertical scope).
+            "_ATTN_DP_RANK",
+            "_ATTN_DP_SIZE",
+        }
+    ),
+}
+
+
+class TestModuleStateRatchet(CustomTestCase):
+    def test_global_statements_match_the_pins(self):
+        for rel, pinned in _PINNED_GLOBALS.items():
+            tree = ast.parse((_SRT_ROOT / rel).read_text())
+            declared = {
+                name
+                for node in ast.walk(tree)
+                if isinstance(node, ast.Global)
+                for name in node.names
+            }
+            grown = declared - pinned
+            self.assertFalse(
+                grown,
+                f"{rel} declares new module-level runtime state {sorted(grown)}; "
+                "put runtime flags on a get_flags() group instead "
+                "(see runtime_context.MoeFlags / DpFlags).",
+            )
+            shrunk = pinned - declared
+            self.assertFalse(
+                shrunk,
+                f"{rel} no longer declares {sorted(shrunk)}; "
+                "shrink the pin in this file to lock in the progress.",
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()

test/registered/unit/test_parallel_adoption_ratchet.py

@@ -1,4 +1,5 @@
-"""Guard: no legacy parallel-getter calls in the swept directories.
+"""Ratchet guard: legacy parallel-getter calls in swept directories may only
+shrink.
 
 ``models/`` and ``layers/`` read parallel topology through
 ``get_parallel().<dim>`` (the read-through wrapper in ``runtime_context``),
@@ -7,10 +8,18 @@ primitive. Exemptions are pinned in ``_EXEMPT``, each with its reason; sweeping
 one must remove it from there.
 """
 
+from sglang.test.ci.ci_register import register_cpu_ci
+
+register_cpu_ci(est_time=5, suite="base-a-test-cpu")
+
 import re
+import unittest
 from pathlib import Path
 
-_SRT_ROOT = Path(__file__).resolve().parents[2] / "python" / "sglang" / "srt"
+import sglang.srt
+from sglang.test.test_utils import CustomTestCase
+
+_SRT_ROOT = Path(next(iter(sglang.srt.__path__)))
 
 _BANNED_CALLS = re.compile(
     r"\b(?:dcp_enabled|get_(?:"
@@ -42,22 +51,23 @@ _EXEMPT = (
 )
 
 
-def check_parallel_adoption_ratchet():
+class TestParallelAdoptionRatchet(CustomTestCase):
-    offenders = []
+    def test_no_legacy_parallel_getters_in_swept_dirs(self):
-    for top in _SWEPT_DIRS:
+        offenders = []
-        for path in sorted((_SRT_ROOT / top).rglob("*.py")):
+        for top in _SWEPT_DIRS:
-            rel = path.relative_to(_SRT_ROOT).as_posix()
+            for path in sorted((_SRT_ROOT / top).rglob("*.py")):
-            if rel.startswith(_EXEMPT):
+                rel = path.relative_to(_SRT_ROOT).as_posix()
-                continue
+                if rel.startswith(_EXEMPT):
-            for line_number, line in enumerate(path.read_text().split("\n"), 1):
+                    continue
-                if _BANNED_CALLS.search(line):
+                for i, line in enumerate(path.read_text().split("\n"), 1):
-                    offenders.append(f"{rel}:{line_number}")
+                    if _BANNED_CALLS.search(line):
-    if offenders:
+                        offenders.append(f"{rel}:{i}")
-        raise AssertionError(
+        self.assertFalse(
+            offenders,
             "legacy parallel-getter calls in swept directories (use "
             f"get_parallel().<dim> instead): {offenders}",
         )
 
 
 if __name__ == "__main__":
-    check_parallel_adoption_ratchet()
+    unittest.main()

test/registered/unit/test_server_args_mutation_ratchet.py

@@ -0,0 +1,81 @@
+"""Ratchet guard: server_args mutations outside the resolution pipeline may
+only decrease.
+
+After ``ServerArgs.__post_init__`` returns, the instance carries the resolved
+configuration and the resolution pipeline (``server_args.py`` and
+``arg_groups/``) is the only place that computes it: resolved config changes go
+to the context bags via ``get_context().override(source, **fields)``, and a
+value one runner or worker owns travels as a constructor argument. The baseline
+is therefore an exact pin at zero -- new mutations must not appear, and removals
+must lower it.
+
+``ServerArgs.__setattr__`` already raises on a bare assignment after
+resolution; this textual scan is what reaches the sites tests never execute.
+"""
+
+from sglang.test.ci.ci_register import register_cpu_ci
+
+register_cpu_ci(est_time=5, suite="base-a-test-cpu")
+
+import re
+import unittest
+from pathlib import Path
+
+import sglang
+from sglang.test.test_utils import CustomTestCase
+
+_SGLANG_ROOT = Path(next(iter(sglang.__path__)))
+
+# Assignments to a server_args attribute (``server_args.x = ...``,
+# ``self.server_args.x = ...``, and the ``sa`` alias used by a few helpers).
+# ``==`` comparisons are excluded by the negative lookahead.
+_MUTATION_PATTERNS = [
+    # (?![=}]) skips ``==`` comparisons and f-string ``{x=}`` debug specs.
+    re.compile(r"\bserver_args\.[a-z0-9_]+\s*=(?![=}])"),
+    re.compile(r"\bsa\.[a-z0-9_]+\s*=(?![=}])"),
+    re.compile(r"get_(?:global_)?server_args\(\)\.[a-z0-9_]+\s*=(?![=}])"),
+    # setattr is the same write with the attribute name behind a variable.
+    re.compile(
+        r"setattr\(\s*(?:[\w.]+\.)?(?:server_args|sa|get_(?:global_)?server_args\(\))\s*,"
+    ),
+]
+
+# The resolution pipeline itself (mutation is its job) and multimodal_gen,
+# whose ServerArgs is a different class outside this contract.
+_EXCLUDED = (
+    "srt/server_args.py",
+    "srt/arg_groups",
+    "multimodal_gen",
+)
+
+_BASELINE = 0
+
+
+class TestServerArgsMutationRatchet(CustomTestCase):
+    def test_out_of_pipeline_mutations_match_the_baseline(self):
+        count = 0
+        for path in sorted(_SGLANG_ROOT.rglob("*.py")):
+            rel = path.relative_to(_SGLANG_ROOT).as_posix()
+            if rel.startswith(_EXCLUDED):
+                continue
+            source = path.read_text()
+            count += sum(len(p.findall(source)) for p in _MUTATION_PATTERNS)
+        if count > _BASELINE:
+            self.fail(
+                f"server_args mutations outside the resolution pipeline grew: "
+                f"{count} > baseline {_BASELINE}. Configuration is resolved in "
+                "ServerArgs.__post_init__; declare through the pipeline "
+                "(passes / declare_late_resolution), change resolved config "
+                "with get_context().override(source, ...), or hand the value "
+                "to its runner as a constructor argument — do not assign fields."
+            )
+        if count < _BASELINE:
+            self.fail(
+                f"server_args mutations outside the resolution pipeline "
+                f"shrank: {count} < baseline {_BASELINE}. Lower the baseline "
+                "in this file to lock in the progress."
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()