From 53dc77ff4e2b5e52259f1a764604d9be5418def8 Mon Sep 17 00:00:00 2001
From: Cheng Wan <54331508+ch-wan@users.noreply.github.com>
Date: Wed, 9 Sep 2026 19:22:06 -0700
Subject: [PATCH] [Config] One writer for the declaration stash; no exception
 to the write seal (#38752)

---
 .../skills/sglang-runtime-context/SKILL.md    |  50 +++-
 python/sglang/benchmark/one_batch.py          |  10 +-
 python/sglang/srt/arg_groups/lora_hook.py     |  18 +-
 python/sglang/srt/arg_groups/overrides.py     | 215 ++++++------------
 python/sglang/srt/arg_groups/pipeline.py      |  11 +-
 python/sglang/srt/arg_groups/platform_hook.py |  24 ++
 .../sglang/srt/arg_groups/speculative_hook.py |  29 ++-
 .../sglang/srt/parser/template_detection.py   |   4 +-
 python/sglang/srt/ray/engine.py               |  11 +-
 python/sglang/srt/ray/scheduler_actor.py      |  11 +-
 python/sglang/srt/runtime_context.py          |   4 +-
 python/sglang/srt/server_args.py              | 103 ++-------
 .../server_args/test_model_config_cache.py    |  15 +-
 .../test_resolution_declarations.py           |  24 +-
 .../test_resolution_is_reproducible.py        |  92 +++-----
 .../test_resolution_reads_the_declarations.py |   5 +-
 .../unit/server_args/test_server_args.py      |  40 ++--
 .../unit/test_chain_read_ratchet.py           |  41 +---
 ...test_supplied_instance_exposure_ratchet.py |  46 ++--
 19 files changed, 333 insertions(+), 420 deletions(-)

diff --git a/.claude/skills/sglang-runtime-context/SKILL.md b/.claude/skills/sglang-runtime-context/SKILL.md
index e59fff1fc..3c8d651fe 100644
--- a/.claude/skills/sglang-runtime-context/SKILL.md
+++ b/.claude/skills/sglang-runtime-context/SKILL.md
@@ -91,8 +91,11 @@ with what the operator typed, not with what resolution decided.**
 - **Late launcher-stage resolution (pre-publish)**: a few rules cannot run inside
   `__post_init__` — LoRA normalization, and the auto-parser detection that needs a
   tokenizer/chat-template load. They are resolution, not mutation, and they
-  **declare** via `arg_groups.overrides.declare_late_resolution(server_args,
-  source, **fields)`, which refuses the published instance. The declaration lands
+  **declare** via `arg_groups.overrides.declare_resolution(server_args, source,
+  **fields)`, the same call the rest of the pipeline makes; there is no
+  `declare_late_resolution` any more. *When* a declaration is made is not
+  something the code marks — the guardrails that used to read that marker
+  cover these sites through the ordinary keyword scan instead. The declaration lands
   in the stash on that very object, so every holder of it carries the decision —
   the HTTP server, the multi-tokenizer workers it is serialized for, the
   schedulers it forks — and each of them publishes bags projected from it. The
@@ -408,9 +411,38 @@ derivation cannot enumerate.
 
 One consequence worth knowing: because the fields are the raw input, resolving a
 bare `dataclasses.replace` copy lands in the same place as the parent — the
-pipeline reads only its own input. `replace_resolved` is the way to copy a
-resolved record (it carries the declarations and the `model_config` memo, so the
-copy does not re-resolve at all).
+pipeline reads only its own input. **So a resolved record is not copied at
+all.** A caller that needs one field different for the process it is about to
+hand the record to — the Ray paths and their `dist_init_addr` — declares it on
+the record it holds (`declare_resolution`) and hands that over: the declaration
+travels inside the object, the receiving process projects its bags from it, and
+nothing re-resolves. There is no `ServerArgs.replace_resolved` any more, and the
+`model_config`-memo bug that copying used to cause (a copy marked resolved but
+arriving without the memo cannot refill it, because the guard refuses the write)
+is gone by construction rather than guarded.
+
+A bag `override` cannot stand in for this. It is *not* because overriding needs
+a publish — `set_server_args` is what projects the bags and `override` works as
+soon as the context holds a record — but because `override` writes bag leaves
+and by contract never touches the record, so its effect cannot travel inside an
+object to another process.
+
+### The declaration stash has one writer
+
+Everything that decides configuration goes through
+`declare_resolution(server_args, source, **fields)`. It validates the names,
+refuses the published config (the stash is projected at publish and never
+again, so a later declaration is a silent no-op), and appends. The other names
+around it are spellings, not mechanisms:
+
+| name | what it adds |
+|---|---|
+| `run_post_process_pass` | runs a pass at its slot and validates its return; declares through `declare_resolution`. A pass returning an **empty** dict is a validation, not a declaration, and stays legal on the published instance — `Engine(server_args=sa)` after `Engine.shutdown()` re-runs `check_server_args` on the very instance the context holds |
+| `record_foreign_defaults` | for a resolver this tree does not own (an out-of-tree platform plugin, a registered speculative algorithm), whose interface is to *assign* fields. It gets a stand-in whose reads fall through to `resolving_view`; what it assigned is declared. The record is never written, so the write seal has no exception. In-tree code does not go through it — `handle_platform_defaults` wraps the platform hook, and the in-tree speculative dispatcher is called directly, because handed the stand-in its own `declare_resolution` calls would stash on that instead |
+
+`resolution_projection` is gone; the whole-object readback is
+`ServerArgs.resolved_dict()`, which is what `/server_info` and its gRPC and
+in-process twins report.
 
 ### Adding a model-specific config adjustment
 
@@ -531,8 +563,10 @@ ONE thread — do not design for TBO threads that don't exist.
 
 ## Guardrails (these fail CI; what to do when they fire)
 
-1. **Strict mutation guard** (always on): bare `server_args.x = ...` after resolution
-   raises unconditionally in `ServerArgs.__setattr__` — this *is* the guarantee that
+1. **Strict mutation guard** (always on, and with no exception): bare
+   `server_args.x = ...` after resolution raises unconditionally in
+   `ServerArgs.__setattr__` — the named lift that out-of-tree plugins used to
+   ask for is gone, they assign onto a stand-in instead — this *is* the guarantee that
    no writer can desync the bags, so there is no writer ratchet any more. Change
    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
@@ -621,7 +655,7 @@ Never module-skip a test "until the migration settles" — seed the context inst
 Key source files: `python/sglang/srt/runtime_context.py` (the container, every tier,
 `publish`, `_ConfigBag`, `override_server_args`),
 `python/sglang/srt/arg_groups/overrides.py` (override registry, passes,
-`declare_late_resolution`), `python/sglang/srt/server_args.py` (`NS` metadata,
+`declare_resolution` and the spellings around it), `python/sglang/srt/server_args.py` (`NS` metadata,
 `Arg(..., resolvable=True)`, `__setattr__` strict guard), and the guardrail tests under
 `test/registered/unit/` (`test_server_args_mutation_ratchet.py`,
 `test_global_config_read_ratchet.py`,
diff --git a/python/sglang/benchmark/one_batch.py b/python/sglang/benchmark/one_batch.py
index 2d891af3f..7bc8da18c 100644
--- a/python/sglang/benchmark/one_batch.py
+++ b/python/sglang/benchmark/one_batch.py
@@ -64,7 +64,11 @@ import numpy as np
 import torch
 import torch.distributed as dist
 
-from sglang.srt.arg_groups.overrides import resolution_result, resolving_view
+from sglang.srt.arg_groups.overrides import (
+    declare_resolution,
+    resolution_result,
+    resolving_view,
+)
 from sglang.srt.configs.model_config import ModelConfig
 from sglang.srt.distributed.parallel_state import (
     destroy_distributed_environment,
@@ -1034,8 +1038,8 @@ def main(server_args, bench_args):
         decode = dict(graph_config.get(Phase.DECODE) or {})
         decode["max_bs"] = max(bench_args.batch_size)
         graph_config[Phase.DECODE] = decode
-    server_args = server_args.replace_resolved(
-        "benchmark.one_batch", cuda_graph_config=graph_config
+    declare_resolution(
+        server_args, "benchmark.one_batch", cuda_graph_config=graph_config
     )
     server_args.resolve_once()
     cfg = resolving_view(server_args)
diff --git a/python/sglang/srt/arg_groups/lora_hook.py b/python/sglang/srt/arg_groups/lora_hook.py
index 82df754e1..35ba2dfa0 100644
--- a/python/sglang/srt/arg_groups/lora_hook.py
+++ b/python/sglang/srt/arg_groups/lora_hook.py
@@ -7,7 +7,7 @@ import logging
 from typing import Any
 
 from sglang.srt.arg_groups.overrides import (
-    declare_late_resolution,
+    declare_resolution,
     resolving_view,
 )
 from sglang.srt.environ import envs
@@ -24,9 +24,7 @@ def check_lora_server_args(server_args: Any):
     # Enable LoRA if any LoRA paths are provided for backward compatibility.
     if cfg.lora_paths:
         if cfg.enable_lora is None:
-            declare_late_resolution(
-                server_args, "check_lora_server_args", enable_lora=True
-            )
+            declare_resolution(server_args, "check_lora_server_args", enable_lora=True)
             logger.warning(
                 "--enable-lora is set to True because --lora-paths is provided."
             )
@@ -37,7 +35,7 @@ def check_lora_server_args(server_args: Any):
 
     if cfg.enable_lora:
         if cfg.enable_lora_overlap_loading is None:
-            declare_late_resolution(
+            declare_resolution(
                 server_args, "check_lora_server_args", enable_lora_overlap_loading=False
             )
 
@@ -93,11 +91,11 @@ def check_lora_server_args(server_args: Any):
                         "Expected a string or a dictionary."
                     )
                 parsed_lora_paths.append(lora_ref)
-            declare_late_resolution(
+            declare_resolution(
                 server_args, "check_lora_server_args", lora_paths=parsed_lora_paths
             )
         elif isinstance(cfg.lora_paths, dict):
-            declare_late_resolution(
+            declare_resolution(
                 server_args,
                 "check_lora_server_args",
                 lora_paths=[
@@ -111,9 +109,7 @@ def check_lora_server_args(server_args: Any):
                 ],
             )
         elif cfg.lora_paths is None:
-            declare_late_resolution(
-                server_args, "check_lora_server_args", lora_paths=[]
-            )
+            declare_resolution(server_args, "check_lora_server_args", lora_paths=[])
         else:
             raise ValueError(
                 f"Invalid type for --lora-paths: {type(cfg.lora_paths)}. "
@@ -123,7 +119,7 @@ def check_lora_server_args(server_args: Any):
         # Normalize target modules to a set; keep {"all"} as a sentinel
         # that gets resolved model-awarely in lora_manager.init_lora_shapes().
         if cfg.lora_target_modules:
-            declare_late_resolution(
+            declare_resolution(
                 server_args,
                 "check_lora_server_args",
                 lora_target_modules=set(cfg.lora_target_modules),
diff --git a/python/sglang/srt/arg_groups/overrides.py b/python/sglang/srt/arg_groups/overrides.py
index 486e56786..796290ee2 100644
--- a/python/sglang/srt/arg_groups/overrides.py
+++ b/python/sglang/srt/arg_groups/overrides.py
@@ -17,10 +17,9 @@ Model-identity adjustments to the server configuration are DECLARED here and
 appended to the record's declaration stash (gate order, last writer wins).
 Nothing here writes back onto ``ServerArgs``: the record holds the user's raw
 input, and a decision is read through ``resolution_result`` or the published
-config bags — model code never mutates ``ServerArgs`` fields imperatively. The
-one channel that still leaves a field changed is ``declare_direct_writes``,
-which does not perform the write: it captures one an out-of-tree plugin already
-made, and undoing it would surprise the plugin's own reads.
+config bags — model code never mutates ``ServerArgs`` fields imperatively. That
+holds without exception: a resolver this tree does not own assigns onto a
+stand-in (``record_foreign_defaults``), and what it set is declared.
 
 Two declaration forms, keyed on ``hf_config.architectures[0]``:
 
@@ -33,7 +32,6 @@ Two declaration forms, keyed on ``hf_config.architectures[0]``:
 
 from __future__ import annotations
 
-import copy
 import dataclasses
 import json
 import logging
@@ -116,9 +114,9 @@ def run_post_process_pass(server_args: Any, fn: Callable[..., dict]) -> None:
     an empty dict is a validation, and it may run on the published instance --
     it has to, because ``Engine(server_args=sa)`` after ``Engine.shutdown()``
     re-runs ``check_server_args`` on the very instance the context still holds.
-    A pass that returns a non-empty dict there is refused, as
-    ``declare_late_resolution`` is -- post-publish changes go to the bags through
-    ``get_context().override(...)``.
+    A pass that returns a non-empty dict there is refused by the guard in
+    ``declare_resolution``, as a late declaration is -- post-publish changes go
+    to the bags through ``get_context().override(...)``.
     """
 
     declared = fn(ResolvedView(server_args, overlay=_declaration_overlay(server_args)))
@@ -133,29 +131,12 @@ def run_post_process_pass(server_args: Any, fn: Callable[..., dict]) -> None:
         # a rebuild: `Engine(server_args=sa)` after `Engine.shutdown()` hands
         # back the same instance while the context still holds it, and
         # refusing on identity alone would fail that launch.
-        try:
-            published = get_context().server_args
-        except ValueError:
-            published = None
-        if published is server_args:
-            raise ValueError(
-                f"run_post_process_pass({fn.__qualname__!r}) declared "
-                f"{sorted(declared)} on the published config; the stash is "
-                "projected at publish and never again, so this would be a "
-                "silent no-op -- post-publish changes go to the bags via "
-                "get_context().override(...)"
-            )
-        entry = (fn.__qualname__, dict(declared))
-        stash = getattr(server_args, "_resolved_overrides", None)
-        if stash is None:
-            # Handlers hosting pass slots may be invoked directly on fixtures
-            # that never ran the monolith dispatch (which owns the stash);
-            # create it lazily. Real publishes always pass through the
-            # dispatch first — the dispatch ASSIGNS the stash, so pass slots
-            # must sit at or after it in __post_init__ order.
-            stash = server_args._resolved_overrides = []
-        stash.append(entry)
-        validate_declarations(server_args, [entry])
+        # Only a non-empty return is a declaration. An empty one is a
+        # validation and may run on the published instance -- see above -- so it
+        # must not reach the guard in `declare_resolution`.
+        if declared:
+            declare_resolution(server_args, fn.__qualname__, **declared)
+            validate_declarations(server_args, [(fn.__qualname__, dict(declared))])
 
 
 def declare_resolution(server_args: Any, source: str, **fields: Any) -> None:
@@ -167,52 +148,32 @@ def declare_resolution(server_args: Any, source: str, **fields: Any) -> None:
     (or `resolved_view(server_args)`), which
     `test_resolution_reads_the_declarations` pins.
 
-    For resolvers inside ``__post_init__``; launcher-stage resolution goes
-    through ``declare_late_resolution``. A name that is not a field is rejected
-    here rather than becoming an attribute nothing reads.
+    Every declaration goes through here, whenever it is made: inside
+    ``__post_init__``, at launcher stage (LoRA normalization, the auto-detected
+    parsers -- they decide what the process will run with, so they belong to the
+    pipeline even though they run after it), and on a copy about to cross a
+    process boundary. A name that is not a field is rejected here rather than
+    becoming an attribute nothing reads.
+
+    Refuses the published config. The stash is projected at publish and never
+    again, so a declaration afterwards is a silent no-op; post-publish changes
+    go to the bags through ``get_context().override(...)``.
     """
     if dataclasses.is_dataclass(type(server_args)):
         unknown = sorted(set(fields) - field_names(type(server_args)))
         if unknown:
             raise AttributeError(f"{source}: {unknown} are not ServerArgs fields")
-    stash = getattr(server_args, "_resolved_overrides", None)
-    if stash is None:
-        stash = []
-        server_args._resolved_overrides = stash
-    stash.append((source, dict(fields)))
-
-
-def declare_late_resolution(server_args: Any, source: str, **fields: Any) -> None:
-    """Resolve fields on a config that is **not published yet**.
-
-    A few resolution rules cannot run inside ``__post_init__``: LoRA
-    normalization and the auto-parser detection need the launcher's validation
-    stage (and, for the parsers, a tokenizer / chat-template load). They still
-    belong to the resolution pipeline — they decide what the process will run
-    with — so their decision goes to the stash like any other, and the record
-    keeps what the caller passed. Every holder of that instance reads the
-    decision the same way the rest of the pipeline does: the bags it publishes,
-    or ``resolution_result``, both of which survive the pickle to a child.
-
-    Refuses to touch the published instance: after publish the bags exist and a
-    field write would desync them, which is what ``get_context().override`` is
-    for.
-    """
-
     try:
         published = get_context().server_args
     except ValueError:
         published = None
     if published is server_args:
         raise ValueError(
-            f"declare_late_resolution({source!r}) called on the published config; "
-            "post-publish changes go to the bags via get_context().override(...)"
+            f"{source}: declared on the published config; the stash is "
+            "projected at publish and never again, so this would be a silent "
+            "no-op -- post-publish changes go to the bags via "
+            "get_context().override(...)"
         )
-    log = getattr(server_args, "_runtime_mutations", None)
-    if log is None:
-        log = []
-        server_args._runtime_mutations = log
-    log.append((source, dict(fields)))
     stash = getattr(server_args, "_resolved_overrides", None)
     if stash is None:
         stash = []
@@ -220,59 +181,65 @@ def declare_late_resolution(server_args: Any, source: str, **fields: Any) -> Non
     stash.append((source, dict(fields)))
 
 
-def declare_direct_writes(
+class _ForeignDefaults:
+    """The stand-in handed to a resolver this tree does not own.
+
+    Reads fall through to the resolving view, so a plugin sees what resolution
+    has decided so far rather than the raw input -- better than what it used to
+    get, which was the record's own fields. Writes are captured here and
+    declared by the caller, so the record is never written and the write seal
+    has no exception.
+    """
+
+    __slots__ = ("_cfg", "_written")
+
+    def __init__(self, server_args: Any):
+        object.__setattr__(self, "_cfg", resolving_view(server_args))
+        object.__setattr__(self, "_written", {})
+
+    def __getattr__(self, name: str) -> Any:
+        written = object.__getattribute__(self, "_written")
+        if name in written:
+            return written[name]
+        return getattr(object.__getattribute__(self, "_cfg"), name)
+
+    def __setattr__(self, name: str, value: Any) -> None:
+        object.__getattribute__(self, "_written")[name] = value
+
+
+def record_foreign_defaults(
     server_args: Any, source: str, resolve: Callable[[Any], Any]
 ) -> Any:
-    """Run a resolver that writes the fields directly, and declare what it moved.
+    """Run a resolver this tree does not own, and declare what it set.
+
+    Out-of-tree platform plugins and registered speculative algorithms are
+    handed a configuration and assign fields on it. That interface is not ours
+    to change, so the assignment stays the contract -- it just lands on a
+    stand-in instead of the record, and what it set is declared like any other
+    decision. Nothing writes the record, which is why there is no longer a
+    named hole in the seal.
 
     Returns whatever the resolver returned, so a provider with a return value
-    can go through the same capture.
+    goes through the same capture.
 
-    Out-of-tree platform plugins are handed the record and set fields on it.
-    Their implementations live outside this tree, so they cannot be converted
-    by editing the resolver; and the raw snapshot is taken before the pipeline
-    starts, so a plugin's default is neither declared nor raw. The write itself
-    stays: this captures it into the stash so the projection and the bags carry
-    it, but reverting the field would break the plugin's own reads of what it
-    just set. It is the only field a record still carries from resolution.
-
-    Rebinding is what the diff sees, and rebinding is all it needs to see: a
-    plugin that mutates a value in place reaches the projection anyway, because
-    the raw snapshot and the stash entries hold the same object it mutated.
+    Non-field names are dropped: a plugin scribbling on an attribute that is
+    not configuration is not a decision, and it was invisible to the previous
+    diff for the same reason.
 
     A stand-in record (tests drive the hooks with a plain namespace) has no
-    fields to diff and no projection to feed, so the resolver runs uncaptured.
+    view to read, so the resolver runs against it directly and uncaptured.
     """
     if not dataclasses.is_dataclass(server_args):
         return resolve(server_args)
-    before = {
-        field.name: getattr(server_args, field.name)
-        for field in dataclasses.fields(server_args)
+    recorder = _ForeignDefaults(server_args)
+    result = resolve(recorder)
+    written = {
+        name: value
+        for name, value in object.__getattribute__(recorder, "_written").items()
+        if name in field_names(type(server_args))
     }
-    already = len(getattr(server_args, "_resolved_overrides", None) or ())
-    # The one place the input seal comes off. The plugin writes the record;
-    # the diff below captures what it moved into the stash so the projection
-    # and the bags carry it.
-    from sglang.srt.server_args import record_writable
-
-    with record_writable(server_args):
-        result = resolve(server_args)
-    stash = getattr(server_args, "_resolved_overrides", None)
-    if stash is None:
-        stash = []
-        server_args._resolved_overrides = stash
-    # A resolver reached this way can also declare properly -- the in-tree
-    # implementations of these hooks do. Those fields are already explained, and
-    # recording them again would attribute them to the wrapper and bury an
-    # actual direct write among the echoes.
-    declared = {name for _source, fields in stash[already:] for name in fields}
-    changed = {
-        name: getattr(server_args, name)
-        for name, previous in before.items()
-        if name not in declared and getattr(server_args, name) is not previous
-    }
-    if changed:
-        stash.append((source, changed))
+    if written:
+        declare_resolution(server_args, source, **written)
     return result
 
 
@@ -303,40 +270,6 @@ def resolution_result(server_args: Any, field: str, default: Any = None) -> Any:
     return with_fallback(type(server_args), field, getattr(server_args, field, default))
 
 
-def resolution_projection(server_args: Any) -> Dict[str, Any]:
-    """Every field's resolved value, nested dataclasses expanded.
-
-    The whole-object shape of ``resolution_result``, for the exits that hand out
-    the entire configuration (``/server_info``, the gRPC and engine readbacks).
-    They used ``dataclasses.asdict``, which reads the fields -- the operator's
-    input, not what resolution decided. Field values only: the private resolution
-    bookkeeping and the ``model_config`` memo that a ``vars()`` dump carried into
-    the readback are not configuration.
-    """
-    return {
-        field.name: _plain(resolution_result(server_args, field.name))
-        for field in dataclasses.fields(server_args)
-    }
-
-
-def _plain(value: Any) -> Any:
-    """``dataclasses.asdict``'s conversion, applied to one value: dataclasses
-    become dicts, containers recurse, everything else is deep-copied (a caller
-    mutating the dump must not reach the live configuration)."""
-    if dataclasses.is_dataclass(value) and not isinstance(value, type):
-        return {
-            field.name: _plain(getattr(value, field.name))
-            for field in dataclasses.fields(value)
-        }
-    if isinstance(value, tuple) and hasattr(value, "_fields"):  # namedtuple
-        return type(value)(*(_plain(item) for item in value))
-    if isinstance(value, (list, tuple)):
-        return type(value)(_plain(item) for item in value)
-    if isinstance(value, dict):
-        return type(value)((_plain(k), _plain(v)) for k, v in value.items())
-    return copy.deepcopy(value)
-
-
 def pre_capture_activation_reserve_mb_of(cfg: Any, gpu_mem: Optional[float]) -> float:
     """The activation working-set reserve held back before cuda-graph capture.
 
diff --git a/python/sglang/srt/arg_groups/pipeline.py b/python/sglang/srt/arg_groups/pipeline.py
index 96c996c8e..fd5cee710 100644
--- a/python/sglang/srt/arg_groups/pipeline.py
+++ b/python/sglang/srt/arg_groups/pipeline.py
@@ -15,11 +15,9 @@ from sglang.srt.arg_groups.overrides import (
     _page_size_default,
     _pipeline_parallel_overlap_disable,
     _sampling_backend_default,
-    declare_direct_writes,
     resolving_view,
     run_post_process_pass,
 )
-from sglang.srt.platforms import current_platform
 from sglang.srt.utils.common import get_device_memory_capacity
 
 
@@ -204,6 +202,7 @@ def run_resolution_pipeline(server_args: Any) -> None:
         handle_mps_backends,
         handle_nccl_pre_warm,
         handle_npu_backends,
+        handle_platform_defaults,
         handle_symm_mem_device_support,
         handle_xpu_backends,
     )
@@ -217,13 +216,7 @@ def run_resolution_pipeline(server_args: Any) -> None:
     # keys off enable_symm_mem.
     handle_symm_mem_device_support(server_args)
 
-    # OOT platform plugins set fields directly (an interface this tree
-    # does not own); the diff records what they applied.
-    declare_direct_writes(
-        server_args,
-        f"platform:{current_platform.device_name}",
-        current_platform.apply_server_args_defaults,
-    )
+    handle_platform_defaults(server_args)
 
     gpu_mem = get_device_memory_capacity(cfg.device)
 
diff --git a/python/sglang/srt/arg_groups/platform_hook.py b/python/sglang/srt/arg_groups/platform_hook.py
index f32876aaa..fac536a04 100644
--- a/python/sglang/srt/arg_groups/platform_hook.py
+++ b/python/sglang/srt/arg_groups/platform_hook.py
@@ -9,6 +9,7 @@ from typing import Any
 
 from sglang.srt.arg_groups.overrides import (
     declare_resolution,
+    record_foreign_defaults,
     resolving_view,
 )
 from sglang.srt.hardware_backend.mlx.runtime import use_mlx
@@ -84,6 +85,29 @@ def handle_nccl_pre_warm(server_args: Any):
         declare_resolution(server_args, "_handle_nccl_pre_warm", pre_warm_nccl=False)
 
 
+def handle_platform_defaults(server_args: Any):
+    """An out-of-tree platform's defaults, declared like every rule beside it.
+
+    `Platform.apply_server_args_defaults` is a plugin interface: the platform is
+    handed a configuration and assigns the fields it wants defaulted. In-tree
+    platforms do not implement it -- the base is a no-op and nothing overrides
+    it -- so this captures nothing here and exists for the platforms that live
+    outside this tree.
+
+    Ordering: it must precede `handle_gpu_memory_settings`, whose symm-mem
+    prealloc default keys off `enable_symm_mem`.
+    """
+    # `current_platform` is the plugin object; `get_platform()` is the facts
+    # view over it and carries neither the name nor the hook.
+    from sglang.srt.platforms import current_platform
+
+    record_foreign_defaults(
+        server_args,
+        f"platform:{current_platform.device_name}",
+        current_platform.apply_server_args_defaults,
+    )
+
+
 def handle_symm_mem_device_support(server_args: Any):
     cfg = resolving_view(server_args)
     # The symm-mem allocator compiles a CUDA plugin and links -lnccl, so off
diff --git a/python/sglang/srt/arg_groups/speculative_hook.py b/python/sglang/srt/arg_groups/speculative_hook.py
index b2f39e61d..24fd98d4b 100644
--- a/python/sglang/srt/arg_groups/speculative_hook.py
+++ b/python/sglang/srt/arg_groups/speculative_hook.py
@@ -8,9 +8,9 @@ from typing import TYPE_CHECKING, Optional
 from sglang.srt.arg_groups.overrides import (
     _speculative_moe_runner_default,
     attention_backends_of,
-    declare_direct_writes,
     declare_resolution,
     model_config_of,
+    record_foreign_defaults,
     resolved_view,
     resolving_view,
     run_post_process_pass,
@@ -157,7 +157,7 @@ def handle_speculative_decoding(server_args: ServerArgs) -> None:
 
         # TODO: move the per-algorithm validation below into spec module hooks.
         if isinstance(algo, CustomSpecAlgo) and algo.validate_server_args is not None:
-            declare_direct_writes(
+            record_foreign_defaults(
                 server_args,
                 "handle_speculative_decoding.custom_validate",
                 algo.validate_server_args,
@@ -175,13 +175,24 @@ def handle_speculative_decoding(server_args: ServerArgs) -> None:
             _init_adaptive_speculative_params(server_args)
 
     if algo is not None:
-        # A registered algorithm's callback lives outside this tree and sets
-        # fields on the record, so the writes are captured around the call.
-        declare_direct_writes(
-            server_args,
-            "handle_speculative_decoding.custom_algo",
-            algo.handle_server_args,
-        )
+        # Imported here and not above: the name is only bound inside the
+        # `speculative_algorithm is not None` branch, and this runs either way.
+        from sglang.srt.speculative.spec_registry import CustomSpecAlgo
+
+        if isinstance(algo, CustomSpecAlgo):
+            # A registered algorithm's callback lives outside this tree and
+            # assigns fields, so it gets the stand-in and its writes are
+            # declared.
+            record_foreign_defaults(
+                server_args,
+                "handle_speculative_decoding.custom_algo",
+                algo.handle_server_args,
+            )
+        else:
+            # The in-tree dispatcher, which declares. It needs the record
+            # itself: handed the stand-in, its `declare_resolution` calls would
+            # stash on that instead.
+            algo.handle_server_args(server_args)
 
 
 def _handle_dflash(server_args: ServerArgs) -> None:
diff --git a/python/sglang/srt/parser/template_detection.py b/python/sglang/srt/parser/template_detection.py
index 6a82786de..c87629ca2 100644
--- a/python/sglang/srt/parser/template_detection.py
+++ b/python/sglang/srt/parser/template_detection.py
@@ -29,7 +29,7 @@ import jinja2.ext
 import jinja2.nodes
 import jinja2.sandbox
 
-from sglang.srt.arg_groups.overrides import declare_late_resolution, resolving_view
+from sglang.srt.arg_groups.overrides import declare_resolution, resolving_view
 
 logger = logging.getLogger(__name__)
 
@@ -829,4 +829,4 @@ def resolve_auto_parsers(server_args) -> None:
                 detected[attr] = _detect_auto_parser(attr, ctx, rules, label)
 
     if detected:
-        declare_late_resolution(server_args, "template-detection", **detected)
+        declare_resolution(server_args, "template-detection", **detected)
diff --git a/python/sglang/srt/ray/engine.py b/python/sglang/srt/ray/engine.py
index 35cb9c9c8..41f667d9e 100644
--- a/python/sglang/srt/ray/engine.py
+++ b/python/sglang/srt/ray/engine.py
@@ -24,6 +24,7 @@ import ray
 from ray.util.placement_group import PlacementGroup
 from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
 
+from sglang.srt.arg_groups.overrides import declare_resolution
 from sglang.srt.entrypoints.engine import (
     Engine,
     SchedulerInitResult,
@@ -471,12 +472,16 @@ class RayEngine(Engine):
             f"enable_dp_attention={parallel.enable_dp_attention}"
         )
 
-        # Set dist_init_addr on server_args so PortArgs.init_new() can compute
-        # TCP addresses correctly (required for DP attention path).
-        dp_server_args = server_args.replace_resolved(
+        # Declared on the record itself so `PortArgs.init_new()` can compute
+        # TCP addresses (required for the DP attention path). No copy: this
+        # process does not publish here, and every other reader of the field
+        # goes through the bags its own process projects.
+        declare_resolution(
+            server_args,
             "ray.dp_controller",
             dist_init_addr=f"{rank0_node_ip}:{port_args.nccl_port}",
         )
+        dp_server_args = server_args
         # Create the DP controller in-process. This blocks until all actors
         # are initialized and their event loops have started.
         controller = RayDataParallelController(
diff --git a/python/sglang/srt/ray/scheduler_actor.py b/python/sglang/srt/ray/scheduler_actor.py
index 098987948..3616d1816 100644
--- a/python/sglang/srt/ray/scheduler_actor.py
+++ b/python/sglang/srt/ray/scheduler_actor.py
@@ -20,6 +20,7 @@ from typing import Any, Dict, Optional
 
 import ray
 
+from sglang.srt.arg_groups.overrides import declare_resolution
 from sglang.srt.runtime_context import publish
 from sglang.srt.server_args import PortArgs, ServerArgs
 
@@ -54,11 +55,13 @@ class SchedulerActor:
             numa_bind_to_node,
         )
 
-        # Override dist_init_addr if provided (for multi-node), through
-        # `replace_resolved` so the copy keeps the parent's resolution.
+        # Declared, not copied: Ray deserializes the argument per call, so this
+        # record is the actor's own and nothing else in the process holds it.
+        # The field stays the operator's input; `PortArgs.init_new` and the bags
+        # this actor publishes read the decision.
         if dist_init_addr:
-            server_args = server_args.replace_resolved(
-                "ray.scheduler_actor", dist_init_addr=dist_init_addr
+            declare_resolution(
+                server_args, "ray.scheduler_actor", dist_init_addr=dist_init_addr
             )
 
         # Get actual GPU IDs from Ray runtime context
diff --git a/python/sglang/srt/runtime_context.py b/python/sglang/srt/runtime_context.py
index 8eff31689..4b0a9739b 100644
--- a/python/sglang/srt/runtime_context.py
+++ b/python/sglang/srt/runtime_context.py
@@ -1282,7 +1282,7 @@ class _ServerArgsOverride:
         self._prev_parallel_config = ctx.parallel._config
         self._prev_capture = ctx.flags.capture.enable_torch_compile
         from sglang.srt.arg_groups.overrides import (
-            declare_late_resolution,
+            declare_resolution,
         )
 
         server_args = ServerArgs(model_path="dummy")
@@ -1306,7 +1306,7 @@ class _ServerArgsOverride:
         fields = set(type(server_args).__dataclass_fields__)
         declared = {n: v for n, v in self._fields.items() if n in fields}
         if declared:
-            declare_late_resolution(server_args, "override_server_args", **declared)
+            declare_resolution(server_args, "override_server_args", **declared)
         # What is left seeds the record's own private caches (`_model_config`
         # and friends), which are not configuration and never were.
         seeds = {n: v for n, v in self._fields.items() if n not in fields}
diff --git a/python/sglang/srt/server_args.py b/python/sglang/srt/server_args.py
index f970c14ef..509129065 100644
--- a/python/sglang/srt/server_args.py
+++ b/python/sglang/srt/server_args.py
@@ -40,7 +40,6 @@ import functools
 import logging
 import tempfile
 import uuid
-from contextlib import contextmanager
 from typing import Any, NoReturn
 
 from sglang.kernels.ops.kv_canary.consts import RealKvHashMode
@@ -53,7 +52,7 @@ from sglang.srt.arg_groups.argparse_actions import (
 from sglang.srt.arg_groups.model_override_base import ep_joiner_of, ep_scale_joiner_of
 from sglang.srt.arg_groups.overrides import (
     remote_instance_transfer_engine_of,
-    resolution_projection,
+    resolution_result,
     resolving_view,
 )
 from sglang.srt.environ import envs
@@ -170,6 +169,24 @@ from sglang.srt.utils.common import (  # noqa: F401
 )
 
 
+def _plain(value: Any) -> Any:
+    """``dataclasses.asdict``'s conversion, applied to one value: dataclasses
+    become dicts, containers recurse, everything else is deep-copied (a caller
+    mutating the dump must not reach the live configuration)."""
+    if dataclasses.is_dataclass(value) and not isinstance(value, type):
+        return {
+            field.name: _plain(getattr(value, field.name))
+            for field in dataclasses.fields(value)
+        }
+    if isinstance(value, tuple) and hasattr(value, "_fields"):  # namedtuple
+        return type(value)(*(_plain(item) for item in value))
+    if isinstance(value, (list, tuple)):
+        return type(value)(_plain(item) for item in value)
+    if isinstance(value, dict):
+        return type(value)((_plain(k), _plain(v)) for k, v in value.items())
+    return copy.deepcopy(value)
+
+
 class ServerArgs:
     """Server-wide configuration for SGLang.
 
@@ -252,9 +269,8 @@ class ServerArgs:
         from sglang.srt.arg_groups.pipeline import run_resolution_pipeline
 
         # Sealed for the duration, not just afterwards: everything below this
-        # line reads the input and declares against it, and the one channel
-        # that still writes the record (`declare_direct_writes`, for
-        # out-of-tree platform plugins) asks for the seal to be lifted by name.
+        # line reads the input and declares against it. No exceptions -- even a
+        # resolver from outside this tree assigns onto a stand-in, not here.
         self._input_frozen = True
         try:
             run_resolution_pipeline(self)
@@ -298,58 +314,10 @@ class ServerArgs:
         `model_config` memo are not fields and do not appear.
         """
 
-        return resolution_projection(self)
-
-    def replace_resolved(self, source: str, **changes: Any) -> ServerArgs:
-        """A copy of this record that stays resolved, and says what it changed.
-
-        `dataclasses.replace` builds a new instance, so the copy carries none of
-        what makes a record resolved: no raw snapshot, no declarations, no
-        finished flag. The next publish therefore resolves it again, which
-        drops every decision the stash held -- the late ones (the auto-detected
-        parsers) and the direct ones alike -- and re-runs the device probes in
-        whatever process opened the copy. The Ray paths replace
-        `dist_init_addr` on a resolved record, which is how they reach this.
-
-        The change is appended to the stash rather than left on the field: the
-        projection reads the raw snapshot plus the declarations, so a field the
-        copy set on its own would publish the parent's raw value instead.
-
-        The carry is shallow. The containers are copied so the copy's own
-        declaration does not travel back into the parent, but everything inside
-        them -- the stash entries, the raw-input values, the memoized
-        `ModelConfig` -- is shared. That is fine for what this is for: a copy
-        that immediately crosses a process boundary (Ray actors, the gateway's
-        workers), where pickling severs the sharing. A caller that mutates the
-        copy's deep structure in-process mutates the parent's too.
-        """
-        replacement = dataclasses.replace(self, **changes)
-        # Provenance, not resolution state: a copy was still launched by
-        # whatever launched its parent, resolved or not.
-        object.__setattr__(replacement, "_launch_command", self.launch_command)
-        if not getattr(self, "_resolution_finished", False):
-            # Not resolved yet: the copy goes through the gate itself.
-            return replacement
-
-        # Everything outside the fields, enumerated from the instance: the raw
-        # snapshot, the stash, and what resolution memoized -- including the
-        # model-configuration memo, which the copy carries over rather than
-        # rebuild.
-        field_names = {field.name for field in dataclasses.fields(self)}
-        for name, value in vars(self).items():
-            if name in field_names or name == "_resolution_finished":
-                continue
-            if isinstance(value, (dict, list, set)):
-                value = copy.copy(value)
-            object.__setattr__(replacement, name, value)
-        stash = getattr(replacement, "_resolved_overrides", None)
-        if stash is None:
-            stash = []
-            object.__setattr__(replacement, "_resolved_overrides", stash)
-        if changes:
-            stash.append((source, dict(changes)))
-        object.__setattr__(replacement, "_resolution_finished", True)
-        return replacement
+        return {
+            field.name: _plain(resolution_result(self, field.name))
+            for field in dataclasses.fields(self)
+        }
 
     # ------------------------------------------------------------------
     # CUDA graph configuration resolution
@@ -663,27 +631,6 @@ def get_global_server_args() -> NoReturn:
     )
 
 
-@contextmanager
-def record_writable(server_args: Any):
-    """Lift the input seal for a resolver that genuinely writes the record.
-
-    There is exactly one: `declare_direct_writes`, which hands the record to an
-    out-of-tree platform plugin that sets fields on it. Those implementations
-    live outside this tree and cannot be converted by editing a resolver here,
-    so the write stays and is captured into the stash afterwards. Naming the
-    exception is the point -- an in-tree resolver that reaches for this is
-    doing something it should be declaring instead.
-    """
-    frozen = getattr(server_args, "_input_frozen", False)
-    if frozen:
-        object.__setattr__(server_args, "_input_frozen", False)
-    try:
-        yield
-    finally:
-        if frozen:
-            object.__setattr__(server_args, "_input_frozen", True)
-
-
 def prepare_server_args(argv: list[str]) -> ServerArgs:
     """
     Prepare the server arguments from the command line arguments.
diff --git a/test/registered/unit/server_args/test_model_config_cache.py b/test/registered/unit/server_args/test_model_config_cache.py
index 8019d427e..0337e6885 100644
--- a/test/registered/unit/server_args/test_model_config_cache.py
+++ b/test/registered/unit/server_args/test_model_config_cache.py
@@ -148,16 +148,19 @@ class TestTheModelConfigCache(CustomTestCase):
         second_checkpoint = self._checkpoint()
 
         server_args = self._resolved(model_path=first_checkpoint)
-        copy_ = server_args.replace_resolved(
+        self.assertEqual(model_config_of(server_args).model_path, first_checkpoint)
+
+        # Declaring a new `model_path` moves what the memo is keyed on, so the
+        # next read rebuilds rather than handing back a configuration that
+        # describes the previous checkpoint.
+        declare_resolution(
+            server_args,
             "test_the_cache_refills_on_a_resolved_record",
             model_path=second_checkpoint,
         )
-
-        rebuilt = model_config_of(copy_)
+        rebuilt = model_config_of(server_args)
         self.assertEqual(rebuilt.model_path, second_checkpoint)
-        self.assertIs(model_config_of(copy_), rebuilt)
-        # The parent keeps the configuration it resolved with.
-        self.assertEqual(model_config_of(server_args).model_path, first_checkpoint)
+        self.assertIs(model_config_of(server_args), rebuilt)
 
     def test_a_supplied_configuration_is_handed_back(self):
         """A configuration nothing in here built carries no key, so nothing
diff --git a/test/registered/unit/server_args/test_resolution_declarations.py b/test/registered/unit/server_args/test_resolution_declarations.py
index 51667b3e6..b7739665a 100644
--- a/test/registered/unit/server_args/test_resolution_declarations.py
+++ b/test/registered/unit/server_args/test_resolution_declarations.py
@@ -436,7 +436,7 @@ 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
-        declare through `declare_late_resolution`. The declaration is the only
+        declare through `declare_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.
 
@@ -444,14 +444,12 @@ class TestResolutionDeclarations(CustomTestCase):
         so its `resolve_once` re-runs and re-snapshots the raw input from
         already-late-resolved fields, which hides exactly this.
         """
-        from sglang.srt.arg_groups.overrides import declare_late_resolution
+        from sglang.srt.arg_groups.overrides import declare_resolution
         from sglang.srt.runtime_context import get_serving, publish, reset_context
 
         server_args = self._resolve({"reasoning_parser": "auto"})
         self.addCleanup(reset_context)
-        declare_late_resolution(
-            server_args, "template-detection", reasoning_parser="qwen3"
-        )
+        declare_resolution(server_args, "template-detection", reasoning_parser="qwen3")
         self.assertEqual(
             resolution_result(server_args, "reasoning_parser"),
             "qwen3",
@@ -469,10 +467,10 @@ class TestResolutionDeclarations(CustomTestCase):
 
     def test_pre_engine_late_resolution_reaches_the_projection(self):
         """A launcher declaration survives the engine's first resolution pass."""
-        from sglang.srt.arg_groups.overrides import declare_late_resolution
+        from sglang.srt.arg_groups.overrides import declare_resolution
 
         server_args = ServerArgs(model_path="dummy")
-        declare_late_resolution(
+        declare_resolution(
             server_args,
             "launcher",
             enable_forward_pass_metrics=True,
@@ -695,11 +693,13 @@ class TestResolutionDeclarations(CustomTestCase):
                 server_args.attention_backend = "triton"
                 server_args.schedule_conservativeness = 0.5
 
-        from sglang.srt.arg_groups import pipeline as pipeline_module
+        from sglang.srt import platforms as platforms_module
 
-        # The write capture runs in the dispatcher, so that is the namespace the
-        # plugin has to be installed in.
-        with unittest.mock.patch.object(pipeline_module, "current_platform", _Plugin()):
+        # `handle_platform_defaults` imports `current_platform` when it runs, so
+        # the platform module is the namespace to install the plugin in.
+        with unittest.mock.patch.object(
+            platforms_module, "current_platform", _Plugin()
+        ):
             server_args = self._resolve({})
         self.assertEqual(
             (
@@ -738,7 +738,7 @@ class TestDeclaredValuesAreNotEditedLater(CustomTestCase):
 
         The property is about the stash, so the seam is the stash: a list that
         snapshots on append. Every declaration path -- `declare_resolution`,
-        `declare_late_resolution`, `declare_direct_writes` and the passes --
+        `declare_resolution`, `record_foreign_defaults` and the passes --
         reaches it through `.append`, whatever it was imported as.
         """
         recorded = []
diff --git a/test/registered/unit/server_args/test_resolution_is_reproducible.py b/test/registered/unit/server_args/test_resolution_is_reproducible.py
index 998692fdf..b6bc293ee 100644
--- a/test/registered/unit/server_args/test_resolution_is_reproducible.py
+++ b/test/registered/unit/server_args/test_resolution_is_reproducible.py
@@ -35,7 +35,10 @@ import unittest.mock
 
 import torch
 
-from sglang.srt.arg_groups.overrides import model_config_of, resolution_result
+from sglang.srt.arg_groups.overrides import (
+    declare_resolution,
+    resolution_result,
+)
 from sglang.srt.environ import EnvField, envs
 from sglang.srt.server_args import ServerArgs
 from sglang.srt.utils import is_cuda
@@ -480,15 +483,15 @@ class TestResolutionIsReproducible(_RestoresProcessState, CustomTestCase):
         self.assertEqual(getattr(first, "_resolved_overrides", None), first_provenance)
 
 
-class TestACopyStaysResolved(_RestoresProcessState, CustomTestCase):
+class TestALateDeclarationKeepsTheResolution(_RestoresProcessState, CustomTestCase):
     """A resolved record copied with `dataclasses.replace` loses what makes it
     resolved, and the next publish resolves it a second time -- over values it
-    already decided. The Ray paths copy a resolved record to set
-    `dist_init_addr`, which is how they reach this.
+    already decided. The Ray paths declare `dist_init_addr` on a record that
+    has already resolved, which is how they reach this.
     """
 
     def _resolved(self):
-        config_dir = tempfile.mkdtemp(prefix="replace_resolved_")
+        config_dir = tempfile.mkdtemp(prefix="late_declaration_")
         self.addCleanup(shutil.rmtree, config_dir, ignore_errors=True)
         with open(os.path.join(config_dir, "config.json"), "w") as handle:
             json.dump(_MINI_CONFIG, handle)
@@ -509,9 +512,9 @@ class TestACopyStaysResolved(_RestoresProcessState, CustomTestCase):
 
         `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.
+        halving and the conservativeness scaling apply once. This is why the Ray
+        paths declare on the record they were handed instead of copying it: the
+        record arrives resolved, and a copy would throw that away.
         """
         parent = self._resolved()
         bare = dataclasses.replace(parent, dist_init_addr="1.2.3.4:5000")
@@ -537,56 +540,34 @@ class TestACopyStaysResolved(_RestoresProcessState, CustomTestCase):
             "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_, "_resolution_finished", False))
-        drifted = {
-            field.name: (getattr(parent, field.name), getattr(copy_, field.name))
-            for field in dataclasses.fields(parent)
-            if field.name != "dist_init_addr"
-            and getattr(parent, field.name) != getattr(copy_, field.name)
-        }
-        self.assertEqual(
-            drifted,
-            {},
-            f"the copy differs from its parent beyond the change: {drifted}",
-        )
-        self.assertEqual(copy_.dist_init_addr, "1.2.3.4:5000")
+    def test_a_late_change_leaves_the_rest_of_the_resolution_alone(self):
+        """What the Ray paths do: declare one field on a record that has already
+        resolved, then hand it to the process that will publish it.
 
-    def test_the_copy_carries_what_resolution_left_on_the_record(self):
-        """Not just the stash and the flag.
-
-        `model_config_of()` memoizes on the record, and that cache is filled
-        during resolution. A copy that is marked resolved but arrives without it
-        cannot fill it -- the read-only guard refuses the cache write -- so the
-        first `model_config_of()` raises. That is what killed the Ray
-        schedulers, and it is why the carry is enumerated from the instance
-        rather than from a list of names.
+        The record stays resolved, so nothing re-derives; the field stays the
+        operator's input, because resolution does not write fields; and the
+        decision is what `resolution_result` answers.
         """
         parent = self._resolved()
-        copy_ = parent.replace_resolved("ray.test", dist_init_addr="1.2.3.4:5000")
-        fields = {field.name for field in dataclasses.fields(parent)}
-        missing = sorted(
-            name
-            for name in vars(parent)
-            if name not in fields and name not in vars(copy_)
-        )
-        self.assertEqual(
-            missing,
-            [],
-            f"the copy did not carry what resolution left on the record: {missing}",
-        )
-        self.assertIsNotNone(model_config_of(copy_))
-        # Containers are copied, so the copy's declaration stays with it.
-        self.assertEqual(
-            len(parent._resolved_overrides) + 1, len(copy_._resolved_overrides)
+        declare_resolution(parent, "ray.test", dist_init_addr="1.2.3.4:5000")
+
+        self.assertTrue(getattr(parent, "_resolution_finished", False))
+        self.assertIsNone(
+            parent.dist_init_addr,
+            "the declaration wrote the field; the record is the operator's input",
         )
+        self.assertEqual(resolution_result(parent, "dist_init_addr"), "1.2.3.4:5000")
 
     def test_the_change_reaches_the_bags(self):
         """The projection reads the raw snapshot plus the declarations, so a
-        change the copy only wrote to the field would publish the parent's raw
-        value."""
+        change written only to the field would publish the raw value instead.
+
+        This is the Ray hop: the actor receives the record by pickle, declares
+        its own `dist_init_addr`, and publishes. Nothing else may move --
+        publishing must not re-run resolution.
+        """
+        import pickle
+
         from sglang.srt.runtime_context import (
             get_parallel,
             get_schedule,
@@ -595,16 +576,17 @@ class TestACopyStaysResolved(_RestoresProcessState, CustomTestCase):
         )
 
         parent = self._resolved()
-        copy_ = parent.replace_resolved("ray.test", dist_init_addr="1.2.3.4:5000")
+        arrived = pickle.loads(pickle.dumps(parent))
+        declare_resolution(arrived, "ray.test", dist_init_addr="1.2.3.4:5000")
         self.addCleanup(reset_context)
         reset_context()
-        publish(copy_, role="scheduler")
+        publish(arrived, role="scheduler")
         self.assertEqual(get_parallel().dist_init_addr, "1.2.3.4:5000")
         self.assertEqual(
             get_schedule().chunked_prefill_size,
             resolution_result(parent, "chunked_prefill_size"),
-            "publishing the copy re-ran resolution; the bag disagrees with what "
-            "the parent's resolution decided",
+            "publishing re-ran resolution; the bag disagrees with what the "
+            "parent's resolution decided",
         )
 
 
diff --git a/test/registered/unit/server_args/test_resolution_reads_the_declarations.py b/test/registered/unit/server_args/test_resolution_reads_the_declarations.py
index acdc926d4..d04a1a0be 100644
--- a/test/registered/unit/server_args/test_resolution_reads_the_declarations.py
+++ b/test/registered/unit/server_args/test_resolution_reads_the_declarations.py
@@ -86,8 +86,7 @@ def _field_reads(fn, holders):
 _DECLARERS = frozenset(
     {
         "declare_resolution",
-        "declare_late_resolution",
-        "declare_direct_writes",
+        "record_foreign_defaults",
     }
 )
 
@@ -271,7 +270,7 @@ def _record_aliases(function):
                 aliases.add(target.id)
         elif (
             isinstance(func, ast.Attribute)
-            and func.attr in ("from_cli_args", "replace_resolved")
+            and func.attr == "from_cli_args"
             and isinstance(func.value, ast.Name)
             and func.value.id == "ServerArgs"
         ):
diff --git a/test/registered/unit/server_args/test_server_args.py b/test/registered/unit/server_args/test_server_args.py
index e6fbffd19..46c4ec554 100644
--- a/test/registered/unit/server_args/test_server_args.py
+++ b/test/registered/unit/server_args/test_server_args.py
@@ -2886,17 +2886,32 @@ class TestTheInputIsSealedDuringResolution(CustomTestCase):
         with self.assertRaisesRegex(AttributeError, "after resolution"):
             server_args.tp_size = 4
 
-    def test_the_named_exception_lifts_it(self):
-        """`declare_direct_writes` hands the record to an out-of-tree platform
-        plugin that sets fields on it; that is the only channel."""
-        from sglang.srt.server_args import record_writable
+    def test_it_has_no_exception(self):
+        """A resolver from outside this tree assigns fields -- an interface this
+        tree does not own -- and it still does not reach the record.
+
+        `record_foreign_defaults` hands it a stand-in: the assignment is
+        captured and declared, the field keeps the operator's input, and the
+        seal stays armed for the whole call. There used to be a named lift for
+        this, which made the record the one thing resolution could write.
+        """
+        from sglang.srt.arg_groups.overrides import (
+            record_foreign_defaults,
+            resolution_result,
+        )
 
         server_args = ServerArgs(model_path="dummy", device="cuda")
         object.__setattr__(server_args, "_input_frozen", True)
-        with record_writable(server_args):
-            server_args.tp_size = 4
-        self.assertEqual(server_args.tp_size, 4)
-        # and it goes back on afterwards
+
+        def foreign(config):
+            # What a plugin does: read what is decided, assign a default.
+            assert config.tp_size == 1
+            config.tp_size = 4
+
+        record_foreign_defaults(server_args, "platform:probe", foreign)
+
+        self.assertEqual(resolution_result(server_args, "tp_size"), 4)
+        self.assertEqual(server_args.tp_size, 1, "the record is the input")
         with self.assertRaisesRegex(AttributeError, "during resolution"):
             server_args.tp_size = 8
 
@@ -2942,15 +2957,6 @@ class TestLaunchCommand(CustomTestCase):
             server_args.launch_command,
         )
 
-    def test_a_copy_keeps_it(self):
-        """`replace_resolved` is how the Ray paths rewrite `dist_init_addr`;
-        the copy was launched by whatever launched its parent."""
-        server_args = prepare_server_args(["--model-path", "/tmp/x"])
-        self.assertEqual(
-            server_args.replace_resolved("test").launch_command,
-            server_args.launch_command,
-        )
-
     def test_it_is_not_a_config_field(self):
         """It describes how the configuration was asked for, so it is not part
         of the configuration: no CLI flag, no namespace, not in the bags."""
diff --git a/test/registered/unit/test_chain_read_ratchet.py b/test/registered/unit/test_chain_read_ratchet.py
index 4eca0f836..362eb008b 100644
--- a/test/registered/unit/test_chain_read_ratchet.py
+++ b/test/registered/unit/test_chain_read_ratchet.py
@@ -40,7 +40,7 @@ _OWNERS = ("server_args.py", "runtime_context.py", "arg_groups/")
 # startup default wherever it is written, and `benchmark/` ships too.
 _READS_SCANNED = _PACKAGE
 
-_DECLARERS = ("declare_resolution", "declare_late_resolution")
+_DECLARERS = ("declare_resolution",)
 
 
 def _declared_by_keyword():
@@ -239,27 +239,6 @@ def _declared_by_registry_and_passes():
     return fields
 
 
-def _declared_by_late_resolution():
-    """Keywords of `declare_late_resolution(record, ...)`, the late spelling.
-
-    The fields sit at the call sites rather than in the declarer, so a scan
-    that only knew the declarer's own definition would find none of them.
-    """
-    # The record plus `arg_groups/`: a hook calls it on the record it was
-    # handed, so scanning the record's file alone finds nothing.
-    sources = [_SRT / "server_args.py", *sorted((_SRT / "arg_groups").rglob("*.py"))]
-    fields = set()
-    for source in sources:
-        for node in ast.walk(ast.parse(source.read_text(encoding="utf-8-sig"))):
-            if (
-                isinstance(node, ast.Call)
-                and isinstance(node.func, ast.Name)
-                and node.func.id == "declare_late_resolution"
-            ):
-                fields |= {keyword.arg for keyword in node.keywords if keyword.arg}
-    return fields
-
-
 def _written_after_publish():
     """Fields the runtime overrides once the bags exist.
 
@@ -286,7 +265,6 @@ def _resolution_written():
     return (
         _declared_by_keyword()
         | _declared_by_registry_and_passes()
-        | _declared_by_late_resolution()
         | _written_after_publish()
     )
 
@@ -561,7 +539,6 @@ class TestNoChainReadsOfResolvedConfig(CustomTestCase):
 
         by_keyword = _declared_by_keyword()
         by_data = _declared_by_registry_and_passes()
-        by_late = _declared_by_late_resolution()
 
         self.assertGreater(
             len(by_keyword),
@@ -586,24 +563,10 @@ class TestNoChainReadsOfResolvedConfig(CustomTestCase):
             f"{len(overrides.POST_PROCESS_PASSES)} passes; the scan of the "
             "dict-key channel broke",
         )
-        self.assertGreaterEqual(
-            len(by_late),
-            3,
-            f"only {len(by_late)} fields are declared late; the "
-            "`declare_late_resolution` keyword scan broke",
-        )
         # The data channel is not the keyword scan's subset: if it became one,
         # that scan would be doing all the work and a regression here would be
-        # invisible. The late channel *is* a subset, and deliberately so --
-        # `declare_late_resolution` is a keyword declarer like the others now
-        # that the record hosts no forwarding member, so its own floor above is
-        # what pins it.
+        # invisible.
         self.assertTrue(by_data - by_keyword, "the data channel adds nothing")
-        self.assertTrue(
-            by_late <= by_keyword,
-            "late resolution declares outside the keyword channel; it is the "
-            "same spelling, so the two cannot disagree",
-        )
 
     def test_nothing_reads_a_resolved_field_off_a_borrowed_record(self):
         found = _chain_reads(_resolution_written())
diff --git a/test/registered/unit/test_supplied_instance_exposure_ratchet.py b/test/registered/unit/test_supplied_instance_exposure_ratchet.py
index 6dd35993f..a2d9219ab 100644
--- a/test/registered/unit/test_supplied_instance_exposure_ratchet.py
+++ b/test/registered/unit/test_supplied_instance_exposure_ratchet.py
@@ -112,7 +112,7 @@ _MATRIX = (
     {"enable_mis": True, "attention_backend": "flashinfer"},
 )
 
-# `declare_late_resolution` call sites whose keyword expansion is built
+# `declare_resolution` call sites whose keyword expansion is built
 # dynamically; the written fields are spelled out here and drift-guarded.
 _LATE_RESOLUTION_DYNAMIC_SITES = {
     "parser/template_detection.py": frozenset({"reasoning_parser", "tool_call_parser"}),
@@ -341,10 +341,10 @@ class TestSuppliedInstanceExposure(CustomTestCase):
         union does not depend on matrix order; and the ambient CI marker is
         cleared, so a runner's identity cannot leak into the measurement --
         the CI-conditioned writes come from `_ENV_MATRIX`'s explicit entry.
-        Late resolution counts too: `declare_late_resolution` writers run at
-        launcher stage (LoRA normalization, parser auto-detection), so their
-        target fields are collected statically from the call sites -- they are
-        resolution writes by definition, just staged after `__post_init__`.
+        Declarers outside `arg_groups/` count too: the parser auto-detection
+        runs at launcher stage and the NPU helper is called by the pipeline, so
+        their target fields are collected statically from the call sites --
+        resolution writes by definition, just not reached by the matrix.
         """
         pristine = (dict(os.environ), self._env_field_flags())
         written = set()
@@ -383,7 +383,7 @@ class TestSuppliedInstanceExposure(CustomTestCase):
         for extra, env in _ENV_MATRIX:
             resolve_one(extra, env)
         self._restore_process_state(pristine)
-        written |= self._late_resolution_written_fields()
+        written |= self._declared_outside_the_pipeline()
         written |= self._hook_assignment_targets()
         written |= self._record_method_assignment_targets()
         written |= self._declarative_override_fields()
@@ -607,22 +607,32 @@ class TestSuppliedInstanceExposure(CustomTestCase):
                         fields.add(key.value)
         return fields
 
-    def _late_resolution_written_fields(self) -> set:
-        """Fields `declare_late_resolution` writes, collected statically.
+    def _declared_outside_the_pipeline(self) -> set:
+        """Fields declared by a `declare_resolution` caller outside
+        `arg_groups/`, collected statically.
 
-        These are resolution's launcher-stage writes (they need a tokenizer or
-        adapter load, so they cannot run in `__post_init__`), which the
-        construct-and-diff pass above never sees. The keywords at the call
-        sites are the written fields; an expansion this cannot resolve fails
-        loudly like the override collector's, except the named dynamic sites
-        below, whose field sets are spelled out and drift-guarded (each name
-        must still appear as a constant in the file)."""
+        Resolution's launcher-stage writes live here -- the auto-detected
+        parsers need a tokenizer or chat-template load, so they cannot run in
+        `__post_init__` -- alongside the NPU default helper and the expert-pack
+        loader, which the pipeline calls the same way. The construct-and-diff
+        pass above never sees any of them.
+
+        `arg_groups/` is deliberately excluded: `_hook_assignment_targets`
+        covers it exactly, and it resolves the pipeline's own computed
+        expansions (`record_foreign_defaults` declares a `**` dict this
+        collector's resolver cannot read). The keywords at the call sites are
+        the written fields; an expansion this cannot resolve fails loudly like
+        the override collector's, except the named dynamic sites below, whose
+        field sets are spelled out and drift-guarded (each name must still
+        appear as a constant in the file)."""
         written = set()
         root = _PACKAGE_ROOT
         for path in sorted(root.rglob("*.py")):
             rel = path.relative_to(root).as_posix()
+            if rel.startswith("arg_groups/"):
+                continue
             source = path.read_text(encoding="utf-8-sig")
-            if "declare_late_resolution" not in source:
+            if "declare_resolution" not in source:
                 continue
             try:
                 tree = ast.parse(source)
@@ -634,12 +644,12 @@ class TestSuppliedInstanceExposure(CustomTestCase):
                     and (
                         (
                             isinstance(node.func, ast.Name)
-                            and node.func.id == "declare_late_resolution"
+                            and node.func.id == "declare_resolution"
                         )
                         or (
                             isinstance(node.func, ast.Attribute)
                             and node.func.attr
-                            in ("declare_late_resolution", "_late_resolution")
+                            in ("declare_resolution", "_declare_resolution")
                         )
                     )
                 ):