refactor: rename FutureMap to Relayer

python/sglang/srt/disaggregation/decode.py

@@ -151,6 +151,9 @@ def alloc(self, reqs: List["Req"]) -> Optional[List[int]]:
         assert (
             len(reusing) <= 1
         ), "only one chunked request may reuse req_pool_idx in a batch"
+        # ``req.kv_committed_len`` is updated in place by
+        # _resolve_spec_overlap_tokens (main-style); attribute is the
+        # post-iter value, no channel resolve needed.
         assert all(
             reqs[i].inflight_middle_chunks > 0 or reqs[i].kv_committed_len > 0
             for i in reusing
@@ -1216,7 +1219,13 @@ def _pre_alloc(
 
         fill_len = len(req.origin_input_ids) + max(len(req.output_ids) - 1, 0)
         req.kv_allocated_len = fill_len
+        # disagg-init kv_committed_len: this is the schedule-side
+        # initialization of a freshly-received KV-ready req, not a
+        # cross-iter relay value, so it remains a direct attribute write.
+        # The Relayer cpu_value channel only owns the per-iter delta
+        # produced by process_batch_result.
         req.kv_committed_len = fill_len
+        req.clear_relayer_kv_committed_ctx()
 
         if prefix_len > 0:
             self.req_to_token_pool.write(
@@ -1331,7 +1340,12 @@ def _pre_alloc(
         # Truncate fill_ids to kv_committed_len so cache_unfinished_req only
         # inserts committed KV into the radix tree. The last output token
         # hasn't had KV committed yet (fill_ids is 1 ahead).
-        req.fill_ids = (req.origin_input_ids + req.output_ids)[: req.kv_committed_len]
+        # Route through Relayer cpu_value channel when a kv_committed ctx is
+        # attached to ``req`` (the channel resolve returns ``baseline +
+        # delta`` from the per-iter store); falls back to the attribute.
+        req.fill_ids = (req.origin_input_ids + req.output_ids)[
+            : req.relayer_resolve_kv_committed_len()
+        ]
         # Set prefix_indices so downstream consumers (init_next_round_input,
         # prepare_for_extend) see the correct prefix length. In the agg path
         # this is done inside init_next_round_input, but decode-disagg needs
@@ -1733,8 +1747,11 @@ def get_new_prebuilt_batch(self: Scheduler) -> Optional[ScheduleBatch]:
                 req.init_next_round_input(tree_cache)
                 # Truncate fill_ids to kv_committed_len so cache_unfinished_req
                 # only sees committed KV (fill_ids includes one uncommitted token).
-                if req.kv_committed_len is not None:
-                    req.fill_ids = req.fill_ids[: req.kv_committed_len]
+                # Route through Relayer cpu_value channel when a kv_committed
+                # ctx is attached to the req; falls back to the attribute.
+                committed_len = req.relayer_resolve_kv_committed_len()
+                if committed_len is not None:
+                    req.fill_ids = req.fill_ids[:committed_len]
                     req.set_extend_input_len(
                         len(req.fill_ids) - len(req.prefix_indices)
                     )
@@ -1760,7 +1777,7 @@ def get_new_prebuilt_batch(self: Scheduler) -> Optional[ScheduleBatch]:
 
         # construct fake completed prefill
         new_batch.prepare_for_prebuilt()
-        new_batch.process_prebuilt(self.server_args, self.future_map)
+        new_batch.process_prebuilt(self.server_args, self.relayer)
 
         return new_batch
 

python/sglang/srt/disaggregation/decode_schedule_batch_mixin.py

@@ -13,7 +13,7 @@
 logger = logging.getLogger(__name__)
 
 if TYPE_CHECKING:
-    from sglang.srt.managers.overlap_utils import FutureMap
+    from sglang.srt.managers.overlap_utils import Relayer
     from sglang.srt.managers.schedule_batch import ScheduleBatch
     from sglang.srt.server_args import ServerArgs
 
@@ -104,7 +104,7 @@ def prepare_for_prebuilt(self: ScheduleBatch):
     def process_prebuilt(
         self: ScheduleBatch,
         server_args: ServerArgs,
-        future_map: FutureMap,
+        relayer: Relayer,
     ):
         """Assign the buffered last input id to schedule batch"""
         self.output_ids = []
@@ -176,10 +176,8 @@ def process_prebuilt(
             spec_info.prepare_for_extend(self)
             spec_info.capture_hidden_mode = CaptureHiddenMode.LAST
             if self.enable_overlap:
-                spec_info.future_indices = future_map.alloc_future_indices(
+                spec_info.future_indices = relayer.alloc_future_indices(
                     len(self.seq_lens)
                 )
-                future_map.store_to_map_for_new_batch(
-                    spec_info.future_indices, spec_info
-                )
+                relayer.store_to_map_for_new_batch(spec_info.future_indices, spec_info)
             self.spec_info = spec_info

python/sglang/srt/environ.py

@@ -218,6 +218,14 @@ class Envs:
     SGLANG_TEST_RETRACT_NO_PREFILL_BS = EnvInt(2 ** 31)
     SGLANG_ENABLE_STRICT_MEM_CHECK_DURING_BUSY = EnvInt(0)
     SGLANG_ENABLE_STRICT_MEM_CHECK_DURING_IDLE = EnvBool(True)
+    SGLANG_RELAYER_LOCKSTEP_ASSERT = EnvBool(True)
+    # Strict mode: raise on any ScheduleBatch cross-iter volatile attribute
+    # read from a worker-stream entry frame. Worker reads must go through
+    # ForwardData snapshot or Relayer channel resolve. Default-on so the
+    # "worker never touches live SB" invariant is locked in CI without
+    # contributors having to remember to flip it; set 0 to opt out if the
+    # stack-walk cost shows up in profiling.
+    SGLANG_RELAYER_DEBUG_STRICT = EnvBool(True)
 
     # Scheduler: new token ratio hyperparameters
     SGLANG_INIT_NEW_TOKEN_RATIO = EnvFloat(0.7)