diff --git a/python/sglang/srt/disaggregation/encoder/http_server.py b/python/sglang/srt/disaggregation/encoder/http_server.py
index d87a020d7..1e952c2a3 100644
--- a/python/sglang/srt/disaggregation/encoder/http_server.py
+++ b/python/sglang/srt/disaggregation/encoder/http_server.py
@@ -206,8 +206,6 @@ def launch_server(server_args: ServerArgs):
     global dp_dispatcher, encoder, encoder_scheduler, local_runtime, send_sockets
 
     configure_logger(server_args, prefix=" encode_server")
-    # Publish before the launch path reads configuration; each encoder built
-    # below re-projects the same object in its process.
     publish(server_args, role="encoder")
     if get_parallel().dp_size > 1:
         dp_dispatcher = launch_dp_runtime(server_args)
diff --git a/python/sglang/srt/disaggregation/encoder/server.py b/python/sglang/srt/disaggregation/encoder/server.py
index eb6703ac6..3d9596d81 100644
--- a/python/sglang/srt/disaggregation/encoder/server.py
+++ b/python/sglang/srt/disaggregation/encoder/server.py
@@ -57,12 +57,12 @@ from sglang.srt.multimodal.encoder_preprocessing import (
 )
 from sglang.srt.observability.metrics_collector import EncoderMetricsCollector
 from sglang.srt.runtime_context import (
+    ensure_published,
     get_device,
     get_disagg,
     get_exec,
     get_mm,
     get_model,
-    publish,
 )
 from sglang.srt.server_args import ServerArgs
 from sglang.srt.utils import configure_media_url_security
@@ -448,9 +448,7 @@ class MMEncoder:
         ``base_gpu_id + rank`` — the DP launcher's per-worker placement. It is
         this instance's value, not a config change, so it travels as an
         argument."""
-        # The DP and TP encoder workers are spawned, so this constructor is
-        # the first publish in those processes.
-        publish(server_args, role="encoder")
+        ensure_published(server_args, role="encoder")
         logger.info(f"init MMEncoder {rank}/{server_args.tp_size}")
         self.server_args = server_args
         configure_media_url_security(
diff --git a/python/sglang/srt/entrypoints/engine.py b/python/sglang/srt/entrypoints/engine.py
index 5e8c2fc8b..59f4e4f35 100644
--- a/python/sglang/srt/entrypoints/engine.py
+++ b/python/sglang/srt/entrypoints/engine.py
@@ -1117,9 +1117,6 @@ class Engine(EngineScoreMixin, EngineBase):
         ):
             resolve_auto_parsers(server_args)
 
-        # Resolution is complete here; this process goes on to host the
-        # tokenizer manager or the multi-tokenizer router, whose own publish
-        # re-projects the same object.
         publish(server_args, role="tokenizer")
 
         # Launch daemons (daemon mode only). The handles travel back to the
diff --git a/python/sglang/srt/managers/tokenizer_manager.py b/python/sglang/srt/managers/tokenizer_manager.py
index 3bef7c669..3297aa8f7 100644
--- a/python/sglang/srt/managers/tokenizer_manager.py
+++ b/python/sglang/srt/managers/tokenizer_manager.py
@@ -122,6 +122,7 @@ from sglang.srt.observability.request_metrics_exporter import (
 )
 from sglang.srt.observability.trace import SpanAttributes, extract_trace_headers
 from sglang.srt.runtime_context import (
+    ensure_published,
     get_context,
     get_device,
     get_disagg,
@@ -138,7 +139,6 @@ from sglang.srt.sampling.sampling_params import SamplingParams
 from sglang.srt.server_args import (
     PortArgs,
     ServerArgs,
-    set_global_server_args_for_tokenizer,
 )
 from sglang.srt.utils import (
     configure_gc_warning,
@@ -405,9 +405,7 @@ class TokenizerManager(TokenizerControlMixin, TokenizerManagerScoreMixin):
     ):
         # Parse args
         self.server_args = server_args
-        # In a tokenizer-worker process this is the process's first publish;
-        # the in-process path re-projects the object the launcher published.
-        set_global_server_args_for_tokenizer(server_args)
+        ensure_published(server_args, role="tokenizer")
         self.startup_time: Optional[Dict[str, Any]] = None
         self.elastic_worker_count = get_parallel().dp_size
         self.elastic_pending_ep_size = None
diff --git a/python/sglang/srt/model_executor/model_runner.py b/python/sglang/srt/model_executor/model_runner.py
index 59583db17..ccee400dd 100644
--- a/python/sglang/srt/model_executor/model_runner.py
+++ b/python/sglang/srt/model_executor/model_runner.py
@@ -168,6 +168,7 @@ from sglang.srt.model_executor.runner import (
 )
 from sglang.srt.platforms import current_platform
 from sglang.srt.runtime_context import (
+    ensure_published,
     get_context,
     get_exec,
     get_global_dwdp_manager,
@@ -188,7 +189,6 @@ from sglang.srt.server_args import (  # noqa: F401  (re-export)
     ServerArgs,
     add_chunked_prefix_cache_attention_backend,
     get_global_server_args,
-    set_global_server_args_for_scheduler,
 )
 from sglang.srt.speculative.spec_info import SpeculativeAlgorithm
 from sglang.srt.speculative.spec_utils import resolve_num_tokens_per_req
@@ -336,7 +336,7 @@ class ModelRunner:
         # construction (benchmark/one_batch, the manual runner tests) has no
         # earlier publish.
         if not is_draft_worker:
-            set_global_server_args_for_scheduler(server_args)
+            ensure_published(server_args, role="scheduler")
         # Set by maybe_init_lora_manager; stays None when LoRA is off and on
         # draft runners, which serve adapters' target model unadapted.
         self.lora_manager: Optional[LoRAManager] = None
diff --git a/python/sglang/srt/runtime_context.py b/python/sglang/srt/runtime_context.py
index f769853fe..4819091f5 100644
--- a/python/sglang/srt/runtime_context.py
+++ b/python/sglang/srt/runtime_context.py
@@ -48,6 +48,7 @@ from __future__ import annotations
 
 import dataclasses
 import functools
+import logging
 import math
 import os
 import sys
@@ -57,6 +58,8 @@ from typing import TYPE_CHECKING, Any
 if TYPE_CHECKING:
     from sglang.srt.server_args import ServerArgs
 
+logger = logging.getLogger(__name__)
+
 
 # Imported lazily so this module has no import-time dependencies: any module can
 # import get_parallel at module level without risking an import cycle.
@@ -1304,7 +1307,18 @@ def publish(server_args, *, role: str, hf_config: Any = None) -> RuntimeContext:
             f"publish role {role!r} has no ROLE_NAMESPACE_SETS entry; declare "
             "its namespace set (None for the full tree)."
         )
+    discarded = _CONTEXT.overrides_log()
     _CONTEXT.set_server_args(server_args)
+    if discarded:
+        logger.warning(
+            "publish(role=%s) re-projected the config bags and dropped %d "
+            "override(s) taken since the last publish: %s",
+            role,
+            len(discarded),
+            ", ".join(
+                f"{source}({', '.join(sorted(fields))})" for source, fields in discarded
+            ),
+        )
     _CONTEXT._publish_role = role
     if _ROLE_NS_MODE == "record":
         # The '-' marker distinguishes a zero-read role from a process where
@@ -1320,6 +1334,31 @@ def publish(server_args, *, role: str, hf_config: Any = None) -> RuntimeContext:
     return _CONTEXT
 
 
+def ensure_published(server_args, *, role: str) -> RuntimeContext:
+    """Publish unless this exact record is already published under this role.
+
+    Three constructors publish defensively, because each can be built with
+    nothing published before it -- `ModelRunner` (a benchmark harness, the
+    manual runner tests), `TokenizerManager`, and `MMEncoder` (spawned encoder
+    workers). Inside a process that already published the same record,
+    publishing again re-projects the bags: every `override()` taken between the
+    two calls is discarded, and the provenance log with it.
+
+    No override sits in one of those windows today, so this removes a hazard
+    rather than a live bug. It is worth removing anyway: the drop is silent, it
+    depends on where a constructor happens to sit relative to the overrides
+    around it, and `publish` now says what a re-projection discarded so the
+    next one is loud.
+
+    So these callers ask for the end state -- this record, this role, published
+    -- and get a no-op when that already holds. An engine rebuild still calls
+    `publish` directly, because there the reset is the point.
+    """
+    if _CONTEXT._server_args is server_args and _CONTEXT._publish_role == role:
+        return _CONTEXT
+    return publish(server_args, role=role)
+
+
 def publish_role() -> str | None:
     """The role recorded by the last ``publish`` (None for a legacy set)."""
     return _CONTEXT._publish_role
diff --git a/python/sglang/test/config_publishers.py b/python/sglang/test/config_publishers.py
new file mode 100644
index 000000000..f473403a7
--- /dev/null
+++ b/python/sglang/test/config_publishers.py
@@ -0,0 +1,117 @@
+"""Who installs the startup record into the runtime context, derived from code.
+
+Two guards need this answer and neither should keep its own list: matching the
+spellings by hand is how `ensure_published` once read as *not* publishing,
+which turned a correct module into a reported violation. A publisher is
+defined by what it does -- it reaches ``RuntimeContext.set_server_args`` --
+and a *constructor* publisher is an ``__init__`` that calls one.
+"""
+
+from __future__ import annotations
+
+import ast
+import pathlib
+from typing import Dict, Set, Tuple
+
+_DEFINING_MODULES = ("runtime_context.py", "server_args.py")
+
+
+def publisher_names(srt_root: pathlib.Path) -> frozenset:
+    """Module-level functions that transitively install the record."""
+    reaches: Set[str] = set()
+    graph: Dict[str, Set[str]] = {}
+    for relative in _DEFINING_MODULES:
+        tree = ast.parse((srt_root / relative).read_text(encoding="utf-8-sig"))
+        for node in tree.body:
+            if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
+                continue
+            called: Set[str] = set()
+            installs = False
+            for inner in ast.walk(node):
+                if not isinstance(inner, ast.Call):
+                    continue
+                if (
+                    isinstance(inner.func, ast.Attribute)
+                    and inner.func.attr == "set_server_args"
+                ):
+                    installs = True
+                elif isinstance(inner.func, ast.Name):
+                    called.add(inner.func.id)
+            graph[node.name] = called
+            if installs:
+                reaches.add(node.name)
+    growing = True
+    while growing:
+        growing = False
+        for name, called in graph.items():
+            if name not in reaches and called & reaches:
+                reaches.add(name)
+                growing = True
+    return frozenset(reaches)
+
+
+def constructor_publishers(srt_root: pathlib.Path) -> Set[Tuple[str, str, str]]:
+    """``{(module, class, publisher)}`` for every ``__init__`` that publishes.
+
+    Keyed by the owning class, not just the module: two constructors in one
+    module would otherwise collapse into a single entry, and a *new* defensive
+    publish added next to a listed one would leave the census unchanged --
+    which is exactly the "adding one fails the pin" property it exists for.
+
+    Reached through a local name or through a module attribute
+    (``runtime_context.publish(...)``), and through a helper defined in the
+    same module -- a constructor that publishes one hop away is the same
+    hazard as one that publishes directly.
+    """
+    publishers = publisher_names(srt_root)
+    found: Set[Tuple[str, str, str]] = set()
+    for path in sorted(srt_root.rglob("*.py")):
+        try:
+            tree = ast.parse(path.read_text(encoding="utf-8-sig"))
+        except SyntaxError:
+            raise AssertionError(f"unparsable module in the census: {path}")
+        local: Dict[str, ast.AST] = {
+            node.name: node
+            for node in ast.walk(tree)
+            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
+        }
+
+        def publishes(node, seen=None):
+            """The publisher this callable reaches, if any."""
+            seen = seen if seen is not None else set()
+            for inner in ast.walk(node):
+                if not isinstance(inner, ast.Call):
+                    continue
+                if isinstance(inner.func, ast.Name):
+                    name = inner.func.id
+                elif isinstance(inner.func, ast.Attribute):
+                    name = inner.func.attr
+                else:
+                    continue
+                if name in publishers:
+                    return name
+                if name in local and name not in seen:
+                    seen.add(name)
+                    reached = publishes(local[name], seen)
+                    if reached is not None:
+                        return reached
+            return None
+
+        for owner in ast.walk(tree):
+            if not isinstance(owner, ast.ClassDef):
+                continue
+            for node in owner.body:
+                if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
+                    continue
+                if node.name != "__init__":
+                    continue
+                reached = publishes(node)
+                if reached is not None:
+                    found.add(
+                        (
+                            path.relative_to(srt_root).as_posix(),
+                            owner.name,
+                            reached,
+                        )
+                    )
+    return found
diff --git a/test/registered/unit/test_legacy_global_ratchet.py b/test/registered/unit/test_legacy_global_ratchet.py
index a0ecc59d2..294d8844b 100644
--- a/test/registered/unit/test_legacy_global_ratchet.py
+++ b/test/registered/unit/test_legacy_global_ratchet.py
@@ -31,7 +31,7 @@ _RATCHETS = [
     (
         "set_global_server_args_for_*",
         r"\bset_global_server_args_for_(?:scheduler|tokenizer)\s*\(",
-        4,
+        2,
     ),
 ]
 
diff --git a/test/registered/unit/test_publish_precedes_bag_reads.py b/test/registered/unit/test_publish_precedes_bag_reads.py
index 377429cf0..04b5a8e1d 100644
--- a/test/registered/unit/test_publish_precedes_bag_reads.py
+++ b/test/registered/unit/test_publish_precedes_bag_reads.py
@@ -42,6 +42,7 @@ import unittest
 
 import sglang
 from sglang.test.ci.ci_register import register_cpu_ci
+from sglang.test.config_publishers import publisher_names
 from sglang.test.test_utils import CustomTestCase
 
 register_cpu_ci(est_time=40, suite="base-a-test-cpu")
@@ -158,6 +159,9 @@ def _calls(fn):
     return out
 
 
+_PUBLISH_NAMES = publisher_names(_PACKAGE_ROOT / "srt")
+
+
 class _Module:
     """One parsed module: what it calls the config API, and what it defines.
 
@@ -183,9 +187,7 @@ class _Module:
                 if node.module in _CONFIG_MODULES:
                     for alias in node.names:
                         local = alias.asname or alias.name
-                        if alias.name == "publish" or alias.name.startswith(
-                            "set_global_server_args"
-                        ):
+                        if alias.name in _PUBLISH_NAMES:
                             self.publishers.add(local)
                         elif alias.name in _ACCESSORS:
                             self.accessors.add(local)
diff --git a/test/registered/unit/test_runtime_context.py b/test/registered/unit/test_runtime_context.py
index 6e08c0dd2..7547f75bd 100644
--- a/test/registered/unit/test_runtime_context.py
+++ b/test/registered/unit/test_runtime_context.py
@@ -19,11 +19,15 @@ from sglang.srt.runtime_context import (
     ParallelContext,
     RuntimeContext,
     _FlagGroupBase,
+    ensure_published,
     get_context,
+    get_exec,
     get_flags,
     get_parallel,
     get_server_args,
     max_speculative_num_draft_tokens,
+    publish,
+    publish_role,
     reset_context,
 )
 from sglang.srt.server_args import ServerArgs
@@ -248,6 +252,119 @@ class TestServerArgsOwnership(_IsolatedServerArgs):
             get_server_args()
 
 
+class TestEnsurePublished(_IsolatedServerArgs):
+    """A defensive publish must not re-project over a live process.
+
+    Three constructors publish because each can be built with nothing published
+    first: `ModelRunner`, `TokenizerManager`, `MMEncoder`. Inside a process that
+    already published the same record, publishing again re-projects the bags --
+    discarding every `override()` taken since, and the provenance log with it.
+
+    No current override sits in one of those windows, so what these assertions
+    protect is the mechanism, not a reproduction: the drop is silent and depends
+    on where a constructor happens to sit relative to the overrides around it.
+    """
+
+    def _record(self, **fields):
+        return ServerArgs(model_path="dummy", **fields)
+
+    def test_a_second_publish_of_the_same_record_keeps_the_overrides(self):
+        record = self._record(grammar_backend="xgrammar")
+        publish(record, role="scheduler")
+        get_context().override("grammar.import_fallback", grammar_backend="none")
+
+        ensure_published(record, role="scheduler")
+
+        self.assertEqual(
+            get_exec().kernel.grammar_backend,
+            "none",
+            "the constructor's publish re-projected the bags, so the import "
+            "fallback was discarded and the process reports a backend it is "
+            "not using",
+        )
+        self.assertEqual(
+            len(get_context().overrides_log()),
+            1,
+            "the provenance of the override went with it",
+        )
+
+    def test_a_different_record_is_published(self):
+        first = self._record(grammar_backend="xgrammar")
+        publish(first, role="scheduler")
+        second = self._record(grammar_backend="llguidance")
+
+        ensure_published(second, role="scheduler")
+
+        self.assertIs(get_server_args(), second)
+        self.assertEqual(get_exec().kernel.grammar_backend, "llguidance")
+
+    def test_an_empty_slot_is_published(self):
+        """The standalone case the defensive publish exists for."""
+        reset_context()
+        record = self._record(grammar_backend="xgrammar")
+
+        ensure_published(record, role="scheduler")
+
+        self.assertIs(get_server_args(), record)
+        self.assertEqual(publish_role(), "scheduler")
+
+    def test_the_same_record_under_a_different_role_is_republished(self):
+        """The role decides which namespaces this process may read."""
+        record = self._record()
+        publish(record, role="tokenizer")
+
+        ensure_published(record, role="scheduler")
+
+        self.assertEqual(publish_role(), "scheduler")
+
+    def test_every_constructor_that_publishes_is_classified(self):
+        """A new constructor publish has to say which of the two it is.
+
+        Publishing in a constructor is right when the constructor *is* the
+        entry -- a spawned worker, the Ray actor that stands in for
+        `run_scheduler_process`, an `Engine` being (re)built, where resetting
+        the bags is the point -- and wrong when the process is already live
+        with the same record, where it silently drops overrides. The
+        difference is not visible in the syntax, so the census is pinned:
+        adding one fails here until it is classified.
+
+        Both the publisher set and "which `__init__` reaches one" come from
+        `sglang.test.config_publishers`, which derives them from the code --
+        a hand-written spelling list here missed a constructor that publishes
+        one hop away through a helper. The derivation follows helpers defined
+        in the same module; a constructor that publishes through a helper in
+        *another* module is not seen, which is the one hole left here.
+        """
+        import pathlib
+
+        import sglang
+        from sglang.test.config_publishers import constructor_publishers
+
+        srt = pathlib.Path(sglang.__file__).resolve().parent / "srt"
+        self.assertEqual(
+            constructor_publishers(srt),
+            {
+                # Entries: nothing published yet, or a rebuild that must not
+                # inherit the previous engine's runtime overrides.
+                ("entrypoints/engine.py", "Engine", "publish"),
+                ("ray/scheduler_actor.py", "SchedulerActor", "publish"),
+                # Defensive: the process is usually already live with this
+                # record, and `launch_server` publishes before building the
+                # in-process encoder.
+                ("disaggregation/encoder/server.py", "MMEncoder", "ensure_published"),
+                (
+                    "managers/tokenizer_manager.py",
+                    "TokenizerManager",
+                    "ensure_published",
+                ),
+                ("model_executor/model_runner.py", "ModelRunner", "ensure_published"),
+            },
+            "a constructor publishes and this census does not know which kind "
+            "it is; an entry uses publish(), one that may run inside a live "
+            "process with the same record uses ensure_published()",
+        )
+
+
 class TestServerArgsScopedOverride(_IsolatedServerArgs):
     """ctx.override_server_args: the config tier's scoped test override —
     tests force execution paths by overriding the context, not by