Bump actions/setup-python from 6.1.0 to 6.3.0

Bumps [actions/setup-python](https://github.com/actions/setup-python) from 6.1.0 to 6.3.0. - [Release notes](https://github.com/actions/setup-python/releases) - [Commits](https://github.com/actions/setup-python/compare/83679a892e2d95755f2dac6acb0bfd1e9ac5d548...ece7cb06caefa5fff74198d8649806c4678c61a1) --- updated-dependencies: - dependency-name: actions/setup-python dependency-version: 6.3.0 dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com>
Bump actions/checkout from 6.0.1 to 7.0.0 (#33057 )
2026-06-30 12:18:49 +00:00 · 2026-06-30 13:16:20 +01:00 · 2026-06-30 13:01:03 +01:00 · 2026-06-30 11:20:33 +00:00 · 2026-06-30 18:17:43 +08:00 · 2026-06-30 09:57:27 +00:00
94 changed files with 4565 additions and 1555 deletions
@@ -23,4 +23,5 @@ steps:
      bash .buildkite/scripts/hardware_ci/run-intel-test.sh
      'cd tests &&
      export VLLM_WORKER_MULTIPROC_METHOD=spawn &&
+      pytest -v -s basic_correctness/test_cpu_offload.py &&
      pytest -v -s basic_correctness/test_mem.py::test_end_to_end'
@@ -128,10 +128,10 @@ steps:
      bash .buildkite/scripts/hardware_ci/run-intel-test.sh
      'cd tests &&
      export VLLM_WORKER_MULTIPROC_METHOD=spawn &&
-      (pytest -v -s lora/test_mixtral.py --deselect="tests/lora/test_mixtral.py::test_mixtral_lora[4]" || true) &&
      pytest -v -s lora/test_quant_model.py --deselect="tests/lora/test_quant_model.py::test_quant_model_lora[model0]" --deselect="tests/lora/test_quant_model.py::test_quant_model_lora[model1]" --deselect="tests/lora/test_quant_model.py::test_quant_model_tp_equality[model0]" &&
      pytest -v -s lora/test_transformers_model.py &&
      pytest -v -s lora/test_chatglm3_tp.py &&
+      pytest -v -s lora/test_llama_tp.py::test_llama_lora &&
      pytest -s -v lora/test_minicpmv_tp.py'

 - label: LoRA Multimodal
@@ -0,0 +1,27 @@
+group: Models - Distributed
+depends_on: 
+  - image-build-xpu
+steps:
+- label: Distributed Model Tests (2 GPUs)
+  key: distributed-model-tests-2-gpus
+  timeout_in_minutes: 50
+  device: intel_gpu
+  agent_tags:
+    label: production
+    gpu: 2+
+    mem: 24+
+  no_plugin: true
+  working_dir: "."
+  env:
+    REGISTRY: "public.ecr.aws/q9t5s3a7"
+    REPO: "vllm-ci-test-repo"
+    VLLM_TEST_DEVICE: "xpu"
+  source_file_dependencies:
+  - vllm/model_executor/model_loader/sharded_state_loader.py
+  - vllm/model_executor/models/
+  - tests/model_executor/model_loader/test_sharded_state_loader.py
+  commands:
+    - >-
+      bash .buildkite/scripts/hardware_ci/run-intel-test.sh
+      'cd tests &&
+      pytest -v -s model_executor/model_loader/test_sharded_state_loader.py -m "not slow_test"'
@@ -327,7 +327,7 @@ jobs:
                message: 'CC {users} for ROCm-related issue',
              },
              mistral: {
-                users: ['patrickvonplaten', 'juliendenize', 'andylolu2'],
+                users: ['patrickvonplaten', 'juliendenize', 'andylolu2', 'NickLucche'],
                message: 'CC {users} for Mistral-related issue',
              },
              // Add more label -> user mappings here
@@ -27,7 +27,7 @@ jobs:
    timeout-minutes: 30

    steps:
-      - uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v6.0.1
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0

      - uses: astral-sh/setup-uv@37802adc94f370d6bfd71619e3f0bf239e1f3b78 # v7.6.0
        with:
@@ -48,8 +48,8 @@ jobs:
    if: always() && (needs.pre-run-check.result == 'success' || needs.pre-run-check.result == 'skipped')
    runs-on: [self-hosted, linux, x64, vllm-runners]
    steps:
-    - uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v6.0.1
-    - uses: actions/setup-python@83679a892e2d95755f2dac6acb0bfd1e9ac5d548 # v6.1.0
+    - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+    - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0
      with:
        python-version: "3.12"
    # Provide shellcheck on PATH so tools/pre_commit/shellcheck.sh skips its
@@ -131,6 +131,19 @@ repos:
        --python-version, "3.12",
      ]
      files: ^requirements/(common|xpu|test/xpu)\.(in|txt)$
+    - id: pip-compile
+      alias: pip-compile-cpu
+      name: pip-compile-cpu
+      args: [
+        requirements/test/cuda.in,
+        -o, requirements/test/cpu.txt,
+        --index-strategy, unsafe-best-match,
+        --torch-backend, cpu,
+        --python-platform, x86_64-manylinux_2_28,
+        --python-version, "3.12",
+      ]
+      files: ^requirements/(common|cpu|test/(cuda|cpu))\.(in|txt)$
+      exclude: ^requirements/test/cuda\.txt$
    - id: pip-compile
      alias: pip-compile-docs
      name: pip-compile-docs
@@ -193,27 +193,16 @@ FROM base AS vllm-test-deps

 WORKDIR /vllm-workspace

-# Copy test requirements
-COPY requirements/common.txt requirements/common.txt
-COPY requirements/test/cuda.in requirements/test/cpu.in
+# Test requirements are compiled from requirements/test/cuda.in into
+# requirements/test/cpu.txt by the pip-compile-cpu pre-commit hook, which
+# resolves CPU wheels via uv's --torch-backend cpu.
+COPY requirements/test/cpu.txt requirements/test/cpu.txt

-RUN \
-    sed -i '/mamba_ssm/d' requirements/test/cpu.in && \
-    remove_packages_not_supported_on_aarch64() { \
-    case "$(uname -m)" in \
-    aarch64|arm64) \
-    sed -i '/decord/d' requirements/test/cpu.in; \
-    sed -i '/terratorch/d' requirements/test/cpu.in; \
-    ;; \
-    esac; \
-    }; \
-    remove_packages_not_supported_on_aarch64 && \
-    sed -i 's/^torch==.*/torch==2.11.0/g' requirements/test/cpu.in && \
-    sed -i 's/torchaudio.*/torchaudio/g' requirements/test/cpu.in && \
-    sed -i 's/torchvision.*/torchvision/g' requirements/test/cpu.in && \
-    # Related issue: https://github.com/vllm-project/vllm/pull/38800#issuecomment-4228314305
-    sed -i 's/^sentence-transformers.*/sentence-transformers==5.3.0/g' requirements/test/cpu.in && \
-    uv pip compile requirements/test/cpu.in -o requirements/test/cpu.txt --index-strategy unsafe-best-match --torch-backend cpu
+# cpu.txt is compiled for x86_64, so platform markers are resolved away. Drop
+# packages unavailable on aarch64 (decord, terratorch) for arm builds.
+RUN case "$(uname -m)" in \
+    aarch64|arm64) sed -i '/^decord==/d; /^terratorch==/d' requirements/test/cpu.txt ;; \
+    esac

 RUN --mount=type=cache,target=/root/.cache/uv \
    uv pip install -r requirements/test/cpu.txt
@@ -167,6 +167,7 @@ Priority is **1 = highest** (tried first).
 | `FLASH_ATTN` | FA4* | fp16, bf16 | `auto`, `float16`, `bfloat16` | %16 | Any | ✅ | ✅ | ❌ | ✅ | All | ≥10.0 |
 | `FLASH_ATTN_DIFFKV` | | fp16, bf16 | `auto` | Any | Any | ❌ | ❌ | ❌ | ✅ | Decoder | Any |
 | `FLEX_ATTENTION` | | fp16, bf16, fp32 | `auto`, `float16`, `bfloat16` | %16 | Any | ❌ | ✅ | ✅ | ❌ | Decoder, Encoder Only | Any |
+| `HPC_ATTN` | | fp16, bf16 | `auto`, `fp8_e4m3` | 64 | 128 | ❌ | ❌ | ❌ | ❌ | Decoder | ≥9.0 |
 | `ROCM_AITER_FA` | | fp16, bf16 | `auto`, `float16`, `bfloat16`, `fp8`, `fp8_e4m3`, `fp8_e5m2` | 16, 32 | 64, 128, 256 | ✅ | ✅ | ❌ | ❌ | Decoder | N/A |
 | `ROCM_AITER_UNIFIED_ATTN` | | bf16 | `auto`, `bfloat16`, `fp8`, `fp8_e4m3` | %16 | Any | ✅ | ❌ | ✅ | ❌ | All | N/A |
 | `ROCM_ATTN` | | fp16, bf16, fp32 | `auto`, `float16`, `bfloat16`, `fp8`, `fp8_e4m3`, `fp8_e5m2` | %16 | 32, 64, 80, 96, 128, 160, 192, 224, 256 | ❌ | ✅ | ✅ | ❌ | Decoder, Encoder, Encoder Only | N/A |
@@ -85,6 +85,21 @@ significantly reduce the attack surface for these types of abuse.
 Also, consider setting `VLLM_MEDIA_URL_ALLOW_REDIRECTS=0` to prevent HTTP
 redirects from being followed to bypass domain restrictions.

+### 5. **Restrict Media Decode Sizes:**
+
+Compressed media files can expand into gigabytes of memory during decoding. vLLM
+enforces decode-size limits to prevent out-of-memory denial of service:
+
+| Environment Variable | Default | Description |
+| --- | --- | --- |
+| `VLLM_MAX_IMAGE_PIXELS` | `178956970` (~179M pixels) | Maximum decoded image size in pixels. Images exceeding this are rejected before raster memory is allocated. Default matches PIL's built-in 2x decompression-bomb threshold (~680 MB for RGB). |
+| `VLLM_MAX_AUDIO_CLIP_FILESIZE_MB` | `25` | Maximum filesize in MB for a single audio file. |
+| `VLLM_MAX_AUDIO_DECODE_DURATION_S` | `600` | Maximum decoded audio duration in seconds. Prevents compressed audio from expanding into gigabytes of float32 PCM. |
+
+Setting any of these to `0` disables the corresponding limit. This is **not
+recommended** for deployments exposed to untrusted users, as it removes the
+protection against resource-exhaustion attacks.
+
 ## Security and Firewalls: Protecting Exposed vLLM Systems

 While vLLM is designed to allow unsafe network services to be isolated to
@@ -5310,7 +5310,6 @@ dependencies = [
 "vllm-llm",
 "vllm-metrics",
 "vllm-text",
- "vllm-tokenizer",
 "zeromq",
 ]

@@ -50,7 +50,6 @@ tokio.workspace = true
 tracing-subscriber.workspace = true
 uuid.workspace = true
 vllm-engine-core-client = { workspace = true, features = ["test-util"] }
-vllm-tokenizer = { workspace = true, features = ["test-utils"] }
 zeromq.workspace = true

 [lints]
@@ -154,8 +154,7 @@ mod tests {
    use thiserror_ext::AsReport as _;
    use vllm_text::Prompt;
    use vllm_text::backend::hf::TokenizerSource;
-    use vllm_text::tokenizer::DynTokenizer;
-    use vllm_tokenizer::test_utils::TestTokenizer;
+    use vllm_text::tokenizer::{DynTokenizer, Tokenizer};

    use super::HfChatBackend;
    use crate::backend::{ChatBackend, LoadModelBackendsOptions, NewChatOutputProcessorOptions};
@@ -197,8 +196,32 @@ mod tests {
        }
    }

+    struct TestTokenizer;
+
+    impl Tokenizer for TestTokenizer {
+        fn encode(
+            &self,
+            _text: &str,
+            _add_special_tokens: bool,
+        ) -> vllm_text::tokenizer::Result<Vec<u32>> {
+            Ok(Vec::new())
+        }
+
+        fn decode(
+            &self,
+            _token_ids: &[u32],
+            _skip_special_tokens: bool,
+        ) -> vllm_text::tokenizer::Result<String> {
+            Ok(String::new())
+        }
+
+        fn token_to_id(&self, _token: &str) -> Option<u32> {
+            None
+        }
+    }
+
    fn test_tokenizer() -> DynTokenizer {
-        Arc::new(TestTokenizer::new())
+        Arc::new(TestTokenizer)
    }

    fn backend_for_selection(
@@ -563,7 +563,7 @@ mod tests {

    use llm_multimodal::TokenId;
    use vllm_engine_core_client::protocol::tensor::WireArrayData;
-    use vllm_tokenizer::test_utils::TestTokenizer;
+    use vllm_text::tokenizer::{IncrementalDecoder, Tokenizer, TokenizerError};

    use super::*;

@@ -574,14 +574,60 @@ mod tests {
    const LLAMA4_TILE_X_SEPARATOR_ID: u32 = 200093;
    const LLAMA4_TILE_Y_SEPARATOR_ID: u32 = 200094;

-    fn llama4_tokenizer() -> TestTokenizer {
-        TestTokenizer::new()
-            .with_regular_token("<|image_start|>", LLAMA4_IMAGE_START_ID)
-            .with_regular_token("<|image_end|>", LLAMA4_IMAGE_END_ID)
-            .with_regular_token("<|image|>", LLAMA4_IMAGE_ID)
-            .with_regular_token("<|patch|>", LLAMA4_PATCH_ID)
-            .with_regular_token("<|tile_x_separator|>", LLAMA4_TILE_X_SEPARATOR_ID)
-            .with_regular_token("<|tile_y_separator|>", LLAMA4_TILE_Y_SEPARATOR_ID)
+    struct TestTokenizer;
+
+    impl Tokenizer for TestTokenizer {
+        fn encode(
+            &self,
+            text: &str,
+            _add_special_tokens: bool,
+        ) -> std::result::Result<Vec<u32>, TokenizerError> {
+            Ok(match text {
+                "<|image|>" => vec![LLAMA4_IMAGE_ID],
+                text => text.bytes().map(u32::from).collect(),
+            })
+        }
+
+        fn decode(
+            &self,
+            _token_ids: &[u32],
+            _skip_special_tokens: bool,
+        ) -> std::result::Result<String, TokenizerError> {
+            Ok(String::new())
+        }
+
+        fn token_to_id(&self, token: &str) -> Option<u32> {
+            match token {
+                "<|image_start|>" => Some(LLAMA4_IMAGE_START_ID),
+                "<|image_end|>" => Some(LLAMA4_IMAGE_END_ID),
+                "<|image|>" => Some(LLAMA4_IMAGE_ID),
+                "<|patch|>" => Some(LLAMA4_PATCH_ID),
+                "<|tile_x_separator|>" => Some(LLAMA4_TILE_X_SEPARATOR_ID),
+                "<|tile_y_separator|>" => Some(LLAMA4_TILE_Y_SEPARATOR_ID),
+                _ => None,
+            }
+        }
+
+        fn id_to_token(&self, id: u32) -> Option<String> {
+            match id {
+                LLAMA4_IMAGE_START_ID => Some("<|image_start|>".to_string()),
+                LLAMA4_IMAGE_END_ID => Some("<|image_end|>".to_string()),
+                LLAMA4_IMAGE_ID => Some("<|image|>".to_string()),
+                LLAMA4_PATCH_ID => Some("<|patch|>".to_string()),
+                LLAMA4_TILE_X_SEPARATOR_ID => Some("<|tile_x_separator|>".to_string()),
+                LLAMA4_TILE_Y_SEPARATOR_ID => Some("<|tile_y_separator|>".to_string()),
+                _ => None,
+            }
+        }
+
+        fn create_decode_stream(
+            &self,
+            _prompt_token_ids: &[u32],
+            _skip_special_tokens: bool,
+            _min_bytes_to_buffer: usize,
+        ) -> Box<dyn IncrementalDecoder + '_> {
+            unreachable!("not used")
+        }
    }

    fn test_info(model_type: &str, config: serde_json::Value) -> MultimodalModelInfo {
@@ -589,7 +635,7 @@ mod tests {
            model_id: format!("{model_type}-test"),
            model_type: Some(model_type.to_string()),
            config,
-            tokenizer: TokenizerResolver(Arc::new(llama4_tokenizer())),
+            tokenizer: TokenizerResolver(Arc::new(TestTokenizer)),
        };
        let spec = context
            .resolve_model_spec()
@@ -189,19 +189,46 @@ impl ChatOutputProcessor for DefaultChatOutputProcessor {
 mod tests {
    use std::sync::Arc;

-    use vllm_tokenizer::test_utils::TestTokenizer;
+    use vllm_tokenizer::Tokenizer;

    use super::DefaultChatOutputProcessor;
    use crate::Error;
    use crate::parser::ParserSelection;
    use crate::request::ChatRequest;

-    fn tokenizer() -> Arc<TestTokenizer> {
-        Arc::new(
-            TestTokenizer::new()
-                .with_regular_token("<|channel>", 256)
-                .with_regular_token("<channel|>", 257),
-        )
+    struct FakeTokenizer;
+
+    impl Tokenizer for FakeTokenizer {
+        fn encode(
+            &self,
+            text: &str,
+            _add_special_tokens: bool,
+        ) -> vllm_tokenizer::Result<Vec<u32>> {
+            Ok(text.chars().map(u32::from).collect())
+        }
+
+        fn decode(
+            &self,
+            token_ids: &[u32],
+            _skip_special_tokens: bool,
+        ) -> vllm_tokenizer::Result<String> {
+            Ok(token_ids
+                .iter()
+                .map(|token_id| char::from_u32(*token_id).unwrap_or('\u{FFFD}'))
+                .collect())
+        }
+
+        fn token_to_id(&self, token: &str) -> Option<u32> {
+            match token {
+                "<|channel>" => Some(1),
+                "<channel|>" => Some(2),
+                _ => None,
+            }
+        }
+    }
+
+    fn tokenizer() -> Arc<FakeTokenizer> {
+        Arc::new(FakeTokenizer)
    }

    #[test]
@@ -1,9 +1,32 @@
 use std::sync::Arc;

-use vllm_tokenizer::test_utils::TestTokenizer;
+use vllm_tokenizer::Tokenizer;

 use super::{ReasoningParserFactory, names};

+struct FakeTokenizer;
+
+impl Tokenizer for FakeTokenizer {
+    fn encode(&self, text: &str, _add_special_tokens: bool) -> vllm_tokenizer::Result<Vec<u32>> {
+        Ok(text.chars().map(u32::from).collect())
+    }
+
+    fn decode(
+        &self,
+        token_ids: &[u32],
+        _skip_special_tokens: bool,
+    ) -> vllm_tokenizer::Result<String> {
+        Ok(token_ids
+            .iter()
+            .map(|token_id| char::from_u32(*token_id).unwrap_or('\u{FFFD}'))
+            .collect())
+    }
+
+    fn token_to_id(&self, _token: &str) -> Option<u32> {
+        None
+    }
+}
+
 #[test]
 fn factory_contains_and_lists_registered_parsers() {
    let factory = ReasoningParserFactory::new();
@@ -84,7 +107,7 @@ fn factory_resolves_minimax_m3_before_generic_minimax() {

 #[test]
 fn factory_rejects_unknown_parser_names() {
-    let tokenizer = Arc::new(TestTokenizer::new());
+    let tokenizer = Arc::new(FakeTokenizer);
    let factory = ReasoningParserFactory::new();
    let error = match factory.create("missing", tokenizer) {
        Ok(_) => panic!("expected parser lookup to fail"),
@@ -75,14 +75,39 @@ impl UnifiedParserFactory {
 mod tests {
    use std::sync::Arc;

-    use vllm_tokenizer::test_utils::TestTokenizer;
+    use vllm_tokenizer::Tokenizer;

    use super::{UnifiedParserFactory, names};

-    fn tokenizer() -> TestTokenizer {
-        TestTokenizer::new()
-            .with_regular_token("<|channel>", 256)
-            .with_regular_token("<channel|>", 257)
+    struct FakeTokenizer;
+
+    impl Tokenizer for FakeTokenizer {
+        fn encode(
+            &self,
+            text: &str,
+            _add_special_tokens: bool,
+        ) -> vllm_tokenizer::Result<Vec<u32>> {
+            Ok(text.chars().map(u32::from).collect())
+        }
+
+        fn decode(
+            &self,
+            token_ids: &[u32],
+            _skip_special_tokens: bool,
+        ) -> vllm_tokenizer::Result<String> {
+            Ok(token_ids
+                .iter()
+                .map(|token_id| char::from_u32(*token_id).unwrap_or('\u{FFFD}'))
+                .collect())
+        }
+
+        fn token_to_id(&self, token: &str) -> Option<u32> {
+            match token {
+                "<|channel>" => Some(1),
+                "<channel|>" => Some(2),
+                _ => None,
+            }
+        }
    }

    #[test]
@@ -94,6 +119,6 @@ mod tests {
            factory.resolve_name_for_model("google/gemma-4-27b-it"),
            Some(names::GEMMA4)
        );
-        factory.create(names::GEMMA4, &[], Arc::new(tokenizer())).unwrap();
+        factory.create(names::GEMMA4, &[], Arc::new(FakeTokenizer)).unwrap();
    }
 }
@@ -21,17 +21,15 @@ use vllm_engine_core_client::protocol::{
 use vllm_engine_core_client::test_utils::{IpcNamespace, spawn_mock_engine_task};
 use vllm_engine_core_client::{EngineCoreClient, EngineCoreClientConfig};
 use vllm_llm::Llm;
-use vllm_text::tokenizer::DynTokenizer;
+use vllm_text::tokenizer::{DynTokenizer, Tokenizer};
 use vllm_text::{
    DecodedLogprobs, DecodedPositionLogprobs, DecodedPromptLogprobs, DecodedTokenLogprob, Prompt,
    TextBackend,
 };
-use vllm_tokenizer::test_utils::TestTokenizer;
 use zeromq::prelude::{SocketRecv, SocketSend};
 use zeromq::{DealerSocket, PushSocket, ZmqMessage};

 const SPECIAL_STOP_TOKEN_ID: u32 = 256;
-const UNKNOWN_DECODE_TOKEN_ID: u32 = 10_000;

 fn request_output(
    request_id: &str,
@@ -160,18 +158,45 @@ async fn connect_chat_llm_with_ipc(
 struct FakeChatBackend {
    has_template: bool,
    model_id: String,
-    tokenizer: DynTokenizer,
 }

-fn fake_chat_tokenizer() -> TestTokenizer {
-    TestTokenizer::new()
-        .with_special_token("<stop>", SPECIAL_STOP_TOKEN_ID)
-        .with_regular_token("<think>", 0xF001)
-        .with_regular_token("</think>", 0xF002)
-        .with_regular_token("<|START_THINKING|>", 0xF003)
-        .with_regular_token("<|END_THINKING|>", 0xF004)
-        .with_regular_token("◁think▷", 0xF005)
-        .with_regular_token("◁/think▷", 0xF006)
+#[derive(Debug)]
+struct FakeChatTokenizer;
+
+impl Tokenizer for FakeChatTokenizer {
+    fn encode(&self, text: &str, _add_special_tokens: bool) -> vllm_tokenizer::Result<Vec<u32>> {
+        Ok(text.bytes().map(u32::from).collect())
+    }
+
+    fn decode(
+        &self,
+        token_ids: &[u32],
+        skip_special_tokens: bool,
+    ) -> vllm_tokenizer::Result<String> {
+        let bytes = token_ids
+            .iter()
+            .filter_map(|id| {
+                if skip_special_tokens && *id == SPECIAL_STOP_TOKEN_ID {
+                    None
+                } else {
+                    Some(*id as u8)
+                }
+            })
+            .collect::<Vec<_>>();
+        Ok(String::from_utf8_lossy(&bytes).into_owned())
+    }
+
+    fn token_to_id(&self, token: &str) -> Option<u32> {
+        match token {
+            "<think>" => Some(0xF001),
+            "</think>" => Some(0xF002),
+            "<|START_THINKING|>" => Some(0xF003),
+            "<|END_THINKING|>" => Some(0xF004),
+            "◁think▷" => Some(0xF005),
+            "◁/think▷" => Some(0xF006),
+            _ => None,
+        }
+    }
 }

 impl fmt::Debug for FakeChatBackend {
@@ -185,7 +210,6 @@ impl FakeChatBackend {
        Self {
            has_template: true,
            model_id: "test-model".to_string(),
-            tokenizer: Arc::new(fake_chat_tokenizer()),
        }
    }

@@ -193,7 +217,6 @@ impl FakeChatBackend {
        Self {
            has_template: false,
            model_id: "test-model".to_string(),
-            tokenizer: Arc::new(fake_chat_tokenizer()),
        }
    }

@@ -201,19 +224,13 @@ impl FakeChatBackend {
        Self {
            has_template: true,
            model_id: model_id.into(),
-            tokenizer: Arc::new(fake_chat_tokenizer()),
        }
    }
-
-    fn with_tokenizer(mut self, tokenizer: DynTokenizer) -> Self {
-        self.tokenizer = tokenizer;
-        self
-    }
 }

 impl TextBackend for FakeChatBackend {
    fn tokenizer(&self) -> DynTokenizer {
-        Arc::clone(&self.tokenizer)
+        Arc::new(FakeChatTokenizer)
    }

    fn model_id(&self) -> &str {
@@ -265,6 +282,65 @@ impl ChatRenderer for FakeChatBackend {
    }
 }

+#[derive(Clone, Debug)]
+struct FailingDecodeBackend {
+    inner: FakeChatBackend,
+}
+
+#[derive(Debug)]
+struct FailingDecodeTokenizer;
+
+impl Tokenizer for FailingDecodeTokenizer {
+    fn encode(&self, text: &str, add_special_tokens: bool) -> vllm_tokenizer::Result<Vec<u32>> {
+        FakeChatTokenizer.encode(text, add_special_tokens)
+    }
+
+    fn decode(
+        &self,
+        token_ids: &[u32],
+        skip_special_tokens: bool,
+    ) -> vllm_tokenizer::Result<String> {
+        if token_ids.contains(&(b'i' as u32)) {
+            return Err(vllm_tokenizer::TokenizerError("decode failed".to_string()));
+        }
+        FakeChatTokenizer.decode(token_ids, skip_special_tokens)
+    }
+
+    fn token_to_id(&self, token: &str) -> Option<u32> {
+        FakeChatTokenizer.token_to_id(token)
+    }
+}
+
+impl TextBackend for FailingDecodeBackend {
+    fn tokenizer(&self) -> DynTokenizer {
+        Arc::new(FailingDecodeTokenizer)
+    }
+
+    fn model_id(&self) -> &str {
+        self.inner.model_id()
+    }
+}
+
+impl ChatBackend for FailingDecodeBackend {
+    fn chat_renderer(&self) -> DynChatRenderer {
+        Arc::new(self.clone())
+    }
+
+    fn new_chat_output_processor(
+        &self,
+        _request: &mut ChatRequest,
+        _options: NewChatOutputProcessorOptions<'_>,
+    ) -> vllm_chat::Result<DynChatOutputProcessor> {
+        Ok(Box::new(DefaultChatOutputProcessor::plain_text_only()))
+    }
+}
+
+impl ChatRenderer for FailingDecodeBackend {
+    fn render(&self, request: &ChatRequest) -> vllm_chat::Result<RenderedPrompt> {
+        self.inner.render(request)
+    }
+}
+
 /// Skip `LogprobsDelta` events that carry only token_ids (no logprobs),
 /// returning the next semantically interesting event.
 async fn next_semantic<S>(stream: &mut S) -> Option<Result<ChatEvent, vllm_chat::Error>>
@@ -662,12 +738,7 @@ async fn chat_stream_reports_decode_failure_as_error_event() {
                send_outputs(
                    push,
                    EngineCoreOutputs {
-                        outputs: vec![request_output(
-                            "chat-4",
-                            vec![UNKNOWN_DECODE_TOKEN_ID],
-                            None,
-                            None,
-                        )],
+                        outputs: vec![request_output("chat-4", vec![b'i' as u32], None, None)],
                        ..Default::default()
                    },
                )
@@ -676,8 +747,9 @@ async fn chat_stream_reports_decode_failure_as_error_event() {
        },
    );

-    let backend: Arc<dyn ChatTextBackend> =
-        Arc::new(FakeChatBackend::new().with_tokenizer(Arc::new(TestTokenizer::new())));
+    let backend: Arc<dyn ChatTextBackend> = Arc::new(FailingDecodeBackend {
+        inner: FakeChatBackend::new(),
+    });
    let chat = connect_chat_llm_with_ipc(
        EngineCoreClientConfig::new_single(handshake_address),
        &ipc,
@@ -697,10 +769,7 @@ async fn chat_stream_reports_decode_failure_as_error_event() {

    match timeout(Duration::from_secs(2), stream.next()).await.unwrap() {
        Some(Err(vllm_chat::Error::Text(vllm_text::Error::Tokenizer(message)))) => {
-            assert_eq!(
-                message,
-                format!("test tokenizer cannot decode unknown token id {UNKNOWN_DECODE_TOKEN_ID}")
-            );
+            assert_eq!(message, "decode failed");
        }
        other => panic!("unexpected event after close: {other:?}"),
    }
@@ -23,7 +23,6 @@ expect-test.workspace = true
 futures.workspace = true
 openai-protocol.workspace = true
 tool-parser.workspace = true
-vllm-tokenizer = { workspace = true, features = ["test-utils"] }

 [[bench]]
 name = "deepseek_v3"
@@ -27,10 +27,6 @@ impl Tokenizer for BenchTokenizer {
    fn token_to_id(&self, _token: &str) -> Option<u32> {
        Some(u32::MAX)
    }
-
-    fn id_to_token(&self, _id: u32) -> Option<String> {
-        Some("\u{FFFD}".to_string())
-    }
 }

 /// Bench-only adapter that exposes a unified parser through the tool-parser
@@ -49,14 +49,11 @@ mod tests {
    use std::sync::Arc;

    use super::SeedOssReasoningParser;
-    use crate::reasoning::{
-        ReasoningParser,
-        tests::{SEED_THINK_END_ID, SEED_THINK_START_ID, fake_tokenizer},
-    };
+    use crate::reasoning::{ReasoningParser, tests::FakeTokenizer};

    #[test]
    fn without_prompt_markers_expects_start_token() {
-        let tokenizer = Arc::new(fake_tokenizer());
+        let tokenizer = Arc::new(FakeTokenizer);
        let mut parser = SeedOssReasoningParser::new(tokenizer).unwrap();

        let delta = parser.push("implicit reasoning</seed:think>answer").unwrap();
@@ -69,10 +66,10 @@ mod tests {

    #[test]
    fn picks_up_prompt_start_boundary() {
-        let tokenizer = Arc::new(fake_tokenizer());
+        let tokenizer = Arc::new(FakeTokenizer);
        let mut parser = SeedOssReasoningParser::new(tokenizer).unwrap();
-        // Prompt prefills `<seed:think>`, opening reasoning before the stream.
-        parser.initialize(&[SEED_THINK_START_ID]).unwrap();
+        // Prompt prefills `<seed:think>` (id 10), opening reasoning before the stream.
+        parser.initialize(&[10]).unwrap();

        let delta = parser.push("reason</seed:think>answer").unwrap();
        assert_eq!(delta.reasoning.as_deref(), Some("reason"));
@@ -81,10 +78,10 @@ mod tests {

    #[test]
    fn respects_prompt_end_boundary() {
-        let tokenizer = Arc::new(fake_tokenizer());
+        let tokenizer = Arc::new(FakeTokenizer);
        let mut parser = SeedOssReasoningParser::new(tokenizer).unwrap();
-        // Prompt already closed reasoning with `</seed:think>`.
-        parser.initialize(&[SEED_THINK_END_ID]).unwrap();
+        // Prompt already closed reasoning with `</seed:think>` (id 11).
+        parser.initialize(&[11]).unwrap();

        let delta = parser.push("answer").unwrap();
        assert_eq!(delta.reasoning, None);
@@ -94,7 +91,7 @@ mod tests {
    #[test]
    fn handles_explicit_start_token() {
        // An explicit start delimiter must not leak into reasoning text.
-        let tokenizer = Arc::new(fake_tokenizer());
+        let tokenizer = Arc::new(FakeTokenizer);
        let mut parser = SeedOssReasoningParser::new(tokenizer).unwrap();

        let delta = parser.push("<seed:think>reason</seed:think>answer").unwrap();
@@ -106,7 +103,7 @@ mod tests {
    fn streams_explicit_start_token_across_pushes() {
        // Start token, reasoning body, end token, and content arrive in separate
        // streaming deltas.
-        let tokenizer = Arc::new(fake_tokenizer());
+        let tokenizer = Arc::new(FakeTokenizer);
        let mut parser = SeedOssReasoningParser::new(tokenizer).unwrap();

        let mut reasoning = String::new();
@@ -134,9 +131,9 @@ mod tests {

    #[test]
    fn handles_partial_delimiters_across_pushes() {
-        let tokenizer = Arc::new(fake_tokenizer());
+        let tokenizer = Arc::new(FakeTokenizer);
        let mut parser = SeedOssReasoningParser::new(tokenizer).unwrap();
-        parser.initialize(&[SEED_THINK_START_ID]).unwrap();
+        parser.initialize(&[10]).unwrap();

        // Closing delimiter `</seed:think>` arrives in two halves.
        let first = parser.push("reason</seed:").unwrap();
@@ -127,17 +127,14 @@ mod tests {
    use std::sync::Arc;

    use super::Step3p5ReasoningParser;
-    use crate::reasoning::{
-        ReasoningParser,
-        tests::{THINK_START_ID, fake_tokenizer},
-    };
+    use crate::reasoning::{ReasoningParser, tests::FakeTokenizer};

    #[test]
    fn picks_up_prompt_start_boundary() {
-        let tokenizer = Arc::new(fake_tokenizer());
+        let tokenizer = Arc::new(FakeTokenizer);
        let mut parser = Step3p5ReasoningParser::new(tokenizer).unwrap();
-        // Prompt prefills `<think>`, opening reasoning before the stream.
-        parser.initialize(&[THINK_START_ID]).unwrap();
+        // Prompt prefills `<think>` (id 1), opening reasoning before the stream.
+        parser.initialize(&[1]).unwrap();

        let delta = parser.push("This is a reasoning section</think>This is the rest").unwrap();
        assert_eq!(
@@ -149,7 +146,7 @@ mod tests {

    #[test]
    fn handles_unterminated_reasoning() {
-        let tokenizer = Arc::new(fake_tokenizer());
+        let tokenizer = Arc::new(FakeTokenizer);
        let mut parser = Step3p5ReasoningParser::new(tokenizer).unwrap();

        let pushed = parser.push("<think>reason without end").unwrap();
@@ -162,7 +159,7 @@ mod tests {

    #[test]
    fn handles_empty_input() {
-        let tokenizer = Arc::new(fake_tokenizer());
+        let tokenizer = Arc::new(FakeTokenizer);
        let mut parser = Step3p5ReasoningParser::new(tokenizer).unwrap();

        let pushed = parser.push("").unwrap();
@@ -175,9 +172,9 @@ mod tests {
    fn complex_newline_pattern_trims_only_single_framing_newline_each_side() {
        // Only the immediately-adjacent framing `\n` is dropped on each side of
        // `</think>`; surrounding newlines remain part of reasoning/content.
-        let tokenizer = Arc::new(fake_tokenizer());
+        let tokenizer = Arc::new(FakeTokenizer);
        let mut parser = Step3p5ReasoningParser::new(tokenizer).unwrap();
-        parser.initialize(&[THINK_START_ID]).unwrap();
+        parser.initialize(&[1]).unwrap();

        let delta = parser
            .push("\n This is a \n reasoning section\n\n\n</think>\n\nThis is the rest")
@@ -191,7 +188,7 @@ mod tests {

    #[test]
    fn drops_framing_newlines_in_single_push() {
-        let tokenizer = Arc::new(fake_tokenizer());
+        let tokenizer = Arc::new(FakeTokenizer);
        let mut parser = Step3p5ReasoningParser::new(tokenizer).unwrap();

        let delta = parser.push("<think>reason\n</think>\nanswer").unwrap();
@@ -201,7 +198,7 @@ mod tests {

    #[test]
    fn drops_framing_newlines_across_pushes() {
-        let tokenizer = Arc::new(fake_tokenizer());
+        let tokenizer = Arc::new(FakeTokenizer);
        let mut parser = Step3p5ReasoningParser::new(tokenizer).unwrap();

        // The trailing `\n` from the first push is held until we know whether
@@ -222,7 +219,7 @@ mod tests {

    #[test]
    fn replays_held_newline_when_more_reasoning_follows() {
-        let tokenizer = Arc::new(fake_tokenizer());
+        let tokenizer = Arc::new(FakeTokenizer);
        let mut parser = Step3p5ReasoningParser::new(tokenizer).unwrap();

        let first = parser.push("<think>reason\n").unwrap();
@@ -235,7 +232,7 @@ mod tests {

    #[test]
    fn finish_flushes_held_newline_in_unterminated_stream() {
-        let tokenizer = Arc::new(fake_tokenizer());
+        let tokenizer = Arc::new(FakeTokenizer);
        let mut parser = Step3p5ReasoningParser::new(tokenizer).unwrap();

        let first = parser.push("<think>reason\n").unwrap();
@@ -248,7 +245,7 @@ mod tests {

    #[test]
    fn preserves_inner_newlines_in_reasoning() {
-        let tokenizer = Arc::new(fake_tokenizer());
+        let tokenizer = Arc::new(FakeTokenizer);
        let mut parser = Step3p5ReasoningParser::new(tokenizer).unwrap();

        let delta = parser.push("<think>line1\nline2</think>tail").unwrap();
@@ -260,7 +257,7 @@ mod tests {
    fn trims_only_one_trailing_reasoning_newline() {
        // Only the single framing newline immediately before `</think>` is
        // dropped; earlier newlines in the reasoning body are preserved.
-        let tokenizer = Arc::new(fake_tokenizer());
+        let tokenizer = Arc::new(FakeTokenizer);
        let mut parser = Step3p5ReasoningParser::new(tokenizer).unwrap();

        let delta = parser.push("<think>reason\n\n</think>answer").unwrap();
@@ -272,7 +269,7 @@ mod tests {
    fn drops_only_first_content_newline_after_transition() {
        // The leading-`\n` drop applies only to the first content delta after
        // `</think>`; later deltas pass through untouched.
-        let tokenizer = Arc::new(fake_tokenizer());
+        let tokenizer = Arc::new(FakeTokenizer);
        let mut parser = Step3p5ReasoningParser::new(tokenizer).unwrap();

        let first = parser.push("<think>reason</think>").unwrap();
@@ -291,7 +288,7 @@ mod tests {

    #[test]
    fn passes_through_clean_boundary_without_framing_newlines() {
-        let tokenizer = Arc::new(fake_tokenizer());
+        let tokenizer = Arc::new(FakeTokenizer);
        let mut parser = Step3p5ReasoningParser::new(tokenizer).unwrap();

        let delta = parser.push("<think>reason</think>tail").unwrap();
@@ -301,7 +298,7 @@ mod tests {

    #[test]
    fn handles_empty_reasoning_section() {
-        let tokenizer = Arc::new(fake_tokenizer());
+        let tokenizer = Arc::new(FakeTokenizer);
        let mut parser = Step3p5ReasoningParser::new(tokenizer).unwrap();

        let delta = parser.push("<think></think>answer").unwrap();
@@ -1,42 +1,54 @@
 use std::sync::Arc;

-use vllm_tokenizer::test_utils::TestTokenizer;
+use vllm_tokenizer::Tokenizer;

 use super::{
    DeepSeekR1ReasoningParser, DelimitedReasoningParser, MiniMaxM3ReasoningParser,
    Qwen3ReasoningParser, ReasoningParser,
 };

-pub(crate) const THINK_START_ID: u32 = 256;
-pub(crate) const THINK_END_ID: u32 = 257;
-pub(crate) const START_THINKING_ID: u32 = 258;
-pub(crate) const END_THINKING_ID: u32 = 259;
-pub(crate) const MINIMAX_THINK_START_ID: u32 = 260;
-pub(crate) const MINIMAX_THINK_END_ID: u32 = 261;
-pub(crate) const SPECIAL_BOUNDARY_ID: u32 = 262;
-pub(crate) const MM_THINK_START_ID: u32 = 263;
-pub(crate) const MM_THINK_END_ID: u32 = 264;
-pub(crate) const SEED_THINK_START_ID: u32 = 265;
-pub(crate) const SEED_THINK_END_ID: u32 = 266;
+pub(crate) struct FakeTokenizer;

-pub(crate) fn fake_tokenizer() -> TestTokenizer {
-    TestTokenizer::new()
-        .with_regular_token("<think>", THINK_START_ID)
-        .with_regular_token("</think>", THINK_END_ID)
-        .with_regular_token("<|START_THINKING|>", START_THINKING_ID)
-        .with_regular_token("<|END_THINKING|>", END_THINKING_ID)
-        .with_regular_token("◁think▷", MINIMAX_THINK_START_ID)
-        .with_regular_token("◁/think▷", MINIMAX_THINK_END_ID)
-        .with_special_token("<special-boundary>", SPECIAL_BOUNDARY_ID)
-        .with_regular_token("<mm:think>", MM_THINK_START_ID)
-        .with_regular_token("</mm:think>", MM_THINK_END_ID)
-        .with_regular_token("<seed:think>", SEED_THINK_START_ID)
-        .with_regular_token("</seed:think>", SEED_THINK_END_ID)
+impl Tokenizer for FakeTokenizer {
+    fn encode(&self, text: &str, _add_special_tokens: bool) -> vllm_tokenizer::Result<Vec<u32>> {
+        Ok(text.chars().map(u32::from).collect())
+    }
+
+    fn decode(
+        &self,
+        token_ids: &[u32],
+        _skip_special_tokens: bool,
+    ) -> vllm_tokenizer::Result<String> {
+        Ok(token_ids
+            .iter()
+            .map(|token_id| char::from_u32(*token_id).unwrap_or('\u{FFFD}'))
+            .collect())
+    }
+
+    fn token_to_id(&self, token: &str) -> Option<u32> {
+        match token {
+            "<think>" => Some(1),
+            "</think>" => Some(2),
+            "<|START_THINKING|>" => Some(3),
+            "<|END_THINKING|>" => Some(4),
+            "◁think▷" => Some(5),
+            "◁/think▷" => Some(6),
+            "<mm:think>" => Some(8),
+            "</mm:think>" => Some(9),
+            "<seed:think>" => Some(10),
+            "</seed:think>" => Some(11),
+            _ => None,
+        }
+    }
+
+    fn is_special_id(&self, token_id: u32) -> bool {
+        token_id == 7
+    }
 }

 #[test]
 fn delimited_content_only_stream() {
-    let tokenizer = Arc::new(fake_tokenizer());
+    let tokenizer = Arc::new(FakeTokenizer);
    let mut parser =
        DelimitedReasoningParser::new(tokenizer, "<think>", "</think>", false).unwrap();

@@ -48,7 +60,7 @@ fn delimited_content_only_stream() {

 #[test]
 fn delimited_single_chunk_with_reasoning_and_content() {
-    let tokenizer = Arc::new(fake_tokenizer());
+    let tokenizer = Arc::new(FakeTokenizer);
    let mut parser =
        DelimitedReasoningParser::new(tokenizer, "<think>", "</think>", false).unwrap();

@@ -59,7 +71,7 @@ fn delimited_single_chunk_with_reasoning_and_content() {

 #[test]
 fn delimited_partial_tokens_across_chunks() {
-    let tokenizer = Arc::new(fake_tokenizer());
+    let tokenizer = Arc::new(FakeTokenizer);
    let mut parser =
        DelimitedReasoningParser::new(tokenizer, "<think>", "</think>", false).unwrap();

@@ -71,10 +83,10 @@ fn delimited_partial_tokens_across_chunks() {

 #[test]
 fn delimited_finish_flushes_buffer() {
-    let tokenizer = Arc::new(fake_tokenizer());
+    let tokenizer = Arc::new(FakeTokenizer);
    let mut parser =
        DelimitedReasoningParser::new(tokenizer, "<think>", "</think>", false).unwrap();
-    parser.initialize(&[THINK_START_ID]);
+    parser.initialize(&[1]);

    let delta = parser.push("unfinished</thi");
    assert_eq!(delta.reasoning.as_deref(), Some("unfinished"));
@@ -84,7 +96,7 @@ fn delimited_finish_flushes_buffer() {

 #[test]
 fn qwen3_without_prompt_markers_expects_start_token() {
-    let tokenizer = Arc::new(fake_tokenizer());
+    let tokenizer = Arc::new(FakeTokenizer);
    let mut parser = Qwen3ReasoningParser::new(tokenizer).unwrap();

    let delta = parser.push("reason</think>answer").unwrap();
@@ -94,9 +106,9 @@ fn qwen3_without_prompt_markers_expects_start_token() {

 #[test]
 fn qwen3_prompt_end_marker_starts_in_content() {
-    let tokenizer = Arc::new(fake_tokenizer());
+    let tokenizer = Arc::new(FakeTokenizer);
    let mut parser = Qwen3ReasoningParser::new(tokenizer).unwrap();
-    parser.initialize(&[THINK_END_ID]).unwrap();
+    parser.initialize(&[2]).unwrap();

    let delta = parser.push("answer").unwrap();
    assert_eq!(delta.reasoning, None);
@@ -105,7 +117,7 @@ fn qwen3_prompt_end_marker_starts_in_content() {

 #[test]
 fn qwen3_tolerates_old_and_new_formats() {
-    let tokenizer = Arc::new(fake_tokenizer());
+    let tokenizer = Arc::new(FakeTokenizer);

    let mut old_parser = Qwen3ReasoningParser::new(tokenizer.clone()).unwrap();
    let old = old_parser.push("<think>reason</think>answer").unwrap();
@@ -113,7 +125,7 @@ fn qwen3_tolerates_old_and_new_formats() {
    assert_eq!(old.content.as_deref(), Some("answer"));

    let mut new_parser = Qwen3ReasoningParser::new(tokenizer).unwrap();
-    new_parser.initialize(&[THINK_START_ID]).unwrap();
+    new_parser.initialize(&[1]).unwrap();
    let new = new_parser.push("reason</think>answer").unwrap();
    assert_eq!(new.reasoning.as_deref(), Some("reason"));
    assert_eq!(new.content.as_deref(), Some("answer"));
@@ -121,10 +133,10 @@ fn qwen3_tolerates_old_and_new_formats() {

 #[test]
 fn qwen3_stops_scanning_at_last_special_token() {
-    let tokenizer = Arc::new(fake_tokenizer());
+    let tokenizer = Arc::new(FakeTokenizer);
    let mut parser = Qwen3ReasoningParser::new(tokenizer).unwrap();

-    parser.initialize(&[THINK_START_ID, SPECIAL_BOUNDARY_ID]).unwrap();
+    parser.initialize(&[1, 7]).unwrap();

    let delta = parser.push("answer").unwrap();
    assert_eq!(delta.reasoning, None);
@@ -133,7 +145,7 @@ fn qwen3_stops_scanning_at_last_special_token() {

 #[test]
 fn deepseek_r1_defaults_to_reasoning_without_prompt_boundary() {
-    let tokenizer = Arc::new(fake_tokenizer());
+    let tokenizer = Arc::new(FakeTokenizer);
    let mut parser = DeepSeekR1ReasoningParser::new(tokenizer).unwrap();

    let delta = parser.push("reason</think>answer").unwrap();
@@ -143,10 +155,10 @@ fn deepseek_r1_defaults_to_reasoning_without_prompt_boundary() {

 #[test]
 fn deepseek_r1_stops_scanning_at_last_special_token() {
-    let tokenizer = Arc::new(fake_tokenizer());
+    let tokenizer = Arc::new(FakeTokenizer);
    let mut parser = DeepSeekR1ReasoningParser::new(tokenizer).unwrap();

-    parser.initialize(&[THINK_END_ID, SPECIAL_BOUNDARY_ID]).unwrap();
+    parser.initialize(&[2, 7]).unwrap();

    let delta = parser.push("reason</think>answer").unwrap();
    assert_eq!(delta.reasoning.as_deref(), Some("reason"));
@@ -155,7 +167,7 @@ fn deepseek_r1_stops_scanning_at_last_special_token() {

 #[test]
 fn minimax_m3_handles_explicit_think_delimiters() {
-    let tokenizer = Arc::new(fake_tokenizer());
+    let tokenizer = Arc::new(FakeTokenizer);
    let mut parser = MiniMaxM3ReasoningParser::new(tokenizer).unwrap();

    let delta = parser.push("<mm:think>reason</mm:think>answer").unwrap();
@@ -165,7 +177,7 @@ fn minimax_m3_handles_explicit_think_delimiters() {

 #[test]
 fn minimax_m3_drops_leading_end_marker() {
-    let tokenizer = Arc::new(fake_tokenizer());
+    let tokenizer = Arc::new(FakeTokenizer);
    let mut parser = MiniMaxM3ReasoningParser::new(tokenizer).unwrap();

    let delta = parser.push("</mm:think>answer").unwrap();
@@ -175,7 +187,7 @@ fn minimax_m3_drops_leading_end_marker() {

 #[test]
 fn minimax_m3_preserves_non_leading_end_marker() {
-    let tokenizer = Arc::new(fake_tokenizer());
+    let tokenizer = Arc::new(FakeTokenizer);
    let mut parser = MiniMaxM3ReasoningParser::new(tokenizer).unwrap();

    let delta = parser.push("XXX</mm:think>YYY").unwrap();
@@ -185,7 +197,7 @@ fn minimax_m3_preserves_non_leading_end_marker() {

 #[test]
 fn minimax_m3_drops_split_leading_end_marker() {
-    let tokenizer = Arc::new(fake_tokenizer());
+    let tokenizer = Arc::new(FakeTokenizer);
    let mut parser = MiniMaxM3ReasoningParser::new(tokenizer).unwrap();

    assert!(parser.push("</mm").unwrap().is_empty());
@@ -196,9 +208,9 @@ fn minimax_m3_drops_split_leading_end_marker() {

 #[test]
 fn minimax_m3_uses_prompt_prefilled_start_marker() {
-    let tokenizer = Arc::new(fake_tokenizer());
+    let tokenizer = Arc::new(FakeTokenizer);
    let mut parser = MiniMaxM3ReasoningParser::new(tokenizer).unwrap();
-    parser.initialize(&[MM_THINK_START_ID]).unwrap();
+    parser.initialize(&[8]).unwrap();

    let delta = parser.push("reason</mm:think>answer").unwrap();
    assert_eq!(delta.reasoning.as_deref(), Some("reason"));
@@ -207,9 +219,9 @@ fn minimax_m3_uses_prompt_prefilled_start_marker() {

 #[test]
 fn minimax_m3_uses_prompt_prefilled_end_marker() {
-    let tokenizer = Arc::new(fake_tokenizer());
+    let tokenizer = Arc::new(FakeTokenizer);
    let mut parser = MiniMaxM3ReasoningParser::new(tokenizer).unwrap();
-    parser.initialize(&[MM_THINK_END_ID]).unwrap();
+    parser.initialize(&[9]).unwrap();

    let delta = parser.push("answer").unwrap();
    assert_eq!(delta.reasoning, None);
@@ -124,17 +124,42 @@ impl UnifiedParser for CombinedParser {
 mod tests {
    use std::sync::Arc;

-    use vllm_tokenizer::test_utils::TestTokenizer;
+    use vllm_tokenizer::Tokenizer;

    use super::CombinedParser;
    use crate::reasoning::{Qwen3ReasoningParser, ReasoningDelta, ReasoningParser};
    use crate::tool::{Qwen3XmlToolParser, Tool, ToolParser};
    use crate::unified::{UnifiedParser, UnifiedParserEvent, UnifiedParserOutput};

-    fn tokenizer() -> TestTokenizer {
-        TestTokenizer::new()
-            .with_regular_token("<think>", 256)
-            .with_regular_token("</think>", 257)
+    struct FakeTokenizer;
+
+    impl Tokenizer for FakeTokenizer {
+        fn encode(
+            &self,
+            text: &str,
+            _add_special_tokens: bool,
+        ) -> vllm_tokenizer::Result<Vec<u32>> {
+            Ok(text.chars().map(u32::from).collect())
+        }
+
+        fn decode(
+            &self,
+            token_ids: &[u32],
+            _skip_special_tokens: bool,
+        ) -> vllm_tokenizer::Result<String> {
+            Ok(token_ids
+                .iter()
+                .map(|token_id| char::from_u32(*token_id).unwrap_or('\u{FFFD}'))
+                .collect())
+        }
+
+        fn token_to_id(&self, token: &str) -> Option<u32> {
+            match token {
+                "<think>" => Some(1),
+                "</think>" => Some(2),
+                _ => None,
+            }
+        }
    }

    fn test_tools() -> Vec<Tool> {
@@ -248,7 +273,7 @@ mod tests {

    #[test]
    fn combined_parser_emits_reasoning_and_text() {
-        let tokenizer = Arc::new(tokenizer());
+        let tokenizer = Arc::new(FakeTokenizer);
        let reasoning = Qwen3ReasoningParser::create(tokenizer).unwrap();
        let mut parser = CombinedParser::new(Some(reasoning), None);

@@ -515,7 +515,7 @@ mod tests {

    use serde_json::{Value, json};
    use thiserror_ext::AsReport;
-    use vllm_tokenizer::test_utils::TestTokenizer;
+    use vllm_tokenizer::Tokenizer;
    use winnow::combinator::{eof, terminated};
    use winnow::error::ErrMode;
    use winnow::prelude::*;
@@ -527,15 +527,66 @@ mod tests {
    use crate::tool::Tool;
    use crate::unified::{UnifiedParserEvent, parsing_failed};

-    const CHANNEL_START_ID: u32 = 256;
-    const CHANNEL_END_ID: u32 = 257;
-    const TURN_BOUNDARY_ID: u32 = 258;
+    struct FakeTokenizer;

-    fn tokenizer() -> TestTokenizer {
-        TestTokenizer::new()
-            .with_special_token(CHANNEL_START, CHANNEL_START_ID)
-            .with_special_token(CHANNEL_END, CHANNEL_END_ID)
-            .with_special_token("<turn-boundary>", TURN_BOUNDARY_ID)
+    impl Tokenizer for FakeTokenizer {
+        fn encode(
+            &self,
+            text: &str,
+            _add_special_tokens: bool,
+        ) -> vllm_tokenizer::Result<Vec<u32>> {
+            Ok(text.chars().map(u32::from).collect())
+        }
+
+        fn decode(
+            &self,
+            token_ids: &[u32],
+            _skip_special_tokens: bool,
+        ) -> vllm_tokenizer::Result<String> {
+            Ok(token_ids
+                .iter()
+                .map(|token_id| char::from_u32(*token_id).unwrap_or('\u{FFFD}'))
+                .collect())
+        }
+
+        fn token_to_id(&self, token: &str) -> Option<u32> {
+            match token {
+                CHANNEL_START => Some(100),
+                CHANNEL_END => Some(101),
+                _ => None,
+            }
+        }
+
+        fn is_special_id(&self, token_id: u32) -> bool {
+            matches!(token_id, 100..=105)
+        }
+    }
+
+    struct MissingTokenTokenizer;
+
+    impl Tokenizer for MissingTokenTokenizer {
+        fn encode(
+            &self,
+            text: &str,
+            _add_special_tokens: bool,
+        ) -> vllm_tokenizer::Result<Vec<u32>> {
+            Ok(text.chars().map(u32::from).collect())
+        }
+
+        fn decode(
+            &self,
+            token_ids: &[u32],
+            _skip_special_tokens: bool,
+        ) -> vllm_tokenizer::Result<String> {
+            Ok(token_ids
+                .iter()
+                .map(|token_id| char::from_u32(*token_id).unwrap_or('\u{FFFD}'))
+                .collect())
+        }
+
+        fn token_to_id(&self, _token: &str) -> Option<u32> {
+            None
+        }
    }

    trait UnifiedParserTestExt {
@@ -665,12 +716,12 @@ mod tests {
    }

    fn test_parser() -> Gemma4UnifiedParser {
-        Gemma4UnifiedParser::new(&test_tools(), Arc::new(tokenizer())).unwrap()
+        Gemma4UnifiedParser::new(&test_tools(), Arc::new(FakeTokenizer)).unwrap()
    }

    #[test]
    fn gemma4_create_requires_channel_start_token() {
-        let error = match Gemma4UnifiedParser::new(&test_tools(), Arc::new(TestTokenizer::new())) {
+        let error = match Gemma4UnifiedParser::new(&test_tools(), Arc::new(MissingTokenTokenizer)) {
            Ok(_) => panic!("expected missing token error"),
            Err(error) => error,
        };
@@ -995,7 +1046,7 @@ mod tests {
    #[test]
    fn gemma4_initialize_open_channel_prompt_starts_in_reasoning() {
        let mut parser = test_parser();
-        parser.initialize(&[CHANNEL_START_ID, 3000, 3001]).unwrap();
+        parser.initialize(&[100, 3000, 3001]).unwrap();

        let output = parser.parse_complete("reason<channel|>answer").unwrap();

@@ -1006,7 +1057,7 @@ mod tests {
    #[test]
    fn gemma4_initialize_turn_prompt_starts_in_text() {
        let mut parser = test_parser();
-        parser.initialize(&[TURN_BOUNDARY_ID, 3000, 3001]).unwrap();
+        parser.initialize(&[104, 3000, 3001]).unwrap();

        let output = parser.parse_complete("<|channel>thought\nreason<channel|>answer").unwrap();

@@ -1017,7 +1068,7 @@ mod tests {
    #[test]
    fn gemma4_initialize_special_token_caps_boundary_scan() {
        let mut parser = test_parser();
-        parser.initialize(&[CHANNEL_START_ID, 3000, TURN_BOUNDARY_ID, 3001]).unwrap();
+        parser.initialize(&[100, 3000, 104, 3001]).unwrap();

        let output = parser.parse_complete("answer").unwrap();

@@ -1028,7 +1079,7 @@ mod tests {
    #[test]
    fn gemma4_initialize_closed_channel_prompt_starts_in_text() {
        let mut parser = test_parser();
-        parser.initialize(&[CHANNEL_START_ID, 3000, 3001, CHANNEL_END_ID]).unwrap();
+        parser.initialize(&[100, 3000, 3001, 101]).unwrap();

        let output = parser.parse_complete("answer").unwrap();

@@ -63,7 +63,6 @@ serial_test.workspace = true
 tempfile.workspace = true
 tower.workspace = true
 vllm-engine-core-client = { workspace = true, features = ["test-util"] }
-vllm-tokenizer = { workspace = true, features = ["test-utils"] }
 zeromq.workspace = true

 [lints]
@@ -24,9 +24,8 @@ use vllm_engine_core_client::protocol::{
 use vllm_engine_core_client::test_utils::{IpcNamespace, spawn_mock_engine_task};
 use vllm_engine_core_client::{EngineCoreClient, EngineCoreClientConfig, EngineId};
 use vllm_llm::Llm;
-use vllm_text::tokenizer::DynTokenizer;
+use vllm_text::tokenizer::{DynTokenizer, Tokenizer};
 use vllm_text::{Prompt, TextBackend};
-use vllm_tokenizer::test_utils::TestTokenizer;
 use zeromq::prelude::{SocketRecv, SocketSend};
 use zeromq::{DealerSocket, PushSocket, ZmqMessage};

@@ -156,9 +155,37 @@ fn test_llm(client: EngineCoreClient) -> Llm {
 #[derive(Clone, Debug)]
 struct FakeTextBackend;

+#[derive(Debug)]
+struct FakeTokenizer;
+
+impl Tokenizer for FakeTokenizer {
+    fn encode(
+        &self,
+        text: &str,
+        _add_special_tokens: bool,
+    ) -> vllm_text::tokenizer::Result<Vec<u32>> {
+        Ok(text.bytes().map(u32::from).collect())
+    }
+
+    fn decode(
+        &self,
+        token_ids: &[u32],
+        _skip_special_tokens: bool,
+    ) -> vllm_text::tokenizer::Result<String> {
+        Ok(
+            String::from_utf8_lossy(&token_ids.iter().map(|id| *id as u8).collect::<Vec<_>>())
+                .into_owned(),
+        )
+    }
+
+    fn token_to_id(&self, token: &str) -> Option<u32> {
+        token.bytes().next().map(u32::from)
+    }
+}
+
 impl TextBackend for FakeTextBackend {
    fn tokenizer(&self) -> DynTokenizer {
-        Arc::new(TestTokenizer::new())
+        Arc::new(FakeTokenizer)
    }

    fn model_id(&self) -> &str {
@@ -24,9 +24,8 @@ use vllm_engine_core_client::protocol::{
 use vllm_engine_core_client::test_utils::{IpcNamespace, spawn_mock_engine_task};
 use vllm_engine_core_client::{EngineCoreClient, EngineCoreClientConfig, EngineId};
 use vllm_llm::Llm;
-use vllm_text::tokenizer::DynTokenizer;
+use vllm_text::tokenizer::{DynTokenizer, Tokenizer};
 use vllm_text::{Prompt, TextBackend};
-use vllm_tokenizer::test_utils::TestTokenizer;
 use zeromq::prelude::{SocketRecv, SocketSend};
 use zeromq::{DealerSocket, PushSocket, ZmqMessage};

@@ -152,9 +151,37 @@ fn test_llm(client: EngineCoreClient) -> Llm {
 #[derive(Clone, Debug)]
 struct FakeChatBackend;

+#[derive(Debug)]
+struct FakeChatTokenizer;
+
+impl Tokenizer for FakeChatTokenizer {
+    fn encode(
+        &self,
+        text: &str,
+        _add_special_tokens: bool,
+    ) -> vllm_text::tokenizer::Result<Vec<u32>> {
+        Ok(text.bytes().map(u32::from).collect())
+    }
+
+    fn decode(
+        &self,
+        token_ids: &[u32],
+        _skip_special_tokens: bool,
+    ) -> vllm_text::tokenizer::Result<String> {
+        Ok(
+            String::from_utf8_lossy(&token_ids.iter().map(|id| *id as u8).collect::<Vec<_>>())
+                .into_owned(),
+        )
+    }
+
+    fn token_to_id(&self, token: &str) -> Option<u32> {
+        token.bytes().next().map(u32::from)
+    }
+}
+
 impl TextBackend for FakeChatBackend {
    fn tokenizer(&self) -> DynTokenizer {
-        Arc::new(TestTokenizer::new())
+        Arc::new(FakeChatTokenizer)
    }

    fn model_id(&self) -> &str {
@@ -194,7 +194,7 @@ mod tests {
    use axum::http::HeaderMap;
    use serde_json::json;
    use vllm_text::Prompt;
-    use vllm_tokenizer::test_utils::TestTokenizer;
+    use vllm_text::tokenizer::Tokenizer;

    use super::prepare_completion_request;
    use crate::lora::LoraModelResolution;
@@ -212,8 +212,32 @@ mod tests {
        }
    }

-    fn test_tokenizer() -> TestTokenizer {
-        TestTokenizer::new()
+    #[derive(Debug)]
+    struct TestTokenizer;
+
+    impl Tokenizer for TestTokenizer {
+        fn encode(
+            &self,
+            text: &str,
+            _add_special_tokens: bool,
+        ) -> vllm_text::tokenizer::Result<Vec<u32>> {
+            Ok(text.bytes().map(u32::from).collect())
+        }
+
+        fn decode(
+            &self,
+            token_ids: &[u32],
+            _skip_special_tokens: bool,
+        ) -> vllm_text::tokenizer::Result<String> {
+            Ok(
+                String::from_utf8_lossy(&token_ids.iter().map(|id| *id as u8).collect::<Vec<_>>())
+                    .into_owned(),
+            )
+        }
+
+        fn token_to_id(&self, _token: &str) -> Option<u32> {
+            None
+        }
    }

    fn base_request_json() -> serde_json::Value {
@@ -273,7 +297,7 @@ mod tests {
            request,
            &served(&["Qwen/Qwen1.5-0.5B-Chat"]),
            ResolvedRequestContext::default(),
-            &test_tokenizer(),
+            &TestTokenizer,
        )
        .expect("prepare");

@@ -316,7 +340,7 @@ mod tests {
                request,
                &served(&["Qwen/Qwen1.5-0.5B-Chat"]),
                ResolvedRequestContext::default(),
-                &test_tokenizer(),
+                &TestTokenizer,
            )
            .expect("prepare")
            .text_request
@@ -350,7 +374,7 @@ mod tests {
            request,
            &served(&["Qwen/Qwen1.5-0.5B-Chat"]),
            ResolvedRequestContext::default(),
-            &test_tokenizer(),
+            &TestTokenizer,
        )
        .expect("prepare");

@@ -375,7 +399,7 @@ mod tests {
            request,
            &served(&["Qwen/Qwen1.5-0.5B-Chat"]),
            ResolvedRequestContext::default(),
-            &test_tokenizer(),
+            &TestTokenizer,
        )
        .expect("prepare");

@@ -398,7 +422,7 @@ mod tests {
            request,
            &served(&["Qwen/Qwen1.5-0.5B-Chat"]),
            ResolvedRequestContext::default(),
-            &test_tokenizer(),
+            &TestTokenizer,
        )
        .expect("prepare");

@@ -422,7 +446,7 @@ mod tests {
            request,
            &served(&["Qwen/Qwen1.5-0.5B-Chat"]),
            ResolvedRequestContext::default(),
-            &test_tokenizer(),
+            &TestTokenizer,
        )
        .expect("prepare");

@@ -447,7 +471,7 @@ mod tests {
            request,
            &served(&["Qwen/Qwen1.5-0.5B-Chat"]),
            ResolvedRequestContext::default(),
-            &test_tokenizer(),
+            &TestTokenizer,
        )
        .expect("prepare");

@@ -474,7 +498,7 @@ mod tests {
            request,
            &served(&["Qwen/Qwen1.5-0.5B-Chat"]),
            ResolvedRequestContext::default(),
-            &test_tokenizer(),
+            &TestTokenizer,
        )
        .expect("prepare");

@@ -499,7 +523,7 @@ mod tests {
            request,
            &served(&["Qwen/Qwen1.5-0.5B-Chat"]),
            ResolvedRequestContext::default(),
-            &test_tokenizer(),
+            &TestTokenizer,
        )
        .expect("prepare");

@@ -525,7 +549,7 @@ mod tests {
            request,
            &served(&["Qwen/Qwen1.5-0.5B-Chat"]),
            ResolvedRequestContext::default(),
-            &test_tokenizer(),
+            &TestTokenizer,
        )
        .expect("prepare");
        assert_eq!(prepared.text_request.sampling_params.logprobs, Some(1));
@@ -550,7 +574,7 @@ mod tests {
            request,
            &served(&["Qwen/Qwen1.5-0.5B-Chat"]),
            request_context(&headers, None),
-            &test_tokenizer(),
+            &TestTokenizer,
        )
        .expect("prepare");
        assert_eq!(prepared.text_request.data_parallel_rank, Some(3));
@@ -569,7 +593,7 @@ mod tests {
            request,
            &served(&["Qwen/Qwen1.5-0.5B-Chat"]),
            ResolvedRequestContext::default(),
-            &test_tokenizer(),
+            &TestTokenizer,
        )
        .expect("prepare");
        assert_eq!(prepared.text_request.data_parallel_rank, None);
@@ -40,9 +40,8 @@ use vllm_engine_core_client::{
 };
 use vllm_llm::Llm;
 use vllm_metrics::METRICS;
-use vllm_text::tokenizer::DynTokenizer;
+use vllm_text::tokenizer::{DynTokenizer, Tokenizer};
 use vllm_text::{Prompt, TextBackend};
-use vllm_tokenizer::test_utils::TestTokenizer;
 use zeromq::prelude::{SocketRecv, SocketSend};
 use zeromq::{DealerSocket, PushSocket, ZmqMessage};

@@ -418,20 +417,82 @@ struct FakeChatBackend {
 }

 /// Synthetic BOS id used when `add_special_tokens` is true in tests.
-const FAKE_BOS_TOKEN_ID: u32 = 256;
-const UNKNOWN_DECODE_TOKEN_ID: u32 = 10_000;
+const FAKE_BOS_TOKEN_ID: u32 = 1;

-fn fake_chat_tokenizer() -> TestTokenizer {
-    TestTokenizer::new()
-        .with_bos_token("<bos>", FAKE_BOS_TOKEN_ID)
-        .with_regular_token("<image>", 999)
-        .with_regular_token("<|image_pad|>", 151655)
-        .with_regular_token("<think>", 0xF001)
-        .with_regular_token("</think>", 0xF002)
-        .with_regular_token("<|START_THINKING|>", 0xF003)
-        .with_regular_token("<|END_THINKING|>", 0xF004)
-        .with_regular_token("◁think▷", 0xF005)
-        .with_regular_token("◁/think▷", 0xF006)
+#[derive(Debug)]
+struct FakeChatTokenizer;
+
+impl Tokenizer for FakeChatTokenizer {
+    fn encode(
+        &self,
+        text: &str,
+        add_special_tokens: bool,
+    ) -> vllm_text::tokenizer::Result<Vec<u32>> {
+        let mut token_ids = Vec::new();
+        if add_special_tokens {
+            token_ids.push(FAKE_BOS_TOKEN_ID);
+        }
+        let mut rest = text;
+        while !rest.is_empty() {
+            if let Some(stripped) = rest.strip_prefix("<image>") {
+                token_ids.push(999);
+                rest = stripped;
+                continue;
+            }
+            if let Some(stripped) = rest.strip_prefix("<|image_pad|>") {
+                token_ids.push(151655);
+                rest = stripped;
+                continue;
+            }
+
+            let ch = rest.chars().next().expect("rest is not empty");
+            let mut buf = [0; 4];
+            token_ids.extend(ch.encode_utf8(&mut buf).bytes().map(u32::from));
+            rest = &rest[ch.len_utf8()..];
+        }
+        Ok(token_ids)
+    }
+
+    fn decode(
+        &self,
+        token_ids: &[u32],
+        _skip_special_tokens: bool,
+    ) -> vllm_text::tokenizer::Result<String> {
+        Ok(
+            String::from_utf8_lossy(&token_ids.iter().map(|id| *id as u8).collect::<Vec<_>>())
+                .into_owned(),
+        )
+    }
+
+    fn token_to_id(&self, token: &str) -> Option<u32> {
+        match token {
+            "<image>" => Some(999),
+            "<|image_pad|>" => Some(151655),
+            "<think>" => Some(0xF001),
+            "</think>" => Some(0xF002),
+            "<|START_THINKING|>" => Some(0xF003),
+            "<|END_THINKING|>" => Some(0xF004),
+            "◁think▷" => Some(0xF005),
+            "◁/think▷" => Some(0xF006),
+            _ => None,
+        }
+    }
+
+    fn id_to_token(&self, id: u32) -> Option<String> {
+        match id {
+            FAKE_BOS_TOKEN_ID => Some("<bos>".to_string()),
+            999 => Some("<image>".to_string()),
+            151655 => Some("<|image_pad|>".to_string()),
+            0xF001 => Some("<think>".to_string()),
+            0xF002 => Some("</think>".to_string()),
+            0xF003 => Some("<|START_THINKING|>".to_string()),
+            0xF004 => Some("<|END_THINKING|>".to_string()),
+            0xF005 => Some("◁think▷".to_string()),
+            0xF006 => Some("◁/think▷".to_string()),
+            id if id < 128 => char::from_u32(id).map(|ch| ch.to_string()),
+            _ => None,
+        }
+    }
 }

 impl FakeChatBackend {
@@ -469,7 +530,7 @@ impl fmt::Debug for FakeChatBackend {

 impl TextBackend for FakeChatBackend {
    fn tokenizer(&self) -> DynTokenizer {
-        Arc::new(fake_chat_tokenizer())
+        Arc::new(FakeChatTokenizer)
    }

    fn model_id(&self) -> &str {
@@ -569,7 +630,7 @@ fn qwen_multimodal_model_info() -> vllm_chat::multimodal::MultimodalModelInfo {
        Some("qwen2_vl".to_string()),
        Some(&config_path),
        None,
-        Arc::new(fake_chat_tokenizer()),
+        Arc::new(FakeChatTokenizer),
    )
    .expect("load multimodal info")
    .expect("qwen multimodal info is registered");
@@ -577,6 +638,70 @@ fn qwen_multimodal_model_info() -> vllm_chat::multimodal::MultimodalModelInfo {
    info
 }

+#[derive(Clone, Debug)]
+struct FailingDecodeChatBackend;
+
+#[derive(Debug)]
+struct FailingDecodeTokenizer;
+
+impl Tokenizer for FailingDecodeTokenizer {
+    fn encode(
+        &self,
+        text: &str,
+        add_special_tokens: bool,
+    ) -> vllm_text::tokenizer::Result<Vec<u32>> {
+        FakeChatTokenizer.encode(text, add_special_tokens)
+    }
+
+    fn decode(
+        &self,
+        token_ids: &[u32],
+        skip_special_tokens: bool,
+    ) -> vllm_text::tokenizer::Result<String> {
+        if token_ids.contains(&(b'i' as u32)) {
+            return Err(vllm_text::tokenizer::TokenizerError(
+                "forced decode failure for streaming test".to_string(),
+            ));
+        }
+
+        FakeChatTokenizer.decode(token_ids, skip_special_tokens)
+    }
+
+    fn token_to_id(&self, token: &str) -> Option<u32> {
+        FakeChatTokenizer.token_to_id(token)
+    }
+}
+
+impl TextBackend for FailingDecodeChatBackend {
+    fn tokenizer(&self) -> DynTokenizer {
+        Arc::new(FailingDecodeTokenizer)
+    }
+
+    fn model_id(&self) -> &str {
+        "test-model"
+    }
+}
+
+impl ChatBackend for FailingDecodeChatBackend {
+    fn chat_renderer(&self) -> DynChatRenderer {
+        Arc::new(self.clone())
+    }
+
+    fn new_chat_output_processor(
+        &self,
+        _request: &mut ChatRequest,
+        _options: NewChatOutputProcessorOptions<'_>,
+    ) -> vllm_chat::Result<DynChatOutputProcessor> {
+        Ok(Box::new(DefaultChatOutputProcessor::plain_text_only()))
+    }
+}
+
+impl ChatRenderer for FailingDecodeChatBackend {
+    fn render(&self, request: &ChatRequest) -> vllm_chat::Result<vllm_chat::RenderedPrompt> {
+        FakeChatBackend::new().render(request)
+    }
+}
+
 async fn test_models_with_engine_outputs_and_backend_inner(
    engine_id: impl Into<EngineId>,
    output_specs: Vec<(Vec<u32>, Option<EngineCoreFinishReason>)>,
@@ -2630,8 +2755,8 @@ async fn load_endpoint_resets_when_stream_response_is_dropped() {
 #[serial]
 async fn stream_error_is_returned_as_openai_error_sse() {
    let (app, engine_task) = test_app_with_backend_and_stream_output_specs(
-        Arc::new(FakeChatBackend::new()),
-        vec![(vec![UNKNOWN_DECODE_TOKEN_ID], None)],
+        Arc::new(FailingDecodeChatBackend),
+        default_stream_output_specs(),
    )
    .await;
    let response = app
@@ -2664,9 +2789,7 @@ async fn stream_error_is_returned_as_openai_error_sse() {
    assert!(text.contains("\"role\":\"assistant\""), "{text}");
    assert!(text.contains("\"type\":\"server_error\""), "{text}");
    assert!(
-        text.contains(&format!(
-            "test tokenizer cannot decode unknown token id {UNKNOWN_DECODE_TOKEN_ID}"
-        )),
+        text.contains("forced decode failure for streaming test"),
        "{text}"
    );
    assert!(!text.contains("\"usage\":"), "{text}");
@@ -31,7 +31,6 @@ serial_test.workspace = true
 tempfile.workspace = true
 tokio.workspace = true
 vllm-llm = { workspace = true, features = ["test-util"] }
-vllm-tokenizer = { workspace = true, features = ["test-utils"] }

 [lints]
 workspace = true
@@ -275,7 +275,6 @@ mod tests {
    use std::collections::{BTreeSet, HashMap};

    use serial_test::file_serial;
-    use vllm_tokenizer::test_utils::TestTokenizer;

    use super::*;
    use crate::backend::hf::HfTextBackend;
@@ -283,8 +282,60 @@ mod tests {
    use crate::error::{LogprobsError, TokenIdsError};
    use crate::request::{Prompt, TextRequest};

-    fn stub_tokenizer() -> TestTokenizer {
-        TestTokenizer::new()
+    /// Stub tokenizer that returns empty token IDs — sufficient for tests that
+    /// don't exercise bad-words tokenization.
+    struct StubTokenizer;
+
+    impl Tokenizer for StubTokenizer {
+        fn encode(
+            &self,
+            _text: &str,
+            _add_special_tokens: bool,
+        ) -> vllm_tokenizer::Result<Vec<u32>> {
+            Ok(vec![])
+        }
+
+        fn decode(
+            &self,
+            _token_ids: &[u32],
+            _skip_special_tokens: bool,
+        ) -> vllm_tokenizer::Result<String> {
+            Ok(String::new())
+        }
+
+        fn token_to_id(&self, _token: &str) -> Option<u32> {
+            None
+        }
+    }
+
+    fn stub_tokenizer() -> StubTokenizer {
+        StubTokenizer
+    }
+
+    struct FixedTokenizer {
+        token_ids: Vec<u32>,
+    }
+
+    impl Tokenizer for FixedTokenizer {
+        fn encode(
+            &self,
+            _text: &str,
+            _add_special_tokens: bool,
+        ) -> vllm_tokenizer::Result<Vec<u32>> {
+            Ok(self.token_ids.clone())
+        }
+
+        fn decode(
+            &self,
+            _token_ids: &[u32],
+            _skip_special_tokens: bool,
+        ) -> vllm_tokenizer::Result<String> {
+            Ok(String::new())
+        }
+
+        fn token_to_id(&self, _token: &str) -> Option<u32> {
+            None
+        }
    }

    fn sample_request() -> TextRequest {
@@ -901,7 +952,9 @@ mod tests {

    #[test]
    fn lower_sampling_params_rejects_out_of_vocab_bad_words() {
-        let tokenizer = TestTokenizer::new().with_regular_token("blocked", 2000);
+        let tokenizer = FixedTokenizer {
+            token_ids: vec![1999, 2000],
+        };
        let error = lower_sampling_params(
            SamplingParams {
                bad_words: Some(vec!["blocked".to_string()]),
@@ -323,11 +323,37 @@ mod tests {
    use futures::{Stream, stream};
    use vllm_engine_core_client::AbortCause;
    use vllm_llm::GenerateOutput;
-    use vllm_tokenizer::test_utils::TestTokenizer;
+    use vllm_tokenizer::Tokenizer;

    use super::*;
    use crate::output::TextOutputStreamExt as _;

+    /// Backend that treats each token ID as a raw byte, producing lossy UTF-8.
+    struct ByteTokenizer;
+
+    impl Tokenizer for ByteTokenizer {
+        fn encode(
+            &self,
+            _text: &str,
+            _add_special_tokens: bool,
+        ) -> vllm_tokenizer::Result<Vec<u32>> {
+            unreachable!()
+        }
+
+        fn decode(
+            &self,
+            token_ids: &[u32],
+            _skip_special_tokens: bool,
+        ) -> vllm_tokenizer::Result<String> {
+            let bytes = token_ids.iter().map(|id| *id as u8).collect::<Vec<_>>();
+            Ok(String::from_utf8_lossy(&bytes).into_owned())
+        }
+
+        fn token_to_id(&self, _token: &str) -> Option<u32> {
+            unreachable!()
+        }
+    }
+
    /// Helper: run `decoded_text_event_stream` to completion and return the
    /// collected output.
    async fn run_to_completion(
@@ -340,7 +366,7 @@ mod tests {
            token_ids,
            Some(FinishReason::Length),
        ))]);
-        let tokenizer: DynTokenizer = Arc::new(TestTokenizer::new());
+        let tokenizer: DynTokenizer = Arc::new(ByteTokenizer);
        decoded_text_event_stream("test".into(), tokenizer, raw_stream, decode_options, false)
            .collect_output()
            .await
@@ -393,7 +419,7 @@ mod tests {
            ))),
            dropped_cause: Arc::clone(&dropped_cause),
        };
-        let tokenizer: DynTokenizer = Arc::new(TestTokenizer::new());
+        let tokenizer: DynTokenizer = Arc::new(ByteTokenizer);

        let output = decoded_text_event_stream(
            "test".into(),
@@ -129,13 +129,40 @@ fn decode_position_logprobs<T: Tokenizer + ?Sized>(
 #[cfg(test)]
 mod tests {
    use vllm_llm::{Logprobs, PositionLogprobs, TokenLogprob};
-    use vllm_tokenizer::test_utils::TestTokenizer;

    use super::*;

+    #[derive(Debug)]
+    struct ByteTokenizer;
+
+    impl vllm_tokenizer::Tokenizer for ByteTokenizer {
+        fn encode(
+            &self,
+            _text: &str,
+            _add_special_tokens: bool,
+        ) -> vllm_tokenizer::Result<Vec<u32>> {
+            unreachable!()
+        }
+
+        fn decode(
+            &self,
+            token_ids: &[u32],
+            _skip_special_tokens: bool,
+        ) -> vllm_tokenizer::Result<String> {
+            Ok(String::from_utf8_lossy(
+                &token_ids.iter().map(|token_id| *token_id as u8).collect::<Vec<_>>(),
+            )
+            .into_owned())
+        }
+
+        fn token_to_id(&self, _token: &str) -> Option<u32> {
+            unreachable!()
+        }
+    }
+
    #[test]
    fn decode_logprobs_decodes_every_candidate_token() {
-        let tokenizer = TestTokenizer::new();
+        let tokenizer = ByteTokenizer;
        let logprobs = Logprobs {
            positions: vec![PositionLogprobs {
                entries: vec![
@@ -178,7 +205,7 @@ mod tests {

    #[test]
    fn decode_prompt_logprobs_separates_first_prompt_token() {
-        let tokenizer = TestTokenizer::new();
+        let tokenizer = ByteTokenizer;
        let logprobs = Logprobs {
            positions: vec![PositionLogprobs {
                entries: vec![TokenLogprob {
@@ -4,9 +4,6 @@ version.workspace = true
 edition.workspace = true
 license.workspace = true

-[features]
-test-utils = []
-
 [dependencies]
 base64.workspace = true
 fastokens.workspace = true
@@ -32,7 +32,6 @@ pub(crate) struct DecodeStream<'a, T: Tokenizer + ?Sized> {
    ids: Vec<u32>,
    prefix: String,
    prefix_index: usize,
-    prefix_seeded: bool,
    cumulative_output: String,
    output_index: usize,
 }
@@ -51,7 +50,6 @@ impl<'a, T: Tokenizer + ?Sized> DecodeStream<'a, T> {
            ids: prompt_token_ids.to_vec(),
            prefix: String::new(),
            prefix_index: 0,
-            prefix_seeded: prompt_token_ids.is_empty(),
            cumulative_output: String::new(),
            output_index: 0,
        }
@@ -65,48 +63,29 @@ const SAFE_SUFFIX_MIN: usize = 4;
 const SAFE_SUFFIX_MAX: usize = 6;

 impl<T: Tokenizer + ?Sized> DecodeStream<'_, T> {
-    /// Return prompt-context ids that have a tokenizer-local raw token string.
-    ///
-    /// `DecodeStream` uses prompt ids only to seed left context before the
-    /// first generated token. Some prompt ids may come from a wider model
-    /// vocabulary than the local tokenizer can decode, so context seeding drops
-    /// ids that [`Tokenizer::id_to_token`] cannot resolve before calling strict
-    /// [`Tokenizer::decode`]. Generated token ids keep the normal strict decode
-    /// path.
-    fn decodable_context_ids(&self, ids: &[u32]) -> Vec<u32> {
-        ids.iter()
-            .copied()
-            .filter(|&id| self.tokenizer.id_to_token(id).is_some())
-            .collect()
-    }
-
    /// Seed `self.prefix` from the shortest trailing suffix whose decoded text
    /// has no U+FFFD — a clean decode means the suffix starts and ends at
    /// valid UTF-8/token boundaries, so priming from it is equivalent to
-    /// priming from the full prompt. Prompt-only ids that the tokenizer cannot
-    /// map back to token text are ignored for decode context; generated ids
-    /// still go through strict decode.
+    /// priming from the full prompt.
    fn seed_prefix(&mut self) -> Result<()> {
        let prompt_len = self.ids.len();
        if prompt_len > SAFE_SUFFIX_MIN {
            let max_try = SAFE_SUFFIX_MAX.min(prompt_len - 1);
            for suffix_len in SAFE_SUFFIX_MIN..=max_try {
                let start = prompt_len - suffix_len;
-                let candidate = self.decodable_context_ids(&self.ids[start..]);
-                let decoded = self.tokenizer.decode(&candidate, self.skip_special_tokens)?;
+                let decoded =
+                    self.tokenizer.decode(&self.ids[start..], self.skip_special_tokens)?;
                if !decoded.contains('\u{FFFD}') {
                    self.prefix = decoded;
-                    self.ids = candidate;
+                    self.ids.drain(..start);
                    self.prefix_index = self.ids.len();
                    return Ok(());
                }
            }
        }
-        let candidate = self.decodable_context_ids(&self.ids);
-        let decoded = self.tokenizer.decode(&candidate, self.skip_special_tokens)?;
+        let decoded = self.tokenizer.decode(&self.ids, self.skip_special_tokens)?;
        if !decoded.ends_with('\u{FFFD}') {
            self.prefix = decoded;
-            self.ids = candidate;
            self.prefix_index = self.ids.len();
        }
        Ok(())
@@ -115,9 +94,8 @@ impl<T: Tokenizer + ?Sized> DecodeStream<'_, T> {

 impl<T: Tokenizer + ?Sized> IncrementalDecoder for DecodeStream<'_, T> {
    fn push_token(&mut self, token_id: u32) -> Result<usize> {
-        if !self.prefix_seeded && !self.ids.is_empty() {
+        if self.prefix.is_empty() && !self.ids.is_empty() {
            self.seed_prefix()?;
-            self.prefix_seeded = true;
        }

        self.ids.push(token_id);
@@ -153,7 +131,6 @@ impl<T: Tokenizer + ?Sized> IncrementalDecoder for DecodeStream<'_, T> {
            self.ids.clear();
            self.prefix.clear();
            self.prefix_index = 0;
-            self.prefix_seeded = true;
            // Ensure we split at a utf-8 char boundary.
            self.cumulative_output
                .push_str(&string[string.floor_char_boundary(prefix_len)..]);
@@ -175,7 +152,6 @@ impl<T: Tokenizer + ?Sized> IncrementalDecoder for DecodeStream<'_, T> {
 #[cfg(test)]
 mod tests {
    use super::*;
-    use crate::test_utils::TestTokenizer;

    /// Backend that treats each token ID as a raw byte, producing lossy UTF-8.
    #[derive(Debug)]
@@ -194,13 +170,6 @@ mod tests {
        fn token_to_id(&self, _token: &str) -> Option<u32> {
            unreachable!()
        }
-
-        fn id_to_token(&self, id: u32) -> Option<String> {
-            u8::try_from(id).ok().map(|byte| {
-                let bytes = [byte];
-                String::from_utf8_lossy(&bytes).into_owned()
-            })
-        }
    }

    #[test]
@@ -279,10 +248,6 @@ mod tests {
        fn token_to_id(&self, _token: &str) -> Option<u32> {
            unreachable!()
        }
-
-        fn id_to_token(&self, _id: u32) -> Option<String> {
-            unreachable!()
-        }
    }

    #[test]
@@ -308,47 +273,6 @@ mod tests {
        assert_eq!(decoder.output(), "!");
    }

-    #[test]
-    fn seed_prefix_filters_unknown_prompt_ids_from_suffix_context() {
-        let tokenizer = TestTokenizer::new();
-        let prompt = &[
-            b'a' as u32,
-            b'b' as u32,
-            b'c' as u32,
-            10_000,
-            b'H' as u32,
-            b'i' as u32,
-        ];
-        let mut decoder = tokenizer.create_decode_stream(prompt, false, 0);
-
-        assert_eq!(decoder.push_token(b'!' as u32).unwrap(), 1);
-        assert_eq!(decoder.output(), "!");
-    }
-
-    #[test]
-    fn seed_prefix_filters_unknown_prompt_ids_from_full_prompt_fallback() {
-        let tokenizer = TestTokenizer::new();
-        let prompt = &[10_000, b'H' as u32, b'i' as u32];
-        let mut decoder = tokenizer.create_decode_stream(prompt, false, 0);
-
-        assert_eq!(decoder.push_token(b'!' as u32).unwrap(), 1);
-        assert_eq!(decoder.output(), "!");
-    }
-
-    #[test]
-    fn generated_unknown_ids_still_return_decode_error() {
-        let tokenizer = TestTokenizer::new();
-        let prompt = &[10_000, b'H' as u32, b'i' as u32];
-        let mut decoder = tokenizer.create_decode_stream(prompt, false, 0);
-
-        let error = decoder.push_token(10_000).unwrap_err();
-        assert!(
-            error
-                .to_string()
-                .contains("test tokenizer cannot decode unknown token id 10000")
-        );
-    }
-
    #[test]
    fn chunks_concatenate_to_full_text() {
        let backend = Utf8Backend;
@@ -396,10 +320,6 @@ mod tests {
        fn token_to_id(&self, _token: &str) -> Option<u32> {
            unreachable!()
        }
-
-        fn id_to_token(&self, _id: u32) -> Option<String> {
-            unreachable!()
-        }
    }

    /// Without the char-boundary fix, this panics slicing mid-emoji.
@@ -8,8 +8,6 @@ mod error;
 mod hf;
 mod incremental;
 mod tekken;
-#[cfg(any(test, feature = "test-utils"))]
-pub mod test_utils;
 mod tiktoken;

 pub use error::{Result, TokenizerError};
@@ -30,7 +28,11 @@ pub trait Tokenizer: Send + Sync {
    fn token_to_id(&self, token: &str) -> Option<u32>;

    /// Convert one token ID into the tokenizer's raw token string.
-    fn id_to_token(&self, id: u32) -> Option<String>;
+    fn id_to_token(&self, _id: u32) -> Option<String> {
+        // TODO: remove default impl and require this to be implemented by all
+        // tokenizers
+        None
+    }

    /// Return the vocabulary size. Backends that cannot report it fall back to
    /// `usize::MAX`, an effectively unbounded value used only by test stubs.
@@ -1,434 +0,0 @@
-use std::collections::BTreeMap;
-
-use crate::{Result, Tokenizer, TokenizerError};
-
-const FIRST_CONFIGURED_TOKEN_ID: u32 = 256;
-
-/// Whether a configured test token should be treated as special.
-///
-/// Special tokens are skipped by [`Tokenizer::decode`] when
-/// `skip_special_tokens` is set. Regular configured tokens are always emitted.
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum TestTokenKind {
-    /// Token is skipped when `skip_special_tokens = true`.
-    Special,
-    /// Token is emitted regardless of `skip_special_tokens`.
-    Regular,
-}
-
-impl TestTokenKind {
-    fn is_special(self) -> bool {
-        matches!(self, Self::Special)
-    }
-}
-
-/// Decode behavior for token ids that are neither configured tokens nor byte ids.
-///
-/// The default is [`UnknownDecode::Error`] so tests notice missing tokenizer
-/// fixtures instead of silently accepting impossible ids. Individual tests can
-/// opt into empty or replacement output when they are explicitly modeling a
-/// lenient detokenization path.
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum UnknownDecode {
-    /// Return a tokenizer error on the first unknown id.
-    Error,
-    /// Drop unknown ids from decoded output.
-    Empty,
-    /// Emit U+FFFD for each unknown id.
-    Replacement,
-}
-
-#[derive(Debug, Clone)]
-struct TestToken {
-    text: String,
-    kind: TestTokenKind,
-}
-
-/// Configurable tokenizer for Rust frontend tests.
-///
-/// `TestTokenizer` is intentionally small, but its methods obey the same basic
-/// contract as production tokenizers:
-///
-/// - ordinary text encodes as UTF-8 byte ids;
-/// - configured token ids start at 256, leaving `0..=255` for byte fallback;
-/// - configured token ids and token text are unique;
-/// - configured tokens are matched before ordinary bytes, using longest-prefix matching so
-///   multi-character markers such as `<think>` work naturally;
-/// - `token_to_id` and `id_to_token` are consistent for configured tokens;
-/// - `decode` is strict by default for ids outside the byte range and the configured token table;
-/// - `vocab_size` is an exclusive upper bound covering byte ids and configured token ids unless a
-///   test sets it explicitly.
-///
-/// Prefer this helper over ad-hoc fake tokenizers for tests that rely on
-/// tokenizer semantics. Keep dedicated tiny fakes for error injection or for
-/// tests that deliberately need a degenerate tokenizer.
-#[derive(Debug, Clone)]
-pub struct TestTokenizer {
-    token_to_id: BTreeMap<String, u32>,
-    id_to_token: BTreeMap<u32, TestToken>,
-    unknown_decode: UnknownDecode,
-    vocab_size: Option<usize>,
-    bos_token_id: Option<u32>,
-}
-
-impl Default for TestTokenizer {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-impl TestTokenizer {
-    /// Create a byte-level test tokenizer with strict unknown-id decode.
-    pub fn new() -> Self {
-        Self {
-            token_to_id: BTreeMap::new(),
-            id_to_token: BTreeMap::new(),
-            unknown_decode: UnknownDecode::Error,
-            vocab_size: None,
-            bos_token_id: None,
-        }
-    }
-
-    /// Add a configured token and return the updated tokenizer.
-    ///
-    /// Configured tokens must use ids outside the byte range and may be marked
-    /// special or regular.
-    pub fn with_token(mut self, token: impl Into<String>, id: u32, kind: TestTokenKind) -> Self {
-        self.insert_token(token, id, kind);
-        self
-    }
-
-    /// Add a special configured token and return the updated tokenizer.
-    pub fn with_special_token(self, token: impl Into<String>, id: u32) -> Self {
-        self.with_token(token, id, TestTokenKind::Special)
-    }
-
-    /// Add a regular configured token and return the updated tokenizer.
-    pub fn with_regular_token(self, token: impl Into<String>, id: u32) -> Self {
-        self.with_token(token, id, TestTokenKind::Regular)
-    }
-
-    /// Add a special BOS token inserted by `encode(..., true)`.
-    ///
-    /// This also registers the token in the normal token/id maps so
-    /// `token_to_id`, `id_to_token`, `decode`, and `is_special_id` stay
-    /// consistent for the inserted id.
-    pub fn with_bos_token(mut self, token: impl Into<String>, id: u32) -> Self {
-        self.insert_token(token, id, TestTokenKind::Special);
-        self.bos_token_id = Some(id);
-        self
-    }
-
-    /// Set decode behavior for unknown non-byte ids.
-    pub fn with_unknown_decode(mut self, behavior: UnknownDecode) -> Self {
-        self.unknown_decode = behavior;
-        self
-    }
-
-    /// Set an explicit vocabulary size.
-    ///
-    /// Use this when a test needs a model-like vocabulary bound that differs
-    /// from the highest configured token id plus one.
-    pub fn with_vocab_size(mut self, vocab_size: usize) -> Self {
-        self.vocab_size = Some(vocab_size);
-        self
-    }
-
-    fn insert_token(&mut self, token: impl Into<String>, id: u32, kind: TestTokenKind) {
-        let token = token.into();
-        assert!(
-            !token.is_empty(),
-            "configured test token text must be non-empty"
-        );
-        assert!(
-            id >= FIRST_CONFIGURED_TOKEN_ID,
-            "configured test token id {id} overlaps byte fallback range 0..=255"
-        );
-        assert!(
-            token.len() > 1,
-            "configured test token text {token:?} overlaps byte fallback token text"
-        );
-        if self.token_to_id.insert(token.clone(), id).is_some() {
-            panic!("configured test token text {token:?} was registered more than once");
-        }
-        if self.id_to_token.insert(id, TestToken { text: token, kind }).is_some() {
-            panic!("configured test token id {id} was registered more than once");
-        }
-    }
-
-    fn byte_to_token(id: u32) -> Option<String> {
-        u8::try_from(id).ok().map(|byte| String::from_utf8_lossy(&[byte]).into_owned())
-    }
-
-    fn flush_bytes(bytes: &mut Vec<u8>, output: &mut String) {
-        if !bytes.is_empty() {
-            output.push_str(&String::from_utf8_lossy(bytes));
-            bytes.clear();
-        }
-    }
-
-    fn configured_token_prefix(&self, text: &str) -> Option<(&str, u32)> {
-        self.token_to_id
-            .iter()
-            .filter_map(|(token, &id)| text.starts_with(token).then_some((token.as_str(), id)))
-            .max_by_key(|(token, _)| token.len())
-    }
-
-    fn inferred_vocab_size(&self) -> usize {
-        let max_configured =
-            self.id_to_token.last_key_value().map(|(&id, _)| id as usize + 1).unwrap_or(0);
-        256.max(max_configured)
-    }
-}
-
-impl Tokenizer for TestTokenizer {
-    fn encode(&self, text: &str, add_special_tokens: bool) -> Result<Vec<u32>> {
-        let mut ids = Vec::new();
-        if add_special_tokens && let Some(bos_token_id) = self.bos_token_id {
-            ids.push(bos_token_id);
-        }
-
-        let mut rest = text;
-        while !rest.is_empty() {
-            if let Some((token, id)) = self.configured_token_prefix(rest) {
-                ids.push(id);
-                rest = &rest[token.len()..];
-                continue;
-            }
-
-            let ch = rest.chars().next().expect("rest is not empty");
-            let mut buf = [0_u8; 4];
-            ids.extend(ch.encode_utf8(&mut buf).bytes().map(u32::from));
-            rest = &rest[ch.len_utf8()..];
-        }
-
-        Ok(ids)
-    }
-
-    fn decode(&self, token_ids: &[u32], skip_special_tokens: bool) -> Result<String> {
-        let mut output = String::new();
-        let mut pending_bytes = Vec::new();
-        for &id in token_ids {
-            if let Some(token) = self.id_to_token.get(&id) {
-                Self::flush_bytes(&mut pending_bytes, &mut output);
-                if !(skip_special_tokens && token.kind.is_special()) {
-                    output.push_str(&token.text);
-                }
-            } else if let Ok(byte) = u8::try_from(id) {
-                pending_bytes.push(byte);
-            } else {
-                Self::flush_bytes(&mut pending_bytes, &mut output);
-                match self.unknown_decode {
-                    UnknownDecode::Error => {
-                        return Err(TokenizerError(format!(
-                            "test tokenizer cannot decode unknown token id {id}"
-                        )));
-                    }
-                    UnknownDecode::Empty => {}
-                    UnknownDecode::Replacement => output.push('\u{FFFD}'),
-                }
-            }
-        }
-        Self::flush_bytes(&mut pending_bytes, &mut output);
-        Ok(output)
-    }
-
-    fn token_to_id(&self, token: &str) -> Option<u32> {
-        self.token_to_id.get(token).copied().or_else(|| {
-            let bytes = token.as_bytes();
-            (bytes.len() == 1).then(|| u32::from(bytes[0]))
-        })
-    }
-
-    fn id_to_token(&self, id: u32) -> Option<String> {
-        self.id_to_token
-            .get(&id)
-            .map(|token| token.text.clone())
-            .or_else(|| Self::byte_to_token(id))
-    }
-
-    fn vocab_size(&self) -> usize {
-        self.vocab_size.unwrap_or_else(|| self.inferred_vocab_size())
-    }
-
-    fn is_special_id(&self, token_id: u32) -> bool {
-        self.id_to_token.get(&token_id).is_some_and(|token| token.kind.is_special())
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn byte_text_roundtrips_and_reports_byte_ids() {
-        let tokenizer = TestTokenizer::new();
-
-        let ids = tokenizer.encode("hi", false).unwrap();
-        assert_eq!(ids, vec![b'h' as u32, b'i' as u32]);
-        assert_eq!(tokenizer.decode(&ids, false).unwrap(), "hi");
-        assert_eq!(tokenizer.token_to_id("h"), Some(b'h' as u32));
-        assert_eq!(tokenizer.id_to_token(b'h' as u32).as_deref(), Some("h"));
-        assert_eq!(tokenizer.vocab_size(), 256);
-    }
-
-    #[test]
-    fn configured_tokens_use_longest_prefix_matching() {
-        let tokenizer = TestTokenizer::new()
-            .with_regular_token("<image>", 999)
-            .with_regular_token("<image></image>", 1000);
-
-        assert_eq!(
-            tokenizer.encode("a<image></image>b", false).unwrap(),
-            vec![b'a' as u32, 1000, b'b' as u32,]
-        );
-        assert_eq!(
-            tokenizer.decode(&[b'a' as u32, 1000, b'b' as u32], false).unwrap(),
-            "a<image></image>b"
-        );
-        assert_eq!(tokenizer.token_to_id("<image>"), Some(999));
-        assert_eq!(
-            tokenizer.id_to_token(1000).as_deref(),
-            Some("<image></image>")
-        );
-        assert_eq!(tokenizer.vocab_size(), 1001);
-    }
-
-    #[test]
-    fn non_ascii_text_roundtrips_through_buffered_byte_decode() {
-        let tokenizer = TestTokenizer::new();
-        let text = "你好, café, 🚀";
-
-        let ids = tokenizer.encode(text, false).unwrap();
-        assert_eq!(
-            ids,
-            text.as_bytes().iter().copied().map(u32::from).collect::<Vec<_>>()
-        );
-        assert_eq!(tokenizer.decode(&ids, false).unwrap(), text);
-    }
-
-    #[test]
-    fn buffered_byte_decode_flushes_around_configured_tokens() {
-        let tokenizer = TestTokenizer::new()
-            .with_regular_token("<image>", 999)
-            .with_special_token("<skip>", 1000);
-
-        assert_eq!(
-            tokenizer.encode("你<image>好<skip>🚀", false).unwrap(),
-            vec![228, 189, 160, 999, 229, 165, 189, 1000, 240, 159, 154, 128]
-        );
-        assert_eq!(
-            tokenizer
-                .decode(
-                    &[228, 189, 160, 999, 229, 165, 189, 1000, 240, 159, 154, 128],
-                    false
-                )
-                .unwrap(),
-            "你<image>好<skip>🚀"
-        );
-        assert_eq!(
-            tokenizer
-                .decode(
-                    &[228, 189, 160, 999, 229, 165, 189, 1000, 240, 159, 154, 128],
-                    true
-                )
-                .unwrap(),
-            "你<image>好🚀"
-        );
-    }
-
-    #[test]
-    fn invalid_utf8_bytes_decode_lossily_as_a_sequence() {
-        let tokenizer = TestTokenizer::new();
-
-        assert_eq!(tokenizer.decode(&[0xE4, 0xBD], false).unwrap(), "\u{FFFD}");
-        assert_eq!(
-            tokenizer.decode(&[0xFF, b'a' as u32], false).unwrap(),
-            "\u{FFFD}a"
-        );
-    }
-
-    #[test]
-    fn special_tokens_respect_skip_special_tokens() {
-        let tokenizer = TestTokenizer::new()
-            .with_bos_token("<bos>", 256)
-            .with_special_token("<think>", 0xF001)
-            .with_regular_token("</think>", 0xF002);
-
-        assert_eq!(
-            tokenizer.encode("<think>x</think>", true).unwrap(),
-            vec![256, 0xF001, b'x' as u32, 0xF002,]
-        );
-        assert_eq!(
-            tokenizer.decode(&[256, 0xF001, b'x' as u32, 0xF002], false).unwrap(),
-            "<bos><think>x</think>"
-        );
-        assert_eq!(
-            tokenizer.decode(&[256, 0xF001, b'x' as u32, 0xF002], true).unwrap(),
-            "x</think>"
-        );
-        assert!(tokenizer.is_special_id(0xF001));
-        assert!(!tokenizer.is_special_id(0xF002));
-    }
-
-    #[test]
-    #[should_panic(expected = "configured test token id 255 overlaps byte fallback range 0..=255")]
-    fn configured_token_id_must_stay_outside_byte_range() {
-        let _ = TestTokenizer::new().with_regular_token("<token>", 255);
-    }
-
-    #[test]
-    #[should_panic(expected = "configured test token text \"a\" overlaps byte fallback token text")]
-    fn configured_token_text_must_not_shadow_byte_tokens() {
-        let _ = TestTokenizer::new().with_regular_token("a", 256);
-    }
-
-    #[test]
-    #[should_panic(
-        expected = "configured test token text \"<token>\" was registered more than once"
-    )]
-    fn configured_token_text_must_be_unique() {
-        let _ = TestTokenizer::new()
-            .with_regular_token("<token>", 256)
-            .with_regular_token("<token>", 257);
-    }
-
-    #[test]
-    #[should_panic(expected = "configured test token id 256 was registered more than once")]
-    fn configured_token_id_must_be_unique() {
-        let _ = TestTokenizer::new()
-            .with_regular_token("<token-a>", 256)
-            .with_regular_token("<token-b>", 256);
-    }
-
-    #[test]
-    fn unknown_decode_is_strict_by_default_and_configurable() {
-        let strict = TestTokenizer::new();
-        assert!(strict.decode(&[300], false).is_err());
-        assert_eq!(
-            TestTokenizer::new()
-                .with_unknown_decode(UnknownDecode::Empty)
-                .decode(&[b'a' as u32, 300, b'b' as u32], false)
-                .unwrap(),
-            "ab"
-        );
-        assert_eq!(
-            TestTokenizer::new()
-                .with_unknown_decode(UnknownDecode::Replacement)
-                .decode(&[300], false)
-                .unwrap(),
-            "\u{FFFD}"
-        );
-        assert_eq!(strict.id_to_token(300), None);
-    }
-
-    #[test]
-    fn explicit_vocab_size_overrides_inferred_bound() {
-        let tokenizer = TestTokenizer::new()
-            .with_regular_token("<high>", 10_000)
-            .with_vocab_size(20_000);
-
-        assert_eq!(tokenizer.vocab_size(), 20_000);
-        assert_eq!(tokenizer.id_to_token(10_000).as_deref(), Some("<high>"));
-    }
-}
@@ -23,7 +23,7 @@ def test_python_error():
    error happening from the C++ side.
    """
    allocator = get_mem_allocator_instance()
-    total_bytes = current_platform.mem_get_info()[1]
+    total_bytes = torch.accelerator.get_memory_info()[1]
    alloc_bytes = int(total_bytes * 0.7)
    tensors = []
    with allocator.use_memory_pool():
@@ -64,9 +64,9 @@ def test_basic_cumem():
    output = x + y + z
    assert torch.allclose(output, torch.ones_like(output) * 3)

-    free_bytes = current_platform.mem_get_info()[0]
+    free_bytes = torch.accelerator.get_memory_info()[0]
    allocator.sleep()
-    free_bytes_after_sleep = current_platform.mem_get_info()[0]
+    free_bytes_after_sleep = torch.accelerator.get_memory_info()[0]
    assert free_bytes_after_sleep > free_bytes
    allocator.wake_up()

@@ -99,9 +99,9 @@ def test_cumem_with_cudagraph():
    with torch.cuda.graph(model_graph):
        y = model(x)

-    free_bytes = current_platform.mem_get_info()[0]
+    free_bytes = torch.accelerator.get_memory_info()[0]
    allocator.sleep()
-    free_bytes_after_sleep = current_platform.mem_get_info()[0]
+    free_bytes_after_sleep = torch.accelerator.get_memory_info()[0]
    assert free_bytes_after_sleep > free_bytes
    allocator.wake_up()

@@ -132,7 +132,7 @@ def test_cumem_with_cudagraph():
    ],
 )
 def test_end_to_end(model: str):
-    free, total = current_platform.mem_get_info()
+    free, total = torch.accelerator.get_memory_info()
    used_bytes_baseline = total - free  # in case other process is running
    llm = LLM(model, enable_sleep_mode=True)
    prompt = "How are you?"
@@ -144,7 +144,7 @@ def test_end_to_end(model: str):
    # test sleep level 1 here.
    llm.sleep(level=1)

-    free_gpu_bytes_after_sleep, total = current_platform.mem_get_info()
+    free_gpu_bytes_after_sleep, total = torch.accelerator.get_memory_info()
    used_bytes = total - free_gpu_bytes_after_sleep - used_bytes_baseline
    # now the memory usage is mostly cudagraph memory pool,
    # and it should be less than the model weights (1B model, 2GiB weights)
@@ -164,7 +164,7 @@ def test_end_to_end(model: str):
    llm.sleep(level=1)
    llm.wake_up(tags=["weights"])

-    free_gpu_bytes_wake_up_w, total = current_platform.mem_get_info()
+    free_gpu_bytes_wake_up_w, total = torch.accelerator.get_memory_info()
    used_bytes = total - free_gpu_bytes_wake_up_w - used_bytes_baseline

    # should just reallocate memory for weights (1B model, ~2GiB weights)
@@ -181,7 +181,7 @@ def test_end_to_end(model: str):
@create_new_process_for_each_test()
 def test_deep_sleep():
    model = "hmellor/tiny-random-LlamaForCausalLM"
-    free, total = current_platform.mem_get_info()
+    free, total = torch.accelerator.get_memory_info()
    used_bytes_baseline = total - free  # in case other process is running
    llm = LLM(model, enable_sleep_mode=True)
    prompt = "How are you?"
@@ -191,13 +191,13 @@ def test_deep_sleep():
    # Put the engine to deep sleep
    llm.sleep(level=2)

-    free_gpu_bytes_after_sleep, total = current_platform.mem_get_info()
+    free_gpu_bytes_after_sleep, total = torch.accelerator.get_memory_info()
    used_bytes = total - free_gpu_bytes_after_sleep - used_bytes_baseline
    assert used_bytes < 3 * GiB_bytes

    llm.wake_up(tags=["weights"])
    llm.collective_rpc("reload_weights")
-    free_gpu_bytes_wake_up_w, total = current_platform.mem_get_info()
+    free_gpu_bytes_wake_up_w, total = torch.accelerator.get_memory_info()
    used_bytes = total - free_gpu_bytes_wake_up_w - used_bytes_baseline
    assert used_bytes < 4 * GiB_bytes

@@ -213,7 +213,7 @@ def test_deep_sleep():
 def test_deep_sleep_async():
    async def test():
        model = "hmellor/tiny-random-LlamaForCausalLM"
-        free, total = current_platform.mem_get_info()
+        free, total = torch.accelerator.get_memory_info()
        used_bytes_baseline = total - free  # in case other process is running
        engine_args = AsyncEngineArgs(
            model=model,
@@ -232,7 +232,7 @@ def test_deep_sleep_async():

        await llm.wake_up(tags=["weights"])
        await llm.collective_rpc("reload_weights")
-        free_gpu_bytes_wake_up_w, total = current_platform.mem_get_info()
+        free_gpu_bytes_wake_up_w, total = torch.accelerator.get_memory_info()
        used_bytes = total - free_gpu_bytes_wake_up_w - used_bytes_baseline
        assert used_bytes < 4 * GiB_bytes

@@ -29,9 +29,43 @@ from vllm.distributed.weight_transfer.nccl_engine import (
    NCCLWeightTransferInitInfo,
    NCCLWeightTransferUpdateInfo,
 )
+from vllm.platforms import current_platform
 from vllm.utils.network_utils import get_open_port


+def _weight_transfer_ray_env_vars() -> dict[str, str]:
+    if not current_platform.is_rocm():
+        return {}
+
+    return {
+        "RAY_EXPERIMENTAL_NOSET_CUDA_VISIBLE_DEVICES": "1",
+        "RAY_EXPERIMENTAL_NOSET_HIP_VISIBLE_DEVICES": "1",
+        "RAY_EXPERIMENTAL_NOSET_ROCR_VISIBLE_DEVICES": "1",
+    }
+
+
+def _init_ray_for_weight_transfer() -> None:
+    if ray.is_initialized():
+        return
+    ray.init(
+        ignore_reinit_error=True,
+        runtime_env={"env_vars": _weight_transfer_ray_env_vars()},
+    )
+
+
+def _get_ray_assigned_device() -> torch.device:
+    gpu_ids = ray.get_gpu_ids()
+    if not gpu_ids:
+        return torch.device("cuda:0")
+    return torch.device(f"cuda:{int(gpu_ids[0])}")
+
+
+def _set_ray_assigned_device() -> torch.device:
+    device = _get_ray_assigned_device()
+    torch.accelerator.set_device(device)
+    return device
+
+
 def create_mock_parallel_config(
    rank: int = 0,
    world_size: int = 1,
@@ -321,6 +355,8 @@ def trainer_broadcast_tensor(
    """Trainer task that broadcasts a tensor via NCCL."""
    import torch

+    device = _set_ray_assigned_device()
+
    from vllm.distributed.device_communicators.pynccl import PyNcclCommunicator
    from vllm.distributed.utils import StatelessProcessGroup

@@ -331,12 +367,11 @@ def trainer_broadcast_tensor(
        rank=0,
        world_size=world_size,
    )
-    # Ray sets CUDA_VISIBLE_DEVICES, so device 0 is the assigned GPU
-    comm = PyNcclCommunicator(pg, device=0)
+    comm = PyNcclCommunicator(pg, device=device.index)

    # Create and broadcast the tensor
    dtype = getattr(torch, tensor_dtype)
-    tensor_to_send = torch.ones(tensor_shape, dtype=dtype, device="cuda:0")
+    tensor_to_send = torch.ones(tensor_shape, dtype=dtype, device=device)
    comm.broadcast(tensor_to_send, src=0, stream=torch.cuda.current_stream())
    torch.accelerator.synchronize()

@@ -356,6 +391,8 @@ def inference_receive_tensor(

    import torch

+    _set_ray_assigned_device()
+
    from vllm.config.parallel import ParallelConfig
    from vllm.config.weight_transfer import WeightTransferConfig
    from vllm.distributed.weight_transfer.nccl_engine import (
@@ -435,7 +472,7 @@ def test_nccl_weight_transfer_between_processes():
    This test verifies that the NCCLWeightTransferEngine can receive
    tensors broadcast by a trainer process via NCCL.
    """
-    ray.init(ignore_reinit_error=True)
+    _init_ray_for_weight_transfer()

    master_address = "127.0.0.1"
    master_port = get_open_port()
@@ -473,6 +510,8 @@ def trainer_broadcast_sparse_tensor(
    """Trainer task that broadcasts sparse patches via NCCL."""
    import torch

+    device = _set_ray_assigned_device()
+
    from vllm.distributed.device_communicators.pynccl import PyNcclCommunicator
    from vllm.distributed.utils import StatelessProcessGroup
    from vllm.distributed.weight_transfer.base import SparseWeightPatch
@@ -487,12 +526,12 @@ def trainer_broadcast_sparse_tensor(
        rank=0,
        world_size=world_size,
    )
-    comm = PyNcclCommunicator(pg, device=0)
+    comm = PyNcclCommunicator(pg, device=device.index)

    patch = SparseWeightPatch(
        name="test.weight",
-        indices=torch.tensor([1, 7, 25], dtype=torch.int32, device="cuda:0"),
-        values=torch.tensor([10.0, 20.0, 30.0], dtype=torch.float32, device="cuda:0"),
+        indices=torch.tensor([1, 7, 25], dtype=torch.int32, device=device),
+        values=torch.tensor([10.0, 20.0, 30.0], dtype=torch.float32, device=device),
    )
    NCCLWeightTransferEngine.trainer_send_sparse_weights(
        iter([patch]),
@@ -513,6 +552,8 @@ def inference_receive_sparse_tensor(

    import torch

+    device = _set_ray_assigned_device()
+
    from vllm.config.parallel import ParallelConfig
    from vllm.config.weight_transfer import WeightTransferConfig
    from vllm.distributed.weight_transfer.nccl_engine import (
@@ -540,7 +581,7 @@ def inference_receive_sparse_tensor(
        )
    )

-    target = torch.zeros(30, dtype=torch.float32, device="cuda")
+    target = torch.zeros(30, dtype=torch.float32, device=device)

    def apply_sparse_patches(patches: list[SparseWeightPatch]):
        for patch in patches:
@@ -556,9 +597,9 @@ def inference_receive_sparse_tensor(
    engine.receive_sparse_weights(update_info, apply_sparse_patches)
    torch.accelerator.synchronize()

-    expected = torch.zeros(30, dtype=torch.float32, device="cuda")
+    expected = torch.zeros(30, dtype=torch.float32, device=device)
    expected[[1, 7, 25]] = torch.tensor(
-        [10.0, 20.0, 30.0], dtype=torch.float32, device="cuda"
+        [10.0, 20.0, 30.0], dtype=torch.float32, device=device
    )
    success = torch.equal(target, expected)
    engine.shutdown()
@@ -574,7 +615,7 @@ def inference_receive_sparse_tensor(
 )
 def test_nccl_sparse_weight_transfer_between_processes():
    """Test NCCL sparse weight transfer from trainer to inference process."""
-    ray.init(ignore_reinit_error=True)
+    _init_ray_for_weight_transfer()

    master_address = "127.0.0.1"
    master_port = get_open_port()
@@ -933,16 +974,18 @@ class TrainerActor:
    """Trainer actor that creates and holds CUDA IPC handles."""

    def __init__(self, tensor_shape: list[int], tensor_dtype: str):
+        device = _set_ray_assigned_device()
+
        # Create tensor on GPU and keep it alive
        dtype = getattr(torch, tensor_dtype)
-        self.tensor = torch.ones(tensor_shape, dtype=dtype, device="cuda:0")
+        self.tensor = torch.ones(tensor_shape, dtype=dtype, device=device)
        self.tensor.fill_(42.0)  # Fill with 42 to verify correct transfer

        # Create IPC handle (tensor must stay alive for IPC to work)
        # reduce_tensor returns (rebuild_func, args); we only send args
        # since the receiver imports rebuild_cuda_tensor directly.
        _, ipc_args = reduce_tensor(self.tensor)
-        gpu_uuid = get_physical_gpu_id(0)
+        gpu_uuid = get_physical_gpu_id(device.index)

        torch.accelerator.synchronize()

@@ -974,6 +1017,8 @@ def inference_receive_ipc_tensor(

    import torch

+    _set_ray_assigned_device()
+
    from vllm.config.parallel import ParallelConfig
    from vllm.config.weight_transfer import WeightTransferConfig
    from vllm.distributed.weight_transfer.ipc_engine import (
@@ -1072,7 +1117,7 @@ def test_ipc_weight_transfer_between_processes(mode: str):
    from ray.util.placement_group import placement_group
    from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy

-    ray.init(ignore_reinit_error=True)
+    _init_ray_for_weight_transfer()

    # Create a placement group to ensure both processes are on the same GPU
    # Use fractional GPUs so both tasks can share the same GPU bundle
@@ -142,6 +142,7 @@ class TestHarmonyToResponseOutput:
        )
        assert output_items[0].call_id.startswith("call_")
        assert output_items[0].id.startswith("fc_")
+        assert output_items[0].status == "completed"

    def test_commentary_with_python_recipient_creates_reasoning(self):
        """Test that commentary with recipient='python' creates reasoning items."""
@@ -347,7 +347,7 @@ def test_selective_state_update(dim, dstate, has_z, itype):
    rtol, atol = (3e-4, 1e-3) if itype == torch.float32 else (5e-3, 1e-2)
    if itype == torch.bfloat16:
        rtol, atol = 1e-2, 5e-2
-        if torch.version.hip:
+        if current_platform.is_rocm() or current_platform.is_xpu():
            atol *= 2
    # set seed
    set_random_seed(0)
@@ -437,7 +437,7 @@ def test_selective_state_update_varlen(dim, dstate, has_z, itype, max_seq_len):
    rtol, atol = (3e-4, 1e-3) if itype == torch.float32 else (5e-3, 1e-2)
    if itype == torch.bfloat16:
        rtol, atol = 5e-2, 1.5e-1
-        if torch.version.hip:
+        if current_platform.is_rocm() or current_platform.is_xpu():
            atol *= 2
    # set seed
    set_random_seed(0)
@@ -700,7 +700,7 @@ def test_selective_state_update_with_batch_indices(
    rtol, atol = (3e-4, 1e-3) if itype == torch.float32 else (5e-3, 1e-2)
    if itype == torch.bfloat16:
        rtol, atol = 1e-1, 1e-1
-        if torch.version.hip:
+        if current_platform.is_rocm() or current_platform.is_xpu():
            atol *= 2
    # set seed
    torch.random.manual_seed(0)
@@ -865,7 +865,7 @@ def test_selective_state_update_with_num_accepted_tokens(
    rtol, atol = (3e-4, 1e-3) if itype == torch.float32 else (5e-3, 1e-2)
    if itype == torch.bfloat16:
        rtol, atol = 5e-2, 1.5e-1
-        if torch.version.hip:
+        if current_platform.is_rocm() or current_platform.is_xpu():
            atol *= 2

    set_random_seed(0)
@@ -991,7 +991,7 @@ def test_selective_state_update_varlen_with_num_accepted(
    rtol, atol = (3e-4, 1e-3) if itype == torch.float32 else (5e-3, 1e-2)
    if itype == torch.bfloat16:
        rtol, atol = 5e-2, 1.5e-1
-        if torch.version.hip:
+        if current_platform.is_rocm() or current_platform.is_xpu():
            atol *= 2

    set_random_seed(0)
@@ -406,7 +406,7 @@ def test_fused_moe_int64_overflow(workspace_init):
    Reproduces the scenario from PR #34279.
    """
    # ~12 GB GPU memory needed for intermediate caches
-    free_mem = torch.cuda.mem_get_info()[0]
+    free_mem = torch.accelerator.get_memory_info()[0]
    if free_mem < 12 * 1024**3:
        pytest.skip("Insufficient GPU memory for overflow test")

@@ -97,7 +97,7 @@ def test_sharded_state_loader(
    ctx = mp.get_context("spawn")

    platform_args = {}
-    if current_platform.is_rocm():
+    if current_platform.is_rocm() or current_platform.is_xpu():
        platform_args["max_num_seqs"] = 1

    # Run in separate processes for memory & CUDA isolation
@@ -83,7 +83,7 @@ def _ru_maxrss_bytes() -> int | None:

 def _gpu_used_bytes() -> int:
    torch.accelerator.synchronize()
-    free_bytes, total_bytes = current_platform.mem_get_info()
+    free_bytes, total_bytes = torch.accelerator.get_memory_info()
    return int(total_bytes - free_bytes)


@@ -205,3 +205,57 @@ def test_image_media_io_load_file(tmp_path):

    with pytest.raises(ValueError, match="Failed to load image"):
        image_io.load_file(truncated_real_file)
+
+
+def test_image_pixel_limit_respected():
+    """A small image within the pixel limit loads successfully."""
+    import vllm.envs as envs
+
+    image = Image.new("RGB", (100, 100), (255, 0, 0))
+    from io import BytesIO
+
+    buf = BytesIO()
+    image.save(buf, format="PNG")
+    data = buf.getvalue()
+
+    assert envs.VLLM_MAX_IMAGE_PIXELS >= 100 * 100
+
+    image_io = ImageMediaIO()
+    result = image_io.load_bytes(data)
+    assert result.media.size == (100, 100)
+
+
+def test_image_pixel_limit_rejected(monkeypatch):
+    """An image exceeding the pixel limit is rejected before raster decode."""
+    import vllm.envs as envs
+
+    monkeypatch.setattr(envs, "VLLM_MAX_IMAGE_PIXELS", 100)
+
+    image = Image.new("RGB", (20, 20), (0, 255, 0))
+    from io import BytesIO
+
+    buf = BytesIO()
+    image.save(buf, format="PNG")
+    data = buf.getvalue()
+
+    image_io = ImageMediaIO()
+    with pytest.raises(ValueError, match="exceed"):
+        image_io.load_bytes(data)
+
+
+def test_image_pixel_limit_disabled(monkeypatch):
+    """Setting VLLM_MAX_IMAGE_PIXELS=0 disables the pixel limit."""
+    import vllm.envs as envs
+
+    monkeypatch.setattr(envs, "VLLM_MAX_IMAGE_PIXELS", 0)
+
+    image = Image.new("RGB", (1000, 1000), (0, 0, 255))
+    from io import BytesIO
+
+    buf = BytesIO()
+    image.save(buf, format="PNG")
+    data = buf.getvalue()
+
+    image_io = ImageMediaIO()
+    result = image_io.load_bytes(data)
+    assert result.media.size == (1000, 1000)
@@ -31,6 +31,7 @@ from vllm.entrypoints.openai.chat_completion.protocol import (
 from vllm.parser.engine.registered_adapters import (
    Gemma4Parser,
    Glm47MoeParser,
+    KimiK2Parser,
    MinimaxM2Parser,
    NemotronV3Parser,
    Qwen3Parser,
@@ -717,6 +718,96 @@ def _build_glm47_moe(scenario: Scenario, validate: bool = True) -> Sample:
    return sample


+# ── Kimi K2 (native tool-call section, starts in REASONING) ──────────
+
+_KIMI_K2_VOCAB: dict[str, int] = {
+    "<think>": 50,
+    "</think>": 51,
+    "<|tool_calls_section_begin|>": 60,
+    "<|tool_calls_section_end|>": 61,
+    "<|tool_call_begin|>": 62,
+    "<|tool_call_end|>": 63,
+    "<|tool_call_argument_begin|>": 64,
+}
+
+
+def _kimi_k2_tool_segments(
+    tool_calls: list[ToolCallSpec],
+) -> list[tuple[str, bool]]:
+    segs: list[tuple[str, bool]] = [("<|tool_calls_section_begin|>", True)]
+    for index, tc in enumerate(tool_calls):
+        args = json.dumps(tc.arguments, ensure_ascii=False, separators=(",", ":"))
+        segs.extend(
+            [
+                ("<|tool_call_begin|>", True),
+                (f"functions.{tc.name}:{index}\n", False),
+                ("<|tool_call_argument_begin|>", True),
+                (args, False),
+                ("<|tool_call_end|>", True),
+            ]
+        )
+    segs.append(("<|tool_calls_section_end|>", True))
+    return segs
+
+
+def _kimi_k2_segments(scenario: Scenario) -> list[tuple[str, bool]]:
+    segs: list[tuple[str, bool]] = []
+    if scenario.reasoning is not None:
+        segs.append(("<think>", True))
+        segs.append((scenario.reasoning, False))
+    if scenario.content is not None or scenario.tool_calls is not None:
+        segs.append(("</think>", True))
+    if scenario.content is not None:
+        segs.append((scenario.content, False))
+    if scenario.tool_calls is not None:
+        segs.extend(_kimi_k2_tool_segments(scenario.tool_calls))
+    return segs
+
+
+def _build_kimi_k2(
+    scenario: Scenario,
+    validate: bool = True,
+    thinking: bool = True,
+) -> Sample:
+    expected_reasoning = (
+        scenario.reasoning.rstrip()
+        if (thinking and scenario.reasoning is not None)
+        else None
+    )
+    if thinking and scenario.reasoning is None:
+        expected_reasoning = ""
+
+    sample = _make_sample(
+        sample_id=f"kimi_k2-{scenario.id}",
+        description=scenario.description,
+        vocab=_KIMI_K2_VOCAB,
+        segments=_kimi_k2_segments(scenario),
+        expected_reasoning=expected_reasoning,
+        expected_content=_qwen3_expected_content(scenario),
+        expected_tool_calls=_expected_tc(scenario),
+        tools=_expected_tools(scenario),
+        chat_template_kwargs=None if thinking else {"thinking": False},
+    )
+    if validate:
+        _validate_sample(
+            sample,
+            KimiK2Parser,
+            chat_template_kwargs=sample.chat_template_kwargs,
+        )
+    return sample
+
+
+_KIMI_K2_SCENARIOS = [
+    *SCENARIOS,
+    Scenario(
+        id="trailing-reasoning-whitespace",
+        description="Reasoning trailing whitespace is stripped",
+        reasoning="Reasoning with trailing whitespace. \n\t",
+        content="Done.",
+    ),
+]
+
+
 # ── Registry and public API ──────────────────────────────────────────

 _BUILDERS: dict[str, Any] = {
@@ -726,6 +817,7 @@ _BUILDERS: dict[str, Any] = {
    "nemotron_v3": _build_nemotron_v3,
    "seed_oss": _build_seed_oss,
    "glm47_moe": _build_glm47_moe,
+    "kimi_k2": _build_kimi_k2,
 }


@@ -733,7 +825,8 @@ _BUILDERS: dict[str, Any] = {
 def build_samples(model: str) -> tuple[Sample, ...]:
    """Build all scenario samples for a model, self-validated."""
    builder = _BUILDERS[model]
-    return tuple(builder(s) for s in SCENARIOS)
+    scenarios = _KIMI_K2_SCENARIOS if model == "kimi_k2" else SCENARIOS
+    return tuple(builder(s) for s in scenarios)


 def build_sample(model: str, scenario: Scenario) -> Sample:
@@ -7,7 +7,6 @@ import pytest

 from vllm.entrypoints.openai.chat_completion.protocol import ChatCompletionRequest
 from vllm.entrypoints.openai.engine.protocol import DeltaMessage
-from vllm.reasoning.identity_reasoning_parser import IdentityReasoningParser
 from vllm.reasoning.kimi_k2_reasoning_parser import KimiK2ReasoningParser
 from vllm.tokenizers import get_tokenizer

@@ -33,20 +32,6 @@ def kimi_k2_tokenizer():
    return get_tokenizer(tokenizer_name=REASONING_MODEL_NAME, trust_remote_code=True)


-def test_parser_selection_thinking_enabled(kimi_k2_tokenizer):
-    parser = KimiK2ReasoningParser(
-        kimi_k2_tokenizer, chat_template_kwargs={"thinking": True}
-    )
-    assert parser._identity_parser is None
-
-
-def test_parser_selection_thinking_disabled(kimi_k2_tokenizer):
-    parser = KimiK2ReasoningParser(
-        kimi_k2_tokenizer, chat_template_kwargs={"thinking": False}
-    )
-    assert isinstance(parser._identity_parser, IdentityReasoningParser)
-
-
 def test_extract_reasoning_with_think_tags(kimi_k2_tokenizer):
    parser = KimiK2ReasoningParser(kimi_k2_tokenizer)
    request = ChatCompletionRequest(model="test-model", messages=[], temperature=1.0)
@@ -65,7 +50,7 @@ def test_extract_reasoning_empty_thinking(kimi_k2_tokenizer):
    reasoning, content = parser.extract_reasoning(
        "<think></think>final answer", request
    )
-    assert reasoning == ""
+    assert reasoning is None
    assert content == "final answer"


@@ -96,8 +81,8 @@ def test_streaming_reasoning_then_content(kimi_k2_tokenizer):
    """Token-by-token streaming: reasoning tokens then content after </think>."""
    parser = KimiK2ReasoningParser(kimi_k2_tokenizer)

-    think_id = parser._start_token_id
-    end_think_id = parser._end_token_id
+    think_id = parser._parser_engine._start_token_id
+    end_think_id = parser._parser_engine._end_token_id
    # Use a real token ID from the tokenizer for regular content
    regular_id = kimi_k2_tokenizer.encode("hello", add_special_tokens=False)[0]

@@ -154,8 +139,8 @@ def test_streaming_tool_section_ends_reasoning(kimi_k2_tokenizer):
    """<|tool_calls_section_begin|> in delta ends reasoning during streaming."""
    parser = KimiK2ReasoningParser(kimi_k2_tokenizer)

-    think_id = parser._start_token_id
-    tool_begin_id = parser._tool_section_start_token_id
+    think_id = parser._parser_engine._start_token_id
+    tool_begin_id = parser._parser_engine._tool_section_start_token_id
    regular_id = kimi_k2_tokenizer.encode("hello", add_special_tokens=False)[0]

    # Tool section token arrives — should transition from reasoning to content
@@ -169,50 +154,3 @@ def test_streaming_tool_section_ends_reasoning(kimi_k2_tokenizer):
    )
    assert isinstance(result, DeltaMessage)
    assert result.content == "<|tool_calls_section_begin|>"
-
-
-def test_streaming_end_token_id_buffered(mock_kimi_k2_tokenizer):
-    """When stop sequences buffer text, </think> ID arrives before its text.
-
-    The token ID is present in delta_token_ids but the actual string is not
-    yet in delta_text (still buffered). The parser must return None to wait
-    for the next delta, instead of calling find() which returns -1 and
-    silently corrupting the text split.
-    """
-    parser = KimiK2ReasoningParser(mock_kimi_k2_tokenizer)
-    think_id = parser._start_token_id
-    end_think_id = parser._end_token_id
-
-    # Simulate: </think> ID arrived but text not yet flushed.
-    # Two token IDs in delta to bypass the single-special-token guard.
-    result = parser.extract_reasoning_streaming(
-        previous_text="some reasoning",
-        current_text="some reasoning extra",
-        delta_text="extra",  # </think> text not yet flushed
-        previous_token_ids=[think_id],
-        current_token_ids=[think_id, end_think_id, 999],
-        delta_token_ids=[end_think_id, 999],
-    )
-    assert result is None
-
-
-def test_streaming_tool_section_id_buffered(mock_kimi_k2_tokenizer):
-    """When stop sequences buffer text, tool section start ID arrives before its text.
-
-    Same buffering scenario as above but for <|tool_calls_section_begin|>.
-    Without the guard, find() returns -1 and delta_text[:tool_index] silently
-    drops the last character of reasoning.
-    """
-    parser = KimiK2ReasoningParser(mock_kimi_k2_tokenizer)
-    think_id = parser._start_token_id
-    tool_begin_id = parser._tool_section_start_token_id
-
-    result = parser.extract_reasoning_streaming(
-        previous_text="some reasoning",
-        current_text="some reasoning extra",
-        delta_text="extra",  # tool section text not yet flushed
-        previous_token_ids=[think_id],
-        current_token_ids=[think_id, tool_begin_id, 999],
-        delta_token_ids=[tool_begin_id, 999],
-    )
-    assert result is None
@@ -604,6 +604,13 @@ class RemoteVLLMServer:
                        mem_info = nvmlDeviceGetMemoryInfo(handle)
                        total_used += mem_info.used
                    return total_used
+            elif current_platform.is_xpu():
+                total_used = 0
+                device_count = current_platform.device_count()
+                for i in range(device_count):
+                    free, total = torch.xpu.mem_get_info(i)
+                    total_used += total - free
+                return total_used
        except Exception as e:
            print(f"[RemoteOpenAIServer] Could not query GPU memory: {e}")
            return None
@@ -36,7 +36,7 @@ def test_memory_profiling():
    weights_memory = 128 * 1024 * 1024 * 4  # 512 MiB

    def measure_current_non_torch():
-        free, total = torch.cuda.mem_get_info()
+        free, total = torch.accelerator.get_memory_info()
        current_used = total - free
        current_torch = torch.accelerator.memory_reserved()
        current_non_torch = current_used - current_torch
@@ -81,8 +81,9 @@ def test_memory_snapshot_uses_psutil_on_integrated_gpu():
    with (
        patch("vllm.utils.mem_utils.current_platform") as mock_platform,
        patch("vllm.utils.mem_utils.psutil") as mock_psutil,
+        patch("torch.accelerator") as mock_accelerator,
    ):
-        mock_platform.mem_get_info.return_value = (
+        mock_accelerator.get_memory_info.return_value = (
            mock_cuda_free,
            mock_cuda_total,
        )
@@ -90,8 +91,8 @@ def test_memory_snapshot_uses_psutil_on_integrated_gpu():
        mock_platform.memory_stats.return_value = {
            "allocated_bytes.all.peak": 0,
        }
-        mock_platform.memory_reserved.return_value = 0
-        mock_platform.current_device = lambda: "cuda:0"
+        mock_accelerator.memory_reserved.return_value = 0
+        mock_accelerator.current_device = lambda: "cuda:0"

        mock_vmem = MagicMock()
        mock_vmem.available = mock_psutil_available
@@ -105,24 +106,25 @@ def test_memory_snapshot_uses_psutil_on_integrated_gpu():


 def test_memory_snapshot_uses_cuda_on_discrete_gpu():
-    """On discrete GPUs, free_memory should come from CUDA mem_get_info."""
+    """On discrete GPUs, free_memory should come from accelerator  get_memory_info."""
    mock_cuda_free = 70 * 1024**3
    mock_cuda_total = 80 * 1024**3

    with (
        patch("vllm.utils.mem_utils.current_platform") as mock_platform,
        patch("vllm.utils.mem_utils.psutil") as mock_psutil,
+        patch("torch.accelerator") as mock_accelerator,
    ):
-        mock_platform.mem_get_info.return_value = (
+        mock_accelerator.get_memory_info.return_value = (
            mock_cuda_free,
            mock_cuda_total,
        )
        mock_platform.is_integrated_gpu.return_value = False
-        mock_platform.memory_stats.return_value = {
+        mock_accelerator.memory_stats.return_value = {
            "allocated_bytes.all.peak": 0,
        }
-        mock_platform.memory_reserved.return_value = 0
-        mock_platform.current_device = lambda: "cuda:0"
+        mock_accelerator.memory_reserved.return_value = 0
+        mock_accelerator.current_device = lambda: "cuda:0"

        snapshot = MemorySnapshot(device="cuda:0")

@@ -33,14 +33,37 @@ from vllm.distributed.kv_transfer.kv_connector.v1.mooncake.mooncake_utils import
 )
 from vllm.utils.network_utils import get_open_port
 from vllm.v1.attention.backends.flash_attn import FlashAttentionBackend
-from vllm.v1.kv_cache_interface import KVCacheConfig
+from vllm.v1.kv_cache_interface import (
+    FullAttentionSpec,
+    KVCacheConfig,
+    KVCacheGroupSpec,
+)
 from vllm.v1.request import RequestStatus

 from .utils import create_request, create_scheduler, create_vllm_config


 def _make_test_kv_cache_config() -> KVCacheConfig:
-    return KVCacheConfig(num_blocks=0, kv_cache_tensors=[], kv_cache_groups=[])
+    return KVCacheConfig(
+        num_blocks=0,
+        kv_cache_tensors=[],
+        kv_cache_groups=[
+            KVCacheGroupSpec(
+                [
+                    "model.layers.0.self_attn",
+                    "model.layers.1.self_attn",
+                    "model.layers.0.mla_attn",
+                    "model.layers.1.eagle_attn",
+                ],
+                FullAttentionSpec(
+                    block_size=16,
+                    num_kv_heads=4,
+                    head_size=64,
+                    dtype=torch.float16,
+                ),
+            )
+        ],
+    )


 class FakeMooncakeWrapper:
@@ -126,6 +149,8 @@ async def test_build_transfer_params_separates_prefill_pp_layers():
    worker.is_kv_producer = True
    worker.tp_rank = 0
    worker.tp_size = 1
+    worker.kv_cache_config = _make_test_kv_cache_config()
+    worker._physical_blocks_per_logical_kv_block = 1
    worker.transfer_topo = SimpleNamespace(local_replicates_kv_cache=False)

    block_len = 256
@@ -206,6 +231,7 @@ async def test_build_transfer_params_separates_prefill_pp_layers():
        req_blocks={"d-req-pp": (transfer_id, [[20, 21]])},
        kv_caches_base_addr=[region.base_addr for region in remote_regions],
        block_lens=[region.block_len for region in remote_regions],
+        kv_block_lens=[region.kv_block_len for region in remote_regions],
        registered_layer_names=[region.layer_name for region in remote_regions],
        registered_layer_indices=[region.layer_index for region in remote_regions],
    )
@@ -266,6 +292,7 @@ async def test_send_kv_to_decode_aligns_consumer_regions_by_layer_metadata(
        kv_half = block_len // 2
        prefill_worker.kv_caches_base_addr = [0x1000]
        prefill_worker.block_len_per_layer = [block_len]
+        prefill_worker.kv_block_len_per_layer = [kv_half]
        prefill_worker.registered_layer_names = ["model.layers.1.self_attn"]
        prefill_worker.registered_layer_indices = [1]

@@ -294,6 +321,7 @@ async def test_send_kv_to_decode_aligns_consumer_regions_by_layer_metadata(
            req_blocks={"d-req-layer-align": (transfer_id, [[20]])},
            kv_caches_base_addr=[0xA000, 0xB000],
            block_lens=[block_len, block_len],
+            kv_block_lens=[kv_half, kv_half],
            registered_layer_names=[
                "model.layers.0.self_attn",
                "model.layers.1.self_attn",
@@ -804,7 +832,9 @@ async def test_kv_producer(monkeypatch):
        prefill_worker = prefill_connector.connector_worker
        prefill_worker.kv_caches_base_addr = [0x1000]
        block_len = 4096
+        kv_half = block_len // 2
        prefill_worker.block_len_per_layer = [block_len]
+        prefill_worker.kv_block_len_per_layer = [kv_half]
        prefill_worker.registered_layer_names = ["model.layers.0.self_attn"]
        prefill_worker.registered_layer_indices = [0]

@@ -832,6 +862,7 @@ async def test_kv_producer(monkeypatch):
            req_blocks={"d-req-1": (transfer_id, [[20, 21]])},
            kv_caches_base_addr=[0x2000],
            block_lens=[block_len],
+            kv_block_lens=[kv_half],
            registered_layer_names=["model.layers.0.self_attn"],
            registered_layer_indices=[0],
        )
@@ -845,8 +876,6 @@ async def test_kv_producer(monkeypatch):
        ) as mock_send_blocks:
            # With blocks-first layout, each block is virtually split
            # into K and V halves, producing non-coalesced transfers.
-            kv_half = block_len // 2
-
            def expected_split_transfers(src_base, dst_base, src_blocks, dst_blocks):
                """Build expected (src_ptrs, dst_ptrs, lengths) for
                virtual-split K/V transfers."""
@@ -981,6 +1010,7 @@ async def test_kv_consumuer(monkeypatch):
        decode_worker = decode_connector.connector_worker
        decode_worker.kv_caches_base_addr = [0x1000]
        decode_worker.block_len_per_layer = [4096]
+        decode_worker.kv_block_len_per_layer = [4096]
        decode_worker.registered_layer_names = ["model.layers.0.self_attn"]
        decode_worker.registered_layer_indices = [0]
        decode_worker.rpc_port = 54321
@@ -1236,6 +1266,7 @@ async def test_kv_producer_heterogeneous_tp(monkeypatch, d_tp_size):

        prefill_worker.kv_caches_base_addr = [0x1000]
        prefill_worker.block_len_per_layer = [local_block_len]
+        prefill_worker.kv_block_len_per_layer = [local_block_len // 2]
        prefill_worker.registered_layer_names = ["model.layers.0.self_attn"]
        prefill_worker.registered_layer_indices = [0]

@@ -1283,6 +1314,7 @@ async def test_kv_producer_heterogeneous_tp(monkeypatch, d_tp_size):
                    },
                    kv_caches_base_addr=[0x2000],
                    block_lens=[remote_block_len],
+                    kv_block_lens=[remote_block_len // 2],
                    registered_layer_names=["model.layers.0.self_attn"],
                    registered_layer_indices=[0],
                )
@@ -257,6 +257,7 @@ async def test_build_transfer_params_multi_group_trimming(monkeypatch):
            },
            kv_caches_base_addr=[0x2000],
            block_lens=[block_len],
+            kv_block_lens=[block_len],
        )

        local_regions = [
@@ -348,6 +349,7 @@ async def test_build_transfer_params_group_count_mismatch(monkeypatch):
            },
            kv_caches_base_addr=[0x2000],
            block_lens=[block_len],
+            kv_block_lens=[block_len],
        )

        local_regions = [
@@ -0,0 +1,386 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Unit tests for MooncakeConnector hybrid FA + GDN support.
+
+GDN is represented as a MambaSpec in vLLM, so these tests exercise the
+Mooncake MambaSpec path with mamba_type=GDN_ATTN. Mamba2 is intentionally not
+validated by this test module.
+"""
+
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+import torch
+
+from vllm.config import set_current_vllm_config
+from vllm.distributed.kv_transfer.kv_connector.v1.mooncake.mooncake_connector import (
+    KVConnectorRole,
+    MooncakeConnector,
+    MooncakeConnectorScheduler,
+    MooncakeConnectorWorker,
+    MooncakeXferMetadata,
+    SendBlockMeta,
+    TransferRegion,
+)
+from vllm.v1.attention.backends.registry import MambaAttentionBackendEnum
+from vllm.v1.attention.backends.utils import NULL_BLOCK_ID
+from vllm.v1.kv_cache_interface import (
+    FullAttentionSpec,
+    KVCacheConfig,
+    KVCacheGroupSpec,
+    MambaSpec,
+)
+
+from .test_mooncake_connector import patch_worker_dependencies
+from .utils import create_request, create_vllm_config
+
+
+def noop_shutdown():
+    pass
+
+
+def make_hybrid_gdn_kv_cache_config(block_size: int) -> KVCacheConfig:
+    gdn_spec = MambaSpec(
+        block_size=block_size,
+        shapes=((6, 3), (1, 2, 2)),
+        dtypes=(torch.float16, torch.float16),
+        mamba_type=MambaAttentionBackendEnum.GDN_ATTN,
+    )
+    assert gdn_spec.mamba_type == MambaAttentionBackendEnum.GDN_ATTN
+    return KVCacheConfig(
+        num_blocks=16,
+        kv_cache_tensors=[],
+        kv_cache_groups=[
+            KVCacheGroupSpec(
+                ["model.layers.0.self_attn"],
+                FullAttentionSpec(
+                    block_size=block_size,
+                    num_kv_heads=1,
+                    head_size=1,
+                    dtype=torch.float16,
+                ),
+            ),
+            KVCacheGroupSpec(
+                ["model.layers.1.linear_attn"],
+                gdn_spec,
+            ),
+        ],
+    )
+
+
+def make_hybrid_gdn_scheduler(kv_role: str) -> MooncakeConnectorScheduler:
+    vllm_config = create_vllm_config(
+        kv_connector="MooncakeConnector",
+        kv_role=kv_role,
+    )
+    vllm_config.scheduler_config.disable_hybrid_kv_cache_manager = False
+    return MooncakeConnectorScheduler(
+        vllm_config=vllm_config,
+        engine_id="test-engine",
+        kv_cache_config=make_hybrid_gdn_kv_cache_config(
+            vllm_config.cache_config.block_size
+        ),
+    )
+
+
+@pytest.mark.cpu_test
+def test_hybrid_gdn_remote_prefill_uses_mamba_n_minus_one():
+    scheduler = make_hybrid_gdn_scheduler(kv_role="kv_consumer")
+    request = create_request(num_tokens=10, do_remote_prefill=True)
+
+    num_new_tokens, is_async = scheduler.get_num_new_matched_tokens(
+        request, num_computed_tokens=0
+    )
+
+    assert num_new_tokens == request.num_prompt_tokens - 1
+    assert is_async is True
+
+
+@pytest.mark.cpu_test
+def test_hybrid_gdn_remote_decode_truncates_prefill_once():
+    scheduler = make_hybrid_gdn_scheduler(kv_role="kv_producer")
+    request = create_request(num_tokens=10, do_remote_decode=True)
+    original_tokens = list(request.prompt_token_ids)
+
+    num_new_tokens, is_async = scheduler.get_num_new_matched_tokens(
+        request, num_computed_tokens=0
+    )
+
+    assert num_new_tokens == 0
+    assert is_async is False
+    assert request.prompt_token_ids == original_tokens[:-1]
+    assert request._all_token_ids == original_tokens[:-1]
+    assert request.num_prompt_tokens == len(original_tokens) - 1
+    assert request.max_tokens == 1
+    assert request.kv_transfer_params["_p_side_truncated"] is True
+
+    scheduler.get_num_new_matched_tokens(request, num_computed_tokens=0)
+    assert request.prompt_token_ids == original_tokens[:-1]
+
+
+def test_register_kv_caches_emits_fa_and_gdn_regions(monkeypatch):
+    monkeypatch.setenv("VLLM_MOONCAKE_ABORT_REQUEST_TIMEOUT", "5")
+    vllm_config = create_vllm_config(
+        kv_connector="MooncakeConnector",
+        kv_role="kv_consumer",
+    )
+    kv_cache_config = make_hybrid_gdn_kv_cache_config(
+        vllm_config.cache_config.block_size
+    )
+
+    with set_current_vllm_config(vllm_config), patch_worker_dependencies():
+        connector = MooncakeConnector(
+            vllm_config,
+            KVConnectorRole.WORKER,
+            kv_cache_config,
+        )
+        worker = connector.connector_worker
+
+        fa_cache = torch.empty((2, 2, 11), dtype=torch.float16)
+        gdn_conv_state = torch.empty((2, 22), dtype=torch.float16)
+        gdn_ssm_state = torch.empty((2, 4), dtype=torch.float16)
+
+        worker.register_kv_caches(
+            {
+                "model.layers.0.self_attn": fa_cache,
+                "model.layers.1.linear_attn": (gdn_conv_state, gdn_ssm_state),
+            }
+        )
+
+        assert worker.transfer_topo.is_mamba is True
+        assert worker.registered_layer_names == [
+            "model.layers.0.self_attn",
+            "model.layers.1.linear_attn",
+        ]
+        assert worker.registered_group_indices == [0, 1]
+        assert worker.kv_caches_base_addr == [
+            fa_cache.data_ptr(),
+            gdn_conv_state.data_ptr(),
+        ]
+
+        worker.shutdown()
+        worker.shutdown = noop_shutdown
+        connector.connector_worker = None
+
+
+def test_register_kv_caches_deduplicates_shared_backing_memory(monkeypatch):
+    monkeypatch.setenv("VLLM_MOONCAKE_ABORT_REQUEST_TIMEOUT", "5")
+    vllm_config = create_vllm_config(
+        kv_connector="MooncakeConnector",
+        kv_role="kv_consumer",
+    )
+    kv_cache_config = make_hybrid_gdn_kv_cache_config(
+        vllm_config.cache_config.block_size
+    )
+
+    with set_current_vllm_config(vllm_config), patch_worker_dependencies():
+        connector = MooncakeConnector(
+            vllm_config,
+            KVConnectorRole.WORKER,
+            kv_cache_config,
+        )
+        worker = connector.connector_worker
+
+        backing = torch.empty((4, 64), dtype=torch.float16)
+        fa_cache = backing[:2, :16]
+        gdn_conv_state = backing[:3]
+        gdn_ssm_state = torch.empty((3, 4), dtype=torch.float16)
+
+        with patch.object(
+            worker.engine, "batch_register_memory", return_value=0
+        ) as batch_register_memory:
+            worker.register_kv_caches(
+                {
+                    "model.layers.0.self_attn": fa_cache,
+                    "model.layers.1.linear_attn": (gdn_conv_state, gdn_ssm_state),
+                }
+            )
+
+        assert worker.kv_caches_base_addr == [
+            fa_cache.data_ptr(),
+            gdn_conv_state.data_ptr(),
+        ]
+        batch_register_memory.assert_called_once()
+        registered_ptrs, registered_lens = batch_register_memory.call_args[0]
+        assert registered_ptrs == [backing.data_ptr()]
+        assert registered_lens == [backing.untyped_storage().nbytes()]
+
+        worker.shutdown()
+        worker.shutdown = noop_shutdown
+        connector.connector_worker = None
+
+
+def test_hybrid_gdn_transfer_params_preserve_group_identity(monkeypatch):
+    monkeypatch.setenv("VLLM_MOONCAKE_ABORT_REQUEST_TIMEOUT", "5")
+    vllm_config = create_vllm_config(
+        kv_connector="MooncakeConnector",
+        kv_role="kv_producer",
+    )
+    kv_cache_config = make_hybrid_gdn_kv_cache_config(
+        vllm_config.cache_config.block_size
+    )
+
+    with set_current_vllm_config(vllm_config), patch_worker_dependencies():
+        connector = MooncakeConnector(
+            vllm_config,
+            KVConnectorRole.WORKER,
+            kv_cache_config,
+        )
+        worker = connector.connector_worker
+
+        block_len = 0x100
+        transfer_id = "xfer-hybrid-gdn"
+
+        async def build_transfer_params():
+            send_meta = SendBlockMeta(
+                p_req_id="p-hybrid-gdn",
+                transfer_id=transfer_id,
+                local_block_ids=[
+                    [10, 11],
+                    [NULL_BLOCK_ID, 4],
+                ],
+                ready=asyncio.Event(),
+            )
+            return await worker._build_transfer_params(
+                [("d-hybrid-gdn", send_meta)],
+                xfer_meta,
+                local_regions,
+                remote_regions,
+            )
+
+        xfer_meta = MooncakeXferMetadata(
+            remote_hostname="consumer-host",
+            remote_port=54321,
+            remote_tp_size=1,
+            remote_tp_rank=0,
+            req_blocks={
+                "d-hybrid-gdn": (
+                    transfer_id,
+                    [
+                        [30, 31],
+                        [NULL_BLOCK_ID, 7],
+                    ],
+                )
+            },
+            kv_caches_base_addr=[],
+            block_lens=[],
+            kv_block_lens=[],
+        )
+
+        local_regions = [
+            TransferRegion(
+                layer_name="model.layers.1.linear_attn",
+                layer_index=1,
+                base_addr=0x5000,
+                block_len=block_len,
+                kv_block_len=block_len,
+                group_index=1,
+            ),
+            TransferRegion(
+                layer_name="model.layers.0.self_attn",
+                layer_index=0,
+                base_addr=0x1000,
+                block_len=block_len,
+                kv_block_len=block_len,
+                group_index=0,
+            ),
+        ]
+        remote_regions = [
+            TransferRegion(
+                layer_name="model.layers.1.linear_attn",
+                layer_index=1,
+                base_addr=0x6000,
+                block_len=block_len,
+                kv_block_len=block_len,
+                group_index=1,
+            ),
+            TransferRegion(
+                layer_name="model.layers.0.self_attn",
+                layer_index=0,
+                base_addr=0x2000,
+                block_len=block_len,
+                kv_block_len=block_len,
+                group_index=0,
+            ),
+        ]
+
+        src_ptrs, dst_ptrs, lengths, err_reqs, err_msg = asyncio.run(
+            build_transfer_params()
+        )
+
+        assert err_reqs == []
+        assert err_msg is None
+        assert src_ptrs == [
+            0x5000 + 4 * block_len,
+            0x1000 + 10 * block_len,
+        ]
+        assert dst_ptrs == [
+            0x6000 + 7 * block_len,
+            0x2000 + 30 * block_len,
+        ]
+        assert lengths == [block_len, 2 * block_len]
+
+        worker.shutdown()
+        worker.shutdown = noop_shutdown
+        connector.connector_worker = None
+
+
+def test_logical_to_kernel_block_ids_expands_fa_not_gdn():
+    worker = object.__new__(MooncakeConnectorWorker)
+    worker.shutdown = noop_shutdown
+    worker._physical_blocks_per_logical_kv_block = 17
+    worker.kv_cache_config = make_hybrid_gdn_kv_cache_config(block_size=544)
+
+    block_ids = [[2], [2]]
+    kernel_block_ids = worker._logical_to_kernel_block_ids(block_ids)
+
+    assert kernel_block_ids == [list(range(34, 51)), [2]]
+
+
+def test_hybrid_gdn_splits_fa_regions_but_keeps_gdn_state_whole(
+    monkeypatch,
+):
+    monkeypatch.setenv("VLLM_MOONCAKE_ABORT_REQUEST_TIMEOUT", "5")
+    vllm_config = create_vllm_config(
+        kv_connector="MooncakeConnector",
+        kv_role="kv_producer",
+    )
+    kv_cache_config = make_hybrid_gdn_kv_cache_config(
+        vllm_config.cache_config.block_size
+    )
+
+    with set_current_vllm_config(vllm_config), patch_worker_dependencies():
+        connector = MooncakeConnector(
+            vllm_config,
+            KVConnectorRole.WORKER,
+            kv_cache_config,
+        )
+        worker = connector.connector_worker
+
+        worker.transfer_topo = SimpleNamespace(virtually_split_kv_in_blocks=True)
+        regions = worker._get_transfer_regions(
+            base_addrs=[0x1000, 0x2000],
+            block_lens=[0x100, 0x100],
+            kv_block_lens=[0x40, 0x100],
+            layer_names=[
+                "model.layers.0.self_attn",
+                "model.layers.1.linear_attn",
+            ],
+            layer_indices=[0, 1],
+            group_indices=[0, 1],
+        )
+
+        assert [
+            (region.group_index, region.base_addr, region.kv_block_len)
+            for region in regions
+        ] == [
+            (0, 0x1000, 0x40),
+            (0, 0x1040, 0x40),
+            (1, 0x2000, 0x100),
+        ]
+
+        worker.shutdown()
+        worker.shutdown = noop_shutdown
+        connector.connector_worker = None
@@ -29,9 +29,8 @@ def _gpu_snapshot(tag: str, prev_alloc: float = 0.0) -> dict:
    torch.accelerator.synchronize()
    alloc = torch.accelerator.memory_allocated()
    reserved = torch.accelerator.memory_reserved()
-    # mem_get_info is not available on torch.accelerator
    try:
-        drv_free, drv_total = torch.cuda.mem_get_info()
+        drv_free, drv_total = torch.accelerator.get_memory_info()
        drv_used = drv_total - drv_free
        drv_pct = drv_used / drv_total * 100
    except Exception:
@@ -1285,7 +1285,7 @@ def test_token_logprobs_large_batch_int64_row_offset():
    batch_size = 2**31 // vocab_size + 64  # batch_size * vocab_size > 2**31
    # logits (the large input) plus small logprob/rank outputs; ~1 GB headroom.
    required_bytes = batch_size * vocab_size * 4 + (1 << 30)
-    if torch.cuda.mem_get_info()[0] < required_bytes:
+    if torch.accelerator.get_memory_info()[0] < required_bytes:
        pytest.skip(f"needs ~{required_bytes / 1e9:.0f} GB of free GPU memory")

    logits = torch.randn(batch_size, vocab_size, device=device, dtype=torch.float32)
@@ -426,7 +426,7 @@ class TestTritonTopkTopp:
        # logits is modified in place; the only extra device memory is the
        # per-SM scratch buffer (~num_sm * vocab), so allow ~1 GB of headroom.
        required_bytes = batch_size * vocab_size * 4 + (1 << 30)
-        if torch.cuda.mem_get_info()[0] < required_bytes:
+        if torch.accelerator.get_memory_info()[0] < required_bytes:
            pytest.skip(f"needs ~{required_bytes / 1e9:.0f} GB of free GPU memory")

        logits = torch.randn(
@@ -8,11 +8,12 @@ import regex as re
 # Regex: match `torch.cuda.xxx` but allow `torch.accelerator.xxx`
 # --------------------------------------------------------------------------- #
 _TORCH_CUDA_PATTERNS = [
-    r"\btorch\.cuda\.(empty_cache|synchronize|device_count|current_device|memory_reserved|memory_allocated|max_memory_allocated|max_memory_reserved|reset_peak_memory_stats|memory_stats|set_device|device\()\b",
+    r"\btorch\.cuda\.(empty_cache|synchronize|device_count|current_device|memory_reserved|memory_allocated|max_memory_allocated|max_memory_reserved|reset_peak_memory_stats|memory_stats|mem_get_info|set_device|device\()\b",
    r"\btorch\.cuda\.(manual_seed|manual_seed_all)\b",
    r"\bwith\storch\.cuda\.device\b",
    # Calls torch.cuda.{_is_compiled/_device_count_amdsmi/_device_count_nvml} internally
    r"\bcuda_device_count_stateless\(\)\b",
+    r"\bcurrent_platform\.mem_get_info\(\)\b",
 ]

 ALLOWED_FILES = {
@@ -759,6 +759,7 @@ class CompilationConfig:
        "vllm::sparse_attn_indexer",
        "vllm::rocm_aiter_sparse_attn_indexer",
        "vllm::deepseek_v4_attention",
+        "vllm::hpc_rope_norm_forward",
    ]

    def compute_hash(self) -> str:
@@ -79,7 +79,7 @@ class SchedulerConfig:

    long_prefill_token_threshold: int = Field(default=0, ge=0)
    """For chunked prefill, a request is considered long if the prompt is
-    longer than this number of tokens."""
+    longer than this number of tokens. 0 disables the cap (default)."""

    enable_chunked_prefill: bool = True
    """If True, prefill requests can be chunked based
@@ -22,7 +22,6 @@ from vllm.config import VllmConfig
 from vllm.distributed.kv_transfer.kv_connector.utils import (
    EngineId,
    TransferTopology,
-    get_current_attn_backend,
    get_current_attn_backends,
 )
 from vllm.distributed.kv_transfer.kv_connector.v1.base import (
@@ -51,10 +50,18 @@ from vllm.platforms import current_platform
 from vllm.utils.math_utils import cdiv
 from vllm.utils.network_utils import get_ip, make_zmq_path, make_zmq_socket
 from vllm.v1.attention.backend import AttentionMetadata
-from vllm.v1.attention.backends.utils import get_kv_cache_layout
+from vllm.v1.attention.backends.utils import NULL_BLOCK_ID, get_kv_cache_layout
 from vllm.v1.core.sched.output import SchedulerOutput
-from vllm.v1.kv_cache_interface import FullAttentionSpec, SlidingWindowSpec
+from vllm.v1.kv_cache_interface import (
+    FullAttentionSpec,
+    KVCacheSpec,
+    MambaSpec,
+    MLAAttentionSpec,
+    SlidingWindowMLASpec,
+    SlidingWindowSpec,
+)
 from vllm.v1.request import RequestStatus
+from vllm.v1.worker.block_table import BlockTable
 from vllm.v1.worker.utils import select_common_block_size

 logger = init_logger(__name__)
@@ -85,6 +92,7 @@ class TransferRegion:
    base_addr: int
    block_len: int
    kv_block_len: int
+    group_index: int = 0


 def _get_tp_ratio(local_tp_size: int, remote_tp_size: int) -> int:
@@ -111,24 +119,58 @@ def _get_tp_ratio(local_tp_size: int, remote_tp_size: int) -> int:
 def _expand_transfer_regions(
    base_addrs: list[int],
    block_lens: list[int],
+    kv_block_lens: list[int],
    layer_names: list[str],
    layer_indices: list[int],
    is_kv_layout_blocks_first: bool,
+    group_indices: list[int] | None = None,
+    split_kv_regions: list[bool] | None = None,
 ) -> list[TransferRegion]:
    """Expand registered KV tensors into the regions transferred by Mooncake."""
    assert (
-        len(base_addrs) == len(block_lens) == len(layer_names) == len(layer_indices)
+        len(base_addrs)
+        == len(block_lens)
+        == len(kv_block_lens)
+        == len(layer_names)
+        == len(layer_indices)
    ), (
        "Mooncake transfer regions require matching metadata lengths, got "
        f"base_addrs={len(base_addrs)}, block_lens={len(block_lens)}, "
+        f"kv_block_lens={len(kv_block_lens)}, "
        f"layer_names={len(layer_names)}, "
        f"layer_indices={len(layer_indices)}."
    )
+    if group_indices is None:
+        group_indices = [0] * len(layer_names)
+    assert len(group_indices) == len(layer_names), (
+        "Mooncake transfer regions require matching group metadata lengths, "
+        f"got group_indices={len(group_indices)}, layer_names={len(layer_names)}."
+    )
+    if split_kv_regions is None:
+        split_kv_regions = [is_kv_layout_blocks_first] * len(layer_names)
+    assert len(split_kv_regions) == len(layer_names), (
+        "Mooncake transfer regions require matching split metadata, "
+        f"got split_kv_regions={len(split_kv_regions)}, "
+        f"layer_names={len(layer_names)}."
+    )
    regions: list[TransferRegion] = []
-    for base_addr, block_len, layer_name, layer_index in zip(
-        base_addrs, block_lens, layer_names, layer_indices
+    for (
+        base_addr,
+        block_len,
+        kv_block_len,
+        layer_name,
+        layer_index,
+        group_index,
+        split_kv_region,
+    ) in zip(
+        base_addrs,
+        block_lens,
+        kv_block_lens,
+        layer_names,
+        layer_indices,
+        group_indices,
+        split_kv_regions,
    ):
-        kv_block_len = block_len // 2 if is_kv_layout_blocks_first else block_len
        regions.append(
            TransferRegion(
                layer_name=layer_name,
@@ -136,9 +178,10 @@ def _expand_transfer_regions(
                base_addr=base_addr,
                block_len=block_len,
                kv_block_len=kv_block_len,
+                group_index=group_index,
            )
        )
-        if is_kv_layout_blocks_first:
+        if split_kv_region:
            regions.append(
                TransferRegion(
                    layer_name=layer_name,
@@ -146,6 +189,7 @@ def _expand_transfer_regions(
                    base_addr=base_addr + kv_block_len,
                    block_len=block_len,
                    kv_block_len=kv_block_len,
+                    group_index=group_index,
                )
            )
    return regions
@@ -308,6 +352,17 @@ def _align_transfer_regions(
                    f"{remote_region.layer_index}."
                ),
            )
+        if local_region.group_index != remote_region.group_index:
+            return (
+                [],
+                [],
+                (
+                    "Mooncake registered group index mismatch for "
+                    f"{local_region.layer_name}: producer="
+                    f"{local_region.group_index}, consumer="
+                    f"{remote_region.group_index}."
+                ),
+            )
        aligned_local.append(local_region)
        aligned_remote.append(remote_region)

@@ -332,8 +387,10 @@ class MooncakeXferMetadata(
    req_blocks: dict[ReqId, tuple[TransferId, list[list[int]]]]
    kv_caches_base_addr: list[int]
    block_lens: list[int]
+    kv_block_lens: list[int]
    registered_layer_names: list[str] = msgspec.field(default_factory=list)
    registered_layer_indices: list[int] = msgspec.field(default_factory=list)
+    registered_group_indices: list[int] = msgspec.field(default_factory=list)


 class MooncakeXferResponseStatus(IntEnum):
@@ -581,6 +638,9 @@ class MooncakeConnectorScheduler:
                for g in kv_cache_config.kv_cache_groups
            )
        )
+        # GDN is represented as a MambaSpec in vLLM. This Mooncake MambaSpec
+        # path is currently tested with GDN; Mamba2 is not validated yet.
+        self._has_mamba = kv_cache_config.has_mamba_layers

        # Requests that need to start recv/send.
        # New requests are added by update_state_after_alloc in
@@ -617,6 +677,38 @@ class MooncakeConnectorScheduler:
            for i, blocks in enumerate(block_ids)
        ]

+    def _get_remote_prefill_token_count(self, num_prompt_tokens: int) -> int:
+        """D-side only. Returns N-1 for Mamba models since the decoder
+        always recomputes the last token and must start from h(N-1)."""
+        if self._has_mamba and num_prompt_tokens > 1:
+            return num_prompt_tokens - 1
+        return num_prompt_tokens
+
+    def _truncate_mamba_request_for_prefill(self, request: "Request") -> None:
+        """P-side only: drop the last prompt token so the prefiller computes
+        h(N-1) instead of h(N). The decoder recomputes the last token to
+        derive h(N) correctly.
+
+        Guarded by ``_p_side_truncated`` to avoid repeated truncation if the
+        request is preempted and rescheduled."""
+        params = request.kv_transfer_params
+        if (
+            params is not None
+            and not params.get("_p_side_truncated")
+            and request.num_prompt_tokens > 1
+        ):
+            if request.prompt_token_ids is not None:
+                request.prompt_token_ids.pop()
+            elif request.prompt_embeds is not None:
+                request.prompt_embeds = request.prompt_embeds[:-1]
+            else:
+                return
+
+            request._all_token_ids.pop()
+            request.num_prompt_tokens -= 1
+            request.max_tokens = 1
+            params["_p_side_truncated"] = True
+
    def get_num_new_matched_tokens(
        self, request: "Request", num_computed_tokens: int
    ) -> tuple[int, bool]:
@@ -650,10 +742,15 @@ class MooncakeConnectorScheduler:
            # Remote prefill: get all prompt blocks from remote.
            assert not self.is_kv_producer
            token_ids = request.prompt_token_ids or []
-            count = len(token_ids) - num_computed_tokens
+            count = self._get_remote_prefill_token_count(len(token_ids)) - (
+                num_computed_tokens
+            )
            if count > 0:
                return count, True

+        if params.get("do_remote_decode") and self._has_mamba:
+            self._truncate_mamba_request_for_prefill(request)
+
        # No remote prefill for this request.
        return 0, False

@@ -802,7 +899,7 @@ class MooncakeConnectorWorker:
        self,
        vllm_config: VllmConfig,
        engine_id: str,
-        kv_cache_config: "KVCacheConfig | None" = None,
+        kv_cache_config: "KVCacheConfig",
    ):
        if TransferEngine is None:
            logger.error("Mooncake is not available")
@@ -831,10 +928,15 @@ class MooncakeConnectorWorker:
        protocol = kv_transfer_config.kv_connector_extra_config.get(  # type: ignore[union-attr]
            "mooncake_protocol", "rdma"
        )
+        device_name = kv_transfer_config.kv_connector_extra_config.get(  # type: ignore[union-attr]
+            "device_name", ""
+        )
        logger.info(
            "The Mooncake Transfer Engine is using %s as its protocol.", protocol
        )
-        ret_value = self.engine.initialize(self.hostname, "P2PHANDSHAKE", protocol, "")
+        ret_value = self.engine.initialize(
+            self.hostname, "P2PHANDSHAKE", protocol, device_name
+        )
        if ret_value != 0:
            raise RuntimeError("Mooncake Transfer Engine initialization failed.")

@@ -852,10 +954,11 @@ class MooncakeConnectorWorker:
        self.engine_id: EngineId = engine_id
        self.tp_rank = get_tensor_model_parallel_rank()
        self.tp_size = get_tensor_model_parallel_world_size()
-        self.num_blocks = 0
        self.block_len_per_layer: list[int] = []
+        self.kv_block_len_per_layer: list[int] = []
        self.registered_layer_names: list[str] = []
        self.registered_layer_indices: list[int] = []
+        self.registered_group_indices: list[int] = []
        self.seen_base_addresses: list[int] = []

        assert (parallel_config := vllm_config.parallel_config)
@@ -916,26 +1019,40 @@ class MooncakeConnectorWorker:
        self.cache_config = vllm_config.cache_config
        self.kv_cache_config = kv_cache_config
        self.use_mla = self.model_config.use_mla
+        self._physical_blocks_per_logical_kv_block = 1
        self._sync_block_size_with_kernel()

-        # Get the attention backend from the first layer
-        # NOTE (NickLucche) models with multiple backends are not supported yet
-        backend = get_current_attn_backend(vllm_config)
-        self.backend_name = backend.get_name()
+        self.attn_backends = get_current_attn_backends(vllm_config)
        self.kv_cache_layout = get_kv_cache_layout()
-        logger.debug("Detected attention backend %s", self.backend_name)
+        logger.debug(
+            "Detected attention backends %s",
+            [backend.get_name() for backend in self.attn_backends],
+        )
        logger.debug("Detected kv cache layout %s", self.kv_cache_layout)

        self._tp_size: dict[EngineId, int] = {self.engine_id: self.tp_size}
+        self._layer_specs: dict[str, KVCacheSpec] = {}
+        for group in kv_cache_config.kv_cache_groups:
+            group_spec = group.kv_cache_spec
+            specs_by_layer = getattr(group_spec, "kv_cache_specs", {})
+            for layer_name in group.layer_names:
+                self._layer_specs[layer_name] = specs_by_layer.get(
+                    layer_name, group_spec
+                )
+        self._layer_group_indices: dict[str, int] = {
+            layer: group_index
+            for group_index, group in enumerate(kv_cache_config.kv_cache_groups)
+            for layer in group.layer_names
+        }
        self.transfer_topo = TransferTopology(
            tp_rank=self.tp_rank,
            tp_size=self.tp_size,
            block_size=self.block_size,
            engine_id=self.engine_id,
            is_mla=self.use_mla,
-            is_mamba=False,
+            is_mamba=kv_cache_config.has_mamba_layers,
            total_num_kv_heads=self.model_config.get_total_num_kv_heads(),
-            attn_backends=[backend],
+            attn_backends=self.attn_backends,
        )

        self.async_zmq_ctx = zmq.asyncio.Context()
@@ -958,6 +1075,9 @@ class MooncakeConnectorWorker:
                kernel_block_size,
            )
            assert self.block_size > kernel_block_size
+            self._physical_blocks_per_logical_kv_block = (
+                self.block_size // kernel_block_size
+            )
            self.block_size = kernel_block_size

    def __del__(self):
@@ -1092,14 +1212,18 @@ class MooncakeConnectorWorker:
        local_regions = self._get_transfer_regions(
            self.kv_caches_base_addr,
            self.block_len_per_layer,
+            self.kv_block_len_per_layer,
            self.registered_layer_names,
            self.registered_layer_indices,
+            self.registered_group_indices,
        )
        remote_regions = self._get_transfer_regions(
            meta.kv_caches_base_addr,
            meta.block_lens,
+            meta.kv_block_lens,
            meta.registered_layer_names,
            meta.registered_layer_indices,
+            meta.registered_group_indices,
        )
        local_regions, remote_regions, align_err = _align_transfer_regions(
            local_regions, remote_regions
@@ -1271,6 +1395,32 @@ class MooncakeConnectorWorker:
            remote_tp_ranks,
        )

+    def _logical_to_kernel_block_ids(
+        self, block_ids: list[list[int]]
+    ) -> list[list[int]]:
+        # For example, if a 544-token logical block is served by 32-token
+        # FA kernel blocks, FA block id k expands to [17k, ..., 17k + 16],
+        # while the matching Mamba/GDN state block remains k. Only attention
+        # groups need logical block ids expanded to kernel block ids; Mamba/GDN
+        # state block ids stay in the logical/page-id space.
+        if self._physical_blocks_per_logical_kv_block == 1:
+            return block_ids
+
+        block_arange = np.arange(self._physical_blocks_per_logical_kv_block).reshape(
+            1, -1
+        )
+        group_specs = self.kv_cache_config.kv_cache_groups
+        return [
+            BlockTable.map_to_kernel_blocks(
+                np.array(group),
+                self._physical_blocks_per_logical_kv_block,
+                block_arange,
+            ).tolist()
+            if not isinstance(group_specs[i].kv_cache_spec, MambaSpec)
+            else group
+            for i, group in enumerate(block_ids)
+        ]
+
    async def _build_transfer_params(
        self,
        ready_reqs: list[tuple[ReqId, SendBlockMeta]],
@@ -1293,14 +1443,6 @@ class MooncakeConnectorWorker:
            ):
                continue

-            # Per-group partial hit trimming, then flatten.
-            # With HMA, groups share the same KV tensor but use different
-            # block ranges.  We trim and concatenate so the coalescer and
-            # address math see one flat block list — same as non-HMA, but
-            # now including blocks from every group.
-            local_block_ids: list[int] = []
-            remote_block_ids: list[int] = []
-            has_block_error = False
            if len(send_meta.local_block_ids) != len(remote_block_ids_per_group):
                logger.error(
                    "req %s: KV group count mismatch: local=%d, remote=%d",
@@ -1312,26 +1454,55 @@ class MooncakeConnectorWorker:
                if err_msg is None:
                    err_msg = "KV group count mismatch"
                continue
-            for local_group, remote_group in zip(
-                send_meta.local_block_ids, remote_block_ids_per_group
+
+            # Keep KV-cache group identity. Hybrid/HMA groups can carry
+            # different semantics (e.g. full-attention KV pages vs GDN/Mamba
+            # inner-state slots), so their block IDs must not be flattened and
+            # reused for every registered region.
+            local_block_ids_by_group: list[list[int]] = []
+            remote_block_ids_by_group: list[list[int]] = []
+            has_block_error = False
+            group_specs = self.kv_cache_config.kv_cache_groups
+            for group_index, (local_group, remote_group) in enumerate(
+                zip(send_meta.local_block_ids, remote_block_ids_per_group)
            ):
+                is_mamba_group = isinstance(
+                    group_specs[group_index].kv_cache_spec,
+                    MambaSpec,
+                )
+                if is_mamba_group:
+                    # Mamba/GDN prefix caching can use null blocks only as
+                    # align-mode placeholders. They do not carry transferable
+                    # state, so skip them on both producer and consumer sides.
+                    local_group = [
+                        block_id
+                        for block_id in local_group
+                        if block_id != NULL_BLOCK_ID
+                    ]
+                    remote_group = [
+                        block_id
+                        for block_id in remote_group
+                        if block_id != NULL_BLOCK_ID
+                    ]
+
                n_local = len(local_group)
                n_remote = len(remote_group)
                if n_local < n_remote:
                    logger.error(
                        "req %s: local blocks(%d) < remote blocks(%d) "
-                        "in a KV cache group",
+                        "in a KV cache group (is_mamba_group=%s)",
                        d_req_id,
                        n_local,
                        n_remote,
+                        is_mamba_group,
                    )
                    has_block_error = True
                    break
-                if n_local > n_remote:
+                elif n_local > n_remote:
                    # Partial prefix cache hit: just read uncomputed blocks.
-                    local_group = local_group[-n_remote:]
-                local_block_ids.extend(local_group)
-                remote_block_ids.extend(remote_group)
+                    local_group = local_group[-n_remote:] if n_remote > 0 else []
+                local_block_ids_by_group.append(local_group)
+                remote_block_ids_by_group.append(remote_group)

            if has_block_error:
                err_reqs.append(d_req_id)
@@ -1339,22 +1510,44 @@ class MooncakeConnectorWorker:
                    err_msg = "P num blocks less than D"
                continue

-            if not local_block_ids:
+            if not any(local_block_ids_by_group):
                continue

-            # Group by indices
-            group_local_block_ids, group_remote_block_ids = group_concurrent_contiguous(
-                local_block_ids, remote_block_ids
+            local_block_ids_by_group = self._logical_to_kernel_block_ids(
+                local_block_ids_by_group
+            )
+            remote_block_ids_by_group = self._logical_to_kernel_block_ids(
+                remote_block_ids_by_group
            )

            for local_region, remote_region in zip(local_regions, remote_regions):
-                should_transfer, src_region_offset, dst_region_offset, transfer_len = (
-                    self._get_sender_transfer_plan(
-                        local_kv_block_len=local_region.kv_block_len,
-                        remote_kv_block_len=remote_region.kv_block_len,
-                        remote_tp_rank=agent_meta.remote_tp_rank,
-                        remote_tp_size=agent_meta.remote_tp_size,
-                    )
+                assert local_region.group_index == remote_region.group_index, (
+                    "Aligned Mooncake transfer regions must belong to the same "
+                    "KV group."
+                )
+                group_index = local_region.group_index
+                assert group_index < len(local_block_ids_by_group), (
+                    "Transfer region references a missing KV group."
+                )
+                local_block_ids = local_block_ids_by_group[group_index]
+                remote_block_ids = remote_block_ids_by_group[group_index]
+                if not local_block_ids:
+                    continue
+
+                # Group by indices within this region's KV-cache group only.
+                group_local_block_ids, group_remote_block_ids = (
+                    group_concurrent_contiguous(local_block_ids, remote_block_ids)
+                )
+                (
+                    should_transfer,
+                    src_region_offset,
+                    dst_region_offset,
+                    transfer_len,
+                ) = self._get_sender_transfer_plan(
+                    local_kv_block_len=local_region.kv_block_len,
+                    remote_kv_block_len=remote_region.kv_block_len,
+                    remote_tp_rank=agent_meta.remote_tp_rank,
+                    remote_tp_size=agent_meta.remote_tp_size,
                )
                if not should_transfer:
                    # Replicated KV cache: only one producer rank in the TP group
@@ -1368,7 +1561,7 @@ class MooncakeConnectorWorker:
                    "Computed source transfer region exceeds local KV block size."
                )
                assert dst_region_offset + transfer_len <= remote_region.kv_block_len, (
-                    "Computed destination transfer region exceeds remote KV block size."
+                    "Destination transfer region exceeds remote KV block size."
                )
                # Collapse one contiguous block group into a single larger
                # transfer descriptor when the per-block copy is identical.
@@ -1411,28 +1604,10 @@ class MooncakeConnectorWorker:
                            )
                            lengths.append(transfer_len)

-                if local_region is local_regions[0]:
-                    logger.debug(
-                        "Mooncake transfer plan for request %s: local_tp=%d "
-                        "remote_tp=%d remote_tp_rank=%d local_block_len=%d "
-                        "remote_block_len=%d src_offset=%d dst_offset=%d "
-                        "transfer_len=%d coalesce=%s",
-                        d_req_id,
-                        self.tp_size,
-                        agent_meta.remote_tp_size,
-                        agent_meta.remote_tp_rank,
-                        local_region.block_len,
-                        remote_region.block_len,
-                        src_region_offset,
-                        dst_region_offset,
-                        transfer_len,
-                        can_coalesce,
-                    )
-
            logger.debug(
                "Sending kv_caches for request %s (%d blocks) to %s",
                d_req_id,
-                len(local_block_ids),
+                sum(len(group) for group in local_block_ids_by_group),
                remote_session,
            )

@@ -1480,18 +1655,33 @@ class MooncakeConnectorWorker:

        logger.info("Registering KV_Caches. use_mla: %s", self.use_mla)

-        kv_data_ptrs = []
-        kv_data_lens = []
-        seen_base_addresses = []
+        kv_data_ptrs: list[int] = []
+        kv_data_lens: list[int] = []
+        region_base_addresses: list[int] = []
+        seen_storage_ptrs: set[int] = set()
        self.block_len_per_layer = []
+        self.kv_block_len_per_layer = []
        self.registered_layer_names = []
        self.registered_layer_indices = []
+        self.registered_group_indices = []

-        split_k_and_v = self.transfer_topo.split_k_and_v
-        tensor_size_bytes = None
        for layer_name, cache_or_caches in kv_caches.items():
            layer_index = extract_layer_index(layer_name)
-            cache_list = cache_or_caches if split_k_and_v else [cache_or_caches]
+            layer_spec = self._layer_specs.get(layer_name)
+            if layer_spec is None:
+                logger.debug(
+                    "Skipping layer %s because no KV cache spec is present.",
+                    layer_name,
+                )
+                continue
+            if isinstance(layer_spec, MambaSpec):
+                conv, _ = cache_or_caches
+                cache_list = [conv]
+            else:
+                cache_list = self.transfer_topo.get_transfer_cache_regions(
+                    cache_or_caches, layer_spec
+                )
+
            logger.debug(
                "registering layer %s with %d cache tensor(s)",
                layer_name,
@@ -1501,45 +1691,46 @@ class MooncakeConnectorWorker:
            for cache in cache_list:
                self._log_debug_cache_registration(layer_name, cache)
                base_addr = cache.data_ptr()
-                if base_addr in seen_base_addresses:
-                    continue
-
-                seen_base_addresses.append(base_addr)
-
-                if tensor_size_bytes is None:
-                    tensor_size_bytes = cache.nbytes
-                    self.num_blocks = cache.shape[0]
-                assert cache.shape[0] == self.num_blocks, (
-                    "All kv cache tensors must have the same number of blocks"
-                )
-
-                # Use stride-based block length so RDMA reaches the last
-                # block's padding (e.g. DeepseekV4 MLA alignment). stride(0)
-                # reflects the actual byte distance between consecutive
-                # blocks in GPU memory, which matches or exceeds the
-                # shape-based size.
                block_len = cache.stride(0) * cache.element_size()
+                region_base_addresses.append(base_addr)

+                if isinstance(layer_spec, (MLAAttentionSpec, SlidingWindowMLASpec)):
+                    kv_block_len = layer_spec.page_size_bytes
+                elif self.transfer_topo.virtually_split_kv_in_blocks and not isinstance(
+                    layer_spec, MambaSpec
+                ):
+                    kv_block_len = block_len // 2
+                else:
+                    kv_block_len = block_len
                self.block_len_per_layer.append(block_len)
+                self.kv_block_len_per_layer.append(kv_block_len)
                self.registered_layer_names.append(layer_name)
                self.registered_layer_indices.append(layer_index)
-                kv_data_ptrs.append(base_addr)
-                kv_data_lens.append(self.num_blocks * block_len)
+                self.registered_group_indices.append(
+                    self._layer_group_indices[layer_name]
+                )
+                storage = cache.untyped_storage()
+                storage_addr = storage.data_ptr()
+                if storage_addr not in seen_storage_ptrs:
+                    seen_storage_ptrs.add(storage_addr)
+                    kv_data_ptrs.append(storage_addr)
+                    kv_data_lens.append(storage.nbytes())

-        self.kv_caches_base_addr = seen_base_addresses
-        self.seen_base_addresses = seen_base_addresses
+        self.kv_caches_base_addr = region_base_addresses
+        self.seen_base_addresses = kv_data_ptrs
+
+        if not kv_data_ptrs:
+            raise RuntimeError("No KV cache tensors were registered with Mooncake.")

        ret_value = self.engine.batch_register_memory(kv_data_ptrs, kv_data_lens)
        if ret_value != 0:
            raise RuntimeError("Mooncake batch memory registration failed.")

-        assert tensor_size_bytes is not None
-        assert self.num_blocks != 0
        self.device_kv_caches = kv_caches
        logger.debug(
-            "registered num_blocks=%d block_lens=%s",
-            self.num_blocks,
+            "registered block_lens=%s kv_block_lens=%s",
            self.block_len_per_layer,
+            self.kv_block_len_per_layer,
        )

        # No need to launch server for D node.
@@ -1642,8 +1833,10 @@ class MooncakeConnectorWorker:
            },
            kv_caches_base_addr=self.kv_caches_base_addr,
            block_lens=self.block_len_per_layer,
+            kv_block_lens=self.kv_block_len_per_layer,
            registered_layer_names=self.registered_layer_names,
            registered_layer_indices=self.registered_layer_indices,
+            registered_group_indices=self.registered_group_indices,
        )

        encoded_data = self._encoder.encode(metadata)
@@ -1852,15 +2045,34 @@ class MooncakeConnectorWorker:
        self,
        base_addrs: list[int],
        block_lens: list[int],
+        kv_block_lens: list[int],
        layer_names: list[str],
        layer_indices: list[int],
+        group_indices: list[int] | None = None,
    ) -> list[TransferRegion]:
+        if not group_indices:
+            group_indices = [
+                self._layer_group_indices.get(layer_name, 0)
+                for layer_name in layer_names
+            ]
+        split_kv_regions = None
+        if self.transfer_topo.virtually_split_kv_in_blocks:
+            split_kv_regions = [
+                not isinstance(
+                    self._layer_specs[layer_name],
+                    (MambaSpec, MLAAttentionSpec, SlidingWindowMLASpec),
+                )
+                for layer_name in layer_names
+            ]
        return _expand_transfer_regions(
            base_addrs=base_addrs,
            block_lens=block_lens,
+            kv_block_lens=kv_block_lens,
            layer_names=layer_names,
            layer_indices=layer_indices,
            is_kv_layout_blocks_first=self.transfer_topo.virtually_split_kv_in_blocks,
+            group_indices=group_indices,
+            split_kv_regions=split_kv_regions,
        )

    def _get_sender_transfer_plan(
@@ -320,7 +320,7 @@ class NixlBaseConnectorScheduler:
                    logger.warning("Connection listener got unexpected message %s", msg)
                sock.send_multipart((identity, b"", encoded_data[target_tp_rank]))

-    def _mamba_prefill_token_count(self, num_prompt_tokens: int) -> int:
+    def _get_remote_prefill_token_count(self, num_prompt_tokens: int) -> int:
        """D-side only. Returns N-1 for Mamba models since the decoder
        always recomputes the last token and must start from h(N-1)."""
        if self._has_mamba and num_prompt_tokens > 1:
@@ -60,7 +60,7 @@ class NixlPullConnectorScheduler(NixlBaseConnectorScheduler):
        if params is not None and params.get("do_remote_prefill"):
            # Remote prefill: get all prompt blocks from remote.
            token_ids = request.prompt_token_ids or []
-            actual = self._mamba_prefill_token_count(len(token_ids))
+            actual = self._get_remote_prefill_token_count(len(token_ids))
            count = actual - num_computed_tokens
            if count > 0:
                return count, True
@@ -116,7 +116,7 @@ class NixlPushConnectorScheduler(NixlBaseConnectorScheduler):

        if params is not None and params.get("do_remote_prefill"):
            token_ids = request.prompt_token_ids or []
-            actual = self._mamba_prefill_token_count(len(token_ids))
+            actual = self._get_remote_prefill_token_count(len(token_ids))
            count = actual - num_computed_tokens
            if count > 0:
                return count, True
@@ -318,6 +318,7 @@ def _parse_function_call(message: Message, recipient: str) -> list[ResponseOutpu
            type="function_call",
            name=function_name,
            id=f"fc_{random_id}",
+            status="completed",
        )
        output_items.append(response_item)
    return output_items
@@ -79,6 +79,7 @@ if TYPE_CHECKING:
    VLLM_MAX_AUDIO_CLIP_FILESIZE_MB: int = 25
    VLLM_MAX_AUDIO_DECODE_DURATION_S: int = 600
    VLLM_MAX_AUDIO_PREPROCESS_WORKERS: int = max(1, min(os.cpu_count() or 1, 2))
+    VLLM_MAX_IMAGE_PIXELS: int = 178_956_970
    VLLM_VIDEO_LOADER_BACKEND: str = "opencv"
    VLLM_MEDIA_CONNECTOR: str = "http"
    VLLM_MM_HASHER_ALGORITHM: str = "blake3"
@@ -954,6 +955,13 @@ environment_variables: dict[str, Callable[[], Any]] = {
            str(max(1, min(os.cpu_count() or 1, 2))),
        )
    ),
+    # Maximum decoded image size in pixels.  Small compressed images can
+    # expand into gigabytes of raster memory.  This limit is enforced before
+    # decoding so the memory is never allocated.  Default matches PIL's
+    # built-in 2x decompression-bomb threshold (~179M pixels, ~680 MB RGB).
+    "VLLM_MAX_IMAGE_PIXELS": lambda: int(
+        os.getenv("VLLM_MAX_IMAGE_PIXELS", "178956970")
+    ),
    # Backend for Video IO — selects the frame-sampling algorithm.
    # - "opencv": uniform sampling.
    # - "opencv_dynamic": duration-aware dynamic sampling.
@@ -2083,6 +2091,7 @@ def compile_factors() -> dict[str, object]:
        "VLLM_MAX_AUDIO_CLIP_FILESIZE_MB",
        "VLLM_MAX_AUDIO_DECODE_DURATION_S",
        "VLLM_MAX_AUDIO_PREPROCESS_WORKERS",
+        "VLLM_MAX_IMAGE_PIXELS",
        "VLLM_VIDEO_LOADER_BACKEND",
        "VLLM_MEDIA_CONNECTOR",
        "VLLM_OBJECT_STORAGE_SHM_BUFFER_NAME",
@@ -458,6 +458,7 @@ class Attention(nn.Module, AttentionLayerBase):
        # shape does not match the query shape, so we optionally let the model
        # definition specify the output tensor shape.
        output_shape: torch.Size | None = None,
+        output_dtype: torch.dtype | None = None,
    ) -> torch.Tensor:
        """
        The KV cache is stored inside this class and is accessed via
@@ -472,7 +473,8 @@ class Attention(nn.Module, AttentionLayerBase):
            torch.ops.vllm.maybe_calc_kv_scales(
                query, key, value, _encode_layer_name(self.layer_name)
            )
-        output_dtype = query.dtype
+        if output_dtype is None:
+            output_dtype = query.dtype
        if self.query_quant is not None:
            # quantizing with a simple torch operation enables
            # torch.compile to fuse this into previous ops
@@ -301,6 +301,13 @@ def convert_to_unquantized_kernel_format(
            is_gated_act_gemm=is_act_and_mul,
        )

+    if (
+        unquantized_backend == UnquantizedMoeBackend.TRITON
+        and current_platform.is_rocm()
+        and envs.VLLM_ROCM_MOE_PADDING
+    ):
+        # Skip .contiguous(): it would undo the ROCm MoE weight padding.
+        return w13_weight, w2_weight
    return w13_weight.contiguous(), w2_weight.contiguous()


@@ -0,0 +1,11 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+
+from vllm.model_executor.layers.hpc.hpc_module import HpcModule
+from vllm.model_executor.layers.hpc.rope_norm import HpcRopeNorm, QkNormPolicy
+
+__all__ = [
+    "HpcModule",
+    "HpcRopeNorm",
+    "QkNormPolicy",
+]
@@ -0,0 +1,18 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+import torch.nn as nn
+
+
+class HpcModule(nn.Module):
+    def __init__(self):
+        super().__init__()
+
+    @classmethod
+    def support(cls, *args, **kwargs):
+        return True
+
+    def process_weights_after_loading(self, model):
+        pass
+
+    def forward(self, *args, **kwargs):
+        pass
@@ -0,0 +1,408 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""HPC fused RoPE + QK-Norm + KV-Cache-Write (+ optional FP8 Q quant).
+
+Decoupled from HpcAttentionImpl; extra params are passed via layer attrs.
+"""
+
+from __future__ import annotations
+
+from enum import IntEnum
+from typing import Any
+
+import torch
+
+from vllm.config import get_current_vllm_config_or_none
+from vllm.forward_context import ForwardContext, get_forward_context
+from vllm.logger import init_logger
+from vllm.model_executor.custom_op import CustomOp
+from vllm.model_executor.layers.hpc.hpc_module import HpcModule
+from vllm.utils.torch_utils import direct_register_custom_op
+from vllm.v1.attention.backends.hpc_attn import HpcAttnMetadata
+from vllm.v1.attention.backends.registry import AttentionBackendEnum
+
+logger = init_logger(__name__)
+
+_hpc_rope_norm_instances: dict[str, HpcRopeNorm] = {}
+
+
+class QkNormPolicy(IntEnum):
+    """Order of QK-RMSNorm relative to RoPE in the fused HPC rope_norm kernel.
+
+    The values are part of the HPC kernel ABI (passed through as ints), so they
+    must stay in sync with the kernel's expectations.
+    """
+
+    # No QK-Norm: apply RoPE only.
+    NONE = 0
+    # Apply RoPE first, then QK-RMSNorm.
+    ROPE_THEN_NORM = 1
+    # Apply QK-RMSNorm first, then RoPE (e.g. HunYuan V3).
+    NORM_THEN_ROPE = 2
+
+
+def hpc_rope_norm_forward(
+    qkv: torch.Tensor,
+    output: torch.Tensor,
+    layer_name: str,
+) -> None:
+    """Top-level custom op: RoPE + QK-Norm + KV-Cache-Write + FP8 Q quant.
+
+    Fully opaque to torch.compile (dynamo).
+    """
+    forward_context: ForwardContext = get_forward_context()
+    attn_metadata: Any = forward_context.attn_metadata
+    if isinstance(attn_metadata, dict):
+        attn_metadata = attn_metadata[layer_name]
+
+    if attn_metadata is None:
+        output.zero_()
+        return
+
+    attn_layer = forward_context.no_compile_layers[layer_name]
+    # bind_kv_cache stores the per-layer KV cache as a single 5D tensor
+    # (num_blocks, 2, block_size, num_kv_heads, head_size), so use it directly.
+    kv_cache = attn_layer.kv_cache
+
+    if kv_cache.numel() == 0:
+        output.zero_()
+        return
+
+    assert kv_cache.dim() == 5, (
+        f"Expected kv_cache to have 5 dims, got {tuple(kv_cache.shape)}"
+    )
+
+    rope_norm = _hpc_rope_norm_instances[layer_name]
+    rope_norm._forward_impl(qkv, kv_cache, attn_metadata, attn_layer, output)
+
+
+def hpc_rope_norm_forward_fake(
+    qkv: torch.Tensor,
+    output: torch.Tensor,
+    layer_name: str,
+) -> None:
+    """Fake impl for torch.compile trace; output is a mutated arg."""
+    return
+
+
+direct_register_custom_op(
+    op_name="hpc_rope_norm_forward",
+    op_func=hpc_rope_norm_forward,
+    mutates_args=["output"],
+    fake_impl=hpc_rope_norm_forward_fake,
+)
+
+
+@CustomOp.register("hpc_rope_norm")
+class HpcRopeNorm(CustomOp, HpcModule):
+    """HPC fused RoPE + QK-Norm + KV-Cache-Write (+ optional FP8 Q quant).
+
+    Registered as a sub-module in model layers (e.g. HunYuanAttention).
+    Norm weights are extracted from fallback norm modules via
+    process_weights_after_loading() after all weights are loaded.
+
+    forward() is dispatched by CustomOp framework:
+    - In compiled mode: forward_cuda() calls torch.ops.vllm.hpc_rope_norm_forward
+      as a splitting point — internal Python control flow is opaque
+      to torch.compile and not captured by CUDA Graph.
+    - In eager/native mode: forward_native() falls back to forward_cuda().
+    """
+
+    def __init__(
+        self,
+        num_heads: int,
+        num_kv_heads: int,
+        head_dim: int,
+        cos_sin_cache: torch.Tensor,
+        use_qk_norm: bool,
+        fallback_qnorm: torch.nn.Module | None,
+        fallback_knorm: torch.nn.Module | None,
+        kv_cache_dtype: str,
+        layer_name: str,
+        qk_norm_policy: QkNormPolicy = QkNormPolicy.ROPE_THEN_NORM,
+    ) -> None:
+        super().__init__()
+        self.num_heads = num_heads
+        self.num_kv_heads = num_kv_heads
+        self.head_dim = head_dim
+
+        self.use_qk_norm = use_qk_norm
+
+        self.q_size = num_heads * head_dim
+        self.kv_size = num_kv_heads * head_dim
+
+        # Register as a non-persistent buffer so it participates in sleep
+        # level-2 save/restore (CuMemAllocator) but is excluded from the
+        # checkpoint state_dict.
+        self.register_buffer("cos_sin_cache", cos_sin_cache.float(), persistent=False)
+
+        self.fallback_qnorm = fallback_qnorm
+        self.fallback_knorm = fallback_knorm
+
+        self.head_per_group = num_heads // num_kv_heads
+
+        # Pre-allocate norm weight tensors as Parameters so they are tracked by
+        # CuMemAllocator (for sleep/wake_up) and have stable addresses for CUDA
+        # Graph replay. process_weights_after_loading() updates them inplace via
+        # copy_() so refit does not invalidate captured graph tensor pointers.
+        # Shape is [head_dim] to match the HPC kernel's q/k_norm_weight layout.
+        if use_qk_norm and fallback_qnorm is not None:
+            self.qnorm_weight: torch.nn.Parameter | None = torch.nn.Parameter(
+                torch.empty(head_dim, dtype=torch.float32),
+                requires_grad=False,
+            )
+        else:
+            self.qnorm_weight = None
+        if use_qk_norm and fallback_knorm is not None:
+            self.knorm_weight: torch.nn.Parameter | None = torch.nn.Parameter(
+                torch.empty(head_dim, dtype=torch.float32),
+                requires_grad=False,
+            )
+        else:
+            self.knorm_weight = None
+
+        self.use_fp8 = "fp8" in kv_cache_dtype
+        # The RMSNorm/RoPE ordering is model dependent (e.g. HunYuan V3 applies
+        # QK-Norm before RoPE -> NORM_THEN_ROPE), so it is supplied by the
+        # caller. When QK-Norm is disabled the policy is forced to NONE.
+        self.qk_norm_policy = qk_norm_policy if use_qk_norm else QkNormPolicy.NONE
+
+        # Register layer_name + add self to the global instance registry so the
+        # module-level custom op (hpc_rope_norm_forward) can route back here.
+        self.layer_name: str | None = None
+        self.register_layer_name(layer_name)
+
+    @classmethod
+    def support(
+        cls,
+        num_heads: int,
+        num_kv_heads: int,
+        head_dim: int,
+        kv_cache_dtype: str,
+    ) -> bool:
+        """Check whether HpcRopeNorm is supported for the given config."""
+        # HpcRopeNorm is only enabled together with the HPC attention backend.
+        vllm_config = get_current_vllm_config_or_none()
+        if (
+            vllm_config is None
+            or vllm_config.attention_config.backend != AttentionBackendEnum.HPC_ATTN
+        ):
+            return False
+
+        if kv_cache_dtype not in ("fp8_e4m3", "auto"):
+            logger.warning_once(
+                f"hpc rope_norm not support kv_cache_dtype:{kv_cache_dtype}, "
+                "only support fp8_e4m3, bfloat16"
+            )
+            return False
+
+        if head_dim not in (128,):
+            logger.warning_once("hpc rope_norm only support head_dim == 128.")
+            return False
+
+        head_per_group = num_heads // num_kv_heads
+        if head_per_group not in (4, 8):
+            logger.warning_once("hpc rope_norm only support head_per_group in [4, 8].")
+            return False
+
+        logger.info_once("enable hpc rope_norm")
+        return True
+
+    def process_weights_after_loading(self, model: torch.nn.Module = None) -> None:
+        """Copy norm weights (float32) from fallback norm modules inplace.
+
+        Uses copy_() to preserve tensor addresses for CUDA Graph / refit
+        compatibility. Called by the model's load_weights() after all weights
+        are loaded (and generically from the model loader for DummyModelLoader
+        / sleep-wake_up reload paths).
+        """
+        if self.use_qk_norm:
+            if self.fallback_qnorm is not None and self.qnorm_weight is not None:
+                self.qnorm_weight.data.copy_(self.fallback_qnorm.weight.data.float())
+            if self.fallback_knorm is not None and self.knorm_weight is not None:
+                self.knorm_weight.data.copy_(self.fallback_knorm.weight.data.float())
+
+    def register_layer_name(self, layer_name: str) -> None:
+        """Register layer_name and add self to the global registry.
+
+        The global registry is needed because the bottom-level torch op
+        (hpc_rope_norm_forward) is a module-level function and needs to
+        route back to the correct instance via layer_name.
+        """
+        self.layer_name = layer_name
+        _hpc_rope_norm_instances[layer_name] = self
+        logger.debug(
+            "[rope_norm] registered HpcRopeNorm for layer: %s",
+            layer_name,
+        )
+
+    def forward_native(
+        self,
+        qkv: torch.Tensor,
+        layer_name: str,
+    ) -> torch.Tensor:
+        """Native fallback path: delegates to forward_cuda().
+
+        For now, the default native path will use CUDA backend path.
+        Other platforms may override via OOT registration.
+        """
+        return self.forward_cuda(qkv, layer_name)
+
+    def forward_cuda(
+        self,
+        qkv: torch.Tensor,
+        layer_name: str,
+    ) -> torch.Tensor:
+        """CUDA path: invoke the torch custom op as a compile splitting point."""
+        num_tokens = qkv.shape[0]
+        output = torch.empty(
+            (num_tokens, self.num_heads, self.head_dim),
+            dtype=torch.float8_e4m3fn if self.use_fp8 else qkv.dtype,
+            device=qkv.device,
+        )
+
+        torch.ops.vllm.hpc_rope_norm_forward(qkv, output, layer_name)
+        return output
+
+    def _forward_impl(
+        self,
+        qkv: torch.Tensor,
+        kv_cache: torch.Tensor,
+        attn_metadata: HpcAttnMetadata,
+        attn_layer: torch.nn.Module,
+        output: torch.Tensor,
+    ) -> None:
+        """Actual forward logic called by the custom op.
+
+        Writes processed q into *output* and attaches extra params
+        (e.g. FP8 scales) to *attn_layer* as attributes.
+        """
+        import hpc
+
+        num_actual_tokens = attn_metadata.num_actual_tokens
+        num_prefill_reqs = attn_metadata.num_prefills
+        num_decode_reqs = attn_metadata.num_decodes
+        num_decode_tokens = attn_metadata.num_decode_tokens
+
+        qkv = qkv[:num_actual_tokens]
+
+        num_prefill_tokens = num_actual_tokens - num_decode_tokens
+
+        # KV cache for the FP8 path is stored as uint8; view it as fp8 so the
+        # rope_norm_store_kv_fp8 kernel can write quantized K/V in-place.
+        if self.use_fp8:
+            kv_cache = kv_cache.view(torch.float8_e4m3fn)
+
+        # Per-tensor K/V scales (shape [1]) used by the FP8 kernel.
+        k_scale = attn_layer._k_scale.reshape(1)
+        v_scale = attn_layer._v_scale.reshape(1)
+
+        q_norm_weight = (
+            self.qnorm_weight if self.qk_norm_policy != QkNormPolicy.NONE else None
+        )
+        k_norm_weight = (
+            self.knorm_weight if self.qk_norm_policy != QkNormPolicy.NONE else None
+        )
+
+        # Dynamic per-token-per-head Q quant + per-tensor K/V (dqskv).
+        # rope_norm_store_kv_fp8 is registered as a torch op whose ``quant_policy``
+        # argument is typed as ``int``; pybind cannot cast the hpc.QuantType enum
+        # automatically, so pass its integer ``.value``.
+        QUANT_POLICY_DQSKV = hpc.QuantType.QPERTOKEN_PERHEAD_KPERTENSOR_VPERTENSOR.value
+
+        # --- Prefill ---
+        if num_prefill_reqs > 0:
+            seq_lens_prefill = attn_metadata.seq_lens[num_decode_reqs:]
+            cu_seqlens_prefill = attn_metadata.qo_indptr
+            max_seqlens = attn_metadata.max_query_len
+            block_table_prefill = attn_metadata.block_table_tensor[num_decode_reqs:]
+            qkv_prefill = qkv[num_decode_tokens:]
+            out_q_prefill = output[
+                num_decode_tokens : num_decode_tokens + num_prefill_tokens
+            ]
+
+            if self.use_fp8:
+                _, q_scale, split_k_flag = hpc.rope_norm_store_kv_fp8(
+                    key_cache=kv_cache[:, 0],
+                    value_cache=kv_cache[:, 1],
+                    qkv=qkv_prefill,
+                    cos_sin=self.cos_sin_cache,
+                    num_seqlen_per_req=seq_lens_prefill,
+                    q_index=cu_seqlens_prefill,
+                    kvcache_indices=block_table_prefill,
+                    is_prefill=True,
+                    k_scale=k_scale,
+                    v_scale=v_scale,
+                    quant_policy=QUANT_POLICY_DQSKV,
+                    max_seqlens=max_seqlens,
+                    q_norm_weight=q_norm_weight,
+                    k_norm_weight=k_norm_weight,
+                    qk_norm_policy=self.qk_norm_policy,
+                    out_q=out_q_prefill,
+                )
+                attn_metadata.hpc_prefill_q_scale = q_scale
+            else:
+                hpc.rope_norm_store_kv(
+                    kv_cache[:, 0],
+                    kv_cache[:, 1],
+                    qkv_prefill,
+                    self.cos_sin_cache,
+                    seq_lens_prefill,
+                    cu_seqlens_prefill,
+                    block_table_prefill,
+                    True,  # is_prefill
+                    q_norm_weight=q_norm_weight,
+                    k_norm_weight=k_norm_weight,
+                    out_q=out_q_prefill,
+                    qk_norm_policy=self.qk_norm_policy,
+                )
+
+        # --- Decode ---
+        if num_decode_reqs > 0:
+            num_seq_kvcache = attn_metadata.seq_lens[:num_decode_reqs]
+            block_table_decode = attn_metadata.block_table_tensor[:num_decode_reqs]
+            qkv_decode = qkv[:num_decode_tokens]
+            # Single-token decode: q_index is the per-request prefix sum
+            # [0, 1, ..., num_decode_reqs].
+            qo_indptr_decode = torch.arange(
+                num_decode_reqs + 1, dtype=torch.int32, device=qkv.device
+            )
+            out_q_decode = output[:num_decode_tokens]
+
+            if self.use_fp8:
+                _, q_scale, split_k_flag = hpc.rope_norm_store_kv_fp8(
+                    key_cache=kv_cache[:, 0],
+                    value_cache=kv_cache[:, 1],
+                    qkv=qkv_decode,
+                    cos_sin=self.cos_sin_cache,
+                    num_seqlen_per_req=num_seq_kvcache,
+                    q_index=qo_indptr_decode,
+                    kvcache_indices=block_table_decode,
+                    is_prefill=False,
+                    k_scale=k_scale,
+                    v_scale=v_scale,
+                    quant_policy=QUANT_POLICY_DQSKV,
+                    max_seqlens=1,
+                    q_norm_weight=q_norm_weight,
+                    k_norm_weight=k_norm_weight,
+                    qk_norm_policy=self.qk_norm_policy,
+                    out_q=out_q_decode,
+                )
+                attn_metadata.hpc_decode_q_scale = q_scale
+                if split_k_flag is not None:
+                    attn_metadata.hpc_split_k_flag = split_k_flag
+            else:
+                hpc.rope_norm_store_kv(
+                    kv_cache[:, 0],
+                    kv_cache[:, 1],
+                    qkv_decode,
+                    self.cos_sin_cache,
+                    num_seq_kvcache,
+                    qo_indptr_decode,
+                    block_table_decode,
+                    False,  # is_prefill
+                    q_norm_weight=q_norm_weight,
+                    k_norm_weight=k_norm_weight,
+                    out_q=out_q_decode,
+                    qk_norm_policy=self.qk_norm_policy,
+                )
@@ -20,6 +20,7 @@ from vllm.model_executor.layers.attention import (
    MLAAttention,
    MMEncoderAttention,
 )
+from vllm.model_executor.layers.hpc import HpcModule
 from vllm.model_executor.layers.quantization.base_config import (
    QuantizationConfig,
    QuantizeMethodBase,
@@ -125,6 +126,15 @@ def process_weights_after_loading(
            with device_loading_context(module, target_device):
                module.process_weights_after_loading(model_config.dtype)

+    # Process HPC modules (HpcRopeNorm, etc.) that rely on
+    # process_weights_after_loading being called from the model's
+    # load_weights(). When using DummyModelLoader (e.g. profiling or
+    # sleep/wake_up reload), the model's load_weights() is not called, so we
+    # must handle HPC modules here generically.
+    for _, module in model.named_modules():
+        if isinstance(module, HpcModule):
+            module.process_weights_after_loading(model)
+
    # Needed for torchao model reloading via model.reload_weights
    # @kylesayrs @jerryzh168 this can be removed if callers move to `reload_weights`
    if model_config.quantization == "torchao":
@@ -63,7 +63,6 @@ from vllm.multimodal.processing.processor import (
    PromptUpdate,
    PromptUpdateDetails,
 )
-from vllm.platforms import current_platform
 from vllm.sequence import IntermediateTensors
 from vllm.utils.tensor_schema import TensorSchema, TensorShape

@@ -1274,7 +1273,7 @@ class Gemma4ForConditionalGeneration(
        # pass has already allocated activations we should account for.
        last_hidden_states_map: dict[int, torch.Tensor] = {}
        for patches, items in buckets.items():
-            free, total = current_platform.mem_get_info()
+            free, total = torch.accelerator.get_memory_info()
            max_batch_size = min(
                len(items),
                self._encoder_chunk(
@@ -1382,7 +1381,7 @@ class Gemma4ForConditionalGeneration(
            fc_list = list(frame_counts)

        total_frames = pixel_values.shape[0]
-        free, total = current_platform.mem_get_info()
+        free, total = torch.accelerator.get_memory_info()
        max_batch_size = min(
            total_frames,
            self._encoder_chunk(
@@ -48,6 +48,7 @@ from vllm.model_executor.layers.fused_moe import (
    GateLinear,
    fused_moe_make_expert_params_mapping,
 )
+from vllm.model_executor.layers.hpc import HpcRopeNorm, QkNormPolicy
 from vllm.model_executor.layers.layernorm import RMSNorm
 from vllm.model_executor.layers.linear import (
    MergedColumnParallelLinear,
@@ -234,6 +235,7 @@ class HYV3Attention(nn.Module):
        dual_chunk_attention_config: dict[str, Any] | None = None,
    ) -> None:
        super().__init__()
+        self.dtype = torch.get_default_dtype()
        self.hidden_size = hidden_size
        tp_size = get_tensor_model_parallel_world_size()
        self.total_num_heads = num_heads
@@ -276,11 +278,18 @@ class HYV3Attention(nn.Module):
            quant_config=quant_config,
            prefix=f"{prefix}.o_proj",
        )
+        # When the HPC fused RoPE+QK-Norm path is enabled, the RoPE cos/sin
+        # cache must be float32 to match the HPC kernel's expectations.
+        kv_cache_dtype = cache_config.cache_dtype if cache_config else "auto"
+        rope_support = HpcRopeNorm.support(
+            self.num_heads, self.num_kv_heads, self.head_dim, kv_cache_dtype
+        )
        self.rotary_emb = get_rope(
            self.head_dim,
            max_position=max_position_embeddings,
            rope_parameters=rope_parameters,
            is_neox_style=True,
+            dtype=torch.float32 if rope_support else torch.get_default_dtype(),
        )
        self.attn = Attention(
            self.num_heads,
@@ -295,6 +304,27 @@ class HYV3Attention(nn.Module):
            self.q_norm = RMSNorm(self.head_dim, rms_norm_eps)
            self.k_norm = RMSNorm(self.head_dim, rms_norm_eps)

+        # HPC fused RoPE + QK-Norm + KV-Cache-Write (+ optional FP8 Q quant).
+        # HunYuan V3 applies QK-Norm *before* RoPE, so NORM_THEN_ROPE.
+        self.hpc_rope_norm: HpcRopeNorm | None = None
+        if rope_support:
+            self.hpc_rope_norm = HpcRopeNorm(
+                num_heads=self.num_heads,
+                num_kv_heads=self.num_kv_heads,
+                head_dim=self.head_dim,
+                cos_sin_cache=self.rotary_emb.cos_sin_cache,
+                use_qk_norm=self.use_qk_norm,
+                fallback_qnorm=self.q_norm if self.use_qk_norm else None,
+                fallback_knorm=self.k_norm if self.use_qk_norm else None,
+                kv_cache_dtype=kv_cache_dtype,
+                layer_name=self.attn.layer_name,
+                qk_norm_policy=QkNormPolicy.NORM_THEN_ROPE,
+            )
+            # FP8 Q is produced by HpcRopeNorm, so the attention layer must not
+            # re-quantize the query.
+            if self.hpc_rope_norm.use_fp8 and hasattr(self.attn, "query_quant"):
+                self.attn.query_quant = None
+
    def forward(
        self,
        positions: torch.Tensor,
@@ -303,20 +333,28 @@ class HYV3Attention(nn.Module):
        qkv, _ = self.qkv_proj(hidden_states)
        q, k, v = qkv.split([self.q_size, self.kv_size, self.kv_size], dim=-1)
        output_shape = None
-        if self.use_qk_norm:
-            q_by_head = q.view(
-                *q.shape[:-1], q.shape[-1] // self.head_dim, self.head_dim
-            )
-            q_by_head = self.q_norm(q_by_head)
-            q = q_by_head.view(q.shape)
+        if self.hpc_rope_norm is not None:
+            # HPC handles QK-Norm + RoPE + KV-cache write (+ optional FP8 Q
+            # quant) internally and returns the processed query. K/V are
+            # written into the paged cache by the fused op.
+            q = self.hpc_rope_norm(qkv, self.attn.layer_name)
+            q = q.view(-1, self.num_heads * self.head_dim)
+            attn_output = self.attn(q, k, v, output_shape, self.dtype)
+        else:
+            if self.use_qk_norm:
+                q_by_head = q.view(
+                    *q.shape[:-1], q.shape[-1] // self.head_dim, self.head_dim
+                )
+                q_by_head = self.q_norm(q_by_head)
+                q = q_by_head.view(q.shape)

-            k_by_head = k.view(
-                *k.shape[:-1], k.shape[-1] // self.head_dim, self.head_dim
-            )
-            k_by_head = self.k_norm(k_by_head)
-            k = k_by_head.view(k.shape)
-        q, k = self.rotary_emb(positions, q, k)
-        attn_output = self.attn(q, k, v, output_shape)
+                k_by_head = k.view(
+                    *k.shape[:-1], k.shape[-1] // self.head_dim, self.head_dim
+                )
+                k_by_head = self.k_norm(k_by_head)
+                k = k_by_head.view(k.shape)
+            q, k = self.rotary_emb(positions, q, k)
+            attn_output = self.attn(q, k, v, output_shape)
        attn_output = attn_output.view(q.shape[0], -1)
        output, _ = self.o_proj(attn_output)
        return output
@@ -9,6 +9,7 @@ import pybase64
 import torch
 from PIL import Image

+import vllm.envs as envs
 from vllm.utils.serial_utils import tensor2base64

 from ..image import convert_image_mode, normalize_image, rgba_to_rgb
@@ -72,6 +73,14 @@ class ImageMediaIO(MediaIO[Image.Image]):
    def load_bytes(self, data: bytes) -> MediaWithBytes[Image.Image]:
        try:
            image = Image.open(BytesIO(data))
+            w, h = image.size
+            max_pixels = envs.VLLM_MAX_IMAGE_PIXELS
+            if max_pixels > 0 and w * h > max_pixels:
+                raise ValueError(
+                    f"Image dimensions {w}x{h} ({w * h} pixels) exceed "
+                    f"the maximum of {max_pixels} pixels. Set "
+                    f"VLLM_MAX_IMAGE_PIXELS to increase this limit."
+                )
            image = normalize_image(image)
            image.load()
            image = self._convert_image_mode(image)
@@ -13,6 +13,7 @@ import numpy as np
 import numpy.typing as npt
 import torch

+from vllm import envs
 from vllm.logger import init_logger
 from vllm.utils.import_utils import PlaceholderModule
 from vllm.utils.mem_constants import MiB_bytes
@@ -88,6 +89,18 @@ def get_video_loader_backend_for_processor(
    return VIDEO_LOADER_REGISTRY.get_backend_for_video_processor(video_processor)


+def _check_frame_pixel_limit(width: int, height: int) -> None:
+    """Reject video frames exceeding VLLM_MAX_IMAGE_PIXELS before decoding."""
+    max_pixels = envs.VLLM_MAX_IMAGE_PIXELS
+    if max_pixels > 0 and width * height > max_pixels:
+        raise ValueError(
+            f"Video frame dimensions {width}x{height} "
+            f"({width * height} pixels) exceed the maximum of "
+            f"{max_pixels} pixels. Set VLLM_MAX_IMAGE_PIXELS to "
+            f"increase this limit."
+        )
+
+
 def resize_video(frames: npt.NDArray, size: tuple[int, int]) -> npt.NDArray:
    num_frames, _, _, channels = frames.shape
    new_height, new_width = size
@@ -733,6 +746,7 @@ class PyNvVideoCodecVideoBackendMixin:
                temp_file.write(data)

            gpu_source = cls._read_source_metadata(temp_path, nvc)
+            _check_frame_pixel_limit(gpu_source.width, gpu_source.height)
            source = cls._prepare_source(gpu_source.source)
            frame_idx = cls.compute_frames_index_to_sample(
                source=source, target=target, **kwargs
@@ -835,6 +849,10 @@ class VideoBackend(

        if backend == "opencv":
            cap = cls.open_video_capture(data)
+            _check_frame_pixel_limit(
+                int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)),
+                int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)),
+            )
            source = cls._prepare_source(cls.get_video_metadata(cap))
            frame_idx = cls.compute_frames_index_to_sample(
                source=source, target=target, **kwargs
@@ -850,6 +868,8 @@ class VideoBackend(
                "frame_recovery is only available for `opencv` backend"
            )
            with av.open(BytesIO(data)) as container:
+                stream = container.streams.video[0]
+                _check_frame_pixel_limit(stream.width, stream.height)
                source = cls._prepare_source(cls.get_metadata(container))
                frame_idx = cls.compute_frames_index_to_sample(
                    source=source, target=target, **kwargs
@@ -1609,6 +1629,10 @@ class Molmo2VideoBackend(VideoLoader, OpenCVVideoBackendMixin):
        **kwargs,
    ) -> tuple[npt.NDArray, dict[str, Any]]:
        cap = cls.open_video_capture(data)
+        _check_frame_pixel_limit(
+            int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)),
+            int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)),
+        )

        source = OpenCVVideoBackendMixin.get_video_metadata(cap)
        target = VideoTargetMetadata(
@@ -1758,6 +1782,10 @@ class OpenCVDynamicOpenPanguVideoBackend(VideoLoader, OpenCVVideoBackendMixin):
            Tuple of (frames_array, metadata_dict)
        """
        cap = cls.open_video_capture(data)
+        _check_frame_pixel_limit(
+            int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)),
+            int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)),
+        )

        source = OpenCVVideoBackendMixin.get_video_metadata(cap)

@@ -10,6 +10,7 @@ names so that :class:`ReasoningParserManager` and
 from vllm.parser.engine.adapters import make_adapters
 from vllm.parser.gemma4 import Gemma4Parser
 from vllm.parser.glm47_moe import Glm47MoeParser
+from vllm.parser.kimi_k2 import KimiK2Parser
 from vllm.parser.minimax_m2 import MinimaxM2Parser
 from vllm.parser.nemotron_v3 import NemotronV3Parser
 from vllm.parser.qwen3 import Qwen3Parser
@@ -44,3 +45,8 @@ from vllm.parser.seed_oss import SeedOssParser
    Glm47MoeParserReasoningAdapter,
    Glm47MoeParserToolAdapter,
 ) = make_adapters(Glm47MoeParser)
+
+(
+    KimiK2ParserReasoningAdapter,
+    KimiK2ParserToolAdapter,
+) = make_adapters(KimiK2Parser)
@@ -0,0 +1,285 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Kimi K2 parser for reasoning and tool calls.
+
+Kimi K2 tool call format::
+
+    <|tool_calls_section_begin|>
+    <|tool_call_begin|>functions.get_weather:0
+    <|tool_call_argument_begin|>{"city": "Tokyo"}<|tool_call_end|>
+    <|tool_calls_section_end|>
+
+The header before ``<|tool_call_argument_begin|>`` is Kimi's native tool
+call id. The function name is the final component before ``:N``.
+"""
+
+from __future__ import annotations
+
+import functools
+from collections.abc import Sequence
+from typing import TYPE_CHECKING
+
+import regex as re
+
+from vllm.entrypoints.openai.engine.protocol import DeltaFunctionCall, DeltaToolCall
+from vllm.parser.engine.events import EventType
+from vllm.parser.engine.parser_engine import ParserEngine
+from vllm.parser.engine.parser_engine_config import (
+    ParserEngineConfig,
+    ParserState,
+    Transition,
+)
+
+if TYPE_CHECKING:
+    from vllm.entrypoints.openai.chat_completion.protocol import (
+        ChatCompletionRequest,
+    )
+    from vllm.entrypoints.openai.responses.protocol import ResponsesRequest
+    from vllm.tokenizers import TokenizerLike
+    from vllm.tool_parsers.abstract_tool_parser import Tool
+
+THINK_START = "<think>"
+THINK_END = "</think>"
+TOOL_SECTION_START = "<|tool_calls_section_begin|>"
+TOOL_SECTION_END = "<|tool_calls_section_end|>"
+TOOL_CALL_START = "<|tool_call_begin|>"
+TOOL_CALL_END = "<|tool_call_end|>"
+TOOL_ARG_START = "<|tool_call_argument_begin|>"
+
+_TOOL_ID_RE = re.compile(r"(?P<id>.+:\d+)")
+
+
+@functools.cache
+def kimi_k2_config(thinking: bool = True) -> ParserEngineConfig:
+    reasoning_terminals = (
+        {
+            "THINK_START": THINK_START,
+            "THINK_END": THINK_END,
+        }
+        if thinking
+        else {}
+    )
+    reasoning_transitions = (
+        {
+            (ParserState.REASONING, "THINK_START"): Transition(
+                ParserState.REASONING,
+                (),
+            ),
+            (ParserState.REASONING, "THINK_END"): Transition(
+                ParserState.CONTENT,
+                (EventType.REASONING_END,),
+            ),
+            (ParserState.CONTENT, "THINK_END"): Transition(
+                ParserState.CONTENT,
+                (),
+            ),
+        }
+        if thinking
+        else {}
+    )
+
+    return ParserEngineConfig(
+        name="kimi_k2",
+        initial_state=ParserState.REASONING if thinking else ParserState.CONTENT,
+        terminals={
+            **reasoning_terminals,
+            "TOOL_SECTION_START": TOOL_SECTION_START,
+            "TOOL_SECTION_END": TOOL_SECTION_END,
+            "TOOL_START": TOOL_CALL_START,
+            "TOOL_END": TOOL_CALL_END,
+            "ARG_START": TOOL_ARG_START,
+        },
+        token_id_terminals={
+            **reasoning_terminals,
+            "TOOL_SECTION_START": TOOL_SECTION_START,
+            "TOOL_SECTION_END": TOOL_SECTION_END,
+            "TOOL_START": TOOL_CALL_START,
+            "TOOL_END": TOOL_CALL_END,
+            "ARG_START": TOOL_ARG_START,
+        },
+        transitions={
+            **reasoning_transitions,
+            (ParserState.REASONING, "TOOL_SECTION_START"): Transition(
+                ParserState.TOOL_PREAMBLE,
+                (EventType.REASONING_END,),
+            ),
+            (ParserState.CONTENT, "TOOL_SECTION_START"): Transition(
+                ParserState.TOOL_PREAMBLE,
+                (),
+            ),
+            (ParserState.TOOL_PREAMBLE, "TOOL_START"): Transition(
+                ParserState.TOOL_NAME,
+                (EventType.TOOL_CALL_START,),
+            ),
+            (ParserState.TOOL_NAME, "ARG_START"): Transition(
+                ParserState.TOOL_ARGS,
+                (),
+            ),
+            (ParserState.TOOL_ARGS, "TOOL_END"): Transition(
+                ParserState.TOOL_BETWEEN,
+                (EventType.TOOL_CALL_END,),
+            ),
+            (ParserState.TOOL_ARGS, "TOOL_SECTION_END"): Transition(
+                ParserState.TOOL_PREAMBLE,
+                (EventType.TOOL_CALL_END,),
+            ),
+            (ParserState.TOOL_BETWEEN, "TOOL_START"): Transition(
+                ParserState.TOOL_NAME,
+                (EventType.TOOL_CALL_START,),
+            ),
+            # Keep the parser in a tool state after the section closes so
+            # trailing model text after native tool calls is suppressed.
+            (ParserState.TOOL_PREAMBLE, "TOOL_SECTION_END"): Transition(
+                ParserState.TOOL_PREAMBLE,
+                (),
+            ),
+            (ParserState.TOOL_BETWEEN, "TOOL_SECTION_END"): Transition(
+                ParserState.TOOL_PREAMBLE,
+                (),
+            ),
+        },
+        stream_arg_deltas=True,
+        tool_args_json=True,
+        strip_trailing_reasoning_whitespace=True,
+        drop_whitespace_only_content_before_tools=True,
+        strip_content_whitespace_with_tools=False,
+        validate_tool_names=False,
+    )
+
+
+class KimiK2Parser(ParserEngine):
+    """Kimi K2 parser backed by the declarative parser engine."""
+
+    def __init__(
+        self,
+        tokenizer: TokenizerLike,
+        tools: list[Tool] | None = None,
+        **kwargs,
+    ) -> None:
+        chat_kwargs = kwargs.get("chat_template_kwargs", {}) or {}
+        thinking = chat_kwargs.get("thinking", None)
+        enable_thinking = chat_kwargs.get("enable_thinking", None)
+        self.thinking_enabled = (
+            True
+            if thinking is None and enable_thinking is None
+            else bool(thinking) or bool(enable_thinking)
+        )
+        kwargs.setdefault(
+            "parser_engine_config",
+            kimi_k2_config(thinking=self.thinking_enabled),
+        )
+        super().__init__(tokenizer, tools, **kwargs)
+
+        vocab = self.vocab
+        self._start_token_id = vocab.get(THINK_START)
+        self._end_token_id = vocab.get(THINK_END)
+        self._tool_section_start_token_id = vocab.get(TOOL_SECTION_START)
+
+    @staticmethod
+    def _extract_tool_id_and_name(header: str | None) -> tuple[str | None, str | None]:
+        if header is None:
+            return None, None
+        match = _TOOL_ID_RE.match(header.strip())
+        if not match:
+            return None, None
+
+        tool_id = match.group("id").strip()
+        tool_name = tool_id.split(":")[0].removeprefix("functions.")
+        return tool_id, tool_name
+
+    def _emit_name_delta(
+        self,
+        idx: int,
+        deltas: list[DeltaToolCall],
+        name: str | None,
+    ) -> None:
+        tool_id, tool_name = self._extract_tool_id_and_name(name)
+        if not tool_name:
+            if 0 <= idx < len(self._tool_slots):
+                self._tool_slots[idx].name = ""
+            return
+
+        slot = self._tool_slots[idx]
+        slot.id = tool_id or ""
+        super()._emit_name_delta(idx, deltas, tool_name)
+
+    def _handle_tool_end(self, event, deltas) -> None:
+        idx = event.tool_index
+        if 0 <= idx < len(self._tool_slots) and not self._tool_slots[idx].name_sent:
+            tool_id, tool_name = self._extract_tool_id_and_name(
+                self._tool_slots[idx].name
+            )
+            if tool_name:
+                self._tool_slots[idx].id = tool_id or ""
+                self._tool_slots[idx].name = tool_name
+        super()._handle_tool_end(event, deltas)
+
+    def _handle_arg_chunk(self, event, deltas) -> None:
+        idx = event.tool_index
+        name_sent_before = (
+            0 <= idx < len(self._tool_slots) and self._tool_slots[idx].name_sent
+        )
+        super()._handle_arg_chunk(event, deltas)
+        if (
+            event.value
+            and not name_sent_before
+            and 0 <= idx < len(self._tool_slots)
+            and self._tool_slots[idx].name_sent
+        ):
+            deltas.append(
+                DeltaToolCall(
+                    index=idx,
+                    function=DeltaFunctionCall(arguments=event.value),
+                )
+            )
+
+    def _extract_args_json(self, raw_args: str, func_name: str) -> str:
+        return raw_args.strip() or "{}"
+
+    def is_reasoning_end(self, input_ids: list[int]) -> bool:
+        if not self.thinking_enabled:
+            return True
+
+        start_id = self._start_token_id
+        end_id = self._end_token_id
+        tool_section_id = self._tool_section_start_token_id
+
+        for i in range(len(input_ids) - 1, -1, -1):
+            token_id = input_ids[i]
+            if start_id is not None and token_id == start_id:
+                return False
+            if end_id is not None and token_id == end_id:
+                return True
+            if tool_section_id is not None and token_id == tool_section_id:
+                return True
+        return False
+
+    def extract_content_ids(self, input_ids: list[int]) -> list[int]:
+        if not self.thinking_enabled:
+            return input_ids
+
+        end_id = self._end_token_id
+        if end_id is not None and end_id in input_ids:
+            end_idx = len(input_ids) - 1 - input_ids[::-1].index(end_id)
+            return input_ids[end_idx + 1 :]
+
+        tool_section_id = self._tool_section_start_token_id
+        if tool_section_id is not None and tool_section_id in input_ids:
+            section_idx = len(input_ids) - 1 - input_ids[::-1].index(tool_section_id)
+            return input_ids[section_idx:]
+
+        return []
+
+    def extract_reasoning(
+        self,
+        model_output: str,
+        request: ChatCompletionRequest | ResponsesRequest,
+    ) -> tuple[str | None, str | None]:
+        if not self.thinking_enabled:
+            return None, model_output
+        return super().extract_reasoning(model_output, request)
+
+    def count_reasoning_tokens(self, token_ids: Sequence[int]) -> int:
+        if not self.thinking_enabled:
+            return 0
+        return super().count_reasoning_tokens(token_ids)
@@ -92,11 +92,6 @@ class CpuPlatform(Platform):

        return meminfo.total_memory

-    @classmethod
-    def mem_get_info(cls) -> tuple[int, int]:
-        meminfo = get_memory_node_info()
-        return meminfo.available_memory, meminfo.total_memory
-
    @classmethod
    def set_device(cls, device: torch.device) -> None:
        """
@@ -243,6 +243,16 @@ class XPUPlatform(Platform):
        if "VLLM_WORKER_MULTIPROC_METHOD" not in os.environ:
            os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn"

+        # XPU requires graceful shutdown to allow oneCCL/Level Zero resources
+        # to be properly released. Without this, subsequent server startups on
+        # the same devices may hang during CCL initialization.
+        if vllm_config.shutdown_timeout == 0:
+            vllm_config.shutdown_timeout = 5
+            logger.info(
+                "XPU platform: set server shutdown_timeout=%d.",
+                vllm_config.shutdown_timeout,
+            )
+
    @classmethod
    def update_block_size_for_backend(cls, vllm_config: "VllmConfig") -> None:
        super().update_block_size_for_backend(vllm_config)
@@ -1,245 +1,8 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project

-from collections.abc import Iterable, Sequence
-from typing import TYPE_CHECKING
+from vllm.parser.engine.registered_adapters import KimiK2ParserReasoningAdapter

-from transformers import PreTrainedTokenizerBase
+KimiK2ReasoningParser = KimiK2ParserReasoningAdapter

-from vllm.entrypoints.openai.engine.protocol import DeltaMessage
-from vllm.reasoning.abs_reasoning_parsers import ReasoningParser
-from vllm.reasoning.identity_reasoning_parser import IdentityReasoningParser
-
-if TYPE_CHECKING:
-    from vllm.entrypoints.openai.chat_completion.protocol import ChatCompletionRequest
-    from vllm.entrypoints.openai.responses.protocol import ResponsesRequest
-
-
-class KimiK2ReasoningParser(ReasoningParser):
-    """
-    Reasoning parser for Kimi K2 model.
-
-    The Kimi K2 model uses <think>...</think> tokens to denote reasoning text,
-    and may implicitly end reasoning by starting a tool call section using
-    <|tool_calls_section_begin|>.
-    Thinking may also begin without a </think> token.
-
-    Kimi's thinking mode can be disabled via chat_template_kwargs.
-    """
-
-    def __init__(self, tokenizer: PreTrainedTokenizerBase, *args, **kwargs):
-        super().__init__(tokenizer, *args, **kwargs)
-
-        if not self.model_tokenizer:
-            raise ValueError(
-                "The model tokenizer must be passed to the ReasoningParser "
-                "constructor during construction."
-            )
-
-        # Check if thinking is disabled via chat_template_kwargs
-        chat_kwargs = kwargs.get("chat_template_kwargs", {}) or {}
-        thinking = bool(chat_kwargs.get("thinking", True))
-
-        # If thinking is not enabled, use identity parser to fall through
-        self._identity_parser: IdentityReasoningParser | None
-        if not thinking:
-            self._identity_parser = IdentityReasoningParser(tokenizer, *args, **kwargs)
-        else:
-            self._identity_parser = None
-
-        # Token definitions
-        self._start_token = "<think>"
-        self._end_token = "</think>"
-        self._tool_section_start_token = "<|tool_calls_section_begin|>"
-
-        # Get token IDs
-        self._start_token_id = self.vocab.get(self._start_token)
-        self._end_token_id = self.vocab.get(self._end_token)
-        self._tool_section_start_token_id = self.vocab.get(
-            self._tool_section_start_token
-        )
-
-        if self._start_token_id is None or self._end_token_id is None:
-            raise RuntimeError(
-                "KimiK2ReasoningParser could not locate think start/end "
-                "tokens in the tokenizer!"
-            )
-
-    @property
-    def reasoning_start_str(self) -> str | None:
-        return self._start_token
-
-    @property
-    def reasoning_end_str(self) -> str | None:
-        return self._end_token
-
-    def is_reasoning_end(self, input_ids: Sequence[int]) -> bool:
-        """
-        Check if the reasoning content ends in the input_ids.
-
-        Reasoning ends when we see either:
-        1. The end token (</think>)
-        2. The tool section start token (<|tool_calls_section_begin|>)
-        """
-        if self._identity_parser is not None:
-            return self._identity_parser.is_reasoning_end(input_ids)
-
-        start_token_id = self._start_token_id
-        end_token_id = self._end_token_id
-        tool_section_start_token_id = self._tool_section_start_token_id
-
-        for i in range(len(input_ids) - 1, -1, -1):
-            if input_ids[i] == start_token_id:
-                return False
-            if input_ids[i] == end_token_id:
-                return True
-            # Implicit reasoning end via tool call section
-            if (
-                tool_section_start_token_id is not None
-                and input_ids[i] == tool_section_start_token_id
-            ):
-                return True
-        return False
-
-    def is_reasoning_end_streaming(
-        self, input_ids: Sequence[int], delta_ids: Iterable[int]
-    ) -> bool:
-        """
-        Check if the reasoning content ends in the input_ids on a decode step.
-        """
-        if self._identity_parser is not None:
-            return self._identity_parser.is_reasoning_end_streaming(
-                input_ids, delta_ids
-            )
-
-        # Materialize iterable for membership checks
-        delta_ids_set = set(delta_ids)
-
-        # Check for explicit end token or implicit tool section start in delta
-        if self._end_token_id in delta_ids_set:
-            return True
-        return (
-            self._tool_section_start_token_id is not None
-            and self._tool_section_start_token_id in delta_ids_set
-        )
-
-    def extract_content_ids(self, input_ids: list[int]) -> list[int]:
-        """
-        Extract content token ids from the input_ids.
-        """
-        if self._identity_parser is not None:
-            return self._identity_parser.extract_content_ids(input_ids)
-
-        if self._end_token_id in input_ids:
-            end_token_index = (
-                len(input_ids) - 1 - input_ids[::-1].index(self._end_token_id)
-            )
-
-            if end_token_index != -1:
-                return input_ids[end_token_index + 1 :]
-
-        if (
-            self._tool_section_start_token_id is not None
-            and self._tool_section_start_token_id in input_ids
-        ):
-            tool_section_index = (
-                len(input_ids)
-                - 1
-                - input_ids[::-1].index(self._tool_section_start_token_id)
-            )
-
-            if tool_section_index != -1:
-                return input_ids[tool_section_index:]
-
-        # still reasoning (no content)
-        return []
-
-    def extract_reasoning(
-        self, model_output: str, request: "ChatCompletionRequest | ResponsesRequest"
-    ) -> tuple[str | None, str | None]:
-        """
-        Extract reasoning content from the model output.
-        """
-        if self._identity_parser is not None:
-            return self._identity_parser.extract_reasoning(model_output, request)
-
-        # thinking does not require a think start token but consume it if present
-        start_token_index = model_output.find(self._start_token)
-        start_token_index = 0 if start_token_index != 0 else len(self._start_token)
-        end_token_index = model_output.find(self._end_token)
-
-        if end_token_index != -1:
-            return (
-                model_output[start_token_index:end_token_index],
-                model_output[end_token_index + len(self._end_token) :] or None,
-            )
-
-        tool_section_index = model_output.find(self._tool_section_start_token)
-        if tool_section_index != -1:
-            return (
-                model_output[start_token_index:tool_section_index],
-                model_output[tool_section_index:] or None,
-            )
-
-        # still reasoning (no content)
-        return (
-            model_output[start_token_index:],
-            None,
-        )
-
-    def extract_reasoning_streaming(
-        self,
-        previous_text: str,
-        current_text: str,
-        delta_text: str,
-        previous_token_ids: Sequence[int],
-        current_token_ids: Sequence[int],
-        delta_token_ids: Sequence[int],
-    ) -> DeltaMessage | None:
-        """
-        Extract reasoning content from a delta message during streaming.
-        """
-        if self._identity_parser is not None:
-            return self._identity_parser.extract_reasoning_streaming(
-                previous_text,
-                current_text,
-                delta_text,
-                previous_token_ids,
-                current_token_ids,
-                delta_token_ids,
-            )
-
-        # If reasoning has already ended in previous tokens, this is content
-        if self.is_reasoning_end(previous_token_ids):
-            return DeltaMessage(content=delta_text)
-
-        # Skip single special tokens
-        if len(delta_token_ids) == 1 and delta_token_ids[0] in [
-            self._start_token_id,
-            self._end_token_id,
-        ]:
-            return None
-
-        if self._end_token_id in delta_token_ids:
-            if self._end_token not in delta_text:
-                # Token ID arrived before text was flushed (stop-sequence buffering).
-                # Wait for the next delta when the text becomes visible.
-                return None
-            end_index = delta_text.find(self._end_token)
-            reasoning = delta_text[:end_index]
-            content = delta_text[end_index + len(self._end_token) :]
-            return DeltaMessage(
-                reasoning=reasoning, content=content if content else None
-            )
-
-        if self._tool_section_start_token_id in delta_token_ids:
-            if self._tool_section_start_token not in delta_text:
-                # Token ID arrived before text was flushed (stop-sequence buffering).
-                return None
-            tool_index = delta_text.find(self._tool_section_start_token)
-            reasoning = delta_text[:tool_index]
-            content = delta_text[tool_index:]
-            return DeltaMessage(reasoning=reasoning, content=content)
-
-        # still reasoning (no end token)
-        return DeltaMessage(reasoning=delta_text)
+__all__ = ["KimiK2ReasoningParser"]
@@ -1,278 +1,20 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project

-from collections.abc import Sequence
-
-import regex as re
-
 from vllm.entrypoints.openai.chat_completion.protocol import (
    ChatCompletionRequest,
 )
-from vllm.entrypoints.openai.engine.protocol import (
-    DeltaFunctionCall,
-    DeltaMessage,
-    DeltaToolCall,
-    ExtractedToolCallInformation,
-    FunctionCall,
-    ToolCall,
-)
 from vllm.entrypoints.openai.responses.protocol import ResponsesRequest
-from vllm.logger import init_logger
-from vllm.tokenizers import TokenizerLike
-from vllm.tool_parsers.abstract_tool_parser import (
-    Tool,
-    ToolParser,
-)
-from vllm.tool_parsers.utils import partial_tag_overlap
-
-logger = init_logger(__name__)
+from vllm.parser.engine.registered_adapters import KimiK2ParserToolAdapter


-class KimiK2ToolParser(ToolParser):
+class KimiK2ToolParser(KimiK2ParserToolAdapter):  # type: ignore[valid-type, misc]
    structural_tag_model = "kimi"

-    def __init__(self, tokenizer: TokenizerLike, tools: list[Tool] | None = None):
-        super().__init__(tokenizer, tools)
-
-        # Streaming state
-        self._sent_content_idx: int = 0
-        self.prev_tool_call_arr: list[dict] = []
-        self.streamed_args_for_tool: list[str] = []
-
-        # Section marker
-        self.tool_calls_start_token: str = "<|tool_calls_section_begin|>"
-
-        # Individual tool call markers
-        self.tool_call_start_token: str = "<|tool_call_begin|>"
-        self.tool_call_end_token: str = "<|tool_call_end|>"
-        self.tool_call_arg_token: str = "<|tool_call_argument_begin|>"
-
-        # Regex for non-streaming extraction
-        self.tool_call_regex = re.compile(
-            r"<\|tool_call_begin\|>\s*(?P<tool_call_id>[^<]+:\d+)\s*"
-            r"<\|tool_call_argument_begin\|>\s*"
-            r"(?P<function_arguments>(?:(?!<\|tool_call_begin\|>).)*?)\s*"
-            r"<\|tool_call_end\|>",
-            re.DOTALL,
-        )
-
-        if not self.model_tokenizer:
-            raise ValueError(
-                "The model tokenizer must be passed to the ToolParser "
-                "constructor during construction."
-            )
-
    def adjust_request(
-        self, request: ChatCompletionRequest | ResponsesRequest
+        self,
+        request: ChatCompletionRequest | ResponsesRequest,
    ) -> ChatCompletionRequest | ResponsesRequest:
-        request = super().adjust_request(request)
        if request.tools and request.tool_choice != "none":
-            # Ensure special-token markers appear as literal text in
-            # current_text so we can do pure text-based parsing.
            request.skip_special_tokens = False
        return request
-
-    def extract_tool_calls(
-        self,
-        model_output: str,
-        request: ChatCompletionRequest,
-    ) -> ExtractedToolCallInformation:
-        # sanity check; avoid unnecessary processing
-        if self.tool_calls_start_token not in model_output:
-            return ExtractedToolCallInformation(
-                tools_called=False, tool_calls=[], content=model_output
-            )
-
-        else:
-            try:
-                # there are two possible captures - between tags, or between a
-                # tag and end-of-string so the result of
-                # findall is an array of tuples where one is a function call and
-                # the other is None
-                function_call_tuples = self.tool_call_regex.findall(model_output)
-
-                logger.debug("function_call_tuples: %s", function_call_tuples)
-
-                tool_calls = []
-                for match in function_call_tuples:
-                    function_id, function_args = match
-                    # function_id: functions.get_weather:0 or get_weather:0
-                    function_name = function_id.split(":")[0].split(".")[-1]
-                    tool_calls.append(
-                        ToolCall(
-                            id=function_id,
-                            type="function",
-                            function=FunctionCall(
-                                name=function_name, arguments=function_args
-                            ),
-                        )
-                    )
-
-                content = model_output[: model_output.find(self.tool_calls_start_token)]
-                return ExtractedToolCallInformation(
-                    tools_called=True,
-                    tool_calls=tool_calls,
-                    content=content if content else None,
-                )
-
-            except Exception:
-                logger.exception("Error in extracting tool call from response.")
-                return ExtractedToolCallInformation(
-                    tools_called=False, tool_calls=[], content=model_output
-                )
-
-    def _extract_content(self, current_text: str) -> str | None:
-        """Return unsent content before the tool-calls section, or None.
-
-        Holds back any trailing suffix that partially matches
-        ``<|tool_calls_section_begin|>`` to avoid leaking marker bytes.
-        """
-        if self.tool_calls_start_token not in current_text:
-            overlap = partial_tag_overlap(current_text, self.tool_calls_start_token)
-            sendable_idx = len(current_text) - overlap
-        else:
-            sendable_idx = current_text.index(self.tool_calls_start_token)
-
-        if sendable_idx > self._sent_content_idx:
-            content = current_text[self._sent_content_idx : sendable_idx]
-            self._sent_content_idx = sendable_idx
-            return content
-        return None
-
-    def _extract_tool_calls(self, current_text: str) -> list[str]:
-        """Extract raw bodies from ``<|tool_call_begin|>…<|tool_call_end|>`` blocks."""
-        if self.tool_calls_start_token not in current_text:
-            return []
-
-        results: list[str] = []
-        pos = current_text.index(self.tool_calls_start_token)
-        while True:
-            start = current_text.find(self.tool_call_start_token, pos)
-            if start == -1:
-                break
-            tc_start = start + len(self.tool_call_start_token)
-            end = current_text.find(self.tool_call_end_token, tc_start)
-
-            if end != -1:
-                tool_call = current_text[tc_start:end]
-                pos = end + len(self.tool_call_end_token)
-            else:
-                tool_call = current_text[tc_start:]
-                overlap = partial_tag_overlap(tool_call, self.tool_call_end_token)
-                if overlap:
-                    tool_call = tool_call[:-overlap]
-
-            results.append(tool_call)
-
-            if end == -1:
-                break
-        return results
-
-    @staticmethod
-    def _extract_tool_id_and_name(
-        header: str | None,
-    ) -> tuple[str | None, str | None]:
-        """Parse ``(tool_id, tool_name)`` from a header
-        like ``"functions.get_weather:0"``."""
-        if header is None:
-            return None, None
-        match = re.match(r"(.+:\d+)", header)
-        if not match:
-            return None, None
-
-        tool_id = match.group(1).strip()
-        tool_name = tool_id.split(":")[0].split(".")[-1]
-        return tool_id, tool_name
-
-    def _split_tool_call(self, tool_call: str) -> tuple[str | None, str | None]:
-        """Split a tool-call body into ``(header, arguments)`` at the argument marker.
-
-        Example::
-            'get_weather:0 <|tool_call_argument_begin|>{"c'
-            -> ("get_weather:0", '{"c')
-        """
-        arg_pos = tool_call.find(self.tool_call_arg_token)
-        if arg_pos == -1:
-            return None, None
-        header = tool_call[:arg_pos].strip()
-        tool_args = tool_call[arg_pos + len(self.tool_call_arg_token) :]
-        return header, tool_args
-
-    def _compute_args_diff(self, index: int, tool_args: str | None) -> str | None:
-        """Return new argument text not yet sent for tool `index`, or None."""
-        if tool_args is None:
-            return None
-        prev = self.streamed_args_for_tool[index]
-        if len(tool_args) <= len(prev):
-            return None
-        diff = tool_args[len(prev) :]
-        self.streamed_args_for_tool[index] = tool_args
-        self.prev_tool_call_arr[index]["arguments"] = tool_args
-        return diff
-
-    def extract_tool_calls_streaming(
-        self,
-        previous_text: str,
-        current_text: str,
-        delta_text: str,
-        previous_token_ids: Sequence[int],
-        current_token_ids: Sequence[int],
-        delta_token_ids: Sequence[int],
-        request: ChatCompletionRequest,
-    ) -> DeltaMessage | None:
-        try:
-            # Extract any content before tool calls.
-            content = self._extract_content(current_text)
-            tool_calls = self._extract_tool_calls(current_text)
-            tool_call_deltas: list[DeltaToolCall] = []
-
-            for i, tool_call in enumerate(tool_calls):
-                # First time seeing tool call at index i.
-                if i >= len(self.prev_tool_call_arr):
-                    # Initialize streaming state.
-                    self.prev_tool_call_arr.append({})
-                    self.streamed_args_for_tool.append("")
-
-                header, tool_args = self._split_tool_call(tool_call)
-
-                # Stream back tool name.
-                if "name" not in self.prev_tool_call_arr[i]:
-                    tool_id, tool_name = self._extract_tool_id_and_name(header)
-                    if not tool_name:
-                        # Can't skip to tool i+1 if i isn't ready
-                        break
-                    self.prev_tool_call_arr[i]["name"] = tool_name
-                    self.prev_tool_call_arr[i]["id"] = tool_id
-                    tool_call_deltas.append(
-                        DeltaToolCall(
-                            index=i,
-                            type="function",
-                            id=tool_id,
-                            function=DeltaFunctionCall(name=tool_name).model_dump(
-                                exclude_none=True
-                            ),
-                        )
-                    )
-
-                # Stream back new tool args by diffing against what was sent.
-                args_diff = self._compute_args_diff(i, tool_args)
-                if args_diff:
-                    tool_call_deltas.append(
-                        DeltaToolCall(
-                            index=i,
-                            function=DeltaFunctionCall(arguments=args_diff).model_dump(
-                                exclude_none=True
-                            ),
-                        )
-                    )
-
-            if content or tool_call_deltas:
-                return DeltaMessage(
-                    content=content,
-                    tool_calls=tool_call_deltas,
-                )
-            return None
-
-        except Exception:
-            logger.exception("Error trying to handle streaming tool call.")
-            return None
@@ -44,12 +44,6 @@ AUDIO_CONTEXT = "<so_embedding>"
 # MAX_FRAMES = 16
 DEFAULT_NUM_TILES = 12

-# Configure PIL to handle large images without warnings
-# This prevents DecompressionBombWarning for legitimate large images
-Image.MAX_IMAGE_PIXELS = None  # Disable the limit entirely
-# Alternative: Set a specific higher limit
-# Image.MAX_IMAGE_PIXELS = 300000000  # ~300M pixels
-

 def calculate_timestamps(
    indices: list[int] | torch.Tensor,
@@ -10,12 +10,6 @@ from vllm.tokenizers.hf import HfTokenizer

 from .internvl import InternVLImageProcessor, InternVLProcessor

-# Configure PIL to handle large images without warnings
-# This prevents DecompressionBombWarning for legitimate large images
-Image.MAX_IMAGE_PIXELS = None  # Disable the limit entirely
-# Alternative: Set a specific higher limit
-# Image.MAX_IMAGE_PIXELS = 300000000  # ~300M pixels
-

 def build_transform(input_size: int):
    return T.Compose(
@@ -143,7 +143,7 @@ class MemorySnapshot:
            "allocated_bytes.all.peak", 0
        )

-        self.free_memory, self.total_memory = current_platform.mem_get_info(device)
+        self.free_memory, self.total_memory = torch.accelerator.get_memory_info(device)
        if current_platform.is_integrated_gpu(device.index):
            # On UMA (Unified Memory Architecture) platforms where CPU and
            # GPU share physical memory (e.g. GH200, DGX Spark, Jetson Orin),
@@ -0,0 +1,469 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""HPC Attention Backend.
+
+Pure attention (prefill + decode), without RoPE or RMSNorm.
+Independent metadata / builder; KV cache layout is NHD:
+(num_blocks, 2, block_size, num_kv_heads, head_size).
+"""
+
+import importlib.util
+from dataclasses import dataclass
+from typing import ClassVar
+
+import torch
+from typing_extensions import override
+
+from vllm.config import VllmConfig
+from vllm.config.cache import CacheDType
+from vllm.logger import init_logger
+from vllm.platforms import current_platform
+from vllm.platforms.interface import DeviceCapability
+from vllm.v1.attention.backend import (
+    AttentionBackend,
+    AttentionCGSupport,
+    AttentionImpl,
+    AttentionMetadata,
+    AttentionMetadataBuilder,
+    AttentionType,
+    CommonAttentionMetadata,
+    MultipleOf,
+)
+from vllm.v1.attention.backends.utils import (
+    KVCacheLayoutType,
+    get_per_layer_parameters,
+    infer_global_hyperparameters,
+    split_decodes_and_prefills,
+)
+from vllm.v1.kv_cache_interface import AttentionSpec
+
+logger = init_logger(__name__)
+
+FP8_DTYPE = current_platform.fp8_dtype()
+
+
+def _get_fp8_dtype_for_kv_cache(kv_cache_dtype: str) -> torch.dtype:
+    """Return the torch FP8 dtype for the given kv_cache_dtype string."""
+    if kv_cache_dtype in ("fp8", "fp8_e4m3"):
+        return torch.float8_e4m3fn
+    elif kv_cache_dtype == "fp8_e5m2":
+        return torch.float8_e5m2
+    else:
+        raise ValueError(f"Unrecognized FP8 dtype: {kv_cache_dtype}")
+
+
+@dataclass
+class HpcAttnMetadata(AttentionMetadata):
+    """Metadata required by the HPC attention kernel."""
+
+    num_actual_tokens: int
+    num_decodes: int
+    num_decode_tokens: int
+    num_prefills: int
+    num_prefill_tokens: int
+    max_query_len: int
+
+    slot_mapping: torch.Tensor
+    """Slot mapping for KV cache writes. shape = [num_actual_tokens]"""
+
+    seq_lens: torch.Tensor
+    """KV cache length per request. shape = [batch_size]"""
+
+    block_table_tensor: torch.Tensor
+    """Paged KV-cache block table.
+    shape = [batch_size, max_num_blocks_per_seq]"""
+
+    qo_indptr: torch.Tensor | None = None
+    """Cumulative query lengths for prefill requests (GPU tensor).
+    shape = [num_prefills + 1]. None when num_prefills == 0."""
+
+    # --- HPC RopeNorm pass-through fields ---
+    # Set by HpcRopeNorm._forward_impl(); consumed & reset by
+    # HpcAttentionImpl.forward().  Defaults are safe for the standard
+    # (non-RopeNorm) path and for profiling runs (attn_metadata=None).
+    hpc_kv_written: bool = False
+    """True when HpcRopeNorm already wrote KV cache."""
+    hpc_prefill_q_scale: torch.Tensor | None = None
+    """FP8 per-token-per-head Q scale for prefill (from RopeNorm)."""
+    hpc_decode_q_scale: torch.Tensor | None = None
+    """FP8 per-token-per-head Q scale for decode (from RopeNorm)."""
+    hpc_split_k_flag: torch.Tensor | None = None
+    """Split-K flag tensor for FP8 decode (from RopeNorm)."""
+
+
+class HpcAttnMetadataBuilder(AttentionMetadataBuilder[HpcAttnMetadata]):
+    """Build HpcAttnMetadata from CommonAttentionMetadata."""
+
+    _cudagraph_support = AttentionCGSupport.UNIFORM_SINGLE_TOKEN_DECODE
+    reorder_batch_threshold: int = 1
+
+    def __init__(
+        self,
+        kv_cache_spec: AttentionSpec,
+        layer_names: list[str],
+        vllm_config: VllmConfig,
+        device: torch.device,
+    ):
+        super().__init__(kv_cache_spec, layer_names, vllm_config, device)
+        self.model_config = vllm_config.model_config
+        self.cache_config = vllm_config.cache_config
+
+        self.num_qo_heads = self.model_config.get_num_attention_heads(
+            vllm_config.parallel_config
+        )
+        self.num_kv_heads = kv_cache_spec.num_kv_heads
+        self.head_dim = kv_cache_spec.head_size
+        self.page_size = kv_cache_spec.block_size
+
+        self.cache_dtype = self.cache_config.cache_dtype
+
+        self.global_hyperparameters = infer_global_hyperparameters(
+            get_per_layer_parameters(vllm_config, layer_names, HpcAttentionImpl)
+        )
+
+    @override  # type: ignore[misc]
+    @classmethod
+    def get_cudagraph_support(
+        cls: type["HpcAttnMetadataBuilder"],
+        vllm_config: VllmConfig,
+        kv_cache_spec: AttentionSpec,
+    ) -> AttentionCGSupport:
+        return AttentionCGSupport.UNIFORM_SINGLE_TOKEN_DECODE
+
+    def build(
+        self,
+        common_prefix_len: int,
+        common_attn_metadata: CommonAttentionMetadata,
+        fast_build: bool = False,
+    ) -> HpcAttnMetadata:
+        """Build HpcAttnMetadata from CommonAttentionMetadata."""
+        num_actual_tokens = common_attn_metadata.num_actual_tokens
+
+        num_decodes, num_prefills, num_decode_tokens, num_prefill_tokens = (
+            split_decodes_and_prefills(
+                common_attn_metadata,
+                decode_threshold=self.reorder_batch_threshold,
+                require_uniform=False,
+            )
+        )
+
+        seq_lens = common_attn_metadata.seq_lens
+        block_table_tensor = common_attn_metadata.block_table_tensor
+        slot_mapping = common_attn_metadata.slot_mapping
+        max_query_len = common_attn_metadata.max_query_len
+
+        qo_indptr = None
+        if num_prefills > 0:
+            qo_indptr_cpu = common_attn_metadata.query_start_loc_cpu
+            prefill_start = num_decodes
+            qo_indptr_prefill_cpu = (
+                qo_indptr_cpu[prefill_start:] - qo_indptr_cpu[prefill_start]
+            )
+            qo_indptr = qo_indptr_prefill_cpu.to(self.device, non_blocking=True)
+
+        return HpcAttnMetadata(
+            num_actual_tokens=num_actual_tokens,
+            num_decodes=num_decodes,
+            num_decode_tokens=num_decode_tokens,
+            num_prefills=num_prefills,
+            num_prefill_tokens=num_prefill_tokens,
+            max_query_len=max_query_len,
+            slot_mapping=slot_mapping,
+            seq_lens=seq_lens,
+            block_table_tensor=block_table_tensor,
+            qo_indptr=qo_indptr,
+            hpc_kv_written=True,
+            hpc_prefill_q_scale=None,
+            hpc_decode_q_scale=None,
+            hpc_split_k_flag=None,
+        )
+
+
+class HpcAttentionBackend(AttentionBackend):
+    """HPC attention backend (pure attention, no RoPE/Norm).
+
+    KV cache layout: NHD (num_blocks, 2, block_size, num_kv_heads, head_size).
+    """
+
+    accept_output_buffer: bool = True
+    supported_dtypes: ClassVar[list[torch.dtype]] = [
+        torch.float16,
+        torch.bfloat16,
+    ]
+    supported_kv_cache_dtypes: ClassVar[list[CacheDType]] = [
+        "auto",
+        "fp8_e4m3",
+    ]
+
+    # Avoid attention abstracted method call cache insert
+    forward_includes_kv_cache_update: bool = True
+
+    @staticmethod
+    def get_supported_kernel_block_sizes() -> list[int | MultipleOf]:
+        return [64]
+
+    @staticmethod
+    def get_name() -> str:
+        return "HPC_ATTN"
+
+    @staticmethod
+    def get_impl_cls() -> type["HpcAttentionImpl"]:
+        return HpcAttentionImpl
+
+    @staticmethod
+    def get_builder_cls() -> type["HpcAttnMetadataBuilder"]:
+        return HpcAttnMetadataBuilder
+
+    @staticmethod
+    def get_kv_cache_shape(
+        num_blocks: int,
+        block_size: int,
+        num_kv_heads: int,
+        head_size: int,
+        cache_dtype_str: str = "auto",
+    ) -> tuple[int, ...]:
+        return (num_blocks, 2, block_size, num_kv_heads, head_size)
+
+    @staticmethod
+    def get_kv_cache_stride_order(
+        include_num_layers_dimension: bool = False,
+    ) -> tuple[int, ...]:
+        if include_num_layers_dimension:
+            return (1, 0, 2, 3, 4, 5)
+        return (0, 1, 2, 3, 4)
+
+    @classmethod
+    def get_supported_head_sizes(cls) -> list[int]:
+        return [128]
+
+    @classmethod
+    def supports_compute_capability(cls, capability: DeviceCapability) -> bool:
+        return capability >= DeviceCapability(9, 0)
+
+    @classmethod
+    def supports_kv_cache_dtype(cls, kv_cache_dtype: "CacheDType | None") -> bool:
+        if kv_cache_dtype is None:
+            return True
+        return kv_cache_dtype in cls.supported_kv_cache_dtypes
+
+    @classmethod
+    def get_required_kv_cache_layout(cls) -> KVCacheLayoutType | None:
+        return "NHD"
+
+
+class HpcAttentionImpl(AttentionImpl[HpcAttnMetadata]):
+    """HPC pure attention implementation (no RoPE/Norm).
+
+    Constraints:
+    - head_dim == 128
+    - num_heads // num_kv_heads in {4, 8}
+    - kv_cache_dtype in {"auto", "fp8_e4m3"}
+    """
+
+    def __init__(
+        self,
+        num_heads: int,
+        head_size: int,
+        scale: float,
+        num_kv_heads: int,
+        alibi_slopes: list[float] | None = None,
+        sliding_window: int | None = None,
+        kv_cache_dtype: str = "auto",
+        logits_soft_cap: float | None = None,
+        attn_type: str = AttentionType.DECODER,
+        kv_sharing_target_layer_name: str | None = None,
+    ) -> None:
+        if importlib.util.find_spec("hpc") is None:
+            raise ImportError(
+                "HPC attention requires the hpc module to be installed. "
+                "Please install it from https://github.com/Tencent/hpc-ops"
+            )
+        if attn_type != AttentionType.DECODER:
+            raise NotImplementedError("HPC attention only supports decoder attention")
+        if alibi_slopes is not None:
+            raise NotImplementedError("HPC attention does not support ALiBi")
+        if logits_soft_cap is not None:
+            raise NotImplementedError("HPC attention does not support logits_soft_cap")
+
+        if head_size != 128:
+            raise ValueError(
+                f"HPC attention only supports head_dim=128, got {head_size}"
+            )
+
+        num_queries_per_kv = num_heads // num_kv_heads
+        if num_queries_per_kv not in (4, 8):
+            raise ValueError(
+                f"HPC attention only supports head_per_group in {{4, 8}}, "
+                f"got {num_queries_per_kv} "
+                f"(num_heads={num_heads}, num_kv_heads={num_kv_heads})"
+            )
+
+        if kv_cache_dtype not in ("auto", "fp8_e4m3"):
+            raise ValueError(
+                f"HPC attention only supports kv_cache_dtype 'auto' or "
+                f"'fp8_e4m3', got '{kv_cache_dtype}'"
+            )
+
+        self.num_heads = num_heads
+        self.head_size = head_size
+        self.scale = float(scale)
+        self.num_kv_heads = num_kv_heads
+        self.kv_cache_dtype = kv_cache_dtype
+        self.kv_sharing_target_layer_name = kv_sharing_target_layer_name
+
+        self.num_queries_per_kv = num_queries_per_kv
+
+        if sliding_window is None:
+            self.sliding_window = (-1, -1)
+        else:
+            self.sliding_window = (sliding_window - 1, 0)
+
+        self.use_fp8 = kv_cache_dtype == "fp8_e4m3"
+
+        self.supports_quant_query_input = False
+        self.splitk = True
+
+    def forward(
+        self,
+        layer: torch.nn.Module,
+        query: torch.Tensor,
+        key: torch.Tensor,
+        value: torch.Tensor,
+        kv_cache: torch.Tensor,
+        attn_metadata: HpcAttnMetadata | None,
+        output: torch.Tensor | None = None,
+        output_scale: torch.Tensor | None = None,
+        output_block_scale: torch.Tensor | None = None,
+    ) -> torch.Tensor:
+        """HPC attention forward (standard vLLM backend interface).
+
+        Two modes:
+        1. Standard: upstream handles RoPE/Norm; this backend writes KV + attn.
+        2. HpcRopeNorm: fused op already did RoPE/Norm/KV-Write/Q-Quant;
+           extra params passed via attn_metadata.hpc_* fields.
+        """
+        import hpc
+
+        assert output is not None, "Output tensor must be provided."
+        assert output_scale is None, "HPC attention does not support fused output quant"
+        assert output_block_scale is None
+
+        if attn_metadata is None:
+            return output.fill_(0)
+
+        hpc_kv_written = attn_metadata.hpc_kv_written
+        hpc_prefill_q_scale = attn_metadata.hpc_prefill_q_scale
+        hpc_decode_q_scale = attn_metadata.hpc_decode_q_scale
+        hpc_split_k_flag = attn_metadata.hpc_split_k_flag
+
+        num_actual_tokens = attn_metadata.num_actual_tokens
+        num_prefill_reqs = attn_metadata.num_prefills
+        num_decode_reqs = attn_metadata.num_decodes
+        num_decode_tokens = attn_metadata.num_decode_tokens
+
+        # Write KV cache if not already done by HpcRopeNorm.
+        if self.kv_sharing_target_layer_name is None and not hpc_kv_written:
+            torch.ops._C_cache_ops.reshape_and_cache_flash(
+                key,
+                value,
+                kv_cache[:, 0],
+                kv_cache[:, 1],
+                attn_metadata.slot_mapping,
+                self.kv_cache_dtype,
+                layer._k_scale,
+                layer._v_scale,
+            )
+
+        if self.use_fp8:
+            torch_dtype = _get_fp8_dtype_for_kv_cache(self.kv_cache_dtype)
+            kv_cache = kv_cache.view(torch_dtype)
+
+        if self.use_fp8:
+            if not hpc_kv_written:
+                raise RuntimeError(
+                    "HpcAttentionImpl: FP8 mode requires HpcRopeNorm. "
+                    "Ensure hpc_rope_norm is enabled or set "
+                    "kv_cache_dtype='auto' for bf16 mode."
+                    f" (layer={getattr(layer, 'layer_name', '?')})"
+                )
+            k_scale = layer._k_scale.reshape(1)
+            v_scale = layer._v_scale.reshape(1)
+
+        query = query[:num_actual_tokens]
+        key = key[:num_actual_tokens]
+        value = value[:num_actual_tokens]
+        output_padded = output
+        output = output[:num_actual_tokens]
+
+        # --- Prefill ---
+        if num_prefill_reqs > 0:
+            seq_lens_prefill = attn_metadata.seq_lens[num_decode_reqs:]
+            cu_seqlens_prefill = attn_metadata.qo_indptr
+            max_seqlens = attn_metadata.max_query_len
+            block_table_prefill = attn_metadata.block_table_tensor[num_decode_reqs:]
+
+            q_prefill = query[num_decode_tokens:]
+            output_prefill = output[num_decode_tokens:]
+
+            if self.use_fp8:
+                hpc.attention_with_kvcache_prefill_fp8(
+                    q_prefill,
+                    kv_cache[:, 0],
+                    kv_cache[:, 1],
+                    hpc_prefill_q_scale,
+                    k_scale,
+                    v_scale,
+                    cu_seqlens_prefill,
+                    block_table_prefill,
+                    seq_lens_prefill,
+                    max_seqlens,
+                    output=output_prefill,
+                )
+            else:
+                hpc.attention_with_kvcache_prefill_bf16(
+                    q_prefill,
+                    kv_cache[:, 0],
+                    kv_cache[:, 1],
+                    cu_seqlens_prefill,
+                    block_table_prefill,
+                    seq_lens_prefill,
+                    max_seqlens,
+                    output=output_prefill,
+                )
+
+        # --- Decode ---
+        if num_decode_reqs > 0:
+            num_seq_kvcache = attn_metadata.seq_lens[:num_decode_reqs]
+            block_table_decode = attn_metadata.block_table_tensor[:num_decode_reqs]
+
+            q_decode = query[:num_decode_tokens]
+            output_decode = output[:num_decode_tokens]
+
+            if self.use_fp8:
+                hpc.attention_decode_fp8(
+                    q_decode,
+                    kv_cache[:, 0],
+                    kv_cache[:, 1],
+                    block_table_decode,
+                    num_seq_kvcache,
+                    hpc_decode_q_scale,
+                    k_scale,
+                    v_scale,
+                    new_kv_included=True,
+                    splitk=self.splitk,
+                    split_flag=hpc_split_k_flag,
+                    output=output_decode,
+                )
+            else:
+                hpc.attention_decode_bf16(
+                    q_decode,
+                    kv_cache[:, 0],
+                    kv_cache[:, 1],
+                    block_table_decode,
+                    num_seq_kvcache,
+                    output=output_decode,
+                    new_kv_included=True,
+                    splitk=self.splitk,
+                )
+
+        return output_padded
@@ -103,6 +103,12 @@ class AttentionBackendEnum(Enum, metaclass=_AttentionBackendEnumMeta):
    )
    NO_ATTENTION = "vllm.v1.attention.backends.no_attention.NoAttentionBackend"
    FLEX_ATTENTION = "vllm.v1.attention.backends.flex_attention.FlexAttentionBackend"
+    # HPC Attention Backend:
+    # powered by operators from https://github.com/Tencent/hpc-ops.
+    # Only supported on NVIDIA Hopper GPUs (e.g. H20, H200),
+    # currently limited to the Hy3 model,
+    # and requires a block size of 64.
+    HPC_ATTN = "vllm.v1.attention.backends.hpc_attn.HpcAttentionBackend"
    ROCM_AITER_UNIFIED_ATTN = (
        "vllm.v1.attention.backends.rocm_aiter_unified_attn."
        "RocmAiterUnifiedAttentionBackend"
@@ -38,6 +38,14 @@ class _StreamPlaceholder:
        pass


+from vllm.utils.cpu_resource_utils import get_memory_node_info
+
+
+def get_memory_info(*args: Any, **kwargs: Any) -> tuple[int, int]:
+    meminfo = get_memory_node_info()
+    return meminfo.available_memory, meminfo.total_memory
+
+
 torch.Event = _EventPlaceholder
 torch.cuda.Event = _EventPlaceholder
 torch.cuda.Stream = _StreamPlaceholder
@@ -46,6 +54,7 @@ torch.cuda.current_stream = lambda *args, **kwargs: _StreamPlaceholder()
 torch.accelerator.synchronize = noop
 torch.accelerator.empty_cache = noop
 torch.Tensor.pin_memory = fake_pin_memory
+torch.accelerator.get_memory_info = get_memory_info

 # Patch vLLM torch utils
 import vllm.utils.torch_utils as torch_utils
@@ -701,7 +701,7 @@ class GPUModelRunner(LoRAModelRunnerMixin):
        start_time = time.perf_counter()
        gc.collect()
        torch.accelerator.empty_cache()
-        start_free_gpu_memory = torch.cuda.mem_get_info()[0]
+        start_free_gpu_memory = torch.accelerator.get_memory_info()[0]

        with self.maybe_setup_dummy_loras(self.lora_config):
            attn_states = self.cudagraph_manager.capture(
@@ -720,7 +720,7 @@ class GPUModelRunner(LoRAModelRunnerMixin):
                self.speculator.capture(attn_states)

        end_time = time.perf_counter()
-        end_free_gpu_memory = torch.cuda.mem_get_info()[0]
+        end_free_gpu_memory = torch.accelerator.get_memory_info()[0]
        elapsed_time = end_time - start_time
        cuda_graph_size = start_free_gpu_memory - end_free_gpu_memory
        # This usually takes 5~20 seconds.
@@ -20,7 +20,7 @@ def _should_share(eagle: nn.Module, flag: str, draft, target) -> bool:
    # Use the faster GPU path when there is plenty of headroom;
    # otherwise compare on CPU.
    w = draft.weight
-    if w.is_cuda and torch.cuda.mem_get_info(w.device)[0] < w.numel() * 2:
+    if w.is_cuda and torch.accelerator.get_memory_info(w.device)[0] < w.numel() * 2:
        return torch.equal(w.cpu(), target.weight.cpu())
    return torch.equal(w, target.weight)

@@ -6385,7 +6385,7 @@ class GPUModelRunner(
        _ROPE_DICT.clear()

        reset_workspace_manager()
-        if current_platform.is_rocm():
+        if current_platform.is_rocm() or current_platform.is_xpu():
            gc.collect()
            torch.accelerator.empty_cache()
            torch.accelerator.synchronize()
@@ -6527,7 +6527,7 @@ class GPUModelRunner(
                    mem_samples: list[int] = []

                    for i, desc in enumerate(profile_descs):
-                        mem_before = torch.cuda.mem_get_info()[0]
+                        mem_before = torch.accelerator.get_memory_info()[0]
                        self._warmup_and_capture(
                            desc,
                            cudagraph_runtime_mode=mode,
@@ -6541,7 +6541,7 @@ class GPUModelRunner(
                            ),
                        )
                        torch.accelerator.synchronize()
-                        free_after = torch.cuda.mem_get_info()[0]
+                        free_after = torch.accelerator.get_memory_info()[0]
                        mem_samples.append(mem_before - free_after)

                    first_capture = mem_samples[0]
@@ -6563,10 +6563,10 @@ class GPUModelRunner(
                    )

                if encoder_cudagraph_manager is not None:
-                    mem_before = torch.cuda.mem_get_info()[0]
+                    mem_before = torch.accelerator.get_memory_info()[0]
                    encoder_cudagraph_manager.capture(graph_pool=encoder_profiling_pool)
                    torch.accelerator.synchronize()
-                    free_after = torch.cuda.mem_get_info()[0]
+                    free_after = torch.accelerator.get_memory_info()[0]
                    encoder_memory_estimate = max(mem_before - free_after, 0)

                    logger.debug(
@@ -6632,7 +6632,7 @@ class GPUModelRunner(
        with self._freeze_gc(), graph_capture(device=self.device):
            torch.accelerator.synchronize()
            torch.accelerator.empty_cache()
-            start_free_gpu_memory = torch.cuda.mem_get_info()[0]
+            start_free_gpu_memory = torch.accelerator.get_memory_info()[0]

            for (
                runtime_mode,
@@ -6650,7 +6650,7 @@ class GPUModelRunner(
                self.encoder_cudagraph_manager.capture(graph_pool=encoder_graph_pool)

            torch.accelerator.synchronize()
-            end_free_gpu_memory = torch.cuda.mem_get_info()[0]
+            end_free_gpu_memory = torch.accelerator.get_memory_info()[0]

        # Disable cudagraph capturing globally, so any unexpected cudagraph
        # capturing will be detected and raise an error after here.
@@ -172,7 +172,7 @@ class Worker(WorkerBase):

    def sleep(self, level: int = 1) -> None:
        torch.accelerator.synchronize()
-        free_bytes_before_sleep = current_platform.mem_get_info()[0]
+        free_bytes_before_sleep = torch.accelerator.get_memory_info()[0]

        # Save the buffers before level 2 sleep
        if level == 2:
@@ -187,7 +187,7 @@ class Worker(WorkerBase):
        torch.accelerator.synchronize()
        deadline = time.monotonic() + (5.0 if current_platform.is_rocm() else 0)
        while True:
-            free_bytes_after_sleep, total = current_platform.mem_get_info()
+            free_bytes_after_sleep, total = torch.accelerator.get_memory_info()
            freed_bytes = free_bytes_after_sleep - free_bytes_before_sleep
            if freed_bytes >= 0 or time.monotonic() >= deadline:
                break
@@ -459,8 +459,8 @@ class Worker(WorkerBase):
            )

            # Profile CUDA graph memory if graphs will be captured.
-            # Skip on ROCm/HIP/XPU as graph pool handles and mem_get_info behave
-            # differently and can produce incorrect/negative estimates.
+            # Skip on ROCm/HIP/XPU as graph pool handles and get_memory_info
+            # behave differently and can produce incorrect/negative estimates.
            cudagraph_memory_estimate = 0
            if (
                current_platform.is_cuda()
@@ -1294,10 +1294,11 @@ class Worker(WorkerBase):
        # Release kept-alive cumem pools while the pluggable allocator wrappers
        # and callbacks are still alive, so MemPool teardown is not deferred to
        # interpreter finalization (pytorch/pytorch#145168).
-        from vllm.device_allocator.cumem import CuMemAllocator
+        if current_platform.is_cuda_alike():
+            from vllm.device_allocator.cumem import CuMemAllocator

-        if CuMemAllocator.instance is not None:
-            CuMemAllocator.instance.release_pools()
+            if CuMemAllocator.instance is not None:
+                CuMemAllocator.instance.release_pools()

    def elastic_ep_execute(self, execute_method: str, *args, **kwargs):
        return self.elastic_ep_executor.execute(execute_method, *args, **kwargs)
@@ -45,7 +45,6 @@ def _torch_cuda_wrapper():
    torch.cuda.default_stream = torch.xpu.current_stream
    torch.cuda.current_stream = torch.xpu.current_stream
    torch.cuda.stream = torch.xpu.stream
-    torch.cuda.mem_get_info = torch.xpu.mem_get_info
    torch.cuda.Event = torch.Event
    torch.cuda.set_stream = torch.xpu.set_stream
    if supports_xpu_graph():
@@ -162,3 +162,20 @@ class XPUWorker(Worker):
            logger.debug("Starting torch profiler with trace name: %s", trace_name)

        super().profile(is_start=is_start, profile_prefix=profile_prefix)
+
+    def shutdown(self) -> None:
+        logger.info(
+            "XPUWorker shutdown: cleaning up (rank=%d, local_rank=%d)",
+            self.rank,
+            self.local_rank,
+        )
+        super().shutdown()
+        from vllm.device_allocator.xpumem import XpuMemAllocator
+
+        if XpuMemAllocator.instance is not None:
+            XpuMemAllocator.instance.release_pools()
+        logger.info(
+            "XPUWorker shutdown: done (rank=%d, local_rank=%d)",
+            self.rank,
+            self.local_rank,
+        )
Author	SHA1	Message	Date
dependabot[bot]andGitHub	95dcefaaa5	Bump actions/setup-python from 6.1.0 to 6.3.0 Bumps [actions/setup-python](https://github.com/actions/setup-python) from 6.1.0 to 6.3.0. - [Release notes](https://github.com/actions/setup-python/releases) - [Commits](https://github.com/actions/setup-python/compare/83679a892e2d95755f2dac6acb0bfd1e9ac5d548...ece7cb06caefa5fff74198d8649806c4678c61a1) --- updated-dependencies: - dependency-name: actions/setup-python dependency-version: 6.3.0 dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com>	2026-06-30 12:18:49 +00:00
dependabot[bot]GitHubdependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	536047755e	Bump actions/checkout from 6.0.1 to 7.0.0 (#33057 ) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-06-30 13:16:20 +01:00
JianweiZheng GitHub Claude Wentao Ye Harry Mellor	1907d3854a	[Bugfix] Reject negative values for max_logprobs and long_prefill_token_threshold (#44002 ) Signed-off-by: jwzheng96 <jianweizheng@pku.edu.cn> Signed-off-by: JianweiZheng <32029023+jwzheng96@users.noreply.github.com> Co-authored-by: Claude <noreply@anthropic.com> Co-authored-by: Wentao Ye <44945378+yewentao256@users.noreply.github.com> Co-authored-by: Harry Mellor <19981378+hmellor@users.noreply.github.com>	2026-06-30 13:01:03 +01:00
Chaojun ZhangandGitHub	ea9ddf59fc	[XPU][CI] Enable shared loader test (#45977 ) Signed-off-by: Chaojun Zhang <chaojun.zhang@intel.com>	2026-06-30 11:20:33 +00:00
Cheng Jiang GitHub chengvjiang Andreas Karatzas	8cf7c4d8ad	[Attention Backend] add HPC-Ops Attention backend (#46020 ) Signed-off-by: chengvjiang <chengvjiang@tencent.com> Co-authored-by: chengvjiang <chengvjiang@tencent.com> Co-authored-by: Andreas Karatzas <akaratza@amd.com>	2026-06-30 18:17:43 +08:00
Agata Dobrzyniewicz GitHub Kunshang Ji	8e9d70fdd5	[Kernel][XPU] Adjust kernel unit tests for XPU (#45140 ) Signed-off-by: Dobrzyniewicz, Agata <agata.dobrzyniewicz@intel.com> Co-authored-by: Kunshang Ji <kunshang.ji@intel.com>	2026-06-30 09:57:27 +00:00
Juan Pérez de AlgabaandGitHub	364ee36af1	fix(security): prevent image decompression bomb OOM denial of service (#47010 ) Signed-off-by: jperezde <jperezde@redhat.com>	2026-06-30 09:39:22 +00:00
Nicolò LucchesiandGitHub	06fae69114	[Misc] Mistral label alert (#47132 ) Signed-off-by: NickLucche <nicolo.lucchesi@mistral.ai>	2026-06-30 09:02:07 +00:00
Li, Jiang GitHub Harry Mellor	14f8660a18	[CI/Build] Add CPU test dependency pre-commit hooks (#47032 ) Signed-off-by: jiang1.li <jiang1.li@intel.com> Signed-off-by: Harry Mellor <19981378+hmellor@users.noreply.github.com> Co-authored-by: Harry Mellor <19981378+hmellor@users.noreply.github.com>	2026-06-30 07:59:13 +00:00
Aman Paswan GitHub amanambak Chauncey	aed541def4	[Bugfix][Responses] Set completed status for Harmony function calls (#46945 ) Signed-off-by: amanambak <aman.paswan@ambak.com> Co-authored-by: amanambak <aman.paswan@ambak.com> Co-authored-by: Chauncey <chaunceyjiang@gmail.com>	2026-06-30 07:55:14 +00:00
Chauncey GitHubmergify[bot] <37929162+mergify[bot]@users.noreply.github.com>	2bc20e8aba	[Frontend] Add Streaming Parser Engine and new Kimi k2.5/k2.6/k2.7 Parser (#46610 ) Signed-off-by: chaunceyjiang <chaunceyjiang@gmail.com> Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>	2026-06-30 07:53:17 +00:00
Chaojun ZhangandGitHub	8cc242335d	[XPU] Optimize XPU worker shutdown logic to prevent resource leak (#46433 ) Signed-off-by: Chaojun Zhang <chaojun.zhang@intel.com>	2026-06-30 15:27:21 +08:00
Andreas KaratzasandGitHub	ba22cb6765	[ROCm][Ray][CI] Keep assigned GPU visible for weight transfer (#47000 ) Signed-off-by: Andreas Karatzas <akaratza@amd.com>	2026-06-30 14:59:18 +08:00
Uros MarkovicandGitHub	81bcced482	[Bugfix][ROCm] Preserve MoE weight padding for unquantized Triton path (#46381 ) Signed-off-by: Uros Markovic <umarkovi@amd.com>	2026-06-30 14:47:57 +08:00
Kunshang JiandGitHub	fb42e5219e	[Platform] Replace `torch.cuda.mem_get_info` with `torch.accelerator.get_memory_info` (#44825 ) Signed-off-by: Kunshang Ji <kunshang.ji@intel.com> Signed-off-by: Kunshang Ji <jikunshang95@gmail.com>	2026-06-30 14:39:52 +08:00
Dakai AnandGitHub	0feca7ffa8	PD disagg with Mooncake Connector: GDN support (Qwen3.5) and MLA support (Deepseek-V4-Flash) (#46807 )	2026-06-29 23:29:04 -07:00