minke.yu/sglang
1
0
Fork 0
Code Issues Pull Requests Actions 13 Packages Projects Releases Wiki Activity

config: ServerArgs holds the raw input (#36255)

Browse Source
This commit is contained in:
Cheng Wan
2026-08-26 05:14:05 -07:00
committed by GitHub
parent 27c36368b6
commit 413df1f8db
34 changed files with 1201 additions and 383 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/registered/unit/server_args/test_model_config_reads_resolved_input.py
+2 -2
View File
@@ -26,8 +26,8 @@ register_cpu_ci(est_time=5, suite="base-a-test-cpu")
_SRT = pathlib.Path(sglang.__file__).resolve().parent / "srt"
# Read for what the caller asked for: the constructor passes it through and
# never stores it, while resolution later overwrites the field with the value
# the architecture implies. Two quantities sharing one name.
# never stores it, while resolution declares the value the architecture implies.
# Two quantities sharing one name.
_READ_BEFORE_RESOLUTION = frozenset({"is_embedding"})
# Declared after the first `get_model_config()`, so the cached configuration
test/registered/unit/server_args/test_resolution_declarations.py
+84 -46
View File
@@ -5,13 +5,11 @@ so a resolution write that only assigns the field is invisible to it. Every
resolver declares now -- the record's handlers through `self._declare`, the
hooks and hardware defaults through `declare_resolution` -- and that is pinned
two ways: no bare assignment to a field survives anywhere a ServerArgs instance
is in reach, and after resolution every declared field agrees with what the
stash says. The second check is the one that keeps the transition honest --
while a declaration still writes the field immediately, a stash entry and a
field can only disagree if something assigned the field behind the stash's
back. A third check runs the other way: every field resolution moved has to
be explained by the stash, which covers the spellings a source scan cannot
see.
is in reach, and after resolution `resolution_result` answers for every declared
field with what the stash holds. The second check is what the stash is measured
against: the two can disagree only if something wrote behind the stash's back. A
third check runs the other way -- every field resolution moved has to be
explained by the stash, which covers the spellings a source scan cannot see.
"""
import ast
@@ -233,6 +231,11 @@ def _bare_assignments():
return sorted(found)
def shape_key(shape):
"""A shape rendered short enough for a failure message."""
return ",".join(f"{k}={v}" for k, v in sorted(shape.items())) or "defaults"
def _stash_overlay(server_args):
"""What the declarations say, last writer wins -- the projection's input."""
overlay = {}
@@ -345,35 +348,43 @@ class TestResolutionDeclarations(CustomTestCase):
+ "\n ".join(unexplained),
)
def test_the_projection_input_is_the_resolved_configuration(self):
"""What the bags are built from equals what the record ends up holding.
def test_a_declaration_only_resolver_leaves_the_field_alone(self):
"""The direction of travel: resolution decides, the record does not move.
The projection reads `raw input + declarations` rather than the
fields, so that it keeps working when the declarations stop
materializing. While they still do, the two have to agree leaf for
leaf -- a difference means the projection would publish something the
record does not say, which is the failure this whole transition is
meant to avoid.
A resolver that only declares -- a model-specific override, a registry
entry -- writes nothing onto the record. The projection carries its
answer and the field still holds what the caller passed.
"""
from sglang.srt.arg_groups.arg_utils import namespace_of
from sglang.srt.arg_groups.overrides import resolution_result
differences = []
found = []
for shape in _SHAPES:
server_args = self._resolve(shape)
raw = getattr(server_args, "_raw_input", None) or {}
for field in namespace_of(type(server_args)):
projected = resolution_result(server_args, field)
if field not in raw:
continue
decided = resolution_result(server_args, field)
on_record = getattr(server_args, field)
if projected != on_record:
differences.append(
f"{shape} -> {field}: projection={projected!r} "
f"record={on_record!r}"
)
self.assertEqual(
differences,
if decided == on_record:
continue
# It moved away from the record's value, so the record must
# still hold exactly what the caller passed.
self.assertEqual(
on_record,
raw[field],
f"{shape} -> {field}: the record holds {on_record!r}, which "
f"is neither the raw input {raw[field]!r} nor what "
f"resolution decided ({decided!r})",
)
found.append((shape_key(shape), field))
self.assertNotEqual(
found,
[],
"the projection and the record disagree about a config leaf:\n "
+ "\n ".join(differences),
"no field is resolved by declaration alone any more, so this check "
"no longer covers anything -- either the shapes stopped reaching "
"one or the declarations are writing the fields again",
)
def test_the_whole_object_readback_carries_only_fields(self):
@@ -561,10 +572,9 @@ class TestResolutionDeclarations(CustomTestCase):
The parser detection and the LoRA normalization run at launcher stage --
they need a tokenizer, a chat template, an adapter directory -- and they
write through `declare_late_resolution`. If those writes only reached
the fields, the bags would describe the *unresolved* value: a server
launched with `--reasoning-parser auto` would advertise and apply
`auto` after detection had already replaced it.
declare through `declare_late_resolution`. The declaration is the only
home for what they decide: the record keeps `--reasoning-parser auto`,
and the bags a process publishes carry the detected parser.
A real model path, not the dummy one: a dummy record never materializes,
so its `resolve_once` re-runs and re-snapshots the raw input from
@@ -586,15 +596,22 @@ class TestResolutionDeclarations(CustomTestCase):
)
publish(server_args, role="tokenizer")
self.assertEqual(get_serving().reasoning_parser, "qwen3")
self.assertEqual(server_args.reasoning_parser, get_serving().reasoning_parser)
self.assertEqual(
server_args.reasoning_parser,
"auto",
"the record is the operator's input; late resolution declares, it "
"does not write back",
)
def test_validation_can_still_resolve_before_the_record_is_published(self):
"""The LoRA checks normalize in place, so they must precede publish.
"""The LoRA checks resolve, so they must precede publish.
`check_server_args` is not read-only: it infers `enable_lora`, parses
adapter paths and normalizes target modules through late resolution,
which a published record refuses. The launcher order is what keeps this
legal, and this is the assertion that notices if it moves.
legal, and this is the assertion that notices if it moves. What those
declarations decide reaches the bags; the record keeps the raw form the
operator passed.
"""
from sglang.srt.runtime_context import get_lora, publish, reset_context
@@ -608,9 +625,17 @@ class TestResolutionDeclarations(CustomTestCase):
self.addCleanup(reset_context)
server_args.check_server_args()
publish(server_args, role="tokenizer")
self.assertEqual(get_lora().enable_lora, server_args.enable_lora)
self.assertEqual(
get_lora().lora_target_modules, server_args.lora_target_modules
get_lora().enable_lora, resolution_result(server_args, "enable_lora")
)
self.assertEqual(
get_lora().lora_target_modules,
resolution_result(server_args, "lora_target_modules"),
)
self.assertEqual(
server_args.lora_target_modules,
["q_proj"],
"normalization is a declaration; the record keeps what was passed",
)
def test_the_launcher_finishes_resolving_before_it_publishes(self):
@@ -661,24 +686,37 @@ class TestResolutionDeclarations(CustomTestCase):
f"written:\n " + "\n ".join(too_late),
)
def test_the_stash_agrees_with_the_fields_it_declared(self):
mismatches = []
def test_an_undeclared_field_still_holds_the_raw_input(self):
"""Nothing writes a field behind the stash's back.
Comparing the stash against `resolution_result` would agree by
construction -- both are the same last-writer-wins walk over
`_resolved_overrides`, spelled forwards and backwards. The independent
source is the record's own `_raw_input` snapshot: a field with no
declaration has to still equal what the caller passed, because the only
sanctioned way to move one is to declare it.
"""
moved = []
for shape in _SHAPES:
server_args = self._resolve(shape)
overlay = _stash_overlay(server_args)
for field, declared in overlay.items():
if field not in _RESOLVED_FIELDS:
raw_input = getattr(server_args, "_raw_input", None)
self.assertTrue(raw_input, f"{shape}: the record kept no raw snapshot")
for field in dataclasses.fields(server_args):
name = field.name
if name in overlay or name not in raw_input:
continue
actual = getattr(server_args, field)
if actual != declared:
mismatches.append(
f"{shape} -> {field}: field={actual!r} stash={declared!r}"
current = getattr(server_args, name, None)
if current != raw_input[name]:
moved.append(
f"{shape} -> {name}: raw={raw_input[name]!r} "
f"field={current!r}"
)
self.assertEqual(
mismatches,
moved,
[],
"a declared field and its stash entry disagree, so something "
"assigned the field behind the declaration:\n " + "\n ".join(mismatches),
"these fields moved without a declaration, so the bags publish one "
"value while the record shows another:\n " + "\n ".join(moved),
)
def test_no_immediate_writer_overrides_a_deferred_one(self):
test/registered/unit/server_args/test_resolution_is_reproducible.py
+25 -13
View File
@@ -768,31 +768,43 @@ class TestACopyStaysResolved(_RestoresProcessState, CustomTestCase):
server_args.resolve_once()
return server_args
def test_a_bare_replace_would_resolve_a_second_time(self):
"""Why the helper exists. If this stops drifting, the pipeline became
idempotent and the helper's reason is gone -- read it again before
deleting either."""
def test_a_bare_replace_resolves_again_and_lands_in_the_same_place(self):
"""A bare copy resolves to the same place: the fields are the raw input.
`dataclasses.replace` copies the fields, so a bare copy re-runs
resolution over the *same input* the parent got -- the DP-attention
halving and the conservativeness scaling apply once. `replace_resolved`
buys something else: it carries the parent's declarations and its
`model_config`, so the copy answers without resolving at all.
"""
parent = self._resolved()
bare = dataclasses.replace(parent, dist_init_addr="1.2.3.4:5000")
self.assertFalse(
getattr(bare, "_declarations_materialized", False),
getattr(bare, "_resolution_finished", False),
"a bare replace carried the flag; then this test proves nothing",
)
bare.resolve_once()
drifted = {
field.name: (
resolution_result(parent, field.name),
resolution_result(bare, field.name),
)
for field in dataclasses.fields(parent)
if field.name not in ("dist_init_addr", "random_seed")
and repr(resolution_result(parent, field.name))
!= repr(resolution_result(bare, field.name))
}
self.assertEqual(
(bare.chunked_prefill_size, round(bare.schedule_conservativeness, 4)),
(
parent.chunked_prefill_size // 2,
round(parent.schedule_conservativeness * 0.3, 4),
),
"the second pass no longer drifts; this is the drift the copy "
"helper exists to avoid",
drifted,
{},
"resolving a bare copy landed somewhere else, so the pipeline is "
"reading its own output again",
)
def test_replace_resolved_keeps_the_parents_resolution(self):
parent = self._resolved()
copy_ = parent.replace_resolved("ray.test", dist_init_addr="1.2.3.4:5000")
self.assertTrue(getattr(copy_, "_declarations_materialized", False))
self.assertTrue(getattr(copy_, "_resolution_finished", False))
drifted = {
field.name: (getattr(parent, field.name), getattr(copy_, field.name))
for field in dataclasses.fields(parent)
test/registered/unit/server_args/test_resolution_reads_the_declarations.py
+612
View File
@@ -0,0 +1,612 @@
"""Resolution reads its own decisions, not the record's fields.
`declare_resolution` records a decision in the declaration stash and writes
nothing. The fields keep what the caller passed, so a resolver that reads a
field another resolver may have decided reads the raw input -- silently, and
only on the configurations where that other resolver fires. The whole pipeline
therefore reads through `resolving_view` (or `ServerArgs._resolved()`, which is
the same view spelled as the record's own member), and this pins that there is
nothing left reading a field directly.
Subjects: every function in `arg_groups/` that takes a config, every
`ServerArgs` handler the dispatcher reaches, and every member of `ServerArgs` /
`PortArgs` -- the members are reached from the hooks and from business code,
which the handler walk cannot see, and a member that recomputes from a raw field
decides from what was typed. All three
are derived -- a new hook file, a new handler or a new member is covered the
moment it is written. Readers *outside* those
two -- the platform defaults, `ModelConfig`, the spec-algo hook -- are reached by
resolution too and have moved to the view as well, but enumerating them needs
the call-graph derivation `test_resolution_reads_no_bag` owns; this file pins
the two scopes it can derive exactly.
"""
import ast
import dataclasses
import pathlib
import re
import sglang
from sglang.srt.server_args import ServerArgs
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")
_SRT = pathlib.Path(sglang.__file__).resolve().parent / "srt"
_FIELDS = frozenset(field.name for field in dataclasses.fields(ServerArgs))
# Names a config travels under. `args` is included because the platform hooks
# use it; a false positive would be a function taking an argparse Namespace and
# reading an attribute that happens to be a ServerArgs field name, which the
# allowlist below would then have to carry.
_HOLDER_NAMES = frozenset({"server_args", "sa", "args"})
def _holders(fn):
names = {
arg.arg
for arg in list(fn.args.posonlyargs)
+ list(fn.args.args)
+ list(fn.args.kwonlyargs)
if arg.arg in _HOLDER_NAMES
}
for arg in (
list(fn.args.posonlyargs) + list(fn.args.args) + list(fn.args.kwonlyargs)
):
annotation = arg.annotation
text = (
annotation.value
if isinstance(annotation, ast.Constant)
else (
annotation.id
if isinstance(annotation, ast.Name)
else annotation.attr if isinstance(annotation, ast.Attribute) else None
)
)
if text == "ServerArgs":
names.add(arg.arg)
return names
def _field_reads(fn, holders):
for node in ast.walk(fn):
if (
isinstance(node, ast.Attribute)
and node.attr in _FIELDS
and isinstance(node.value, ast.Name)
and node.value.id in holders
and isinstance(node.ctx, ast.Load)
):
yield node.lineno, node.attr
def _resolution_handlers():
"""The `ServerArgs` methods the dispatcher reaches, transitively."""
tree = ast.parse((_SRT / "server_args.py").read_text(encoding="utf-8-sig"))
cls = next(
node
for node in ast.walk(tree)
if isinstance(node, ast.ClassDef) and node.name == "ServerArgs"
)
methods = {
node.name: node
for node in cls.body
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
}
assert "_run_resolution_pipeline" in methods, "the dispatcher was renamed"
seen, stack = set(), ["_run_resolution_pipeline"]
while stack:
name = stack.pop()
if name in seen or name not in methods:
continue
seen.add(name)
for node in ast.walk(methods[name]):
if (
isinstance(node, ast.Call)
and isinstance(node.func, ast.Attribute)
and isinstance(node.func.value, ast.Name)
and node.func.value.id == "self"
):
stack.append(node.func.attr)
return {name: methods[name] for name in seen}
_DECLARERS = frozenset(
{
"_declare",
"declare_resolution",
"declare_late_resolution",
"declare_direct_writes",
}
)
def _declared_fields():
"""The fields resolution decides, read off every shape that reaches the stash.
A keyword on a `declare_*` call is only one shape: the model-override and
post-process passes build a mapping instead (`MODEL_OVERRIDES` literals,
`overrides["dtype"] = ...`, a returned dict), and late resolution splats a
variable-keyed one. Deriving from keywords alone leaves nineteen fields
outside the subject set, `dtype` and `reasoning_parser` among them.
"""
fields = set()
# The declaration calls live wherever a resolver does; the mapping channels
# only exist where the override providers and post-process passes are.
keyword_sources = [_SRT / "server_args.py"]
for sub in ("arg_groups", "hardware_backend", "parser"):
keyword_sources += sorted((_SRT / sub).rglob("*.py"))
mapping_sources = {_SRT / "server_args.py", *(_SRT / "arg_groups").rglob("*.py")}
for path in keyword_sources:
tree = ast.parse(path.read_text(encoding="utf-8-sig"))
for node in ast.walk(tree):
# 1. `declare_resolution(sa, src, page_size=64)` and its siblings
if isinstance(node, ast.Call):
name = (
node.func.id
if isinstance(node.func, ast.Name)
else getattr(node.func, "attr", None)
)
if name in _DECLARERS:
for keyword in node.keywords:
if keyword.arg:
fields.add(keyword.arg)
elif isinstance(keyword.value, ast.Dict):
fields.update(_string_keys(keyword.value))
# 2. every mapping literal in the files that declare through one:
# the MODEL_OVERRIDES tables, the dicts the override providers
# return, the ones the post-process passes build. Scanning
# unrelated files here would collect a plain kwarg dict
# (`tokenizer_config={"trust_remote_code": ...}`) and turn a
# passthrough read into a violation.
if isinstance(node, ast.Dict) and path in mapping_sources:
fields.update(_string_keys(node))
# 3. `overrides["field"] = ...`
if (
path in mapping_sources
and isinstance(node, ast.Assign)
and isinstance(node.targets[0], ast.Subscript)
and isinstance(node.targets[0].slice, ast.Constant)
and isinstance(node.targets[0].slice.value, str)
):
fields.add(node.targets[0].slice.value)
return frozenset(fields & _FIELDS)
def _string_keys(node: ast.Dict) -> set:
return {
key.value
for key in node.keys
if isinstance(key, ast.Constant) and isinstance(key.value, str)
}
def _record_members():
"""Every member of `ServerArgs` / `PortArgs`, by class and name."""
tree = ast.parse((_SRT / "server_args.py").read_text(encoding="utf-8-sig"))
members = {}
for node in ast.walk(tree):
if isinstance(node, ast.ClassDef) and node.name in ("ServerArgs", "PortArgs"):
for member in node.body:
if isinstance(member, (ast.FunctionDef, ast.AsyncFunctionDef)):
members[f"{node.name}.{member.name}"] = member
return members
def _config_reading_helpers():
"""Module functions that load a decided field off the config they are handed.
A member that hands them `self`, or a call site that hands them a record,
reads the raw input through the callee -- the shape neither an attribute
scan nor a `getattr` scan can see, because the field name is spelled in the
helper and the record is spelled at the call site.
"""
decided = _declared_fields()
helpers = {}
sources = [_SRT / "server_args.py"] + sorted((_SRT / "arg_groups").rglob("*.py"))
for path in sources:
tree = ast.parse(path.read_text(encoding="utf-8-sig"))
for fn in ast.walk(tree):
if not isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)):
continue
params = {
arg.arg
for arg in list(fn.args.posonlyargs)
+ list(fn.args.args)
+ list(fn.args.kwonlyargs)
} - {"self", "cls"}
if not params:
continue
reads = {
node.attr
for node in ast.walk(fn)
if isinstance(node, ast.Attribute)
and isinstance(node.value, ast.Name)
and node.value.id in params
and isinstance(node.ctx, ast.Load)
and node.attr in decided
}
if reads:
helpers[fn.name] = sorted(reads)
return helpers
# The accessors that hand back the process-global record itself. A helper that
# is handed one of these reads the raw input exactly as a bare `self` would.
_RECORD_ACCESSORS = frozenset({"get_server_args", "global_server_args"})
# `self._server_args` is the same record under a private name; the scan has to
# see it or a reader inside the context object escapes every shape above.
_RECORD_ATTR = re.compile(r"^_*(server_args|sa)$")
def _record_arguments(node, aliases=frozenset()):
"""The bare-record arguments of a call.
Four spellings reach a helper with a record: the bare name (`self`, `sa`),
an attribute (`runner.server_args`), the process-global accessor called
inline (`get_server_args()`), and a local bound to either of the last two
earlier in the same function.
"""
out = []
for arg in node.args:
if isinstance(arg, ast.Name) and arg.id in ("self", "server_args", "sa"):
out.append(arg.id)
elif isinstance(arg, ast.Attribute) and _RECORD_ATTR.match(arg.attr or ""):
out.append(ast.unparse(arg))
elif (
isinstance(arg, ast.Call)
and isinstance(arg.func, ast.Name)
and arg.func.id in _RECORD_ACCESSORS
):
out.append(ast.unparse(arg))
elif isinstance(arg, ast.Name) and arg.id in aliases:
out.append(arg.id)
return out
def _record_aliases(function):
"""Locals bound to the record under a name of their own.
`_sa = getattr(runner, "server_args", None)`, `cfg = get_server_args()` and
`engine_args = ServerArgs.from_cli_args(args)` all put the record behind a
name the argument scan does not recognise, so a later
`getattr(_sa, "<decided leaf>")` reads what the operator typed.
"""
aliases = set()
for node in ast.walk(function):
if not isinstance(node, ast.Assign) or len(node.targets) != 1:
continue
target = node.targets[0]
if not isinstance(target, ast.Name):
continue
value = node.value
if isinstance(value, ast.Attribute):
if _RECORD_ATTR.match(value.attr or ""):
aliases.add(target.id)
continue
if not isinstance(value, ast.Call):
continue
func = value.func
if isinstance(func, ast.Name):
if func.id in _RECORD_ACCESSORS or func.id == "ServerArgs":
aliases.add(target.id)
elif (
func.id == "getattr"
and len(value.args) >= 2
and isinstance(value.args[1], ast.Constant)
and _RECORD_ATTR.match(str(value.args[1].value))
):
aliases.add(target.id)
elif (
isinstance(func, ast.Attribute)
and func.attr in ("from_cli_args", "replace_resolved")
and isinstance(func.value, ast.Name)
and func.value.id == "ServerArgs"
):
aliases.add(target.id)
return aliases
# The one reader for which the raw field is the right answer. The gateway sizes
# its worker pool from the operator's requested replica count; `--dwdp-size`
# makes resolution declare a `dp_size` describing one multi-rank server's
# internal topology, so reading the decision there would spawn dp_size
# single-rank children and ask for dp_size^2 GPUs. A new entry here needs that
# kind of reason next to it.
_NO_RESOLVED_SURFACE = frozenset(
{
(
"sgl-model-gateway/bindings/python/src/sglang_router/launch_server.py",
"server_args.dp_size",
),
}
)
def _is_record_base(node, aliases):
"""Is this expression the record itself?
A local bound to one, a parameter that carries one (`server_args`, `sa`,
`engine_args`), or an attribute holding one (`self._server_args`).
"""
if isinstance(node, ast.Name):
return node.id in aliases
if isinstance(node, ast.Attribute):
return bool(_RECORD_ATTR.match(node.attr or ""))
return False
def _record_handoff_offenders(rel, tree, helpers, decided, is_record=False):
"""Every way a decided leaf is reached through a record in one module.
Two shapes, both scanned under the record aliases the function binds:
handing the record to a helper that loads a decided field, and loading one
off the alias directly (`alias.<leaf>` or `getattr(alias, "<leaf>")`). The
second is what the MiniMax backend spelled, and an argument scan cannot see
it -- the leaf never appears at a call site.
"""
offenders, seen = [], set()
def record(lineno, text):
if (lineno, text) in seen:
return
seen.add((lineno, text))
offenders.append(f"{rel}:{lineno} {text}")
scopes = [(tree, frozenset())] + [
(fn, _record_aliases(fn))
for fn in ast.walk(tree)
if isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef))
]
for scope, aliases in scopes:
for node in ast.walk(scope):
if isinstance(node, ast.Call):
name = (
node.func.id
if isinstance(node.func, ast.Name)
else getattr(node.func, "attr", None)
)
if name in helpers:
for arg in _record_arguments(node, aliases):
if arg == "self" and not is_record:
continue
record(
node.lineno,
f"{name}({arg}) reads {', '.join(helpers[name])}",
)
if (
isinstance(node.func, ast.Name)
and node.func.id == "getattr"
and len(node.args) >= 2
and isinstance(node.args[0], ast.Name)
and node.args[0].id in aliases
and isinstance(node.args[1], ast.Constant)
and node.args[1].value in decided
):
record(
node.lineno,
f'getattr({node.args[0].id}, "{node.args[1].value}")',
)
elif (
isinstance(node, ast.Attribute)
and isinstance(node.ctx, ast.Load)
and node.attr in decided
and _is_record_base(node.value, aliases)
):
record(node.lineno, f"{ast.unparse(node.value)}.{node.attr}")
return offenders
# Source the scanner must read the same way whether or not the tree happens to
# contain these shapes today. The first four are the spellings that reached
# production and were converted; the last two are the legal forms next to them,
# which have to stay quiet or the guard is unusable.
_SPELLINGS = """
def hands_the_alias_to_a_helper(runner):
_sa = getattr(runner, "server_args", None)
return m3_fp8_attn_gemm_enabled(_sa)
def loads_a_leaf_off_the_alias(runner):
_sa = getattr(runner, "server_args", None)
return getattr(_sa, "speculative_num_draft_tokens", None)
def reads_a_leaf_through_the_alias(runner):
sa_local = runner.server_args
return sa_local.attention_backend
def hands_the_accessor_to_a_helper():
return compute_world_size(get_server_args())
def reads_the_view(runner):
cfg = resolving_view(runner.server_args)
return cfg.attention_backend
def reads_an_undecided_leaf(runner):
_sa = runner.server_args
return _sa.tp_size
def reads_a_leaf_off_a_private_attribute(self):
return self._server_args.attention_backend
def reads_a_leaf_off_a_constructed_record(cli):
engine_args = ServerArgs.from_cli_args(cli)
engine_args.resolve_once()
return engine_args.attention_backend
"""
class TestResolutionReadsTheDeclarations(CustomTestCase):
def test_no_hook_reads_a_field_off_the_record(self):
offenders = []
files = sorted((_SRT / "arg_groups").glob("*.py"))
self.assertGreater(len(files), 5, "the hook scan found almost nothing")
for path in files:
rel = f"arg_groups/{path.name}"
tree = ast.parse(path.read_text(encoding="utf-8-sig"))
for fn in ast.walk(tree):
if not isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)):
continue
holders = _holders(fn)
if not holders:
continue
for lineno, field in _field_reads(fn, holders):
offenders.append(f"{rel}:{lineno} {fn.name} reads .{field}")
self.assertEqual(
offenders,
[],
"a resolution hook reads a field off the record; the field holds the "
"raw input, so this decides from what was typed rather than from "
"what resolution decided. Read `resolving_view(server_args)`:\n "
+ "\n ".join(offenders),
)
def test_no_handler_reads_a_field_off_self(self):
handlers = _resolution_handlers()
self.assertGreater(
len(handlers), 50, f"only {len(handlers)} handlers were reached"
)
offenders = []
for name, fn in sorted(handlers.items()):
for lineno, field in _field_reads(fn, {"self"}):
offenders.append(f"server_args.py:{lineno} {name} reads self.{field}")
self.assertEqual(
offenders,
[],
"a resolution handler reads its own field; the field holds the raw "
"input. Bind `cfg = resolving_view(self)` and read that:\n "
+ "\n ".join(offenders),
)
def test_no_member_recomputes_from_a_raw_field(self):
decided = _declared_fields()
self.assertGreater(
len(decided), 100, f"the declaration set derived only {len(decided)} fields"
)
members = _record_members()
self.assertGreater(len(members), 100, f"only {len(members)} members were found")
offenders = []
for name, fn in sorted(members.items()):
holders = _holders(fn) | {"self"}
for lineno, field in _field_reads(fn, holders):
if field in decided:
offenders.append(f"server_args.py:{lineno} {name} reads .{field}")
self.assertEqual(
offenders,
[],
"a record member recomputes from a field resolution decides; the "
"field holds the raw input, so the member answers for what was "
"typed. Bind `cfg = resolving_view(self)` and read that:\n "
+ "\n ".join(offenders),
)
def test_no_reader_hands_the_record_to_a_config_helper(self):
helpers = _config_reading_helpers()
self.assertGreater(
len(helpers), 5, f"the helper derivation found only {len(helpers)}"
)
decided = _declared_fields()
offenders = []
# `scripts/`, `examples/` and the gateway binding are outside the
# package but hold records they resolve themselves, and every reader
# this scan found in them was reading a field resolution fills in.
_REPO = _SRT.parent.parent.parent
roots = (
[_SRT]
+ [_SRT.parent / d for d in ("benchmark", "lang")]
+ [
_REPO / d
for d in (
"scripts",
"examples",
"sgl-model-gateway/bindings/python/src",
)
]
)
for root in roots:
if not root.exists():
continue
for path in sorted(root.rglob("*.py")):
# Package files keep their `srt/...` spelling (the skips below
# key on it); the repo-level roots are named from the repo.
try:
rel = path.relative_to(_SRT.parent).as_posix()
except ValueError:
rel = path.relative_to(_REPO).as_posix()
if rel.startswith(("srt/arg_groups/", "multimodal_gen/")):
continue
try:
tree = ast.parse(path.read_text(encoding="utf-8-sig"))
except SyntaxError:
continue
offenders += _record_handoff_offenders(
rel, tree, helpers, decided, is_record=rel == "srt/server_args.py"
)
offenders = [
line
for line in offenders
if (line.split(":", 1)[0], line.split(" ", 1)[1])
not in _NO_RESOLVED_SURFACE
]
self.assertEqual(
offenders,
[],
"a caller hands the record to a helper that loads a field "
"resolution decides; the helper then reads the raw input. Hand it "
"`resolving_view(record)` (or the published bag):\n "
+ "\n ".join(offenders),
)
def test_the_scan_sees_every_spelling_that_reached_production(self):
"""Every spelling that reached production, pinned next to the scanner.
A shape the scan stops seeing is a silent hole, so each one is listed
here with the legal forms beside it and the flagged set compared
exactly.
What it does not reach: a record that arrives as a *parameter* and was
resolved by the caller (`scripts/playground/bench_speculative.py` hands
`main(args, server_args)` one). Binding that would need the call graph,
and naming a parameter `server_args` is also how the resolution-time
readers spell a view.
"""
helpers = _config_reading_helpers()
decided = _declared_fields()
for name in ("m3_fp8_attn_gemm_enabled", "compute_world_size"):
self.assertIn(name, helpers, f"the helper derivation lost {name}")
for field in ("speculative_num_draft_tokens", "attention_backend"):
self.assertIn(field, decided, f"the declared set lost {field}")
offenders = _record_handoff_offenders(
"sample.py", ast.parse(_SPELLINGS), helpers, decided
)
flagged = {line.split(" ", 1)[1] for line in offenders}
self.assertEqual(
flagged,
{
"m3_fp8_attn_gemm_enabled(_sa)"
" reads " + ", ".join(helpers["m3_fp8_attn_gemm_enabled"]),
'getattr(_sa, "speculative_num_draft_tokens")',
"sa_local.attention_backend",
"self._server_args.attention_backend",
"engine_args.attention_backend",
"compute_world_size(get_server_args())"
" reads " + ", ".join(helpers["compute_world_size"]),
},
"the scan lost a spelling, or started flagging a legal one:\n "
+ "\n ".join(sorted(flagged)),
)
if __name__ == "__main__":
import unittest
unittest.main()