diff --git a/megatron/core/inference/config.py b/megatron/core/inference/config.py
index a334d58c1ac..4063ffbc977 100644
--- a/megatron/core/inference/config.py
+++ b/megatron/core/inference/config.py
@@ -260,6 +260,12 @@ class InferenceConfig:
     Only applies when enable_prefix_caching is True and using a coordinator.
     """
 
+    prefix_caching_routing_alpha: float = 0.5
+    """Weight for prefix-aware scoring: score = alpha * match + (1 - alpha) * normalized_load.
+    Higher alpha favors prefix cache hits; lower alpha favors load balance.
+    Must be in [0, 1]. Only applies when enable_prefix_caching is True and using a coordinator.
+    """
+
     prefix_caching_mamba_gb: Optional[float] = None
     """GPU memory budget (in GB) for the Mamba state cache used by prefix caching
     on hybrid models. Each cache slot stores SSM and conv states for all Mamba layers
@@ -298,7 +304,14 @@ class InferenceConfig:
     """
 
     use_synchronous_zmq_collectives: bool = False
-    """Whether to use synchronous ZMQ collectives for inference. If True, the 
-    all_reduce_max operation will be performed synchronously, which can help reduce 
+    """Whether to use synchronous ZMQ collectives for inference. If True, the
+    all_reduce_max operation will be performed synchronously, which can help reduce
     performance variability for MoEs.
     """
+
+    def __post_init__(self):
+        if not (0.0 <= self.prefix_caching_routing_alpha <= 1.0):
+            raise ValueError(
+                f"prefix_caching_routing_alpha must be in [0, 1], "
+                f"got {self.prefix_caching_routing_alpha}"
+            )
diff --git a/megatron/core/inference/contexts/dynamic_context.py b/megatron/core/inference/contexts/dynamic_context.py
index 5a1dbb5cf97..1117f2b9c4b 100644
--- a/megatron/core/inference/contexts/dynamic_context.py
+++ b/megatron/core/inference/contexts/dynamic_context.py
@@ -252,10 +252,11 @@ def __init__(self, model_config: TransformerConfig, inference_config: InferenceC
         self.enable_prefix_caching = inference_config.enable_prefix_caching
         self.prefix_caching_eviction_policy = inference_config.prefix_caching_eviction_policy
         self.prefix_caching_coordinator_policy = inference_config.prefix_caching_coordinator_policy
-        # Engine step counter (used for logging, metrics, and event tracking)
-        self.step_count = 0
 
-        # Separate monotonic clock for prefix caching LRU eviction ordering.
+        # Hyperparameter for choosing to prioritize prefix hit matches vs minimizing idle load
+        self.prefix_caching_routing_alpha = inference_config.prefix_caching_routing_alpha
+
+        # Monotonic clock for prefix caching LRU eviction ordering.
         # Incremented each engine step but kept independent so the engine step
         # counter is not overloaded with cache-eviction semantics.
         self.prefix_cache_lru_clock = 0
@@ -264,6 +265,9 @@ def __init__(self, model_config: TransformerConfig, inference_config: InferenceC
         self.prefix_cache_hits = 0  # requests that matched at least one cached block
         self.prefix_cache_blocks_matched = 0  # total matched blocks across all requests
 
+        # Engine step counter (used for logging, metrics, and event tracking)
+        self.step_count = 0
+
         self.cache_mla_latent = (
             isinstance(model_config, MLATransformerConfig) and model_config.cache_mla_latents
         )
diff --git a/megatron/core/inference/data_parallel_inference_coordinator.py b/megatron/core/inference/data_parallel_inference_coordinator.py
index fab9b4acfe6..50f586cc598 100644
--- a/megatron/core/inference/data_parallel_inference_coordinator.py
+++ b/megatron/core/inference/data_parallel_inference_coordinator.py
@@ -11,6 +11,7 @@
 from multiprocessing import Event
 from multiprocessing.connection import Connection
 
+import numpy as np
 import torch
 
 from megatron.core.inference.config import PrefixCachingCoordinatorPolicy
@@ -87,6 +88,7 @@ def __init__(
         pipe_connection: Connection,
         data_parallel_size: int,
         tokenizer,
+        max_requests,
         inference_coordinator_port: int | None = None,
         deterministic_mode: bool = False,
         block_size_tokens: int | None = None,
@@ -94,6 +96,7 @@ def __init__(
         prefix_caching_coordinator_policy: PrefixCachingCoordinatorPolicy = (
             PrefixCachingCoordinatorPolicy.FIRST_PREFIX_BLOCK
         ),
+        prefix_caching_routing_alpha: float = 0.5,
         schedule_output_path: str | None = None,
         hostname: str | None = None,
     ):
@@ -110,6 +113,10 @@ def __init__(
                 expected to connect.
             tokenizer: The tokenizer to use for prompt tokenization and detokenization.
             inference_coordinator_port (Optional[int]): The TCP port number to bind the server to.
+            prefix_caching_routing_alpha (float): Weight for prefix-aware routing score:
+                score = alpha * match + (1 - alpha) * normalized_load.
+            max_requests (int): Max concurrent requests per rank, used to
+                compute normalized_load for prefix-aware scoring.
         """
         assert HAVE_ZMQ, (
             "please install the pyzmq library to use DataParallelInferenceCoordinator\n"
@@ -179,6 +186,7 @@ def __init__(
 
         self.request_id_to_client_id = {}
         self.request_id_to_client_request_id = {}
+        self.request_id_to_rank = {}  # Maps request_id → rank identity for pending count tracking
 
         self.next_request_id = 0
         self.tokenizer = tokenizer
@@ -188,8 +196,9 @@ def __init__(
         self.block_size_tokens = block_size_tokens
         self.enable_prefix_caching = enable_prefix_caching
         self.prefix_caching_coordinator_policy = prefix_caching_coordinator_policy
-        self.hash_to_rank_info = {}  # Dict[int, Dict[bytes, int]]: hash → {rank → timestamp}
-        self._assignment_counter = 0
+        self.prefix_caching_routing_alpha = prefix_caching_routing_alpha
+        self.max_requests = max_requests
+        assert self.max_requests is not None and self.max_requests > 0
 
         # Schedule recording.
         self.schedule_output_path = schedule_output_path
@@ -201,6 +210,17 @@ def __init__(
             identity: idx for idx, identity in enumerate(sorted_identities)
         }
 
+        # Numpy arrays for vectorized scoring (indexed by rank index).
+        n_ranks = len(sorted_identities)
+        self._identities_list = list(sorted_identities)  # rank_index → identity
+        self._pending_counts = np.zeros(n_ranks, dtype=np.int32)
+
+        # Hash → {rank_idx: timestamp} dict for prefix cache affinity routing.
+        # Each key is a block hash; each value maps rank indices to assignment
+        # timestamps (positive int).  Missing entries are implicitly zero.
+        self._hash_table: dict[int, dict[int, int]] = {}
+        self._hash_assignment_counter = 0
+
     def get_next_data_parallel_rank(self):
         """
         Selects the next data parallel rank using round-robin scheduling.
@@ -215,6 +235,26 @@ def get_next_data_parallel_rank(self):
         self._round_robin_idx = idx + 1
         return identities[idx]
 
+    def _register_rank_identity(self, identity):
+        """Register a new rank identity in the scoring data structures.
+
+        Called when a rank dynamically connects to a running coordinator
+        (e.g. in tests that spawn the coordinator with data_parallel_size=0
+        and let engines register after the fact).
+        """
+        if identity in self.identity_to_rank_index:
+            return
+        new_idx = len(self._identities_list)
+        self.identity_to_rank_index[identity] = new_idx
+        self._identities_list.append(identity)
+        self._pending_counts = np.append(self._pending_counts, np.int32(0))
+        logging.info(
+            "Coordinator: registered engine %s as rank index %d (now %d engines)",
+            identity,
+            new_idx,
+            len(self._identities_list),
+        )
+
     def _remove_engine(self, identity):
         """Remove a disconnected engine from the routing pool."""
         self.identities_of_data_parallel_ranks.remove(identity)
@@ -258,14 +298,13 @@ def compute_request_hashes(self, prompt):
         return compute_block_hashes_batched(token_tensor, self.block_size_tokens)
 
     def get_best_data_parallel_rank(self, request_hashes):
-        """Select the best DP rank based on prefix cache affinity.
+        """Select the best DP rank based on prefix cache affinity and load.
 
-        Iterates request hashes in reverse order and picks the rank that cached
-        the longest matching prefix (the furthest hash found). Since hashes are
-        parent-chained, finding hash[i] in a rank guarantees hash[0..i-1] are
-        also present. Among ranks that share the longest match, the most recently
-        assigned rank (highest timestamp) is preferred. Falls back to round-robin
-        when no rank matches.
+        Uses a scoring function: score = alpha * match + (1 - alpha) * normalized_load
+        where *match* is a policy-dependent affinity score in [0, 1] (binary for
+        ``first_prefix_block``, normalized prefix depth for ``longest_prefix``)
+        and normalized_load = free_slots / max_requests (higher means more free
+        capacity).
 
         Args:
             request_hashes: List of block hashes for the request.
@@ -273,22 +312,25 @@ def get_best_data_parallel_rank(self, request_hashes):
         Returns:
             bytes: The ZMQ identity of the selected data parallel rank.
         """
-        if (
-            not self.enable_prefix_caching
-            or not request_hashes
-            or self.prefix_caching_coordinator_policy == PrefixCachingCoordinatorPolicy.ROUND_ROBIN
-        ):
+        if self.prefix_caching_coordinator_policy == PrefixCachingCoordinatorPolicy.ROUND_ROBIN:
+            return self.get_next_data_parallel_rank()
+
+        if not self.enable_prefix_caching or not request_hashes:
             return self.get_next_data_parallel_rank()
 
-        # Reverse scan: first match is the longest prefix (parent-chained hashes).
-        for h in reversed(request_hashes):
-            rank_info = self.hash_to_rank_info.get(h)
-            if rank_info:
-                # Pick the most recently assigned rank.
-                best_rank = max(rank_info, key=rank_info.get)
-                return best_rank
+        match, recency = self._match_vector(request_hashes)
+
+        alpha = self.prefix_caching_routing_alpha
 
-        return self.get_next_data_parallel_rank()
+        # Vectorized score: alpha * match + (1-alpha) * free_capacity_fraction.
+        free_slots = np.maximum(0, self.max_requests - self._pending_counts).astype(np.float64)
+        scores = alpha * match + (1.0 - alpha) * (free_slots / self.max_requests)
+
+        # Tiebreak: highest score, then highest recency, then lowest rank index.
+        n_ranks = len(self._identities_list)
+        order = np.lexsort((np.arange(n_ranks), -recency, -scores))
+        best_idx = int(order[0])
+        return self._identities_list[best_idx]
 
     def _update_rank_hashes(self, rank_identity, request_hashes):
         """Record that a rank owns the given hashes.
@@ -297,10 +339,39 @@ def _update_rank_hashes(self, rank_identity, request_hashes):
             rank_identity: ZMQ identity of the target rank.
             request_hashes: List of block hashes assigned to this rank.
         """
-        self._assignment_counter += 1
-        ts = self._assignment_counter
+        rank_idx = self.identity_to_rank_index[rank_identity]
+        self._hash_assignment_counter += 1
+        ts = self._hash_assignment_counter
         for h in request_hashes:
-            self.hash_to_rank_info.setdefault(h, {})[rank_identity] = ts
+            self._hash_table.setdefault(h, {})[rank_idx] = ts
+
+    def _match_vector(self, hashes):
+        """Return ``(match, recency)`` vectors of shape ``(n_ranks,)``.
+
+        *match* is binary depth: ``(depth + 1) / len(hashes)`` for ranks that
+        have the deepest cached block, 0 otherwise.  *recency* is the raw
+        assignment timestamp for each matching rank (0 for non-matching ranks).
+
+        For ``FIRST_PREFIX_BLOCK`` the caller already truncates *hashes* to a
+        single element, so the same logic yields a binary 0/1 match score.
+        """
+        n_ranks = len(self._identities_list)
+        n = len(hashes)
+        zeros = np.zeros(n_ranks, dtype=np.float64)
+        if n == 0:
+            return zeros, zeros.copy()
+        for i in range(n - 1, -1, -1):
+            row = self._hash_table.get(hashes[i])
+            if row is None:
+                continue
+            rank_idxs = np.fromiter(row.keys(), dtype=np.intp)
+            present = np.zeros(n_ranks, dtype=bool)
+            present[rank_idxs] = True
+            recency = np.zeros(n_ranks, dtype=np.float64)
+            recency[rank_idxs] = np.fromiter(row.values(), dtype=np.float64)
+            if present.any():
+                return present.astype(np.float64) * ((i + 1.0) / n), recency
+        return zeros, zeros.copy()
 
     def start(self):
         """
@@ -321,6 +392,7 @@ def start(self):
             if serialized_payload == b"":
                 if sender_identity not in self.identities_of_data_parallel_ranks:
                     self.identities_of_data_parallel_ranks.append(sender_identity)
+                    self._register_rank_identity(sender_identity)
                 continue
 
             deserialized_payload = msgpack.unpackb(serialized_payload, raw=False)
@@ -392,6 +464,8 @@ def start(self):
                     del self.request_id_to_client_request_id[request_id]
                     return
 
+                self.request_id_to_rank[request_id] = next_identity
+                self._pending_counts[self.identity_to_rank_index[next_identity]] += 1
                 if request_hashes:
                     self._update_rank_hashes(next_identity, request_hashes)
                 if self.schedule_records is not None:
@@ -471,6 +545,12 @@ def start(self):
                     client_request_identity = self.request_id_to_client_request_id[fid]
                     del self.request_id_to_client_id[fid]
                     del self.request_id_to_client_request_id[fid]
+                    assigned_rank = self.request_id_to_rank.pop(fid, None)
+                    if assigned_rank is not None:
+                        idx = self.identity_to_rank_index.get(assigned_rank)
+                        if idx is not None:
+                            assert self._pending_counts[idx] >= 1
+                            self._pending_counts[idx] -= 1
 
                     self.router_socket.send_multipart(
                         [
@@ -526,6 +606,7 @@ def entrypoint(
         ready_event: Event,
         data_parallel_size: int,
         tokenizer,
+        max_requests,
         inference_coordinator_port: int | None = None,
         deterministic_mode: bool = False,
         block_size_tokens: int | None = None,
@@ -533,6 +614,7 @@ def entrypoint(
         prefix_caching_coordinator_policy: PrefixCachingCoordinatorPolicy = (
             PrefixCachingCoordinatorPolicy.FIRST_PREFIX_BLOCK
         ),
+        prefix_caching_routing_alpha: float = 0.5,
         schedule_output_path: str | None = None,
         hostname: str | None = None,
     ):
@@ -553,16 +635,20 @@ def entrypoint(
             enable_prefix_caching (bool): Whether prefix caching is enabled.
             prefix_caching_coordinator_policy (PrefixCachingCoordinatorPolicy): Routing policy.
             schedule_output_path (Optional[str]): Path to write scheduling decisions JSON.
+            prefix_caching_routing_alpha (float): Weight for prefix-aware routing score.
+            max_requests (int): Max concurrent requests per rank.
         """
         coordinator = cls(
             pipe_connection,
             data_parallel_size,
             tokenizer,
+            max_requests,
             inference_coordinator_port,
             deterministic_mode=deterministic_mode,
             block_size_tokens=block_size_tokens,
             enable_prefix_caching=enable_prefix_caching,
             prefix_caching_coordinator_policy=prefix_caching_coordinator_policy,
+            prefix_caching_routing_alpha=prefix_caching_routing_alpha,
             schedule_output_path=schedule_output_path,
             hostname=hostname,
         )
diff --git a/megatron/core/inference/engines/dynamic_engine.py b/megatron/core/inference/engines/dynamic_engine.py
index 4d5562bf39d..a9c8337271f 100644
--- a/megatron/core/inference/engines/dynamic_engine.py
+++ b/megatron/core/inference/engines/dynamic_engine.py
@@ -509,19 +509,21 @@ async def start_listening_to_data_parallel_coordinator(
             coordinator_ready_event = spawn_context.Event()
             self.inference_coordinator_process = spawn_context.Process(
                 target=DataParallelInferenceCoordinator.entrypoint,
-                args=(
-                    dp_process_pipe,
-                    coordinator_ready_event,
-                    get_pg_size(self.pg_collection.dp),
-                    self.controller.tokenizer,
-                    inference_coordinator_port,
-                    deterministic_mode,
-                    self.context.block_size_tokens,
-                    self.context.enable_prefix_caching,
-                    self.context.prefix_caching_coordinator_policy,
-                    coordinator_schedule_output_path,
-                    hostname,
-                ),
+                kwargs={
+                    "pipe_connection": dp_process_pipe,
+                    "ready_event": coordinator_ready_event,
+                    "data_parallel_size": get_pg_size(self.pg_collection.dp),
+                    "tokenizer": self.controller.tokenizer,
+                    "max_requests": self.context.max_requests,
+                    "inference_coordinator_port": inference_coordinator_port,
+                    "deterministic_mode": deterministic_mode,
+                    "block_size_tokens": self.context.block_size_tokens,
+                    "enable_prefix_caching": self.context.enable_prefix_caching,
+                    "prefix_caching_coordinator_policy": self.context.prefix_caching_coordinator_policy,
+                    "prefix_caching_routing_alpha": self.context.prefix_caching_routing_alpha,
+                    "schedule_output_path": coordinator_schedule_output_path,
+                    "hostname": hostname,
+                },
             )
             self.inference_coordinator_process.start()
             await await_process_call(dp_pipe.poll, self.inference_coordinator_process)
diff --git a/megatron/inference/utils.py b/megatron/inference/utils.py
index 5fce910a203..b59e2b77247 100644
--- a/megatron/inference/utils.py
+++ b/megatron/inference/utils.py
@@ -355,6 +355,7 @@ def get_inference_config_from_model_and_args(model: MegatronModule, args):
         enable_prefix_caching=args.inference_dynamic_batching_enable_prefix_caching,
         prefix_caching_eviction_policy=PrefixCachingEvictionPolicy(args.inference_dynamic_batching_prefix_caching_eviction_policy),
         prefix_caching_coordinator_policy=PrefixCachingCoordinatorPolicy(args.inference_dynamic_batching_prefix_caching_coordinator_policy),
+        prefix_caching_routing_alpha=getattr(args, 'inference_dynamic_batching_prefix_caching_routing_alpha', 0.5),
         prefix_caching_mamba_gb=getattr(args, 'inference_dynamic_batching_prefix_caching_mamba_gb', None),
         metrics_writer=metrics_writer,
         logging_step_interval=args.inference_logging_step_interval,
diff --git a/megatron/training/arguments.py b/megatron/training/arguments.py
index 595305a7780..d3026d94b68 100644
--- a/megatron/training/arguments.py
+++ b/megatron/training/arguments.py
@@ -1881,6 +1881,13 @@ def _add_inference_args(parser):
                        'block hash only. "longest_prefix" routes to the rank with '
                        'the longest matching prefix. "round_robin" ignores prefix '
                        'affinity and cycles through ranks.')
+    group.add_argument('--inference-dynamic-batching-prefix-caching-routing-alpha',
+                       type=float, default=0.5,
+                       dest='inference_dynamic_batching_prefix_caching_routing_alpha',
+                       help='Weight for prefix-aware routing score: '
+                       'score = alpha * match + (1 - alpha) * normalized_load. '
+                       'Higher alpha favors prefix cache hits; lower alpha '
+                       'favors load balance. Default: 0.5.')
     group.add_argument('--inference-dynamic-batching-prefix-caching-mamba-gb',
                        type=float, default=None,
                        dest='inference_dynamic_batching_prefix_caching_mamba_gb',
diff --git a/tests/unit_tests/inference/coordinator_test_utils.py b/tests/unit_tests/inference/coordinator_test_utils.py
new file mode 100644
index 00000000000..d33d8790ef9
--- /dev/null
+++ b/tests/unit_tests/inference/coordinator_test_utils.py
@@ -0,0 +1,77 @@
+# Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+
+"""Shared test fixtures and helpers for inference tests."""
+
+import itertools
+from collections import deque
+
+import numpy as np
+
+from megatron.core.inference.config import PrefixCachingCoordinatorPolicy
+from megatron.core.inference.data_parallel_inference_coordinator import (
+    DataParallelInferenceCoordinator,
+)
+
+
+def make_coordinator_direct(
+    data_parallel_size=2,
+    block_size_tokens=4,
+    enable_prefix_caching=True,
+    deterministic_mode=True,
+    prefix_caching_routing_alpha=0.5,
+    max_requests=10,
+    policy=PrefixCachingCoordinatorPolicy.LONGEST_PREFIX,
+    tokenizer=None,
+    rank_name_template="rank_{}",
+):
+    """Create a coordinator with mock ZMQ, for unit testing routing logic.
+
+    Returns the coordinator instance with fake rank identities.
+
+    Args:
+        data_parallel_size: Number of DP ranks.
+        block_size_tokens: Block size in tokens.
+        enable_prefix_caching: Whether prefix caching is enabled.
+        deterministic_mode: If True, sort identities for deterministic ordering.
+        prefix_caching_routing_alpha: Alpha for prefix-aware scoring.
+        max_requests: Max requests per rank (None disables vectorized scoring).
+        policy: Prefix caching coordinator routing policy.
+        tokenizer: Optional tokenizer instance (set on the coordinator).
+        rank_name_template: Format string for rank names, e.g. ``"rank_{}"``
+            or ``"rank-{}"``.  The integer rank index is substituted.
+    """
+    coordinator = object.__new__(DataParallelInferenceCoordinator)
+    coordinator.tokenizer = tokenizer
+    coordinator.data_parallel_size = data_parallel_size
+    coordinator.block_size_tokens = block_size_tokens
+    coordinator.enable_prefix_caching = enable_prefix_caching
+    coordinator.prefix_caching_coordinator_policy = policy
+    coordinator.prefix_caching_routing_alpha = prefix_caching_routing_alpha
+    coordinator.max_requests = max_requests
+
+    # Create fake rank identities.
+    coordinator.identities_of_data_parallel_ranks = deque(
+        [rank_name_template.format(i).encode() for i in range(data_parallel_size)]
+    )
+    if deterministic_mode:
+        coordinator.identities_of_data_parallel_ranks = deque(
+            sorted(coordinator.identities_of_data_parallel_ranks)
+        )
+    coordinator.data_parallel_rank_iterator = itertools.cycle(
+        coordinator.identities_of_data_parallel_ranks
+    )
+
+    n_ranks = data_parallel_size
+    coordinator._hash_table = {}
+    coordinator._hash_assignment_counter = 0
+    coordinator._round_robin_idx = 0
+
+    sorted_identities = sorted(coordinator.identities_of_data_parallel_ranks)
+    coordinator.identity_to_rank_index = {
+        identity: idx for idx, identity in enumerate(sorted_identities)
+    }
+
+    coordinator._pending_counts = np.zeros(n_ranks, dtype=np.int32)
+    coordinator._identities_list = list(sorted_identities)
+
+    return coordinator
diff --git a/tests/unit_tests/inference/test_data_parallel_inference_coordinator.py b/tests/unit_tests/inference/test_data_parallel_inference_coordinator.py
index f53ee707c21..b2e94bc54f9 100644
--- a/tests/unit_tests/inference/test_data_parallel_inference_coordinator.py
+++ b/tests/unit_tests/inference/test_data_parallel_inference_coordinator.py
@@ -10,9 +10,11 @@
 from typing import Dict, Optional
 
 import msgpack
+import numpy as np
 import pytest
 import torch
 
+from megatron.core.inference.config import PrefixCachingCoordinatorPolicy
 from megatron.core.inference.data_parallel_inference_coordinator import (
     DataParallelInferenceCoordinator,
 )
@@ -304,7 +306,15 @@ def coordinator():
         ready_event = spawn_context.Event()
         proc = spawn_context.Process(
             target=DataParallelInferenceCoordinator.entrypoint,
-            args=(pipe_child, ready_event, 0, DummyTokenizer(), DEFAULT_PORT, False),
+            kwargs={
+                "pipe_connection": pipe_child,
+                "ready_event": ready_event,
+                "data_parallel_size": 0,
+                "tokenizer": DummyTokenizer(),
+                "max_requests": 16,
+                "inference_coordinator_port": DEFAULT_PORT,
+                "deterministic_mode": False,
+            },
         )
         proc.start()
 
@@ -670,3 +680,119 @@ async def test_throughput(self, initialize_model_parallel, coordinator, test_cas
             await asyncio.wait_for(test_case_communicator.all_reduce_max(1), timeout=30.0)
         finally:
             await cleanup_engine(engine, client, timeout=60.0)
+
+
+def _set_hash_rank(coord, h, rank_identity, timestamp):
+    """Test helper: set a hash→rank timestamp in the coordinator's dict."""
+    rank_idx = coord.identity_to_rank_index[rank_identity]
+    coord._hash_table.setdefault(h, {})[rank_idx] = timestamp
+
+
+def _make_routing_coordinator(
+    num_ranks=4, enable_prefix_caching=False, policy=PrefixCachingCoordinatorPolicy.LONGEST_PREFIX
+):
+    """Create a coordinator with fake rank identities for routing-only tests.
+
+    Thin wrapper around the shared helper in coordinator_test_utils.py.
+    """
+    from tests.unit_tests.inference.coordinator_test_utils import (
+        make_coordinator_direct as _make_coordinator,
+    )
+
+    return _make_coordinator(
+        data_parallel_size=num_ranks,
+        block_size_tokens=64,
+        enable_prefix_caching=enable_prefix_caching,
+        policy=policy,
+        rank_name_template="rank-{}",
+    )
+
+
+class TestRoutingPolicies:
+    """Unit tests for routing behavior under different policies and load conditions."""
+
+    def test_no_prefix_caching_uses_round_robin(self):
+        """When prefix caching is off, round-robin is used regardless of load."""
+        coord = _make_routing_coordinator(num_ranks=3, enable_prefix_caching=False)
+        coord._pending_counts[coord.identity_to_rank_index[b"rank-0"]] = 2
+        coord._pending_counts[coord.identity_to_rank_index[b"rank-1"]] = 1
+
+        results = [coord.get_best_data_parallel_rank([]) for _ in range(6)]
+        assert results == [b"rank-0", b"rank-1", b"rank-2", b"rank-0", b"rank-1", b"rank-2"]
+
+    def test_empty_hashes_uses_round_robin(self):
+        """Empty hash list falls back to round-robin."""
+        coord = _make_routing_coordinator(num_ranks=4)
+        coord._pending_counts[coord.identity_to_rank_index[b"rank-1"]] = 5
+
+        results = [coord.get_best_data_parallel_rank([]) for _ in range(4)]
+        assert results == [b"rank-0", b"rank-1", b"rank-2", b"rank-3"]
+
+    def test_prefix_affinity_routing(self):
+        """When prefix caching is on with hashes, scoring picks the best rank."""
+        coord = _make_routing_coordinator(
+            num_ranks=3,
+            enable_prefix_caching=True,
+            policy=PrefixCachingCoordinatorPolicy.LONGEST_PREFIX,
+        )
+        for ident in coord.identities_of_data_parallel_ranks:
+            coord._pending_counts[coord.identity_to_rank_index[ident]] = 1
+
+        # Seed a hash on rank-2 so prefix routing prefers it.
+        fake_hash = 12345
+        _set_hash_rank(coord, fake_hash, b"rank-2", 1)
+
+        chosen = coord.get_best_data_parallel_rank([fake_hash])
+        assert chosen == b"rank-2"
+
+    def test_prefix_affinity_beats_free_capacity(self):
+        """A rank with a prefix match and capacity is preferred over a free rank."""
+        coord = _make_routing_coordinator(
+            num_ranks=3,
+            enable_prefix_caching=True,
+            policy=PrefixCachingCoordinatorPolicy.LONGEST_PREFIX,
+        )
+        coord._pending_counts[coord.identity_to_rank_index[b"rank-0"]] = 2
+        coord._pending_counts[coord.identity_to_rank_index[b"rank-1"]] = 1
+
+        fake_hash = 99999
+        _set_hash_rank(coord, fake_hash, b"rank-1", 1)
+
+        # Scoring: rank-1 gets prefix match bonus, which outweighs rank-2's
+        # free capacity advantage.
+        chosen = coord.get_best_data_parallel_rank([fake_hash])
+        assert chosen == b"rank-1"
+
+    def test_free_capacity_wins_when_prefix_rank_is_full(self):
+        """A free rank wins when the prefix-matched rank is full and alpha is low."""
+        coord = _make_routing_coordinator(
+            num_ranks=2,
+            enable_prefix_caching=True,
+            policy=PrefixCachingCoordinatorPolicy.FIRST_PREFIX_BLOCK,
+        )
+        coord.prefix_caching_routing_alpha = 0.1
+        coord._pending_counts[coord.identity_to_rank_index[b"rank-0"]] = 10
+
+        fake_hash = 42
+        _set_hash_rank(coord, fake_hash, b"rank-0", 1)
+
+        # score(rank-0) = 0.1*1 + 0.9*(0/10) = 0.1
+        # score(rank-1) = 0.1*0 + 0.9*(10/10) = 0.9
+        chosen = coord.get_best_data_parallel_rank([fake_hash])
+        assert chosen == b"rank-1"
+
+    def test_round_robin_policy_ignores_load(self):
+        """ROUND_ROBIN policy does naive round-robin regardless of load."""
+        coord = _make_routing_coordinator(
+            num_ranks=3,
+            enable_prefix_caching=True,
+            policy=PrefixCachingCoordinatorPolicy.ROUND_ROBIN,
+        )
+        coord._pending_counts[coord.identity_to_rank_index[b"rank-0"]] = 1
+        coord._pending_counts[coord.identity_to_rank_index[b"rank-1"]] = 1
+
+        coord._round_robin_idx = 0
+        identities = list(coord.identities_of_data_parallel_ranks)
+        for i in range(len(identities)):
+            chosen = coord.get_best_data_parallel_rank([99])
+            assert chosen == identities[i]
diff --git a/tests/unit_tests/inference/test_dynamic_prefix_caching_coordinator.py b/tests/unit_tests/inference/test_dynamic_prefix_caching_coordinator.py
index e8d2bc728ef..b4b8da0e538 100644
--- a/tests/unit_tests/inference/test_dynamic_prefix_caching_coordinator.py
+++ b/tests/unit_tests/inference/test_dynamic_prefix_caching_coordinator.py
@@ -14,6 +14,7 @@
 from unittest.mock import MagicMock
 
 import msgpack
+import numpy as np
 import pytest
 import torch
 
@@ -53,6 +54,12 @@
 BLOCK_SIZE = 4
 
 
+def _set_hash_rank(coordinator, h, rank_identity, timestamp):
+    """Test helper: set a hash→rank timestamp in the coordinator's dict."""
+    rank_idx = coordinator.identity_to_rank_index[rank_identity]
+    coordinator._hash_table.setdefault(h, {})[rank_idx] = timestamp
+
+
 class DummyTokenizer:
     """Dummy tokenizer that splits on whitespace and converts to ints."""
 
@@ -191,35 +198,27 @@ def make_coordinator_direct(
     block_size_tokens=BLOCK_SIZE,
     enable_prefix_caching=True,
     deterministic_mode=True,
+    prefix_caching_routing_alpha=0.5,
+    max_requests=10,
 ):
     """Create a coordinator with mock ZMQ, for unit testing routing logic.
 
-    Returns the coordinator instance with fake rank identities.
+    Thin wrapper around the shared helper in coordinator_test_utils.py that
+    supplies a DummyTokenizer and this module's BLOCK_SIZE default.
     """
-    coordinator = object.__new__(DataParallelInferenceCoordinator)
-    coordinator.tokenizer = DummyTokenizer()
-    coordinator.data_parallel_size = data_parallel_size
-    coordinator.block_size_tokens = block_size_tokens
-    coordinator.enable_prefix_caching = enable_prefix_caching
-    coordinator.prefix_caching_coordinator_policy = PrefixCachingCoordinatorPolicy.LONGEST_PREFIX
-
-    # Create fake rank identities.
-    coordinator.identities_of_data_parallel_ranks = deque(
-        [f"rank_{i}".encode() for i in range(data_parallel_size)]
+    from tests.unit_tests.inference.coordinator_test_utils import (
+        make_coordinator_direct as _make_coordinator,
     )
-    if deterministic_mode:
-        coordinator.identities_of_data_parallel_ranks = deque(
-            sorted(coordinator.identities_of_data_parallel_ranks)
-        )
-    coordinator.data_parallel_rank_iterator = itertools.cycle(
-        coordinator.identities_of_data_parallel_ranks
-    )
-
-    coordinator.hash_to_rank_info = {}
-    coordinator._round_robin_idx = 0
-    coordinator._assignment_counter = 0
 
-    return coordinator
+    return _make_coordinator(
+        data_parallel_size=data_parallel_size,
+        block_size_tokens=block_size_tokens,
+        enable_prefix_caching=enable_prefix_caching,
+        deterministic_mode=deterministic_mode,
+        prefix_caching_routing_alpha=prefix_caching_routing_alpha,
+        max_requests=max_requests,
+        tokenizer=DummyTokenizer(),
+    )
 
 
 # ============================================================================
@@ -292,15 +291,20 @@ def test_hash_parent_chaining(self):
 class TestCoordinatorPrefixRouting:
     """Test routing decisions based on prefix cache affinity."""
 
-    def test_no_match_uses_round_robin(self):
-        """When no rank has matching hashes, falls back to round-robin."""
+    def test_no_match_prefers_least_loaded(self):
+        """When no rank has matching hashes, the rank with most free capacity wins."""
         coordinator = make_coordinator_direct()
         hashes = coordinator.compute_request_hashes([1, 2, 3, 4])
 
-        rank1 = coordinator.get_best_data_parallel_rank(hashes)
-        rank2 = coordinator.get_best_data_parallel_rank(hashes)
-        # Round-robin cycles through ranks.
-        assert rank1 != rank2
+        rank_0 = coordinator.identities_of_data_parallel_ranks[0]
+        rank_1 = coordinator.identities_of_data_parallel_ranks[1]
+
+        # rank_1 has fewer pending requests, so more free capacity.
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_0]] = 5
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_1]] = 1
+
+        selected = coordinator.get_best_data_parallel_rank(hashes)
+        assert selected == rank_1
 
     def test_routes_to_rank_with_longest_match(self):
         """Request is routed to the rank with the longest consecutive prefix match."""
@@ -312,18 +316,22 @@ def test_routes_to_rank_with_longest_match(self):
         rank_0 = coordinator.identities_of_data_parallel_ranks[0]
         rank_1 = coordinator.identities_of_data_parallel_ranks[1]
 
+        # Ensure no rank is idle so prefix-matching logic is exercised.
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_0]] = 1
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_1]] = 1
+
         # rank_0 has first block only.
-        coordinator.hash_to_rank_info.setdefault(hashes[0], {})[rank_0] = 1
+        _set_hash_rank(coordinator, hashes[0], rank_0, 1)
 
         # rank_1 has first two blocks.
-        coordinator.hash_to_rank_info.setdefault(hashes[0], {})[rank_1] = 1
-        coordinator.hash_to_rank_info.setdefault(hashes[1], {})[rank_1] = 1
+        _set_hash_rank(coordinator, hashes[0], rank_1, 1)
+        _set_hash_rank(coordinator, hashes[1], rank_1, 1)
 
         selected = coordinator.get_best_data_parallel_rank(hashes)
         assert selected == rank_1
 
-    def test_recency_tiebreaker(self):
-        """When two ranks tie on match length, prefer the more recent one."""
+    def test_equal_scores_tiebreak_by_rank_index(self):
+        """When two ranks have equal scores, the lower rank index wins."""
         coordinator = make_coordinator_direct()
         tokens = [1, 2, 3, 4, 5, 6, 7, 8]
         hashes = coordinator.compute_request_hashes(tokens)
@@ -331,17 +339,23 @@ def test_recency_tiebreaker(self):
         rank_0 = coordinator.identities_of_data_parallel_ranks[0]
         rank_1 = coordinator.identities_of_data_parallel_ranks[1]
 
-        # Both ranks have both blocks, but rank_1 has higher timestamps.
+        # Both ranks have same pending counts, same match, and same timestamp.
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_0]] = 1
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_1]] = 1
+
         for h in hashes:
-            coordinator.hash_to_rank_info.setdefault(h, {})[rank_0] = 1
-            coordinator.hash_to_rank_info.setdefault(h, {})[rank_1] = 5
+            _set_hash_rank(coordinator, h, rank_0, 1)
+            _set_hash_rank(coordinator, h, rank_1, 1)
 
+        # Equal scores → lowest rank index (rank_0) wins.
         selected = coordinator.get_best_data_parallel_rank(hashes)
-        assert selected == rank_1
+        assert selected == rank_0
 
     def test_empty_hashes_uses_round_robin(self):
         """Empty hash list falls back to round-robin."""
         coordinator = make_coordinator_direct()
+        for identity in coordinator.identities_of_data_parallel_ranks:
+            coordinator._pending_counts[coordinator.identity_to_rank_index[identity]] = 1
         rank1 = coordinator.get_best_data_parallel_rank([])
         rank2 = coordinator.get_best_data_parallel_rank([])
         assert rank1 != rank2
@@ -349,6 +363,8 @@ def test_empty_hashes_uses_round_robin(self):
     def test_disabled_prefix_caching_uses_round_robin(self):
         """With prefix caching disabled, always uses round-robin."""
         coordinator = make_coordinator_direct(enable_prefix_caching=False)
+        for identity in coordinator.identities_of_data_parallel_ranks:
+            coordinator._pending_counts[coordinator.identity_to_rank_index[identity]] = 1
         rank1 = coordinator.get_best_data_parallel_rank([1, 2, 3])
         rank2 = coordinator.get_best_data_parallel_rank([1, 2, 3])
         assert rank1 != rank2
@@ -361,31 +377,33 @@ def test_update_rank_hashes_adds_to_set(self):
         """_update_rank_hashes adds hashes to the rank's set."""
         coordinator = make_coordinator_direct()
         rank_0 = coordinator.identities_of_data_parallel_ranks[0]
+        idx_0 = coordinator.identity_to_rank_index[rank_0]
 
         coordinator._update_rank_hashes(rank_0, [100, 200, 300])
-        assert all(rank_0 in coordinator.hash_to_rank_info[h] for h in [100, 200, 300])
+        assert all(coordinator._hash_table.get(h, {}).get(idx_0, 0) > 0 for h in [100, 200, 300])
 
     def test_update_rank_hashes_increments_counter(self):
         """Each call to _update_rank_hashes increments the assignment counter."""
         coordinator = make_coordinator_direct()
         rank_0 = coordinator.identities_of_data_parallel_ranks[0]
 
-        assert coordinator._assignment_counter == 0
+        assert coordinator._hash_assignment_counter == 0
         coordinator._update_rank_hashes(rank_0, [100])
-        assert coordinator._assignment_counter == 1
+        assert coordinator._hash_assignment_counter == 1
         coordinator._update_rank_hashes(rank_0, [200])
-        assert coordinator._assignment_counter == 2
+        assert coordinator._hash_assignment_counter == 2
 
     def test_timestamps_updated_on_reassignment(self):
         """Re-assigning a hash to the same rank updates its timestamp."""
         coordinator = make_coordinator_direct()
         rank_0 = coordinator.identities_of_data_parallel_ranks[0]
+        idx_0 = coordinator.identity_to_rank_index[rank_0]
 
         coordinator._update_rank_hashes(rank_0, [100])
-        ts1 = coordinator.hash_to_rank_info[100][rank_0]
+        ts1 = coordinator._hash_table[100][idx_0]
 
         coordinator._update_rank_hashes(rank_0, [100])
-        ts2 = coordinator.hash_to_rank_info[100][rank_0]
+        ts2 = coordinator._hash_table[100][idx_0]
 
         assert ts2 > ts1
 
@@ -393,22 +411,25 @@ def test_multiple_requests_accumulate_hashes(self):
         """Multiple requests to the same rank accumulate their hashes."""
         coordinator = make_coordinator_direct()
         rank_0 = coordinator.identities_of_data_parallel_ranks[0]
+        idx_0 = coordinator.identity_to_rank_index[rank_0]
 
         coordinator._update_rank_hashes(rank_0, [10, 20])
         coordinator._update_rank_hashes(rank_0, [30, 40])
-        assert all(rank_0 in coordinator.hash_to_rank_info[h] for h in [10, 20, 30, 40])
+        assert all(coordinator._hash_table.get(h, {}).get(idx_0, 0) > 0 for h in [10, 20, 30, 40])
 
     def test_hash_can_appear_in_multiple_ranks(self):
         """The same hash can be owned by multiple ranks."""
         coordinator = make_coordinator_direct()
         rank_0 = coordinator.identities_of_data_parallel_ranks[0]
         rank_1 = coordinator.identities_of_data_parallel_ranks[1]
+        idx_0 = coordinator.identity_to_rank_index[rank_0]
+        idx_1 = coordinator.identity_to_rank_index[rank_1]
 
         coordinator._update_rank_hashes(rank_0, [100])
         coordinator._update_rank_hashes(rank_1, [100])
 
-        assert rank_0 in coordinator.hash_to_rank_info[100]
-        assert rank_1 in coordinator.hash_to_rank_info[100]
+        assert coordinator._hash_table[100].get(idx_0, 0) > 0
+        assert coordinator._hash_table[100].get(idx_1, 0) > 0
 
     def test_routing_then_state_update_flow(self):
         """Full flow: compute hashes, route, update state, then re-route to same rank."""
@@ -425,6 +446,29 @@ def test_routing_then_state_update_flow(self):
         rank2 = coordinator.get_best_data_parallel_rank(hashes)
         assert rank2 == rank
 
+    def test_recency_breaks_tie_at_equal_load(self):
+        """When two ranks match the same prefix and have equal load, the more
+        recently assigned rank wins."""
+        coordinator = make_coordinator_direct()
+        tokens = [1, 2, 3, 4, 5, 6, 7, 8]
+        hashes = coordinator.compute_request_hashes(tokens)
+
+        rank_0 = coordinator.identities_of_data_parallel_ranks[0]
+        rank_1 = coordinator.identities_of_data_parallel_ranks[1]
+
+        # Equal load on both ranks.
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_0]] = 1
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_1]] = 1
+
+        # Both have the prefix, but rank_1 was assigned more recently.
+        for h in hashes:
+            _set_hash_rank(coordinator, h, rank_0, 1)
+            _set_hash_rank(coordinator, h, rank_1, 5)
+
+        # rank_1 wins via recency despite rank_0 having a lower index.
+        selected = coordinator.get_best_data_parallel_rank(hashes)
+        assert selected == rank_1
+
 
 @pytest.mark.skipif(ZMQ_FLAKY_SHUTDOWN, reason="ZMQ shutdown is flaky")
 class TestCoordinatorEndToEnd:
@@ -496,8 +540,12 @@ def test_first_block_match_routes_to_rank(self):
         rank_0 = coordinator.identities_of_data_parallel_ranks[0]
         rank_1 = coordinator.identities_of_data_parallel_ranks[1]
 
+        # Ensure no rank is idle so prefix-matching logic is exercised.
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_0]] = 1
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_1]] = 1
+
         # Only rank_1 has the first block.
-        coordinator.hash_to_rank_info.setdefault(hashes[0], {})[rank_1] = 1
+        _set_hash_rank(coordinator, hashes[0], rank_1, 1)
 
         selected = coordinator.get_best_data_parallel_rank(hashes[:1])
         assert selected == rank_1
@@ -512,20 +560,24 @@ def test_first_block_ignores_longer_match(self):
         rank_0 = coordinator.identities_of_data_parallel_ranks[0]
         rank_1 = coordinator.identities_of_data_parallel_ranks[1]
 
+        # Ensure no rank is idle so prefix-matching logic is exercised.
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_0]] = 1
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_1]] = 1
+
         # rank_0 has first block only, with higher timestamp.
-        coordinator.hash_to_rank_info.setdefault(hashes[0], {})[rank_0] = 10
+        _set_hash_rank(coordinator, hashes[0], rank_0, 10)
 
         # rank_1 has all three blocks, but lower timestamp on first block.
         for h in hashes:
-            coordinator.hash_to_rank_info.setdefault(h, {})[rank_1] = 1
+            _set_hash_rank(coordinator, h, rank_1, 1)
 
         # rank_0 wins because it has higher recency on the first block.
         # Caller truncates to [:1] before calling get_best_data_parallel_rank.
         selected = coordinator.get_best_data_parallel_rank(hashes[:1])
         assert selected == rank_0
 
-    def test_first_block_recency_tiebreaker(self):
-        """When multiple ranks have the first block, higher timestamp wins."""
+    def test_first_block_equal_match_tiebreaks_by_rank_index(self):
+        """When multiple ranks have the first block with equal load, lowest index wins."""
         coordinator = make_first_prefix_block_coordinator()
         tokens = [1, 2, 3, 4, 5, 6, 7, 8]
         hashes = coordinator.compute_request_hashes(tokens)
@@ -533,28 +585,36 @@ def test_first_block_recency_tiebreaker(self):
         rank_0 = coordinator.identities_of_data_parallel_ranks[0]
         rank_1 = coordinator.identities_of_data_parallel_ranks[1]
 
-        # Both ranks have the first block.
-        coordinator.hash_to_rank_info.setdefault(hashes[0], {})[rank_0] = 3
-        coordinator.hash_to_rank_info.setdefault(hashes[0], {})[rank_1] = 7
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_0]] = 1
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_1]] = 1
+
+        # Both ranks have the first block with the same timestamp.
+        _set_hash_rank(coordinator, hashes[0], rank_0, 3)
+        _set_hash_rank(coordinator, hashes[0], rank_1, 3)
 
+        # Equal scores → lowest rank index wins.
         selected = coordinator.get_best_data_parallel_rank(hashes[:1])
-        assert selected == rank_1
+        assert selected == rank_0
 
-    def test_no_first_block_match_uses_round_robin(self):
-        """Falls back to round-robin when no rank has the first block."""
+    def test_no_first_block_match_prefers_least_loaded(self):
+        """When no rank has the first block, the least loaded rank wins."""
         coordinator = make_first_prefix_block_coordinator()
         tokens = [1, 2, 3, 4, 5, 6, 7, 8]
         hashes = coordinator.compute_request_hashes(tokens)
 
         rank_0 = coordinator.identities_of_data_parallel_ranks[0]
+        rank_1 = coordinator.identities_of_data_parallel_ranks[1]
 
         # rank_0 has block 1 (second block), but not block 0.
-        coordinator.hash_to_rank_info.setdefault(hashes[1], {})[rank_0] = 1
+        _set_hash_rank(coordinator, hashes[1], rank_0, 1)
 
-        # No rank has the first block, so round-robin.
-        r1 = coordinator.get_best_data_parallel_rank(hashes[:1])
-        r2 = coordinator.get_best_data_parallel_rank(hashes[:1])
-        assert r1 != r2
+        # rank_1 has fewer pending requests.
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_0]] = 5
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_1]] = 1
+
+        # No rank has the first block → load determines winner.
+        selected = coordinator.get_best_data_parallel_rank(hashes[:1])
+        assert selected == rank_1
 
     def test_first_block_policy_with_single_block_prompt(self):
         """Works correctly with a prompt that has only one block."""
@@ -565,7 +625,269 @@ def test_first_block_policy_with_single_block_prompt(self):
 
         rank_1 = coordinator.identities_of_data_parallel_ranks[1]
 
-        coordinator.hash_to_rank_info.setdefault(hashes[0], {})[rank_1] = 1
+        # Ensure no rank is idle so prefix-matching logic is exercised.
+        for identity in coordinator.identities_of_data_parallel_ranks:
+            coordinator._pending_counts[coordinator.identity_to_rank_index[identity]] = 1
+
+        _set_hash_rank(coordinator, hashes[0], rank_1, 1)
 
         selected = coordinator.get_best_data_parallel_rank(hashes[:1])
         assert selected == rank_1
+
+
+class TestLoadAwarePrefixRouting:
+    """Test that prefix routing spreads load across ranks with the same prefix."""
+
+    def test_spreads_across_ranks_with_same_prefix(self):
+        """When three ranks all cache the same prefix, requests spread by load."""
+        coordinator = make_coordinator_direct(data_parallel_size=3)
+        tokens = [1, 2, 3, 4, 5, 6, 7, 8]
+        hashes = coordinator.compute_request_hashes(tokens)
+
+        rank_0 = coordinator.identities_of_data_parallel_ranks[0]
+        rank_1 = coordinator.identities_of_data_parallel_ranks[1]
+        rank_2 = coordinator.identities_of_data_parallel_ranks[2]
+
+        # All three ranks have both blocks cached with the same timestamp.
+        for h in hashes:
+            _set_hash_rank(coordinator, h, rank_0, 1)
+            _set_hash_rank(coordinator, h, rank_1, 1)
+            _set_hash_rank(coordinator, h, rank_2, 1)
+
+        # Simulate sending 6 requests. With load-aware routing, they should
+        # spread across ranks rather than all going to one.
+        assigned_ranks = []
+        for _ in range(6):
+            rank = coordinator.get_best_data_parallel_rank(hashes)
+            coordinator._pending_counts[coordinator.identity_to_rank_index[rank]] += 1
+            assigned_ranks.append(rank)
+
+        # Each rank should get exactly 2 of the 6 requests.
+        from collections import Counter
+
+        counts = Counter(assigned_ranks)
+        assert counts[rank_0] == 2
+        assert counts[rank_1] == 2
+        assert counts[rank_2] == 2
+
+    def test_load_overrides_recency(self):
+        """A rank with a higher timestamp but more pending requests is not preferred."""
+        coordinator = make_coordinator_direct()
+        tokens = [1, 2, 3, 4]
+        hashes = coordinator.compute_request_hashes(tokens)
+
+        rank_0 = coordinator.identities_of_data_parallel_ranks[0]
+        rank_1 = coordinator.identities_of_data_parallel_ranks[1]
+
+        # Both ranks have the prefix. rank_1 has a higher (more recent) timestamp.
+        _set_hash_rank(coordinator, hashes[0], rank_0, 1)
+        _set_hash_rank(coordinator, hashes[0], rank_1, 10)
+
+        # But rank_1 already has 5 pending requests while rank_0 has only 1.
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_0]] = 1
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_1]] = 5
+
+        selected = coordinator.get_best_data_parallel_rank(hashes)
+        assert selected == rank_0
+
+    def test_pending_count_decremented_on_completion(self):
+        """Completing a request frees capacity on the assigned rank."""
+        coordinator = make_coordinator_direct()
+        tokens = [1, 2, 3, 4]
+        hashes = coordinator.compute_request_hashes(tokens)
+
+        rank_0 = coordinator.identities_of_data_parallel_ranks[0]
+        rank_1 = coordinator.identities_of_data_parallel_ranks[1]
+
+        _set_hash_rank(coordinator, hashes[0], rank_0, 1)
+        _set_hash_rank(coordinator, hashes[0], rank_1, 1)
+
+        # Simulate assigning a request to rank_0.
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_0]] = 1
+        coordinator.request_id_to_rank = {42: rank_0}
+
+        # Simulate completion: decrement pending count.
+        assigned_rank = coordinator.request_id_to_rank.pop(42, None)
+        if assigned_rank is not None:
+            idx = coordinator.identity_to_rank_index.get(assigned_rank)
+            if idx is not None:
+                coordinator._pending_counts[idx] = max(0, coordinator._pending_counts[idx] - 1)
+
+        assert coordinator._pending_counts[coordinator.identity_to_rank_index[rank_0]] == 0
+
+    def test_equal_load_tiebreaks_by_rank_index(self):
+        """With equal pending counts and match, lowest rank index wins."""
+        coordinator = make_coordinator_direct()
+        tokens = [1, 2, 3, 4]
+        hashes = coordinator.compute_request_hashes(tokens)
+
+        rank_0 = coordinator.identities_of_data_parallel_ranks[0]
+        rank_1 = coordinator.identities_of_data_parallel_ranks[1]
+
+        # Equal pending counts, both have the prefix cached.
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_0]] = 1
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_1]] = 1
+        _set_hash_rank(coordinator, hashes[0], rank_0, 10)
+        _set_hash_rank(coordinator, hashes[0], rank_1, 1)
+
+        # Equal scores → lowest rank index (rank_0) wins.
+        selected = coordinator.get_best_data_parallel_rank(hashes)
+        assert selected == rank_0
+
+
+class TestScoringFunctionRouting:
+    """Test the alpha-based scoring function: score = alpha * match + (1 - alpha) * normalized_load."""
+
+    def test_high_alpha_prefers_prefix_match(self):
+        """With alpha=1.0, a rank with a prefix hit is always preferred over a free rank."""
+        coordinator = make_coordinator_direct(prefix_caching_routing_alpha=1.0, max_requests=10)
+        tokens = [1, 2, 3, 4]
+        hashes = coordinator.compute_request_hashes(tokens)
+
+        rank_0 = coordinator.identities_of_data_parallel_ranks[0]
+        rank_1 = coordinator.identities_of_data_parallel_ranks[1]
+
+        # rank_0 has the prefix but is heavily loaded (9/10 slots used).
+        _set_hash_rank(coordinator, hashes[0], rank_0, 1)
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_0]] = 9
+
+        # rank_1 has no prefix match but is idle.
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_1]] = 0
+
+        # alpha=1.0: score(rank_0) = 1*1 + 0*0.1 = 1.0
+        #            score(rank_1) = 1*0 + 0*1.0 = 0.0
+        selected = coordinator.get_best_data_parallel_rank(hashes)
+        assert selected == rank_0
+
+    def test_low_alpha_prefers_free_capacity(self):
+        """With alpha=0.0, the rank with the most free capacity is preferred."""
+        coordinator = make_coordinator_direct(prefix_caching_routing_alpha=0.0, max_requests=10)
+        tokens = [1, 2, 3, 4]
+        hashes = coordinator.compute_request_hashes(tokens)
+
+        rank_0 = coordinator.identities_of_data_parallel_ranks[0]
+        rank_1 = coordinator.identities_of_data_parallel_ranks[1]
+
+        # rank_0 has the prefix but is heavily loaded.
+        _set_hash_rank(coordinator, hashes[0], rank_0, 1)
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_0]] = 8
+
+        # rank_1 has no prefix match but is nearly idle.
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_1]] = 1
+
+        # alpha=0.0: score(rank_0) = 0*1 + 1*(2/10) = 0.2
+        #            score(rank_1) = 0*0 + 1*(9/10) = 0.9
+        selected = coordinator.get_best_data_parallel_rank(hashes)
+        assert selected == rank_1
+
+    def test_balanced_alpha_trades_off(self):
+        """With alpha=0.5, prefix match and load are balanced."""
+        coordinator = make_coordinator_direct(prefix_caching_routing_alpha=0.5, max_requests=10)
+        tokens = [1, 2, 3, 4]
+        hashes = coordinator.compute_request_hashes(tokens)
+
+        rank_0 = coordinator.identities_of_data_parallel_ranks[0]
+        rank_1 = coordinator.identities_of_data_parallel_ranks[1]
+
+        # rank_0 has prefix match, 7 pending (3 free).
+        _set_hash_rank(coordinator, hashes[0], rank_0, 1)
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_0]] = 7
+
+        # rank_1 has no prefix match, 0 pending (10 free).
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_1]] = 0
+
+        # alpha=0.5: score(rank_0) = 0.5*1 + 0.5*(3/10) = 0.5 + 0.15 = 0.65
+        #            score(rank_1) = 0.5*0 + 0.5*(10/10) = 0.0 + 0.5  = 0.5
+        selected = coordinator.get_best_data_parallel_rank(hashes)
+        assert selected == rank_0
+
+    def test_balanced_alpha_prefers_free_when_heavily_loaded(self):
+        """With alpha=0.5, a completely free rank beats a nearly-full rank with prefix match."""
+        coordinator = make_coordinator_direct(prefix_caching_routing_alpha=0.5, max_requests=10)
+        tokens = [1, 2, 3, 4]
+        hashes = coordinator.compute_request_hashes(tokens)
+
+        rank_0 = coordinator.identities_of_data_parallel_ranks[0]
+        rank_1 = coordinator.identities_of_data_parallel_ranks[1]
+
+        # rank_0 has prefix match, 10 pending (0 free).
+        _set_hash_rank(coordinator, hashes[0], rank_0, 1)
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_0]] = 10
+
+        # rank_1 has no prefix match, 0 pending (10 free).
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_1]] = 0
+
+        # alpha=0.5: score(rank_0) = 0.5*1 + 0.5*(0/10) = 0.5
+        #            score(rank_1) = 0.5*0 + 0.5*(10/10) = 0.5
+        # Tie broken by rank index: rank_0 has lower index.
+        selected = coordinator.get_best_data_parallel_rank(hashes)
+        assert selected == rank_0
+
+    def test_scoring_tiebreak_by_rank_index(self):
+        """When scores are equal, the rank with lower index is preferred."""
+        coordinator = make_coordinator_direct(prefix_caching_routing_alpha=0.5, max_requests=10)
+        tokens = [1, 2, 3, 4]
+        hashes = coordinator.compute_request_hashes(tokens)
+
+        rank_0 = coordinator.identities_of_data_parallel_ranks[0]
+        rank_1 = coordinator.identities_of_data_parallel_ranks[1]
+
+        # Both ranks have prefix match and same load.
+        _set_hash_rank(coordinator, hashes[0], rank_0, 1)
+        _set_hash_rank(coordinator, hashes[0], rank_1, 1)
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_0]] = 5
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_1]] = 5
+
+        selected = coordinator.get_best_data_parallel_rank(hashes)
+        assert selected == rank_0
+
+    def test_scoring_spreads_load_across_ranks(self):
+        """Scoring function distributes requests when all ranks have prefix match."""
+        coordinator = make_coordinator_direct(
+            data_parallel_size=3, prefix_caching_routing_alpha=0.5, max_requests=10
+        )
+        tokens = [1, 2, 3, 4, 5, 6, 7, 8]
+        hashes = coordinator.compute_request_hashes(tokens)
+
+        rank_0 = coordinator.identities_of_data_parallel_ranks[0]
+        rank_1 = coordinator.identities_of_data_parallel_ranks[1]
+        rank_2 = coordinator.identities_of_data_parallel_ranks[2]
+
+        # All three ranks have both blocks cached.
+        for h in hashes:
+            _set_hash_rank(coordinator, h, rank_0, 1)
+            _set_hash_rank(coordinator, h, rank_1, 1)
+            _set_hash_rank(coordinator, h, rank_2, 1)
+
+        # Simulate sending 6 requests.
+        assigned_ranks = []
+        for _ in range(6):
+            rank = coordinator.get_best_data_parallel_rank(hashes)
+            coordinator._pending_counts[coordinator.identity_to_rank_index[rank]] += 1
+            assigned_ranks.append(rank)
+
+        from collections import Counter
+
+        counts = Counter(assigned_ranks)
+        # Each rank should get exactly 2 of the 6 requests.
+        assert counts[rank_0] == 2
+        assert counts[rank_1] == 2
+        assert counts[rank_2] == 2
+
+    def test_scoring_with_no_prefix_match_anywhere(self):
+        """When no rank has a prefix match, load alone determines the winner."""
+        coordinator = make_coordinator_direct(prefix_caching_routing_alpha=0.5, max_requests=10)
+        tokens = [1, 2, 3, 4]
+        hashes = coordinator.compute_request_hashes(tokens)
+
+        rank_0 = coordinator.identities_of_data_parallel_ranks[0]
+        rank_1 = coordinator.identities_of_data_parallel_ranks[1]
+
+        # No prefix matches for either rank.
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_0]] = 5
+        coordinator._pending_counts[coordinator.identity_to_rank_index[rank_1]] = 2
+
+        # alpha=0.5: score(rank_0) = 0 + 0.5*(5/10) = 0.25
+        #            score(rank_1) = 0 + 0.5*(8/10) = 0.4
+        selected = coordinator.get_best_data_parallel_rank(hashes)
+        assert selected == rank_1