runtime_context: per-role namespace enforcement behind SGLANG_ROLE_NAMESPACES (#33172)

publish(role=...) has recorded provenance since the namespace split; this
wires the enforcement the role was reserved for. SGLANG_ROLE_NAMESPACES
selects the mode:

- off (default): no bookkeeping; the mode gate in config_bag stays a
  dead-branch-prunable check under dynamo (bag reads run inside compiled
  forwards — pinned by a fullgraph test).
- record: audit mode — collect (role, namespace) pairs per process and
  persist each new pair immediately to SGLANG_ROLE_NAMESPACES_OUT (worker
  teardown skips atexit), plus a per-process stderr summary at exit.
- enforce: a bag read outside the role's ROLE_NAMESPACE_SETS entry fails
  closed with an actionable error; None entries mean full tree.

Sets are filled only where audits back them: dp_controller reads only
exec (record-mode plain + DP-attention smokes agree with the module's
static read set — the elastic-EP gate). tokenizer observed zero bag
reads (per-instance managers read self.server_args by design) but keeps
the full tree until the multi-tokenizer disagg shape (TokenizerWorker's
get_disagg read) is audited; encoder / expert_backup /
weight_cache_daemon likewise await their deployment shapes.

Verified end-to-end: DP-attention smoke under enforce boots and serves
with zero violations.

test/registered/unit/test_runtime_context_config_bags.py

@@ -7,6 +7,7 @@ by bare assignment and fail closed until published.
 
 import dataclasses
 import unittest
+from unittest import mock
 
 from sglang.srt import runtime_context as rc
 from sglang.srt.arg_groups.arg_utils import NS, A
@@ -127,5 +128,114 @@ class TestConfigBagTree(CustomTestCase):
             rc._build_config_bags(_CollisionFake())
 
 
+class TestRoleNamespaceEnforcement(CustomTestCase):
+    """SGLANG_ROLE_NAMESPACES: off (default) is free; record collects the
+    per-role read audit; enforce fails closed on reads outside the role's
+    declared set."""
+
+    def setUp(self):
+        rc.reset_context()
+
+    def tearDown(self):
+        rc.reset_context()
+
+    def _publish(self, role):
+        rc.publish(ServerArgs(model_path="dummy"), role=role)
+
+    def test_off_mode_ignores_declared_sets(self):
+        self._publish("test")
+        with mock.patch.dict(rc.ROLE_NAMESPACE_SETS, {"test": frozenset({"serving"})}):
+            rc.get_exec()  # off mode: no enforcement despite the narrow set
+
+    def test_enforce_blocks_reads_outside_the_declared_set(self):
+        self._publish("test")
+        with mock.patch.object(rc, "_ROLE_NS_MODE", "enforce"), mock.patch.dict(
+            rc.ROLE_NAMESPACE_SETS, {"test": frozenset({"serving", "schedule"})}
+        ):
+            rc.get_serving()
+            rc.get_schedule()
+            with self.assertRaisesRegex(ValueError, "outside the declared set"):
+                rc.get_exec()
+
+    def test_enforce_full_tree_role_and_roleless_install_are_unrestricted(self):
+        with mock.patch.object(rc, "_ROLE_NS_MODE", "enforce"):
+            self._publish("scheduler")  # None in the table = full tree
+            rc.get_exec()
+            rc.get_mm()
+            # A direct set_server_args install is roleless; enforcement only
+            # keys off a recorded publish role.
+            rc.get_context().set_server_args(ServerArgs(model_path="dummy"))
+            rc.get_exec()
+
+    def test_off_mode_bag_read_traces_under_torch_compile(self):
+        # config_bag runs inside compiled model forwards; the mode gate must
+        # stay a dead-branch-prunable check in the default "off" mode.
+        import torch
+
+        self._publish("test")
+
+        @torch.compile(fullgraph=True, backend="eager", dynamic=False)
+        def probe(x):
+            if rc.get_schedule().max_running_requests is None:
+                return x + 1
+            return x * 2
+
+        self.assertEqual(probe(torch.zeros(())).item(), 1.0)
+
+    def test_record_mode_collects_the_audit(self):
+        self._publish("test")
+        with mock.patch.object(rc, "_ROLE_NS_MODE", "record"), mock.patch.object(
+            rc, "_RECORDED_NS_READS", set()
+        ):
+            rc.get_exec()
+            rc.get_disagg()
+            self.assertIn(("test", "exec"), rc._RECORDED_NS_READS)
+            self.assertIn(("test", "disagg"), rc._RECORDED_NS_READS)
+
+    def test_enforce_rejects_roles_missing_from_the_table(self):
+        # Fail closed: an unknown/misspelled role must not silently inherit
+        # the full tree — rejected at publish, and defensively at read time.
+        with mock.patch.object(rc, "_ROLE_NS_MODE", "enforce"):
+            with self.assertRaisesRegex(ValueError, "no ROLE_NAMESPACE_SETS entry"):
+                self._publish("not_a_registered_role")
+            self._publish("scheduler")
+            with mock.patch.dict(rc.ROLE_NAMESPACE_SETS, {}, clear=True):
+                with self.assertRaisesRegex(ValueError, "no ROLE_NAMESPACE_SETS entry"):
+                    rc.get_exec()
+
+    def test_mode_env_value_is_validated(self):
+        self.assertEqual(rc._validated_role_ns_mode(" Enforce "), "enforce")
+        with self.assertRaisesRegex(ValueError, "SGLANG_ROLE_NAMESPACES"):
+            rc._validated_role_ns_mode("bogus")
+
+    def test_record_mode_registers_the_exit_summary_at_publish(self):
+        # A role that reads no bags must still emit its audit line; the exit
+        # hook therefore registers at publish, not at the first read.
+        with mock.patch.object(rc, "_ROLE_NS_MODE", "record"), mock.patch.object(
+            rc, "_RECORD_DUMP_REGISTERED", False
+        ):
+            self._publish("test")
+            self.assertTrue(rc._RECORD_DUMP_REGISTERED)
+
+    def test_record_mode_bag_read_traces_under_torch_compile(self):
+        # Recording has side effects dynamo must never trace; the
+        # is_compiling() probe prunes them, keeping fullgraph capture legal
+        # even in record mode.
+        import torch
+
+        self._publish("test")
+        with mock.patch.object(rc, "_ROLE_NS_MODE", "record"), mock.patch.object(
+            rc, "_RECORDED_NS_READS", set()
+        ):
+
+            @torch.compile(fullgraph=True, backend="eager", dynamic=False)
+            def probe(x):
+                if rc.get_schedule().max_running_requests is None:
+                    return x + 1
+                return x * 2
+
+            self.assertEqual(probe(torch.zeros(())).item(), 1.0)
+
+
 if __name__ == "__main__":
     unittest.main()