feat: kvbm + connector (#2258)

Signed-off-by: Ryan Olson <rolson@nvidia.com>
Co-authored-by: Olga Andreeva <oandreeva@nvidia.com>
Co-authored-by: Ziqi Fan <ziqif@nvidia.com>
Co-authored-by: John Thompson <jothomson@nvidia.com>
Co-authored-by: Richard Huo <rihuo@nvidia.com>
Co-authored-by: Zicheng Ma <zichengm@nvidia.com>

lib/llm/tests/block_manager.rs

@@ -481,7 +481,7 @@ mod tests {
             .build()
             .unwrap();
 
-        ReferenceBlockManager::new(config).unwrap()
+        ReferenceBlockManager::new(config).await.unwrap()
     }
 
     async fn setup_kvbm_component(

lib/runtime/src/distributed.rs

@@ -43,6 +43,12 @@ impl MetricsRegistry for DistributedRuntime {
     }
 }
 
+impl std::fmt::Debug for DistributedRuntime {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "DistributedRuntime")
+    }
+}
+
 impl DistributedRuntime {
     pub async fn new(runtime: Runtime, config: DistributedConfig) -> Result<Self> {
         let (etcd_config, nats_config, is_static) = config.dissolve();

lib/runtime/src/pipeline/network/ingress/push_endpoint.rs

 // SPDX-FileCopyrightText: Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 // SPDX-License-Identifier: Apache-2.0
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
 
 use std::sync::atomic::{AtomicU64, Ordering};
 
@@ -136,7 +124,7 @@ impl PushEndpoint {
                             tracing::trace!(instance_id, "request handled successfully");
                         }
                         Err(e) => {
-                            tracing::warn!("Failed to handle request: {:?}", e);
+                            tracing::warn!("Failed to handle request: {}", e.to_string());
                         }
                     }
 

lib/runtime/src/pipeline/network/ingress/push_handler.rs

@@ -211,8 +211,21 @@ where
                 stream
             }
             Err(e) => {
-                tracing::error!("Failed to generate response stream: {:?}", e);
-                let _result = publisher.send_prologue(Some(e.to_string())).await;
+                let error_string = e.to_string();
+
+                #[cfg(debug_assertions)]
+                {
+                    tracing::debug!(
+                        "Failed to generate response stream (with debug backtrace): {:?}",
+                        e
+                    );
+                }
+                #[cfg(not(debug_assertions))]
+                {
+                    tracing::error!("Failed to generate response stream: {}", error_string);
+                }
+
+                let _result = publisher.send_prologue(Some(error_string)).await;
                 Err(e)?
             }
         };

lib/runtime/src/worker.rs

@@ -88,6 +88,16 @@ impl Worker {
         Ok(Worker { runtime, config })
     }
 
+    pub fn runtime_from_existing() -> Result<Runtime> {
+        if let Some(rt) = RT.get() {
+            Ok(Runtime::from_handle(rt.handle().clone())?)
+        } else if let Some(rt) = RTHANDLE.get() {
+            Ok(Runtime::from_handle(rt.clone())?)
+        } else {
+            Runtime::from_settings()
+        }
+    }
+
     pub fn tokio_runtime(&self) -> Result<&'static tokio::runtime::Runtime> {
         RT.get().ok_or_else(|| error!("Worker not initialized"))
     }

pyproject.toml

@@ -135,6 +135,7 @@ addopts = [
     "--strict-config",
     "--mypy",
     "--ignore-glob=*model.py",
+    "--ignore-glob=*vllm_integration*",
     "--ignore-glob=*_inc.py",
     "--ignore-glob=*/llm/tensorrtllm*",
     "--ignore-glob=docs/*",
@@ -173,7 +174,8 @@ markers = [
     "vllm: marks tests as requiring vllm",
     "trtllm_marker: marks tests as requiring trtllm",
     "sglang: marks tests as requiring sglang",
-    "slow: marks tests as known to be slow"
+    "slow: marks tests as known to be slow",
+    "kvbm: marks tests for KV behavior and model determinism"
 ]
 
 # Linting/formatting

tests/kvbm/README.md

+# KV Behavior & Model Determinism Tests (kvbm)
+
+## Overview
+
+This suite validates determinism properties of the API-backed LLM under fixed sampling parameters and optionally across prefix cache resets. The tests can automatically start a local vLLM server, warm it up, and compare responses for identical prompts over multiple iterations.
+
+## Files
+
+- `test_determinism.py` — comprehensive determinism tests with automatic vLLM server lifecycle and warmup.
+  - `test_determinism_with_cache_reset` — run test with warmup, reset cache, then run again without warmup to test determinism across cache reset boundary
+  - `test_concurrent_determinism_with_ifeval` — send parametrized number of IFEval prompts (default: 120) with controlled concurrency, with warmup, then reset cache and test again without warmup to validate determinism across cache reset
+
+## Markers
+
+- `kvbm` — KV behavior and model determinism tests
+- `e2e` — end-to-end tests
+- `slow` — tests may take a while due to warmup/iterations
+- `nightly` — preferred for nightly runs
+
+## How It Works
+
+- A `VLLMServerManager` fixture (`vllm_server`) launches `vllm serve` with the Dynamo connector and optional cache block overrides.
+- A `tester` fixture binds the test client to the running server's base URL.
+- The test performs a comprehensive warmup across prompts, then executes repeated requests and checks that responses are identical (deterministic). An optional cache reset phase re-validates determinism across the reset boundary.
+
+## Running
+
+Run all kvbm tests:
+
+```bash
+pytest -v -m "kvbm" -s
+```
+
+Run the determinism test file directly:
+
+```bash
+pytest -v dynamo/tests/kvbm/test_determinism.py -s
+```
+
+## Configuration
+
+Environment variables control server settings and test load:
+
+- Server/model
+  - `KVBM_MODEL_ID` (default: `deepseek-ai/DeepSeek-R1-Distill-Llama-8B`)
+  - `KVBM_VLLM_PORT` (default: `8000`)
+  - `KVBM_VLLM_START_TIMEOUT` (default: `300` seconds)
+
+- Cache size overrides
+  - `KVBM_CPU_BLOCKS` (used via test parametrization; default: `10000`)
+  - `--num-gpu-blocks-override` is applied when `gpu_blocks` is parametrized
+
+- Request/test parameters
+  - `KVBM_MAX_TOKENS` (default: `48`)
+  - `KVBM_SEED` (default: `42`)
+  - `KVBM_MAX_ITERATIONS` (default: `500`)
+  - `KVBM_WORD_COUNT` (default: `200`)
+  - `KVBM_CONTROL_INTERVAL` (default: `10`)
+  - `KVBM_SHAKESPEARE_INTERVAL` (default: `1`)
+  - `KVBM_RANDOM_INTERVAL` (default: `7`)
+  - `KVBM_HTTP_TIMEOUT` (default: `30` seconds)
+  - `KVBM_SHAKESPEARE_URL` (default: MIT OCW Shakespeare text)
+
+- Concurrent testing
+  - `KVBM_CONCURRENT_REQUESTS` (default: `"3"` - comma-separated list for parametrization of max concurrent workers)
+  - `KVBM_MAX_TOKENS` (default: `"10"` - comma-separated list for parametrization of max_tokens in concurrent tests)
+  - `KVBM_IFEVAL_PROMPTS` (default: `"120"` - comma-separated list for parametrization of number of IFEval prompts to use)
+
+Example:
+
+```bash
+KVBM_MODEL_ID=Qwen/Qwen3-0.6B \
+KVBM_CPU_BLOCKS=12000 \
+KVBM_MAX_ITERATIONS=100 \
+KVBM_CONCURRENT_REQUESTS="10,25,50" \
+KVBM_MAX_TOKENS="48,128,256" \
+KVBM_IFEVAL_PROMPTS="50,120,200" \
+pytest -v -m "kvbm" -s
+```
+
+## Requirements
+
+- `vllm` executable available in PATH inside the test environment.
+- The connector module path must be valid: `dynamo.llm.vllm_integration.connector`.
+- NATS and etcd services (provided automatically by the `runtime_services` fixture).
+- `datasets` library for IFEval concurrent testing (included in test dependencies).
+- For containerized workflows, follow the top-level `tests/README.md` guidance to build/run the appropriate image, then execute pytest inside the container.
+
+## Notes
+
+- Warmup is critical to avoid initialization effects impacting determinism.
+- For faster local iteration, reduce `KVBM_MAX_ITERATIONS` and/or increase intervals.
+- Logs are written under the per-test directory created by `tests/conftest.py` and include the vLLM server stdout/stderr.
+- Tests use the static port defined by `KVBM_VLLM_PORT` for vLLM server communication.
\ No newline at end of file

tests/kvbm/__init__.py

+# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0

tests/kvbm/test_determinism.py

+#!/usr/bin/env python3
+# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""
+Determinism test for language model API using pytest.
+
+This test suite checks if the model produces deterministic outputs
+when given the same inputs with fixed seed and temperature=0.
+
+The test uses comprehensive server warmup (sending all test prompts
+before validation) to avoid server initialization effects that could
+impact determinism measurements.
+"""
+
+import logging
+import os
+import signal
+import subprocess
+import time
+from collections import defaultdict
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from datetime import datetime
+from pathlib import Path
+from typing import Dict, List, Optional, TextIO, Tuple
+
+import pytest
+import requests
+
+# Test markers to align with repository conventions
+pytestmark = [
+    pytest.mark.kvbm,
+    pytest.mark.e2e,
+    pytest.mark.slow,
+    pytest.mark.nightly,
+    pytest.mark.gpu_1,
+]
+
+
+class VLLMServerManager:
+    """Manages vLLM server lifecycle for determinism testing."""
+
+    def __init__(
+        self,
+        base_url: Optional[str] = None,
+        port: Optional[int] = None,
+        cpu_cache_blocks: Optional[int] = None,
+        gpu_cache_blocks: Optional[int] = None,
+        log_dir: Optional[Path] = None,
+    ):
+        self.port = port or int(os.environ.get("KVBM_VLLM_PORT", "8000"))
+        self.base_url = base_url or f"http://localhost:{self.port}"
+        self.process: Optional[subprocess.Popen] = None
+        self.cpu_cache_blocks = cpu_cache_blocks
+        self.gpu_cache_blocks = gpu_cache_blocks
+
+        # Prepare logging
+        self.log_dir = log_dir or Path(".")
+        self.log_dir.mkdir(parents=True, exist_ok=True)
+        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+        config_str = (
+            f"cpu{cpu_cache_blocks or 'default'}_gpu{gpu_cache_blocks or 'default'}"
+        )
+        self.server_log_file = (
+            self.log_dir / f"vllm_server_{config_str}_{timestamp}.log"
+        )
+        self.server_stdout_file: Optional[TextIO] = None
+        self.server_stderr_file: Optional[TextIO] = None
+
+        # Construct serve command
+        self.server_cmd = [
+            "vllm",
+            "serve",
+            "--block-size",
+            "16",
+            "--port",
+            str(self.port),
+            "--kv-transfer-config",
+            '{"kv_connector":"DynamoConnector","kv_role":"kv_both", "kv_connector_module_path": "dynamo.llm.vllm_integration.connector"}',
+            os.environ.get("KVBM_MODEL_ID", "deepseek-ai/DeepSeek-R1-Distill-Llama-8B"),
+        ]
+
+        # GPU blocks override
+        if gpu_cache_blocks is not None:
+            self.server_cmd.extend(["--num-gpu-blocks-override", str(gpu_cache_blocks)])
+
+        # Environment for the process
+        self.env = os.environ.copy()
+        self.env.update(
+            {
+                "RUST_BACKTRACE": "1",
+                "DYN_LOG": os.environ.get(
+                    "DYN_LOG", "debug,dynamo_llm::block_manager::layout=error"
+                ),
+                "VLLM_SERVER_DEV_MODE": "1",
+                # DynamoConnector connection settings
+                "NATS_SERVER": "nats://localhost:4222",
+                "ETCD_ENDPOINTS": "http://localhost:2379",
+            }
+        )
+
+        # CPU cache blocks override via env
+        if cpu_cache_blocks is not None:
+            self.env["DYN_KVBM_CPU_CACHE_OVERRIDE_NUM_BLOCKS"] = str(cpu_cache_blocks)
+
+    def start_server(self, timeout: int = 300) -> bool:
+        """Start vLLM server and wait for readiness."""
+        if self.is_server_running():
+            self.stop_server()
+            time.sleep(2)
+
+        # Open log files
+        self.server_stdout_file = open(
+            self.server_log_file.with_suffix(".stdout.log"), "w"
+        )
+        self.server_stderr_file = open(
+            self.server_log_file.with_suffix(".stderr.log"), "w"
+        )
+        if self.server_stdout_file is not None:
+            self.server_stdout_file.write(
+                f"=== vLLM Server Started at {datetime.now()} ===\nCommand: {' '.join(self.server_cmd)}\n"
+            )
+            self.server_stdout_file.flush()
+
+        # Launch
+        self.process = subprocess.Popen(
+            self.server_cmd,
+            stdout=self.server_stdout_file,
+            stderr=self.server_stderr_file,
+            env=self.env,
+            preexec_fn=os.setsid,
+        )
+
+        # Wait for health
+        start_time = time.time()
+        while time.time() - start_time < timeout:
+            if self.is_server_running():
+                return True
+            if self.process.poll() is not None:
+                self._close_log_files()
+                return False
+            time.sleep(5)
+
+        # Timeout
+        self.stop_server()
+        return False
+
+    def stop_server(self):
+        """Stop vLLM server and close logs."""
+        if self.process:
+            try:
+                os.killpg(os.getpgid(self.process.pid), signal.SIGTERM)
+                try:
+                    self.process.wait(timeout=30)
+                except subprocess.TimeoutExpired:
+                    os.killpg(os.getpgid(self.process.pid), signal.SIGKILL)
+                    self.process.wait()
+            except (ProcessLookupError, OSError):
+                pass
+            finally:
+                self.process = None
+        self._close_log_files()
+
+    def _close_log_files(self):
+        if self.server_stdout_file:
+            self.server_stdout_file.write(
+                f"\n=== Server Stopped at {datetime.now()} ===\n"
+            )
+            self.server_stdout_file.close()
+            self.server_stdout_file = None
+        if self.server_stderr_file:
+            self.server_stderr_file.close()
+            self.server_stderr_file = None
+
+    def is_server_running(self) -> bool:
+        try:
+            # First check basic health
+            response = requests.get(f"{self.base_url}/health", timeout=5)
+            if response.status_code != 200:
+                return False
+
+            # Then check if the model endpoint is ready with a simple test request
+            test_payload = {
+                "model": os.environ.get(
+                    "KVBM_MODEL_ID", "deepseek-ai/DeepSeek-R1-Distill-Llama-8B"
+                ),
+                "messages": [{"role": "user", "content": "test"}],
+                "max_completion_tokens": 1,
+                "temperature": 0,
+            }
+
+            response = requests.post(
+                f"{self.base_url}/v1/chat/completions",
+                headers={"Content-Type": "application/json"},
+                json=test_payload,
+                timeout=10,
+            )
+            return response.status_code == 200
+
+        except requests.exceptions.RequestException:
+            return False
+
+
+class DeterminismTester:
+    """Test class for model determinism validation."""
+
+    def __init__(self, base_url: Optional[str] = None, model_id: Optional[str] = None):
+        # Allow environment override for flexibility in CI/local runs
+        self.base_url = (
+            base_url or os.environ.get("DYNAMO_API_BASE_URL") or "http://localhost:8000"
+        )
+        self.model_id = (
+            model_id
+            or os.environ.get("KVBM_MODEL_ID")
+            or "deepseek-ai/DeepSeek-R1-Distill-Llama-8B"
+        )
+
+        self.shakespeare_file = Path("t8.shakespeare.txt")
+        self.max_iterations = int(os.environ.get("KVBM_MAX_ITERATIONS", "500"))
+        self.word_count = int(os.environ.get("KVBM_WORD_COUNT", "200"))
+
+        # Test intervals
+        self.control_interval = int(os.environ.get("KVBM_CONTROL_INTERVAL", "10"))
+        self.shakespeare_interval = int(
+            os.environ.get("KVBM_SHAKESPEARE_INTERVAL", "1")
+        )
+        self.random_interval = int(os.environ.get("KVBM_RANDOM_INTERVAL", "7"))
+
+        # Response storage
+        self.control_responses: Dict[int, List[str]] = defaultdict(list)
+        self.shakespeare_responses: Dict[int, List[str]] = defaultdict(list)
+        self.random_responses: Dict[int, List[str]] = defaultdict(list)
+
+        # Control sequences
+        self.control_sequences = [
+            "Hello world",
+            "The quick brown fox jumps over the lazy dog. This is a standard pangram that contains all letters of the alphabet.",
+            "Find light in the beautiful sea, I choose to be happy, You and I, you and I, we are like a beautiful melody that never ends, dancing through the night with stars as our companions, whispering secrets to the wind as we journey through life together, hand in hand, heart to heart, forever and always.",
+            "The advancement of technology has fundamentally transformed the way we live, work, and communicate in the modern world. From the invention of the printing press to the development of the internet, each technological breakthrough has opened new possibilities and created unprecedented opportunities for human progress. Today, artificial intelligence and machine learning are reshaping industries, healthcare, education, and countless other fields, promising to solve complex problems and improve the quality of life for people around the globe.",
+            "In the heart of Eldoria, an ancient land of boundless magic and mysterious creatures, lies the long-forgotten city of Aeloria. Once a beacon of knowledge and power, Aeloria was buried beneath the shifting sands of time, lost to the world for centuries. You are an intrepid explorer, known for your unparalleled curiosity and courage, who has stumbled upon an ancient map hinting at ests that Aeloria holds a secret so profound that it has the potential to reshape the very fabric of reality. Your journey will take you through treacherous deserts, enchanted forests, and across perilous mountain ranges. Your Task: Character Background: Develop a detailed background for your character. Describe their motivations for seeking out Aeloria, their skills and weaknesses, and any personal connections to the ancient city or its legends. Are they driven by a quest for knowledge, a search for lost familt clue is hidden.",
+            "The human brain is the most complex organ in the known universe, containing approximately 86 billion neurons, each connected to thousands of others through intricate networks of synapses. This biological supercomputer processes information at speeds that would make even the most advanced artificial intelligence systems seem primitive by comparison. Every thought, memory, emotion, and decision we make is the result of electrical and chemical signals traveling through this vast neural network. The brain's ability to learn, adapt, and create is unmatched by any machine we have ever built. It can recognize patterns in milliseconds, solve complex problems through intuition, and generate creative ideas that have never existed before. Yet despite our incredible advances in neuroscience, we still understand only a fraction of how this remarkable organ truly works. The mysteries of consciousness, memory formation, and the nature of human intelligence continue to challenge the brightest minds in science and philosophy.",
+        ]
+
+        # Random sequences
+        self.random_sequences = [
+            "Coffee is ready",
+            "The cat sat on the mat while the dog slept peacefully in the corner, creating a perfect picture of domestic tranquility that warmed the heart of anyone who witnessed this simple moment of harmony between two natural enemies turned friends.",
+            "Mathematics is the language of the universe, and numbers are its alphabet. Through the elegant dance of equations and the symphony of algorithms, we unlock the secrets of nature's most profound mysteries. From the simple beauty of prime numbers to the complex elegance of calculus, mathematics provides us with the tools to understand everything from the smallest subatomic particles to the vast expanse of galaxies stretching across the cosmic void.",
+            "A journey of a thousand miles begins with a single step, as the ancient Chinese proverb wisely reminds us. This timeless wisdom speaks to the fundamental truth that every great achievement, every monumental discovery, and every life-changing transformation starts with that crucial moment of decision - the moment when we choose to take action instead of remaining in the comfort of inaction. Whether it's learning a new skill, starting a business, writing a novel, or embarking on a spiritual quest, the path to success is paved with countless small steps, each one building upon the last, until we find ourselves transformed by the journey itself.",
+            "Technology evolves rapidly, but human nature remains constant through the ages. Despite the incredible advances in artificial intelligence, virtual reality, and biotechnology, the fundamental desires, fears, and aspirations that drive human behavior have remained remarkably consistent throughout history. We still seek connection, meaning, and purpose in our lives. We still fear the unknown and crave security. We still dream of a better future and work to create it for ourselves and our loved ones. This paradox - the ever-changing nature of our tools and the unchanging nature of our hearts - is perhaps the most fascinating aspect of the human condition, reminding us that while we may build increasingly sophisticated machines, we remain fundamentally human in our core essence.",
+        ]
+
+    def download_shakespeare_text(self):
+        """Download Shakespeare text if not present."""
+        if not self.shakespeare_file.exists():
+            print("Downloading Shakespeare text...")
+            import urllib.request
+
+            url = os.environ.get(
+                "KVBM_SHAKESPEARE_URL",
+                "https://ocw.mit.edu/ans7870/6/6.006/s08/lecturenotes/files/t8.shakespeare.txt",
+            )
+            urllib.request.urlretrieve(url, self.shakespeare_file)
+
+            # Remove double newlines
+            with open(self.shakespeare_file, "r", encoding="utf-8") as f:
+                content = f.read()
+            content = content.replace("\n\n", "")
+            with open(self.shakespeare_file, "w", encoding="utf-8") as f:
+                f.write(content)
+
+    def make_request(self, content: str) -> str:
+        """Make API request and return completion text."""
+        payload = {
+            "model": self.model_id,
+            "messages": [
+                {"role": "user", "content": content},
+            ],
+            "stream": False,
+            "max_completion_tokens": int(os.environ.get("KVBM_MAX_TOKENS", "48")),
+            "temperature": 0,
+            "top_p": 0.0001,
+            "seed": int(os.environ.get("KVBM_SEED", "42")),
+        }
+
+        response = requests.post(
+            f"{self.base_url}/v1/chat/completions",
+            headers={"Content-Type": "application/json"},
+            json=payload,
+            timeout=int(os.environ.get("KVBM_HTTP_TIMEOUT", "30")),
+        )
+        response.raise_for_status()
+
+        data = response.json()
+        return data["choices"][0]["message"]["content"]
+
+    def reset_prefix_cache(self):
+        """Reset the prefix cache."""
+        print("Resetting prefix cache...")
+        response = requests.post(
+            f"{self.base_url}/reset_prefix_cache",
+            timeout=int(os.environ.get("KVBM_HTTP_TIMEOUT", "30")),
+        )
+        response.raise_for_status()
+        print("Cache reset done")
+
+    def warmup_server(self):
+        """Perform comprehensive server warmup with all test prompts."""
+        print("=" * 70)
+        print("PERFORMING COMPREHENSIVE SERVER WARMUP")
+        print("=" * 70)
+        print(
+            "Sending all control, Shakespeare, and random prompts to warm up the server..."
+        )
+
+        # Warmup with all control sequences
+        print("Warming up with control sequences...")
+        for i, control_seq in enumerate(self.control_sequences):
+            print(f"  Warmup control sequence {i + 1}: {control_seq[:50]}...")
+            try:
+                self.make_request(control_seq)
+            except Exception as e:
+                print(f"  Warning: Warmup request failed: {e}")
+
+        # Warmup with Shakespeare sequences that will be used in testing
+        print("Warming up with Shakespeare sequences...")
+        shakespeare_count = self.max_iterations // self.shakespeare_interval
+        for seq_idx in range(1, shakespeare_count + 1):
+            start_word = (seq_idx - 1) * self.word_count
+            content = self.get_shakespeare_content(start_word)
+
+            if content:
+                print(
+                    f"  Warmup Shakespeare sequence {seq_idx} (words {start_word}-{start_word + self.word_count - 1})..."
+                )
+                try:
+                    self.make_request(content)
+                except Exception as e:
+                    print(f"  Warning: Warmup request failed: {e}")
+
+        # Warmup with all random sequences
+        print("Warming up with random sequences...")
+        for i, random_seq in enumerate(self.random_sequences):
+            print(f"  Warmup random sequence {i + 1}: {random_seq[:50]}...")
+            try:
+                self.make_request(random_seq)
+            except Exception as e:
+                print(f"  Warning: Warmup request failed: {e}")
+
+        print("Server warmup completed!")
+        print("=" * 70)
+
+    def get_shakespeare_content(self, start_word: int) -> str:
+        """Get Shakespeare content starting from a specific word."""
+        with open(self.shakespeare_file, "r", encoding="utf-8") as f:
+            words = f.read().split()
+
+        end_word = min(start_word + self.word_count, len(words))
+        return " ".join(words[start_word:end_word])
+
+    def download_ifeval_dataset(self) -> List[str]:
+        """Download and extract all prompts from IFEval dataset."""
+        try:
+            from datasets import load_dataset
+
+            print("Loading complete IFEval dataset...")
+            dataset = load_dataset("google/IFEval", split="train")
+
+            # Extract all prompts from the dataset
+            prompts = []
+
+            for example in dataset:
+                # IFEval has 'prompt' field with the instruction
+                if "prompt" in example:
+                    prompt_text = example["prompt"].strip()
+                    if prompt_text:  # Only skip empty prompts
+                        prompts.append(prompt_text)
+
+            print(f"Loaded {len(prompts)} prompts from complete IFEval dataset")
+            return prompts
+
+        except ImportError:
+            print(
+                "Warning: datasets library not available, falling back to default prompts"
+            )
+            return self.control_sequences + self.random_sequences
+        except Exception as e:
+            print(
+                f"Warning: Failed to load IFEval dataset ({e}), falling back to default prompts"
+            )
+            return self.control_sequences + self.random_sequences
+
+    def run_test_iterations(self):
+        """Run the test iterations with comprehensive warmup."""
+        # Perform initial warmup before testing
+        self.warmup_server()
+
+        for iteration in range(1, self.max_iterations + 1):
+            print(f"Iteration {iteration}/{self.max_iterations}")
+
+            # Control sequence test
+            if iteration % self.control_interval == 0:
+                control_idx = (iteration // self.control_interval - 1) % len(
+                    self.control_sequences
+                )
+                control_content = self.control_sequences[control_idx]
+
+                print(
+                    f"  Running control sequence {control_idx + 1}: {control_content[:50]}..."
+                )
+                completion = self.make_request(control_content)
+                self.control_responses[control_idx].append(completion)
+                print(f"  Response: {completion}")
+
+            # Shakespeare sequence test
+            if iteration % self.shakespeare_interval == 0:
+                start_word = (
+                    iteration // self.shakespeare_interval - 1
+                ) * self.word_count
+                content = self.get_shakespeare_content(start_word)
+
+                if content:
+                    shakespeare_idx = iteration // self.shakespeare_interval - 1
+                    print(
+                        f"  Running Shakespeare sequence {shakespeare_idx + 1} (words {start_word}-{start_word + self.word_count - 1})..."
+                    )
+                    completion = self.make_request(content)
+                    self.shakespeare_responses[shakespeare_idx].append(completion)
+                    print(f"  Response: {completion}")
+
+            # Random sequence test
+            if iteration % self.random_interval == 0:
+                random_idx = (iteration // self.random_interval - 1) % len(
+                    self.random_sequences
+                )
+                random_content = self.random_sequences[random_idx]
+
+                print(
+                    f"  Running random sequence {random_idx + 1}: {random_content[:50]}..."
+                )
+                completion = self.make_request(random_content)
+                self.random_responses[random_idx].append(completion)
+                print(f"  Response: {completion}")
+
+    def analyze_responses(
+        self, responses: Dict[int, List[str]], sequence_type: str
+    ) -> Tuple[int, int]:
+        """Analyze responses for determinism."""
+        passed = 0
+        failed = 0
+
+        print(f"\n=== {sequence_type.upper()} SEQUENCES ===")
+
+        for idx, response_list in responses.items():
+            if not response_list:
+                continue
+
+            print(f"\n{sequence_type} sequence {idx + 1}:")
+            print(f"Total responses: {len(response_list)}")
+
+            if len(response_list) == 1:
+                print("Single response - cannot check determinism")
+                continue
+
+            reference = response_list[0]
+            differences = 0
+
+            print(f"Reference response: {reference}")
+
+            for i, response in enumerate(response_list[1:], 2):
+                if response == reference:
+                    print(f"Response {i}: MATCHES reference")
+                else:
+                    print(f"Response {i}: DIFFERS from reference")
+                    print(f"  Expected: {reference}")
+                    print(f"  Got:      {response}")
+                    differences += 1
+
+            if differences == 0:
+                print(" ALL RESPONSES IDENTICAL - DETERMINISTIC")
+                passed += 1
+            else:
+                print(f" {differences} DIFFERENCES DETECTED - NON-DETERMINISTIC")
+                failed += 1
+
+        return passed, failed
+
+    def test_concurrent_determinism(
+        self, prompts: List[str], num_workers: int = 4, requests_per_prompt: int = 3
+    ) -> bool:
+        """Test determinism with concurrent requests to the same prompts."""
+        print("\n=== CONCURRENT DETERMINISM TEST ===")
+        print(f"Workers: {num_workers}, Requests per prompt: {requests_per_prompt}")
+
+        # Prepare test data: each prompt will get multiple concurrent requests
+        test_tasks = []
+        for i, prompt in enumerate(prompts):
+            for req_num in range(requests_per_prompt):
+                test_tasks.append(
+                    {
+                        "prompt_idx": i,
+                        "prompt": prompt,
+                        "request_id": f"p{i}_r{req_num}",
+                    }
+                )
+
+        print(f"Total concurrent requests: {len(test_tasks)}")
+
+        # Storage for responses grouped by prompt
+        concurrent_responses: Dict[int, List[Tuple[str, str]]] = defaultdict(list)
+
+        def make_concurrent_request(task):
+            """Worker function for concurrent requests."""
+            try:
+                response = self.make_request(task["prompt"])
+                return {
+                    "prompt_idx": task["prompt_idx"],
+                    "request_id": task["request_id"],
+                    "response": response,
+                    "success": True,
+                    "error": None,
+                }
+            except Exception as e:
+                return {
+                    "prompt_idx": task["prompt_idx"],
+                    "request_id": task["request_id"],
+                    "response": None,
+                    "success": False,
+                    "error": str(e),
+                }
+
+        # Execute concurrent requests
+        print("Executing concurrent requests...")
+        start_time = time.time()
+
+        with ThreadPoolExecutor(max_workers=num_workers) as executor:
+            # Submit all tasks
+            future_to_task = {
+                executor.submit(make_concurrent_request, task): task
+                for task in test_tasks
+            }
+
+            # Collect results
+            completed = 0
+            failed = 0
+            for future in as_completed(future_to_task):
+                result = future.result()
+                completed += 1
+
+                if result["success"]:
+                    concurrent_responses[result["prompt_idx"]].append(
+                        (result["request_id"], result["response"])
+                    )
+                    if completed % 10 == 0:
+                        print(f"  Completed: {completed}/{len(test_tasks)}")
+                else:
+                    failed += 1
+                    print(f"  Failed request {result['request_id']}: {result['error']}")
+
+        elapsed = time.time() - start_time
+        print(
+            f"Completed {completed} requests in {elapsed:.2f}s ({completed/elapsed:.1f} req/s)"
+        )
+        print(f"Failed requests: {failed}")
+
+        # Analyze concurrent determinism
+        print("\n=== CONCURRENT DETERMINISM ANALYSIS ===")
+        total_prompts_tested = 0
+        deterministic_prompts = 0
+
+        for prompt_idx, responses in concurrent_responses.items():
+            if len(responses) < 2:
+                print(
+                    f"Prompt {prompt_idx}: Only {len(responses)} response(s), skipping"
+                )
+                continue
+
+            total_prompts_tested += 1
+            prompt_text = prompts[prompt_idx]
+            print(f"\nPrompt {prompt_idx}: {prompt_text[:50]}...")
+            print(f"Concurrent responses: {len(responses)}")
+
+            # Extract just the response text
+            response_texts = [resp[1] for resp in responses]
+            request_ids = [resp[0] for resp in responses]
+
+            # Check if all responses are identical
+            reference_response = response_texts[0]
+            mismatches = []
+
+            for req_id, response_text in zip(request_ids[1:], response_texts[1:]):
+                if response_text != reference_response:
+                    mismatches.append((req_id, response_text))
+
+            if not mismatches:
+                print(
+                    f"   DETERMINISTIC: All {len(responses)} concurrent responses identical"
+                )
+                print(f"     Response: {reference_response}")
+                deterministic_prompts += 1
+            else:
+                print(f"    NON-DETERMINISTIC: {len(mismatches)} different responses")
+                print(f"    Reference ({request_ids[0]}): {reference_response}")
+                for req_id, diff_response in mismatches:
+                    print(f"     Different ({req_id}): {diff_response}")
+
+        # Final assessment
+        success_rate = (
+            deterministic_prompts / total_prompts_tested
+            if total_prompts_tested > 0
+            else 0
+        )
+        print("\n=== FINAL CONCURRENT DETERMINISM RESULT ===")
+        print(f"Prompts tested: {total_prompts_tested}")
+        print(f"Deterministic: {deterministic_prompts}")
+        print(f"Non-deterministic: {total_prompts_tested - deterministic_prompts}")
+        print(f"Success rate: {success_rate:.1%}")
+        print(f"Concurrency level: {num_workers} workers")
+        print(f"Request rate: {completed/elapsed:.1f} req/s")
+
+        return success_rate == 1.0
+
+
+@pytest.fixture(scope="function")
+def vllm_server(request, runtime_services):
+    """Start and stop vLLM server for each test with optional cache block overrides.
+
+    To parametrize, use:
+      @pytest.mark.parametrize("vllm_server", [{"cpu_blocks": 10000, "gpu_blocks": 2048}], indirect=True)
+    """
+    logger = logging.getLogger("pytest")
+    logger.setLevel(logging.INFO)
+
+    cpu_blocks = getattr(request, "param", {}).get("cpu_blocks", None)
+    gpu_blocks = getattr(request, "param", {}).get("gpu_blocks", None)
+    port = getattr(request, "param", {}).get("port", None)
+
+    # Put logs in the per-test directory set up by tests/conftest.py
+    log_dir = Path(request.node.name)
+
+    server_manager = VLLMServerManager(
+        port=port,
+        cpu_cache_blocks=cpu_blocks,
+        gpu_cache_blocks=gpu_blocks,
+        log_dir=log_dir,
+    )
+
+    start_timeout = int(os.environ.get("KVBM_VLLM_START_TIMEOUT", "300"))
+    if not server_manager.start_server(timeout=start_timeout):
+        pytest.fail(
+            f"Failed to start vLLM server (cpu_blocks={cpu_blocks}, gpu_blocks={gpu_blocks}, port={server_manager.port})"
+        )
+
+    yield server_manager
+
+    server_manager.stop_server()
+
+
+@pytest.fixture(scope="function")
+def tester(vllm_server):
+    """Create determinism tester bound to the running server's base URL."""
+    t = DeterminismTester(base_url=vllm_server.base_url)
+    t.download_shakespeare_text()
+    return t
+
+
+class TestDeterminism:
+    """Test class for determinism validation."""
+
+    @pytest.mark.parametrize(
+        "vllm_server",
+        [
+            {"cpu_blocks": int(os.environ.get("KVBM_CPU_BLOCKS", "10000"))},
+        ],
+        indirect=True,
+    )
+    def test_determinism_with_cache_reset(self, tester, vllm_server, runtime_services):
+        """Test determinism across cache reset: run test with warmup, reset cache, run again without warmup."""
+        print("\n" + "=" * 70)
+        print("STARTING DETERMINISM TEST (WITH CACHE RESET)")
+        print("=" * 70)
+
+        # Phase 1: Run test with warmup
+        print("\n=== PHASE 1: BEFORE CACHE RESET (WITH WARMUP) ===")
+        tester.run_test_iterations()
+
+        # Store Phase 1 results
+        phase1_control = {k: v.copy() for k, v in tester.control_responses.items()}
+        phase1_shakespeare = {
+            k: v.copy() for k, v in tester.shakespeare_responses.items()
+        }
+        phase1_random = {k: v.copy() for k, v in tester.random_responses.items()}
+
+        # Reset cache
+        print("\n" + "=" * 50)
+        print("RESETTING CACHE")
+        print("=" * 50)
+        tester.reset_prefix_cache()
+
+        # Clear response storage for Phase 2 (they are defaultdict, so they'll auto-initialize)
+        tester.control_responses.clear()
+        tester.shakespeare_responses.clear()
+        tester.random_responses.clear()
+
+        # Phase 2: Run test without warmup
+        print("\n=== PHASE 2: AFTER CACHE RESET (NO WARMUP) ===")
+        # Temporarily disable warmup by modifying the method
+        original_warmup = tester.warmup_server
+        tester.warmup_server = lambda: print(
+            "Skipping warmup (testing determinism across cache reset)"
+        )
+
+        try:
+            tester.run_test_iterations()
+        finally:
+            # Restore original warmup method
+            tester.warmup_server = original_warmup
+
+        # Compare Phase 1 vs Phase 2 results
+        print("\n" + "=" * 70)
+        print("CROSS-CACHE-RESET DETERMINISM ANALYSIS")
+        print("=" * 70)
+
+        total_passed = 0
+        total_failed = 0
+
+        # Compare control sequences
+        for seq_idx in phase1_control:
+            if seq_idx in tester.control_responses:
+                phase1_responses = phase1_control[seq_idx]
+                phase2_responses = tester.control_responses[seq_idx]
+
+                min_responses = min(len(phase1_responses), len(phase2_responses))
+                for i in range(min_responses):
+                    if phase1_responses[i] == phase2_responses[i]:
+                        total_passed += 1
+                        print(f"   Control {seq_idx}, response {i}: DETERMINISTIC")
+                    else:
+                        total_failed += 1
+                        print(f"   Control {seq_idx}, response {i}: NON-DETERMINISTIC")
+                        print(f"     Before: {phase1_responses[i]}")
+                        print(f"     After:  {phase2_responses[i]}")
+
+        # Compare Shakespeare sequences
+        for seq_idx in phase1_shakespeare:
+            if seq_idx in tester.shakespeare_responses:
+                phase1_responses = phase1_shakespeare[seq_idx]
+                phase2_responses = tester.shakespeare_responses[seq_idx]
+
+                min_responses = min(len(phase1_responses), len(phase2_responses))
+                for i in range(min_responses):
+                    if phase1_responses[i] == phase2_responses[i]:
+                        total_passed += 1
+                        print(f"   Shakespeare {seq_idx}, response {i}: DETERMINISTIC")
+                    else:
+                        total_failed += 1
+                        print(
+                            f"   Shakespeare {seq_idx}, response {i}: NON-DETERMINISTIC"
+                        )
+                        print(f"     Before: {phase1_responses[i]}")
+                        print(f"     After:  {phase2_responses[i]}")
+
+        # Compare random sequences
+        for seq_idx in phase1_random:
+            if seq_idx in tester.random_responses:
+                phase1_responses = phase1_random[seq_idx]
+                phase2_responses = tester.random_responses[seq_idx]
+
+                min_responses = min(len(phase1_responses), len(phase2_responses))
+                for i in range(min_responses):
+                    if phase1_responses[i] == phase2_responses[i]:
+                        total_passed += 1
+                        print(f"   Random {seq_idx}, response {i}: DETERMINISTIC")
+                    else:
+                        total_failed += 1
+                        print(f"   Random {seq_idx}, response {i}: NON-DETERMINISTIC")
+                        print(f"     Before: {phase1_responses[i]}")
+                        print(f"     After:  {phase2_responses[i]}")
+
+        # Final assessment
+        print("\n" + "=" * 70)
+        print("FINAL CROSS-CACHE-RESET DETERMINISM ASSESSMENT")
+        print("=" * 70)
+        print(f"Total comparisons: {total_passed + total_failed}")
+        print(f"Passed (deterministic): {total_passed}")
+        print(f"Failed (non-deterministic): {total_failed}")
+        print(
+            "Test compared responses before cache reset (with warmup) vs after cache reset (no warmup)."
+        )
+
+        if total_passed + total_failed == 0:
+            pytest.skip("No tests were completed - insufficient data")
+
+        assert (
+            total_failed == 0
+        ), f"Model is not deterministic across cache reset: {total_failed} comparisons failed"
+
+    @pytest.mark.parametrize(
+        "vllm_server",
+        [
+            {"cpu_blocks": int(os.environ.get("KVBM_CPU_BLOCKS", "20000"))},
+        ],
+        indirect=True,
+    )
+    @pytest.mark.parametrize(
+        "num_concurrent",
+        [int(x) for x in os.environ.get("KVBM_CONCURRENT_REQUESTS", "3").split(",")],
+    )
+    @pytest.mark.parametrize(
+        "max_tokens",
+        [int(x) for x in os.environ.get("KVBM_MAX_TOKENS", "10").split(",")],
+    )
+    @pytest.mark.parametrize(
+        "num_prompts",
+        [int(x) for x in os.environ.get("KVBM_IFEVAL_PROMPTS", "120").split(",")],
+    )
+    def test_concurrent_determinism_with_ifeval(
+        self,
+        tester,
+        vllm_server,
+        runtime_services,
+        num_concurrent,
+        max_tokens,
+        num_prompts,
+    ):
+        """Simple concurrent determinism test: send IFEval prompts concurrently, with cache reset."""
+        print("\n" + "=" * 70)
+        print("CONCURRENT DETERMINISM TEST WITH IFEVAL")
+        print("=" * 70)
+
+        # Override max_tokens for this test iteration
+        original_max_tokens = os.environ.get("KVBM_MAX_TOKENS")
+        os.environ["KVBM_MAX_TOKENS"] = str(max_tokens)
+        print(
+            f"Using KVBM_MAX_TOKENS={max_tokens} (parametrized, original: {original_max_tokens or '48'})"
+        )
+
+        # Configuration comes from parametrize
+        print(
+            f"Configuration: {num_concurrent} concurrent requests, {max_tokens} max tokens"
+        )
+
+        # Load IFEval prompts
+        ifeval_prompts = tester.download_ifeval_dataset()
+        if not ifeval_prompts:
+            pytest.skip("IFEval dataset not available")
+
+        # Use parametrized number of IFEval prompts
+        test_prompts = ifeval_prompts[:num_prompts]
+        print(
+            f"Using {len(test_prompts)} IFEval prompts for concurrent testing (parametrized: {num_prompts})"
+        )
+        print(f"Concurrency level: {num_concurrent} simultaneous requests")
+
+        # Show sample prompts
+        print("\nSample prompts:")
+        for i, prompt in enumerate(test_prompts[:3]):
+            print(f"  {i+1}. {prompt[:80]}{'...' if len(prompt) > 80 else ''}")
+        if len(test_prompts) > 3:
+            print(f"  ... and {len(test_prompts) - 3} more")
+
+        def run_concurrent_test(phase_name, do_warmup=False):
+            """Run one phase of concurrent testing."""
+            print(f"\n=== {phase_name} ===")
+
+            if do_warmup:
+                # KV Cache warmup - send ALL test prompts to compute KV caches
+                print(
+                    f"Warming up KV caches with all {len(test_prompts)} test prompts..."
+                )
+                warmup_failed = 0
+
+                for i, prompt in enumerate(test_prompts):
+                    if (
+                        i % 5 == 0 or i == len(test_prompts) - 1
+                    ):  # Progress every 5 prompts
+                        print(f"  Warmup progress: {i+1}/{len(test_prompts)}")
+
+                    try:
+                        tester.make_request(prompt)
+                    except Exception as e:
+                        warmup_failed += 1
+                        if warmup_failed <= 3:  # Show first few failures
+                            print(f"    Warmup failed for prompt {i}: {e}")
+
+                if warmup_failed > 0:
+                    print(
+                        f"Warmup completed with {warmup_failed} failures out of {len(test_prompts)} prompts"
+                    )
+                else:
+                    print(
+                        f"Warmup completed successfully - all {len(test_prompts)} KV caches computed"
+                    )
+
+                # Wait for 10 seconds to make sure all transfers are complete
+                time.sleep(10)
+                # Reset cache
+                print("\n" + "=" * 50)
+                print("RESETTING CACHE AFTER WARMUP")
+                print("=" * 50)
+                tester.reset_prefix_cache()
+                time.sleep(10)
+            else:
+                print("Skipping warmup (already done in previous phase)")
+
+            # Run concurrent requests
+            print(
+                f"Sending {len(test_prompts)} requests with {num_concurrent} max concurrent..."
+            )
+            start_time = time.time()
+
+            def make_request_wrapper(prompt_and_idx):
+                idx, prompt = prompt_and_idx
+                try:
+                    response = tester.make_request(prompt)
+                    return {
+                        "idx": idx,
+                        "prompt": prompt,
+                        "response": response,
+                        "success": True,
+                    }
+                except Exception as e:
+                    return {
+                        "idx": idx,
+                        "prompt": prompt,
+                        "error": str(e),
+                        "success": False,
+                    }
+
+            # Execute all requests concurrently
+            with ThreadPoolExecutor(max_workers=num_concurrent) as executor:
+                results = list(
+                    executor.map(make_request_wrapper, enumerate(test_prompts))
+                )
+
+            elapsed = time.time() - start_time
+            successful = [r for r in results if r["success"]]
+            failed = [r for r in results if not r["success"]]
+
+            print(
+                f"Completed in {elapsed:.2f}s - Success: {len(successful)}, Failed: {len(failed)}"
+            )
+
+            if failed:
+                for fail in failed[:3]:  # Show first few failures
+                    print(f"  Failed: {fail['error']}")
+
+            return successful
+
+        # Phase 1: Before cache reset
+        results_before = run_concurrent_test(
+            "PHASE 1: BEFORE CACHE RESET", do_warmup=True
+        )
+
+        # Reset cache
+        print("\n" + "=" * 50)
+        print("RESETTING CACHE")
+        print("=" * 50)
+        tester.reset_prefix_cache()
+
+        # Phase 2: After cache reset
+        results_after = run_concurrent_test("PHASE 2: AFTER CACHE RESET")
+
+        # Compare results between phases
+        print("\n" + "=" * 70)
+        print("DETERMINISM ANALYSIS")
+        print("=" * 70)
+
+        # Create lookup for before results
+        before_responses = {r["idx"]: r["response"] for r in results_before}
+        after_responses = {r["idx"]: r["response"] for r in results_after}
+
+        deterministic_count = 0
+        total_compared = 0
+
+        for idx in before_responses:
+            if idx in after_responses:
+                total_compared += 1
+                before_resp = before_responses[idx]
+                after_resp = after_responses[idx]
+
+                if before_resp == after_resp:
+                    deterministic_count += 1
+                    print(f"   Prompt {idx}: DETERMINISTIC")
+                else:
+                    print(f"   Prompt {idx}: NON-DETERMINISTIC")
+                    print(f"     Before: {before_resp}")
+                    print(f"     After:  {after_resp}")
+
+        # Final assessment
+        success_rate = deterministic_count / total_compared if total_compared > 0 else 0
+        print("\n=== FINAL RESULT ===")
+        print(f"Prompts compared: {total_compared}")
+        print(f"Deterministic: {deterministic_count}")
+        print(f"Success rate: {success_rate:.1%}")
+        print(f"Concurrent requests: {num_concurrent}")
+
+        # Restore original max_tokens setting
+        if original_max_tokens is not None:
+            os.environ["KVBM_MAX_TOKENS"] = original_max_tokens
+        else:
+            os.environ.pop("KVBM_MAX_TOKENS", None)
+
+        assert (
+            success_rate == 1.0
+        ), f"Determinism failed: {deterministic_count}/{total_compared} prompts deterministic"
+
+
+if __name__ == "__main__":
+    # Allow running as script
+    pytest.main([__file__, "-v", "-s"])