[Perf] Optimize Qwen3-VL unique-image serving on H100 (#36411)

Co-authored-by: Cursor <cursoragent@cursor.com>

python/sglang/srt/arg_groups/fields/memory.py

@@ -41,8 +41,12 @@ class Memory(msgspec.Struct):
                 "for what each policy optimizes for."
             ),
             choices=RADIX_EVICTION_POLICY_CHOICES,
+            resolvable=True,
         ),
     ] = "lru"
+    # The value alone cannot distinguish the default from an explicit LRU
+    # choice, which model-specific defaults must preserve.
+    _radix_eviction_policy_explicitly_set: A[bool, Arg(no_cli=True)] = False
     radix_eviction_policy_config: A[
         Optional[Dict[str, Any]],
         Arg(

python/sglang/srt/arg_groups/fields/mm.py

@@ -99,10 +99,15 @@ class Mm(msgspec.Struct):
     ] = 64
     mm_preprocess_cache_size_mb: A[
         Optional[int],
-        "CPU memory budget for content-addressed multimodal preprocessing "
+        Arg(
-        "artifacts. Unset selects a model-specific default (256 MiB for "
+            help=(
-        "Kimi-K3); 0 disables the cache. The budget is divided across "
+                "CPU memory budget for content-addressed multimodal preprocessing "
-        "tokenizer workers and does not reserve GPU memory.",
+                "artifacts. Unset selects a model-specific default (256 MiB for "
+                "Kimi-K3); 0 disables the cache. The budget is divided across "
+                "tokenizer workers and does not reserve GPU memory."
+            ),
+            resolvable=True,
+        ),
     ] = None
     trust_mm_content_hashes: A[
         bool,
@@ -139,13 +144,18 @@ class Mm(msgspec.Struct):
     ] = False
     mm_feature_transport: A[
         Optional[Literal["cpu", "cuda_ipc", "cuda_vmm"]],
-        "Transport multimodal features through CPU memory, a bounded CUDA IPC "
+        Arg(
-        "pool, or a bounded CUDA VMM pool. "
+            help=(
-        "Unset uses cpu except for validated multi-node GB200/GB300 MNNVL models, "
+                "Transport multimodal features through CPU memory, a bounded CUDA IPC "
-        "which use cuda_vmm when an IMEX channel is available. Select cuda_ipc "
+                "pool, or a bounded CUDA VMM pool. "
-        "explicitly for single-node GPU transport. GPU transports reserve "
+                "Unset uses cpu except for validated multi-node GB200/GB300 MNNVL models, "
-        "SGLANG_MM_FEATURE_CACHE_MB (default 1024 MiB) on the base GPU and fall "
+                "which use cuda_vmm when an IMEX channel is available. Select cuda_ipc "
-        "back to CPU transport when the pool is full.",
+                "explicitly for single-node GPU transport. GPU transports reserve "
+                "SGLANG_MM_FEATURE_CACHE_MB (default 1024 MiB) on the base GPU and fall "
+                "back to CPU transport when the pool is full."
+            ),
+            resolvable=True,
+        ),
     ] = None
     keep_mm_feature_on_device: A[
         bool,

python/sglang/srt/arg_groups/fields/schedule.py

@@ -61,9 +61,12 @@ class Schedule(msgspec.Struct):
         "The maximum number of tokens in a chunk for the chunked prefill. Setting this to -1 means disabling chunked prefill.",
     ] = None
     prefill_decode_interval: A[
-        int,
+        Optional[int],
-        "The number of decode rounds to run after a prefill batch before scheduling the next prefill. In data-parallel attention mode, the interval is synchronized across all DP ranks. Set to 0 to disable.",
+        Arg(
-    ] = 0
+            help="The number of decode rounds to run after a prefill batch before scheduling the next prefill. By default, this is disabled except for profiled Qwen3-VL serving configurations on Hopper. In data-parallel attention mode, the interval is synchronized across all DP ranks. Set to 0 to disable.",
+            resolvable=True,
+        ),
+    ] = None
     enable_dynamic_chunking: A[
         bool,
         "Enable dynamic chunk size adjustment for pipeline parallelism. When enabled, chunk sizes are dynamically calculated based on fitted function to maintain consistent execution time across chunks.",

python/sglang/srt/arg_groups/memory_hook.py

@@ -170,10 +170,17 @@ def handle_gpu_memory_settings(server_args: Any, gpu_mem):
         if decode_cuda_graph_config.max_bs is None:
             decode_cuda_graph_config.max_bs = 160
 
+    from sglang.srt.arg_groups.model_overrides.qwen3_vl import (
+        expand_multimodal_decode_graph_to_running_limit,
+    )
+
+    expand_multimodal_decode_graph_to_running_limit(
+        server_args, decode_cuda_graph_config, gpu_mem
+    )
+
     # ------------------------------------------------------------------
     # CUDA graph batch-size materialization
     # ------------------------------------------------------------------
-
     if cfg.device != "cpu":
         if decode_cuda_graph_config.bs is None:
             decode_cuda_graph_config.bs = generate_decode_cuda_graph_batch_sizes(

python/sglang/srt/arg_groups/model_overrides/qwen3_vl.py

@@ -1,20 +1,69 @@
 """Config-time override declarations for qwen3_vl.
 
-Architectures: Qwen3VLForConditionalGeneration.
+Architectures: Qwen3VLForConditionalGeneration, Qwen3VLMoeForConditionalGeneration.
 """
 
 import logging
-from typing import Any
+from typing import Any, Optional
 
 from sglang.srt.arg_groups.model_override_base import (
     _register_for,
+    model_config_of,
     resolving_view,
 )
 from sglang.srt.environ import envs
 from sglang.srt.runtime_context import get_platform
+from sglang.srt.utils.common import get_device_memory_capacity, is_sm90_supported
 
 logger = logging.getLogger(__name__)
 
+_HOPPER_QWEN3_VL_TYPES = {"qwen3_vl", "qwen3_vl_moe"}
+
+
+def large_hopper_qwen3_vl_model_type(server_args: Any, gpu_mem=None) -> Optional[str]:
+    """Return the HF model_type on large Hopper Qwen3-VL, else None."""
+    cfg = resolving_view(server_args)
+    if gpu_mem is None:
+        gpu_mem = get_device_memory_capacity(cfg.device)
+    if not is_sm90_supported() or gpu_mem is None or gpu_mem < 60 * 1024:
+        return None
+    model_config = model_config_of(server_args)
+    model_type = getattr(model_config.hf_config, "model_type", "")
+    if model_config.is_multimodal and model_type in _HOPPER_QWEN3_VL_TYPES:
+        return model_type
+    return None
+
+
+def expand_multimodal_decode_graph_to_running_limit(
+    server_args: Any, decode_config: Any, gpu_mem
+) -> None:
+    """Keep profiled high-concurrency Qwen3-VL decode inside CUDA graph."""
+    from sglang.srt.model_executor.cuda_graph_config import Phase
+
+    cfg = resolving_view(server_args)
+    locked = getattr(server_args, "_cuda_graph_config_locked", set())
+    max_running_requests = cfg.max_running_requests
+    if not (
+        gpu_mem is not None
+        and max_running_requests is not None
+        and max_running_requests <= 512
+        and decode_config.max_bs is not None
+        and decode_config.max_bs < max_running_requests
+        and (Phase.DECODE, "max_bs") not in locked
+        and (Phase.DECODE, "bs") not in locked
+    ):
+        return
+    if large_hopper_qwen3_vl_model_type(server_args, gpu_mem) is None:
+        return
+
+    logger.info(
+        "Expanding multimodal decode CUDA graph max_bs from %d to "
+        "max_running_requests=%d.",
+        decode_config.max_bs,
+        max_running_requests,
+    )
+    decode_config.max_bs = max_running_requests
+
 
 @_register_for("Qwen3VLForConditionalGeneration")
 def _qwen3vl_overrides(server_args: Any, hf_config: Any) -> dict:
@@ -30,3 +79,63 @@ def _qwen3vl_overrides(server_args: Any, hf_config: Any) -> dict:
         )
         return {"page_size": 16}
     return {}
+
+
+@_register_for(
+    "Qwen3VLForConditionalGeneration",
+    "Qwen3VLMoeForConditionalGeneration",
+)
+def _qwen3vl_hopper_serving_overrides(server_args: Any, hf_config: Any) -> dict:
+    """Select the profiled Qwen3-VL serving path on large Hopper GPUs."""
+    model_type = large_hopper_qwen3_vl_model_type(server_args)
+    if model_type is None:
+        return {}
+
+    cfg = resolving_view(server_args)
+    if not envs.SGLANG_VLM_CACHE_SIZE_MB.is_set():
+        # Repeated-image traffic can opt in to embedding retention. Disable
+        # it by default for streaming traffic, where every image is used once.
+        envs.SGLANG_VLM_CACHE_SIZE_MB.set(0)
+
+    updates = {}
+    preprocess_cache_size_mb = cfg.mm_preprocess_cache_size_mb
+    if preprocess_cache_size_mb is None:
+        preprocess_cache_size_mb = 0
+        updates["mm_preprocess_cache_size_mb"] = 0
+    cache_retention_enabled = (
+        preprocess_cache_size_mb > 0 or envs.SGLANG_VLM_CACHE_SIZE_MB.get() > 0
+    )
+    if (
+        cfg.mm_feature_transport is None
+        and cfg.nnodes == 1
+        and not cache_retention_enabled
+    ):
+        updates["mm_feature_transport"] = "cuda_ipc"
+    resolved_transport = updates.get("mm_feature_transport", cfg.mm_feature_transport)
+    if (
+        not envs.SGLANG_MM_FEATURE_CACHE_MB.is_set()
+        and cfg.max_running_requests is not None
+        and cfg.max_running_requests >= 400
+        and not cache_retention_enabled
+        and resolved_transport == "cuda_ipc"
+    ):
+        # Keep a full high-concurrency wave GPU-resident instead of
+        # falling back to CPU transport while the scheduler drains it.
+        envs.SGLANG_MM_FEATURE_CACHE_MB.set(3 * 1024)
+    if (
+        cfg.radix_eviction_policy == "lru"
+        and not cfg._radix_eviction_policy_explicitly_set
+    ):
+        updates["radix_eviction_policy"] = "priority"
+    if cfg.prefill_decode_interval is None:
+        updates["prefill_decode_interval"] = 22
+    if cfg.attention_backend is None and cfg.decode_attention_backend is None:
+        updates["decode_attention_backend"] = "flashinfer"
+
+    if updates:
+        logger.info(
+            "Applying profiled %s serving defaults on a large Hopper GPU: %s",
+            model_type,
+            updates,
+        )
+    return updates

python/sglang/srt/arg_groups/pipeline.py

@@ -93,6 +93,7 @@ def run_resolution_pipeline(server_args: Any) -> None:
 
     handle_offload_compatibility(server_args)
     from sglang.srt.arg_groups.validation_hook import (
+        default_unset_prefill_decode_interval,
         validate_experimental_sgl_marlin,
         validate_prefill_decode_interval,
         validate_sampling_mask_max_tokens,
@@ -234,6 +235,7 @@ def run_resolution_pipeline(server_args: Any) -> None:
     )
 
     handle_model_specific_adjustments(server_args)
+    default_unset_prefill_decode_interval(server_args)
     # After the model overrides: Qwen4-Exp declares the PLE offload default there.
     handle_offload_compatibility(server_args)
 

python/sglang/srt/arg_groups/validation_hook.py

@@ -435,10 +435,23 @@ def validate_experimental_sgl_marlin(server_args: Any):
 
 def validate_prefill_decode_interval(server_args: Any):
     cfg = resolving_view(server_args)
-    if cfg.prefill_decode_interval < 0:
+    if cfg.prefill_decode_interval is not None and cfg.prefill_decode_interval < 0:
         raise ValueError("--prefill-decode-interval must be non-negative.")
 
 
+def default_unset_prefill_decode_interval(server_args: Any):
+    """Leave Qwen3-VL Hopper free to pick 22; everyone else stays disabled."""
+    from sglang.srt.arg_groups.overrides import declare_resolution
+
+    cfg = resolving_view(server_args)
+    if cfg.prefill_decode_interval is None:
+        declare_resolution(
+            server_args,
+            "prefill_decode_interval_default",
+            prefill_decode_interval=0,
+        )
+
+
 def validate_sampling_mask_max_tokens(server_args: Any):
     if envs.SGLANG_DISAGGREGATION_SAMPLING_MASK_MAX_TOKENS.is_set():
         raise ValueError(

python/sglang/srt/entrypoints/openai/serving_chat.py

@@ -6,6 +6,7 @@ import logging
 import math
 import time
 import uuid
+from collections import OrderedDict
 from enum import Enum
 from http import HTTPStatus
 from typing import TYPE_CHECKING, Any, AsyncGenerator, Dict, List, Optional, Union
@@ -112,6 +113,7 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 _MEDIA_CONTENT_PART_TYPES = frozenset({"image_url", "video_url", "audio_url"})
+_CHAT_TEMPLATE_CACHE_MAX_SIZE = 128
 
 
 def normalize_tool_content(role: str, content):
@@ -351,6 +353,9 @@ class OpenAIServingChat(OpenAIServingBase):
             )
         except Exception:
             self._tokenizer_auto_adds_specials = True
+        self._chat_template_cache: OrderedDict[
+            bytes, tuple[str, tuple[int, ...], str]
+        ] = OrderedDict()
 
     def _handle_last_assistant_message(
         self,
@@ -1342,6 +1347,7 @@ class OpenAIServingChat(OpenAIServingBase):
         """Apply Jinja chat template"""
         prompt = ""
         prompt_ids = []
+        decoded_prompt = None
         openai_compatible_messages = []
         image_data = []
         video_data = []
@@ -1515,16 +1521,14 @@ class OpenAIServingChat(OpenAIServingBase):
                 else {}
             )
             try:
-                rendered_prompt = self.tokenizer_manager.tokenizer.apply_chat_template(
+                rendered_prompt, prompt_ids, decoded_prompt = (
-                    openai_compatible_messages,
+                    self._render_and_encode_chat_template(
-                    tokenize=False,
+                        openai_compatible_messages,
-                    add_generation_prompt=True,
+                        tools=tools,
-                    tools=tools,
+                        template_kwargs=extra_template_kwargs,
-                    return_dict=False,
+                        encode_kwargs=encode_kwargs,
-                    **extra_template_kwargs,
+                        use_cache=is_multimodal,
-                )
+                    )
-                prompt_ids = self.tokenizer_manager.tokenizer.encode(
-                    rendered_prompt, **encode_kwargs
                 )
             except Exception:
                 # If the first attempt fails, try with flat function-only format.
@@ -1535,19 +1539,15 @@ class OpenAIServingChat(OpenAIServingBase):
                     else None
                 )
                 try:
-                    rendered_prompt = (
+                    rendered_prompt, prompt_ids, decoded_prompt = (
-                        self.tokenizer_manager.tokenizer.apply_chat_template(
+                        self._render_and_encode_chat_template(
                             openai_compatible_messages,
-                            tokenize=False,
-                            add_generation_prompt=True,
                             tools=tools,
-                            return_dict=False,
+                            template_kwargs=extra_template_kwargs,
-                            **extra_template_kwargs,
+                            encode_kwargs=encode_kwargs,
+                            use_cache=is_multimodal,
                         )
                     )
-                    prompt_ids = self.tokenizer_manager.tokenizer.encode(
-                        rendered_prompt, **encode_kwargs
-                    )
                 except _CHAT_TEMPLATE_CLIENT_ERRORS as template_error:
                     # Template errors (e.g., from raise_exception in Jinja templates)
                     # and TypeError (e.g., tojson filter on Jinja2 Undefined variables)
@@ -1559,9 +1559,15 @@ class OpenAIServingChat(OpenAIServingBase):
                 prompt_ids = self._append_assistant_prefix_to_prompt_ids(
                     prompt_ids, assistant_prefix
                 )
+                # The cached decode corresponds to prompt_ids before the prefix.
+                decoded_prompt = None
 
             if is_multimodal:
-                prompt = self.tokenizer_manager.tokenizer.decode(prompt_ids)
+                prompt = (
+                    decoded_prompt
+                    if decoded_prompt is not None
+                    else self.tokenizer_manager.tokenizer.decode(prompt_ids)
+                )
 
         stop = request.stop
         image_data = image_data if image_data else None
@@ -1578,6 +1584,70 @@ class OpenAIServingChat(OpenAIServingBase):
             stop=stop,
         )
 
+    def _render_and_encode_chat_template(
+        self,
+        messages: List[Dict[str, Any]],
+        *,
+        tools: Optional[List[Dict]],
+        template_kwargs: Dict[str, Any],
+        encode_kwargs: Dict[str, Any],
+        use_cache: bool,
+    ) -> tuple[str, List[int], Optional[str]]:
+        cache_key = None
+        if use_cache:
+            try:
+                cache_key = orjson.dumps(
+                    (
+                        getattr(
+                            self.tokenizer_manager.tokenizer,
+                            "chat_template",
+                            None,
+                        ),
+                        messages,
+                        tools,
+                        template_kwargs,
+                        encode_kwargs,
+                    ),
+                    option=orjson.OPT_SORT_KEYS,
+                )
+            except TypeError:
+                pass
+
+        if cache_key is not None:
+            cached = self._chat_template_cache.get(cache_key)
+            if cached is not None:
+                self._chat_template_cache.move_to_end(cache_key)
+                rendered_prompt, prompt_ids, decoded_prompt = cached
+                return rendered_prompt, list(prompt_ids), decoded_prompt
+
+        rendered_prompt = self.tokenizer_manager.tokenizer.apply_chat_template(
+            messages,
+            tokenize=False,
+            add_generation_prompt=True,
+            tools=tools,
+            return_dict=False,
+            **template_kwargs,
+        )
+        prompt_ids = self.tokenizer_manager.tokenizer.encode(
+            rendered_prompt, **encode_kwargs
+        )
+        decoded_prompt = (
+            self.tokenizer_manager.tokenizer.decode(prompt_ids)
+            if cache_key is not None
+            else None
+        )
+
+        if cache_key is not None:
+            self._chat_template_cache[cache_key] = (
+                rendered_prompt,
+                tuple(prompt_ids),
+                decoded_prompt,
+            )
+            if len(self._chat_template_cache) > _CHAT_TEMPLATE_CACHE_MAX_SIZE:
+                self._chat_template_cache.popitem(last=False)
+
+        return rendered_prompt, prompt_ids, decoded_prompt
+
     def _apply_conversation_template(
         self,
         request: ChatCompletionRequest,

python/sglang/srt/managers/mm_schedule.py

@@ -8,6 +8,7 @@ import torch
 from sglang.srt.managers.schedule_batch import MultimodalDataItem
 from sglang.srt.mem_cache.multimodal_cache import EmbeddingResult, MultiModalStaticCache
 from sglang.srt.multimodal.evs import EVSEmbeddingResult
+from sglang.srt.multimodal.transport.cuda_ipc import BORROW_CUDA_IPC_FEATURE_KEY
 from sglang.srt.runtime_context import get_parallel, get_schedule
 from sglang.srt.utils import is_hip, is_npu, is_xpu
 from sglang.srt.utils.async_probe import maybe_assert_sum
@@ -368,6 +369,12 @@ def _batch_encode_per_image_misses(
                     )
                     unique_misses[cache_key] = (item, expected_token_count)
             elif cache_key not in unique_misses:
+                if (
+                    start >= chunk_start
+                    and end < chunk_end
+                    and item.can_defer_cuda_ipc_feature_reconstruction()
+                ):
+                    item.model_specific_data[BORROW_CUDA_IPC_FEATURE_KEY] = True
                 unique_misses[cache_key] = (item, expected_token_count)
 
     # Phase 1b: single ViT call for all unique cache misses

python/sglang/srt/managers/schedule_batch.py

@@ -126,7 +126,9 @@ from sglang.srt.model_executor.forward_batch_info import (
     ForwardMode,
 )
 from sglang.srt.multimodal.transport.cuda_ipc import (
+    CUDA_IPC_FEATURE_COPY_EVENT_KEY,
     DEFER_CUDA_IPC_FEATURE_RECONSTRUCTION_KEY,
+    RETAINED_CUDA_IPC_FEATURE_PROXY_KEY,
     CudaIpcTensorTransportProxy,
 )
 from sglang.srt.observability.metrics_collector import (
@@ -494,6 +496,8 @@ class MultimodalDataItem(msgspec.Struct, kw_only=True, dict=True, array_like=Tru
                 self.precomputed_embeddings.reconstruct_on_target_device(target_device)
             )
         for extra_key in self.model_specific_data:
+            if extra_key == RETAINED_CUDA_IPC_FEATURE_PROXY_KEY:
+                continue
             if isinstance(
                 self.model_specific_data[extra_key], CudaIpcTensorTransportProxy
             ):
@@ -533,7 +537,11 @@ class MultimodalDataItem(msgspec.Struct, kw_only=True, dict=True, array_like=Tru
 
     def release_transport_proxies(self, consumer_count: int = 1) -> None:
         """Best-effort release of proxies left by an abandoned request."""
-        values = [self.feature, self.precomputed_embeddings]
+        retained_proxy = self.model_specific_data.pop(
+            RETAINED_CUDA_IPC_FEATURE_PROXY_KEY, None
+        )
+        self.model_specific_data.pop(CUDA_IPC_FEATURE_COPY_EVENT_KEY, None)
+        values = [self.feature, self.precomputed_embeddings, retained_proxy]
         values.extend(self.model_specific_data.values())
         for value in values:
             if not isinstance(value, CudaIpcTensorTransportProxy):
@@ -682,10 +690,9 @@ class MultimodalInputs:
         """Release feature tensors to free GPU memory."""
         for item in self.mm_items:
             try:
-                # A request can be rejected before a deferred GPU feature is
+                # Release both deferred features that were never used and
-                # reconstructed. Acknowledge that transport lease before the
+                # borrowed features retained for possible re-prefill.
-                # proxy is dropped so the tokenizer pool can reuse its slice.
+                item.release_transport_proxies()
-                item.acknowledge_deferred_cuda_ipc_feature()
             except Exception:
                 logger.warning(
                     "Failed to release an unused multimodal feature transport",

python/sglang/srt/managers/scheduler.py

@@ -1283,7 +1283,7 @@ class Scheduler(
 
     def init_chunked_prefill(self):
         self.chunked_prefill_size = get_schedule().chunked_prefill_size
-        self.prefill_decode_interval = get_schedule().prefill_decode_interval
+        self.prefill_decode_interval = get_schedule().prefill_decode_interval or 0
         self._prefill_decode_interval_remaining = 0
         uses_transformers_backend = (
             get_resolved_model_impl(self.model_config) == ModelImpl.TRANSFORMERS

python/sglang/srt/managers/tokenizer_manager.py

@@ -976,6 +976,7 @@ class TokenizerManager(TokenizerControlMixin, TokenizerManagerScoreMixin):
         input_embeds = None
         input_text = obj.text
         token_type_ids = None
+        contains_mm_input = obj.contains_mm_input()
         is_cross_encoder_request = (
             isinstance(obj, EmbeddingReqInput) and obj.is_cross_encoder_request
         )
@@ -998,9 +999,21 @@ class TokenizerManager(TokenizerControlMixin, TokenizerManagerScoreMixin):
                     "the engine with skip_tokenizer_init=False."
                 )
 
+            # Avoid tokenizing a raw multimodal prompt when the processor will
+            # expand its placeholders and replace input_ids below.
+            if (
+                self.mm_processor
+                and contains_mm_input
+                and getattr(
+                    self.mm_processor,
+                    "generates_input_ids_from_raw_prompt",
+                    False,
+                )
+            ):
+                input_ids = None
             # For audio-only requests (e.g., Whisper), text may be empty.
             # The multimodal processor will provide input_ids later.
-            if not input_text and self.mm_processor and obj.contains_mm_input():
+            elif not input_text and self.mm_processor and contains_mm_input:
                 # Use empty placeholder - multimodal processor will override
                 input_ids = []
             else:
@@ -1008,7 +1021,6 @@ class TokenizerManager(TokenizerControlMixin, TokenizerManagerScoreMixin):
                     input_text, is_cross_encoder_request
                 )
 
-        contains_mm_input = obj.contains_mm_input()
         if contains_mm_input and get_disagg().language_model_only:
             raise ValueError(
                 "Multimodal inputs are not supported when --language-model-only "

python/sglang/srt/mem_cache/radix_cache.py

@@ -517,6 +517,31 @@ class RadixCache(BasePrefixCache):
             result = self.insert(
                 InsertParams(key=radix_key, value=values, priority=priority)
             )
+            # A request that was never cached while unfinished can add its
+            # whole prompt and generated output as one leaf. Split that leaf at
+            # the prompt boundary so LRU eviction can discard output KV without
+            # also losing the reusable prompt KV. Reinserting a prefix only
+            # changes radix topology; it reuses the indices inserted above.
+            prompt_key = RadixKey(
+                token_ids[: len(req.origin_input_ids)],
+                req.extra_key,
+                is_bigram=self.is_eagle,
+                cache_salt=req.cache_salt,
+            ).page_aligned(self.page_size)
+            if 0 < len(prompt_key) < key_len:
+                self.insert(
+                    InsertParams(
+                        key=prompt_key,
+                        value=values[: len(prompt_key)],
+                        priority=priority + 1,
+                        # Topology-only re-insert: this request created these
+                        # nodes moments ago, so counting it as a hit is the
+                        # same self-referencing inflation `chunked` exists to
+                        # suppress. hit_count drives eviction order, so an
+                        # extra bump would silently promote every prompt node.
+                        chunked=True,
+                    )
+                )
             freed_end = result.prefix_len
         else:
             freed_end = key_len

python/sglang/srt/mem_cache/unified_radix_cache.py

@@ -995,6 +995,44 @@ class UnifiedRadixCache(BasePrefixCache):
             insert_params.value = values
             result = self.insert(insert_params)
 
+            # Keep the prompt as an independent radix node. Finished requests
+            # append a short, request-specific output to a much longer prompt;
+            # without this split the prompt and output form one leaf and are
+            # evicted together. Re-inserting the prompt only changes topology:
+            # prev_prefix_len prevents the overlapping KV indices from being
+            # treated as duplicate allocations and freed. A declined rotation
+            # tail releases everything past the protected prefix below, so the
+            # split is skipped there rather than handing the tree rows that
+            # are about to be freed.
+            prompt_key = RadixKey(
+                req.origin_input_ids,
+                req.extra_key,
+                is_bigram=self.tree_core.is_eagle,
+                cache_salt=req.cache_salt,
+            ).page_aligned(self.page_size)
+            if (
+                not result.rotation_tail_declined
+                and len(self._components_tuple) == 1
+                and self._components_tuple[0].component_type == BASE_COMPONENT_TYPE
+                and 0 < len(prompt_key) < len(radix_key)
+            ):
+                self.insert(
+                    replace(
+                        insert_params,
+                        key=prompt_key,
+                        value=values[: len(prompt_key)],
+                        prev_prefix_len=len(prompt_key),
+                        priority=insert_params.priority + 1,
+                        # Topology-only re-insert: the request itself created
+                        # these nodes moments ago, so counting it as a hit is
+                        # the same self-referencing inflation `chunked` exists
+                        # to suppress. hit_count drives eviction order, so an
+                        # extra bump here would silently promote every prompt
+                        # node into the protected segment.
+                        chunked=True,
+                    )
+                )
+
             # Free unaligned tail (+ deferred truncation tail). A rotation
             # decline inserted nothing, so the whole span past the protected
             # prefix stayed request-owned and is released here instead.

python/sglang/srt/models/qwen3_vl.py

@@ -72,6 +72,12 @@ from sglang.srt.multimodal.mm_utils import (
     materialize_multimodal_features,
     run_dp_sharded_mrope_vision_model,
 )
+from sglang.srt.multimodal.transport.cuda_ipc import (
+    BORROW_CUDA_IPC_FEATURE_KEY,
+    CUDA_IPC_FEATURE_COPY_EVENT_KEY,
+    RETAINED_CUDA_IPC_FEATURE_PROXY_KEY,
+    CudaIpcTensorTransportProxy,
+)
 from sglang.srt.multimodal.vit_cuda_graph_runner import ViTCudaGraphRunner
 from sglang.srt.runtime_context import get_exec, get_mm, get_parallel
 from sglang.srt.utils import (
@@ -1435,13 +1441,23 @@ class Qwen3VLForConditionalGeneration(nn.Module):
                 pixel_values_device=self.visual.device,
                 pixel_values_dtype=self.visual.dtype,
             )
-        pixel_values = self._materialize_visual_items(items, range(len(items)))
+        pixel_values, borrowed_items, packed_ready = self._materialize_visual_items(
+            items, range(len(items)), preserve_for_reprefill=True
+        )
         assert pixel_values.dim() == 2, pixel_values.dim()
-        return self.visual(pixel_values, grid_thw=grid_thw)
+        visual_features = self.visual(pixel_values, grid_thw=grid_thw)
+        if borrowed_items:
+            self._offload_packed_visual_inputs(
+                pixel_values, borrowed_items, packed_ready
+            )
+        return visual_features
 
     def _materialize_visual_items(
-        self, items: List[MultimodalDataItem], indices: Iterable[int]
+        self,
-    ) -> torch.Tensor:
+        items: List[MultimodalDataItem],
+        indices: Iterable[int],
+        preserve_for_reprefill: bool = False,
+    ) -> Union[torch.Tensor, Tuple[torch.Tensor, list, Optional[torch.cuda.Event]]]:
         device = self.visual.device
         device_index = device.index
         if device.type == "cuda" and device_index is None:
@@ -1451,14 +1467,114 @@ class Qwen3VLForConditionalGeneration(nn.Module):
             consumer_count = max(parallel.tp_size, 1)
 
         features = []
+        borrowed_items = []
+        feature_offset = 0
         for index in indices:
             item = items[index]
             if device.type == "cuda":
-                item.reconstruct(device_index, ipc_consumer_count=consumer_count)
+                proxy = item.feature
+                model_specific_data = getattr(item, "model_specific_data", {})
+                pending_copy = model_specific_data.pop(
+                    CUDA_IPC_FEATURE_COPY_EVENT_KEY, None
+                )
+                if pending_copy is not None:
+                    torch.cuda.current_stream(device_index).wait_event(pending_copy)
+                borrow_requested = model_specific_data.pop(
+                    BORROW_CUDA_IPC_FEATURE_KEY, False
+                )
+                can_borrow = (
+                    preserve_for_reprefill
+                    and consumer_count == 1
+                    and isinstance(proxy, CudaIpcTensorTransportProxy)
+                    and borrow_requested
+                )
+                borrowed = (
+                    proxy.borrow_on_target_device(device_index) if can_borrow else None
+                )
+                if borrowed is not None:
+                    item.feature = borrowed
+                    model_specific_data[RETAINED_CUDA_IPC_FEATURE_PROXY_KEY] = proxy
+                    borrowed_items.append(
+                        (item, proxy, feature_offset, borrowed.shape[0])
+                    )
+                elif RETAINED_CUDA_IPC_FEATURE_PROXY_KEY not in model_specific_data:
+                    item.reconstruct(device_index, ipc_consumer_count=consumer_count)
             features.append(item.feature)
-        return materialize_multimodal_features(
+            feature_offset += item.feature.shape[0]
-            features, device=device, dtype=self.visual.dtype
+        try:
-        )
+            materialized = materialize_multimodal_features(
+                features, device=device, dtype=self.visual.dtype
+            )
+        except Exception:
+            for item, proxy, _, _ in borrowed_items:
+                try:
+                    proxy.release_borrowed_on_current_stream()
+                    item.model_specific_data.pop(
+                        RETAINED_CUDA_IPC_FEATURE_PROXY_KEY, None
+                    )
+                except Exception:
+                    logger.warning(
+                        "Failed to release a borrowed CUDA IPC feature after "
+                        "materialization failed",
+                        exc_info=True,
+                    )
+                item.feature = None
+            raise
+        packed_ready = None
+        if borrowed_items:
+            source_stream = torch.cuda.current_stream(device_index)
+            packed_ready = torch.cuda.Event()
+            packed_ready.record(source_stream)
+            for item, proxy, offset, length in borrowed_items:
+                item.feature = materialized.narrow(0, offset, length)
+                item.model_specific_data[CUDA_IPC_FEATURE_COPY_EVENT_KEY] = packed_ready
+                try:
+                    proxy.release_borrowed_on_current_stream()
+                    item.model_specific_data.pop(
+                        RETAINED_CUDA_IPC_FEATURE_PROXY_KEY, None
+                    )
+                except Exception:
+                    logger.warning(
+                        "Failed to release a copied CUDA IPC feature; retaining "
+                        "its lease until request cleanup",
+                        exc_info=True,
+                    )
+        if preserve_for_reprefill:
+            return materialized, borrowed_items, packed_ready
+        return materialized
+
+    def _offload_packed_visual_inputs(
+        self, materialized: torch.Tensor, borrowed_items: list, packed_ready
+    ) -> None:
+        """Preserve inputs for re-prefill without delaying the ViT launch."""
+        try:
+            host_features = torch.empty(
+                materialized.shape,
+                dtype=materialized.dtype,
+                device="cpu",
+                pin_memory=torch.cuda.is_available(),
+            )
+            copy_stream = getattr(self, "_mm_feature_copy_stream", None)
+            if copy_stream is None:
+                copy_stream = torch.cuda.Stream(device=self.visual.device)
+                self._mm_feature_copy_stream = copy_stream
+            with torch.cuda.stream(copy_stream):
+                copy_stream.wait_event(packed_ready)
+                host_features.copy_(materialized, non_blocking=True)
+                host_ready = torch.cuda.Event()
+                host_ready.record(copy_stream)
+            if materialized.is_cuda:
+                materialized.record_stream(copy_stream)
+            for item, _, offset, length in borrowed_items:
+                item.feature = host_features.narrow(0, offset, length)
+                item.model_specific_data[CUDA_IPC_FEATURE_COPY_EVENT_KEY] = host_ready
+        except Exception:
+            # The generic multimodal path will offload the owned CUDA slices.
+            logger.warning(
+                "Failed to preserve CUDA IPC features on the copy stream; "
+                "falling back to the generic offload path",
+                exc_info=True,
+            )
 
     def get_input_embeddings(self):
         return self.model.embed_tokens

python/sglang/srt/multimodal/media_artifacts/base.py

@@ -412,6 +412,66 @@ class MediaArtifactCacheMixin:
             raise RuntimeError("Artifact cache did not resolve every media item")
         return [artifact for artifact in artifacts if artifact is not None]
 
+    async def prepare_media_artifacts_without_cache(
+        self,
+        media_data: Sequence[Any],
+        *,
+        content_hashes: Optional[Sequence[Optional[str]]] = None,
+        modality: Optional[Modality] = None,
+    ) -> list[MediaArtifact]:
+        """Build request-local artifacts without cache lookup or retention."""
+        modality = self._resolve_artifact_modality(modality)
+        media_count = len(media_data)
+        if content_hashes is None:
+            content_hashes = [None] * media_count
+        if len(content_hashes) != media_count:
+            raise ValueError(
+                f"mm_content_hashes has {len(content_hashes)} entries for "
+                f"{media_count} {modality.name.lower()} items"
+            )
+        content_hashes = [parse_content_hash(value) for value in content_hashes]
+
+        snapshot_futures = [
+            self.io_executor.submit(self.snapshot_media_source, source, modality)
+            for source in media_data
+        ]
+        snapshots = []
+        for index, future in enumerate(snapshot_futures):
+            snapshot = await asyncio.wrap_future(future)
+            caller_hash = content_hashes[index]
+            if caller_hash is not None and caller_hash != snapshot.content_digest:
+                raise ValueError(
+                    f"content hash mismatch for media_data[{index}]: "
+                    f"expected {caller_hash}, got {snapshot.content_digest}"
+                )
+            snapshots.append(snapshot)
+
+        decode_futures = [
+            self.io_executor.submit(self.decode_media_snapshot, snapshot, modality)
+            for snapshot in snapshots
+        ]
+        entries = []
+        for source, snapshot, future in zip(media_data, snapshots, decode_futures):
+            entries.append(
+                MediaArtifactInput(
+                    content_digest=snapshot.content_digest,
+                    artifact_key=self._artifact_key(
+                        snapshot.content_digest, source, modality=modality
+                    ),
+                    modality=modality,
+                    media=await asyncio.wrap_future(future),
+                )
+            )
+
+        artifacts = await self._run_preprocess_and_build_artifact_batch(entries)
+        if len(artifacts) != len(entries):
+            raise ValueError(
+                "prepare_artifact_batch must return one artifact per media input"
+            )
+        for artifact, entry in zip(artifacts, entries):
+            self.validate_artifact(artifact, entry)
+        return artifacts
+
     async def _compute_cache_misses(
         self,
         misses_to_compute: Sequence[CacheMiss[str, MediaArtifact]],

python/sglang/srt/multimodal/processors/base_processor.py

@@ -227,6 +227,9 @@ class BaseMultimodalProcessor(ABC):
     # Models opt in by assigning a non-zero default. A user-provided server
     # argument overrides this value; zero disables storage and cache-key work.
     auto_mm_preprocess_cache_size_mb = 0
+    # Artifact-based processors may keep their prompt/M-RoPE fast path even
+    # when artifact retention is disabled.
+    uses_media_artifacts_without_cache = False
     # Processors opt out only when their preprocessing is not thread-safe. The
     # worker pool gives each thread its own `copy.deepcopy` of the HF processor
     # and injects it, and the single function it runs --
@@ -287,6 +290,7 @@ class BaseMultimodalProcessor(ABC):
         self.processor_fingerprint = (
             build_processor_fingerprint(self, hf_config)
             if self.mm_preprocess_cache.enabled
+            or self.uses_media_artifacts_without_cache
             else None
         )
         if self.mm_preprocess_cache.enabled:

python/sglang/srt/multimodal/processors/qwen_vl.py

@@ -2,13 +2,17 @@ import math
 import os
 import re
 import time
-from typing import List, Optional, Union
+from copy import deepcopy
+from dataclasses import dataclass, replace
+from functools import lru_cache
+from typing import Any, List, Optional, Union
 
 import numpy as np
 import torch
 import torchvision
 from PIL import Image
 from torchvision.transforms import InterpolationMode
+from transformers import BaseImageProcessor
 
 from sglang.srt.environ import envs
 from sglang.srt.layers.rotary_embedding import MRotaryEmbedding
@@ -33,6 +37,11 @@ from sglang.srt.models.qwen3_omni_moe import Qwen3OmniMoeForConditionalGeneratio
 from sglang.srt.models.qwen3_vl import Qwen3VLForConditionalGeneration
 from sglang.srt.models.qwen3_vl_moe import Qwen3VLMoeForConditionalGeneration
 from sglang.srt.models.qwen4_exp import Qwen4ExpForConditionalGeneration
+from sglang.srt.multimodal.cache import resolve_multimodal_item_hash
+from sglang.srt.multimodal.media_artifacts.base import (
+    MediaArtifactCacheMixin,
+    MediaArtifactInput,
+)
 from sglang.srt.multimodal.processors.base_processor import (
     BaseMultimodalProcessor as SGLangBaseProcessor,
 )
@@ -42,7 +51,7 @@ from sglang.srt.multimodal.processors.base_processor import (
 from sglang.srt.multimodal.transport.cuda_ipc import (
     DEFER_CUDA_IPC_FEATURE_RECONSTRUCTION_KEY,
 )
-from sglang.srt.runtime_context import get_mm
+from sglang.srt.runtime_context import get_mm, get_parallel
 from sglang.srt.utils import cpu_has_amx_support, is_cpu
 from sglang.srt.utils.video_decoder import VideoDecoderWrapper
 from sglang.utils import logger
@@ -68,6 +77,37 @@ FPS = 2.0
 FPS_MIN_FRAMES = 4
 FPS_MAX_FRAMES = 768
 
+
+@dataclass(frozen=True)
+class QwenVLImagePreprocessArtifact:
+    """Prompt-independent Qwen-VL processor output for one image."""
+
+    content_digest: str
+    artifact_key: str
+    feature_hash: int
+    feature: Optional[torch.Tensor]
+    model_specific_data: dict[str, Any]
+
+    @property
+    def has_feature(self) -> bool:
+        return self.feature is not None
+
+    def cache_value(self) -> "QwenVLImagePreprocessArtifact":
+        """Keep CPU processor outputs but never retain a CUDA tensor."""
+        if self.feature is None or self.feature.device.type == "cpu":
+            return self
+        return replace(self, feature=None)
+
+    def cache_size_items(self) -> tuple:
+        return (
+            self.content_digest,
+            self.artifact_key,
+            self.feature_hash,
+            self.feature,
+            self.model_specific_data,
+        )
+
+
 QWEN_VIDEO_PREPROCESS_CONFIG_KEYS = frozenset(
     {
         "fps",
@@ -289,8 +329,10 @@ async def preprocess_video(
 
 
 # Compatible with Qwen-VL & Qwen-Omni Series
-class QwenVLImageProcessor(SGLangBaseProcessor):
+class QwenVLImageProcessor(MediaArtifactCacheMixin, SGLangBaseProcessor):
     supports_transformers_backend = True
+    generates_input_ids_from_raw_prompt = True
+    artifact_modality = Modality.IMAGE
     models = [
         Qwen2VLForConditionalGeneration,
         Qwen2_5_VLForConditionalGeneration,
@@ -308,6 +350,10 @@ class QwenVLImageProcessor(SGLangBaseProcessor):
 
     def __init__(self, hf_config, server_args, _processor, *args, **kwargs):
         self.model_type = hf_config.model_type
+        self.uses_media_artifacts_without_cache = self.model_type in (
+            "qwen3_vl",
+            "qwen3_vl_moe",
+        )
         if self.model_type in (
             "qwen2_vl",
             "qwen2_5_vl",
@@ -732,7 +778,194 @@ class QwenVLImageProcessor(SGLangBaseProcessor):
             mrope_position_delta=mrope_position_delta,
         )
 
-    async def process_mm_data_async(
+    def prepare_artifact_batch(
+        self,
+        entries: list[MediaArtifactInput],
+        *,
+        processor=None,
+    ) -> list[QwenVLImagePreprocessArtifact]:
+        """Preprocess image cache misses without retaining prompt-specific state."""
+        if not entries:
+            return []
+
+        processor, _ = self._resolve_processor(processor)
+        image_kwargs = dict(self.image_config or {})
+        processor_device = None
+        if (
+            isinstance(processor.image_processor, BaseImageProcessor)
+            and not self.disable_fast_image_processor
+        ):
+            processor_device = self._fast_image_processor_device(processor)
+            if processor_device is not None:
+                image_kwargs["device"] = processor_device
+
+        with self._temporary_fast_processor_cuda_pool(processor_device):
+            result = processor.image_processor(
+                images=[entry.media for entry in entries],
+                return_tensors="pt",
+                **image_kwargs,
+            )
+            features = self._get_processor_output_value(result, "pixel_values")
+            image_grid_thw = self._get_processor_output_value(result, "image_grid_thw")
+            if (
+                isinstance(features, torch.Tensor)
+                and (
+                    self.mm_preprocess_cache.enabled
+                    or not self.keep_mm_features_on_device
+                )
+                and not self.precompute_hash_before_cpu_transfer
+            ):
+                features = features.cpu()
+
+        if not isinstance(features, torch.Tensor):
+            raise TypeError("Qwen-VL image processor must return pixel_values")
+        if not isinstance(image_grid_thw, torch.Tensor):
+            image_grid_thw = torch.as_tensor(image_grid_thw, dtype=torch.long)
+        if image_grid_thw.ndim != 2 or image_grid_thw.shape != (len(entries), 3):
+            raise ValueError(
+                "Qwen-VL image processor returned an invalid image_grid_thw shape: "
+                f"{tuple(image_grid_thw.shape)}"
+            )
+        feature_lengths = image_grid_thw.prod(dim=1).tolist()
+        if sum(feature_lengths) != features.shape[0]:
+            raise ValueError(
+                "Qwen-VL image processor feature count does not match image grids: "
+                f"{features.shape[0]} != {sum(feature_lengths)}"
+            )
+
+        artifacts = []
+        feature_offset = 0
+        for entry, grid, feature_length in zip(
+            entries, image_grid_thw, feature_lengths
+        ):
+            feature = features[
+                feature_offset : feature_offset + feature_length
+            ].contiguous()
+            feature_offset += feature_length
+            # The artifact key already commits to the media content, processor
+            # fingerprint, and every preprocessing kwarg. Derive the downstream
+            # cache identity from it so independently preprocessed copies in
+            # different tokenizer workers share the same radix/VLM cache key.
+            feature_hash = resolve_multimodal_item_hash(
+                existing_hash=0, namespace=entry.artifact_key
+            )
+            artifacts.append(
+                QwenVLImagePreprocessArtifact(
+                    content_digest=entry.content_digest,
+                    artifact_key=entry.artifact_key,
+                    feature_hash=feature_hash,
+                    feature=feature,
+                    model_specific_data={"image_grid_thw": grid.unsqueeze(0)},
+                )
+            )
+        return artifacts
+
+    def compose_image_artifacts(
+        self,
+        input_text,
+        artifacts: list[QwenVLImagePreprocessArtifact],
+    ) -> MultimodalProcessorOutput:
+        """Compose prompt tokens and request-owned items from cached images."""
+        image_grids = []
+        for artifact in artifacts:
+            grid = self._as_grid_batch(
+                artifact.model_specific_data.get("image_grid_thw")
+            )
+            if grid is None or grid.shape[0] != 1:
+                raise ValueError("Each Qwen-VL image artifact requires one image grid")
+            image_grids.append(grid)
+        image_grid_thw = torch.cat(image_grids, dim=0)
+
+        grid_key = tuple(
+            tuple(int(value) for value in row.tolist()) for row in image_grid_thw
+        )
+        if isinstance(input_text, str):
+            (
+                input_ids_tuple,
+                offsets,
+                mrope_positions,
+                mrope_position_delta,
+            ) = self._cached_image_prompt_template(input_text, grid_key)
+        else:
+            (
+                input_ids_tuple,
+                offsets,
+                mrope_positions,
+                mrope_position_delta,
+            ) = self._build_image_prompt_template(input_text, grid_key)
+        input_ids_list = list(input_ids_tuple)
+
+        mm_items = []
+        for artifact, offset in zip(artifacts, offsets):
+            item = MultimodalDataItem(
+                modality=Modality.IMAGE,
+                feature=artifact.feature,
+                offsets=[offset],
+                model_specific_data=deepcopy(artifact.model_specific_data),
+            )
+            item.set_hash(artifact.feature_hash)
+            mm_items.append(item)
+
+        padded_input_ids = MultimodalProcessorOutput.build_padded_input_ids(
+            input_ids_list, mm_items
+        )
+        mm_items = self._prepare_mm_items_for_transport(mm_items)
+        self._mark_cuda_ipc_features_for_deferred_reconstruction(mm_items)
+        return MultimodalProcessorOutput(
+            input_ids=input_ids_list,
+            padded_input_ids=padded_input_ids,
+            mm_items=mm_items,
+            im_start_id=self.vision_start_token_id,
+            im_end_id=self.vision_end_token_id,
+            im_token_id=self.mm_tokens.image_token_id,
+            video_token_id=self.mm_tokens.video_token_id,
+            audio_token_id=self.mm_tokens.audio_token_id,
+            mrope_positions=mrope_positions.clone(),
+            mrope_position_delta=mrope_position_delta.clone(),
+        )
+
+    @lru_cache(maxsize=256)
+    def _cached_image_prompt_template(self, input_text: str, grid_key: tuple):
+        """Cache prompt expansion and M-RoPE by prompt and image-grid shape."""
+        return self._build_image_prompt_template(input_text, grid_key)
+
+    def _build_image_prompt_template(self, input_text, grid_key: tuple):
+        image_grid_thw = torch.tensor(grid_key, dtype=torch.long)
+        input_ids_list, offsets, modalities = self.build_input_ids(
+            input_text, img_grid_thw=image_grid_thw
+        )
+        if modalities != [Modality.IMAGE] * len(grid_key):
+            raise ValueError("Qwen-VL image artifacts do not match prompt placeholders")
+
+        template_items = [
+            MultimodalDataItem(
+                modality=Modality.IMAGE,
+                feature=None,
+                offsets=[offset],
+                model_specific_data={"image_grid_thw": grid.unsqueeze(0)},
+            )
+            for grid, offset in zip(image_grid_thw, offsets)
+        ]
+        input_ids = torch.tensor(input_ids_list, dtype=torch.long)
+        mrope_result = self._compute_image_only_mrope_positions_from_offsets(
+            input_len=input_ids.numel(),
+            mm_items=template_items,
+            dtype=input_ids.dtype,
+            device=input_ids.device,
+        )
+        if mrope_result is None:
+            mrope_result = self.compute_mrope_positions(input_ids_list, template_items)
+        mrope_positions, mrope_position_delta = mrope_result
+        if mrope_positions is not None and mrope_positions.ndim == 3:
+            mrope_positions = mrope_positions.squeeze(1)
+        return (
+            tuple(input_ids_list),
+            tuple(offsets),
+            mrope_positions,
+            mrope_position_delta,
+        )
+
+    async def _process_mm_data_uncached(
         self,
         image_data: List[Union[str, bytes]],
         input_text,
@@ -788,7 +1021,7 @@ class QwenVLImageProcessor(SGLangBaseProcessor):
             base_output, self.mm_tokens, **processor_kwargs
         )
 
-        self._mark_dp_encoder_features_for_deferred_reconstruction(mm_items)
+        self._mark_cuda_ipc_features_for_deferred_reconstruction(mm_items)
 
         audio_feature_lengths = None
 
@@ -902,10 +1135,46 @@ class QwenVLImageProcessor(SGLangBaseProcessor):
             mrope_position_delta=mrope_position_delta,
         )
 
-    def _mark_dp_encoder_features_for_deferred_reconstruction(self, mm_items):
+    async def process_mm_data_async(
+        self,
+        image_data: List[Union[str, bytes]],
+        input_text,
+        request_obj,
+        *args,
+        **kwargs,
+    ):
+        if (
+            not image_data
+            or request_obj.video_data
+            or request_obj.audio_data
+            or any(self._is_preprocessed_input(item) for item in image_data)
+            or (
+                not self.mm_preprocess_cache.enabled
+                and not self.uses_media_artifacts_without_cache
+            )
+        ):
+            return await self._process_mm_data_uncached(
+                image_data, input_text, request_obj, *args, **kwargs
+            )
+
+        prepare_artifacts = (
+            self.prepare_media_artifacts
+            if self.mm_preprocess_cache.enabled
+            else self.prepare_media_artifacts_without_cache
+        )
+        artifacts = await prepare_artifacts(
+            image_data, content_hashes=getattr(request_obj, "mm_content_hashes", None)
+        )
+        return self.compose_image_artifacts(input_text, artifacts)
+
+    def _mark_cuda_ipc_features_for_deferred_reconstruction(self, mm_items):
+        supports_deferred_reconstruction = get_mm().mm_enable_dp_encoder or (
+            get_parallel().tp_size == 1
+            and self.model_type in ("qwen3_vl", "qwen3_vl_moe")
+        )
         if not (
             self.keep_mm_features_on_device
-            and get_mm().mm_enable_dp_encoder
+            and supports_deferred_reconstruction
             and self.model_type
             in ("qwen3_vl", "qwen3_vl_moe", "qwen3_5", "qwen3_5_moe")
         ):

python/sglang/srt/multimodal/transport/cuda_ipc.py

@@ -25,6 +25,9 @@ MM_ITEM_MEMORY_POOL_RECYCLE_INTERVAL = (
 DEFER_CUDA_IPC_FEATURE_RECONSTRUCTION_KEY = (
     "_sglang_defer_cuda_ipc_feature_reconstruction"
 )
+BORROW_CUDA_IPC_FEATURE_KEY = "_sglang_borrow_cuda_ipc_feature"
+CUDA_IPC_FEATURE_COPY_EVENT_KEY = "_sglang_cuda_ipc_feature_copy_event"
+RETAINED_CUDA_IPC_FEATURE_PROXY_KEY = "_sglang_retained_cuda_ipc_feature_proxy"
 
 
 def get_mm_feature_pool_size_per_worker(
@@ -233,6 +236,9 @@ class CudaIpcTensorTransportProxy(StreamOrderedPoolConsumerMixin):
         # Keep uncached mappings alive until the work enqueued on the consumer
         # stream has completed.
         self._pool_storage = None
+        self._borrowed_storage = None
+        self._borrowed_base_address = None
+        self._borrowed_device_id = None
 
     def _reconstruct_from_ipc_extra(
         self, ipc_extra, *, use_cache: bool, rebuild_device_idx: int
@@ -321,9 +327,53 @@ class CudaIpcTensorTransportProxy(StreamOrderedPoolConsumerMixin):
             )
         self._retain_storage_until_stream_completes(storage, device_id)
 
+    def borrow_on_target_device(
+        self, rebuild_device_idx: int
+    ) -> Optional[torch.Tensor]:
+        """Return a zero-copy view whose lease remains owned by this proxy."""
+        ipc_extra = self.proxy_state["ipc_extra"]
+        if not ipc_extra["use_pool_handle_cache"] or self._consumer_acknowledged:
+            return None
+
+        with torch.cuda.device(rebuild_device_idx):
+            slice_tensor, storage = self._open_pool_slice(rebuild_device_idx)
+            base_address = storage.data_ptr()
+            self._wait_until_ready(base_address, rebuild_device_idx)
+            borrowed = slice_tensor.view(ipc_extra["recons_dtype"]).reshape(
+                ipc_extra["recons_shape"]
+            )
+
+        self._borrowed_storage = storage
+        self._borrowed_base_address = base_address
+        self._borrowed_device_id = rebuild_device_idx
+        return borrowed
+
+    def release_borrowed_on_current_stream(
+        self, consumer_count: int = 1, consumer_rank: Optional[int] = None
+    ) -> None:
+        """Release a borrowed view after all current-stream reads are enqueued."""
+        storage = getattr(self, "_borrowed_storage", None)
+        if storage is None:
+            return
+        device_id = self._borrowed_device_id
+        with torch.cuda.device(device_id):
+            self._acknowledge_on_stream(
+                self._borrowed_base_address,
+                device_id,
+                consumer_count,
+                consumer_rank,
+            )
+        self._retain_storage_until_stream_completes(storage, device_id)
+        self._borrowed_storage = None
+        self._borrowed_base_address = None
+        self._borrowed_device_id = None
+
     def release_without_reconstruction(self, consumer_count: int = 1) -> None:
         """Release a pool slice when its request abandons this proxy."""
-        self.acknowledge_consumption(consumer_count)
+        if getattr(self, "_borrowed_storage", None) is not None:
+            self.release_borrowed_on_current_stream(consumer_count)
+        else:
+            self.acknowledge_consumption(consumer_count)
 
     def reconstruct_on_target_device(
         self,

python/sglang/srt/server_args.py

@@ -710,7 +710,15 @@ def prepare_server_args(argv: list[str]) -> ServerArgs:
         config_merger = ConfigArgumentMerger(parser)
         argv = config_merger.merge_config_with_args(argv)
 
+    radix_eviction_policy_explicitly_set = any(
+        arg == "--radix-eviction-policy" or arg.startswith("--radix-eviction-policy=")
+        for arg in argv
+    )
+
     raw_args = parser.parse_args(argv)
+    raw_args._radix_eviction_policy_explicitly_set = (
+        radix_eviction_policy_explicitly_set
+    )
 
     # Set up basic logging before ServerArgs.__post_init__ so that
     # logger.info / logger.warning calls there are properly formatted.

test/registered/chunked_prefill/test_mm_chunked_embedding_unit.py

@@ -17,6 +17,11 @@ import torch
 
 from sglang.srt.managers import mm_schedule
 from sglang.srt.managers.schedule_batch import Modality, MultimodalDataItem
+from sglang.srt.multimodal.transport.cuda_ipc import (
+    BORROW_CUDA_IPC_FEATURE_KEY,
+    DEFER_CUDA_IPC_FEATURE_RECONSTRUCTION_KEY,
+    CudaIpcTensorTransportProxy,
+)
 from sglang.srt.runtime_context import get_context, get_parallel
 from sglang.test.ci.ci_register import register_cpu_ci
 
@@ -235,6 +240,30 @@ def test_batched_mismatched_cache_entry_is_reencoded():
     encoder.assert_called_once()
 
 
+def test_full_deferred_ipc_item_is_marked_for_borrow():
+    mm_schedule.init_mm_embedding_cache(1 << 30)
+    proxy = object.__new__(CudaIpcTensorTransportProxy)
+    item = MultimodalDataItem(
+        modality=Modality.IMAGE,
+        hash=1000,
+        pad_value=1000,
+        feature=proxy,
+        offsets=[ITEM_OFFSETS[0]],
+        model_specific_data={DEFER_CUDA_IPC_FEATURE_RECONSTRUCTION_KEY: True},
+    )
+    request = mm_schedule.PerImageRequestInfo(
+        req_idx=0,
+        items=[item],
+        items_offset=[ITEM_OFFSETS[0]],
+        extend_prefix_len=0,
+        extend_seq_len=TOTAL_LEN,
+    )
+
+    mm_schedule._batch_encode_per_image_misses(_encoder_list, [request], _CPU)
+
+    assert item.model_specific_data[BORROW_CUDA_IPC_FEATURE_KEY]
+
+
 def test_batched_colliding_hashes_with_different_lengths_are_not_deduplicated():
     mm_schedule.init_mm_embedding_cache(1 << 30)
     items = _make_items()

test/registered/unit/entrypoints/openai/test_serving_chat.py

@@ -6,7 +6,7 @@ or
     python -m unittest discover -s tests -p "test_*unit.py" -v
 """
 
-from sglang.test.test_utils import enter_override, maybe_stub_sgl_kernel
+from sglang.test.test_utils import CustomTestCase, enter_override, maybe_stub_sgl_kernel
 
 maybe_stub_sgl_kernel()  # must precede any import that pulls in sgl_kernel
 
@@ -195,6 +195,81 @@ class _MockTemplateManager:
         self.jinja_template_may_reorder_tool_results = False
 
 
+class TestChatTemplateCache(CustomTestCase):
+    def setUp(self):
+        super().setUp()
+        reset_context()
+        self.addCleanup(reset_context)
+        publish(
+            ServerArgs(model_path="dummy", default_chat_template_kwargs=None),
+            role="tokenizer",
+        )
+        self.tokenizer_manager = _MockTokenizerManager()
+        self.chat = OpenAIServingChat(
+            self.tokenizer_manager,
+            _MockTemplateManager(),
+        )
+        self.tokenizer_manager.tokenizer.apply_chat_template.return_value = "rendered"
+        self.tokenizer_manager.tokenizer.encode.return_value = [11, 12]
+        self.tokenizer_manager.tokenizer.decode.return_value = "decoded"
+        self.tokenizer_manager.tokenizer.reset_mock()
+
+    def _render(self, **overrides):
+        kwargs = {
+            "messages": [{"role": "user", "content": "same text prefix"}],
+            "tools": None,
+            "template_kwargs": {"enable_thinking": False},
+            "encode_kwargs": {"add_special_tokens": False},
+            "use_cache": True,
+        }
+        kwargs.update(overrides)
+        return self.chat._render_and_encode_chat_template(**kwargs)
+
+    def test_cache_hit_reuses_render_encode_and_returns_an_owned_id_list(self):
+        first = self._render()
+        first[1].append(99)
+        second = self._render()
+
+        self.assertEqual(second, ("rendered", [11, 12], "decoded"))
+        self.tokenizer_manager.tokenizer.apply_chat_template.assert_called_once()
+        self.tokenizer_manager.tokenizer.encode.assert_called_once()
+        self.tokenizer_manager.tokenizer.decode.assert_called_once()
+
+    def test_cache_key_includes_template_and_encode_options(self):
+        self._render()
+        self._render(template_kwargs={"enable_thinking": True})
+        self._render(encode_kwargs={"add_special_tokens": True})
+
+        self.assertEqual(
+            self.tokenizer_manager.tokenizer.apply_chat_template.call_count,
+            3,
+        )
+        self.assertEqual(self.tokenizer_manager.tokenizer.encode.call_count, 3)
+
+    def test_cache_key_tracks_tokenizer_chat_template_updates(self):
+        self.tokenizer_manager.tokenizer.chat_template = "template-v1"
+        self._render()
+        self.tokenizer_manager.tokenizer.chat_template = "template-v2"
+        self._render()
+
+        self.assertEqual(
+            self.tokenizer_manager.tokenizer.apply_chat_template.call_count,
+            2,
+        )
+
+    def test_non_serializable_input_bypasses_cache(self):
+        messages = [{"role": "user", "content": object()}]
+        self._render(messages=messages)
+        self._render(messages=messages)
+
+        self.assertEqual(
+            self.tokenizer_manager.tokenizer.apply_chat_template.call_count,
+            2,
+        )
+        self.assertEqual(self.tokenizer_manager.tokenizer.encode.call_count, 2)
+        self.tokenizer_manager.tokenizer.decode.assert_not_called()
+
+
 class ServingChatTestCase(unittest.TestCase):
     # ------------- common fixtures -------------
     def setUp(self):
@@ -386,9 +461,7 @@ class ServingChatTestCase(unittest.TestCase):
         )
         self.tm.tokenizer.apply_chat_template.reset_mock()
         self.chat._apply_jinja_template(ordered_request, None, is_multimodal=True)
-        self.assertEqual(
+        self.tm.tokenizer.apply_chat_template.assert_not_called()
-            rendered_messages, self.tm.tokenizer.apply_chat_template.call_args[0][0]
-        )
 
         self.template_manager.jinja_template_may_reorder_tool_results = False
         self.tm.tokenizer.apply_chat_template.reset_mock()

test/registered/unit/mem_cache/test_radix_cache_unit.py

@@ -50,6 +50,7 @@ from sglang.srt.mem_cache.events import KVCacheEventRecorder
 from sglang.srt.mem_cache.mamba_radix_cache import TreeNode as MambaTreeNode
 from sglang.srt.mem_cache.radix_cache import RadixCache, RadixKey, TreeNode
 from sglang.srt.utils import get_device
+from sglang.test.test_utils import CustomTestCase
 
 # Test constants
 DEFAULT_PAGE_SIZE = 4
@@ -368,7 +369,7 @@ class TestTreeNode(unittest.TestCase):
                 self.assertEqual(n4.get_prefix_hash_values(n3), ["h1", "h2", "h3"])
 
 
-class TestRadixCache(unittest.TestCase):
+class TestRadixCache(CustomTestCase):
     """Test cases for RadixCache class."""
 
     def setUp(self):
@@ -540,6 +541,52 @@ class TestRadixCache(unittest.TestCase):
             cache.req_to_token_pool.req_to_token[0], tree_indices
         )
 
+    def test_finished_request_splits_prompt_from_output_for_eviction(self):
+        class ReqToTokenPool:
+            def __init__(self, row):
+                self.req_to_token = row.unsqueeze(0)
+
+        allocator = TokenToKVPoolAllocator(
+            size=16,
+            dtype=torch.float16,
+            device="cpu",
+            kvcache=None,
+            need_sort=False,
+        )
+        cache = RadixCache.create_simulated(mock_allocator=allocator)
+        prompt_ids = array("q", [1, 2, 3])
+        output_ids = array("q", [4, 5])
+        kv_indices = allocator.alloc(len(prompt_ids) + len(output_ids))
+        self.assertIsNotNone(kv_indices)
+        cache.req_to_token_pool = ReqToTokenPool(kv_indices)
+        req = unittest.mock.Mock(
+            origin_input_ids=prompt_ids,
+            output_ids=output_ids,
+            kv=ReqKvInfo(req_pool_idx=0, cache_protected_len=0),
+            extra_key=None,
+            cache_salt=None,
+            priority=0,
+            last_node=cache.root_node,
+        )
+
+        cache.cache_finished_req(
+            req,
+            is_insert=True,
+            kv_len_to_handle=len(prompt_ids) + len(output_ids),
+        )
+
+        (prompt_node,) = cache.root_node.children.values()
+        (output_node,) = prompt_node.children.values()
+        self.assertEqual(len(prompt_node.key), len(prompt_ids))
+        self.assertEqual(len(output_node.key), len(output_ids))
+
+        result = cache.evict(EvictParams(num_tokens=len(output_ids)))
+        self.assertEqual(result.num_tokens_evicted, len(output_ids))
+        match = cache.match_prefix(
+            MatchPrefixParams(key=RadixKey(prompt_ids + output_ids))
+        )
+        self.assertEqual(len(match.device_indices), len(prompt_ids))
+
     def test_kv_cache_events(self):
         """Test KV cache events functionality."""
         test_cases = [

test/registered/unit/mem_cache/test_unified_radix_cache_unittest.py

@@ -1558,6 +1558,28 @@ class UnifiedRadixCacheSuite:
             MatchPrefixParams(key=RadixKey(array("q", all_ids[:aligned_len])))
         )
         self.assertEqual(len(m.device_indices), aligned_len)
+
+        prompt_aligned_len = (len(input_ids) // ps) * ps
+        if self.cfg.components == (ComponentType.FULL,):
+            (prompt_node,) = _node_children(cache, cache.root_node_handle())
+            (output_node,) = _node_children(cache, prompt_node)
+            self.assertEqual(_node_key_length(cache, prompt_node), prompt_aligned_len)
+            self.assertEqual(
+                _node_key_length(cache, output_node),
+                aligned_len - prompt_aligned_len,
+            )
+
+            result = cache.evict(
+                EvictParams(num_tokens=aligned_len - prompt_aligned_len)
+            )
+            self.assertEqual(
+                result.num_tokens_evicted,
+                aligned_len - prompt_aligned_len,
+            )
+            prompt_only = cache.match_prefix(
+                MatchPrefixParams(key=RadixKey(array("q", all_ids[:aligned_len])))
+            )
+            self.assertEqual(len(prompt_only.device_indices), prompt_aligned_len)
         cache.sanity_check()
 
     def test_cache_finished_req_strips_thinking(self):

test/registered/unit/models/test_qwen3_vl_feature_materialization.py

@@ -1,16 +1,24 @@
 """Regression tests for Qwen3-VL multimodal feature materialization."""
 
 import unittest
+from contextlib import nullcontext
 from types import SimpleNamespace
 from unittest.mock import Mock, patch
 
 import torch
 
-from sglang.srt.managers.schedule_batch import Modality, MultimodalDataItem
+from sglang.srt.managers.schedule_batch import (
+    Modality,
+    MultimodalDataItem,
+    MultimodalInputs,
+)
 from sglang.srt.models.qwen3_vl import Qwen3VLForConditionalGeneration
 from sglang.srt.multimodal.processors.qwen_vl import QwenVLImageProcessor
 from sglang.srt.multimodal.transport.cuda_ipc import (
+    BORROW_CUDA_IPC_FEATURE_KEY,
+    CUDA_IPC_FEATURE_COPY_EVENT_KEY,
     DEFER_CUDA_IPC_FEATURE_RECONSTRUCTION_KEY,
+    CudaIpcTensorTransportProxy,
 )
 from sglang.srt.runtime_context import get_context
 from sglang.test.ci.ci_register import register_cpu_ci
@@ -69,7 +77,7 @@ class TestQwen3VLFeatureMaterialization(CustomTestCase):
                     MultimodalDataItem(modality=Modality.AUDIO),
                 ]
 
-                processor._mark_dp_encoder_features_for_deferred_reconstruction(items)
+                processor._mark_cuda_ipc_features_for_deferred_reconstruction(items)
 
                 self.assertTrue(
                     items[0].model_specific_data[
@@ -87,19 +95,104 @@ class TestQwen3VLFeatureMaterialization(CustomTestCase):
                 )
 
     def test_processor_does_not_defer_cpu_transport(self):
-        processor = QwenVLImageProcessor.__new__(QwenVLImageProcessor)
+        with get_context().override_server_args(mm_enable_dp_encoder=True):
-        processor.mm_feature_transport = "cpu"
+            processor = QwenVLImageProcessor.__new__(QwenVLImageProcessor)
-        processor.server_args = SimpleNamespace(mm_enable_dp_encoder=True)
+            processor.mm_feature_transport = "cpu"
-        processor.model_type = "qwen3_vl"
+            processor.model_type = "qwen3_vl"
-        item = MultimodalDataItem(modality=Modality.IMAGE)
+            item = MultimodalDataItem(modality=Modality.IMAGE)
 
-        processor._mark_dp_encoder_features_for_deferred_reconstruction([item])
+            processor._mark_cuda_ipc_features_for_deferred_reconstruction([item])
 
         self.assertNotIn(
             DEFER_CUDA_IPC_FEATURE_RECONSTRUCTION_KEY,
             item.model_specific_data,
         )
 
+    def test_processor_defers_cuda_ipc_for_single_tp_qwen3_vl(self):
+        processor = QwenVLImageProcessor.__new__(QwenVLImageProcessor)
+        processor.mm_feature_transport = "cuda_ipc"
+        processor.model_type = "qwen3_vl"
+        item = MultimodalDataItem(modality=Modality.IMAGE)
+
+        processor._mark_cuda_ipc_features_for_deferred_reconstruction([item])
+
+        self.assertTrue(
+            item.model_specific_data[DEFER_CUDA_IPC_FEATURE_RECONSTRUCTION_KEY]
+        )
+
+    def test_retract_reprefill_waits_for_preserved_visual_input(self):
+        visual = Mock()
+        visual.device = torch.device("cuda:0")
+        visual.dtype = torch.bfloat16
+        visual.side_effect = lambda pixel_values, *, grid_thw: pixel_values
+        model = self._model(visual, use_data_parallel=False)
+
+        proxy = CudaIpcTensorTransportProxy.__new__(CudaIpcTensorTransportProxy)
+        proxy.total_consumer_count = 1
+        borrowed_feature = torch.ones(2, 3)
+        packed_ready = Mock()
+        host_ready = Mock()
+        current_stream = Mock()
+        copy_stream = Mock()
+        proxy.reconstruct_on_target_device = Mock()
+        proxy.borrow_on_target_device = Mock(return_value=borrowed_feature)
+        proxy.release_borrowed_on_current_stream = Mock()
+        proxy.release_without_reconstruction = Mock()
+        item = MultimodalDataItem(
+            modality=Modality.IMAGE,
+            feature=proxy,
+            model_specific_data={BORROW_CUDA_IPC_FEATURE_KEY: True},
+        )
+        item.image_grid_thw = torch.tensor([[1, 1, 2]])
+
+        with (
+            patch(
+                "sglang.srt.models.qwen3_vl.get_parallel",
+                return_value=SimpleNamespace(tp_size=1),
+            ),
+            patch(
+                "sglang.srt.models.qwen3_vl.materialize_multimodal_features",
+                side_effect=lambda features, **_kwargs: torch.cat(features),
+            ),
+            patch(
+                "sglang.srt.models.qwen3_vl.torch.cuda.current_stream",
+                return_value=current_stream,
+            ),
+            patch(
+                "sglang.srt.models.qwen3_vl.torch.cuda.Event",
+                side_effect=(packed_ready, host_ready),
+            ),
+            patch(
+                "sglang.srt.models.qwen3_vl.torch.cuda.Stream",
+                return_value=copy_stream,
+            ),
+            patch(
+                "sglang.srt.models.qwen3_vl.torch.cuda.stream",
+                return_value=nullcontext(),
+            ),
+        ):
+            first = model.get_image_feature([item])
+            second = model.get_image_feature([item])
+
+        self.assertIsNot(item.feature, borrowed_feature)
+        self.assertTrue(torch.equal(item.feature, borrowed_feature))
+        self.assertTrue(torch.equal(first, second))
+        proxy.reconstruct_on_target_device.assert_not_called()
+        proxy.borrow_on_target_device.assert_called_once_with(0)
+        proxy.release_borrowed_on_current_stream.assert_called_once_with()
+        proxy.release_without_reconstruction.assert_not_called()
+        packed_ready.record.assert_called_once_with(current_stream)
+        copy_stream.wait_event.assert_called_once_with(packed_ready)
+        host_ready.record.assert_called_once_with(copy_stream)
+        current_stream.wait_event.assert_called_once_with(host_ready)
+        self.assertEqual(visual.call_count, 2)
+
+        MultimodalInputs(mm_items=[item]).release_features()
+
+        proxy.release_without_reconstruction.assert_not_called()
+        self.assertIsNone(item.feature)
+        self.assertNotIn(CUDA_IPC_FEATURE_COPY_EVENT_KEY, item.model_specific_data)
+
     def test_image_features_are_packed_on_the_visual_device(self):
         visual = _RecordingVisual()
         model = self._model(visual, use_data_parallel=False)

test/registered/unit/multimodal/rust/qwen/_fixtures.py

@@ -71,6 +71,7 @@ def make_processor(case, config, image_processor_cls=None):
         model_impl="sglang",
         keep_mm_feature_on_device=False,
         mm_feature_transport="cpu",
+        mm_enable_dp_encoder=False,
         image_processor_backend="auto",
         disable_fast_image_processor=True,
         skip_tokenizer_init=False,

test/registered/unit/multimodal/test_cuda_ipc_transport.py

@@ -128,6 +128,61 @@ class TestCudaIpcTransport(CustomTestCase):
                     producer.join(timeout=10)
             self.assertEqual(producer.exitcode, 0)
 
+    def test_borrowed_tensor_keeps_lease_until_explicit_release(self):
+        ctx = mp.get_context("spawn")
+        proxy_queue = ctx.Queue()
+        producer_results = ctx.Queue()
+        consumer_done = ctx.Event()
+        producer = ctx.Process(
+            target=_produce_pooled_tensor,
+            args=(proxy_queue, consumer_done, producer_results),
+        )
+        producer.start()
+        proxy = borrowed = consumed = None
+        producer_result = None
+        try:
+            try:
+                proxy, expected = proxy_queue.get(timeout=60)
+            except queue.Empty:
+                producer_result = producer_results.get(timeout=5)
+                _status, payload = producer_result
+                self.fail(
+                    f"CUDA IPC producer failed before sending its proxy: {payload}"
+                )
+
+            borrowed = proxy.borrow_on_target_device(0)
+            self.assertIsNotNone(borrowed)
+            consumed = borrowed + 1
+            torch.cuda.synchronize()
+
+            self.assertFalse(proxy._consumer_acknowledged)
+            proxy.release_without_reconstruction()
+            torch.cuda.synchronize()
+
+            self.assertEqual(
+                consumed.cpu().tolist(),
+                (torch.tensor(expected) + 1).tolist(),
+            )
+            self.assertTrue(proxy._consumer_acknowledged)
+            self.assertIsNone(proxy._borrowed_storage)
+        finally:
+            del consumed, borrowed, proxy
+            _pool_handle_cache_clear()
+            gc.collect()
+            torch.cuda.ipc_collect()
+            consumer_done.set()
+            producer.join(timeout=60)
+            try:
+                if producer_result is None:
+                    producer_result = producer_results.get(timeout=5)
+                status, payload = producer_result
+                self.assertEqual(status, "ok", payload)
+            finally:
+                if producer.is_alive():
+                    producer.terminate()
+                    producer.join(timeout=10)
+            self.assertEqual(producer.exitcode, 0)
+
     def test_failed_reconstruction_releases_pooled_tensor(self):
         ctx = mp.get_context("spawn")
         proxy_queue = ctx.Queue()

test/registered/unit/multimodal/test_media_artifact_processor.py

@@ -15,6 +15,7 @@ from sglang.srt.multimodal.media_artifacts import (
     MediaArtifactInput,
 )
 from sglang.test.ci.ci_register import register_cpu_ci
+from sglang.test.test_utils import CustomTestCase
 
 register_cpu_ci(est_time=11, suite="base-a-test-cpu")
 
@@ -88,7 +89,7 @@ class _Processor(MediaArtifactCacheMixin):
         self.io_executor.shutdown()
 
 
-class TestMediaArtifactProcessor(unittest.TestCase):
+class TestMediaArtifactProcessor(CustomTestCase):
     def test_default_image_decoder_rejects_lazy_pil_failure(self):
         malformed_png = base64.b64decode(
             "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVQIHWP4z8DwHwAFgAI/ScLJSwAAAABJRU5ErkJggg=="
@@ -253,6 +254,39 @@ class TestMediaArtifactProcessor(unittest.TestCase):
         self.assertEqual(artifacts[0].feature, b"fresh")
         self.assertEqual(len(processor.batches), 1)
 
+    def test_without_cache_builds_request_local_artifacts_and_does_not_retain(self):
+        processor = _Processor()
+        digest = snapshot_media(b"image").content_digest
+        key = processor._artifact_key(digest, b"image")
+        cached = _Artifact(digest, key, 1, b"cached")
+        processor.mm_preprocess_cache.put(key, cached)
+
+        try:
+            artifacts = asyncio.run(
+                processor.prepare_media_artifacts_without_cache([b"image", b"image"])
+            )
+            cached_after = asyncio.run(processor.prepare_media_artifacts([b"image"]))
+        finally:
+            processor.close()
+
+        self.assertEqual([artifact.feature for artifact in artifacts], [b"image"] * 2)
+        self.assertEqual(len(processor.batches), 1)
+        self.assertEqual(len(processor.batches[0]), 2)
+        self.assertIs(cached_after[0], cached)
+
+    def test_without_cache_still_validates_caller_content_hash(self):
+        processor = _Processor()
+        try:
+            with self.assertRaisesRegex(ValueError, "content hash mismatch"):
+                asyncio.run(
+                    processor.prepare_media_artifacts_without_cache(
+                        [b"image"],
+                        content_hashes=[snapshot_media(b"different").content_digest],
+                    )
+                )
+        finally:
+            processor.close()
+
 
 if __name__ == "__main__":
     unittest.main()

test/registered/unit/server_args/test_server_args.py

@@ -111,6 +111,25 @@ _mock_device.start()
 
 
 class TestPrepareServerArgs(CustomTestCase):
+    def test_radix_eviction_policy_explicitness_is_preserved(self):
+        omitted = prepare_server_args(["--model-path", "dummy"])
+        separated = prepare_server_args(
+            ["--model-path", "dummy", "--radix-eviction-policy", "lru"]
+        )
+        joined = prepare_server_args(
+            ["--model-path", "dummy", "--radix-eviction-policy=lru"]
+        )
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
+            f.write("model-path: dummy\nradix-eviction-policy: lru\n")
+            config_path = f.name
+        self.addCleanup(os.unlink, config_path)
+        configured = prepare_server_args(["--config", config_path])
+
+        self.assertFalse(omitted._radix_eviction_policy_explicitly_set)
+        self.assertTrue(separated._radix_eviction_policy_explicitly_set)
+        self.assertTrue(joined._radix_eviction_policy_explicitly_set)
+        self.assertTrue(configured._radix_eviction_policy_explicitly_set)
+
     def test_ple_embedding_offload_rejects_generic_weight_offload(self):
         for generic_offload in (
             {"cpu_offload_gb": 1},

test/registered/unit/test_model_overrides.py

@@ -20,6 +20,7 @@ from sglang.srt.arg_groups import overrides as overrides_module
 from sglang.srt.arg_groups.arg_utils import A, Arg, resolvable_fields
 from sglang.srt.arg_groups.model_overrides import minicpm as minicpm_module
 from sglang.srt.arg_groups.model_overrides import qwen3_5 as qwen3_5_module
+from sglang.srt.arg_groups.model_overrides import qwen3_vl as qwen3_vl_module
 from sglang.srt.arg_groups.overrides import (
     collect_model_override_declarations,
     register_model_override,
@@ -108,6 +109,10 @@ class TestModelOverridableWhitelist(CustomTestCase):
                     "enable_symm_mem",
                     "speculative_attention_mode",
                     "speculative_draft_attention_backend",
+                    "prefill_decode_interval",
+                    "radix_eviction_policy",
+                    "mm_preprocess_cache_size_mb",
+                    "mm_feature_transport",
                 }
             ),
         )
@@ -3107,5 +3112,107 @@ class TestDeclarationValidation(CustomTestCase):
             validate_declarations(args, [("src", {"nope": 1})])
 
 
+class TestQwen3VLHopperServingOverrides(CustomTestCase):
+    def setUp(self):
+        super().setUp()
+        self.addCleanup(envs.SGLANG_VLM_CACHE_SIZE_MB.clear)
+        self.addCleanup(envs.SGLANG_MM_FEATURE_CACHE_MB.clear)
+        envs.SGLANG_VLM_CACHE_SIZE_MB.clear()
+        envs.SGLANG_MM_FEATURE_CACHE_MB.clear()
+
+    @staticmethod
+    def _args(**overrides):
+        from sglang.srt.server_args import ServerArgs
+
+        values = {
+            "mm_preprocess_cache_size_mb": None,
+            "mm_feature_transport": None,
+            "max_running_requests": 400,
+            "radix_eviction_policy": "lru",
+            "prefill_decode_interval": None,
+            "attention_backend": None,
+            "decode_attention_backend": None,
+        }
+        values.update(overrides)
+        return ServerArgs(model_path="dummy", **values)
+
+    @patch.object(
+        qwen3_vl_module,
+        "large_hopper_qwen3_vl_model_type",
+        return_value="qwen3_vl",
+    )
+    def test_profiled_defaults_are_valid_model_overrides(self, _mock_model_type):
+        server_args = self._args()
+        updates = qwen3_vl_module._qwen3vl_hopper_serving_overrides(server_args, None)
+
+        self.assertEqual(
+            updates,
+            {
+                "mm_preprocess_cache_size_mb": 0,
+                "mm_feature_transport": "cuda_ipc",
+                "radix_eviction_policy": "priority",
+                "prefill_decode_interval": 22,
+                "decode_attention_backend": "flashinfer",
+            },
+        )
+        validate_declarations(
+            server_args,
+            [("_qwen3vl_hopper_serving_overrides", updates)],
+        )
+        self.assertEqual(envs.SGLANG_VLM_CACHE_SIZE_MB.get(), 0)
+        self.assertEqual(envs.SGLANG_MM_FEATURE_CACHE_MB.get(), 3 * 1024)
+
+    @patch.object(
+        qwen3_vl_module,
+        "large_hopper_qwen3_vl_model_type",
+        return_value="qwen3_vl",
+    )
+    def test_multinode_does_not_auto_select_cuda_ipc(self, _mock_model_type):
+        updates = qwen3_vl_module._qwen3vl_hopper_serving_overrides(
+            self._args(nnodes=2), None
+        )
+
+        self.assertNotIn("mm_feature_transport", updates)
+        self.assertFalse(envs.SGLANG_MM_FEATURE_CACHE_MB.is_set())
+
+    @patch.object(
+        qwen3_vl_module,
+        "large_hopper_qwen3_vl_model_type",
+        side_effect=AssertionError("must not load model config without GPU memory"),
+    )
+    def test_decode_graph_expansion_skips_unknown_gpu_memory(self, _mock_model_type):
+        decode_config = SimpleNamespace(max_bs=256)
+
+        qwen3_vl_module.expand_multimodal_decode_graph_to_running_limit(
+            self._args(), decode_config, gpu_mem=None
+        )
+
+        self.assertEqual(decode_config.max_bs, 256)
+
+    @patch.object(
+        qwen3_vl_module,
+        "large_hopper_qwen3_vl_model_type",
+        return_value="qwen3_vl",
+    )
+    def test_explicit_choices_are_not_replaced(self, _mock_model_type):
+        envs.SGLANG_VLM_CACHE_SIZE_MB.set(512)
+        envs.SGLANG_MM_FEATURE_CACHE_MB.set(2048)
+        updates = qwen3_vl_module._qwen3vl_hopper_serving_overrides(
+            self._args(
+                mm_preprocess_cache_size_mb=256,
+                mm_feature_transport="cpu",
+                radix_eviction_policy="lru",
+                _radix_eviction_policy_explicitly_set=True,
+                prefill_decode_interval=0,
+                decode_attention_backend="fa3",
+            ),
+            None,
+        )
+
+        self.assertEqual(updates, {})
+        self.assertEqual(envs.SGLANG_VLM_CACHE_SIZE_MB.get(), 512)
+        self.assertEqual(envs.SGLANG_MM_FEATURE_CACHE_MB.get(), 2048)
+
+
 if __name__ == "__main__":
     unittest.main()