[CI] Inline env vars lost by CONTINUE_ON_FAILURE subshell wrapping

Postmerge/nightly/daily builds set CONTINUE_ON_FAILURE=1, which wraps
each YAML command in `(cmd) || CI_OVERALL_STATUS=1`. A standalone
`export VAR=val` executes inside the subshell and the variable is
immediately lost — subsequent commands never see it.

This caused real failures (model_executor fastsafetensors crash from
missing VLLM_WORKER_MULTIPROC_METHOD=spawn) and silent coverage gaps
(rust_frontend tests silently running the Python frontend because
VLLM_USE_RUST_FRONTEND=1 was never set).

Fix: inline env vars as command prefixes (`VAR=val command`) instead
of standalone `export` lines across all 11 affected CI YAML files.

Co-authored-by: Claude <noreply@anthropic.com>
Signed-off-by: Tyler Michael Smith <tyler@tylermsmith.com>
Signed-off-by: Tyler Michael Smith <tlrmchlsmth@gmail.com>

.buildkite/hardware_tests/intel_xpu_ci/test-intel.yaml

@@ -21,6 +21,10 @@ steps:
     timeout_in_minutes: 30
     optional: true
     device: intel_gpu
+    agent_tags:
+      label: production
+      gpu: 2+
+      mem: 24+
     no_plugin: true
     env:
       REGISTRY: "public.ecr.aws/q9t5s3a7"
@@ -38,6 +42,10 @@ steps:
     timeout_in_minutes: 30
     optional: true
     device: intel_gpu
+    agent_tags:
+      label: production
+      gpu: 1+
+      mem: 16+
     no_plugin: true
     env:
       REGISTRY: "public.ecr.aws/q9t5s3a7"
@@ -55,6 +63,10 @@ steps:
     timeout_in_minutes: 30
     optional: true
     device: intel_gpu
+    agent_tags:
+      label: production
+      gpu: 1+
+      mem: 16+
     no_plugin: true
     env:
       REGISTRY: "public.ecr.aws/q9t5s3a7"

.buildkite/intel_jobs/basic_correctness.yaml

@@ -5,6 +5,10 @@ steps:
 - label: XPU Sleep Mode
   timeout_in_minutes: 30
   device: intel_gpu
+  agent_tags:
+    label: production
+    gpu: 1+
+    mem: 16+
   no_plugin: true
   working_dir: "."
   env:

.buildkite/intel_jobs/engine_intel.yaml

@@ -5,6 +5,10 @@ steps:
 - label: Engine (1 GPU)
   timeout_in_minutes: 30
   device: intel_gpu
+  agent_tags:
+    label: production
+    gpu: 1+
+    mem: 16+
   no_plugin: true
   working_dir: "."
   env:

.buildkite/intel_jobs/expert_parallelism_intel.yaml

@@ -6,6 +6,10 @@ steps:
   key: eplb-algorithm
   timeout_in_minutes: 45
   device: intel_gpu
+  agent_tags:
+    label: production
+    gpu: 1+
+    mem: 16+
   no_plugin: true
   working_dir: "."
   env:

.buildkite/intel_jobs/kernels_intel.yaml

@@ -5,6 +5,10 @@ steps:
 - label: vLLM IR Tests
   timeout_in_minutes: 30
   device: intel_gpu
+  agent_tags:
+    label: production
+    gpu: 1+
+    mem: 16+
   no_plugin: true
   working_dir: "."
   env:

.buildkite/intel_jobs/lora_intel.yaml

@@ -5,6 +5,10 @@ steps:
 - label: LoRA Runtime + Utils
   timeout_in_minutes: 45
   device: intel_gpu
+  agent_tags:
+    label: production
+    gpu: 1+
+    mem: 24+
   no_plugin: true
   working_dir: "."
   env:
@@ -34,6 +38,10 @@ steps:
 - label: LoRA Fused/MoE Kernels
   timeout_in_minutes: 45
   device: intel_gpu
+  agent_tags:
+    label: production
+    gpu: 1+
+    mem: 16+
   no_plugin: true
   working_dir: "."
   env:
@@ -54,6 +62,10 @@ steps:
 - label: LoRA Punica Kernels
   timeout_in_minutes: 45
   device: intel_gpu
+  agent_tags:
+    label: production
+    gpu: 1+
+    mem: 16+
   no_plugin: true
   working_dir: "."
   env:
@@ -74,6 +86,10 @@ steps:
 - label: LoRA Punica FP8/XPU Ops
   timeout_in_minutes: 45
   device: intel_gpu
+  agent_tags:
+    label: production
+    gpu: 1+
+    mem: 16+
   no_plugin: true
   working_dir: "."
   env:
@@ -94,6 +110,10 @@ steps:
 - label: LoRA Models
   timeout_in_minutes: 45
   device: intel_gpu
+  agent_tags:
+    label: production
+    gpu: 2+
+    mem: 24+
   no_plugin: true
   working_dir: "."
   env:
@@ -117,6 +137,10 @@ steps:
 - label: LoRA Multimodal
   timeout_in_minutes: 45
   device: intel_gpu
+  agent_tags:
+    label: production
+    gpu: 1+
+    mem: 16+
   no_plugin: true
   working_dir: "."
   env:

.buildkite/intel_jobs/misc_intel.yaml

@@ -5,6 +5,10 @@ steps:
 - label: V1 Core + KV + Metrics
   timeout_in_minutes: 30
   device: intel_gpu
+  agent_tags:
+    label: production
+    gpu: 1+
+    mem: 16+
   no_plugin: true
   working_dir: "."
   env:
@@ -31,6 +35,10 @@ steps:
 - label: V1 Sample + Logits
   timeout_in_minutes: 30
   device: intel_gpu
+  agent_tags:
+    label: production
+    gpu: 1+
+    mem: 16+
   no_plugin: true
   working_dir: "."
   env:
@@ -71,6 +79,10 @@ steps:
 - label: XPU CPU Offload
   timeout_in_minutes: 60
   device: intel_gpu
+  agent_tags:
+    label: production
+    gpu: 1+
+    mem: 16+
   no_plugin: true
   working_dir: "."
   env:
@@ -95,6 +107,10 @@ steps:
   key: regression
   timeout_in_minutes: 30
   device: intel_gpu
+  agent_tags:
+    label: production
+    gpu: 1+
+    mem: 16+
   no_plugin: true
   working_dir: "."
   env:
@@ -126,6 +142,10 @@ steps:
   timeout_in_minutes: 30
   num_devices: 2
   device: intel_gpu
+  agent_tags:
+    label: production
+    gpu: 2+
+    mem: 16+
   no_plugin: true
   working_dir: "."
   env:
@@ -157,6 +177,10 @@ steps:
   key: async-engine-inputs-utils-worker
   timeout_in_minutes: 30
   device: intel_gpu
+  agent_tags:
+    label: production
+    gpu: 1+
+    mem: 24+
   no_plugin: true
   working_dir: "."
   env:

.buildkite/intel_jobs/model_runner_v2_intel.yaml

@@ -5,6 +5,10 @@ steps:
 - label: Model Runner V2 Core Tests (Intel)
   timeout_in_minutes: 45
   device: intel_gpu
+  agent_tags:
+    label: production
+    gpu: 2+
+    mem: 16+
   no_plugin: true
   working_dir: "."
   env:
@@ -30,6 +34,10 @@ steps:
 - label: Model Runner V2 Examples (Intel)
   timeout_in_minutes: 45
   device: intel_gpu
+  agent_tags:
+    label: production
+    gpu: 1+
+    mem: 24+
   no_plugin: true
   working_dir: "."
   env:

.buildkite/intel_jobs/models_multimodal_intel.yaml

@@ -6,6 +6,10 @@ steps:
   key: multi-modal-models-standard-1-qwen2
   timeout_in_minutes: 45
   device: intel_gpu
+  agent_tags:
+    label: production
+    gpu: 1+
+    mem: 16+
   no_plugin: true
   working_dir: "."
   env:
@@ -27,6 +31,10 @@ steps:
   key: multi-modal-models-standard-2-qwen3-gemma
   timeout_in_minutes: 45
   device: intel_gpu
+  agent_tags:
+    label: production
+    gpu: 1+
+    mem: 16+
   no_plugin: true
   working_dir: "."
   env:
@@ -47,6 +55,10 @@ steps:
   key: multi-modal-models-standard-3-llava-qwen2-vl
   timeout_in_minutes: 45
   device: intel_gpu
+  agent_tags:
+    label: production
+    gpu: 1+
+    mem: 24+
   no_plugin: true
   working_dir: "."
   env:
@@ -68,6 +80,10 @@ steps:
   key: multi-modal-models-standard-4-other-whisper
   timeout_in_minutes: 45
   device: intel_gpu
+  agent_tags:
+    label: production
+    gpu: 1+
+    mem: 16+
   no_plugin: true
   working_dir: "."
   env:
@@ -88,6 +104,10 @@ steps:
   key: multi-modal-processor
   timeout_in_minutes: 45
   device: intel_gpu
+  agent_tags:
+    label: production
+    gpu: 1+
+    mem: 16+
   no_plugin: true
   working_dir: "."
   env:

.buildkite/intel_jobs/test-intel.yaml

@@ -19,6 +19,10 @@ steps:
       - image-build-xpu
     timeout_in_minutes: 30
     device: intel_gpu
+    agent_tags:
+      label: production
+      gpu: 2+
+      mem: 24+
     no_plugin: true
     env:
       REGISTRY: "public.ecr.aws/q9t5s3a7"
@@ -49,6 +53,10 @@ steps:
       - image-build-xpu
     timeout_in_minutes: 30
     device: intel_gpu
+    agent_tags:
+      label: production
+      gpu: 1+
+      mem: 16+
     no_plugin: true
     env:
       REGISTRY: "public.ecr.aws/q9t5s3a7"
@@ -74,6 +82,10 @@ steps:
       - image-build-xpu
     timeout_in_minutes: 30
     device: intel_gpu
+    agent_tags:
+      label: production
+      gpu: 1+
+      mem: 16+
     no_plugin: true
     env:
       REGISTRY: "public.ecr.aws/q9t5s3a7"
@@ -93,6 +105,10 @@ steps:
       - image-build-xpu
     timeout_in_minutes: 30
     device: intel_gpu
+    agent_tags:
+      label: production
+      gpu: 1+
+      mem: 16+
     no_plugin: true
     env:
       REGISTRY: "public.ecr.aws/q9t5s3a7"

.buildkite/scripts/hardware_ci/run-intel-ci-test.sh

@@ -34,7 +34,7 @@ case "${test_suite}" in
     pytest -v -s v1/worker --ignore=v1/worker/test_gpu_model_runner.py --ignore=v1/worker/test_worker_memory_snapshot.py
     pytest -v -s v1/structured_output
     pytest -v -s v1/test_serial_utils.py
-    pytest -v -s v1/spec_decode --ignore=v1/spec_decode/test_max_len.py --ignore=v1/spec_decode/test_speculators_eagle3.py --ignore=v1/spec_decode/test_acceptance_length.py
+    pytest -v -s v1/spec_decode --ignore=v1/spec_decode/test_max_len.py --ignore=v1/spec_decode/test_speculators_eagle3.py --ignore=v1/spec_decode/test_acceptance_length.py --ignore=v1/spec_decode/test_speculators_correctness.py
     pytest -v -s v1/kv_connector/unit --ignore=v1/kv_connector/unit/test_multi_connector.py --ignore=v1/kv_connector/unit/test_example_connector.py --ignore=v1/kv_connector/unit/test_lmcache_integration.py --ignore=v1/kv_connector/unit/test_hf3fs_client.py --ignore=v1/kv_connector/unit/test_hf3fs_connector.py --ignore=v1/kv_connector/unit/test_hf3fs_metadata_server.py --ignore=v1/kv_connector/unit/test_offloading_connector.py
     ;;
   server)

.buildkite/scripts/install-kv-connectors.sh

@@ -4,6 +4,11 @@
 
 set -euo pipefail
 
+if python3 -c "import torch; raise SystemExit(0 if torch.version.hip is not None else 1)"; then
+    uv pip install --system -r /vllm-workspace/requirements/kv_connectors_rocm.txt
+    exit 0
+fi
+
 REQUIREMENTS_FILE="${KV_CONNECTORS_REQUIREMENTS:-/vllm-workspace/requirements/kv_connectors.txt}"
 
 uv pip install --system -r "${REQUIREMENTS_FILE}"

.buildkite/test_areas/basic_correctness.yaml

@@ -12,7 +12,6 @@ steps:
   - tests/basic_correctness/test_cpu_offload
   - tests/basic_correctness/test_mem.py
   commands:
-  - export VLLM_WORKER_MULTIPROC_METHOD=spawn
-  - pytest -v -s basic_correctness/test_mem.py
-  - pytest -v -s basic_correctness/test_basic_correctness.py
-  - pytest -v -s basic_correctness/test_cpu_offload.py
+  - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s basic_correctness/test_mem.py
+  - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s basic_correctness/test_basic_correctness.py
+  - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s basic_correctness/test_cpu_offload.py

.buildkite/test_areas/compile.yaml

@@ -14,8 +14,7 @@ steps:
   - vllm/v1/cudagraph_dispatcher.py
   - tests/compile/correctness_e2e/test_sequence_parallel.py
   commands:
-  - export VLLM_TEST_CLEAN_GPU_MEMORY=1
-  - pytest -v -s tests/compile/correctness_e2e/test_sequence_parallel.py
+  - VLLM_TEST_CLEAN_GPU_MEMORY=1 pytest -v -s tests/compile/correctness_e2e/test_sequence_parallel.py
 
 - label: Sequence Parallel Correctness Tests (2xH100)
   key: sequence-parallel-correctness-tests-2xh100
@@ -25,8 +24,7 @@ steps:
   optional: true
   num_devices: 2
   commands:
-  - export VLLM_TEST_CLEAN_GPU_MEMORY=1
-  - pytest -v -s tests/compile/correctness_e2e/test_sequence_parallel.py
+  - VLLM_TEST_CLEAN_GPU_MEMORY=1 pytest -v -s tests/compile/correctness_e2e/test_sequence_parallel.py
 
 - label: AsyncTP Correctness Tests (2xH100)
   key: asynctp-correctness-tests-2xh100
@@ -36,8 +34,7 @@ steps:
   optional: true
   num_devices: 2
   commands:
-  - export VLLM_TEST_CLEAN_GPU_MEMORY=1
-  - pytest -v -s tests/compile/correctness_e2e/test_async_tp.py
+  - VLLM_TEST_CLEAN_GPU_MEMORY=1 pytest -v -s tests/compile/correctness_e2e/test_async_tp.py
 
 - label: AsyncTP Correctness Tests (B200)
   key: asynctp-correctness-tests-b200
@@ -47,8 +44,7 @@ steps:
   optional: true
   num_devices: 2
   commands:
-  - export VLLM_TEST_CLEAN_GPU_MEMORY=1
-  - pytest -v -s tests/compile/correctness_e2e/test_async_tp.py
+  - VLLM_TEST_CLEAN_GPU_MEMORY=1 pytest -v -s tests/compile/correctness_e2e/test_async_tp.py
 
 - label: Distributed Compile Unit Tests (2xH100)
   key: distributed-compile-unit-tests-2xh100
@@ -61,8 +57,7 @@ steps:
   - vllm/model_executor/layers
   - tests/compile/passes/distributed/
   commands:
-  - export VLLM_TEST_CLEAN_GPU_MEMORY=1
-  - pytest -s -v tests/compile/passes/distributed
+  - VLLM_TEST_CLEAN_GPU_MEMORY=1 pytest -s -v tests/compile/passes/distributed
 
 - label: Fusion and Compile Unit Tests (2xB200)
   key: fusion-and-compile-unit-tests-2xb200

.buildkite/test_areas/distributed.yaml

@@ -32,11 +32,10 @@ steps:
   - tests/entrypoints/openai/test_multi_api_servers.py
   commands:
   # https://github.com/NVIDIA/nccl/issues/1838
-  - export NCCL_CUMEM_HOST_ENABLE=0
-  - TP_SIZE=1 DP_SIZE=2 pytest -v -s v1/distributed/test_async_llm_dp.py
-  - TP_SIZE=1 DP_SIZE=2 pytest -v -s v1/distributed/test_eagle_dp.py
-  - TP_SIZE=1 DP_SIZE=2 pytest -v -s v1/distributed/test_external_lb_dp.py
-  - DP_SIZE=2 pytest -v -s entrypoints/openai/test_multi_api_servers.py
+  - NCCL_CUMEM_HOST_ENABLE=0 TP_SIZE=1 DP_SIZE=2 pytest -v -s v1/distributed/test_async_llm_dp.py
+  - NCCL_CUMEM_HOST_ENABLE=0 TP_SIZE=1 DP_SIZE=2 pytest -v -s v1/distributed/test_eagle_dp.py
+  - NCCL_CUMEM_HOST_ENABLE=0 TP_SIZE=1 DP_SIZE=2 pytest -v -s v1/distributed/test_external_lb_dp.py
+  - NCCL_CUMEM_HOST_ENABLE=0 DP_SIZE=2 pytest -v -s entrypoints/openai/test_multi_api_servers.py
 
 - label: Distributed Compile + RPC Tests (2 GPUs)
   key: distributed-compile-rpc-tests-2-gpus
@@ -56,10 +55,9 @@ steps:
   - tests/entrypoints/llm/test_collective_rpc.py
   commands:
   # https://github.com/NVIDIA/nccl/issues/1838
-  - export NCCL_CUMEM_HOST_ENABLE=0
-  - pytest -v -s entrypoints/llm/test_collective_rpc.py
-  - pytest -v -s ./compile/fullgraph/test_basic_correctness.py
-  - pytest -v -s ./compile/test_wrapper.py
+  - NCCL_CUMEM_HOST_ENABLE=0 pytest -v -s entrypoints/llm/test_collective_rpc.py
+  - NCCL_CUMEM_HOST_ENABLE=0 pytest -v -s ./compile/fullgraph/test_basic_correctness.py
+  - NCCL_CUMEM_HOST_ENABLE=0 pytest -v -s ./compile/test_wrapper.py
 
 - label: Distributed Torchrun + Shutdown Tests (2 GPUs)
   key: distributed-torchrun-shutdown-tests-2-gpus
@@ -78,11 +76,10 @@ steps:
   - tests/v1/worker/test_worker_memory_snapshot.py
   commands:
   # https://github.com/NVIDIA/nccl/issues/1838
-  - export NCCL_CUMEM_HOST_ENABLE=0
-  - VLLM_TEST_SAME_HOST=1 torchrun --nproc-per-node=4 distributed/test_same_node.py | grep 'Same node test passed'
-  - VLLM_TEST_SAME_HOST=1 VLLM_TEST_WITH_DEFAULT_DEVICE_SET=1 torchrun --nproc-per-node=4 distributed/test_same_node.py | grep 'Same node test passed'
-  - CUDA_VISIBLE_DEVICES=0,1 pytest -v -s v1/shutdown
-  - pytest -v -s v1/worker/test_worker_memory_snapshot.py
+  - NCCL_CUMEM_HOST_ENABLE=0 VLLM_TEST_SAME_HOST=1 torchrun --nproc-per-node=4 distributed/test_same_node.py | grep 'Same node test passed'
+  - NCCL_CUMEM_HOST_ENABLE=0 VLLM_TEST_SAME_HOST=1 VLLM_TEST_WITH_DEFAULT_DEVICE_SET=1 torchrun --nproc-per-node=4 distributed/test_same_node.py | grep 'Same node test passed'
+  - NCCL_CUMEM_HOST_ENABLE=0 CUDA_VISIBLE_DEVICES=0,1 pytest -v -s v1/shutdown
+  - NCCL_CUMEM_HOST_ENABLE=0 pytest -v -s v1/worker/test_worker_memory_snapshot.py
 
 - label: Distributed Torchrun + Examples (4 GPUs)
   key: distributed-torchrun-examples-4-gpus
@@ -97,24 +94,23 @@ steps:
   - tests/examples/features/data_parallel/data_parallel_offline.py
   commands:
   # https://github.com/NVIDIA/nccl/issues/1838
-  - export NCCL_CUMEM_HOST_ENABLE=0
   # test with torchrun tp=2 and external_dp=2
-  - torchrun --nproc-per-node=4 tests/distributed/test_torchrun_example.py
+  - NCCL_CUMEM_HOST_ENABLE=0 torchrun --nproc-per-node=4 tests/distributed/test_torchrun_example.py
   # test with torchrun tp=2 and pp=2
-  - PP_SIZE=2 torchrun --nproc-per-node=4 tests/distributed/test_torchrun_example.py
+  - NCCL_CUMEM_HOST_ENABLE=0 PP_SIZE=2 torchrun --nproc-per-node=4 tests/distributed/test_torchrun_example.py
   # test with torchrun tp=4 and dp=1
-  - TP_SIZE=4 torchrun --nproc-per-node=4 tests/distributed/test_torchrun_example_moe.py
+  - NCCL_CUMEM_HOST_ENABLE=0 TP_SIZE=4 torchrun --nproc-per-node=4 tests/distributed/test_torchrun_example_moe.py
   # test with torchrun tp=2, pp=2 and dp=1
-  - PP_SIZE=2 TP_SIZE=2 torchrun --nproc-per-node=4 tests/distributed/test_torchrun_example_moe.py
+  - NCCL_CUMEM_HOST_ENABLE=0 PP_SIZE=2 TP_SIZE=2 torchrun --nproc-per-node=4 tests/distributed/test_torchrun_example_moe.py
   # test with torchrun tp=1 and dp=4 with ep
-  - DP_SIZE=4 ENABLE_EP=1 torchrun --nproc-per-node=4 tests/distributed/test_torchrun_example_moe.py
+  - NCCL_CUMEM_HOST_ENABLE=0 DP_SIZE=4 ENABLE_EP=1 torchrun --nproc-per-node=4 tests/distributed/test_torchrun_example_moe.py
   # test with torchrun tp=2 and dp=2 with ep
-  - TP_SIZE=2 DP_SIZE=2 ENABLE_EP=1 torchrun --nproc-per-node=4 tests/distributed/test_torchrun_example_moe.py
+  - NCCL_CUMEM_HOST_ENABLE=0 TP_SIZE=2 DP_SIZE=2 ENABLE_EP=1 torchrun --nproc-per-node=4 tests/distributed/test_torchrun_example_moe.py
   # test with internal dp
-  - python3 examples/features/data_parallel/data_parallel_offline.py --enforce-eager
+  - NCCL_CUMEM_HOST_ENABLE=0 python3 examples/features/data_parallel/data_parallel_offline.py --enforce-eager
   # rlhf examples
-  - VLLM_ALLOW_INSECURE_SERIALIZATION=1 python3 examples/rl/rlhf_nccl.py
-  - VLLM_ALLOW_INSECURE_SERIALIZATION=1 python3 examples/rl/rlhf_ipc.py
+  - NCCL_CUMEM_HOST_ENABLE=0 VLLM_ALLOW_INSECURE_SERIALIZATION=1 python3 examples/rl/rlhf_nccl.py
+  - NCCL_CUMEM_HOST_ENABLE=0 VLLM_ALLOW_INSECURE_SERIALIZATION=1 python3 examples/rl/rlhf_ipc.py
 
 - label: Distributed DP Tests (4 GPUs)
   key: distributed-dp-tests-4-gpus
@@ -128,14 +124,13 @@ steps:
   - tests/distributed/test_utils
   commands:
   # https://github.com/NVIDIA/nccl/issues/1838
-  - export NCCL_CUMEM_HOST_ENABLE=0
-  - TP_SIZE=2 DP_SIZE=2 pytest -v -s v1/distributed/test_async_llm_dp.py
-  - TP_SIZE=2 DP_SIZE=2 pytest -v -s v1/distributed/test_eagle_dp.py
-  - TP_SIZE=2 DP_SIZE=2 pytest -v -s v1/distributed/test_external_lb_dp.py
-  - TP_SIZE=1 DP_SIZE=4 pytest -v -s v1/distributed/test_internal_lb_dp.py
-  - TP_SIZE=1 DP_SIZE=4 pytest -v -s v1/distributed/test_hybrid_lb_dp.py
-  - pytest -v -s v1/engine/test_engine_core_client.py::test_kv_cache_events_dp
-  - pytest -v -s distributed/test_utils.py
+  - NCCL_CUMEM_HOST_ENABLE=0 TP_SIZE=2 DP_SIZE=2 pytest -v -s v1/distributed/test_async_llm_dp.py
+  - NCCL_CUMEM_HOST_ENABLE=0 TP_SIZE=2 DP_SIZE=2 pytest -v -s v1/distributed/test_eagle_dp.py
+  - NCCL_CUMEM_HOST_ENABLE=0 TP_SIZE=2 DP_SIZE=2 pytest -v -s v1/distributed/test_external_lb_dp.py
+  - NCCL_CUMEM_HOST_ENABLE=0 TP_SIZE=1 DP_SIZE=4 pytest -v -s v1/distributed/test_internal_lb_dp.py
+  - NCCL_CUMEM_HOST_ENABLE=0 TP_SIZE=1 DP_SIZE=4 pytest -v -s v1/distributed/test_hybrid_lb_dp.py
+  - NCCL_CUMEM_HOST_ENABLE=0 pytest -v -s v1/engine/test_engine_core_client.py::test_kv_cache_events_dp
+  - NCCL_CUMEM_HOST_ENABLE=0 pytest -v -s distributed/test_utils.py
 
 - label: Distributed Compile + Comm (4 GPUs)
   key: distributed-compile-comm-4-gpus
@@ -151,13 +146,12 @@ steps:
   - tests/distributed/test_multiproc_executor.py
   commands:
   # https://github.com/NVIDIA/nccl/issues/1838
-  - export NCCL_CUMEM_HOST_ENABLE=0
-  - pytest -v -s compile/fullgraph/test_basic_correctness.py
-  - pytest -v -s distributed/test_pynccl.py
-  - pytest -v -s distributed/test_events.py
-  - pytest -v -s distributed/test_symm_mem_allreduce.py
+  - NCCL_CUMEM_HOST_ENABLE=0 pytest -v -s compile/fullgraph/test_basic_correctness.py
+  - NCCL_CUMEM_HOST_ENABLE=0 pytest -v -s distributed/test_pynccl.py
+  - NCCL_CUMEM_HOST_ENABLE=0 pytest -v -s distributed/test_events.py
+  - NCCL_CUMEM_HOST_ENABLE=0 pytest -v -s distributed/test_symm_mem_allreduce.py
   # test multi-node TP with multiproc executor (simulated on single node)
-  - pytest -v -s distributed/test_multiproc_executor.py::test_multiproc_executor_multi_node
+  - NCCL_CUMEM_HOST_ENABLE=0 pytest -v -s distributed/test_multiproc_executor.py::test_multiproc_executor_multi_node
 
 - label: Distributed Tests (8 GPUs)(H100)
   key: distributed-tests-8-gpus-h100
@@ -176,9 +170,8 @@ steps:
 
   commands:
   # https://github.com/NVIDIA/nccl/issues/1838
-  - export NCCL_CUMEM_HOST_ENABLE=0
   # test with torchrun tp=2 and dp=4 with ep
-  - torchrun --nproc-per-node=8 ../examples/features/torchrun/torchrun_dp_example_offline.py --tp-size=2 --pp-size=1 --dp-size=4 --enable-ep
+  - NCCL_CUMEM_HOST_ENABLE=0 torchrun --nproc-per-node=8 ../examples/features/torchrun/torchrun_dp_example_offline.py --tp-size=2 --pp-size=1 --dp-size=4 --enable-ep
 
 - label: Distributed Tests (4 GPUs)(A100)
   key: distributed-tests-4-gpus-a100
@@ -271,9 +264,7 @@ steps:
   - tests/distributed/test_pipeline_parallel.py
   - tests/basic_correctness/test_basic_correctness.py
   commands:
-  - export VLLM_USE_RAY_V2_EXECUTOR_BACKEND=1
-  - export NCCL_CUMEM_HOST_ENABLE=0
-  - pytest -v -s distributed/test_ray_v2_executor.py
-  - pytest -v -s distributed/test_ray_v2_executor_e2e.py
-  - pytest -v -s distributed/test_pipeline_parallel.py -k "ray"
-  - TARGET_TEST_SUITE=L4 pytest -v -s basic_correctness/test_basic_correctness.py -k "ray"
+  - VLLM_USE_RAY_V2_EXECUTOR_BACKEND=1 NCCL_CUMEM_HOST_ENABLE=0 pytest -v -s distributed/test_ray_v2_executor.py
+  - VLLM_USE_RAY_V2_EXECUTOR_BACKEND=1 NCCL_CUMEM_HOST_ENABLE=0 pytest -v -s distributed/test_ray_v2_executor_e2e.py
+  - VLLM_USE_RAY_V2_EXECUTOR_BACKEND=1 NCCL_CUMEM_HOST_ENABLE=0 pytest -v -s distributed/test_pipeline_parallel.py -k "ray"
+  - VLLM_USE_RAY_V2_EXECUTOR_BACKEND=1 NCCL_CUMEM_HOST_ENABLE=0 TARGET_TEST_SUITE=L4 pytest -v -s basic_correctness/test_basic_correctness.py -k "ray"

.buildkite/test_areas/entrypoints.yaml

@@ -22,10 +22,9 @@ steps:
   - vllm/
   - tests/entrypoints/llm
   commands:
-  - export VLLM_WORKER_MULTIPROC_METHOD=spawn
-  - pytest -v -s entrypoints/llm --ignore=entrypoints/llm/test_generate.py --ignore=entrypoints/llm/test_collective_rpc.py --ignore=entrypoints/llm/offline_mode
-  - pytest -v -s entrypoints/llm/test_generate.py # it needs a clean process
-  - pytest -v -s entrypoints/llm/offline_mode # Needs to avoid interference with other tests
+  - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s entrypoints/llm --ignore=entrypoints/llm/test_generate.py --ignore=entrypoints/llm/test_collective_rpc.py --ignore=entrypoints/llm/offline_mode
+  - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s entrypoints/llm/test_generate.py # it needs a clean process
+  - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s entrypoints/llm/offline_mode # Needs to avoid interference with other tests
   mirror:
     amd:
       device: mi325_1
@@ -41,9 +40,8 @@ steps:
   - vllm/
   - tests/entrypoints/serve
   commands:
-  - export VLLM_WORKER_MULTIPROC_METHOD=spawn
-  - pytest -v -s entrypoints/serve --ignore=entrypoints/serve/dev/rpc
-  - PYTHONPATH=/vllm-workspace pytest -v -s entrypoints/serve/dev/rpc
+  - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s entrypoints/serve --ignore=entrypoints/serve/dev/rpc
+  - VLLM_WORKER_MULTIPROC_METHOD=spawn PYTHONPATH=/vllm-workspace pytest -v -s entrypoints/serve/dev/rpc
   mirror:
     amd:
       device: mi325_1
@@ -59,8 +57,7 @@ steps:
   - tests/entrypoints/openai
   - tests/entrypoints/test_chat_utils
   commands:
-  - export VLLM_WORKER_MULTIPROC_METHOD=spawn
-  - pytest -v -s entrypoints/openai --ignore=entrypoints/openai/completion --ignore=entrypoints/openai/chat_completion --ignore=entrypoints/openai/responses --ignore=entrypoints/openai/correctness
+  - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s entrypoints/openai --ignore=entrypoints/openai/completion --ignore=entrypoints/openai/chat_completion --ignore=entrypoints/openai/responses --ignore=entrypoints/openai/correctness
   mirror:
     amd:
       device: mi325_1
@@ -77,9 +74,8 @@ steps:
   - tests/entrypoints/openai
   - tests/entrypoints/test_chat_utils
   commands:
-  - export VLLM_WORKER_MULTIPROC_METHOD=spawn
-  - pytest -v -s entrypoints/openai/chat_completion
-  - pytest -v -s entrypoints/openai/completion --ignore=entrypoints/openai/completion/test_tensorizer_entrypoint.py
+  - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s entrypoints/openai/chat_completion
+  - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s entrypoints/openai/completion --ignore=entrypoints/openai/completion/test_tensorizer_entrypoint.py
   mirror:
     amd:
       device: mi325_1
@@ -128,8 +124,7 @@ steps:
   - vllm/
   - tests/entrypoints/speech_to_text
   commands:
-  - export VLLM_WORKER_MULTIPROC_METHOD=spawn
-  - pytest -v -s entrypoints/speech_to_text
+  - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s entrypoints/speech_to_text
 
 - label: Entrypoints Integration (Multimodal)
   device: h200_35gb
@@ -140,8 +135,7 @@ steps:
   - vllm/
   - tests/entrypoints/multimodal
   commands:
-  - export VLLM_WORKER_MULTIPROC_METHOD=spawn
-  - pytest -v -s entrypoints/multimodal
+  - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s entrypoints/multimodal
 
 - label: Entrypoints Integration (Pooling)
   key: entrypoints-integration-pooling
@@ -151,8 +145,7 @@ steps:
   - vllm/
   - tests/entrypoints/pooling
   commands:
-  - export VLLM_WORKER_MULTIPROC_METHOD=spawn
-  - pytest -v -s entrypoints/pooling
+  - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s entrypoints/pooling
 
 - label: OpenAI API Correctness
   key: openai-api-correctness

.buildkite/test_areas/lm_eval.yaml

@@ -50,8 +50,7 @@ steps:
   - csrc/
   - vllm/model_executor/layers/quantization
   commands:
-    - export VLLM_USE_DEEP_GEMM=0  # We found Triton is faster than DeepGEMM for H100
-    - pytest -s -v test_lm_eval_correctness.py --config-list-file=configs/models-large-hopper.txt --tp-size=4
+    - VLLM_USE_DEEP_GEMM=0 pytest -s -v test_lm_eval_correctness.py --config-list-file=configs/models-large-hopper.txt --tp-size=4  # Triton is faster than DeepGEMM for H100
 
 - label: LM Eval Small Models (B200)
   key: lm-eval-small-models-b200
@@ -108,9 +107,7 @@ steps:
       depends_on:
       - image-build-amd
       commands:
-      - export VLLM_WORKER_MULTIPROC_METHOD=spawn
-      - export PYTORCH_ROCM_ARCH=gfx942 # Limit Quark compilation to save time
-      - pytest -s -v evals/gsm8k/test_gsm8k_correctness.py --config-list-file=configs/models-mi3xx.txt
+      - VLLM_WORKER_MULTIPROC_METHOD=spawn PYTORCH_ROCM_ARCH=gfx942 pytest -s -v evals/gsm8k/test_gsm8k_correctness.py --config-list-file=configs/models-mi3xx.txt  # Limit Quark compilation to save time
 
 - label: MoE Refactor Integration Test (H100 - TEMPORARY)
   key: moe-refactor-integration-test-h100-temporary

.buildkite/test_areas/lora.yaml

@@ -36,14 +36,14 @@ steps:
   commands:
     # FIXIT: find out which code initialize cuda before running the test
     # before the fix, we need to use spawn to test it
-    - export VLLM_WORKER_MULTIPROC_METHOD=spawn
+    #
     # Alot of these tests are on the edge of OOMing
-    - export PYTORCH_CUDA_ALLOC_CONF=expandable_segments:True
+    #
     # There is some Tensor Parallelism related processing logic in LoRA that
     # requires multi-GPU testing for validation.
-    - pytest -v -s -x lora/test_chatglm3_tp.py
-    - pytest -v -s -x lora/test_llama_tp.py
-    - pytest -v -s -x lora/test_qwen3_with_multi_loras.py
-    - pytest -v -s -x lora/test_olmoe_tp.py
-    - pytest -v -s -x lora/test_gptoss_tp.py
-    - pytest -v -s -x lora/test_qwen35_densemodel_lora.py
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn PYTORCH_CUDA_ALLOC_CONF=expandable_segments:True pytest -v -s -x lora/test_chatglm3_tp.py
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn PYTORCH_CUDA_ALLOC_CONF=expandable_segments:True pytest -v -s -x lora/test_llama_tp.py
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn PYTORCH_CUDA_ALLOC_CONF=expandable_segments:True pytest -v -s -x lora/test_qwen3_with_multi_loras.py
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn PYTORCH_CUDA_ALLOC_CONF=expandable_segments:True pytest -v -s -x lora/test_olmoe_tp.py
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn PYTORCH_CUDA_ALLOC_CONF=expandable_segments:True pytest -v -s -x lora/test_gptoss_tp.py
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn PYTORCH_CUDA_ALLOC_CONF=expandable_segments:True pytest -v -s -x lora/test_qwen35_densemodel_lora.py

.buildkite/test_areas/misc.yaml

@@ -18,9 +18,8 @@ steps:
     - vllm/v1/
     - tests/v1/spec_decode
   commands:
-    - export VLLM_WORKER_MULTIPROC_METHOD=spawn
     # TODO: create another `optional` test group for slow tests
-    - pytest -v -s -m 'not slow_test' v1/spec_decode
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s -m 'not slow_test' v1/spec_decode
   mirror:
     amd:
       device: mi300_1
@@ -50,12 +49,11 @@ steps:
     - tests/v1/test_request.py
     - tests/v1/test_outputs.py
   commands:
-    - export VLLM_WORKER_MULTIPROC_METHOD=spawn
-    - pytest -v -s v1/sample
-    - pytest -v -s v1/logits_processors
-    - pytest -v -s v1/test_oracle.py
-    - pytest -v -s v1/test_request.py
-    - pytest -v -s v1/test_outputs.py
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s v1/sample
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s v1/logits_processors
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s v1/test_oracle.py
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s v1/test_request.py
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s v1/test_outputs.py
   mirror:
     amd:
       device: mi325_1
@@ -93,18 +91,23 @@ steps:
     - tests/entrypoints/openai/correctness/test_lmeval.py
   commands:
     - bash /vllm-workspace/.buildkite/scripts/install-kv-connectors.sh
-    - export VLLM_WORKER_MULTIPROC_METHOD=spawn
     # split the test to avoid interference
-    - pytest -v -s -m 'not cpu_test' v1/core
-    - pytest -v -s v1/executor
-    - pytest -v -s v1/kv_offload
-    - pytest -v -s v1/simple_kv_offload
-    - pytest -v -s v1/worker
-    - pytest -v -s -m 'not cpu_test' v1/kv_connector/unit
-    - pytest -v -s -m 'not cpu_test' v1/metrics
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s -m 'not cpu_test' v1/core
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s v1/executor
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s v1/kv_offload
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s v1/simple_kv_offload
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s v1/worker
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s -m 'not cpu_test' v1/kv_connector/unit
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s -m 'not cpu_test' v1/metrics
     # Integration test for streaming correctness (requires special branch).
     - pip install -U git+https://github.com/robertgshaw2-redhat/lm-evaluation-harness.git@streaming-api
-    - pytest -v -s entrypoints/openai/correctness/test_lmeval.py::test_lm_eval_accuracy_v1_engine
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s entrypoints/openai/correctness/test_lmeval.py::test_lm_eval_accuracy_v1_engine
+  mirror:
+    amd:
+      device: mi325_1
+      timeout_in_minutes: 60
+      depends_on:
+      - image-build-amd
 
 - label: V1 Others (CPU)
   key: v1-others-cpu
@@ -147,8 +150,7 @@ steps:
     - vllm/distributed/kv_transfer/kv_connector/v1/example_hidden_states_connector.py
     - tests/v1/kv_connector/extract_hidden_states_integration
   commands:
-    - export VLLM_WORKER_MULTIPROC_METHOD=spawn
-    - pytest -v -s v1/kv_connector/extract_hidden_states_integration
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s v1/kv_connector/extract_hidden_states_integration
 
 - label: Extract Hidden States Integration (2 GPUs)
   key: extract-hidden-states-integration-2-gpus
@@ -161,8 +163,7 @@ steps:
     - vllm/distributed/kv_transfer/kv_connector/v1/example_hidden_states_connector.py
     - tests/v1/kv_connector/extract_hidden_states_integration
   commands:
-    - export VLLM_WORKER_MULTIPROC_METHOD=spawn
-    - pytest -v -s -m 'distributed' v1/kv_connector/extract_hidden_states_integration
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s -m 'distributed' v1/kv_connector/extract_hidden_states_integration
 
 - label: Regression
   key: regression
@@ -354,10 +355,9 @@ steps:
     - vllm/model_executor/layers
     - tests/v1/determinism/
   commands:
-    - export VLLM_WORKER_MULTIPROC_METHOD=spawn
     - pip install pytest-timeout pytest-forked
-    - pytest -v -s v1/determinism/test_batch_invariance.py
-    - VLLM_TEST_MODEL=deepseek-ai/DeepSeek-V2-Lite-Chat pytest -v -s v1/determinism/test_batch_invariance.py::test_v1_generation_is_deterministic_across_batch_sizes_with_needle[TRITON_MLA]
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s v1/determinism/test_batch_invariance.py
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn VLLM_TEST_MODEL=deepseek-ai/DeepSeek-V2-Lite-Chat pytest -v -s v1/determinism/test_batch_invariance.py::test_v1_generation_is_deterministic_across_batch_sizes_with_needle[TRITON_MLA]
 
 - label: Batch Invariance (H100)
   key: batch-invariance-h100
@@ -368,12 +368,11 @@ steps:
     - vllm/model_executor/layers
     - tests/v1/determinism/
   commands:
-    - export VLLM_WORKER_MULTIPROC_METHOD=spawn
     - pip install pytest-timeout pytest-forked
-    - pytest -v -s v1/determinism/test_batch_invariance.py
-    - pytest -v -s v1/determinism/test_rms_norm_batch_invariant.py
-    - VLLM_TEST_MODEL=deepseek-ai/DeepSeek-V2-Lite-Chat pytest -v -s v1/determinism/test_batch_invariance.py::test_v1_generation_is_deterministic_across_batch_sizes_with_needle[TRITON_MLA]
-    - VLLM_TEST_MODEL=Qwen/Qwen3-30B-A3B-Thinking-2507-FP8 pytest -v -s v1/determinism/test_batch_invariance.py::test_v1_generation_is_deterministic_across_batch_sizes_with_needle[FLASH_ATTN]
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s v1/determinism/test_batch_invariance.py
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s v1/determinism/test_rms_norm_batch_invariant.py
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn VLLM_TEST_MODEL=deepseek-ai/DeepSeek-V2-Lite-Chat pytest -v -s v1/determinism/test_batch_invariance.py::test_v1_generation_is_deterministic_across_batch_sizes_with_needle[TRITON_MLA]
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn VLLM_TEST_MODEL=Qwen/Qwen3-30B-A3B-Thinking-2507-FP8 pytest -v -s v1/determinism/test_batch_invariance.py::test_v1_generation_is_deterministic_across_batch_sizes_with_needle[FLASH_ATTN]
 
 - label: Batch Invariance (B200)
   key: batch-invariance-b200
@@ -384,14 +383,13 @@ steps:
     - vllm/model_executor/layers
     - tests/v1/determinism/
   commands:
-    - export VLLM_WORKER_MULTIPROC_METHOD=spawn
     - pip install pytest-timeout pytest-forked
-    - pytest -v -s v1/determinism/test_batch_invariance.py
-    - pytest -v -s v1/determinism/test_rms_norm_batch_invariant.py
-    - VLLM_TEST_MODEL=deepseek-ai/DeepSeek-V2-Lite-Chat pytest -v -s v1/determinism/test_batch_invariance.py::test_v1_generation_is_deterministic_across_batch_sizes_with_needle[TRITON_MLA]
-    - VLLM_TEST_MODEL=Qwen/Qwen3-30B-A3B-Thinking-2507-FP8 pytest -v -s v1/determinism/test_batch_invariance.py::test_v1_generation_is_deterministic_across_batch_sizes_with_needle[FLASH_ATTN]
-    - pytest -v -s v1/determinism/test_nvfp4_batch_invariant.py
-    - pytest -v -s v1/determinism/test_nvfp4_batch_invariant_scaled_mm.py
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s v1/determinism/test_batch_invariance.py
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s v1/determinism/test_rms_norm_batch_invariant.py
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn VLLM_TEST_MODEL=deepseek-ai/DeepSeek-V2-Lite-Chat pytest -v -s v1/determinism/test_batch_invariance.py::test_v1_generation_is_deterministic_across_batch_sizes_with_needle[TRITON_MLA]
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn VLLM_TEST_MODEL=Qwen/Qwen3-30B-A3B-Thinking-2507-FP8 pytest -v -s v1/determinism/test_batch_invariance.py::test_v1_generation_is_deterministic_across_batch_sizes_with_needle[FLASH_ATTN]
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s v1/determinism/test_nvfp4_batch_invariant.py
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s v1/determinism/test_nvfp4_batch_invariant_scaled_mm.py
   
 - label: Acceptance Length Test (Large Models) # optional
   device: h200_35gb
@@ -406,5 +404,4 @@ steps:
   - vllm/model_executor/models/mlp_speculator.py
   - tests/v1/spec_decode/test_acceptance_length.py
   commands:
-    - export VLLM_ALLOW_INSECURE_SERIALIZATION=1
-    - pytest -v -s v1/spec_decode/test_acceptance_length.py -m slow_test
+    - VLLM_ALLOW_INSECURE_SERIALIZATION=1 pytest -v -s v1/spec_decode/test_acceptance_length.py -m slow_test

.buildkite/test_areas/model_executor.yaml

@@ -13,13 +13,12 @@ steps:
   - tests/entrypoints/openai/completion/test_tensorizer_entrypoint.py
   commands:
     - apt-get update && apt-get install -y curl libsodium23
-    - export VLLM_WORKER_MULTIPROC_METHOD=spawn
-    # Dump tracebacks of all threads if a test hangs, so a wedged GPU/CUDA
-    # init surfaces a stack instead of silently stalling.
-    - export PYTHONFAULTHANDLER=1
     # Per-test watchdog: a single hung test (e.g. stuck during engine/CUDA
     # init) fails fast with a traceback instead of running until the global
     # build timeout. The `thread` method also handles hangs inside C/CUDA
     # calls that the signal method cannot interrupt.
-    - pytest -v -s model_executor -m '(not slow_test)' --timeout=900 --timeout-method=thread
-    - pytest -v -s entrypoints/openai/completion/test_tensorizer_entrypoint.py --timeout=900 --timeout-method=thread
+    #
+    # Env vars are inlined because CONTINUE_ON_FAILURE wraps each command
+    # in a subshell, so a standalone `export` would be lost.
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn PYTHONFAULTHANDLER=1 pytest -v -s model_executor -m '(not slow_test)' --timeout=900 --timeout-method=thread
+    - VLLM_WORKER_MULTIPROC_METHOD=spawn PYTHONFAULTHANDLER=1 pytest -v -s entrypoints/openai/completion/test_tensorizer_entrypoint.py --timeout=900 --timeout-method=thread

.buildkite/test_areas/model_runner_v2.yaml

@@ -16,15 +16,14 @@ steps:
   - tests/entrypoints/llm/test_struct_output_generate.py
   commands:
   - set -x
-  - export VLLM_USE_V2_MODEL_RUNNER=1
-  - pytest -v -s v1/engine/test_llm_engine.py -k "not test_engine_metrics"
+  - VLLM_USE_V2_MODEL_RUNNER=1 pytest -v -s v1/engine/test_llm_engine.py -k "not test_engine_metrics"
   # This requires eager until we sort out CG correctness issues.
   # TODO: remove ENFORCE_EAGER here after https://github.com/vllm-project/vllm/pull/32936 is merged.
-  - ENFORCE_EAGER=1 pytest -v -s v1/e2e/general/test_async_scheduling.py -k "not ngram"
-  - pytest -v -s v1/e2e/general/test_context_length.py
-  - pytest -v -s v1/e2e/general/test_min_tokens.py
+  - VLLM_USE_V2_MODEL_RUNNER=1 ENFORCE_EAGER=1 pytest -v -s v1/e2e/general/test_async_scheduling.py -k "not ngram"
+  - VLLM_USE_V2_MODEL_RUNNER=1 pytest -v -s v1/e2e/general/test_context_length.py
+  - VLLM_USE_V2_MODEL_RUNNER=1 pytest -v -s v1/e2e/general/test_min_tokens.py
   # Temporary hack filter to exclude ngram spec decoding based tests.
-  - pytest -v -s entrypoints/llm/test_struct_output_generate.py -k "xgrammar and not speculative_config6 and not speculative_config7 and not speculative_config8 and not speculative_config0"
+  - VLLM_USE_V2_MODEL_RUNNER=1 pytest -v -s entrypoints/llm/test_struct_output_generate.py -k "xgrammar and not speculative_config6 and not speculative_config7 and not speculative_config8 and not speculative_config0"
 
 - label: Model Runner V2 Examples
   device: h200_35gb
@@ -42,26 +41,25 @@ steps:
     - examples/features/tensorize_vllm_model.py
   commands:
     - set -x
-    - export VLLM_USE_V2_MODEL_RUNNER=1
     - pip install tensorizer # for tensorizer test
-    - python3 basic/offline_inference/chat.py # for basic
-    - python3 basic/offline_inference/generate.py --model facebook/opt-125m
-    #- python3 basic/offline_inference/generate.py --model meta-llama/Llama-2-13b-chat-hf --cpu-offload-gb 10  # TODO
-    #- python3 basic/offline_inference/embed.py   # TODO
+    - VLLM_USE_V2_MODEL_RUNNER=1 python3 basic/offline_inference/chat.py # for basic
+    - VLLM_USE_V2_MODEL_RUNNER=1 python3 basic/offline_inference/generate.py --model facebook/opt-125m
+    #- VLLM_USE_V2_MODEL_RUNNER=1 python3 basic/offline_inference/generate.py --model meta-llama/Llama-2-13b-chat-hf --cpu-offload-gb 10  # TODO
+    #- VLLM_USE_V2_MODEL_RUNNER=1 python3 basic/offline_inference/embed.py   # TODO
     # for multi-modal models
-    - python3 generate/multimodal/audio_language_offline.py --seed 0
-    - python3 generate/multimodal/vision_language_offline.py --seed 0
-    - python3 generate/multimodal/vision_language_multi_image_offline.py --seed 0
-    - python3 generate/multimodal/encoder_decoder_multimodal_offline.py --model-type whisper --seed 0
+    - VLLM_USE_V2_MODEL_RUNNER=1 python3 generate/multimodal/audio_language_offline.py --seed 0
+    - VLLM_USE_V2_MODEL_RUNNER=1 python3 generate/multimodal/vision_language_offline.py --seed 0
+    - VLLM_USE_V2_MODEL_RUNNER=1 python3 generate/multimodal/vision_language_multi_image_offline.py --seed 0
+    - VLLM_USE_V2_MODEL_RUNNER=1 python3 generate/multimodal/encoder_decoder_multimodal_offline.py --model-type whisper --seed 0
     # for pooling models
-    - python3 pooling/embed/vision_embedding_offline.py --seed 0
+    - VLLM_USE_V2_MODEL_RUNNER=1 python3 pooling/embed/vision_embedding_offline.py --seed 0
     # for features demo
-    - python3 features/automatic_prefix_caching/prefix_caching_offline.py
-    - python3 deployment/llm_engine_example.py
-    - python3 features/tensorize_vllm_model.py --model facebook/opt-125m serialize --serialized-directory /tmp/ --suffix v1 && python3 features/tensorize_vllm_model.py --model facebook/opt-125m deserialize --path-to-tensors /tmp/vllm/facebook/opt-125m/v1/model.tensors
-    - python3 features/speculative_decoding/spec_decode_offline.py --test --method eagle --num_spec_tokens 3 --dataset-name hf --dataset-path philschmid/mt-bench --num-prompts 80 --temp 0 --top-p 1.0 --top-k -1 --tp 1 --enable-chunked-prefill --max-model-len 2048
+    - VLLM_USE_V2_MODEL_RUNNER=1 python3 features/automatic_prefix_caching/prefix_caching_offline.py
+    - VLLM_USE_V2_MODEL_RUNNER=1 python3 deployment/llm_engine_example.py
+    - VLLM_USE_V2_MODEL_RUNNER=1 python3 features/tensorize_vllm_model.py --model facebook/opt-125m serialize --serialized-directory /tmp/ --suffix v1 && VLLM_USE_V2_MODEL_RUNNER=1 python3 features/tensorize_vllm_model.py --model facebook/opt-125m deserialize --path-to-tensors /tmp/vllm/facebook/opt-125m/v1/model.tensors
+    - VLLM_USE_V2_MODEL_RUNNER=1 python3 features/speculative_decoding/spec_decode_offline.py --test --method eagle --num_spec_tokens 3 --dataset-name hf --dataset-path philschmid/mt-bench --num-prompts 80 --temp 0 --top-p 1.0 --top-k -1 --tp 1 --enable-chunked-prefill --max-model-len 2048
     # https://github.com/vllm-project/vllm/pull/26682 uses slightly more memory in PyTorch 2.9+ causing this test to OOM in 1xL4 GPU
-    - python3 features/speculative_decoding/spec_decode_offline.py --test --method eagle3 --num_spec_tokens 3 --dataset-name hf --dataset-path philschmid/mt-bench --num-prompts 80 --temp 0 --top-p 1.0 --top-k -1 --tp 1 --enable-chunked-prefill --max-model-len 1536
+    - VLLM_USE_V2_MODEL_RUNNER=1 python3 features/speculative_decoding/spec_decode_offline.py --test --method eagle3 --num_spec_tokens 3 --dataset-name hf --dataset-path philschmid/mt-bench --num-prompts 80 --temp 0 --top-p 1.0 --top-k -1 --tp 1 --enable-chunked-prefill --max-model-len 1536
 
 - label: Model Runner V2 Distributed (2 GPUs)
   key: model-runner-v2-distributed-2-gpus
@@ -76,13 +74,11 @@ steps:
     - tests/v1/distributed/test_eagle_dp.py
   commands:
     - set -x
-    - export VLLM_USE_V2_MODEL_RUNNER=1
     # The "and not True" here is a hacky way to exclude the prompt_embeds cases which aren't yet supported.
-    - TARGET_TEST_SUITE=L4 pytest -v -s basic_correctness/test_basic_correctness.py -m 'distributed(num_gpus=2)' -k "not ray and not True"
+    - VLLM_USE_V2_MODEL_RUNNER=1 TARGET_TEST_SUITE=L4 pytest -v -s basic_correctness/test_basic_correctness.py -m 'distributed(num_gpus=2)' -k "not ray and not True"
     # https://github.com/NVIDIA/nccl/issues/1838
-    - export NCCL_CUMEM_HOST_ENABLE=0
-    - TP_SIZE=1 DP_SIZE=2 pytest -v -s v1/distributed/test_async_llm_dp.py -k "not ray"
-    - TP_SIZE=1 DP_SIZE=2 pytest -v -s v1/distributed/test_eagle_dp.py
+    - VLLM_USE_V2_MODEL_RUNNER=1 NCCL_CUMEM_HOST_ENABLE=0 TP_SIZE=1 DP_SIZE=2 pytest -v -s v1/distributed/test_async_llm_dp.py -k "not ray"
+    - VLLM_USE_V2_MODEL_RUNNER=1 NCCL_CUMEM_HOST_ENABLE=0 TP_SIZE=1 DP_SIZE=2 pytest -v -s v1/distributed/test_eagle_dp.py
 
 - label: Model Runner V2 Pipeline Parallelism (4 GPUs)
   key: model-runner-v2-pipeline-parallelism-4-gpus
@@ -97,10 +93,9 @@ steps:
     - tests/v1/distributed/test_pp_dp_v2.py
   commands:
     - set -x
-    - export VLLM_USE_V2_MODEL_RUNNER=1
-    - pytest -v -s distributed/test_pipeline_parallel.py -k "not ray and not Jamba"
-    - pytest -v -s distributed/test_pp_cudagraph.py -k "not ray"
-    - pytest -v -s v1/distributed/test_pp_dp_v2.py
+    - VLLM_USE_V2_MODEL_RUNNER=1 pytest -v -s distributed/test_pipeline_parallel.py -k "not ray and not Jamba"
+    - VLLM_USE_V2_MODEL_RUNNER=1 pytest -v -s distributed/test_pp_cudagraph.py -k "not ray"
+    - VLLM_USE_V2_MODEL_RUNNER=1 pytest -v -s v1/distributed/test_pp_dp_v2.py
 
 - label: Model Runner V2 Spec Decode
   device: h200_35gb
@@ -115,8 +110,7 @@ steps:
   - tests/v1/e2e/spec_decode/test_spec_decode.py
   commands:
   - set -x
-  - export VLLM_USE_V2_MODEL_RUNNER=1
-  - pytest -v -s v1/spec_decode/test_max_len.py -k "eagle or mtp"
-  - pytest -v -s v1/spec_decode/test_rejection_sampler_utils.py
-  - pytest -v -s v1/spec_decode/test_synthetic_rejection_sampler_utils.py
-  - pytest -v -s v1/e2e/spec_decode/test_spec_decode.py -k "eagle or mtp"
+  - VLLM_USE_V2_MODEL_RUNNER=1 pytest -v -s v1/spec_decode/test_max_len.py -k "eagle or mtp"
+  - VLLM_USE_V2_MODEL_RUNNER=1 pytest -v -s v1/spec_decode/test_rejection_sampler_utils.py
+  - VLLM_USE_V2_MODEL_RUNNER=1 pytest -v -s v1/spec_decode/test_synthetic_rejection_sampler_utils.py
+  - VLLM_USE_V2_MODEL_RUNNER=1 pytest -v -s v1/e2e/spec_decode/test_spec_decode.py -k "eagle or mtp"

.buildkite/test_areas/rust_frontend.yaml

@@ -23,17 +23,15 @@ steps:
   # - tests/entrypoints/openai/test_uds.py
   - tests/v1/sample/test_logprobs_e2e.py
   commands:
-  - export VLLM_USE_RUST_FRONTEND=1
-  - export VLLM_WORKER_MULTIPROC_METHOD=spawn
-  - pytest -v -s benchmarks/test_serve_cli.py -k "not insecure and not (test_bench_serve and not test_bench_serve_chat)"
-  - pytest -v -s entrypoints/openai/chat_completion/test_chat_completion.py
-  # - pytest -v -s entrypoints/openai/chat_completion/test_chat_logit_bias_validation.py -k "not invalid"
+  - VLLM_USE_RUST_FRONTEND=1 VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s benchmarks/test_serve_cli.py -k "not insecure and not (test_bench_serve and not test_bench_serve_chat)"
+  - VLLM_USE_RUST_FRONTEND=1 VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s entrypoints/openai/chat_completion/test_chat_completion.py
+  # - VLLM_USE_RUST_FRONTEND=1 VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s entrypoints/openai/chat_completion/test_chat_logit_bias_validation.py -k "not invalid"
 
-  # - pytest -v -s entrypoints/openai/completion/test_prompt_validation.py -k "not prompt_embeds"
-  - pytest -v -s entrypoints/openai/completion/test_shutdown.py -k "not engine_failure and not test_abort_timeout_exits_quickly"
-  # - pytest -v -s entrypoints/openai/test_return_token_ids.py
-  # - pytest -v -s entrypoints/openai/test_uds.py
-  - pytest -v -s v1/sample/test_logprobs_e2e.py -k "test_prompt_logprobs_e2e_server"
+  # - VLLM_USE_RUST_FRONTEND=1 VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s entrypoints/openai/completion/test_prompt_validation.py -k "not prompt_embeds"
+  - VLLM_USE_RUST_FRONTEND=1 VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s entrypoints/openai/completion/test_shutdown.py -k "not engine_failure and not test_abort_timeout_exits_quickly"
+  # - VLLM_USE_RUST_FRONTEND=1 VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s entrypoints/openai/test_return_token_ids.py
+  # - VLLM_USE_RUST_FRONTEND=1 VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s entrypoints/openai/test_uds.py
+  - VLLM_USE_RUST_FRONTEND=1 VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s v1/sample/test_logprobs_e2e.py -k "test_prompt_logprobs_e2e_server"
 
 - label: Rust Frontend Serve/Admin Coverage
   timeout_in_minutes: 60
@@ -51,13 +49,11 @@ steps:
   - tests/entrypoints/serve/instrumentator/test_metrics.py
   # - tests/entrypoints/serve/dev/test_sleep.py
   commands:
-  - export VLLM_USE_RUST_FRONTEND=1
-  - export VLLM_WORKER_MULTIPROC_METHOD=spawn
-  # - pytest -v -s entrypoints/serve/dev/rpc/test_collective_rpc.py
-  - pytest -v -s entrypoints/serve/instrumentator/test_basic.py -k "not show_version and not server_load"
-  - pytest -v -s entrypoints/serve/disagg/test_serving_tokens.py -k "not stream and not lora and not test_generate_logprobs and not stop_string_workflow"
-  - pytest -v -s entrypoints/serve/instrumentator/test_metrics.py -k "text and not show and not run_batch and not test_metrics_counts and not test_metrics_exist"
-  # - pytest -v -s entrypoints/serve/dev/test_sleep.py
+  # - VLLM_USE_RUST_FRONTEND=1 VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s entrypoints/serve/dev/rpc/test_collective_rpc.py
+  - VLLM_USE_RUST_FRONTEND=1 VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s entrypoints/serve/instrumentator/test_basic.py -k "not show_version and not server_load"
+  - VLLM_USE_RUST_FRONTEND=1 VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s entrypoints/serve/disagg/test_serving_tokens.py -k "not stream and not lora and not test_generate_logprobs and not stop_string_workflow"
+  - VLLM_USE_RUST_FRONTEND=1 VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s entrypoints/serve/instrumentator/test_metrics.py -k "text and not show and not run_batch and not test_metrics_counts and not test_metrics_exist"
+  # - VLLM_USE_RUST_FRONTEND=1 VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s entrypoints/serve/dev/test_sleep.py
 
 - label: Rust Frontend Core Correctness
   timeout_in_minutes: 30
@@ -69,9 +65,7 @@ steps:
   - tests/utils.py
   - tests/entrypoints/openai/correctness/test_lmeval.py
   commands:
-  - export VLLM_USE_RUST_FRONTEND=1
-  - export VLLM_WORKER_MULTIPROC_METHOD=spawn
-  - pytest -s entrypoints/openai/correctness/test_lmeval.py::test_lm_eval_accuracy_v1_engine
+  - VLLM_USE_RUST_FRONTEND=1 VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -s entrypoints/openai/correctness/test_lmeval.py::test_lm_eval_accuracy_v1_engine
 
 - label: Rust Frontend Tool Use
   timeout_in_minutes: 60
@@ -83,9 +77,7 @@ steps:
   - tests/utils.py
   - tests/tool_use/
   commands:
-  - export VLLM_USE_RUST_FRONTEND=1
-  - export VLLM_WORKER_MULTIPROC_METHOD=spawn
-  - pytest -v -s tool_use --ignore=tool_use/mistral --models llama3.2 -k "not test_response_format_with_tool_choice_required and not test_parallel_tool_calls_false and not test_tool_call_and_choice"
+  - VLLM_USE_RUST_FRONTEND=1 VLLM_WORKER_MULTIPROC_METHOD=spawn pytest -v -s tool_use --ignore=tool_use/mistral --models llama3.2 -k "not test_response_format_with_tool_choice_required and not test_parallel_tool_calls_false and not test_tool_call_and_choice"
 
 - label: Rust Frontend Distributed
   timeout_in_minutes: 30
@@ -103,9 +95,6 @@ steps:
   - tests/v1/distributed/test_hybrid_lb_dp.py
   - tests/v1/distributed/test_internal_lb_dp.py
   commands:
-  - export VLLM_USE_RUST_FRONTEND=1
-  - export VLLM_WORKER_MULTIPROC_METHOD=spawn
-  - export NCCL_CUMEM_HOST_ENABLE=0
-  - TP_SIZE=1 DP_SIZE=4 pytest -v -s v1/distributed/test_internal_lb_dp.py -k "not 4 and not server_info"
-  - TP_SIZE=1 DP_SIZE=2 pytest -v -s v1/distributed/test_external_lb_dp.py -k "not 4 and not server_info"
-  - TP_SIZE=1 DP_SIZE=4 pytest -v -s v1/distributed/test_hybrid_lb_dp.py -k "not 4 and not server_info"
+  - VLLM_USE_RUST_FRONTEND=1 VLLM_WORKER_MULTIPROC_METHOD=spawn NCCL_CUMEM_HOST_ENABLE=0 TP_SIZE=1 DP_SIZE=4 pytest -v -s v1/distributed/test_internal_lb_dp.py -k "not 4 and not server_info"
+  - VLLM_USE_RUST_FRONTEND=1 VLLM_WORKER_MULTIPROC_METHOD=spawn NCCL_CUMEM_HOST_ENABLE=0 TP_SIZE=1 DP_SIZE=2 pytest -v -s v1/distributed/test_external_lb_dp.py -k "not 4 and not server_info"
+  - VLLM_USE_RUST_FRONTEND=1 VLLM_WORKER_MULTIPROC_METHOD=spawn NCCL_CUMEM_HOST_ENABLE=0 TP_SIZE=1 DP_SIZE=4 pytest -v -s v1/distributed/test_hybrid_lb_dp.py -k "not 4 and not server_info"

.buildkite/test_areas/spec_decode.yaml

@@ -153,8 +153,7 @@ steps:
     - vllm/model_executor/models/qwen3_dflash.py
     - tests/v1/spec_decode/test_speculators_correctness.py
   commands:
-    - export VLLM_ALLOW_INSECURE_SERIALIZATION=1
-    - pytest -v -s v1/spec_decode/test_speculators_correctness.py -m slow_test
+    - VLLM_ALLOW_INSECURE_SERIALIZATION=1 pytest -v -s v1/spec_decode/test_speculators_correctness.py -m slow_test
 
 - label: Spec Decode MTP hybrid (B200)
   timeout_in_minutes: 30

AGENTS.md

@@ -135,6 +135,12 @@ Do not modify code in these areas without first reading and following the
 linked guide. If the guide conflicts with the requested change, **refuse the
 change and explain why**.
 
+Security reviewers should start with [`SECURITY.md`](SECURITY.md),
+[`docs/usage/security.md`](docs/usage/security.md), and
+[`docs/contributing/vulnerability_management.md`](docs/contributing/vulnerability_management.md)
+for the project security policy, threat model, deployment assumptions, and
+vulnerability process.
+
 - **Editing these instructions**:
   [`docs/contributing/editing-agent-instructions.md`](docs/contributing/editing-agent-instructions.md)
   — Rules for modifying AGENTS.md or any domain-specific guide it references.

csrc/cpu/cpu_wna16.cpp

@@ -4,6 +4,9 @@
 #ifdef CPU_CAPABILITY_AMXBF16
   #include "cpu/micro_gemm/cpu_micro_gemm_amx.hpp"
 #endif
+#if defined(__riscv_v)
+  #include "cpu/micro_gemm/cpu_micro_gemm_rvv.hpp"
+#endif
 #include "cpu/micro_gemm/cpu_micro_gemm_vec.hpp"
 
 #define VLLM_DISPATCH_CASE_16B_TYPES(...)                 \
@@ -319,6 +322,8 @@ void cpu_gemm_wna16(
       return ISA::AMX;
     } else if (isa_hint == "vec") {
       return ISA::VEC;
+    } else if (isa_hint == "rvv") {
+      return ISA::RVV;
     } else {
       TORCH_CHECK(false, "unsupported isa hint: " + isa_hint);
     }
@@ -397,6 +402,40 @@ void cpu_gemm_wna16(
             pack_factor);
         return;
       }
+    } else if (isa == ISA::RVV) {
+      using gemm_t = cpu_micro_gemm::MicroGemm<ISA::RVV, scalar_t>;
+      if (has_zp) {
+        using dequantizer_t = Dequantizer4b<scalar_t, ISA::RVV, true, false>;
+        cpu_gemm_wna16_impl<scalar_t, dequantizer_t, gemm_t>(
+            input.data_ptr<scalar_t>(), q_weight.data_ptr<int32_t>(),
+            output.data_ptr<scalar_t>(), scales.data_ptr<scalar_t>(), zeros_ptr,
+            g_idx_ptr, bias.has_value() ? bias->data_ptr<scalar_t>() : nullptr,
+            a_m_size, b_n_size, a_k_size, a_m_stride, output_m_stride,
+            scales_group_stride, zeros_group_stride, group_num, group_size,
+            pack_factor);
+        return;
+      }
+      if (use_desc_act) {
+        using dequantizer_t = Dequantizer4b<scalar_t, ISA::RVV, false, true>;
+        cpu_gemm_wna16_impl<scalar_t, dequantizer_t, gemm_t>(
+            input.data_ptr<scalar_t>(), q_weight.data_ptr<int32_t>(),
+            output.data_ptr<scalar_t>(), scales.data_ptr<scalar_t>(), zeros_ptr,
+            g_idx_ptr, bias.has_value() ? bias->data_ptr<scalar_t>() : nullptr,
+            a_m_size, b_n_size, a_k_size, a_m_stride, output_m_stride,
+            scales_group_stride, zeros_group_stride, group_num, group_size,
+            pack_factor);
+        return;
+      } else {
+        using dequantizer_t = Dequantizer4b<scalar_t, ISA::RVV, false, false>;
+        cpu_gemm_wna16_impl<scalar_t, dequantizer_t, gemm_t>(
+            input.data_ptr<scalar_t>(), q_weight.data_ptr<int32_t>(),
+            output.data_ptr<scalar_t>(), scales.data_ptr<scalar_t>(), zeros_ptr,
+            g_idx_ptr, bias.has_value() ? bias->data_ptr<scalar_t>() : nullptr,
+            a_m_size, b_n_size, a_k_size, a_m_stride, output_m_stride,
+            scales_group_stride, zeros_group_stride, group_num, group_size,
+            pack_factor);
+        return;
+      }
     }
   });
 }

csrc/cpu/micro_gemm/cpu_micro_gemm_rvv.hpp

@@ -0,0 +1,228 @@
+#ifndef CPU_MICRO_GEMM_RVV_HPP
+#define CPU_MICRO_GEMM_RVV_HPP
+
+#include "cpu/micro_gemm/cpu_micro_gemm_impl.hpp"
+
+#if defined(__riscv_v)
+
+namespace cpu_micro_gemm {
+namespace {
+
+constexpr int32_t RVV_MGEMM_N8 = 8;
+constexpr int32_t RVV_MGEMM_B_GROUP_STRIDE = 16;
+
+template <typename scalar_t>
+FORCE_INLINE fixed_fp32x8_t load_row8_b_as_f32(const scalar_t* ptr);
+
+template <>
+FORCE_INLINE fixed_fp32x8_t load_row8_b_as_f32<float>(const float* ptr) {
+  return RVVI(__riscv_vle32_v_f32, LMUL_256)(ptr, RVV_MGEMM_N8);
+}
+
+template <>
+FORCE_INLINE fixed_fp32x8_t
+load_row8_b_as_f32<c10::Half>(const c10::Half* ptr) {
+  #if defined(__riscv_zvfh)
+  fixed_fp16x8_t vec = RVVI(__riscv_vle16_v_f16, LMUL_128)(
+      reinterpret_cast<const _Float16*>(ptr), RVV_MGEMM_N8);
+  return RVVI(__riscv_vfwcvt_f_f_v_f32, LMUL_256)(vec, RVV_MGEMM_N8);
+  #else
+  alignas(32) float values[RVV_MGEMM_N8];
+  for (int32_t i = 0; i < RVV_MGEMM_N8; ++i) {
+    values[i] = static_cast<float>(ptr[i]);
+  }
+  return RVVI(__riscv_vle32_v_f32, LMUL_256)(values, RVV_MGEMM_N8);
+  #endif
+}
+
+template <>
+FORCE_INLINE fixed_fp32x8_t
+load_row8_b_as_f32<c10::BFloat16>(const c10::BFloat16* ptr) {
+  #if defined(__riscv_zvfbfmin)
+  fixed_u16x8_t raw = RVVI(__riscv_vle16_v_u16, LMUL_128)(
+      reinterpret_cast<const uint16_t*>(ptr), RVV_MGEMM_N8);
+  fixed_bf16x8_t vec =
+      RVVI4(__riscv_vreinterpret_v_u16, LMUL_128, _bf16, LMUL_128)(raw);
+  return RVVI(__riscv_vfwcvtbf16_f_f_v_f32, LMUL_256)(vec, RVV_MGEMM_N8);
+  #else
+  fixed_u16x8_t raw = RVVI(__riscv_vle16_v_u16, LMUL_128)(
+      reinterpret_cast<const uint16_t*>(ptr), RVV_MGEMM_N8);
+  auto wide = RVVI(__riscv_vzext_vf2_u32, LMUL_256)(raw, RVV_MGEMM_N8);
+  auto shifted = RVVI(__riscv_vsll_vx_u32, LMUL_256)(wide, 16, RVV_MGEMM_N8);
+  return RVVI4(__riscv_vreinterpret_v_u32, LMUL_256, _f32, LMUL_256)(shifted);
+  #endif
+}
+
+// Mx8 RVV kernel. B points at one 8-channel half of a 16-channel packed group,
+// with rows separated by RVV_MGEMM_B_GROUP_STRIDE scalar elements.
+template <int32_t M, typename scalar_t>
+FORCE_INLINE void gemm_micro_rvv_fma_mx8_ku4(const scalar_t* __restrict__ a_ptr,
+                                             const scalar_t* __restrict__ b_ptr,
+                                             float* __restrict__ c_ptr,
+                                             const int64_t lda,
+                                             const int64_t ldc, const int32_t k,
+                                             const bool accum_c) {
+  static_assert(0 < M && M <= 8);
+
+  #define RVV_ROWS_APPLY(OP) OP(0) OP(1) OP(2) OP(3) OP(4) OP(5) OP(6) OP(7)
+  #define RVV_IF_M(i) if constexpr (M > (i))
+
+  #define RVV_DECL_A(i) const scalar_t* __restrict__ a##i = a_ptr + (i) * lda;
+  RVV_ROWS_APPLY(RVV_DECL_A)
+  #undef RVV_DECL_A
+
+  #define RVV_DECL_ACC(i) fixed_fp32x8_t acc##i;
+  RVV_ROWS_APPLY(RVV_DECL_ACC)
+  #undef RVV_DECL_ACC
+
+  #define RVV_INIT_ACC(i)                                                  \
+    RVV_IF_M(i) {                                                          \
+      if (accum_c) {                                                       \
+        acc##i = RVVI(__riscv_vle32_v_f32, LMUL_256)(c_ptr + (i) * ldc,    \
+                                                     RVV_MGEMM_N8);        \
+      } else {                                                             \
+        acc##i = RVVI(__riscv_vfmv_v_f_f32, LMUL_256)(0.0f, RVV_MGEMM_N8); \
+      }                                                                    \
+    }
+  RVV_ROWS_APPLY(RVV_INIT_ACC)
+  #undef RVV_INIT_ACC
+
+  int32_t k_idx = 0;
+  for (; k_idx + 3 < k; k_idx += 4) {
+  #define RVV_FMA_ROW(i, K_OFFSET)                                     \
+    RVV_IF_M(i) {                                                      \
+      acc##i = RVVI(__riscv_vfmacc_vf_f32, LMUL_256)(                  \
+          acc##i, static_cast<float>(*(a##i + k_idx + (K_OFFSET))), b, \
+          RVV_MGEMM_N8);                                               \
+    }
+
+  #define RVV_STEP_K(K_OFFSET)                                      \
+    {                                                               \
+      fixed_fp32x8_t b = load_row8_b_as_f32<scalar_t>(              \
+          b_ptr + (k_idx + (K_OFFSET)) * RVV_MGEMM_B_GROUP_STRIDE); \
+      RVV_FMA_ROW(0, K_OFFSET)                                      \
+      RVV_FMA_ROW(1, K_OFFSET)                                      \
+      RVV_FMA_ROW(2, K_OFFSET)                                      \
+      RVV_FMA_ROW(3, K_OFFSET)                                      \
+      RVV_FMA_ROW(4, K_OFFSET)                                      \
+      RVV_FMA_ROW(5, K_OFFSET)                                      \
+      RVV_FMA_ROW(6, K_OFFSET)                                      \
+      RVV_FMA_ROW(7, K_OFFSET)                                      \
+    }
+
+    RVV_STEP_K(0)
+    RVV_STEP_K(1)
+    RVV_STEP_K(2)
+    RVV_STEP_K(3)
+  #undef RVV_STEP_K
+  #undef RVV_FMA_ROW
+  }
+
+  for (; k_idx < k; ++k_idx) {
+    fixed_fp32x8_t b =
+        load_row8_b_as_f32<scalar_t>(b_ptr + k_idx * RVV_MGEMM_B_GROUP_STRIDE);
+  #define RVV_TAIL_ROW(i)                                                \
+    RVV_IF_M(i) {                                                        \
+      acc##i = RVVI(__riscv_vfmacc_vf_f32, LMUL_256)(                    \
+          acc##i, static_cast<float>(*(a##i + k_idx)), b, RVV_MGEMM_N8); \
+    }
+    RVV_ROWS_APPLY(RVV_TAIL_ROW)
+  #undef RVV_TAIL_ROW
+  }
+
+  #define RVV_STORE_ROW(i)                                           \
+    RVV_IF_M(i) {                                                    \
+      RVVI(__riscv_vse32_v_f32, LMUL_256)(c_ptr + (i) * ldc, acc##i, \
+                                          RVV_MGEMM_N8);             \
+    }
+  RVV_ROWS_APPLY(RVV_STORE_ROW)
+  #undef RVV_STORE_ROW
+
+  #undef RVV_ROWS_APPLY
+  #undef RVV_IF_M
+}
+
+template <int32_t M, typename scalar_t>
+FORCE_INLINE void gemm_micro_rvv_mx32_ku4(DEFINE_CPU_MICRO_GEMM_PARAMS) {
+  static_assert(0 < M && M <= 8);
+  scalar_t* __restrict__ curr_b_0 = b_ptr;
+  scalar_t* __restrict__ curr_b_1 = b_ptr + b_n_group_stride;
+
+  gemm_micro_rvv_fma_mx8_ku4<M>(a_ptr, curr_b_0, c_ptr, lda, ldc, k, accum_c);
+  gemm_micro_rvv_fma_mx8_ku4<M>(a_ptr, curr_b_0 + RVV_MGEMM_N8,
+                                c_ptr + RVV_MGEMM_N8, lda, ldc, k, accum_c);
+  gemm_micro_rvv_fma_mx8_ku4<M>(a_ptr, curr_b_1, c_ptr + 16, lda, ldc, k,
+                                accum_c);
+  gemm_micro_rvv_fma_mx8_ku4<M>(a_ptr, curr_b_1 + RVV_MGEMM_N8, c_ptr + 24, lda,
+                                ldc, k, accum_c);
+}
+
+class TileGemmRVV {
+ public:
+  template <typename scalar_t>
+  FORCE_INLINE static void gemm(DEFINE_CPU_MICRO_GEMM_PARAMS) {
+    switch (m) {
+      case 1:
+        gemm_micro_rvv_mx32_ku4<1>(CPU_MICRO_GEMM_PARAMS);
+        break;
+      case 2:
+        gemm_micro_rvv_mx32_ku4<2>(CPU_MICRO_GEMM_PARAMS);
+        break;
+      case 3:
+        gemm_micro_rvv_mx32_ku4<3>(CPU_MICRO_GEMM_PARAMS);
+        break;
+      case 4:
+        gemm_micro_rvv_mx32_ku4<4>(CPU_MICRO_GEMM_PARAMS);
+        break;
+      case 5:
+        gemm_micro_rvv_mx32_ku4<5>(CPU_MICRO_GEMM_PARAMS);
+        break;
+      case 6:
+        gemm_micro_rvv_mx32_ku4<6>(CPU_MICRO_GEMM_PARAMS);
+        break;
+      case 7:
+        gemm_micro_rvv_mx32_ku4<7>(CPU_MICRO_GEMM_PARAMS);
+        break;
+      case 8:
+        gemm_micro_rvv_mx32_ku4<8>(CPU_MICRO_GEMM_PARAMS);
+        break;
+    }
+  }
+};
+
+}  // namespace
+
+template <typename scalar_t>
+class MicroGemm<cpu_utils::ISA::RVV, scalar_t> {
+ public:
+  static constexpr int32_t MaxMSize = 8;
+  static constexpr int32_t NSize = 32;
+
+ public:
+  void gemm(DEFINE_CPU_MICRO_GEMM_PARAMS) {
+    TileGemmRVV::gemm<scalar_t>(CPU_MICRO_GEMM_PARAMS);
+  }
+
+  static void pack_weight(const scalar_t* __restrict__ weight,
+                          scalar_t* __restrict__ packed_weight,
+                          const int32_t output_size, const int32_t input_size) {
+    TORCH_CHECK_EQ(output_size % 16, 0);
+    for (int32_t o_idx = 0; o_idx < output_size; ++o_idx) {
+      const scalar_t* __restrict__ curr_weight = weight + o_idx * input_size;
+      scalar_t* __restrict__ curr_packed_weight =
+          packed_weight + (o_idx / 16) * (16 * input_size) + o_idx % 16;
+      for (int32_t i_idx = 0; i_idx < input_size; ++i_idx) {
+        *curr_packed_weight = *curr_weight;
+
+        curr_packed_weight += 16;
+        ++curr_weight;
+      }
+    }
+  }
+};
+
+}  // namespace cpu_micro_gemm
+
+#endif  // defined(__riscv_v)
+
+#endif  // CPU_MICRO_GEMM_RVV_HPP

csrc/cpu/utils.hpp

@@ -8,13 +8,15 @@
 #include "cpu/cpu_types.hpp"
 
 namespace cpu_utils {
-enum class ISA { AMX, VEC };
+enum class ISA { AMX, VEC, RVV };
 
 inline ISA get_isa(const std::string& isa) {
   if (isa == "amx") {
     return ISA::AMX;
   } else if (isa == "vec") {
     return ISA::VEC;
+  } else if (isa == "rvv") {
+    return ISA::RVV;
   } else {
     TORCH_CHECK(false, "Invalid isa type: " + isa);
   }

csrc/libtorch_stable/attention/mla/sm100_cutlass_mla_kernel.cu

@@ -136,8 +136,12 @@ typename T::Fmha::Arguments args_from_options(
   StrideQ stride_Q_pe = cute::make_tuple(
       static_cast<int64_t>(q_pe.stride(1)), _1{}, static_cast<int64_t>(q_pe.stride(0)));
 
+  // Read the token and page strides from the cache tensor instead of assuming
+  // packed pages, so strided views (e.g. per-layer views into a cross-layer
+  // block-major cache) are addressed correctly.
   StrideK stride_C = cute::make_tuple(
-      static_cast<int64_t>(0 + D_latent + D_rope), _1{}, static_cast<int64_t>(page_size * (D_latent + D_rope)));
+      static_cast<int64_t>(kv_c_and_k_pe_cache.stride(1)), _1{},
+      static_cast<int64_t>(kv_c_and_k_pe_cache.stride(0)));
   StrideLSE stride_PT = cute::make_stride(_1{}, page_count_per_seq);
   StrideLSE stride_LSE = cute::make_tuple(_1{}, 0 + H);
   StrideO stride_O = cute::make_tuple(static_cast<int64_t>(0 + D_latent), _1{}, static_cast<int64_t>(0 + H * D_latent));

csrc/libtorch_stable/cache_kernels.cu

@@ -549,7 +549,7 @@ __global__ void indexer_k_quant_and_cache_kernel(
     const int head_dim,                        // dimension of each head
     const int quant_block_size,                // quantization block size
     const int cache_block_size,                // cache block size
-    const int cache_stride,  // stride for each token in kv_cache
+    const int64_t cache_block_stride,  // stride for each block in kv_cache
 
     const bool use_ue8m0  // use ue8m0 scale format
 ) {
@@ -590,16 +590,15 @@ __global__ void indexer_k_quant_and_cache_kernel(
     scale = exp2f(ceilf(log2f(scale)));
   }
 
-  const int64_t dst_offset = block_idx * cache_block_size * cache_stride +
-                             block_offset * head_dim + head_dim_idx;
+  const int64_t dst_offset =
+      block_idx * cache_block_stride + block_offset * head_dim + head_dim_idx;
   for (int i = 0; i < VEC_SIZE; i++) {
     kv_cache[dst_offset + i] =
         fp8::scaled_convert<cache_t, scalar_t, kv_dt>(k_val_ptr[i], scale);
   }
   if (threadIdx.x == 0) {
     const int64_t dst_scale_idx =
-        block_idx * cache_block_size * cache_stride +
-        cache_block_size * head_dim +
+        block_idx * cache_block_stride + cache_block_size * head_dim +
         (block_offset * head_dim + head_dim_idx) * 4 / quant_block_size;
     reinterpret_cast<float*>(kv_cache)[dst_scale_idx / 4] = scale;
   }
@@ -1452,7 +1451,7 @@ void cp_gather_and_upconvert_fp8_kv_cache(
           reinterpret_cast<KV_T*>(k.data_ptr()),                              \
           reinterpret_cast<CACHE_T*>(kv_cache.data_ptr()),                    \
           slot_mapping.const_data_ptr<int64_t>(), head_dim, quant_block_size, \
-          cache_block_size, cache_stride, use_ue8m0);
+          cache_block_size, cache_block_stride, use_ue8m0);
 
 void indexer_k_quant_and_cache(
     torch::stable::Tensor& k,         // [num_tokens, head_dim]
@@ -1463,7 +1462,7 @@ void indexer_k_quant_and_cache(
   int num_tokens = k.size(0);
   int head_dim = k.size(1);
   int cache_block_size = kv_cache.size(1);
-  int cache_stride = kv_cache.size(2);
+  int64_t cache_block_stride = kv_cache.stride(0);
   bool use_ue8m0 = scale_fmt == "ue8m0";
 
   STD_TORCH_CHECK(k.device() == kv_cache.device(),

docker/Dockerfile.xpu

@@ -132,7 +132,7 @@ CMD ["/bin/bash"]
 FROM vllm-base AS ucx-nixl-build
 
 ARG UCX_VERSION=v1.21.0-rc2
-ARG NIXL_VERSION=0.10.1
+ARG NIXL_VERSION=v1.2.0
 
 # Build-time only: compiler, autotools, and verbs dev headers
 RUN apt-get update -y && apt-get install -y --no-install-recommends \
@@ -149,25 +149,25 @@ RUN apt-get update -y && apt-get install -y --no-install-recommends \
 # patchelf (installed via uv) is used by the NIXL wheel build to rewrite
 # RPATH entries, making the wheel portable across stages.
 RUN --mount=type=cache,target=/root/.cache/uv \
-    git clone https://github.com/openucx/ucx /tmp/ucx_source && \
-    cd /tmp/ucx_source && git checkout "${UCX_VERSION}" && \
+    git clone --depth 1 --branch "${UCX_VERSION}" https://github.com/openucx/ucx /tmp/ucx_source && \
+    cd /tmp/ucx_source && \
     bash autogen.sh && \
     ./configure --prefix=/tmp/ucx_install --with-ze=yes --enable-examples --enable-mt && \
-    make CFLAGS="-Wno-error=incompatible-pointer-types" -j8 && make install && \
-    git clone https://github.com/ai-dynamo/nixl /tmp/nixl_source && \
-    cd /tmp/nixl_source && git checkout "${NIXL_VERSION}" && \
+    make CFLAGS="-Wno-error=incompatible-pointer-types" -j"$(nproc)" && make install && \
+    git clone --depth 1 --branch "${NIXL_VERSION}" https://github.com/ai-dynamo/nixl /tmp/nixl_source && \
+    cd /tmp/nixl_source && \
     uv pip install --upgrade meson pybind11 patchelf && \
     uv pip install -r requirements.txt && \
     PKG_CONFIG_PATH=/tmp/ucx_install/lib/pkgconfig \
     LD_LIBRARY_PATH=/tmp/ucx_install/lib \
     python -m pip wheel --no-deps . -w /tmp/nixl_wheels/ && \
     find /tmp/ucx_install -type f \( -name '*.a' -o -name '*.la' \) -delete && \
-    rm -rf /tmp/ucx_install/include /tmp/ucx_install/share /tmp/ucx_install/etc /tmp/ucx_install/lib/cmake /tmp/ucx_install/bin && \
-    rm -rf /tmp/ucx_source /tmp/nixl_source
+    rm -rf /tmp/ucx_install/{include,share,etc,bin} /tmp/ucx_install/lib/cmake \
+           /tmp/ucx_source /tmp/nixl_source
 
 FROM vllm-base AS vllm-openai
 
-ARG NIXL_VERSION=0.10.1
+ARG NIXL_VERSION=v1.2.0
 
 # Copy compiled UCX runtime libraries and the pre-built NIXL wheel.
 # No compiler or autotools are installed in this stage.
@@ -192,7 +192,7 @@ RUN --mount=type=cache,target=/root/.cache/uv \
     librdmacm1t64 \
     && rm -rf /var/lib/apt/lists/* \
     && uv pip install --no-deps /tmp/nixl_wheels/nixl*.whl \
-    && uv pip install nixl==${NIXL_VERSION} \
+    && uv pip install nixl==${NIXL_VERSION} && uv pip uninstall nixl-cu13 \
     && rm -rf /tmp/nixl_wheels
 
 RUN --mount=type=cache,target=/root/.cache/uv \

docs/contributing/README.md

@@ -304,9 +304,15 @@ review process:
   resources. The reviewer will add `ready` label to the PR when the PR is
   ready to merge or a full CI run is needed.
 
-### Escalating Stalled Contributions
+### Pull Request Limits and Escalation
 
-If you have an important contribution that has not yet received maintainer attention, please email us at:
+vLLM uses GitHub's [pull request limit](https://github.blog/open-source/maintainers/how-pull-request-limits-are-cutting-down-the-noise/)
+for contributors without write access. The current cap is 6 open PRs. If this
+blocks well-intentioned critical work, contact a committer to request bypass
+list access.
+
+If you need an expedited review for an important contribution, please email us
+at:
 
 <pr-review-request@vllm.ai>
 

docs/contributing/model/basic.md

@@ -136,7 +136,7 @@ The model should also be added to the `MODELS_CONFIG_MAP` dictionary in [vllm/mo
 For case (2), we recommend using as a reference the implementation of [`JambaForCausalLM`](../../../vllm/model_executor/models/jamba.py) (for an example of a model that uses Mamba-1 and attention together) or [`NemotronHForCausalLM`](../../../vllm/model_executor/models/nemotron_h.py) (for an example of a model that uses Mamba-2 and attention together).
 These models should follow the same instructions as case (1), but they should inherit protocol `IsHybrid` (instead of `IsAttentionFree`) and it is *not* necessary to add them to the `MODELS_CONFIG_MAP` (their runtime defaults will be inferred from the protocol).
 
-For case (3), we recommend looking at the implementation of [`MiniMaxText01ForCausalLM`](../../../vllm/model_executor/models/minimax_text_01.py) or [`Lfm2ForCausalLM`](../../../vllm/model_executor/models/lfm2.py) as a reference, which use custom "mamba-like" layers `MiniMaxText01LinearAttention` and `ShortConv` respectively.
+For case (3), we recommend looking at the implementation of [`Lfm2ForCausalLM`](../../../vllm/model_executor/models/lfm2.py) as a reference, which uses a custom "mamba-like" layer `ShortConv`.
 Please follow the same guidelines as case (2) for implementing these models.
 We use "mamba-like" to refer to layers that possess a state that is updated in-place, rather than being appended-to (like KV cache for attention).
 For implementing new custom mamba-like layers, one should inherit from `MambaBase` and implement the methods `get_state_dtype`, `get_state_shape` to calculate the data types and state shapes at runtime, as well as `mamba_type` and `get_attn_backend`.
@@ -144,5 +144,5 @@ It is also necessary to implement the "attention meta-data" class which handles
 Please see [`LinearAttentionMetadata`](../../../vllm/v1/attention/backends/linear_attn.py) or [`ShortConvAttentionMetadata`](../../../vllm/v1/attention/backends/short_conv_attn.py) for examples of this.
 It is also worth noting that we should update `MambaAttentionBackendEnum` in [`registry.py`](../../../vllm/v1/attention/backends/registry.py) when adding a new mamba backend.
 Finally, if one wants to support torch compile and CUDA graphs, it necessary to wrap the call to the mamba-like layer inside a custom op and register it.
-Please see the calls to `direct_register_custom_op` in [vllm/model_executor/models/minimax_text_01.py](../../../vllm/model_executor/models/minimax_text_01.py) or [vllm/model_executor/layers/mamba/short_conv.py](../../../vllm/model_executor/layers/mamba/short_conv.py) for examples of this.
+Please see the calls to `direct_register_custom_op` in [vllm/model_executor/layers/mamba/linear/minimax_linear_attn.py](../../../vllm/model_executor/layers/mamba/linear/minimax_linear_attn.py) or [vllm/model_executor/layers/mamba/short_conv.py](../../../vllm/model_executor/layers/mamba/short_conv.py) for examples of this.
 The new custom op should then be added to the list `_attention_ops` in [vllm/config/compilation.py](../../../vllm/config/compilation.py) to ensure that piecewise CUDA graphs works as intended.

docs/design/attention_backends.md

@@ -170,8 +170,8 @@ Priority is **1 = highest** (tried first).
 | Backend | Version | Dtypes | KV Dtypes | Block Sizes | Head Sizes | Sink | Non-Causal | MM Prefix | DCP | Attention Types | Compute Cap. |
 | ------- | ------- | ------ | --------- | ----------- | ---------- | ---- | ---------- | --------- | --- | --------------- | ------------ |
 | `CPU_ATTN` | | fp16, bf16, fp32 | `auto`, `fp8`, `fp8_e4m3`, `fp8_e5m2` | %16 | 32, 64, 80, 96, 112, 128, 160, 192, 224, 256, 512 | ❌ | ❌ | ❌ | ❌ | All | N/A |
-| `FLASHINFER` | Native† | fp16, bf16 | `auto`, `float16`, `bfloat16`, `fp8`, `fp8_e4m3`, `fp8_e5m2` | 16, 32, 64, 128, 256, 512, 1024 | 64, 128, 256, 512 | ❌ | ❌ | ❌ | ✅ | Decoder | 7.x-9.x |
-| `FLASHINFER` | TRTLLM† | fp16, bf16 | `auto`, `float16`, `bfloat16`, `fp8`, `fp8_e4m3`, `fp8_e5m2`, `nvfp4` | 16, 32, 64, 128, 256, 512, 1024 | 64, 128, 256, 512 | ✅ | ❌ | ❌ | ✅ | Decoder | 10.x |
+| `FLASHINFER` | Native† | fp16, bf16 | `auto`, `float16`, `bfloat16`, `fp8`, `fp8_e4m3`, `fp8_e5m2` | 16, 32, 64, 128, 256, 512, 1024 | 64, 128, 256, 512 | ❌ | ✅ | ❌ | ✅ | Decoder | 7.x-9.x |
+| `FLASHINFER` | TRTLLM† | fp16, bf16 | `auto`, `float16`, `bfloat16`, `fp8`, `fp8_e4m3`, `fp8_e5m2`, `nvfp4` | 16, 32, 64, 128, 256, 512, 1024 | 64, 128, 256, 512 | ✅ | ✅ | ❌ | ✅ | Decoder | 10.x |
 | `FLASH_ATTN` | FA2* | fp16, bf16 | `auto`, `float16`, `bfloat16` | %16 | Any | ❌ | ✅ | ❌ | ✅ | All | ≥8.0 |
 | `FLASH_ATTN` | FA3* | fp16, bf16 | `auto`, `float16`, `bfloat16`, `fp8`, `fp8_e4m3`, `fp8_e5m2` | %16 | Any | ✅ | ✅ | ❌ | ✅ | All | 9.x |
 | `FLASH_ATTN` | FA4* | fp16, bf16 | `auto`, `float16`, `bfloat16` | %16 | Any | ✅ | ✅ | ❌ | ✅ | All | ≥10.0 |

docs/features/kv_offloading_usage.md

@@ -74,6 +74,7 @@ vllm serve <model> \
 | `max_tracker_size` | no | `64000` | single-tier | Max entries in the lookup tracker. |
 | `secondary_tiers` | no | `[]` | multi-tier | List of secondary tier configs (see below). |
 | `offload_prompt_only` | no | `true` | both | If `true`, only prompt (prefill) blocks are offloaded; decode blocks are skipped. |
+| `self_describing_kv_events` | no | `false` | single-tier | Opt-in. When `true` *and* KV cache events are enabled (`--kv-events-config` with `enable_kv_cache_events`), the connector emits self-describing block-granular `BlockStored`/`BlockRemoved` payloads (constituent block hashes, whole-chunk `token_ids`, per-block `block_size`, parent hash, LoRA + group/cache-spec metadata) instead of the placeholder fallback, so external KV-event consumers can index offloaded blocks. Inert unless events are enabled. Currently rejected by `TieringOffloadingSpec`. Full-attention groups only; sliding-window/SSM groups keep the placeholder fallback. In chunk mode (`block_size` > GPU block size), overlapping chunks re-announce shared per-block hashes, so consumers must reference-count (deduplicate) repeated store/remove announcements. |
 | `spec_module_path` | no | — | both | Python import path for a custom `OffloadingSpec` not in the built-in registry. Required only when `spec_name` is not built-in (advanced). |
 
 ## Secondary Tiers

docs/features/tool_calling.md

@@ -321,15 +321,6 @@ For Qwen2.5, the chat template in tokenizer_config.json has already included sup
 
 Flags: `--tool-call-parser hermes`
 
-### MiniMax Models (`minimax_m1`)
-
-Supported models:
-
-* `MiniMaxAi/MiniMax-M1-40k` (use with [examples/tool_chat_template_minimax_m1.jinja](../../examples/tool_chat_template_minimax_m1.jinja))
-* `MiniMaxAi/MiniMax-M1-80k` (use with [examples/tool_chat_template_minimax_m1.jinja](../../examples/tool_chat_template_minimax_m1.jinja))
-
-Flags: `--tool-call-parser minimax --chat-template examples/tool_chat_template_minimax_m1.jinja`
-
 ### DeepSeek-V3 Models (`deepseek_v3`)
 
 Supported models:

docs/models/pooling_models/token_embed.md

@@ -61,7 +61,7 @@ Models of any architecture can be converted into embedding models using `--conve
 | `ColModernVBertForRetrieval` | ColModernVBERT | T / I | `ModernVBERT/colmodernvbert-merged` | | |
 | `ColPaliForRetrieval` | ColPali | T / I | `vidore/colpali-v1.3-hf` | | |
 | `ColQwen3` | Qwen3-VL | T / I | `TomoroAI/tomoro-colqwen3-embed-4b`, `TomoroAI/tomoro-colqwen3-embed-8b` | | |
-| `ColQwen3_5` | ColQwen3.5 | T + I + V | `athrael-soju/colqwen3.5-4.5B-v3` | | |
+| `ColQwen3_5` | ColQwen3.5 | T + I + V | `athrael-soju/colqwen3.5-4.5B-v3`, `vultr/VultronRetrieverPrime-Qwen3.5-8B` | | |
 | `OpsColQwen3Model` | Qwen3-VL | T / I | `OpenSearch-AI/Ops-Colqwen3-4B`, `OpenSearch-AI/Ops-Colqwen3-8B` | | |
 | `Qwen3VLNemotronEmbedModel` | Qwen3-VL | T / I | `nvidia/nemotron-colembed-vl-4b-v2`, `nvidia/nemotron-colembed-vl-8b-v2` | ✅︎ | ✅︎ |
 | `*ForConditionalGeneration`<sup>C</sup>, `*ForCausalLM`<sup>C</sup>, etc. | Generative models | \* | N/A | \* | \* |

docs/models/supported_models.md

@@ -441,7 +441,6 @@ th {
 | `MiMoV2ForCausalLM` | MiMoV2Pro | `XiaomiMiMo/MiMo-V2.5-Pro`, etc. | | ✅︎ |
 | `MiniCPMForCausalLM` | MiniCPM | `openbmb/MiniCPM-2B-sft-bf16`, `openbmb/MiniCPM-2B-dpo-bf16`, `openbmb/MiniCPM-S-1B-sft`, etc. | ✅︎ | ✅︎ |
 | `MiniCPM3ForCausalLM` | MiniCPM3 | `openbmb/MiniCPM3-4B`, etc. | ✅︎ | ✅︎ |
-| `MiniMaxForCausalLM` | MiniMax-Text | `MiniMaxAI/MiniMax-Text-01-hf`, etc. | | |
 | `MiniMaxM2ForCausalLM` | MiniMax-M2, MiniMax-M2.1 | `MiniMaxAI/MiniMax-M2`, etc. | ✅︎ | ✅︎ |
 | `MistralForCausalLM` | Ministral-3, Mistral, Mistral-Instruct | `mistralai/Ministral-3-3B-Instruct-2512`, `mistralai/Mistral-7B-v0.1`, `mistralai/Mistral-7B-Instruct-v0.1`, etc. | ✅︎ | ✅︎ |
 | `MistralLarge3ForCausalLM` | Mistral-Large-3-675B-Base-2512, Mistral-Large-3-675B-Instruct-2512 | `mistralai/Mistral-Large-3-675B-Base-2512`, `mistralai/Mistral-Large-3-675B-Instruct-2512`, etc. | ✅︎ | ✅︎ |
@@ -487,8 +486,6 @@ th {
 | `TeleChat2ForCausalLM` | TeleChat2 | `Tele-AI/TeleChat2-3B`, `Tele-AI/TeleChat2-7B`, `Tele-AI/TeleChat2-35B`, etc. | ✅︎ | ✅︎ |
 | `TeleChat3ForCausalLM` | TeleChat3 | `Tele-AI/TeleChat3-36B-Thinking`, `Tele-AI/TeleChat3-Coder-36B-Thinking`, etc. | ✅︎ | ✅︎ |
 | `TeleFLMForCausalLM` | TeleFLM | `CofeAI/FLM-2-52B-Instruct-2407`, `CofeAI/Tele-FLM`, etc. | ✅︎ | ✅︎ |
-| `MiniMaxM1ForCausalLM` | MiniMax-Text | `MiniMaxAI/MiniMax-M1-40k`, `MiniMaxAI/MiniMax-M1-80k`, etc. | | |
-| `MiniMaxText01ForCausalLM` | MiniMax-Text | `MiniMaxAI/MiniMax-Text-01`, etc. | | |
 | `Zamba2ForCausalLM` | Zamba2 | `Zyphra/Zamba2-7B-instruct`, `Zyphra/Zamba2-2.7B-instruct`, `Zyphra/Zamba2-1.2B-instruct`, etc. | | |
 
 !!! note
@@ -595,6 +592,7 @@ These models primarily accept the [`LLM.generate`](./generative_models.md#llmgen
 | `MiMoV2OmniForCausalLM` | MiMo-V2.5-Omni | T + I<sup>E+</sup> + V<sup>E+</sup> + A<sup>+</sup> | `XiaomiMiMo/MiMo-V2.5-Omni` | | ✅︎ |
 | `MiniCPMO` | MiniCPM-O | T + I<sup>E+</sup> + V<sup>E+</sup> + A<sup>E+</sup> | `openbmb/MiniCPM-o-2_6`, etc. | ✅︎ | ✅︎ |
 | `MiniCPMV` | MiniCPM-V | T + I<sup>E+</sup> + V<sup>E+</sup> | `openbmb/MiniCPM-V-2` (see note), `openbmb/MiniCPM-Llama3-V-2_5`, `openbmb/MiniCPM-V-2_6`, `openbmb/MiniCPM-V-4`, `openbmb/MiniCPM-V-4_5`, etc. | ✅︎ | |
+| `MiniMaxM3SparseForConditionalGeneration` | MiniMax-M3 | T + I<sup>+</sup> + V<sup>+</sup> | `MiniMaxAI/MiniMax-M3`, `MiniMaxAI/MiniMax-M3-MXFP8`, etc. | | |
 | `MiniMaxVL01ForConditionalGeneration` | MiniMax-VL | T + I<sup>E+</sup> | `MiniMaxAI/MiniMax-VL-01`, etc. | | ✅︎ |
 | `Mistral3ForConditionalGeneration` | Mistral3 (HF Transformers) | T + I<sup>+</sup> | `mistralai/Mistral-Small-3.1-24B-Instruct-2503`, etc. | ✅︎ | ✅︎ |
 | `MolmoForCausalLM` | Molmo | T + I<sup>+</sup> | `allenai/Molmo-7B-D-0924`, `allenai/Molmo-7B-O-0924`, etc. | ✅︎ | ✅︎ |

docs/usage/v1_guide.md

@@ -128,7 +128,7 @@ Models that use Mamba-2 and Mamba-1 layers (e.g., `Mamba2ForCausalLM`, `MambaFor
 Hybrid models that combine Mamba-2 and Mamba-1 layers with standard attention layers are also supported (e.g., `BambaForCausalLM`,
 `Zamba2ForCausalLM`, `NemotronHForCausalLM`, `FalconH1ForCausalLM` and `GraniteMoeHybridForCausalLM`, `JambaForCausalLM`, `Plamo2ForCausalLM`).
 
-Hybrid models with mechanisms different to Mamba are also supported (e.g, `MiniMaxText01ForCausalLM`, `MiniMaxM1ForCausalLM`, `Lfm2ForCausalLM`).
+Hybrid models with mechanisms different to Mamba are also supported (e.g, `Lfm2ForCausalLM`).
 
 Please note that prefix caching is not yet supported for any of the above models.
 

examples/generate/multimodal/vision_language_offline.py

@@ -1481,39 +1481,6 @@ def run_minicpmv(questions: list[str], modality: str) -> ModelRequestData:
     return run_minicpmv_base(questions, modality, "openbmb/MiniCPM-V-2_6")
 
 
-def run_minimax_vl_01(questions: list[str], modality: str) -> ModelRequestData:
-    assert modality == "image"
-
-    model_name = "MiniMaxAI/MiniMax-VL-01"
-
-    engine_args = EngineArgs(
-        model=model_name,
-        max_num_seqs=2,
-        limit_mm_per_prompt={modality: 1},
-        trust_remote_code=True,
-        tensor_parallel_size=8,
-    )
-
-    tokenizer = AutoTokenizer.from_pretrained(model_name)
-    messages = [
-        [
-            {
-                "role": "user",
-                "content": [{"type": "image"}, {"type": "text", "text": question}],
-            }
-        ]
-        for question in questions
-    ]
-    prompts = tokenizer.apply_chat_template(
-        messages, add_generation_prompt=True, tokenize=False
-    )
-
-    return ModelRequestData(
-        engine_args=engine_args,
-        prompts=prompts,
-    )
-
-
 # Mistral-3 HF-format
 def run_mistral3(questions: list[str], modality: str) -> ModelRequestData:
     assert modality == "image"
@@ -2485,7 +2452,6 @@ model_example_map = {
     "mantis": run_mantis,
     "minicpmo": run_minicpmo,
     "minicpmv": run_minicpmv,
-    "minimax_vl_01": run_minimax_vl_01,
     "mistral3": run_mistral3,
     "molmo": run_molmo,
     "molmo2": run_molmo2,

examples/pooling/score/colqwen3_5_rerank_online.py

@@ -7,11 +7,27 @@ ColQwen3.5 is a multi-modal ColBERT-style model based on Qwen3.5.
 It produces per-token embeddings and uses MaxSim scoring for retrieval
 and reranking. Supports both text and image inputs.
 
+Works for any ColQwen3.5 checkpoint, e.g. `athrael-soju/colqwen3.5-4.5B-v3`
+or `vultr/VultronRetrieverPrime-Qwen3.5-8B`.
+
 Start the server with:
-    vllm serve athrael-soju/colqwen3.5-4.5B --max-model-len 4096
+    vllm serve athrael-soju/colqwen3.5-4.5B-v3 --max-model-len 4096 \
+        --mm-processor-kwargs '{"min_pixels": 65536, "max_pixels": 1835008}'
 
 Then run this script:
     python colqwen3_5_rerank_online.py
+
+Parity note (matching the native colpali ColQwen3_5Processor pipeline):
+  - Visual-token budget: ColQwen3_5Processor uses max_num_visual_tokens=1792,
+    i.e. max_pixels = 1792 * (patch_size*merge_size)^2 = 1792 * 32^2 = 1835008
+    (with min_pixels = shortest_edge = 65536). Pass these via --mm-processor-kwargs
+    as above; the default budget gives fewer visual tokens and lower retrieval ndcg.
+  - When you build prompts yourself (token_embed), reproduce the processor exactly:
+      image (document): wrap in the instruction template
+        "<|im_start|>user\n<|vision_start|><|image_pad|><|vision_end|>"
+        "Describe the image.<|im_end|><|endoftext|>"
+      query: append the augmentation suffix  <text> + "<|endoftext|>" * 10
+    Omitting these reproduces a silent ~2.5 ndcg@10 drop vs the native pipeline.
 """
 
 import requests

examples/tool_chat_template_minimax_m1.jinja

@@ -1,91 +0,0 @@
-{{ '<begin_of_document>' -}}
-{%- if custom_tools is defined %}
-    {%- set tools = custom_tools %}
-{%- endif %}
-{%- if not tools is defined %}
-    {%- set tools = none %}
-{%- endif %}
-
-{#- Extract system message #}
-{% set ns = namespace(system_prompt='') -%}
-{%- if messages[0]['role'] == 'system' %}
-    {%- if messages[0]['content'] is string %}
-        {%- set ns.system_prompt = messages[0]['content']|trim %}
-    {%- else %}
-        {%- set ns.system_prompt = messages[0]['content'][0]['text']|trim %}
-    {%- endif %}
-    {%- set messages = messages[1:] %}
-{%- else %}
-    {%- if tools is not none %}
-        {%- set ns.system_prompt = "You are a helpful assistant created by Minimax based on MiniMax-M1 model." %}
-    {%- else %}
-        {%- set ns.system_prompt = "You are a helpful assistant created by Minimax based on MiniMax-M1 model." %}
-    {%- endif %}
-{%- endif %}
-
-{#- System message #}
-{%- if ns.system_prompt != '' %}
-{{ '<beginning_of_sentence>system ai_setting=assistant\n' + ns.system_prompt + '<end_of_sentence>\n' -}}
-{%- endif %}
-
-{#- Tools configuration #}
-{%- if tools is not none %}
-{{ '<beginning_of_sentence>system tool_setting=tools\nYou are provided with these tools:\n<tools>\n' -}}
-{%- for tool in tools %}
-{{ tool | tojson ~ '\n' -}}
-{%- endfor %}
-{{ '</tools>\n\nIf you need to call tools, please respond with <tool_calls></tool_calls> XML tags, and provide tool-name and json-object of arguments, following the format below:\n<tool_calls>\n{"name": <tool-name>, "arguments": <args-json-object>}\n...\n</tool_calls><end_of_sentence>\n' -}}
-{%- endif %}
-
-{#- Process messages #}
-{%- for message in messages %}
-    {%- if not (message.role == 'ipython' or message.role == 'tool' or 'tool_calls' in message) %}
-        {%- if message['role'] == 'user' %}
-{{ '<beginning_of_sentence>user name=user\n' -}}
-{%- if message['content'] is string %}
-{{ message['content']|trim -}}
-{%- else %}
-{%- for content in message['content'] %}
-{%- if content['type'] == 'text' %}
-{{ content['text']|trim -}}
-{%- endif %}
-{%- endfor %}
-{%- endif %}
-{{ '<end_of_sentence>\n' -}}
-        {%- elif message['role'] == 'assistant' %}
-{{ '<beginning_of_sentence>ai name=assistant\n' -}}
-{%- if message['content'] is string %}
-{{ message['content']|trim -}}
-{%- else %}
-{%- for content in message['content'] | selectattr('type', 'equalto', 'text') %}
-{{ content['text']|trim -}}
-{%- endfor %}
-{%- endif %}
-{{ '<end_of_sentence>\n' -}}
-        {%- endif %}
-    {%- elif 'tool_calls' in message %}
-{{ '<beginning_of_sentence>ai name=assistant\n<tool_calls>\n' -}}
-{%- for tool_call in message.tool_calls %}
-{{ '{"name": "' + tool_call.function.name + '", "arguments": ' + tool_call.function.arguments | tojson + '}\n' -}}
-{%- endfor %}
-{{ '</tool_calls><end_of_sentence>\n' -}}
-    {%- elif message.role == "tool" or message.role == "ipython" %}
-{{ '<beginning_of_sentence>tool name=tools\n' -}}
-{%- if message.content is string %}
-{{ 'tool result: ' + message.content + '\n\n' -}}
-{%- else %}
-{%- for content in message['content'] %}
-{%- if content['type'] == 'text' %}
-{{ 'tool result: ' + content['text'] + '\n\n' -}}
-{%- elif content.get('name') %}
-{{ 'tool name: ' + content['name'] + '\ntool result: ' + content['text'] + '\n\n' -}}
-{%- endif %}
-{%- endfor %}
-{%- endif %}
-{{ '<end_of_sentence>\n' -}}
-    {%- endif %}
-{%- endfor %}
-
-{%- if add_generation_prompt %}
-{{ '<beginning_of_sentence>ai name=assistant\n' -}}
-{%- endif %}

rust/src/chat/src/renderer/hf/format.rs

@@ -386,7 +386,6 @@ mod tests {
             tool_chat_template_llama3.2_pythonic.jinja         => String
             tool_chat_template_llama4_json.jinja               => OpenAi
             tool_chat_template_llama4_pythonic.jinja           => OpenAi
-            tool_chat_template_minimax_m1.jinja                => OpenAi
             tool_chat_template_mistral.jinja                   => String
             tool_chat_template_mistral3.jinja                  => OpenAi
             tool_chat_template_mistral_parallel.jinja          => String

rust/src/chat/tests/templates/vllm_examples/tool_chat_template_minimax_m1.jinja

@@ -1,91 +0,0 @@
-{{ '<begin_of_document>' -}}
-{%- if custom_tools is defined %}
-    {%- set tools = custom_tools %}
-{%- endif %}
-{%- if not tools is defined %}
-    {%- set tools = none %}
-{%- endif %}
-
-{#- Extract system message #}
-{% set ns = namespace(system_prompt='') -%}
-{%- if messages[0]['role'] == 'system' %}
-    {%- if messages[0]['content'] is string %}
-        {%- set ns.system_prompt = messages[0]['content']|trim %}
-    {%- else %}
-        {%- set ns.system_prompt = messages[0]['content'][0]['text']|trim %}
-    {%- endif %}
-    {%- set messages = messages[1:] %}
-{%- else %}
-    {%- if tools is not none %}
-        {%- set ns.system_prompt = "You are a helpful assistant created by Minimax based on MiniMax-M1 model." %}
-    {%- else %}
-        {%- set ns.system_prompt = "You are a helpful assistant created by Minimax based on MiniMax-M1 model." %}
-    {%- endif %}
-{%- endif %}
-
-{#- System message #}
-{%- if ns.system_prompt != '' %}
-{{ '<beginning_of_sentence>system ai_setting=assistant\n' + ns.system_prompt + '<end_of_sentence>\n' -}}
-{%- endif %}
-
-{#- Tools configuration #}
-{%- if tools is not none %}
-{{ '<beginning_of_sentence>system tool_setting=tools\nYou are provided with these tools:\n<tools>\n' -}}
-{%- for tool in tools %}
-{{ tool | tojson ~ '\n' -}}
-{%- endfor %}
-{{ '</tools>\n\nIf you need to call tools, please respond with <tool_calls></tool_calls> XML tags, and provide tool-name and json-object of arguments, following the format below:\n<tool_calls>\n{"name": <tool-name>, "arguments": <args-json-object>}\n...\n</tool_calls><end_of_sentence>\n' -}}
-{%- endif %}
-
-{#- Process messages #}
-{%- for message in messages %}
-    {%- if not (message.role == 'ipython' or message.role == 'tool' or 'tool_calls' in message) %}
-        {%- if message['role'] == 'user' %}
-{{ '<beginning_of_sentence>user name=user\n' -}}
-{%- if message['content'] is string %}
-{{ message['content']|trim -}}
-{%- else %}
-{%- for content in message['content'] %}
-{%- if content['type'] == 'text' %}
-{{ content['text']|trim -}}
-{%- endif %}
-{%- endfor %}
-{%- endif %}
-{{ '<end_of_sentence>\n' -}}
-        {%- elif message['role'] == 'assistant' %}
-{{ '<beginning_of_sentence>ai name=assistant\n' -}}
-{%- if message['content'] is string %}
-{{ message['content']|trim -}}
-{%- else %}
-{%- for content in message['content'] | selectattr('type', 'equalto', 'text') %}
-{{ content['text']|trim -}}
-{%- endfor %}
-{%- endif %}
-{{ '<end_of_sentence>\n' -}}
-        {%- endif %}
-    {%- elif 'tool_calls' in message %}
-{{ '<beginning_of_sentence>ai name=assistant\n<tool_calls>\n' -}}
-{%- for tool_call in message.tool_calls %}
-{{ '{"name": "' + tool_call.function.name + '", "arguments": ' + tool_call.function.arguments | tojson + '}\n' -}}
-{%- endfor %}
-{{ '</tool_calls><end_of_sentence>\n' -}}
-    {%- elif message.role == "tool" or message.role == "ipython" %}
-{{ '<beginning_of_sentence>tool name=tools\n' -}}
-{%- if message.content is string %}
-{{ 'tool result: ' + message.content + '\n\n' -}}
-{%- else %}
-{%- for content in message['content'] %}
-{%- if content['type'] == 'text' %}
-{{ 'tool result: ' + content['text'] + '\n\n' -}}
-{%- elif content.get('name') %}
-{{ 'tool name: ' + content['name'] + '\ntool result: ' + content['text'] + '\n\n' -}}
-{%- endif %}
-{%- endfor %}
-{%- endif %}
-{{ '<end_of_sentence>\n' -}}
-    {%- endif %}
-{%- endfor %}
-
-{%- if add_generation_prompt %}
-{{ '<beginning_of_sentence>ai name=assistant\n' -}}
-{%- endif %}

rust/src/engine-core-client/src/protocol/mod.rs

@@ -277,6 +277,12 @@ pub struct EngineCoreSamplingParams {
     pub max_tokens: u32,
     /// Minimum number of tokens to generate before EOS or stop-token handling.
     pub min_tokens: u32,
+    /// Maximum number of reasoning ("thinking") tokens to emit before the
+    /// reasoning section is force-closed. `None` means unlimited; the
+    /// user-facing `-1` sentinel is normalized to `None` by the frontend before
+    /// reaching this DTO, so only non-negative values are sent. Enforced
+    /// engine-side (and only when a reasoning parser is configured).
+    pub thinking_token_budget: Option<u64>,
     /// Number of log probabilities to return per generated token.
     ///
     /// `None` disables sample logprobs. `-1` requests the full vocabulary.
@@ -345,6 +351,7 @@ impl EngineCoreSamplingParams {
             seed: None,
             max_tokens: 65536,
             min_tokens: 0,
+            thinking_token_budget: None,
             logprobs: None,
             prompt_logprobs: None,
             min_p: 0.0,

rust/src/engine-core-client/src/tests/client.rs

@@ -150,6 +150,7 @@ fn sample_request_with_id(request_id: &str) -> EngineCoreRequest {
             top_k: 8,
             max_tokens: 32,
             min_tokens: 1,
+            thinking_token_budget: Some(256),
             stop_token_ids: vec![151643],
             eos_token_id: Some(151645),
             all_stop_token_ids: BTreeSet::from([151643, 151645]),
@@ -2502,6 +2503,7 @@ fn python_msgpack_fixtures_match_rust_encoding() {
             seed: None,
             max_tokens: 16,
             min_tokens: 0,
+            thinking_token_budget: None,
             logprobs: None,
             prompt_logprobs: None,
             min_p: 0.0,

rust/src/engine-core-client/src/tests/python_compat.py

@@ -39,6 +39,7 @@ class EngineCoreSamplingParams(msgspec.Struct, dict=True, omit_defaults=True):
     seed: int | None = None
     max_tokens: int = 16
     min_tokens: int = 0
+    thinking_token_budget: int | None = None
     min_p: float = 0.0
     frequency_penalty: float = 0.0
     presence_penalty: float = 0.0
@@ -122,6 +123,7 @@ request = EngineCoreRequest(
         seed=None,
         max_tokens=32,
         min_tokens=1,
+        thinking_token_budget=256,
         min_p=0.0,
         frequency_penalty=0.0,
         presence_penalty=0.0,

rust/src/server/src/error.rs

@@ -103,6 +103,7 @@ fn is_request_validation_error(error: &vllm_text::Error) -> bool {
             | vllm_text::Error::EmptyPromptTokenIds { .. }
             | vllm_text::Error::Logprobs(_)
             | vllm_text::Error::OutOfVocab(_)
+            | vllm_text::Error::InvalidThinkingTokenBudget
             // An empty tokenized prompt detected later, at request prepare
             // time, surfaces through the transparent Llm wrapper.
             | vllm_text::Error::Llm(vllm_llm::Error::EmptyPromptTokenIds { .. })
@@ -127,6 +128,18 @@ mod tests {
         assert!(response.error.message.contains("9000"));
     }
 
+    #[test]
+    fn invalid_thinking_token_budget_maps_to_invalid_request() {
+        let api_error = text_submit_error(
+            "failed to submit completion request",
+            vllm_text::Error::InvalidThinkingTokenBudget,
+        );
+        assert_eq!(api_error.status_code(), StatusCode::BAD_REQUEST);
+        let response = api_error.to_error_response();
+        assert_eq!(response.error.error_type, "invalid_request_error");
+        assert!(response.error.message.contains("thinking_token_budget"));
+    }
+
     #[test]
     fn chat_wrapped_prompt_too_long_maps_to_invalid_request() {
         let error = vllm_chat::Error::Text(vllm_text::Error::PromptTooLong {

rust/src/server/src/routes/inference/generate/convert.rs

@@ -150,6 +150,33 @@ mod tests {
         );
     }
 
+    #[test]
+    fn prepare_generate_request_forwards_thinking_token_budget() {
+        let request: GenerateRequest = serde_json::from_value(json!({
+            "model": "Qwen/Qwen1.5-0.5B-Chat",
+            "token_ids": [11, 22, 33],
+            "sampling_params": {
+                "thinking_token_budget": 64
+            }
+        }))
+        .expect("parse request");
+
+        let prepared = prepare_generate_request(
+            request,
+            &served(&["Qwen/Qwen1.5-0.5B-Chat"]),
+            ResolvedRequestContext::default(),
+        )
+        .expect("prepare");
+
+        // The raw inference route shares `vllm_text::SamplingParams`, so the
+        // field is carried through to lowering exactly like the OpenAI routes
+        // (normalization/validation then happens in `lower_sampling_params`).
+        assert_eq!(
+            prepared.text_request.sampling_params.thinking_token_budget,
+            Some(64)
+        );
+    }
+
     #[test]
     fn prepare_generate_request_gates_continuous_usage_on_include_usage() {
         let request: GenerateRequest = serde_json::from_value(json!({

rust/src/server/src/routes/openai/chat_completions/convert.rs

@@ -115,6 +115,7 @@ pub(super) fn prepare_chat_request(
             seed: request.seed,
             max_tokens: request.max_completion_tokens,
             min_tokens: request.min_tokens,
+            thinking_token_budget: request.thinking_token_budget,
             logprobs: request.logprobs.then_some(top_logprobs),
             prompt_logprobs,
             min_p: request.min_p,
@@ -613,6 +614,31 @@ mod tests {
         assert_eq!(prepared.chat_request.sampling_params, expected);
     }
 
+    #[test]
+    fn prepare_chat_request_passes_through_thinking_token_budget() {
+        let prepare = |budget: Option<i64>| {
+            prepare_chat_request(
+                ChatCompletionRequest {
+                    thinking_token_budget: budget,
+                    ..base_request()
+                },
+                &served(&["Qwen/Qwen1.5-0.5B-Chat"]),
+                ResolvedRequestContext::default(),
+            )
+            .expect("request is valid")
+            .chat_request
+            .sampling_params
+            .thinking_token_budget
+        };
+
+        // The convert layer forwards the raw value verbatim (including the `-1`
+        // "unlimited" sentinel); normalization/validation happens during
+        // lowering (see `vllm_text::lower`).
+        assert_eq!(prepare(Some(64)), Some(64));
+        assert_eq!(prepare(Some(-1)), Some(-1));
+        assert_eq!(prepare(None), None);
+    }
+
     #[test]
     fn prepare_chat_request_accepts_developer_messages() {
         let request = ChatCompletionRequest {

rust/src/server/src/routes/openai/chat_completions/types.rs

@@ -165,8 +165,10 @@ pub struct ChatCompletionRequest {
     pub bad_words: Option<Vec<String>>,
 
     // -------- Extra vLLM Parameters --------
-    /// Token budget for reasoning/thinking
-    pub thinking_token_budget: Option<u32>,
+    /// Token budget for reasoning/thinking. Accepts a non-negative integer, or
+    /// `-1` for unlimited (mirroring the Python frontend, which normalizes `-1`
+    /// to "no budget").
+    pub thinking_token_budget: Option<i64>,
 
     /// Whether to include reasoning content in the response
     #[serde(default = "default_true")]

rust/src/server/src/routes/openai/chat_completions/validate.rs

@@ -108,11 +108,6 @@ pub(super) fn validate_request_compat(
         "truncate_prompt_tokens",
         "truncate_prompt_tokens is not supported.",
     )?;
-    reject_non_default(
-        request.thinking_token_budget.as_ref(),
-        "thinking_token_budget",
-        "thinking_token_budget is not supported.",
-    )?;
     reject_non_default(
         request.media_io_kwargs.as_ref(),
         "media_io_kwargs",

rust/src/server/src/routes/openai/completions/convert.rs

@@ -108,6 +108,7 @@ pub(super) fn prepare_completion_request(
             seed: request.seed,
             max_tokens,
             min_tokens: request.min_tokens,
+            thinking_token_budget: request.thinking_token_budget,
             logprobs,
             prompt_logprobs,
             min_p: request.min_p,
@@ -266,6 +267,34 @@ mod tests {
         assert!(!prepared.text_request.decode_options.skip_special_tokens);
     }
 
+    #[test]
+    fn prepare_completion_request_passes_through_thinking_token_budget() {
+        let prepare = |budget: serde_json::Value| {
+            let request: CompletionRequest = serde_json::from_value(json!({
+                "model": "Qwen/Qwen1.5-0.5B-Chat",
+                "prompt": "hello",
+                "thinking_token_budget": budget,
+            }))
+            .expect("parse request");
+            prepare_completion_request(
+                request,
+                &served(&["Qwen/Qwen1.5-0.5B-Chat"]),
+                ResolvedRequestContext::default(),
+            )
+            .expect("prepare")
+            .text_request
+            .sampling_params
+            .thinking_token_budget
+        };
+
+        // The convert layer forwards the raw value verbatim (including the `-1`
+        // "unlimited" sentinel); normalization/validation happens during
+        // lowering (see `vllm_text::lower`).
+        assert_eq!(prepare(json!(64)), Some(64));
+        assert_eq!(prepare(json!(-1)), Some(-1));
+        assert_eq!(prepare(json!(null)), None);
+    }
+
     #[test]
     fn prepare_completion_request_maps_stream_usage_and_token_format_options() {
         let request: CompletionRequest = serde_json::from_value(json!({

rust/src/server/src/routes/openai/completions/types.rs

@@ -146,6 +146,11 @@ pub struct CompletionRequest {
     /// Additional kwargs for structured outputs
     pub structured_outputs: Option<Value>,
 
+    /// Token budget for reasoning/thinking. Accepts a non-negative integer, or
+    /// `-1` for unlimited (mirroring the Python frontend, which normalizes `-1`
+    /// to "no budget").
+    pub thinking_token_budget: Option<i64>,
+
     /// Request scheduling priority (lower means earlier; default 0)
     pub priority: Option<i32>,
 

rust/src/text/src/error.rs

@@ -20,6 +20,8 @@ pub enum Error {
     Logprobs(#[from] LogprobsError),
     #[error(transparent)]
     OutOfVocab(#[from] OutOfVocabError),
+    #[error("`thinking_token_budget` must be a non-negative integer or -1 for unlimited.")]
+    InvalidThinkingTokenBudget,
     #[error("text request stream `{request_id}` closed before terminal output")]
     StreamClosedBeforeTerminalOutput { request_id: String },
     #[error(transparent)]

rust/src/text/src/lower.rs

@@ -87,6 +87,7 @@ pub fn lower_sampling_params(
         seed,
         max_tokens,
         min_tokens,
+        thinking_token_budget,
         logprobs,
         prompt_logprobs,
         min_p,
@@ -128,6 +129,7 @@ pub fn lower_sampling_params(
         prompt_len,
     )?;
     let min_tokens = min_tokens.unwrap_or(0);
+    let thinking_token_budget = normalize_thinking_token_budget(thinking_token_budget)?;
     let frequency_penalty = frequency_penalty.unwrap_or(0.0);
     let presence_penalty = presence_penalty.unwrap_or(0.0);
 
@@ -149,6 +151,7 @@ pub fn lower_sampling_params(
         seed,
         max_tokens,
         min_tokens,
+        thinking_token_budget,
         logprobs,
         prompt_logprobs,
         min_p,
@@ -170,6 +173,21 @@ pub fn lower_sampling_params(
     Ok(params)
 }
 
+/// Normalize the user-facing `thinking_token_budget` into the engine value.
+///
+/// Mirrors Python's `validate_thinking_token_budget`
+/// (<https://github.com/vllm-project/vllm/blob/ecf9d83520eb217401b47d8a5451a27c5231b8c2/vllm/sampling_params.py#L35-L55>):
+/// `None` and the `-1` "unlimited" sentinel both map to `None`; any other
+/// negative value is rejected; non-negative values pass through unchanged. Like
+/// Python's `int`, no upper bound is imposed.
+fn normalize_thinking_token_budget(value: Option<i64>) -> Result<Option<u64>> {
+    match value {
+        None | Some(-1) => Ok(None),
+        Some(budget) if budget >= 0 => Ok(Some(budget as u64)),
+        Some(_) => Err(Error::InvalidThinkingTokenBudget),
+    }
+}
+
 /// Convert bad-word strings into token-ID sequences, following the Python vLLM
 /// logic in `SamplingParams.update_from_tokenizer()`.
 ///
@@ -366,6 +384,36 @@ mod tests {
         )
     }
 
+    #[test]
+    fn lower_sampling_params_normalizes_thinking_token_budget() {
+        let lower = |budget: Option<i64>| {
+            lower_sampling_params_with_limits(
+                SamplingParams {
+                    thinking_token_budget: budget,
+                    ..SamplingParams::default()
+                },
+                sample_sampling_limits(),
+            )
+        };
+
+        // Non-negative budgets (including 0) pass through unchanged.
+        assert_eq!(lower(Some(256)).unwrap().thinking_token_budget, Some(256));
+        assert_eq!(lower(Some(0)).unwrap().thinking_token_budget, Some(0));
+        // `None` and the `-1` "unlimited" sentinel both disable the budget.
+        assert_eq!(lower(None).unwrap().thinking_token_budget, None);
+        assert_eq!(lower(Some(-1)).unwrap().thinking_token_budget, None);
+        // No upper bound is imposed, matching Python's `int`.
+        assert_eq!(
+            lower(Some(i64::from(u32::MAX) + 1)).unwrap().thinking_token_budget,
+            Some(u64::from(u32::MAX) + 1)
+        );
+        // Other negatives are rejected.
+        assert!(matches!(
+            lower(Some(-2)),
+            Err(Error::InvalidThinkingTokenBudget)
+        ));
+    }
+
     #[test]
     fn lower_text_request_applies_python_style_eos_hints() {
         let prepared = lower_text_request(
@@ -386,6 +434,7 @@ mod tests {
                 seed: None,
                 max_tokens: 999997,
                 min_tokens: 0,
+                thinking_token_budget: None,
                 logprobs: None,
                 prompt_logprobs: None,
                 min_p: 0.0,
@@ -437,6 +486,7 @@ mod tests {
                 seed: None,
                 max_tokens: 999997,
                 min_tokens: 0,
+                thinking_token_budget: None,
                 logprobs: None,
                 prompt_logprobs: None,
                 min_p: 0.0,
@@ -567,6 +617,7 @@ mod tests {
                 seed: None,
                 max_tokens: 40957,
                 min_tokens: 0,
+                thinking_token_budget: None,
                 logprobs: None,
                 prompt_logprobs: None,
                 min_p: 0.0,
@@ -628,6 +679,7 @@ mod tests {
                 seed: None,
                 max_tokens: 999997,
                 min_tokens: 0,
+                thinking_token_budget: None,
                 logprobs: None,
                 prompt_logprobs: None,
                 min_p: 0.0,
@@ -697,6 +749,7 @@ mod tests {
                 seed: None,
                 max_tokens: 32,
                 min_tokens: 2,
+                thinking_token_budget: None,
                 logprobs: None,
                 prompt_logprobs: None,
                 min_p: 0.1,
@@ -929,6 +982,7 @@ mod tests {
                 seed: None,
                 max_tokens: 128,
                 min_tokens: 0,
+                thinking_token_budget: None,
                 logprobs: None,
                 prompt_logprobs: None,
                 min_p: 0.1,

rust/src/text/src/output/decoded.rs

@@ -309,7 +309,7 @@ fn matches_stop_string(stops: &[String], output: &str, new_bytes: usize) -> Opti
         .find_map(|(ss_idx, (ss, len, start_off))| {
             output[start_off..]
                 .windows(len)
-                .rposition(|w| w == ss)
+                .position(|w| w == ss)
                 .map(|pos| (ss_idx, start_off + pos))
         })
 }
@@ -562,6 +562,13 @@ mod tests {
         assert_eq!(result, Some((0, 4)));
     }
 
+    #[test]
+    fn stop_string_matches_leftmost_with_multiple_new_bytes() {
+        let stops = vec!["\n".to_string()];
+        let result = matches_stop_string(&stops, "Answer\n\n", 2);
+        assert_eq!(result, Some((0, 6)));
+    }
+
     #[test]
     fn stop_string_matches_at_beginning() {
         let stops = vec!["say".to_string()];

rust/src/text/src/request.rs

@@ -56,6 +56,12 @@ pub struct SamplingParams {
     pub max_tokens: Option<u32>,
     /// Minimum number of tokens to generate before EOS or stop-token handling.
     pub min_tokens: Option<u32>,
+    /// Maximum number of reasoning ("thinking") tokens to emit before the
+    /// reasoning section is force-closed. `None` or the user-facing `-1`
+    /// "unlimited" sentinel both disable the budget. The raw value is carried
+    /// here; `-1` is normalized to `None` (and other negatives rejected) during
+    /// lowering (see `lower_sampling_params`).
+    pub thinking_token_budget: Option<i64>,
     /// Number of log probabilities to return per generated token.
     ///
     /// `None` disables sample logprobs. `-1` requests the full vocabulary.
@@ -116,6 +122,7 @@ impl Default for SamplingParams {
             seed: None,
             max_tokens: None,
             min_tokens: None,
+            thinking_token_budget: None,
             logprobs: None,
             prompt_logprobs: None,
             min_p: None,

tests/entrypoints/serve/tokenize/test_serving_tokenization.py

@@ -15,7 +15,7 @@ from vllm.entrypoints.serve.tokenize.protocol import (
     TokenizeChatRequest,
     TokenizeCompletionRequest,
 )
-from vllm.entrypoints.serve.tokenize.serving import OpenAIServingTokenization
+from vllm.entrypoints.serve.tokenize.serving import ServingTokenization
 from vllm.v1.engine.async_llm import AsyncLLM
 
 MODEL_NAME = "openai-community/gpt2"
@@ -58,7 +58,7 @@ class MockModelConfig:
         return self.diff_sampling_param or {}
 
 
-def _build_serving_tokenization(engine: AsyncLLM) -> OpenAIServingTokenization:
+def _build_serving_tokenization(engine: AsyncLLM) -> ServingTokenization:
     models = OpenAIServingModels(
         engine_client=engine,
         base_model_paths=BASE_MODEL_PATHS,
@@ -71,8 +71,7 @@ def _build_serving_tokenization(engine: AsyncLLM) -> OpenAIServingTokenization:
         chat_template=None,
         chat_template_content_format="auto",
     )
-    return OpenAIServingTokenization(
-        engine,
+    return ServingTokenization(
         models,
         openai_serving_render=serving_render,
         request_logger=None,

tests/kernels/attention/test_attention_selector.py

@@ -435,7 +435,7 @@ def test_per_head_quant_scales_backend_selection(
     ]
     + (
         [
-            ("FLASHINFER", True, False),  # FlashInfer does not support non-causal
+            ("FLASHINFER", True, True),  # FlashInfer supports non-causal
             ("FLASHINFER", False, True),  # FlashInfer works with causal
         ]
         if CudaPlatform is not None

tests/kernels/attention/test_cutlass_mla_decode.py

@@ -212,3 +212,69 @@ def test_cutlass_mla_decode(
     print(
         f"{t:.3f} ms, {FLOPS / 10**9 / t:.0f} TFLOPS,", f"{bytes / 10**6 / t:.0f} GB/s"
     )
+
+
+@pytest.mark.skipif(
+    not current_platform.has_device_capability(100),
+    reason=CUTLASS_MLA_UNSUPPORTED_REASON,
+)
+@torch.inference_mode()
+def test_cutlass_mla_decode_cross_layer_view():
+    """The kernel must read the cache's page-dim stride instead of assuming
+    pages are packed back-to-back. A per-layer view into a cross-layer
+    (block-major) cache has stride(0) inflated by num_layers; outputs must
+    match a contiguous cache holding the same data exactly."""
+    device = torch.device("cuda:0")
+    torch.set_default_dtype(torch.bfloat16)
+    torch.set_default_device(device)
+    torch.manual_seed(42)
+
+    b, mean_sk, d, dv, block_size = 4, 512, 576, 512, 64
+    num_layers, layer_idx = 3, 1
+    scale = math.sqrt(d) ** (-1)
+
+    num_pages = b * (mean_sk // block_size)
+    cache_seqlens = torch.full((b,), mean_sk, dtype=torch.int32)
+    block_table = torch.arange(num_pages, dtype=torch.int32).view(
+        b, mean_sk // block_size
+    )
+
+    kv_contig = torch.randn(num_pages, block_size, d)
+    # Neighbor layers hold random data so packed-pages addressing reads
+    # garbage rather than zeros.
+    kv_cross_layer = torch.randn(num_pages, num_layers, block_size, d)
+    kv_view = kv_cross_layer[:, layer_idx]
+    kv_view.copy_(kv_contig)
+    assert kv_view.stride(0) == num_layers * block_size * d
+
+    q_nope = torch.randn(b, 128, dv)
+    q_pe = torch.randn(b, 128, d - dv)
+    sm_count = num_compute_units(device.index)
+    workspace_size = ops.sm100_cutlass_mla_get_workspace_size(
+        mean_sk, b, sm_count, num_kv_splits=1
+    )
+    workspace = torch.empty(workspace_size, dtype=torch.uint8)
+
+    def run(cache):
+        out = torch.empty(b, 128, dv)
+        lse = torch.empty(b, 128, dtype=torch.float32)
+        ops.sm100_cutlass_mla_decode(
+            out,
+            lse,
+            q_nope,
+            q_pe,
+            cache,
+            cache_seqlens,
+            block_table,
+            workspace,
+            scale,
+            1,
+        )
+        return out, lse
+
+    out_contig, lse_contig = run(kv_contig)
+    out_view, lse_view = run(kv_view)
+
+    # Same data and same compute order; only addressing differs.
+    assert torch.equal(out_contig, out_view)
+    assert torch.equal(lse_contig, lse_view)

tests/kernels/attention/test_mla_cross_layer_kernel_equivalence.py

@@ -0,0 +1,566 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Bit-exact kernel equivalence for MLA decode/write kernels on the
+cross-layer (block-major) KV cache layout.
+
+The cross-layer layout carves each layer's per-block page out of a single
+unified slot, so the per-layer view has an inflated ``stride(0)`` (the full
+unified slot) and a non-zero storage offset. These tests confirm the MLA
+kernels behind the backends that opt in to the layout (FlashMLA dense,
+FlashInfer MLA dense, FlashMLA fp8 sparse, plus the ``concat_and_cache_mla``
+write) honor that strided view bit-identically to a contiguous per-layer
+cache, and that writes do not bleed into neighbouring layers' segments.
+"""
+
+import pytest
+import torch
+
+pytestmark = pytest.mark.skipif(
+    not torch.cuda.is_available(), reason="MLA cache kernels require CUDA"
+)
+
+
+def test_concat_and_cache_mla_into_unified_slot_view():
+    """concat_and_cache_mla must write correctly into a per-layer view whose
+    block stride is the full unified slot (block-major), with zero bleed into
+    the other layers' segments of the same slot."""
+    from vllm import _custom_ops as ops
+
+    torch.manual_seed(0)
+    dev = "cuda"
+    kv_lora_rank = 512
+    pe = 64
+    entry = kv_lora_rank + pe
+    page = 64
+    num_blocks = 32
+    ntok = 200
+
+    kv_c = torch.randn(ntok, kv_lora_rank, device=dev, dtype=torch.bfloat16)
+    k_pe = torch.randn(ntok, pe, device=dev, dtype=torch.bfloat16)
+    slot = torch.randperm(num_blocks * page, device=dev, dtype=torch.int64)[:ntok]
+    scale = torch.tensor(1.0, device=dev)
+
+    def write(cache):
+        ops.concat_and_cache_mla(kv_c, k_pe, cache, slot, "auto", scale)
+
+    # Contiguous per-layer reference: (num_blocks, page, entry).
+    ref = torch.zeros(num_blocks, page, entry, device=dev, dtype=torch.bfloat16)
+    write(ref)
+
+    # Unified slot holding three layer pages per block. Carve the middle
+    # layer's view (non-zero offset, block stride == full unified slot).
+    layer_page_elems = page * entry
+    n_layers = 3
+    unified_slot_elems = n_layers * layer_page_elems
+    big = torch.zeros(num_blocks, unified_slot_elems, device=dev, dtype=torch.bfloat16)
+    flat = big.view(-1)
+    offset = layer_page_elems  # middle layer
+    view = torch.as_strided(
+        flat,
+        size=(num_blocks, page, entry),
+        stride=(unified_slot_elems, entry, 1),
+        storage_offset=offset,
+    )
+    assert not view.is_contiguous()
+    assert view.stride(0) == unified_slot_elems
+    write(view)
+
+    # Bit-exact equivalence and zero bleed into the neighbour segments.
+    max_diff = (ref.float() - view.float()).abs().max().item()
+    assert max_diff == 0.0, f"max|Δ| = {max_diff}"
+
+    neighbour_lo = torch.as_strided(
+        flat, (num_blocks, layer_page_elems), (unified_slot_elems, 1), 0
+    )
+    neighbour_hi = torch.as_strided(
+        flat,
+        (num_blocks, layer_page_elems),
+        (unified_slot_elems, 1),
+        2 * layer_page_elems,
+    )
+    assert neighbour_lo.abs().max().item() == 0.0
+    assert neighbour_hi.abs().max().item() == 0.0
+
+
+def test_flashmla_dense_decode_unified_slot_view():
+    """FlashMLA dense decode (FLASHMLA backend, e.g. Kimi-K2-style dense MLA
+    on Hopper) must read a unified-slot block-major view bit-identically to a
+    contiguous per-layer cache."""
+    import vllm.v1.attention.ops.flashmla as fm
+
+    ok, reason = fm.is_flashmla_dense_supported()
+    if not ok:
+        pytest.skip(reason)
+
+    torch.manual_seed(0)
+    dev = "cuda"
+    dt = torch.bfloat16
+    head_dim = 576
+    hdv = 512
+    h_q = 128
+    page = 64
+    num_blocks = 64
+    bs = 4
+    n_layers = 3
+    layer = 1
+
+    q = torch.randn(bs, 1, h_q, head_dim, device=dev, dtype=dt) * 0.1
+    kv_data = torch.randn(num_blocks, page, 1, head_dim, device=dev, dtype=dt) * 0.1
+
+    # (A) contiguous per-layer reference.
+    cache_contiguous = kv_data.clone().contiguous()
+
+    # (B) unified slot: view one layer -> inflated stride(0), non-zero offset.
+    unified = (
+        torch.randn(num_blocks, n_layers, page, 1, head_dim, device=dev, dtype=dt) * 0.1
+    )
+    unified[:, layer].copy_(kv_data)
+    cache_view = unified[:, layer]
+    assert not cache_view.is_contiguous()
+    assert cache_view.stride(0) == n_layers * page * 1 * head_dim
+
+    max_blk = num_blocks // bs
+    block_table = torch.arange(num_blocks, device=dev, dtype=torch.int32).view(
+        bs, max_blk
+    )
+    cache_seqlens = torch.full((bs,), max_blk * page, device=dev, dtype=torch.int32)
+
+    def run(kc):
+        meta, num_splits = fm.get_mla_metadata()
+        out, _ = fm.flash_mla_with_kvcache(
+            q=q,
+            k_cache=kc,
+            block_table=block_table,
+            cache_seqlens=cache_seqlens,
+            head_dim_v=hdv,
+            tile_scheduler_metadata=meta,
+            num_splits=num_splits,
+            softmax_scale=head_dim**-0.5,
+            causal=True,
+        )
+        return out.clone().float()
+
+    out_ref = run(cache_contiguous)
+    out_view = run(cache_view)
+    assert torch.isfinite(out_ref).all()
+    assert out_ref.abs().max().item() > 0.0
+    assert (out_ref - out_view).abs().max().item() == 0.0
+
+
+def test_flashinfer_mla_dense_decode_unified_slot_view():
+    """FlashInfer MLA dense decode must read a unified-slot block-major view
+    (inflated stride(0), non-zero storage offset) bit-identically to a
+    contiguous per-layer cache."""
+    try:
+        from flashinfer.decode import trtllm_batch_decode_with_kv_cache_mla
+    except ImportError:
+        pytest.skip("flashinfer is not available")
+    from vllm.platforms import current_platform
+
+    if not current_platform.is_device_capability_family(100):
+        pytest.skip("FlashInfer trtllm-gen MLA requires sm100")
+
+    torch.manual_seed(0)
+    dev = "cuda"
+    dt = torch.bfloat16
+    kv_lora_rank = 512
+    qk_rope_head_dim = 64
+    qk_nope_head_dim = 128
+    head_dim = kv_lora_rank + qk_rope_head_dim  # 576
+    num_qo_heads = 128
+    page = 64
+    num_blocks = 64
+    bs = 4
+    n_layers = 3  # >1 so the per-layer view's block stride is inflated.
+    layer = 1
+
+    q = torch.randn(bs, 1, num_qo_heads, head_dim, device=dev, dtype=dt)
+    kv_data = torch.randn(num_blocks, 1, page, head_dim, device=dev, dtype=dt)
+
+    # (A) contiguous per-layer reference.
+    kv_contiguous = kv_data.clone().contiguous()
+
+    # (B) unified slot: block b of every layer packed together; view one layer
+    # -> stride(0) is n_layers x larger and storage offset is non-zero.
+    unified = torch.randn(num_blocks, n_layers, 1, page, head_dim, device=dev, dtype=dt)
+    unified[:, layer].copy_(kv_data)
+    kv_view = unified[:, layer]
+    assert not kv_view.is_contiguous()
+    assert kv_view.stride(0) == n_layers * 1 * page * head_dim
+
+    max_blk = num_blocks // bs
+    block_tables = torch.arange(num_blocks, device=dev, dtype=torch.int32).view(
+        bs, max_blk
+    )
+    seq_lens = torch.full((bs,), max_blk * page, device=dev, dtype=torch.int32)
+    ws = torch.empty(128 * 1024 * 1024, dtype=torch.int8, device=dev)
+    scale = head_dim**-0.5
+
+    def run(kv):
+        return trtllm_batch_decode_with_kv_cache_mla(
+            query=q,
+            kv_cache=kv,
+            workspace_buffer=ws,
+            qk_nope_head_dim=qk_nope_head_dim,
+            kv_lora_rank=kv_lora_rank,
+            qk_rope_head_dim=qk_rope_head_dim,
+            block_tables=block_tables,
+            seq_lens=seq_lens,
+            max_seq_len=int(seq_lens.max().item()),
+            bmm1_scale=scale,
+            bmm2_scale=1.0,
+        ).clone()
+
+    out_ref = run(kv_contiguous).float()
+    out_view = run(kv_view).float()
+    assert torch.isfinite(out_ref).all()
+    assert (out_ref - out_view).abs().max().item() == 0.0
+
+
+def test_flashmla_fp8_sparse_decode_unified_slot_view():
+    """FlashMLA fp8 sparse decode (DeepSeek V3.2/V4 DSA path) must read a
+    unified-slot block-major view bit-identically to a contiguous fp8_ds_mla
+    cache, with finite nonzero output."""
+    import vllm.v1.attention.ops.flashmla as fm
+
+    ok, reason = fm.is_flashmla_sparse_supported()
+    if not ok:
+        pytest.skip(reason)
+
+    torch.manual_seed(0)
+    dev = "cuda"
+    entry = 656  # fp8_ds_mla bytes per token
+    page = 64
+    num_blocks = 32
+    h_q = 128
+    head_dim = 576
+    hdv = 512
+    batch = 2
+    topk = 128
+    n_layers = 3
+    layer = 1
+
+    q = torch.randn(batch, 1, h_q, head_dim, device=dev, dtype=torch.bfloat16) * 0.1
+
+    # Structurally valid fp8 ds_mla payload: 512B fp8 + 16B f32 scales + 128B
+    # bf16 rope (random bytes corrupt the scale region and yield NaNs).
+    nope = (torch.randn(num_blocks, page, 1, 512, device=dev) * 0.1).to(
+        torch.float8_e4m3fn
+    )
+    scales = torch.ones(num_blocks, page, 1, 4, device=dev, dtype=torch.float32)
+    rope = (torch.randn(num_blocks, page, 1, 64, device=dev) * 0.1).to(torch.bfloat16)
+    payload = torch.cat(
+        [
+            nope.view(torch.uint8).view(num_blocks, page, 1, 512),
+            scales.view(torch.uint8).view(num_blocks, page, 1, 16),
+            rope.view(torch.uint8).view(num_blocks, page, 1, 128),
+        ],
+        dim=-1,
+    ).contiguous()
+    assert payload.shape[-1] == entry and payload.dtype == torch.uint8
+
+    # (A) contiguous reference.
+    cache_contiguous = payload.clone().contiguous()
+
+    # (B) unified slot: view one layer -> inflated stride(0), non-zero offset.
+    unified = torch.randint(
+        0, 256, (num_blocks, n_layers, page, 1, entry), device=dev, dtype=torch.uint8
+    )
+    unified[:, layer].copy_(payload)
+    cache_view = unified[:, layer]
+    assert not cache_view.is_contiguous()
+    assert cache_view.stride(0) == n_layers * page * 1 * entry
+
+    # Sparse indices: each batch uses its own disjoint blocks.
+    blocks_per_batch = num_blocks // batch
+    idx = torch.full((batch, 1, topk), -1, device=dev, dtype=torch.int32)
+    for b in range(batch):
+        slots: list[int] = []
+        for blk in range(b * blocks_per_batch, (b + 1) * blocks_per_batch):
+            slots.extend(blk * page + off for off in range(page))
+        slots_t = torch.tensor(slots[:topk], device=dev, dtype=torch.int32)
+        idx[b, 0, : slots_t.numel()] = slots_t
+
+    def run(kc):
+        meta, num_splits = fm.get_mla_metadata()
+        out, _ = fm.flash_mla_with_kvcache(
+            q=q,
+            k_cache=kc,
+            block_table=None,
+            cache_seqlens=None,
+            head_dim_v=hdv,
+            tile_scheduler_metadata=meta,
+            is_fp8_kvcache=True,
+            indices=idx,
+            softmax_scale=head_dim**-0.5,
+        )
+        return out.clone().float()
+
+    out_ref = run(cache_contiguous)
+    out_view = run(cache_view)
+    assert torch.isfinite(out_ref).all()
+    assert out_ref.abs().max().item() > 0.0
+    assert (out_ref - out_view).abs().max().item() == 0.0
+
+
+def test_indexer_k_quant_and_cache_into_unified_slot_view():
+    """indexer_k_quant_and_cache (DeepSeek V3.2/V4 DSA indexer K write) must
+    write correctly into a per-layer view whose block stride is the full
+    unified slot, with zero bleed into the other layers' segments."""
+    from vllm import _custom_ops as ops
+
+    torch.manual_seed(0)
+    dev = "cuda"
+    head_dim = 128
+    quant_block_size = 128
+    block_size = 64
+    num_blocks = 16
+    ntok = 100
+    # Indexer cache layout per token: head_dim fp8 bytes followed by
+    # head_dim * 4 / quant_block_size scale bytes.
+    cache_stride = head_dim + head_dim * 4 // quant_block_size
+
+    k = torch.randn(ntok, head_dim, device=dev, dtype=torch.bfloat16)
+    slot = torch.randperm(num_blocks * block_size, device=dev, dtype=torch.int64)[:ntok]
+
+    def write(cache):
+        ops.indexer_k_quant_and_cache(k, cache, slot, quant_block_size, "ue8m0")
+
+    # Contiguous per-layer reference.
+    ref = torch.zeros(
+        num_blocks, block_size, cache_stride, device=dev, dtype=torch.uint8
+    )
+    write(ref)
+
+    # Unified slot holding three layer pages per block; carve the middle one.
+    n_layers = 3
+    layer = 1
+    unified = torch.zeros(
+        num_blocks, n_layers, block_size, cache_stride, device=dev, dtype=torch.uint8
+    )
+    view = unified[:, layer]
+    assert not view.is_contiguous()
+    assert view.stride(0) == n_layers * block_size * cache_stride
+    write(view)
+
+    assert torch.equal(ref, view.contiguous())
+    # Zero bleed into the neighbour layers' segments.
+    assert unified[:, 0].abs().max().item() == 0
+    assert unified[:, 2].abs().max().item() == 0
+
+
+def test_flashattn_mla_dense_decode_unified_slot_view():
+    """FA3 decode (FLASH_ATTN_MLA backend) must read a unified-slot
+    block-major view bit-identically to a contiguous per-layer cache."""
+    try:
+        from vllm.vllm_flash_attn import flash_attn_varlen_func
+    except ImportError:
+        pytest.skip("vllm_flash_attn is not available")
+    from vllm.v1.attention.backends.fa_utils import flash_attn_supports_mla
+
+    if not flash_attn_supports_mla():
+        pytest.skip("FA3 MLA requires a Hopper device")
+
+    torch.manual_seed(0)
+    dev = "cuda"
+    dt = torch.bfloat16
+    kv_lora_rank = 512
+    rope_dim = 64
+    entry = kv_lora_rank + rope_dim  # 576
+    h_q = 16
+    page = 64
+    num_blocks = 64
+    bs = 4
+    n_layers = 3
+    layer = 1
+
+    q_pe = torch.randn(bs, h_q, rope_dim, device=dev, dtype=dt) * 0.1
+    q_nope = torch.randn(bs, h_q, kv_lora_rank, device=dev, dtype=dt) * 0.1
+    kv_data = torch.randn(num_blocks, page, entry, device=dev, dtype=dt) * 0.1
+
+    # (A) contiguous per-layer reference.
+    cache_contiguous = kv_data.clone().contiguous()
+
+    # (B) unified slot: view one layer -> inflated stride(0), non-zero offset.
+    unified = torch.randn(num_blocks, n_layers, page, entry, device=dev, dtype=dt) * 0.1
+    unified[:, layer].copy_(kv_data)
+    cache_view = unified[:, layer]
+    assert not cache_view.is_contiguous()
+    assert cache_view.stride(0) == n_layers * page * entry
+
+    max_blk = num_blocks // bs
+    block_table = torch.arange(num_blocks, device=dev, dtype=torch.int32).view(
+        bs, max_blk
+    )
+    seq_lens = torch.full((bs,), max_blk * page, device=dev, dtype=torch.int32)
+    cu_seqlens_q = torch.arange(bs + 1, device=dev, dtype=torch.int32)
+
+    def run(cache):
+        kv_c_cache = cache[..., :kv_lora_rank]
+        k_pe_cache = cache[..., kv_lora_rank:]
+        out = flash_attn_varlen_func(
+            q=q_pe,
+            k=k_pe_cache.unsqueeze(-2),  # Add head dim of 1
+            v=kv_c_cache.unsqueeze(-2),  # Add head dim of 1
+            q_v=q_nope,
+            max_seqlen_q=1,
+            cu_seqlens_q=cu_seqlens_q,
+            max_seqlen_k=int(seq_lens.max().item()),
+            seqused_k=seq_lens,
+            block_table=block_table,
+            softmax_scale=entry**-0.5,
+            causal=True,
+            fa_version=3,
+        )
+        return out.clone().float()
+
+    out_ref = run(cache_contiguous)
+    out_view = run(cache_view)
+    assert torch.isfinite(out_ref).all()
+    assert out_ref.abs().max().item() > 0.0
+    assert (out_ref - out_view).abs().max().item() == 0.0
+
+
+def test_flashmla_dense_fp8_decode_unified_slot_view():
+    """FlashMLA dense fp8 decode (FLASHMLA backend with quantized KV cache)
+    must read a unified-slot block-major view bit-identically to a contiguous
+    per-layer fp8 cache."""
+    import vllm.v1.attention.ops.flashmla as fm
+
+    ok, reason = fm.is_flashmla_dense_supported()
+    if not ok:
+        pytest.skip(reason)
+
+    torch.manual_seed(0)
+    dev = "cuda"
+    head_dim = 576
+    hdv = 512
+    h_q = 128
+    page = 64
+    num_blocks = 64
+    bs = 4
+    n_layers = 3
+    layer = 1
+
+    q = torch.randn(bs, 1, h_q, head_dim, device=dev, dtype=torch.bfloat16) * 0.1
+    kv_data = (torch.randn(num_blocks, page, head_dim, device=dev) * 0.1).to(
+        torch.float8_e4m3fn
+    )
+
+    # (A) contiguous per-layer reference.
+    cache_contiguous = kv_data.clone().contiguous()
+
+    # (B) unified slot: view one layer -> inflated stride(0), non-zero offset.
+    unified = (torch.randn(num_blocks, n_layers, page, head_dim, device=dev) * 0.1).to(
+        torch.float8_e4m3fn
+    )
+    unified[:, layer].copy_(kv_data)
+    cache_view = unified[:, layer]
+    assert not cache_view.is_contiguous()
+    assert cache_view.stride(0) == n_layers * page * head_dim
+
+    max_blk = num_blocks // bs
+    block_table = torch.arange(num_blocks, device=dev, dtype=torch.int32).view(
+        bs, max_blk
+    )
+    cache_seqlens = torch.full((bs,), max_blk * page, device=dev, dtype=torch.int32)
+    descale = torch.ones(1, device=dev, dtype=torch.float32)
+
+    def run(kc):
+        tile_md, num_splits = fm.get_mla_metadata_dense_fp8(cache_seqlens, h_q, 1)
+        out, _ = fm.flash_mla_with_kvcache_fp8(
+            q=q,
+            k_cache=kc.unsqueeze(-2),  # Add head dim of 1
+            block_table=block_table,
+            cache_seqlens=cache_seqlens,
+            head_dim_v=hdv,
+            tile_scheduler_metadata=tile_md,
+            num_splits=num_splits,
+            softmax_scale=head_dim**-0.5,
+            causal=True,
+            descale_q=descale,
+            descale_k=descale,
+        )
+        return out.clone().float()
+
+    out_ref = run(cache_contiguous)
+    out_view = run(cache_view)
+    assert torch.isfinite(out_ref).all()
+    assert out_ref.abs().max().item() > 0.0
+    assert (out_ref - out_view).abs().max().item() == 0.0
+
+
+def test_flashinfer_mla_dense_fp8_decode_unified_slot_view():
+    """FlashInfer MLA dense decode with an fp8 KV cache must read a
+    unified-slot block-major view bit-identically to a contiguous per-layer
+    cache."""
+    try:
+        from flashinfer.decode import trtllm_batch_decode_with_kv_cache_mla
+    except ImportError:
+        pytest.skip("flashinfer is not available")
+    from vllm.platforms import current_platform
+
+    if not current_platform.is_device_capability_family(100):
+        pytest.skip("FlashInfer trtllm-gen MLA requires sm100")
+
+    torch.manual_seed(0)
+    dev = "cuda"
+    kv_lora_rank = 512
+    qk_rope_head_dim = 64
+    qk_nope_head_dim = 128
+    head_dim = kv_lora_rank + qk_rope_head_dim  # 576
+    num_qo_heads = 128
+    page = 64
+    num_blocks = 64
+    bs = 4
+    n_layers = 3
+    layer = 1
+
+    # With a quantized KV cache the decode query is quantized to fp8 as well
+    # (trtllm-gen has no bf16-query x fp8-cache decode kernel).
+    q = (torch.randn(bs, 1, num_qo_heads, head_dim, device=dev) * 0.1).to(
+        torch.float8_e4m3fn
+    )
+    kv_data = (torch.randn(num_blocks, 1, page, head_dim, device=dev) * 0.1).to(
+        torch.float8_e4m3fn
+    )
+
+    # (A) contiguous per-layer reference.
+    kv_contiguous = kv_data.clone().contiguous()
+
+    # (B) unified slot: view one layer -> inflated stride(0), non-zero offset.
+    unified = (
+        torch.randn(num_blocks, n_layers, 1, page, head_dim, device=dev) * 0.1
+    ).to(torch.float8_e4m3fn)
+    unified[:, layer].copy_(kv_data)
+    kv_view = unified[:, layer]
+    assert not kv_view.is_contiguous()
+    assert kv_view.stride(0) == n_layers * 1 * page * head_dim
+
+    max_blk = num_blocks // bs
+    block_tables = torch.arange(num_blocks, device=dev, dtype=torch.int32).view(
+        bs, max_blk
+    )
+    seq_lens = torch.full((bs,), max_blk * page, device=dev, dtype=torch.int32)
+    ws = torch.empty(128 * 1024 * 1024, dtype=torch.int8, device=dev)
+    scale = head_dim**-0.5
+
+    def run(kv):
+        return trtllm_batch_decode_with_kv_cache_mla(
+            query=q,
+            kv_cache=kv,
+            workspace_buffer=ws,
+            qk_nope_head_dim=qk_nope_head_dim,
+            kv_lora_rank=kv_lora_rank,
+            qk_rope_head_dim=qk_rope_head_dim,
+            block_tables=block_tables,
+            seq_lens=seq_lens,
+            max_seq_len=int(seq_lens.max().item()),
+            bmm1_scale=scale,
+            bmm2_scale=1.0,
+        ).clone()
+
+    out_ref = run(kv_contiguous).float()
+    out_view = run(kv_view).float()
+    assert torch.isfinite(out_ref).all()
+    assert (out_ref - out_view).abs().max().item() == 0.0

tests/kernels/attention/test_triton_decode_attention.py

@@ -231,3 +231,95 @@ def test_decode_attention_fp8(B, L, H_Q, H_KV, D_QK, D_V, CACHE_SIZE, PAGE_SIZE)
 
     # FP8 tolerances match test_mla_backends.py test_backend_correctness.
     torch.testing.assert_close(o_ref, o_fp8, atol=5e-1, rtol=1e-2)
+
+
+@pytest.mark.parametrize(
+    "H_Q,H_KV,D_QK,D_V,is_mla",
+    [
+        (16, 1, 576, 512, True),  # MLA path (grouped kernel, v = trans(k))
+        (32, 8, 128, 128, False),  # GQA path (grouped kernel)
+        (32, 32, 128, 128, False),  # MHA path (normal kernel)
+    ],
+)
+@pytest.mark.parametrize("PAGE_SIZE", [16])
+def test_decode_attention_cross_layer_view(H_Q, H_KV, D_QK, D_V, is_mla, PAGE_SIZE):
+    """The kernel must honor the cache's page-dim stride, not assume pages are
+    packed back-to-back. A per-layer view into a cross-layer (block-major)
+    cache has stride(0) inflated by num_layers; outputs must match a
+    contiguous cache holding the same data exactly."""
+    B = 3
+    seq_len = 1027
+    CACHE_SIZE = 16384
+    NUM_LAYERS = 3
+    LAYER_IDX = 1
+    dtype = torch.bfloat16
+    sm_scale = 1.0 / (D_QK**0.5)
+    num_kv_splits = 8
+    num_pages = CACHE_SIZE // PAGE_SIZE
+
+    num_pages_per_batch = cdiv(seq_len, PAGE_SIZE)
+    req_to_page = torch.randint(
+        0, num_pages, (B, num_pages_per_batch), device=DEVICE_TYPE
+    )
+
+    q = torch.randn(B, H_Q, D_QK, dtype=dtype, device=DEVICE_TYPE)
+    b_seq_len = torch.full((B,), seq_len, device=DEVICE_TYPE)
+
+    # Reference: contiguous paged cache.
+    k_ref = torch.randn(
+        num_pages, PAGE_SIZE, H_KV, D_QK, dtype=dtype, device=DEVICE_TYPE
+    )
+    if is_mla:
+        v_ref = k_ref[..., :D_V]
+    else:
+        v_ref = torch.randn(
+            num_pages, PAGE_SIZE, H_KV, D_V, dtype=dtype, device=DEVICE_TYPE
+        )
+
+    # Cross-layer cache: all layers' pages for a block are adjacent. The
+    # per-layer view has the same shape as the contiguous cache but
+    # stride(0) is NUM_LAYERS x larger. Neighbor layers hold random data so
+    # any packed-pages addressing reads garbage rather than zeros.
+    k_xl = torch.randn(
+        num_pages, NUM_LAYERS, PAGE_SIZE, H_KV, D_QK, dtype=dtype, device=DEVICE_TYPE
+    )
+    k_view = k_xl[:, LAYER_IDX]
+    k_view.copy_(k_ref)
+    assert k_view.stride(0) == NUM_LAYERS * PAGE_SIZE * H_KV * D_QK
+    if is_mla:
+        v_view = k_view[..., :D_V]
+    else:
+        v_xl = torch.randn(
+            num_pages, NUM_LAYERS, PAGE_SIZE, H_KV, D_V, dtype=dtype, device=DEVICE_TYPE
+        )
+        v_view = v_xl[:, LAYER_IDX]
+        v_view.copy_(v_ref)
+
+    def run(k_buffer, v_buffer):
+        o = torch.zeros(B, H_Q, D_V, dtype=dtype, device=DEVICE_TYPE)
+        lse = torch.zeros(B, H_Q, dtype=dtype, device=DEVICE_TYPE)
+        attn_logits = torch.empty(
+            (B, H_Q, num_kv_splits, D_V + 1), dtype=torch.float32, device=DEVICE_TYPE
+        )
+        decode_attention_fwd(
+            q,
+            k_buffer,
+            v_buffer,
+            o,
+            lse,
+            req_to_page,
+            b_seq_len,
+            attn_logits,
+            num_kv_splits,
+            sm_scale,
+            PAGE_SIZE,
+            is_mla=is_mla,
+        )
+        return o, lse
+
+    o_ref, lse_ref = run(k_ref, v_ref)
+    o_xl, lse_xl = run(k_view, v_view)
+
+    # Same data and same compute order; only addressing differs.
+    assert torch.equal(o_ref, o_xl)
+    assert torch.equal(lse_ref, lse_xl)

tests/kernels/test_mhc_kernels.py

@@ -8,8 +8,8 @@ from vllm.model_executor.kernels.mhc.tilelang import (
     _tilelang_hc_prenorm_gemm,
     _torch_hc_prenorm_gemm,
 )
+from vllm.model_executor.layers.mhc import HAS_TILELANG_MHC
 from vllm.platforms import current_platform
-from vllm.utils.import_utils import has_tilelang
 from vllm.utils.torch_utils import set_random_seed
 
 DEVICE = current_platform.device_type
@@ -97,8 +97,8 @@ def hc_head_ref(
 
 
 @pytest.mark.skipif(
-    not (current_platform.is_cuda_alike() and has_tilelang()),
-    reason="CUDA or ROCm and tilelang required",
+    not HAS_TILELANG_MHC,
+    reason="TileLang MHC support required",
 )
 @pytest.mark.parametrize("num_tokens", [1, 4, 8, 128])
 @pytest.mark.parametrize("hidden_size", [4096, 7168])
@@ -150,8 +150,8 @@ def test_mhc_pre_tilelang(num_tokens, hidden_size, hc_mult):
 
 
 @pytest.mark.skipif(
-    not (current_platform.is_cuda_alike() and has_tilelang()),
-    reason="CUDA or ROCm and tilelang required",
+    not HAS_TILELANG_MHC,
+    reason="TileLang MHC support required",
 )
 @pytest.mark.parametrize(
     ("num_tokens", "hidden_size"),
@@ -190,8 +190,8 @@ def test_hc_prenorm_gemm_tilelang(num_tokens, hidden_size):
 
 
 @pytest.mark.skipif(
-    not (current_platform.is_cuda_alike() and has_tilelang()),
-    reason="CUDA or ROCm and tilelang required",
+    not HAS_TILELANG_MHC,
+    reason="TileLang MHC support required",
 )
 @pytest.mark.parametrize("num_tokens", [1, 4, 8, 128])
 @pytest.mark.parametrize("hidden_size", [4096, 7168])
@@ -217,8 +217,8 @@ def test_mhc_post_tilelang(num_tokens, hidden_size, hc_mult):
 
 
 @pytest.mark.skipif(
-    not (current_platform.is_cuda_alike() and has_tilelang()),
-    reason="CUDA or ROCm and tilelang required",
+    not HAS_TILELANG_MHC,
+    reason="TileLang MHC support required",
 )
 @pytest.mark.parametrize("num_tokens", [1, 4, 8, 128])
 @pytest.mark.parametrize("hidden_size", [4096, 7168])
@@ -324,8 +324,8 @@ def test_hc_head_triton(num_tokens, hidden_size, hc_mult):
 
 
 @pytest.mark.skipif(
-    not (current_platform.is_cuda_alike() and has_tilelang()),
-    reason="CUDA or ROCm and tilelang required",
+    not HAS_TILELANG_MHC,
+    reason="TileLang MHC support required",
 )
 @pytest.mark.parametrize("num_tokens", [1, 4, 8, 128])
 @pytest.mark.parametrize("hidden_size", [4096, 7168])

tests/lora/test_lora_manager.py

@@ -124,6 +124,7 @@ def test_replace_submodules(default_vllm_config, dist_init, dummy_model):
             max_lora_rank=8, max_cpu_loras=8, max_loras=8, lora_dtype=DEFAULT_DTYPE
         ),
         torch.device(DEVICES[0]),
+        default_vllm_config,
     )
     model = manager.model
     assert isinstance(model.get_submodule("dense1"), ColumnParallelLinearWithLoRA)
@@ -152,6 +153,7 @@ def test_wrap_replicated_linear_subclasses(default_vllm_config, dist_init, dummy
             max_lora_rank=8, max_cpu_loras=8, max_loras=8, lora_dtype=DEFAULT_DTYPE
         ),
         torch.device(DEVICES[0]),
+        default_vllm_config,
     )
 
     assert isinstance(
@@ -172,6 +174,7 @@ def test_wrap_gate_linear(default_vllm_config, dist_init, dummy_model):
             max_lora_rank=8, max_cpu_loras=8, max_loras=8, lora_dtype=DEFAULT_DTYPE
         ),
         torch.device(DEVICES[0]),
+        default_vllm_config,
     )
 
     assert isinstance(
@@ -219,6 +222,7 @@ def test_dedup_shared_module_across_paths(default_vllm_config, dist_init, dummy_
             max_lora_rank=8, max_cpu_loras=8, max_loras=8, lora_dtype=DEFAULT_DTYPE
         ),
         torch.device(DEVICES[0]),
+        default_vllm_config,
     )
 
     canonical = manager.model.get_submodule("moe.gate")
@@ -263,6 +267,7 @@ def test_lm_head_exempt_from_dedup(default_vllm_config, dist_init, dummy_model):
             max_lora_rank=8, max_cpu_loras=8, max_loras=8, lora_dtype=DEFAULT_DTYPE
         ),
         torch.device(DEVICES[0]),
+        default_vllm_config,
     )
 
     # lm_head's special handling still ran: logits_processor got wrapped
@@ -293,6 +298,7 @@ def test_skip_unsupported_matched_modules(default_vllm_config, dist_init, dummy_
             max_lora_rank=8, max_cpu_loras=8, max_loras=8, lora_dtype=DEFAULT_DTYPE
         ),
         torch.device(DEVICES[0]),
+        default_vllm_config,
     )
 
     # Should not crash and should keep unsupported matched modules unchanged.
@@ -325,6 +331,7 @@ def test_target_modules_fail_closed_on_unsupported_matched_modules(
                 target_modules=["dense1"],
             ),
             torch.device(DEVICES[0]),
+            default_vllm_config,
         )
 
 
@@ -374,6 +381,7 @@ def test_lora_model_manager(default_vllm_config, dist_init, dummy_model, device)
             max_lora_rank=8, max_cpu_loras=3, max_loras=2, lora_dtype=DEFAULT_DTYPE
         ),
         device=device,
+        vllm_config=default_vllm_config,
     )
     assert all(x is None for x in manager.lora_index_to_id)
     assert manager.add_adapter(model_lora1)
@@ -442,6 +450,7 @@ def test_lora_lru_cache_model_manager(
             max_lora_rank=8, max_cpu_loras=3, max_loras=2, lora_dtype=DEFAULT_DTYPE
         ),
         device=device,
+        vllm_config=default_vllm_config,
     )
     assert all(x is None for x in manager.lora_index_to_id)
     assert manager.add_adapter(model_lora1)
@@ -535,6 +544,7 @@ def test_lru_lora_model_manager(default_vllm_config, dist_init, dummy_model, dev
             max_lora_rank=8, max_cpu_loras=2, max_loras=2, lora_dtype=DEFAULT_DTYPE
         ),
         device=device,
+        vllm_config=default_vllm_config,
     )
     assert all(x is None for x in manager.lora_index_to_id)
 
@@ -642,9 +652,7 @@ def test_lru_lora_model_manager(default_vllm_config, dist_init, dummy_model, dev
 
 
 @pytest.mark.parametrize("device", DEVICES)
-def test_lru_cache_worker_adapter_manager(
-    default_vllm_config, dist_init, dummy_model, device, tmp_path
-):
+def test_lru_cache_worker_adapter_manager(dist_init, dummy_model, device, tmp_path):
     lora_config = LoRAConfig(
         max_lora_rank=8, max_cpu_loras=4, max_loras=4, lora_dtype=DEFAULT_DTYPE
     )
@@ -670,7 +678,7 @@ def test_lru_cache_worker_adapter_manager(
     worker_adapter_manager.max_num_seqs = 4
     worker_adapter_manager.max_num_batched_tokens = 2
 
-    worker_adapter_manager.create_lora_manager(dummy_model)
+    worker_adapter_manager.create_lora_manager(dummy_model, vllm_config)
 
     mapping = LoRAMapping([], [])
     worker_adapter_manager.set_active_adapters(
@@ -758,9 +766,7 @@ def test_lru_cache_worker_adapter_manager(
 
 
 @pytest.mark.parametrize("device", DEVICES)
-def test_worker_adapter_manager(
-    default_vllm_config, dist_init, dummy_model_gate_up, device, tmp_path
-):
+def test_worker_adapter_manager(dist_init, dummy_model_gate_up, device, tmp_path):
     # Should remove every LoRA not specified in the request.
     lora_config = LoRAConfig(
         max_lora_rank=8, max_cpu_loras=4, max_loras=4, lora_dtype=DEFAULT_DTYPE
@@ -774,7 +780,7 @@ def test_worker_adapter_manager(
 
     worker_adapter_manager = WorkerLoRAManager(vllm_config, device, EMBEDDING_MODULES)
     worker_adapter_manager.vocab_size = dummy_model_gate_up.unpadded_vocab_size
-    worker_adapter_manager.create_lora_manager(dummy_model_gate_up)
+    worker_adapter_manager.create_lora_manager(dummy_model_gate_up, vllm_config)
 
     dummy_lora_files = f"{tmp_path}/lora_adapter"
     os.makedirs(dummy_lora_files, exist_ok=True)
@@ -894,6 +900,7 @@ def test_packed_loras(default_vllm_config, dist_init, dummy_model_gate_up, devic
             max_lora_rank=8, max_cpu_loras=2, max_loras=2, lora_dtype=DEFAULT_DTYPE
         ),
         device=device,
+        vllm_config=default_vllm_config,
     )
     model = manager.model
 
@@ -944,6 +951,7 @@ def _test_target_modules(
     device: str,
     expected_lora: list[tuple[str, type]],
     expected_no_lora: list[tuple[str, type]],
+    vllm_config,
 ):
     """Create a LoRAModelManager and assert which modules have LoRA applied."""
     LoRAModelManager(
@@ -959,6 +967,7 @@ def _test_target_modules(
             target_modules=target_modules,
         ),
         device=device,
+        vllm_config=vllm_config,
     )
     for module_path, lora_cls in expected_lora:
         assert isinstance(model.get_submodule(module_path), lora_cls)
@@ -981,6 +990,7 @@ def test_target_modules_config(default_vllm_config, dist_init, dummy_model, devi
             ("dense2", RowParallelLinearWithLoRA),
             ("layer1.dense2", RowParallelLinearWithLoRA),
         ],
+        vllm_config=default_vllm_config,
     )
 
 
@@ -998,6 +1008,7 @@ def test_target_modules_multiple(default_vllm_config, dist_init, dummy_model, de
             ("layer1.dense2", RowParallelLinearWithLoRA),
         ],
         expected_no_lora=[],
+        vllm_config=default_vllm_config,
     )
 
 
@@ -1017,6 +1028,7 @@ def test_target_modules_none_uses_all(
             ("layer1.dense2", RowParallelLinearWithLoRA),
         ],
         expected_no_lora=[],
+        vllm_config=default_vllm_config,
     )
 
 
@@ -1036,4 +1048,5 @@ def test_target_modules_match_packed_runtime_modules(
             ("layer1.dense1", ColumnParallelLinearWithLoRA),
             ("layer1.dense2", RowParallelLinearWithLoRA),
         ],
+        vllm_config=default_vllm_config,
     )

tests/models/multimodal/generation/test_common.py

@@ -810,29 +810,6 @@ VLM_TEST_SETTINGS = {
         hf_output_post_proc=model_utils.minicpmv_trunc_hf_output,
         patch_hf_runner=model_utils.minicpmv_26_patch_hf_runner,
     ),
-    "minimax_vl_01": VLMTestInfo(
-        models=["MiniMaxAI/MiniMax-VL-01"],
-        prompt_formatter=lambda img_prompt: f"<beginning_of_sentence>user: {img_prompt} assistant:<end_of_sentence>",  # noqa: E501
-        img_idx_to_prompt=lambda _: "<image>",
-        test_type=(VLMTestType.IMAGE, VLMTestType.MULTI_IMAGE),
-        max_model_len=8192,
-        max_num_seqs=4,
-        dtype="bfloat16",
-        hf_output_post_proc=model_utils.minimax_vl_01_hf_output,
-        patch_hf_runner=model_utils.minimax_vl_01_patch_hf_runner,
-        auto_cls=AutoModelForImageTextToText,
-        marks=[
-            large_gpu_mark(min_gb=80),
-            # TODO: [ROCm] Fix pickle issue with ROCm spawn and tp>1
-            pytest.mark.skipif(
-                current_platform.is_rocm(),
-                reason=(
-                    "ROCm: Model too large for single GPU; "
-                    "multi-GPU blocked by HF _LazyConfigMapping pickle issue with spawn"
-                ),
-            ),
-        ],
-    ),
     "molmo": VLMTestInfo(
         models=["allenai/Molmo-7B-D-0924"],
         test_type=(VLMTestType.IMAGE, VLMTestType.MULTI_IMAGE),

tests/models/multimodal/generation/vlm_utils/model_utils.py

@@ -245,13 +245,6 @@ def minicpmv_trunc_hf_output(hf_output: RunnerOutput, model: str) -> RunnerOutpu
     return output_ids, output_str, out_logprobs
 
 
-def minimax_vl_01_hf_output(hf_output: RunnerOutput, model: str) -> RunnerOutput:
-    output_ids, output_str, out_logprobs = hf_output
-    if output_str.endswith("<end_of_sentence>"):
-        output_str = output_str.split("<end_of_sentence>")[0]
-    return output_ids, output_str, out_logprobs
-
-
 def ultravox_trunc_hf_output(hf_output: RunnerOutput, model: str) -> RunnerOutput:
     output_ids, output_str, out_logprobs = hf_output
 
@@ -1023,17 +1016,6 @@ def minicpmv_26_patch_hf_runner(hf_model: HfRunner) -> HfRunner:
     return hf_model
 
 
-def minimax_vl_01_patch_hf_runner(hf_model: HfRunner) -> HfRunner:
-    orig_generate = hf_model.model.generate
-
-    def _generate(self, *args, image_sizes=None, **kwargs):
-        return orig_generate(*args, decode_text=False, **kwargs)
-
-    hf_model.model.generate = types.MethodType(_generate, hf_model.model)
-
-    return hf_model
-
-
 def molmo_patch_hf_runner(hf_model: HfRunner) -> HfRunner:
     """Patches and returns an instance of the HfRunner to use for Molmo."""
     hf_processor = hf_model.processor

tests/models/multimodal/pooling/test_colqwen3_5.py

@@ -152,3 +152,21 @@ def test_colqwen3_5_relevance_ordering(
     dtype: str,
 ) -> None:
     _run_relevance_test(vllm_runner, model, dtype=dtype)
+
+
+def test_colqwen3_5_config_enables_bidirectional_attention() -> None:
+    """ColQwen3.5 retrieval must be served BIDIRECTIONAL (is_causal=False) so the
+    full_attention layers build with AttentionType.ENCODER_ONLY. This guards the
+    silent-causal regression (no GPU / model load needed)."""
+    from types import SimpleNamespace
+
+    from vllm.model_executor.models.config import (
+        MODELS_CONFIG_MAP,
+        ColQwen3_5Config,
+    )
+
+    assert MODELS_CONFIG_MAP["ColQwen3_5"] is ColQwen3_5Config
+
+    model_config = SimpleNamespace(hf_config=SimpleNamespace())
+    ColQwen3_5Config.verify_and_update_model_config(model_config)
+    assert model_config.hf_config.is_causal is False

tests/models/multimodal/processing/test_minimax_vl_01.py

@@ -1,113 +0,0 @@
-# SPDX-License-Identifier: Apache-2.0
-# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
-
-import pytest
-from PIL import Image
-
-from vllm.multimodal import MULTIMODAL_REGISTRY
-from vllm.multimodal.parse import ImageSize
-from vllm.multimodal.processing import BaseMultiModalProcessor
-
-from ....conftest import ImageTestAssets
-from ...utils import build_model_context
-
-
-@pytest.mark.parametrize("model_id", ["MiniMaxAI/MiniMax-VL-01"])
-@pytest.mark.parametrize("num_imgs", [1, 2])
-def test_processor_override(
-    image_assets: ImageTestAssets,
-    model_id: str,
-    num_imgs: int,
-):
-    ctx = build_model_context(
-        model_id,
-        mm_processor_kwargs=None,
-        limit_mm_per_prompt={"image": num_imgs},
-    )
-    processor = MULTIMODAL_REGISTRY.create_processor(ctx.model_config)
-    prompt = "<image>" * num_imgs
-    image = Image.new("RGB", size=(364, 364))
-    mm_data = {"image": [image] * num_imgs}
-
-    processed_inputs = processor(
-        prompt,
-        mm_items=processor.info.parse_mm_data(mm_data),
-        hf_processor_mm_kwargs={},
-    )
-    image_placeholders = processed_inputs["mm_placeholders"]["image"]
-
-    assert len(image_placeholders) == num_imgs
-
-
-def _validate_image_prompt_replacements_one(
-    processor: BaseMultiModalProcessor,
-    num_imgs: int,
-    failed_size_excs: list[tuple[ImageSize, Exception]],
-    image_size: ImageSize,
-) -> None:
-    prompt = "<image>" * num_imgs
-    image = Image.new("RGB", size=image_size)
-    mm_data = {"image": [image] * num_imgs}
-
-    try:
-        processed_inputs = processor(
-            prompt,
-            mm_items=processor.info.parse_mm_data(mm_data),
-            hf_processor_mm_kwargs={},
-        )
-
-        image_placeholders = processed_inputs["mm_placeholders"]["image"]
-        assert len(image_placeholders) == num_imgs
-
-    except Exception as exc:
-        failed_size_excs.append((image_size, exc))
-
-
-def _test_image_prompt_replacements(
-    processor,
-    *,
-    num_imgs: int,
-    image_sizes: list[ImageSize],
-) -> None:
-    failed_size_excs = list[tuple[ImageSize, Exception]]()
-
-    for size in image_sizes:
-        _validate_image_prompt_replacements_one(
-            processor, num_imgs, failed_size_excs, size
-        )
-
-    if failed_size_excs:
-        msg = "Found failing image sizes:" + "\n========\n".join(
-            f"[{size}]\n{exc}" for size, exc in failed_size_excs
-        )
-        raise AssertionError(msg)
-
-
-@pytest.mark.parametrize("model_id", ["MiniMaxAI/MiniMax-VL-01"])
-@pytest.mark.parametrize("num_imgs", [1, 2])
-def test_processor_prompt_replacements_regression(model_id, num_imgs):
-    ctx = build_model_context(
-        model_id,
-        mm_processor_kwargs=None,
-        limit_mm_per_prompt={"image": num_imgs},
-    )
-    processor = MULTIMODAL_REGISTRY.create_processor(ctx.model_config)
-
-    image_ratios = [
-        (171, 152),
-        (184, 161),
-        (198, 176),
-        (333, 296),
-        (369, 328),
-        (488, 183),
-        (2560, 1669),
-    ]
-    image_sizes = [
-        size for w, h in image_ratios for size in [ImageSize(w, h), ImageSize(h, w)]
-    ]
-
-    _test_image_prompt_replacements(
-        processor,
-        num_imgs=num_imgs,
-        image_sizes=image_sizes,
-    )

tests/models/registry.py

@@ -421,15 +421,6 @@ _TEXT_GENERATION_EXAMPLE_MODELS = {
         },
         trust_remote_code=True,
     ),
-    "MiniMaxForCausalLM": _HfExamplesInfo("MiniMaxAI/MiniMax-Text-01-hf"),
-    "MiniMaxText01ForCausalLM": _HfExamplesInfo(
-        "MiniMaxAI/MiniMax-Text-01",
-        trust_remote_code=True,
-        revision="a59aa9cbc53b9fb8742ca4e9e1531b9802b6fdc3",
-    ),
-    "MiniMaxM1ForCausalLM": _HfExamplesInfo(
-        "MiniMaxAI/MiniMax-M1-40k", trust_remote_code=True
-    ),
     "MiniMaxM2ForCausalLM": _HfExamplesInfo(
         "MiniMaxAI/MiniMax-M2",
         trust_remote_code=True,
@@ -1113,10 +1104,6 @@ _MULTIMODAL_EXAMPLE_MODELS = {
         "openbmb/MiniCPM-V-4_6",
         min_transformers_version="5.7.0",
     ),
-    "MiniMaxVL01ForConditionalGeneration": _HfExamplesInfo(
-        "MiniMaxAI/MiniMax-VL-01",
-        trust_remote_code=True,
-    ),
     "MiniMaxM3SparseForConditionalGeneration": _HfExamplesInfo(
         "MiniMaxAI/MiniMax-M3",
         trust_remote_code=True,

tests/models/test_initialization.py

@@ -98,11 +98,6 @@ def can_initialize(
         vllm_config.validate_block_size()
         return scheduler_kv_cache_config
 
-    if model_arch == "MiniMaxVL01ForConditionalGeneration":
-        pytest.skip(
-            "pickle error when loading `transformers.models.auto.CONFIG_MAPPING`"
-        )
-
     if model_arch == "MoonshotKimiaForCausalLM":
         pytest.skip(
             "Kimi-Audio requires SpeechToTextConfig "

tests/models/utils.py

@@ -507,7 +507,13 @@ def dummy_hf_overrides(
     # Only set MoE related config when the model has MoE layers.
     # Otherwise all models detected as MoE by _get_transformers_backend_cls.
     if model_arch_config.num_experts > 0:
-        num_experts_per_tok = 1 if model_arch == "Llama4ForConditionalGeneration" else 2
+        num_experts_per_tok = 2
+        if model_arch in (
+            "Llama4ForConditionalGeneration",
+            "Llama4ForCausalLM",
+            "EagleLlama4ForCausalLM",
+        ):
+            num_experts_per_tok = 1
         update_dict.update(
             {
                 "num_experts": num_experts,

tests/quantization/test_auto_round.py

@@ -43,8 +43,8 @@ MODELS = [
     pytest.param(
         "Intel/Qwen2-0.5B-Instruct-int4-sym-AutoRound",
         marks=pytest.mark.skipif(
-            not current_platform.is_cuda(),
-            reason="AWQ AutoRound model only supports CUDA backend for now.",
+            not (current_platform.is_cuda() or current_platform.is_xpu()),
+            reason="AWQ AutoRound model only supports CUDA/XPU backend for now.",
         ),
         id="auto_round:auto_awq",
     ),

tests/test_pooling_params.py

@@ -74,6 +74,29 @@ def test_embed_dimensions(model_info: EmbedModelInfo):
         pooling_params.verify(model_config)
 
 
+@dataclass()
+class MockMatryoshkaModelConfig:
+    pooler_config: PoolerConfig
+    is_matryoshka: bool = True
+    matryoshka_dimensions: list[int] | None = None
+    served_model_name: str = "mock-matryoshka-model"
+    embedding_size: int = 32
+
+
+def test_embed_dimensions_matryoshka_without_list_upper_bound():
+    task = "embed"
+    model_config = MockMatryoshkaModelConfig(
+        pooler_config=PoolerConfig(seq_pooling_type="CLS"),
+        matryoshka_dimensions=None,
+        embedding_size=32,
+    )
+
+    PoolingParams(task=task, dimensions=16).verify(model_config)
+
+    with pytest.raises(ValueError):
+        PoolingParams(task=task, dimensions=64).verify(model_config)
+
+
 @pytest.mark.parametrize("task", ["classify"])
 def test_classify(task):
     model_config = MockModelConfig(pooler_config=PoolerConfig(seq_pooling_type="CLS"))

tests/v1/attention/test_attention_backends_selection.py

@@ -6,10 +6,12 @@ from types import SimpleNamespace
 
 import pytest
 
+from vllm.model_executor.layers.mamba.linear.minimax_linear_attn import (
+    MiniMaxText01LinearAttention,
+)
 from vllm.model_executor.layers.mamba.mamba_mixer import MambaMixer
 from vllm.model_executor.layers.mamba.mamba_mixer2 import MambaMixer2
 from vllm.model_executor.layers.mamba.short_conv import ShortConv
-from vllm.model_executor.models.minimax_text_01 import MiniMaxText01LinearAttention
 from vllm.v1.attention.backends.linear_attn import LinearAttentionBackend
 from vllm.v1.attention.backends.mamba1_attn import Mamba1AttentionBackend
 from vllm.v1.attention.backends.mamba2_attn import Mamba2AttentionBackend

tests/v1/attention/test_gdn_metadata_builder.py

@@ -16,6 +16,7 @@ from tests.v1.attention.utils import (
     create_vllm_config,
 )
 from vllm.config import SpeculativeConfig
+from vllm.config.compilation import CUDAGraphMode
 from vllm.v1.attention.backends.gdn_attn import (
     GDNAttentionMetadata,
     GDNAttentionMetadataBuilder,
@@ -123,9 +124,15 @@ GDN_BUILD_TEST_CASES = {
 
 def _create_gdn_builder(
     num_speculative_tokens: int = 0,
+    full_cuda_graph: bool = False,
 ) -> GDNAttentionMetadataBuilder:
     """Create a GDNAttentionMetadataBuilder with minimal config."""
-    vllm_config = create_vllm_config(block_size=BLOCK_SIZE)
+    vllm_config = create_vllm_config(
+        model_name="Qwen/Qwen3.5-0.8B",
+        block_size=BLOCK_SIZE,
+    )
+    if full_cuda_graph:
+        vllm_config.compilation_config.cudagraph_mode = CUDAGraphMode.FULL_AND_PIECEWISE
     if num_speculative_tokens > 0:
         vllm_config.speculative_config = SpeculativeConfig(
             method="ngram",
@@ -189,3 +196,28 @@ def test_has_initial_state_after_reclassification():
     assert meta.has_initial_state is not None
     # req0 has context_lens = 65 - 1 = 64 > 0, so has_initial_state[0] = True
     assert meta.has_initial_state[0].item() is True
+
+
+def test_full_cudagraph_spec_metadata_uses_request_count():
+    """FULL cudagraph token padding must not pad request-indexed metadata."""
+    num_speculative_tokens = 3
+    builder = _create_gdn_builder(
+        num_speculative_tokens=num_speculative_tokens,
+        full_cuda_graph=True,
+    )
+    batch = BatchSpec(seq_lens=[80, 96], query_lens=[4, 4])
+    meta = _build(builder, batch, num_decode_draft_tokens=[3, 3])
+
+    assert meta.num_spec_decodes == batch.batch_size
+    assert meta.num_spec_decode_tokens == batch.compute_num_tokens()
+    assert meta.spec_state_indices_tensor is not None
+    assert meta.spec_state_indices_tensor.shape == (
+        batch.batch_size,
+        num_speculative_tokens + 1,
+    )
+    assert meta.spec_sequence_masks is not None
+    assert meta.spec_sequence_masks.shape == (batch.batch_size,)
+    assert meta.spec_query_start_loc is not None
+    assert meta.spec_query_start_loc.shape == (batch.batch_size + 1,)
+    assert meta.num_accepted_tokens is not None
+    assert meta.num_accepted_tokens.shape == (batch.batch_size,)

tests/v1/core/prefix_cache/test_partial_prefix_cache_primitives.py

@@ -0,0 +1,460 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+
+from collections.abc import Callable
+
+import pytest
+
+import vllm.v1.core.kv_cache_utils as kv_cache_utils
+from vllm.distributed.kv_events import BlockRemoved, BlockStored
+from vllm.sampling_params import SamplingParams
+from vllm.utils.hashing import sha256
+from vllm.v1.core.block_pool import BlockPool
+from vllm.v1.core.kv_cache_utils import (
+    BlockHash,
+    BlockHashListWithBlockSize,
+    KVCacheBlock,
+    get_request_block_hasher,
+    hash_block_tokens,
+    init_none_hash,
+)
+from vllm.v1.request import Request
+
+pytestmark = pytest.mark.cpu_test
+
+
+@pytest.fixture(autouse=True)
+def _auto_init_hash_fn():
+    init_none_hash(sha256)
+
+
+def make_request(
+    request_id: str,
+    prompt_token_ids: list[int],
+    hash_block_size: int,
+    hash_fn: Callable,
+) -> Request:
+    sampling_params = SamplingParams(max_tokens=17)
+    sampling_params.update_from_generation_config({}, eos_token_id=100)
+    return Request(
+        request_id=request_id,
+        prompt_token_ids=prompt_token_ids,
+        sampling_params=sampling_params,
+        pooling_params=None,
+        block_hasher=get_request_block_hasher(hash_block_size, hash_fn),
+    )
+
+
+def boundary_hash(req: Request, hash_block_size: int, num_tokens: int) -> BlockHash:
+    # Every boundary at a hash_block_size multiple is just the fine-grained
+    # chain hash ending there.
+    return req.block_hashes[num_tokens // hash_block_size - 1]
+
+
+def cache_full_block_and_partial_tail(
+    token_ids: list[int],
+    *,
+    enable_kv_cache_events: bool = False,
+) -> tuple[BlockPool, Request, list[KVCacheBlock], BlockHash]:
+    hash_block_size = 2
+    block_size = 6
+    kv_cache_group_id = 0
+    req = make_request("0", token_ids, hash_block_size, sha256)
+    pool = BlockPool(
+        num_gpu_blocks=3,
+        enable_caching=True,
+        hash_block_size=hash_block_size,
+        enable_kv_cache_events=enable_kv_cache_events,
+    )
+    blocks = pool.get_new_blocks(2)
+
+    pool.cache_full_blocks(
+        request=req,
+        blocks=blocks,
+        num_cached_blocks=0,
+        num_full_blocks=1,
+        block_size=block_size,
+        kv_cache_group_id=kv_cache_group_id,
+    )
+    partial_hash = boundary_hash(req, hash_block_size, len(token_ids))
+    assert pool.cache_partial_block(
+        request=req,
+        block=blocks[1],
+        num_tokens=len(token_ids),
+        kv_cache_group_id=kv_cache_group_id,
+        block_size=block_size,
+    )
+    return pool, req, blocks, partial_hash
+
+
+def test_boundary_hashes_reuse_fine_grained_chain():
+    hash_block_size = 2
+    block_size = 6
+    token_ids = [0, 0, 1, 1, 2, 2, 3, 3, 4, 4]
+    req = make_request("0", token_ids, hash_block_size, sha256)
+
+    coarse = BlockHashListWithBlockSize(req.block_hashes, hash_block_size, block_size)
+    # The block_size=6 full-block hash is the fine hash at the 6-token boundary,
+    # not a concatenation of the three fine hashes inside the block.
+    assert coarse[0] == req.block_hashes[6 // hash_block_size - 1]
+    assert coarse[0] != BlockHash(
+        req.block_hashes[0] + req.block_hashes[1] + req.block_hashes[2]
+    )
+    # A partial tail at 10 tokens is the fine hash at the 10-token boundary,
+    # which chains over the entire prefix.
+    tail_hash = boundary_hash(req, hash_block_size, 10)
+    assert tail_hash == req.block_hashes[4]
+    assert tail_hash == hash_block_tokens(sha256, req.block_hashes[3], token_ids[8:10])
+
+
+def test_cache_partial_block_kv_cache_events():
+    hash_block_size = 4
+    block_size = 12
+    kv_cache_group_id = 2
+
+    pool = BlockPool(
+        num_gpu_blocks=2,
+        enable_caching=True,
+        hash_block_size=hash_block_size,
+        enable_kv_cache_events=True,
+    )
+    req = make_request(
+        "req_partial_events",
+        prompt_token_ids=list(range(hash_block_size * 2)),
+        hash_block_size=hash_block_size,
+        hash_fn=sha256,
+    )
+
+    block = pool.get_new_blocks(1)[0]
+    partial_entry_hash = pool.cache_partial_block(
+        request=req,
+        block=block,
+        num_tokens=hash_block_size * 2,
+        kv_cache_group_id=kv_cache_group_id,
+        block_size=block_size,
+    )
+
+    events = pool.take_events()
+    assert len(events) == 1
+    stored_event = events[0]
+    assert isinstance(stored_event, BlockStored)
+    assert partial_entry_hash is not None
+    assert stored_event.block_hashes == [
+        kv_cache_utils.maybe_convert_block_hash(req.block_hashes[1])
+    ]
+    assert stored_event.parent_block_hash == kv_cache_utils.maybe_convert_block_hash(
+        req.block_hashes[0]
+    )
+    assert stored_event.token_ids == req.all_token_ids[hash_block_size:]
+    assert stored_event.block_size == 4
+    assert stored_event.group_idx == kv_cache_group_id
+
+    duplicate_entry_hash = pool.cache_partial_block(
+        request=req,
+        block=block,
+        num_tokens=hash_block_size * 2,
+        kv_cache_group_id=kv_cache_group_id,
+        block_size=block_size,
+    )
+    assert duplicate_entry_hash == partial_entry_hash
+    assert pool.take_events() == []
+
+    pool.free_blocks([block])
+    pool.get_new_blocks(1)
+    events = pool.take_events()
+    assert len(events) == 1
+    removed_event = events[0]
+    assert isinstance(removed_event, BlockRemoved)
+    assert removed_event.block_hashes == stored_event.block_hashes
+    assert removed_event.group_idx == kv_cache_group_id
+
+
+def test_partial_block_replacement_emits_remove_then_store_events():
+    hash_block_size = 2
+    block_size = 6
+    kv_cache_group_id = 0
+    req = make_request("0", [0, 0, 1, 1, 2, 2, 3, 3], hash_block_size, sha256)
+    pool = BlockPool(
+        num_gpu_blocks=3,
+        enable_caching=True,
+        hash_block_size=hash_block_size,
+        enable_kv_cache_events=True,
+    )
+    blocks = pool.get_new_blocks(2)
+
+    pool.cache_full_blocks(
+        request=req,
+        blocks=blocks,
+        num_cached_blocks=0,
+        num_full_blocks=1,
+        block_size=block_size,
+        kv_cache_group_id=kv_cache_group_id,
+    )
+    partial_hash_8 = boundary_hash(req, hash_block_size, 8)
+    assert pool.cache_partial_block(
+        request=req,
+        block=blocks[1],
+        num_tokens=8,
+        kv_cache_group_id=kv_cache_group_id,
+        block_size=block_size,
+    )
+    assert pool.get_cached_block(partial_hash_8, [kv_cache_group_id]) == [blocks[1]]
+    pool.take_events()
+
+    req.append_output_token_ids([4, 4])
+    partial_hash_10 = boundary_hash(req, hash_block_size, 10)
+    assert pool.cache_partial_block(
+        request=req,
+        block=blocks[1],
+        num_tokens=10,
+        kv_cache_group_id=kv_cache_group_id,
+        block_size=block_size,
+    )
+    events = pool.take_events()
+
+    assert len(events) == 2
+    removed_event, stored_event = events
+    assert isinstance(removed_event, BlockRemoved)
+    assert removed_event.block_hashes == [
+        kv_cache_utils.maybe_convert_block_hash(partial_hash_8)
+    ]
+    assert removed_event.group_idx == kv_cache_group_id
+    assert isinstance(stored_event, BlockStored)
+    assert stored_event.block_hashes == [
+        kv_cache_utils.maybe_convert_block_hash(partial_hash_10)
+    ]
+    assert stored_event.parent_block_hash == kv_cache_utils.maybe_convert_block_hash(
+        boundary_hash(req, hash_block_size, 8)
+    )
+    assert stored_event.token_ids == req.all_token_ids[8:10]
+    assert stored_event.block_size == hash_block_size
+    assert stored_event.group_idx == kv_cache_group_id
+    assert pool.get_cached_block(partial_hash_8, [kv_cache_group_id]) is None
+    assert pool.get_cached_block(partial_hash_10, [kv_cache_group_id]) == [blocks[1]]
+
+
+def test_later_request_hits_cached_partial_tail():
+    hash_block_size = 2
+    block_size = 6
+    kv_cache_group_id = 0
+    cached_token_ids = [0, 0, 1, 1, 2, 2, 3, 3, 4, 4]
+    req = make_request("0", cached_token_ids, hash_block_size, sha256)
+    pool = BlockPool(
+        num_gpu_blocks=3,
+        enable_caching=True,
+        hash_block_size=hash_block_size,
+    )
+    blocks = pool.get_new_blocks(2)
+
+    pool.cache_full_blocks(
+        request=req,
+        blocks=blocks,
+        num_cached_blocks=0,
+        num_full_blocks=1,
+        block_size=block_size,
+        kv_cache_group_id=kv_cache_group_id,
+    )
+    partial_hash_10 = boundary_hash(req, hash_block_size, 10)
+    assert pool.cache_partial_block(
+        request=req,
+        block=blocks[1],
+        num_tokens=10,
+        kv_cache_group_id=kv_cache_group_id,
+        block_size=block_size,
+    )
+
+    replay = make_request("1", cached_token_ids, hash_block_size, sha256)
+    replay_hash_10 = boundary_hash(replay, hash_block_size, 10)
+    assert replay_hash_10 == partial_hash_10
+    assert pool.get_cached_block(replay_hash_10, [kv_cache_group_id]) == [blocks[1]]
+
+    extended = make_request("2", cached_token_ids + [10], hash_block_size, sha256)
+    extended_hash_10 = boundary_hash(extended, hash_block_size, 10)
+    assert extended_hash_10 == partial_hash_10
+    assert pool.get_cached_block(extended_hash_10, [kv_cache_group_id]) == [blocks[1]]
+
+
+def test_cache_partial_block_uses_fine_grained_boundary_hash():
+    hash_block_size = 2
+    block_size = 6
+    kv_cache_group_id = 0
+    token_ids = [0, 0, 1, 1, 2, 2, 3, 3, 4, 4]
+    req = make_request("0", token_ids, hash_block_size, sha256)
+    pool = BlockPool(
+        num_gpu_blocks=3,
+        enable_caching=True,
+        hash_block_size=hash_block_size,
+    )
+    blocks = pool.get_new_blocks(2)
+
+    pool.cache_full_blocks(
+        request=req,
+        blocks=blocks,
+        num_cached_blocks=0,
+        num_full_blocks=1,
+        block_size=block_size,
+        kv_cache_group_id=kv_cache_group_id,
+    )
+
+    partial_entry_hash = pool.cache_partial_block(
+        request=req,
+        block=blocks[1],
+        num_tokens=10,
+        kv_cache_group_id=kv_cache_group_id,
+        block_size=block_size,
+    )
+    # The partial entry is keyed by the fine-grained hash at the 10-token
+    # boundary, regardless of the owning group's block_size.
+    expected = boundary_hash(req, hash_block_size, 10)
+    assert partial_entry_hash == kv_cache_utils.make_block_hash_with_group_id(
+        expected, kv_cache_group_id
+    )
+    assert pool.get_cached_block(expected, [kv_cache_group_id]) == [blocks[1]]
+
+
+def test_cache_partial_block_requires_hash_boundary():
+    hash_block_size = 2
+    block_size = 4
+    req = make_request("0", [0, 0, 1, 1], hash_block_size, sha256)
+    pool = BlockPool(
+        num_gpu_blocks=2,
+        enable_caching=True,
+        hash_block_size=hash_block_size,
+    )
+    block = pool.get_new_blocks(1)[0]
+
+    with pytest.raises(AssertionError):
+        pool.cache_partial_block(
+            request=req,
+            block=block,
+            num_tokens=3,
+            kv_cache_group_id=0,
+            block_size=block_size,
+        )
+
+
+def test_cache_partial_block_duplicate_checks_all_blocks_for_hash():
+    hash_block_size = 2
+    block_size = 4
+    kv_cache_group_id = 0
+    req = make_request("0", [0, 0, 1, 1], hash_block_size, sha256)
+    pool = BlockPool(
+        num_gpu_blocks=4,
+        enable_caching=True,
+        hash_block_size=hash_block_size,
+    )
+    blocks = pool.get_new_blocks(2)
+
+    first_entry_hash = pool.cache_partial_block(
+        request=req,
+        block=blocks[0],
+        num_tokens=2,
+        kv_cache_group_id=kv_cache_group_id,
+        block_size=block_size,
+    )
+    second_entry_hash = pool.cache_partial_block(
+        request=req,
+        block=blocks[1],
+        num_tokens=2,
+        kv_cache_group_id=kv_cache_group_id,
+        block_size=block_size,
+    )
+    assert first_entry_hash == second_entry_hash
+
+    duplicate_entry_hash = pool.cache_partial_block(
+        request=req,
+        block=blocks[1],
+        num_tokens=2,
+        kv_cache_group_id=kv_cache_group_id,
+        block_size=block_size,
+    )
+    assert duplicate_entry_hash == second_entry_hash
+    assert pool.cached_block_hashes_by_block == {}
+
+
+def test_reset_prefix_cache_clears_partial_entry_metadata():
+    pool, req, blocks, partial_hash_10 = cache_full_block_and_partial_tail(
+        [0, 0, 1, 1, 2, 2, 3, 3, 4, 4]
+    )
+    full_hash = BlockHashListWithBlockSize(req.block_hashes, 2, 6)[0]
+
+    assert pool.get_cached_block(full_hash, [0]) == [blocks[0]]
+    assert pool.get_cached_block(partial_hash_10, [0]) == [blocks[1]]
+
+    pool.free_blocks(blocks)
+    assert pool.reset_prefix_cache()
+
+    assert pool.get_cached_block(full_hash, [0]) is None
+    assert pool.get_cached_block(partial_hash_10, [0]) is None
+    assert pool.cached_block_hashes_by_block == {}
+
+
+def test_evict_cached_block_removes_full_hash_and_partial_entry():
+    pool, req, blocks, partial_hash_10 = cache_full_block_and_partial_tail(
+        [0, 0, 1, 1, 2, 2, 3, 3, 4, 4]
+    )
+    full_hash = BlockHashListWithBlockSize(req.block_hashes, 2, 6)[0]
+
+    assert pool.get_cached_block(full_hash, [0]) == [blocks[0]]
+    assert pool.get_cached_block(partial_hash_10, [0]) == [blocks[1]]
+
+    pool.evict_blocks({blocks[0].block_id, blocks[1].block_id})
+
+    assert pool.get_cached_block(full_hash, [0]) is None
+    assert pool.get_cached_block(partial_hash_10, [0]) is None
+    assert pool.cached_block_hashes_by_block == {}
+
+
+def test_partial_block_promotes_to_direct_full_block_hash():
+    hash_block_size = 2
+    block_size = 6
+    kv_cache_group_id = 0
+    token_ids = [0, 0, 1, 1, 2, 2, 3, 3, 4, 4]
+    req = make_request("0", token_ids, hash_block_size, sha256)
+    pool = BlockPool(
+        num_gpu_blocks=3,
+        enable_caching=True,
+        hash_block_size=hash_block_size,
+    )
+    blocks = pool.get_new_blocks(2)
+
+    pool.cache_full_blocks(
+        request=req,
+        blocks=blocks,
+        num_cached_blocks=0,
+        num_full_blocks=1,
+        block_size=block_size,
+        kv_cache_group_id=kv_cache_group_id,
+    )
+    partial_hash_10 = boundary_hash(req, hash_block_size, 10)
+    assert pool.cache_partial_block(
+        request=req,
+        block=blocks[1],
+        num_tokens=10,
+        kv_cache_group_id=kv_cache_group_id,
+        block_size=block_size,
+    )
+    assert pool.get_cached_block(partial_hash_10, [kv_cache_group_id]) == [blocks[1]]
+
+    req.append_output_token_ids([5, 5])
+    full_hashes = BlockHashListWithBlockSize(
+        req.block_hashes, hash_block_size, block_size
+    )
+    promoted_full_hash = full_hashes[1]
+    # The promoted full-block hash is the fine hash at the 12-token boundary,
+    # not a concatenation of the fine hashes inside the block.
+    assert promoted_full_hash == req.block_hashes[12 // hash_block_size - 1]
+    assert promoted_full_hash != BlockHash(
+        req.block_hashes[3] + req.block_hashes[4] + req.block_hashes[5]
+    )
+
+    pool.cache_full_blocks(
+        request=req,
+        blocks=blocks,
+        num_cached_blocks=1,
+        num_full_blocks=2,
+        block_size=block_size,
+        kv_cache_group_id=kv_cache_group_id,
+    )
+    assert pool.get_cached_block(promoted_full_hash, [kv_cache_group_id]) == [blocks[1]]
+    assert pool.get_cached_block(partial_hash_10, [kv_cache_group_id]) is None

tests/v1/core/test_kv_cache_utils.py

@@ -225,7 +225,7 @@ def test_kv_cache_block():
 
     # Test block hash setting and resetting
     block_hash = make_block_hash_with_group_id(BlockHash(b"abc"), 0)
-    block.block_hash = block_hash
+    block.set_block_hash(block_hash)
     assert block.block_hash == block_hash
 
     block.reset_hash()

tests/v1/core/test_prefix_caching.py

@@ -2003,7 +2003,7 @@ def test_maybe_evict_cached_block():
     assert len(pool.blocks) == len(block_hashes)
     # Manually add all blocks to cached_blocks
     for block, block_hash in zip(pool.blocks, block_hashes):
-        block.block_hash = block_hash
+        block.set_block_hash(block_hash)
         pool.cached_block_hash_to_block.insert(block_hash, block)
 
     block0, block1, block2, block3 = pool.blocks

tests/v1/kv_connector/unit/offloading_connector/test_events.py

@@ -0,0 +1,355 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+from unittest.mock import MagicMock
+
+import pytest
+import torch
+
+from tests.v1.kv_connector.unit.utils import create_vllm_config
+from vllm.config import KVEventsConfig, KVTransferConfig
+from vllm.distributed.kv_events import BlockRemoved, BlockStored
+from vllm.distributed.kv_transfer.kv_connector.v1.offloading.events import (
+    OffloadingEventGroupSpec,
+    OffloadingEventsTracker,
+)
+from vllm.distributed.kv_transfer.kv_connector.v1.offloading.scheduler import (
+    GroupOffloadConfig,
+)
+from vllm.v1.core.kv_cache_utils import BlockHash, maybe_convert_block_hash
+from vllm.v1.kv_cache_interface import (
+    FullAttentionSpec,
+    KVCacheConfig,
+    KVCacheGroupSpec,
+    KVCacheSpecKind,
+)
+from vllm.v1.kv_offload.base import (
+    OffloadingEvent,
+    OffloadingKVEventsConfig,
+    OffloadKey,
+    make_offload_key,
+)
+from vllm.v1.kv_offload.cpu.common import CPULoadStoreSpec
+from vllm.v1.kv_offload.tiering.spec import TieringOffloadingSpec
+
+_CPU_MEDIUM = CPULoadStoreSpec.medium()
+_FULL_ATTENTION_EVENT_SPEC = OffloadingEventGroupSpec(
+    kv_cache_spec_kind=KVCacheSpecKind.FULL_ATTENTION.value,
+    kv_cache_spec_sliding_window=None,
+)
+
+
+def _tracker(
+    *,
+    enable_kv_cache_events: bool = True,
+    self_describing_kv_events: bool = True,
+) -> OffloadingEventsTracker:
+    return OffloadingEventsTracker(
+        OffloadingKVEventsConfig(
+            enable_kv_cache_events=enable_kv_cache_events,
+            self_describing_kv_events=self_describing_kv_events,
+        )
+    )
+
+
+def _hash(i: int) -> BlockHash:
+    return BlockHash(str(i).encode())
+
+
+def _wire_hash(block_hash: BlockHash):
+    return maybe_convert_block_hash(block_hash)
+
+
+def _request(*, block_hashes: list[BlockHash], token_count: int):
+    req = MagicMock()
+    req.block_hashes = block_hashes
+    req.all_token_ids = list(range(1, token_count + 1))
+    req.lora_request = None
+    return req
+
+
+def _group_config(
+    *,
+    group_idx: int = 0,
+    block_size: int = 4,
+    block_size_factor: int = 1,
+    sliding_window_size_in_blocks: int | None = None,
+) -> GroupOffloadConfig:
+    return GroupOffloadConfig(
+        group_idx=group_idx,
+        gpu_block_size=block_size,
+        offloaded_block_size=block_size * block_size_factor,
+        hash_block_size_factor=block_size_factor,
+        sliding_window_size_in_blocks=sliding_window_size_in_blocks,
+        kv_event_group_spec=_FULL_ATTENTION_EVENT_SPEC,
+    )
+
+
+def _record_chunks(
+    tracker: OffloadingEventsTracker,
+    req,
+    group_config: GroupOffloadConfig,
+    num_chunks: int,
+) -> list[OffloadKey]:
+    keys: list[OffloadKey] = []
+    hbf = group_config.hash_block_size_factor
+    for chunk_idx in range(num_chunks):
+        tail_hash = req.block_hashes[(chunk_idx + 1) * hbf - 1]
+        assert tail_hash is not None
+        key = make_offload_key(tail_hash, group_config.group_idx)
+        tracker.record_store(req, group_config, chunk_idx, key)
+        keys.append(key)
+    return keys
+
+
+def _stored_event(keys: list[OffloadKey]) -> OffloadingEvent:
+    return OffloadingEvent(keys=keys, medium=_CPU_MEDIUM, removed=False)
+
+
+def _removed_event(keys: list[OffloadKey]) -> OffloadingEvent:
+    return OffloadingEvent(keys=keys, medium=_CPU_MEDIUM, removed=True)
+
+
+def test_take_events_publishes_routable_block_stored():
+    block_size = 4
+    tracker = _tracker()
+    group_config = _group_config(block_size=block_size)
+    req = _request(
+        block_hashes=[_hash(i) for i in range(6)],
+        token_count=block_size * 6,
+    )
+    keys = _record_chunks(tracker, req, group_config, num_chunks=6)
+
+    batch1 = list(tracker.take_events([_stored_event(keys[:3])]))
+    assert len(batch1) == 3
+
+    for i, event in enumerate(batch1):
+        assert isinstance(event, BlockStored)
+        assert event.medium == _CPU_MEDIUM
+        assert event.block_hashes == [_wire_hash(_hash(i))]
+        assert event.block_size == block_size
+        assert event.token_ids == list(
+            range(i * block_size + 1, (i + 1) * block_size + 1)
+        )
+        if i == 0:
+            assert event.parent_block_hash is None
+        else:
+            assert event.parent_block_hash == _wire_hash(_hash(i - 1))
+        assert event.lora_id is None
+        assert event.lora_name is None
+        assert event.extra_keys is None
+        assert event.group_idx == 0
+        assert event.kv_cache_spec_kind == KVCacheSpecKind.FULL_ATTENTION.value
+        assert event.kv_cache_spec_sliding_window is None
+
+    batch2 = list(tracker.take_events([_stored_event(keys[3:])]))
+    assert len(batch2) == 3
+    assert batch2[0].parent_block_hash == batch1[-1].block_hashes[-1]
+
+    assert len(tracker._pending_event_metadata) == 6
+
+
+def test_take_events_factor_gt_1_chunk_store_and_remove():
+    block_size = 4
+    block_size_factor = 3
+    tracker = _tracker()
+    group_config = _group_config(
+        block_size=block_size, block_size_factor=block_size_factor
+    )
+    req = _request(
+        block_hashes=[_hash(i) for i in range(6)],
+        token_count=block_size * block_size_factor * 2,
+    )
+    keys = _record_chunks(tracker, req, group_config, num_chunks=2)
+
+    stored = list(tracker.take_events([_stored_event(keys)]))
+    assert len(stored) == 2
+
+    expected_hashes = []
+    for chunk_idx, event in enumerate(stored):
+        assert isinstance(event, BlockStored)
+        expected_chunk_hashes = [
+            _wire_hash(_hash(i))
+            for i in range(
+                chunk_idx * block_size_factor,
+                (chunk_idx + 1) * block_size_factor,
+            )
+        ]
+        assert event.block_hashes == expected_chunk_hashes
+        assert event.block_size == block_size
+        assert len(event.token_ids) == block_size * block_size_factor
+        if chunk_idx == 0:
+            assert event.parent_block_hash is None
+        else:
+            assert event.parent_block_hash == _wire_hash(_hash(block_size_factor - 1))
+        expected_hashes.extend(expected_chunk_hashes)
+
+    assert len(tracker._pending_event_metadata) == 2
+
+    removed = list(tracker.take_events([_removed_event(keys)]))
+    assert len(removed) == 1
+    assert isinstance(removed[0], BlockRemoved)
+    assert removed[0].block_hashes == expected_hashes
+    assert removed[0].medium == _CPU_MEDIUM
+    assert removed[0].group_idx == 0
+    assert not tracker._pending_event_metadata
+
+
+def test_take_events_factor_gt_1_store_is_order_independent():
+    block_size_factor = 3
+    tracker = _tracker()
+    group_config = _group_config(block_size_factor=block_size_factor)
+    req = _request(
+        block_hashes=[_hash(i) for i in range(6)],
+        token_count=4 * block_size_factor * 2,
+    )
+    keys = _record_chunks(tracker, req, group_config, num_chunks=2)
+    unknown_key = make_offload_key(_hash(12345), 0)
+
+    events = list(tracker.take_events([_stored_event([keys[1], unknown_key, keys[0]])]))
+
+    assert len(events) == 3
+    chunk1, placeholder, chunk0 = events
+    assert [len(event.block_hashes) for event in events] == [3, 1, 3]
+    assert placeholder.block_size == 0
+    assert placeholder.token_ids == []
+    assert chunk0.parent_block_hash is None
+    assert chunk1.parent_block_hash == chunk0.block_hashes[-1]
+
+
+def test_take_events_opt_out_keeps_placeholders():
+    tracker = _tracker(self_describing_kv_events=False)
+    group_config = _group_config()
+    req = _request(block_hashes=[_hash(i) for i in range(3)], token_count=12)
+    keys = _record_chunks(tracker, req, group_config, num_chunks=3)
+
+    assert not tracker.self_describing_enabled
+    assert not tracker._pending_event_metadata
+
+    events = list(
+        tracker.take_events(
+            [
+                _stored_event(keys),
+                _removed_event(keys),
+            ]
+        )
+    )
+    assert len(events) == 4
+    for event in events[:3]:
+        assert isinstance(event, BlockStored)
+        assert event.block_size == 0
+        assert event.token_ids == []
+        assert event.parent_block_hash is None
+    assert isinstance(events[3], BlockRemoved)
+    assert len(events[3].block_hashes) == 3
+
+
+def test_record_store_skips_sliding_window_group():
+    tracker = _tracker()
+    group_config = _group_config(sliding_window_size_in_blocks=2)
+    req = _request(block_hashes=[_hash(i) for i in range(3)], token_count=12)
+    keys = _record_chunks(tracker, req, group_config, num_chunks=3)
+
+    assert not tracker._pending_event_metadata
+
+    events = list(tracker.take_events([_stored_event(keys[:1])]))
+    assert len(events) == 1
+    assert isinstance(events[0], BlockStored)
+    assert events[0].block_size == 0
+
+
+def test_take_events_groups_removed_hashes_by_kv_group():
+    tracker = _tracker()
+    group0_config = _group_config(group_idx=0, block_size_factor=2)
+    group1_config = _group_config(group_idx=1, block_size_factor=2)
+    req0 = _request(block_hashes=[_hash(0), _hash(1)], token_count=8)
+    req1 = _request(block_hashes=[_hash(10), _hash(11)], token_count=8)
+    key0 = _record_chunks(tracker, req0, group0_config, num_chunks=1)[0]
+    key1 = _record_chunks(tracker, req1, group1_config, num_chunks=1)[0]
+
+    removed = list(tracker.take_events([_removed_event([key0, key1])]))
+
+    assert len(removed) == 2
+    by_group = {event.group_idx: event.block_hashes for event in removed}
+    assert by_group == {
+        0: [_wire_hash(_hash(0)), _wire_hash(_hash(1))],
+        1: [_wire_hash(_hash(10)), _wire_hash(_hash(11))],
+    }
+
+
+def test_take_events_supports_restore_after_eviction():
+    block_size = 4
+    tracker = _tracker()
+    group_config = _group_config(block_size=block_size)
+    req = _request(block_hashes=[_hash(0)], token_count=block_size)
+    key = _record_chunks(tracker, req, group_config, num_chunks=1)[0]
+
+    first_store = list(tracker.take_events([_stored_event([key])]))
+    assert len(first_store) == 1
+    assert isinstance(first_store[0], BlockStored)
+    assert first_store[0].token_ids == [1, 2, 3, 4]
+
+    removed = list(tracker.take_events([_removed_event([key])]))
+    assert len(removed) == 1
+    assert isinstance(removed[0], BlockRemoved)
+    assert not tracker._pending_event_metadata
+
+    req.all_token_ids = [5, 6, 7, 8]
+    tracker.record_store(req, group_config, offload_block_idx=0, offload_key=key)
+
+    second_store = list(tracker.take_events([_stored_event([key])]))
+    assert len(second_store) == 1
+    assert isinstance(second_store[0], BlockStored)
+    assert second_store[0].token_ids == [5, 6, 7, 8]
+
+
+def test_reset_cache_clears_side_table():
+    tracker = _tracker()
+    group_config = _group_config()
+    req = _request(block_hashes=[_hash(i) for i in range(3)], token_count=12)
+    _record_chunks(tracker, req, group_config, num_chunks=3)
+
+    assert tracker._pending_event_metadata
+
+    tracker.reset()
+
+    assert not tracker._pending_event_metadata
+
+
+def test_tiering_rejects_self_describing_kv_events():
+    vllm_config = create_vllm_config(
+        block_size=4,
+        max_num_batched_tokens=16,
+        disable_hybrid_kv_cache_manager=False,
+    )
+    vllm_config.kv_transfer_config = KVTransferConfig(
+        kv_connector="OffloadingConnector",
+        kv_role="kv_both",
+        kv_connector_extra_config={
+            "spec_name": "TieringOffloadingSpec",
+            "cpu_bytes_to_use": 1 << 20,
+            "self_describing_kv_events": True,
+            "secondary_tiers": [{"type": "example"}],
+        },
+    )
+    vllm_config.kv_events_config = KVEventsConfig(
+        enable_kv_cache_events=True,
+        publisher="null",
+    )
+    kv_cache_config = KVCacheConfig(
+        num_blocks=0,
+        kv_cache_tensors=[],
+        kv_cache_groups=[
+            KVCacheGroupSpec(
+                ["layer"],
+                FullAttentionSpec(
+                    block_size=4,
+                    num_kv_heads=1,
+                    head_size=1,
+                    dtype=torch.float32,
+                ),
+            )
+        ],
+    )
+
+    with pytest.raises(ValueError, match="TieringOffloadingSpec"):
+        TieringOffloadingSpec(vllm_config, kv_cache_config)

tests/v1/kv_connector/unit/offloading_connector/test_scheduler.py

@@ -1,6 +1,5 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
-from collections.abc import Iterable
 from types import SimpleNamespace
 from unittest.mock import MagicMock
 
@@ -12,19 +11,16 @@ from tests.v1.kv_connector.unit.offloading_connector.utils import (
     to_keys,
 )
 from tests.v1.kv_connector.unit.utils import EOS_TOKEN_ID
-from vllm.distributed.kv_events import BlockRemoved, BlockStored
 from vllm.distributed.kv_transfer.kv_connector.v1.offloading.scheduler import (
     OffloadingConnectorScheduler,
     RequestOffloadState,
 )
-from vllm.v1.core.kv_cache_utils import BlockHash
 from vllm.v1.kv_cache_interface import (
     FullAttentionSpec,
     KVCacheGroupSpec,
     SlidingWindowSpec,
 )
 from vllm.v1.kv_offload.base import (
-    OffloadingEvent,
     OffloadingManager,
     OffloadPolicy,
     ReqContext,
@@ -146,31 +142,6 @@ def test_offloading_connector(request_runner, async_scheduling: bool):
     runner.connector_scheduler._maximal_prefix_lookup = lambda key, req_context: 1
     runner.run(decoded_tokens=[EOS_TOKEN_ID], expected_loaded=(3, 4, 5))
 
-    # test take_events
-    def to_hashes(int_hashes: list[int]) -> list[BlockHash]:
-        return [BlockHash(str(i).encode()) for i in int_hashes]
-
-    def take_events() -> Iterable[OffloadingEvent]:
-        yield OffloadingEvent(keys=to_keys([1, 2, 3]), medium="A", removed=False)
-        yield OffloadingEvent(keys=to_keys([4, 5, 6]), medium="B", removed=True)
-
-    runner.manager.take_events.side_effect = take_events
-    events = list(runner.scheduler_connector.take_events())
-    assert len(events) == 2
-    event = events[0]
-    assert isinstance(event, BlockStored)
-    assert event.block_hashes == to_hashes([1, 2, 3])
-    assert event.block_size == 0
-    assert event.medium == "A"
-    assert event.token_ids == []
-    assert event.parent_block_hash is None
-    assert event.lora_id is None
-    assert event.lora_name is None
-    event = events[1]
-    assert isinstance(event, BlockRemoved)
-    assert event.block_hashes == to_hashes([4, 5, 6])
-    assert event.medium == "B"
-
 
 @pytest.mark.parametrize("async_scheduling", [True, False])
 def test_request_preemption(request_runner, async_scheduling: bool):

tests/v1/kv_connector/unit/offloading_connector/utils.py

@@ -14,7 +14,12 @@ from tests.v1.kv_connector.unit.utils import (
     create_vllm_config,
 )
 from vllm import SamplingParams
-from vllm.config import KVTransferConfig, VllmConfig, set_current_vllm_config
+from vllm.config import (
+    KVEventsConfig,
+    KVTransferConfig,
+    VllmConfig,
+    set_current_vllm_config,
+)
 from vllm.distributed.kv_transfer.kv_connector.v1 import KVConnectorRole
 from vllm.distributed.kv_transfer.kv_connector.v1.offloading.common import (
     OffloadingConnectorMetadata,
@@ -198,6 +203,9 @@ class RequestRunner:
             "spec_module_path": "tests.v1.kv_connector.unit.offloading_connector.utils",  # noqa: E501
             # Preserve legacy behavior for tests; new opt-in tests override.
             "offload_prompt_only": False,
+            # Exercise the self-describing KV events path by default;
+            # opt-out tests override this to cover the legacy placeholders.
+            "self_describing_kv_events": True,
         }
         if block_size_factor > 1:
             extra_config["block_size"] = block_size * block_size_factor
@@ -209,6 +217,13 @@ class RequestRunner:
             kv_role="kv_both",
             kv_connector_extra_config=extra_config,
         )
+        vllm_config.kv_events_config = KVEventsConfig(
+            # Enable so the offloading events tracker is active, but use the
+            # null publisher: these tests drain take_events directly and a
+            # real ZMQ publisher would bind a port per test.
+            enable_kv_cache_events=True,
+            publisher="null",
+        )
 
         if kv_cache_groups is None:
             kv_cache_groups = [

tests/v1/kv_connector/unit/test_kv_cache_layout.py

@@ -1,10 +1,13 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
 
+import pytest
 
-def test_mla_backend_rejects_cross_layer_kv_cache():
-    """MLA backends return identity permutation (layers dim first)
-    to signal cross-layer KV cache is unsupported."""
+
+def test_mla_common_backend_rejects_cross_layer_kv_cache():
+    """MLACommonBackend defaults to the identity permutation (layers dim
+    first) so MLA backends whose decode kernels are not verified to honor
+    the cache's block-dim stride stay opted out of cross-layer KV cache."""
     from vllm.model_executor.layers.attention.mla_attention import (
         MLACommonBackend,
     )
@@ -19,6 +22,35 @@ def test_mla_backend_rejects_cross_layer_kv_cache():
     ) == (0, 1, 2)
 
 
+@pytest.mark.parametrize(
+    "backend_path",
+    [
+        "vllm.v1.attention.backends.mla.triton_mla.TritonMLABackend",
+        "vllm.v1.attention.backends.mla.cutlass_mla.CutlassMLABackend",
+        "vllm.v1.attention.backends.mla.flashattn_mla.FlashAttnMLABackend",
+        "vllm.v1.attention.backends.mla.flashmla.FlashMLABackend",
+        "vllm.v1.attention.backends.mla.flashinfer_mla.FlashInferMLABackend",
+    ],
+)
+def test_verified_mla_backends_support_cross_layer_kv_cache(backend_path):
+    """Backends whose decode kernels honor the cache's block-dim stride opt
+    in to the cross-layer layout with a non-identity permutation placing
+    num_blocks first in physical layout."""
+    module_path, name = backend_path.rsplit(".", 1)
+    backend = getattr(
+        pytest.importorskip(module_path, reason="backend deps unavailable"), name
+    )
+
+    stride_order = backend.get_kv_cache_stride_order(include_num_layers_dimension=True)
+    assert stride_order == (1, 0, 2, 3)
+    assert stride_order[0] != 0  # num_blocks first => cross-layer supported
+    assert backend.get_kv_cache_stride_order(include_num_layers_dimension=False) == (
+        0,
+        1,
+        2,
+    )
+
+
 def test_deepseek_v32_indexer_rejects_cross_layer_kv_cache():
     """DeepseekV32Indexer returns identity permutation (layers dim first)
     to signal cross-layer KV cache is unsupported."""

tests/v1/kv_connector/unit/test_multi_connector.py

@@ -219,6 +219,7 @@ def test_multi_example_connector_consistency():
         enforce_eager=True,
         gpu_memory_utilization=0.5,
         kv_transfer_config=kv_transfer_config,
+        async_scheduling=False,
     )
     # Run generation - this should trigger saving KV cache
     # Use a single prompt to avoid race conditions depending on the order of scheduling

tests/v1/kv_connector/unit/test_offloading_connector.py

@@ -138,6 +138,10 @@ def _wait_for_prefix_cache_reset(llm: LLM) -> None:
 
 
 def _latency_test(llm: LLM, subscriber: MockSubscriber | None):
+    # TODO: Reintroduce latency test on ROCm once MRV2 supports cross
+    # layer KV Cache. See https://github.com/vllm-project/vllm/pull/45947
+    if current_platform.is_rocm():
+        return
     sampling_params = SamplingParams(max_tokens=1)
 
     num_times_cpu_better_than_cold = 0

tests/v1/kv_offload/cpu/test_manager.py

@@ -294,25 +294,25 @@ def test_cpu_manager():
     # prepare store with no space ([2, 3] is being loaded)
     assert cpu_manager.prepare_store(to_keys([6, 7, 8]), _EMPTY_REQ_CTX) is None
 
-    # complete load [2, 3]
+    # complete load [2, 3]. Load changes the eviction list, making 2, 3 recent.
     cpu_manager.complete_load(to_keys([2, 3]), _EMPTY_REQ_CTX)
 
-    # prepare store [6, 7, 8] -> evicts [2, 3, 4] (oldest)
+    # prepare store [6, 7, 8] -> evicts [4, 5, 2] (oldest)
     prepare_store_output = cpu_manager.prepare_store(to_keys([6, 7, 8]), _EMPTY_REQ_CTX)
     verify_store_output(
         prepare_store_output,
         ExpectedPrepareStoreOutput(
             keys_to_store=[6, 7, 8],
-            store_block_ids=[3, 2, 1],
-            evicted_keys=[2, 3, 4],
+            store_block_ids=[1, 0, 3],
+            evicted_keys=[4, 5, 2],
         ),
     )
 
     # complete store [6, 7, 8]
     cpu_manager.complete_store(to_keys([6, 7, 8]), _EMPTY_REQ_CTX)
 
-    # touch [5, 6, 7] (move to end of LRU order)
-    cpu_manager.touch(to_keys([5, 6, 7]), _EMPTY_REQ_CTX)
+    # touch [3, 6, 7] (move to end of LRU order)
+    cpu_manager.touch(to_keys([3, 6, 7]), _EMPTY_REQ_CTX)
 
     # prepare store [7, 9] -> evicts [8] (oldest following previous touch)
     prepare_store_output = cpu_manager.prepare_store(to_keys([9]), _EMPTY_REQ_CTX)
@@ -320,7 +320,7 @@ def test_cpu_manager():
         prepare_store_output,
         ExpectedPrepareStoreOutput(
             keys_to_store=[9],
-            store_block_ids=[1],
+            store_block_ids=[3],
             evicted_keys=[8],
         ),
     )
@@ -335,7 +335,7 @@ def test_cpu_manager():
     verify_events(
         cpu_manager.take_events(),
         expected_stores=({3, 4, 5}, {6, 7, 8}),
-        expected_evictions=({2, 3, 4}, {8}),
+        expected_evictions=({4, 5, 2}, {8}),
     )
 
 

tests/v1/kv_offload/test_factory.py

@@ -0,0 +1,272 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""
+Unit tests for OffloadingSpecFactory.
+
+These tests verify:
+1. Pre-registration integrity — registered module paths can actually import
+   and yield correct OffloadingSpec subclasses (CI sentinel against file moves).
+2. End-to-end factory → spec construction with real configs.
+3. Downstream collaboration — build_metric_definitions delegation.
+4. Error paths — unregistered specs, missing config, duplicate registration.
+"""
+
+import pytest
+import torch
+
+from vllm.config import KVTransferConfig
+from vllm.v1.kv_cache_interface import (
+    FullAttentionSpec,
+    KVCacheConfig,
+    KVCacheGroupSpec,
+    KVCacheTensor,
+)
+from vllm.v1.kv_offload.base import OffloadingSpec
+from vllm.v1.kv_offload.cpu.spec import CPUOffloadingSpec
+from vllm.v1.kv_offload.factory import OffloadingSpecFactory
+from vllm.v1.kv_offload.tiering.spec import TieringOffloadingSpec
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture(autouse=True)
+def restore_registry():
+    """Save and restore OffloadingSpecFactory._registry between tests."""
+    original = dict(OffloadingSpecFactory._registry)
+    yield
+    OffloadingSpecFactory._registry = original
+
+
+def _make_vllm_config(
+    spec_name: str | None = "CPUOffloadingSpec",
+    cpu_bytes_to_use: int | None = None,
+    store_threshold: int = 0,
+    extra_config: dict | None = None,
+):
+    """Build a real VllmConfig with kv_transfer_config set for offloading."""
+    from vllm.config import (
+        CacheConfig,
+        DeviceConfig,
+        ModelConfig,
+        SchedulerConfig,
+        VllmConfig,
+    )
+
+    model_config = ModelConfig(
+        model="facebook/opt-125m",
+        trust_remote_code=True,
+        dtype="float16",
+        seed=42,
+    )
+    scheduler_config = SchedulerConfig(
+        max_num_seqs=16,
+        max_num_batched_tokens=64,
+        max_model_len=10000,
+        enable_chunked_prefill=True,
+        is_encoder_decoder=model_config.is_encoder_decoder,
+    )
+    cache_config = CacheConfig(
+        block_size=16,
+        gpu_memory_utilization=0.9,
+        cache_dtype="auto",
+        enable_prefix_caching=True,
+    )
+
+    cfg = extra_config or {}
+    if cpu_bytes_to_use is not None:
+        cfg["cpu_bytes_to_use"] = cpu_bytes_to_use
+    cfg["spec_name"] = spec_name
+    if store_threshold > 0:
+        cfg["store_threshold"] = store_threshold
+
+    kv_transfer_config = KVTransferConfig(
+        kv_connector="OffloadingConnector",
+        kv_role="kv_both",
+        kv_connector_extra_config=cfg,
+    )
+    return VllmConfig(
+        scheduler_config=scheduler_config,
+        model_config=model_config,
+        cache_config=cache_config,
+        kv_transfer_config=kv_transfer_config,
+        device_config=DeviceConfig("cpu"),
+    )
+
+
+def _make_kv_cache_config():
+    """Build a minimal KVCacheConfig with one KV cache tensor."""
+    num_blocks = 16
+    num_kv_heads = 1
+    head_size = 1
+    dtype = torch.float32
+    page_size = 2 * num_kv_heads * head_size * torch.finfo(dtype).bits // 8
+    kv_tensor = KVCacheTensor(
+        size=num_blocks * page_size, shared_by=["layer"], block_stride=0
+    )
+    return KVCacheConfig(
+        num_blocks=num_blocks,
+        kv_cache_tensors=[kv_tensor],
+        kv_cache_groups=[
+            KVCacheGroupSpec(
+                ["layer"],
+                FullAttentionSpec(
+                    block_size=16,
+                    num_kv_heads=num_kv_heads,
+                    head_size=head_size,
+                    dtype=dtype,
+                ),
+            )
+        ],
+    )
+
+
+# ---------------------------------------------------------------------------
+# Pre-registration integrity (CI sentinel)
+# ---------------------------------------------------------------------------
+
+
+def test_pre_registered_specs_can_be_imported():
+    """If someone moves cpu/spec.py but forgets to update factory.py, CI fails."""
+    for name in OffloadingSpecFactory._registry:
+        cls = OffloadingSpecFactory._registry[name]()
+        assert issubclass(cls, OffloadingSpec)
+
+
+def test_cpu_spec_registered():
+    """CPUOffloadingSpec is registered and importable."""
+    cls = OffloadingSpecFactory._registry["CPUOffloadingSpec"]()
+    assert cls is CPUOffloadingSpec
+
+
+def test_tiering_spec_registered():
+    """TieringOffloadingSpec is registered and importable."""
+    cls = OffloadingSpecFactory._registry["TieringOffloadingSpec"]()
+    assert cls is TieringOffloadingSpec
+
+
+# ---------------------------------------------------------------------------
+# Normal path — get_spec_cls
+# ---------------------------------------------------------------------------
+
+
+def test_get_spec_cls_returns_registered_class():
+    """Registered spec_name returns correct class."""
+    config = _make_vllm_config(spec_name="CPUOffloadingSpec")
+    spec_cls = OffloadingSpecFactory.get_spec_cls(config)
+    assert spec_cls is CPUOffloadingSpec
+
+
+def test_get_spec_cls_default_to_cpu():
+    """Default spec_name (absent from config) resolves to CPUOffloadingSpec."""
+    config = _make_vllm_config(spec_name=None)
+    config.kv_transfer_config.kv_connector_extra_config.pop("spec_name", None)
+    spec_cls = OffloadingSpecFactory.get_spec_cls(config)
+    assert spec_cls is CPUOffloadingSpec
+
+
+# ---------------------------------------------------------------------------
+# End-to-end — create_spec
+# ---------------------------------------------------------------------------
+
+
+def test_create_cpu_offloading_spec_end_to_end():
+    """Full factory → spec construction with real VllmConfig/KVCacheConfig.
+
+    Verifies:
+    - cpu_bytes_to_use validation and num_blocks calculation
+    - block_size % hash_block_size assertion
+    - spec instance is CPUOffloadingSpec
+    """
+    config = _make_vllm_config(cpu_bytes_to_use=65536)
+    kv_cache_config = _make_kv_cache_config()
+    spec = OffloadingSpecFactory.create_spec(config, kv_cache_config)
+    assert isinstance(spec, CPUOffloadingSpec)
+    assert spec.num_blocks > 0
+
+
+# ---------------------------------------------------------------------------
+# Dynamic import via spec_module_path
+# ---------------------------------------------------------------------------
+
+
+def test_dynamic_load_via_spec_module_path():
+    """External spec loaded via spec_module_path.
+
+    This is how external projects (e.g., llm-d-kv-cache SharedStorageOffloadingSpec)
+    integrate with vLLM without being pre-registered in the factory.
+    The fallback path: registry miss → spec_module_path → importlib.import_module.
+    """
+    config = _make_vllm_config(spec_name="CPUOffloadingSpec")
+    # Delete from registry to force the dynamic import path
+    del OffloadingSpecFactory._registry["CPUOffloadingSpec"]
+    # spec_name not in registry → falls through to spec_module_path
+    config.kv_transfer_config.kv_connector_extra_config["spec_module_path"] = (
+        "vllm.v1.kv_offload.cpu.spec"
+    )
+    spec_cls = OffloadingSpecFactory.get_spec_cls(config)
+    assert spec_cls is CPUOffloadingSpec
+
+
+# ---------------------------------------------------------------------------
+# Error paths
+# ---------------------------------------------------------------------------
+
+
+def test_unregistered_spec_without_module_path_raises():
+    """spec_name not in registry + no spec_module_path → ValueError."""
+    config = _make_vllm_config(spec_name="NonexistentSpec")
+    with pytest.raises(ValueError, match="Unsupported spec type"):
+        OffloadingSpecFactory.get_spec_cls(config)
+
+    # create_spec should also fail (calls get_spec_cls internally)
+    kv_cache_config = _make_kv_cache_config()
+    with pytest.raises(ValueError, match="Unsupported spec type"):
+        OffloadingSpecFactory.create_spec(config, kv_cache_config)
+
+
+def test_cpu_spec_missing_cpu_bytes_to_use_raises():
+    """CPUOffloadingSpec requires cpu_bytes_to_use → Exception."""
+    config = _make_vllm_config(cpu_bytes_to_use=None)
+    config.kv_transfer_config.kv_connector_extra_config.pop("cpu_bytes_to_use", None)
+    kv_cache_config = _make_kv_cache_config()
+    with pytest.raises(Exception, match="cpu_bytes_to_use must be specified"):
+        OffloadingSpecFactory.create_spec(config, kv_cache_config)
+
+
+def test_duplicate_registration_raises():
+    """register_spec with existing name → ValueError."""
+    with pytest.raises(ValueError, match="is already registered"):
+        OffloadingSpecFactory.register_spec(
+            "CPUOffloadingSpec", "some.module", "SomeClass"
+        )
+
+
+# ---------------------------------------------------------------------------
+# Downstream collaboration — build_metric_definitions
+# ---------------------------------------------------------------------------
+
+
+def test_build_metric_definitions_empty_below_threshold():
+    """store_threshold < 2 → only base metric (no stores_skipped)."""
+    from vllm.v1.kv_offload.cpu.common import CPUOffloadingMetrics
+
+    config = _make_vllm_config(store_threshold=1)
+    spec_cls = OffloadingSpecFactory.get_spec_cls(config)
+    metrics = spec_cls.build_metric_definitions(
+        config.kv_transfer_config.kv_connector_extra_config
+    )
+    assert CPUOffloadingMetrics.STORES_SKIPPED not in metrics
+
+
+def test_build_metric_definitions_returns_counter_at_threshold():
+    """store_threshold >= 2 → returns stores_skipped counter definition."""
+    from vllm.v1.kv_offload.cpu.common import CPUOffloadingMetrics
+
+    config = _make_vllm_config(store_threshold=2)
+    spec_cls = OffloadingSpecFactory.get_spec_cls(config)
+    metrics = spec_cls.build_metric_definitions(
+        config.kv_transfer_config.kv_connector_extra_config
+    )
+    assert CPUOffloadingMetrics.STORES_SKIPPED in metrics

tests/v1/kv_offload/tiering/test_factory.py

@@ -0,0 +1,152 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""
+Unit tests for SecondaryTierFactory.
+
+These tests verify:
+1. Pre-registration integrity — registered tier module paths can import
+   and yield correct SecondaryTierManager subclasses (CI sentinel).
+2. Multi-tier creation via factory with correct tier_type propagation.
+3. Error paths — missing tier_type, unknown tier_type, duplicate registration.
+"""
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from vllm.v1.kv_offload.tiering.base import SecondaryTierManager
+from vllm.v1.kv_offload.tiering.example.manager import ExampleSecondaryTierManager
+from vllm.v1.kv_offload.tiering.factory import SecondaryTierFactory
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture(autouse=True)
+def restore_registry():
+    """Save and restore SecondaryTierFactory._registry between tests."""
+    original = dict(SecondaryTierFactory._registry)
+    yield
+    SecondaryTierFactory._registry = original
+
+
+def _make_mock_args():
+    """Build common mock args for create_secondary_tier."""
+    return MagicMock(), MagicMock()  # primary_kv_view, offloading_spec
+
+
+# ---------------------------------------------------------------------------
+# Pre-registration integrity (CI sentinel)
+# ---------------------------------------------------------------------------
+
+
+def test_pre_registered_tiers_can_be_imported():
+    """CI sentinel: example/fs/obj paths must import and yield SecondaryTierManager."""
+    for tier_type in SecondaryTierFactory._registry:
+        cls = SecondaryTierFactory._registry[tier_type]()
+        assert issubclass(cls, SecondaryTierManager)
+
+
+def test_example_tier_registered():
+    """Example tier is registered."""
+    cls = SecondaryTierFactory._registry["example"]()
+    assert cls is ExampleSecondaryTierManager
+
+
+# ---------------------------------------------------------------------------
+# Normal path — create_secondary_tier
+# ---------------------------------------------------------------------------
+
+
+def test_create_tier_from_registry():
+    """Registered tier_type creates instance with correct tier_type."""
+    primary_kv_view, offloading_spec = _make_mock_args()
+    tier_config = {"type": "example"}
+
+    tier = SecondaryTierFactory.create_secondary_tier(
+        tier_config, primary_kv_view, offloading_spec
+    )
+
+    assert isinstance(tier, SecondaryTierManager)
+    assert tier.tier_type == "example"
+
+
+def test_create_multiple_tiers():
+    """Multiple tier configs can be created with correct tier_types."""
+    primary_kv_view, offloading_spec = _make_mock_args()
+    configs = [
+        {"type": "example", "custom_param": 1},
+        {"type": "example", "custom_param": 2},
+    ]
+
+    tiers = [
+        SecondaryTierFactory.create_secondary_tier(
+            cfg.copy(), primary_kv_view, offloading_spec
+        )
+        for cfg in configs
+    ]
+
+    assert len(tiers) == 2
+    assert all(tier.tier_type == "example" for tier in tiers)
+    assert all(isinstance(tier, ExampleSecondaryTierManager) for tier in tiers)
+
+
+def test_register_new_tier_type():
+    """Verify that new tier types can be registered and created.
+
+    This is how external projects add custom secondary tiers
+    (e.g., llm-d FS backend was upstreamed as "fs" tier via this mechanism).
+    """
+    # Register a new tier type (reuse example manager for simplicity)
+    SecondaryTierFactory.register_tier(
+        "custom_tier",
+        "vllm.v1.kv_offload.tiering.example.manager",
+        "ExampleSecondaryTierManager",
+    )
+
+    primary_kv_view, offloading_spec = _make_mock_args()
+    tier = SecondaryTierFactory.create_secondary_tier(
+        {"type": "custom_tier", "custom_param": 99},
+        primary_kv_view,
+        offloading_spec,
+    )
+
+    assert tier.tier_type == "custom_tier"
+    assert isinstance(tier, ExampleSecondaryTierManager)
+
+
+# ---------------------------------------------------------------------------
+# Error paths
+# ---------------------------------------------------------------------------
+
+
+def test_missing_tier_type_raises():
+    """tier_config without 'type' → ValueError."""
+    primary_kv_view, offloading_spec = _make_mock_args()
+    tier_config: dict[str, str] = {}
+
+    with pytest.raises(ValueError, match="must include 'type'"):
+        SecondaryTierFactory.create_secondary_tier(
+            tier_config, primary_kv_view, offloading_spec
+        )
+
+
+def test_unknown_tier_type_raises():
+    """Unrecognized tier_type → ValueError with supported types list."""
+    primary_kv_view, offloading_spec = _make_mock_args()
+    tier_config = {"type": "nonexistent_tier"}
+
+    with pytest.raises(
+        ValueError,
+        match=r"Unknown secondary tier type.*Supported types:",
+    ):
+        SecondaryTierFactory.create_secondary_tier(
+            tier_config, primary_kv_view, offloading_spec
+        )
+
+
+def test_duplicate_registration_raises():
+    """register_tier with existing type → ValueError."""
+    with pytest.raises(ValueError, match="is already registered"):
+        SecondaryTierFactory.register_tier("example", "some.module", "SomeClass")

tests/v1/kv_offload/tiering/test_tiering_offloading.py

@@ -295,6 +295,8 @@ class TestTieringOffloadingManager:
         self.manager.prepare_store(blocks, _CTX)
         self.manager.complete_store(blocks, _CTX, success=True)
         self._simulate_on_schedule_end()
+        # for secondary tiers to drain jobs, so primary tier's blocks are evictable.
+        self._simulate_on_schedule_end()
 
         self.secondary_tier1.touch = MagicMock(wraps=self.secondary_tier1.touch)
         self.secondary_tier2.touch = MagicMock(wraps=self.secondary_tier2.touch)
@@ -303,7 +305,7 @@ class TestTieringOffloadingManager:
         self.manager.touch(blocks, _CTX)
 
         # Verify touch was called on primary tier (check LRU order)
-        primary_keys = list(self.primary_tier._policy.blocks.keys())
+        primary_keys = list(self.primary_tier._policy.evictable_blocks.keys())
         assert primary_keys[-3:] == list(reversed(blocks))
 
         # Verify touch was propagated to all secondary tiers

tools/pre_commit/mypy.py

@@ -25,8 +25,6 @@ import regex as re
 # from "skip" to "silent", remove its directory from SEPARATE_GROUPS.
 SEPARATE_GROUPS = [
     "tests",
-    # v0 related
-    "vllm/lora",
 ]
 
 # TODO(woosuk): Include the code from Megatron and HuggingFace.

vllm/distributed/kv_transfer/kv_connector/v1/offloading/events.py

@@ -0,0 +1,286 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Self-describing KV cache events for the offloading connector.
+
+The OffloadingManager identifies an offloaded chunk only by its OffloadKey,
+so its raw events carry no token ids, parent hash, or block size.
+:class:`OffloadingEventsTracker` snapshots each chunk's full ``BlockStored``
+payload while the ``Request`` is alive and publishes stores as block-granular
+payloads: a chunk event may carry multiple constituent per-block hashes, and
+evictions fan out to the same hashes. Chunks overlapping a non-chunk-aligned
+shared prefix re-announce the shared hashes once per chunk; consumers are
+expected to deduplicate (reference-count) repeated store/remove announcements
+of the same hash. Opt-in via
+``kv_connector_extra_config["self_describing_kv_events"]``; inert unless
+KV cache events are enabled. See the PR description for the full design.
+"""
+
+from collections.abc import Iterable
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any, NamedTuple
+
+from vllm.distributed.kv_events import BlockRemoved, BlockStored, KVCacheEvent
+from vllm.logger import init_logger
+from vllm.v1.core.kv_cache_utils import BlockHash, maybe_convert_block_hash
+from vllm.v1.kv_cache_interface import (
+    KVCacheGroupSpec,
+    get_kv_cache_spec_kind,
+    get_kv_cache_spec_sliding_window,
+)
+from vllm.v1.kv_offload.base import (
+    OffloadingEvent,
+    OffloadingKVEventsConfig,
+    OffloadKey,
+    get_offload_block_hash,
+    get_offload_group_idx,
+)
+from vllm.v1.request import Request
+
+if TYPE_CHECKING:
+    from vllm.distributed.kv_transfer.kv_connector.v1.offloading.scheduler import (
+        GroupOffloadConfig,
+    )
+
+logger = init_logger(__name__)
+
+
+class OffloadingEventGroupSpec(NamedTuple):
+    kv_cache_spec_kind: str | None
+    kv_cache_spec_sliding_window: int | None
+
+
+def get_offloading_event_group_spec(
+    kv_cache_group: KVCacheGroupSpec,
+) -> OffloadingEventGroupSpec:
+    kv_cache_spec = kv_cache_group.kv_cache_spec
+    return OffloadingEventGroupSpec(
+        kv_cache_spec_kind=get_kv_cache_spec_kind(kv_cache_spec).value,
+        kv_cache_spec_sliding_window=get_kv_cache_spec_sliding_window(kv_cache_spec),
+    )
+
+
+@dataclass(slots=True)
+class _OffloadEventMetadata:
+    """BlockStored payload snapshot for one OffloadKey, captured at store
+    time and kept until the matching eviction event. ``medium`` is forwarded
+    from the OffloadingEvent."""
+
+    # The chunk's constituent block hashes; the last one is the OffloadKey.
+    block_hashes: tuple[BlockHash, ...]
+    parent_block_hash: BlockHash | None
+    token_ids: tuple[int, ...]
+    block_size: int
+    lora_id: int | None
+    lora_name: str | None
+    # Deferred: needs the same incremental curr_mm_idx handling as GPU events.
+    extra_keys: tuple[tuple[Any, ...] | None, ...] | None
+    group_idx: int
+    kv_cache_spec: OffloadingEventGroupSpec
+
+
+class OffloadingEventsTracker:
+    """Tracks offloaded chunks' KV event payloads from store to eviction.
+
+    The scheduler calls :meth:`record_store` from ``_build_store_jobs``
+    while the ``Request`` is available, and routes the manager's raw
+    :class:`OffloadingEvent` stream through :meth:`take_events`. All state
+    is bounded by the CPU pool capacity and cleared by :meth:`reset`.
+    """
+
+    def __init__(self, config: OffloadingKVEventsConfig):
+        self.config = config
+        self.self_describing_enabled = (
+            config.enable_kv_cache_events and config.self_describing_kv_events
+        )
+
+        # OffloadKey -> payload snapshot, kept until the eviction event so
+        # BlockRemoved can fan out. Bounded: one entry per offloaded chunk.
+        self._pending_event_metadata: dict[OffloadKey, _OffloadEventMetadata] = {}
+
+    def record_store(
+        self,
+        req: Request,
+        group_config: "GroupOffloadConfig",
+        offload_block_idx: int,
+        offload_key: OffloadKey,
+    ) -> None:
+        """Snapshot the KV cache event payload for one offloaded chunk.
+
+        No-op when self-describing event capture is disabled or for
+        sliding-window / SSM groups, which keep the legacy placeholder payload.
+        """
+        if not self.self_describing_enabled:
+            return
+        if group_config.sliding_window_size_in_blocks is not None:
+            return
+        meta = self._build_event_metadata(req, group_config, offload_block_idx)
+        self._pending_event_metadata[offload_key] = meta
+
+    def take_events(self, events: Iterable[OffloadingEvent]) -> Iterable[KVCacheEvent]:
+        """Translate raw OffloadingEvents into self-describing KV events.
+
+        Complete metadata is available only for full-attention groups when
+        the tracker is enabled. Other shapes retain the legacy placeholder
+        payload so consumers can ignore them.
+
+        Yields:
+            ``BlockStored`` or ``BlockRemoved`` events corresponding to
+            the underlying :class:`OffloadingEvent` stream.
+        """
+        for event in events:
+            if event.removed:
+                yield from self._take_removed_event(event)
+            else:
+                yield from self._take_stored_event(event)
+
+    def reset(self) -> None:
+        """Drop all tracked state; pending payloads are stale after a
+        manager cache reset."""
+        self._pending_event_metadata.clear()
+
+    def _build_event_metadata(
+        self,
+        req: Request,
+        group_config: "GroupOffloadConfig",
+        offload_block_idx: int,
+    ) -> _OffloadEventMetadata:
+        """Build the payload snapshot for one offloaded chunk: its
+        constituent per-block hashes, the whole chunk's tokens, and the
+        per-block ``block_size``."""
+        hbf = group_config.hash_block_size_factor
+        assert hbf > 0
+        assert offload_block_idx >= 0
+        # per-block token count (= the GPU/hash block size)
+        sub_block_size = group_config.offloaded_block_size // hbf
+        # chunk c covers hash-blocks [c*hbf, (c+1)*hbf); its tail block's hash
+        # is the chunk's OffloadKey.
+        first_hash_idx = offload_block_idx * hbf
+        last_hash_idx = first_hash_idx + hbf
+        assert first_hash_idx >= 0
+        assert last_hash_idx <= len(req.block_hashes)
+        chunk_hashes: list[BlockHash] = []
+        for block_hash in req.block_hashes[first_hash_idx:last_hash_idx]:
+            assert block_hash is not None
+            chunk_hashes.append(block_hash)
+        assert len(chunk_hashes) == hbf
+
+        if group_config.sliding_window_size_in_blocks is not None:
+            # record_store filters these out before calling this helper.
+            raise AssertionError("self-describing events only support full attention")
+
+        parent_block_hash: BlockHash | None
+        if first_hash_idx == 0:
+            parent_block_hash = None
+        else:
+            parent_block_hash = req.block_hashes[first_hash_idx - 1]
+            assert parent_block_hash is not None
+
+        tok_start = offload_block_idx * group_config.offloaded_block_size
+        tok_end = tok_start + group_config.offloaded_block_size
+        assert tok_end <= len(req.all_token_ids)
+        token_ids = tuple(req.all_token_ids[tok_start:tok_end])
+
+        lora_id: int | None = None
+        lora_name: str | None = None
+        if req.lora_request is not None:
+            lora_id = req.lora_request.adapter_id
+            lora_name = req.lora_request.name
+
+        return _OffloadEventMetadata(
+            block_hashes=tuple(chunk_hashes),
+            parent_block_hash=parent_block_hash,
+            token_ids=token_ids,
+            block_size=sub_block_size,
+            lora_id=lora_id,
+            lora_name=lora_name,
+            extra_keys=None,
+            group_idx=group_config.group_idx,
+            kv_cache_spec=group_config.kv_event_group_spec,
+        )
+
+    def _placeholder_stored(self, key: OffloadKey, medium: str) -> BlockStored:
+        return BlockStored(
+            block_hashes=[
+                maybe_convert_block_hash(BlockHash(get_offload_block_hash(key)))
+            ],
+            parent_block_hash=None,
+            token_ids=[],
+            lora_id=None,
+            block_size=0,
+            medium=medium,
+            lora_name=None,
+            group_idx=get_offload_group_idx(key),
+        )
+
+    def _take_stored_event(self, event: OffloadingEvent) -> Iterable[KVCacheEvent]:
+        # Metadata is read, NOT popped: the entry must survive until the
+        # eviction event so BlockRemoved can fan out to the same hashes.
+        # Events are self-contained (own parent), so key order is free.
+        for key in event.keys:
+            meta = self._pending_event_metadata.get(key)
+            if meta is None:
+                if self.self_describing_enabled:
+                    # Expected for unsupported shapes; warn once only.
+                    logger.warning_once(
+                        "OffloadingEventsTracker: no event metadata for "
+                        "offload key during BlockStored emission; emitting a "
+                        "placeholder payload. Expected for non-full-attention "
+                        "groups; otherwise indicates a missing populate path."
+                    )
+                yield self._placeholder_stored(key, event.medium)
+                continue
+
+            yield BlockStored(
+                block_hashes=list(
+                    maybe_convert_block_hash(h) for h in meta.block_hashes
+                ),
+                parent_block_hash=(
+                    maybe_convert_block_hash(meta.parent_block_hash)
+                    if meta.parent_block_hash is not None
+                    else None
+                ),
+                token_ids=list(meta.token_ids),
+                block_size=meta.block_size,
+                lora_id=meta.lora_id,
+                medium=event.medium,
+                lora_name=meta.lora_name,
+                extra_keys=(
+                    list(meta.extra_keys) if meta.extra_keys is not None else None
+                ),
+                group_idx=meta.group_idx,
+                kv_cache_spec_kind=meta.kv_cache_spec.kv_cache_spec_kind,
+                kv_cache_spec_sliding_window=(
+                    meta.kv_cache_spec.kv_cache_spec_sliding_window
+                ),
+            )
+
+    def _take_removed_event(self, event: OffloadingEvent) -> Iterable[KVCacheEvent]:
+        # Keep group_idx unambiguous if a manager batch spans groups.
+        by_group: dict[int, list] = {}
+        for key in event.keys:
+            meta = self._pending_event_metadata.pop(key, None)
+            if meta is not None:
+                group_idx = meta.group_idx
+                by_group.setdefault(group_idx, []).extend(
+                    maybe_convert_block_hash(h) for h in meta.block_hashes
+                )
+            else:
+                if self.self_describing_enabled:
+                    logger.warning_once(
+                        "OffloadingEventsTracker: no event metadata for "
+                        "offload key during BlockRemoved emission; emitting a "
+                        "placeholder removal. Expected if the matching store "
+                        "used the legacy placeholder payload; otherwise "
+                        "indicates missing store metadata."
+                    )
+                group_idx = get_offload_group_idx(key)
+                by_group.setdefault(group_idx, []).append(
+                    maybe_convert_block_hash(BlockHash(get_offload_block_hash(key)))
+                )
+
+        for group_idx, hashes in by_group.items():
+            yield BlockRemoved(
+                block_hashes=hashes,
+                medium=event.medium,
+                group_idx=group_idx,
+            )

vllm/distributed/kv_transfer/kv_connector/v1/offloading/scheduler.py

@@ -5,7 +5,7 @@ from dataclasses import dataclass, field
 from itertools import islice
 from typing import Any, NamedTuple
 
-from vllm.distributed.kv_events import BlockRemoved, BlockStored, KVCacheEvent
+from vllm.distributed.kv_events import KVCacheEvent
 from vllm.distributed.kv_transfer.kv_connector.utils import yield_req_data
 from vllm.distributed.kv_transfer.kv_connector.v1.base import KVConnectorMetadata
 from vllm.distributed.kv_transfer.kv_connector.v1.offloading.common import (
@@ -14,6 +14,11 @@ from vllm.distributed.kv_transfer.kv_connector.v1.offloading.common import (
     ReqId,
     TransferJob,
 )
+from vllm.distributed.kv_transfer.kv_connector.v1.offloading.events import (
+    OffloadingEventGroupSpec,
+    OffloadingEventsTracker,
+    get_offloading_event_group_spec,
+)
 from vllm.distributed.kv_transfer.kv_connector.v1.offloading.metrics import (
     OffloadingConnectorStats,
     _TransferMetricName,
@@ -36,7 +41,6 @@ from vllm.v1.kv_offload.base import (
     OffloadPolicy,
     ReqContext,
     RequestOffloadingContext,
-    get_offload_block_hash,
     make_offload_key,
 )
 from vllm.v1.outputs import KVConnectorOutput
@@ -69,6 +73,9 @@ class GroupOffloadConfig(NamedTuple):
     gpu_block_size: int
     offloaded_block_size: int
     hash_block_size_factor: int
+    # KV cache spec metadata propagated onto emitted BlockStored events so
+    # KV-aware consumers can classify and filter the group.
+    kv_event_group_spec: OffloadingEventGroupSpec
     # None below means full attention
     sliding_window_size_in_blocks: int | None
     # Number of this group's offloaded blocks per full-attention alignment
@@ -200,6 +207,9 @@ class SchedulerOffloadConfig(NamedTuple):
                     alignment_block_count=_alignment_block_count(
                         gpu_block_size * spec.block_size_factor, sw
                     ),
+                    kv_event_group_spec=get_offloading_event_group_spec(
+                        spec.kv_cache_config.kv_cache_groups[idx]
+                    ),
                     is_eagle_group=idx in eagle_groups,
                 )
                 for idx, gpu_block_size in enumerate(spec.gpu_block_size)
@@ -361,6 +371,8 @@ class OffloadingConnectorScheduler:
         # be freed before a request finishes).
         self._block_id_to_pending_jobs: dict[int, set[int]] = {}
 
+        self._events_tracker = OffloadingEventsTracker(spec.kv_events_config)
+
     def _generate_job_id(self) -> int:
         job_id = self._job_counter
         self._job_counter += 1
@@ -934,6 +946,11 @@ class OffloadingConnectorScheduler:
                         continue
 
                     offloaded_block_idx = start_block_idx + idx
+
+                    self._events_tracker.record_store(
+                        req, group_config, offloaded_block_idx, offload_key
+                    )
+
                     gpu_block_idx = offloaded_block_idx * block_size_factor
                     for i in range(block_size_factor):
                         block_id = block_ids[gpu_block_idx + i]
@@ -1184,25 +1201,17 @@ class OffloadingConnectorScheduler:
         return False, None
 
     def take_events(self) -> Iterable[KVCacheEvent]:
-        """Take the KV cache events from the connector.
+        """Drain pending KV cache events.
 
-        Returns:
-            A list of KV cache events.
+        Complete metadata is available only when self-describing KV events
+        are enabled, and only for full-attention groups. Other shapes retain
+        the previous placeholder payload so consumers can ignore them.
+
+        Yields:
+            ``BlockStored`` or ``BlockRemoved`` events corresponding to
+            the underlying :class:`OffloadingEvent` stream.
         """
-        for event in self.manager.take_events():
-            block_hashes = [get_offload_block_hash(key) for key in event.keys]
-            if event.removed:
-                yield BlockRemoved(block_hashes=block_hashes, medium=event.medium)
-            else:
-                yield BlockStored(
-                    block_hashes=block_hashes,
-                    parent_block_hash=None,
-                    token_ids=[],
-                    lora_id=None,
-                    block_size=0,
-                    medium=event.medium,
-                    lora_name=None,
-                )
+        yield from self._events_tracker.take_events(self.manager.take_events())
 
     def reset_cache(self) -> None:
         """Reset the offloading manager cache, evicting all stored blocks."""
@@ -1238,6 +1247,10 @@ class OffloadingConnectorScheduler:
         self._jobs.clear()
         self._block_id_to_pending_jobs.clear()
 
+        # The manager pool is empty; pending event payloads and announced
+        # reference counts are stale.
+        self._events_tracker.reset()
+
         # Note: _current_batch_jobs_to_flush is intentionally NOT cleared.
         # The load flush IDs collected above must be delivered to workers.
         if self._blocks_being_loaded is not None:

vllm/entrypoints/anthropic/api_router.py

@@ -61,7 +61,7 @@ def translate_error_response(response: ErrorResponse) -> JSONResponse:
 async def create_messages(request: AnthropicMessagesRequest, raw_request: Request):
     handler = messages(raw_request)
     if handler is None:
-        base_server = raw_request.app.state.openai_serving_tokenization
+        base_server = raw_request.app.state.serving_tokenization
         error = base_server.create_error_response(
             NotImplementedError("The model does not support Messages API")
         )
@@ -107,7 +107,7 @@ async def create_messages(request: AnthropicMessagesRequest, raw_request: Reques
 async def count_tokens(request: AnthropicCountTokensRequest, raw_request: Request):
     handler = messages(raw_request)
     if handler is None:
-        base_server = raw_request.app.state.openai_serving_tokenization
+        base_server = raw_request.app.state.serving_tokenization
         error = base_server.create_error_response(
             NotImplementedError("The model does not support Messages API")
         )

vllm/entrypoints/openai/api_server.py

@@ -34,7 +34,7 @@ from vllm.entrypoints.openai.models.serving import OpenAIServingModels
 from vllm.entrypoints.serve.elastic_ep.middleware import ScalingMiddleware
 from vllm.entrypoints.serve.render.serving import OpenAIServingRender
 from vllm.entrypoints.serve.sagemaker.api_router import sagemaker_standards_bootstrap
-from vllm.entrypoints.serve.tokenize.serving import OpenAIServingTokenization
+from vllm.entrypoints.serve.tokenize.serving import ServingTokenization
 from vllm.entrypoints.serve.utils.api_utils import (
     cli_env_setup,
     log_non_default_args,
@@ -376,8 +376,7 @@ async def init_app_state(
         log_error_stack=args.log_error_stack,
     )
 
-    state.openai_serving_tokenization = OpenAIServingTokenization(
-        engine_client,
+    state.serving_tokenization = ServingTokenization(
         state.openai_serving_models,
         state.openai_serving_render,
         request_logger=request_logger,
@@ -461,9 +460,15 @@ async def init_render_app_state(
     )
 
     state.openai_serving_models = model_registry
-
-    # Expose tokenization via the render handler (no engine required).
-    state.openai_serving_tokenization = state.openai_serving_render
+    state.serving_tokenization = ServingTokenization(
+        model_registry,
+        state.openai_serving_render,
+        request_logger=request_logger,
+        chat_template=resolved_chat_template,
+        chat_template_content_format=args.chat_template_content_format,
+        default_chat_template_kwargs=args.default_chat_template_kwargs,
+        trust_request_chat_template=args.trust_request_chat_template,
+    )
 
     state.vllm_config = vllm_config
     # Disable stats logging — there is no engine to poll.

vllm/entrypoints/openai/engine/serving.py

@@ -5,25 +5,19 @@ import time
 from collections.abc import Awaitable, Mapping
 from dataclasses import dataclass, field
 from http import HTTPStatus
-from typing import Any, ClassVar, Generic, Protocol, TypeAlias, TypeVar
+from typing import ClassVar, Generic, TypeVar
 
 from fastapi import Request
 from pydantic import ConfigDict
 from starlette.datastructures import Headers
 
-import vllm.envs as envs
-from vllm.config import ModelConfig
 from vllm.engine.protocol import EngineClient
-from vllm.entrypoints.chat_utils import ChatTemplateContentFormatOption
 from vllm.entrypoints.generate.beam_search.online import BeamSearchOnlineMixin
 from vllm.entrypoints.openai.chat_completion.protocol import (
-    BatchChatCompletionRequest,
     ChatCompletionRequest,
-    ChatCompletionResponse,
 )
 from vllm.entrypoints.openai.completion.protocol import (
     CompletionRequest,
-    CompletionResponse,
 )
 from vllm.entrypoints.openai.engine.protocol import (
     ErrorResponse,
@@ -31,81 +25,22 @@ from vllm.entrypoints.openai.engine.protocol import (
 )
 from vllm.entrypoints.openai.models.serving import OpenAIServingModels
 from vllm.entrypoints.openai.responses.protocol import ResponsesRequest
-from vllm.entrypoints.serve.disagg.protocol import GenerateRequest, GenerateResponse
-from vllm.entrypoints.serve.tokenize.protocol import (
-    DetokenizeRequest,
-    TokenizeChatRequest,
-    TokenizeCompletionRequest,
-    TokenizeResponse,
-)
-from vllm.entrypoints.serve.utils.error_response import create_error_response
+from vllm.entrypoints.serve.engine.serving import BaseServing
+from vllm.entrypoints.serve.engine.typing import AnyRequest
 from vllm.entrypoints.serve.utils.request_logger import RequestLogger
-from vllm.entrypoints.speech_to_text.transcription.protocol import (
-    TranscriptionRequest,
-    TranscriptionResponse,
-)
-from vllm.entrypoints.speech_to_text.translation.protocol import TranslationRequest
-from vllm.inputs import EngineInput, PromptType
+from vllm.inputs import EngineInput
 from vllm.logger import init_logger
 from vllm.logprobs import Logprob, PromptLogprobs
 from vllm.lora.request import LoRARequest
-from vllm.renderers import ChatParams, TokenizeParams
-from vllm.renderers.inputs.preprocess import (
-    extract_prompt_components,
-    extract_prompt_len,
-)
-from vllm.sampling_params import BeamSearchParams, SamplingParams
 from vllm.tokenizers import TokenizerLike
 from vllm.tracing import (
     contains_trace_headers,
     extract_trace_headers,
     log_tracing_disabled_warning,
 )
-from vllm.utils import random_uuid
 
 logger = init_logger(__name__)
 
-
-class RendererRequest(Protocol):
-    def build_tok_params(self, model_config: ModelConfig) -> TokenizeParams:
-        raise NotImplementedError
-
-
-class RendererChatRequest(RendererRequest, Protocol):
-    def build_chat_params(
-        self,
-        default_template: str | None,
-        default_template_content_format: ChatTemplateContentFormatOption,
-    ) -> ChatParams:
-        raise NotImplementedError
-
-
-CompletionLikeRequest: TypeAlias = (
-    CompletionRequest | TokenizeCompletionRequest | DetokenizeRequest
-)
-
-ChatLikeRequest: TypeAlias = (
-    ChatCompletionRequest | BatchChatCompletionRequest | TokenizeChatRequest
-)
-
-SpeechToTextRequest: TypeAlias = TranscriptionRequest | TranslationRequest
-
-AnyRequest: TypeAlias = (
-    CompletionLikeRequest
-    | ChatLikeRequest
-    | SpeechToTextRequest
-    | ResponsesRequest
-    | GenerateRequest
-)
-
-AnyResponse: TypeAlias = (
-    CompletionResponse
-    | ChatCompletionResponse
-    | TranscriptionResponse
-    | TokenizeResponse
-    | GenerateResponse
-)
-
 RequestT = TypeVar("RequestT", bound=AnyRequest)
 _T = TypeVar("_T")
 
@@ -122,7 +57,7 @@ class ServeContext(Generic[RequestT]):
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
 
-class OpenAIServing(BeamSearchOnlineMixin):
+class OpenAIServing(BaseServing, BeamSearchOnlineMixin):
     request_id_prefix: ClassVar[str] = """
     A short string prepended to every request’s ID.
     """
@@ -135,15 +70,14 @@ class OpenAIServing(BeamSearchOnlineMixin):
         request_logger: RequestLogger | None,
         return_tokens_as_token_ids: bool = False,
     ):
-        super().__init__()
+        super().__init__(
+            models=models,
+            model_config=engine_client.model_config,
+            request_logger=request_logger,
+        )
 
         self.engine_client = engine_client
-        self.models = models
-
-        self.request_logger = request_logger
         self.return_tokens_as_token_ids = return_tokens_as_token_ids
-
-        self.model_config = engine_client.model_config
         self.renderer = engine_client.renderer
         self.input_processor = engine_client.input_processor
         vllm_config = getattr(engine_client, "vllm_config", None)
@@ -163,15 +97,6 @@ class OpenAIServing(BeamSearchOnlineMixin):
             # Never fail server startup over the fingerprint.
             self.system_fingerprint = None
 
-    @staticmethod
-    def create_error_response(
-        message: str | Exception,
-        err_type: str = "BadRequestError",
-        status_code: HTTPStatus = HTTPStatus.BAD_REQUEST,
-        param: str | None = None,
-    ) -> ErrorResponse:
-        return create_error_response(message, err_type, status_code, param)
-
     def create_streaming_error_response(
         self,
         message: str | Exception,
@@ -208,167 +133,6 @@ class OpenAIServing(BeamSearchOnlineMixin):
             status_code=e.status_code,
         )
 
-    async def _check_model(
-        self,
-        request: AnyRequest,
-    ) -> ErrorResponse | None:
-        error_response = None
-
-        if self._is_model_supported(request.model):
-            return None
-        if request.model in self.models.lora_requests:
-            return None
-        if (
-            envs.VLLM_ALLOW_RUNTIME_LORA_UPDATING
-            and request.model
-            and (load_result := await self.models.resolve_lora(request.model))
-        ):
-            if isinstance(load_result, LoRARequest):
-                return None
-            if (
-                isinstance(load_result, ErrorResponse)
-                and load_result.error.code == HTTPStatus.BAD_REQUEST.value
-            ):
-                error_response = load_result
-
-        return error_response or self.create_error_response(
-            message=f"The model `{request.model}` does not exist.",
-            err_type="NotFoundError",
-            status_code=HTTPStatus.NOT_FOUND,
-            param="model",
-        )
-
-    def _get_active_default_mm_loras(self, request: AnyRequest) -> LoRARequest | None:
-        """Determine if there are any active default multimodal loras."""
-        # TODO: Currently this is only enabled for chat completions
-        # to be better aligned with only being enabled for .generate
-        # when run offline. It would be nice to support additional
-        # tasks types in the future.
-        message_types = self._get_message_types(request)
-        default_mm_loras = set()
-
-        for lora in self.models.lora_requests.values():
-            # Best effort match for default multimodal lora adapters;
-            # There is probably a better way to do this, but currently
-            # this matches against the set of 'types' in any content lists
-            # up until '_', e.g., to match audio_url -> audio
-            if lora.lora_name in message_types:
-                default_mm_loras.add(lora)
-
-        # Currently only support default modality specific loras if
-        # we have exactly one lora matched on the request.
-        if len(default_mm_loras) == 1:
-            return default_mm_loras.pop()
-        return None
-
-    def _maybe_get_adapters(
-        self,
-        request: AnyRequest,
-        supports_default_mm_loras: bool = False,
-    ) -> LoRARequest | None:
-        if request.model in self.models.lora_requests:
-            return self.models.lora_requests[request.model]
-
-        # Currently only support default modality specific loras
-        # if we have exactly one lora matched on the request.
-        if supports_default_mm_loras:
-            default_mm_lora = self._get_active_default_mm_loras(request)
-            if default_mm_lora is not None:
-                return default_mm_lora
-
-        if self._is_model_supported(request.model):
-            return None
-
-        # if _check_model has been called earlier, this will be unreachable
-        raise ValueError(f"The model `{request.model}` does not exist.")
-
-    def _get_message_types(self, request: AnyRequest) -> set[str]:
-        """Retrieve the set of types from message content dicts up
-        until `_`; we use this to match potential multimodal data
-        with default per modality loras.
-        """
-        message_types: set[str] = set()
-
-        if not hasattr(request, "messages"):
-            return message_types
-
-        messages = request.messages
-        if messages is None or isinstance(messages, (str, bytes)):
-            return message_types
-
-        for message in messages:
-            if (
-                isinstance(message, dict)
-                and "content" in message
-                and isinstance(message["content"], list)
-            ):
-                for content_dict in message["content"]:
-                    if "type" in content_dict:
-                        message_types.add(content_dict["type"].split("_")[0])
-        return message_types
-
-    def _validate_chat_template(
-        self,
-        request_chat_template: str | None,
-        chat_template_kwargs: dict[str, Any] | None,
-        trust_request_chat_template: bool,
-    ) -> ErrorResponse | None:
-        if not trust_request_chat_template and (
-            request_chat_template is not None
-            or (
-                chat_template_kwargs
-                and chat_template_kwargs.get("chat_template") is not None
-            )
-        ):
-            return self.create_error_response(
-                "Chat template is passed with request, but "
-                "--trust-request-chat-template is not set. "
-                "Refused request with untrusted chat template."
-            )
-        return None
-
-    @staticmethod
-    def _prepare_extra_chat_template_kwargs(
-        request_chat_template_kwargs: dict[str, Any] | None = None,
-        default_chat_template_kwargs: dict[str, Any] | None = None,
-    ) -> dict[str, Any]:
-        """Helper to merge server-default and request-specific chat template kwargs."""
-        request_chat_template_kwargs = request_chat_template_kwargs or {}
-        if default_chat_template_kwargs is None:
-            return request_chat_template_kwargs
-        # Apply server defaults first, then request kwargs override.
-        return default_chat_template_kwargs | request_chat_template_kwargs
-
-    def _extract_prompt_components(self, prompt: PromptType | EngineInput):
-        return extract_prompt_components(self.model_config, prompt)
-
-    def _extract_prompt_text(self, prompt: PromptType | EngineInput):
-        return self._extract_prompt_components(prompt).text
-
-    def _extract_prompt_len(self, prompt: EngineInput):
-        return extract_prompt_len(self.model_config, prompt)
-
-    def _log_inputs(
-        self,
-        request_id: str,
-        inputs: PromptType | EngineInput,
-        params: SamplingParams | BeamSearchParams | None,
-        lora_request: LoRARequest | None,
-    ) -> None:
-        if self.request_logger is None:
-            return
-
-        components = self._extract_prompt_components(inputs)
-
-        self.request_logger.log_inputs(
-            request_id,
-            components.text,
-            components.token_ids,
-            components.embeds,
-            params=params,
-            lora_request=lora_request,
-        )
-
     async def _get_trace_headers(
         self,
         headers: Headers,
@@ -383,18 +147,6 @@ class OpenAIServing(BeamSearchOnlineMixin):
 
         return None
 
-    @staticmethod
-    def _base_request_id(
-        raw_request: Request | None, default: str | None = None
-    ) -> str | None:
-        """Pulls the request id to use from a header, if provided"""
-        if raw_request is not None and (
-            (req_id := raw_request.headers.get("X-Request-Id")) is not None
-        ):
-            return req_id
-
-        return random_uuid() if default is None else default
-
     @staticmethod
     def _get_data_parallel_rank(raw_request: Request | None) -> int | None:
         """Pulls the data parallel rank from a header, if provided"""
@@ -464,13 +216,6 @@ class OpenAIServing(BeamSearchOnlineMixin):
 
         return tokenizer.decode([token_id])
 
-    def _is_model_supported(self, model_name: str | None) -> bool:
-        if not model_name:
-            return True
-        if envs.VLLM_SKIP_MODEL_NAME_VALIDATION:
-            return True
-        return self.models.is_base_model(model_name)
-
 
 def format_token_id_placeholder(token_id: int) -> str:
     return f"token_id:{token_id}"

vllm/entrypoints/openai/models/serving.py

@@ -42,6 +42,10 @@ class OpenAIModelRegistry:
     ) -> None:
         self.model_config = model_config
         self.base_model_paths = base_model_paths
+        self.lora_requests: dict[str, LoRARequest] = {}
+
+    def model_name(self, lora_request: LoRARequest | None = None) -> str:
+        return self.base_model_paths[0].name
 
     def is_base_model(self, model_name: str) -> bool:
         return any(model.name == model_name for model in self.base_model_paths)
@@ -72,6 +76,9 @@ class OpenAIModelRegistry:
             ]
         )
 
+    async def resolve_lora(self, lora_name: str):
+        raise RuntimeError("The OpenAIModelRegistry has no LoRA support.")
+
 
 class OpenAIServingModels:
     """Shared instance to hold data about the loaded base model(s) and adapters.

vllm/entrypoints/pooling/base/io_processor.py

@@ -12,7 +12,7 @@ from vllm.entrypoints.chat_utils import (
     ChatTemplateContentFormatOption,
     ConversationMessage,
 )
-from vllm.entrypoints.openai.engine.serving import RendererChatRequest, RendererRequest
+from vllm.entrypoints.serve.engine.typing import RendererChatRequest, RendererRequest
 from vllm.inputs import EngineInput, SingletonPrompt
 from vllm.renderers import BaseRenderer, TokenizeParams, merge_kwargs
 from vllm.renderers.inputs.preprocess import parse_model_prompt, prompt_to_seq

vllm/entrypoints/pooling/base/serving.py

@@ -52,7 +52,7 @@ class PoolingServingBase(ABC):
         self.engine_client = engine_client
         self.models = models
         self.model_config = models.model_config
-        self.renderer = models.renderer
+        self.renderer = engine_client.renderer
         self.vllm_config = engine_client.vllm_config
         self.max_model_len = self.model_config.max_model_len
         self.request_logger = request_logger
@@ -61,7 +61,7 @@ class PoolingServingBase(ABC):
         self.chat_template_config = chat_template_config
 
         # Shared thread pool executor for preprocessing and postprocessing.
-        self._executor: Executor = models.renderer._executor
+        self._executor: Executor = self.renderer._executor
         self._preprocessing_async = make_async(
             self._preprocessing, executor=self._executor
         )