[PD] Emit inactive KV blocks for decode affinity

tests/v1/core/test_prefix_caching.py

@@ -15,6 +15,7 @@
 from vllm.distributed.kv_events import (
     MEDIUM_GPU,
     AllBlocksCleared,
+    BlockInactive,
     BlockRemoved,
     BlockStored,
 )
@@ -2579,6 +2580,211 @@ def test_emit_cached_block_events_zero_cached():
     assert pool.take_events() == []
 
 
+def test_emit_remote_recv_block_stored_when_caching_disabled():
+    """Decode remote-recv path emits BlockStored when prefix caching is off."""
+    block_size = 4
+    num_full_blocks = 3
+    num_tokens = block_size * num_full_blocks
+    kv_cache_config = make_kv_cache_config(block_size=block_size, num_blocks=16)
+    manager = make_kv_cache_manager(
+        kv_cache_config,
+        max_model_len=128,
+        enable_caching=False,
+        hash_block_size=block_size,
+        enable_kv_cache_events=True,
+    )
+    req = make_request(
+        "req_remote_recv",
+        prompt_token_ids=list(range(num_tokens)),
+        block_size=block_size,
+        hash_fn=sha256,
+    )
+
+    manager.emit_remote_recv_block_stored(req, num_tokens)
+
+    events = manager.take_events()
+    assert len(events) == 1
+    event = events[0]
+    assert isinstance(event, BlockStored)
+    assert event.medium == MEDIUM_GPU
+    assert event.group_idx == 0
+    assert event.block_size == block_size
+    assert len(event.block_hashes) == num_full_blocks
+    assert event.token_ids == list(req.all_token_ids[:num_tokens])
+
+
+def test_emit_remote_recv_block_stored_skipped_when_caching_enabled():
+    """Avoid double-emit: caching on already emits via cache_full_blocks."""
+    block_size = 4
+    num_tokens = block_size * 3
+    kv_cache_config = make_kv_cache_config(block_size=block_size, num_blocks=16)
+    manager = make_kv_cache_manager(
+        kv_cache_config,
+        max_model_len=128,
+        enable_caching=True,
+        hash_block_size=block_size,
+        enable_kv_cache_events=True,
+    )
+    req = make_request(
+        "req_remote_recv_cached",
+        prompt_token_ids=list(range(num_tokens)),
+        block_size=block_size,
+        hash_fn=sha256,
+    )
+
+    manager.emit_remote_recv_block_stored(req, num_tokens)
+    assert manager.take_events() == []
+
+
+def test_emit_remote_recv_block_stored_skipped_when_events_disabled():
+    block_size = 4
+    num_tokens = block_size * 2
+    kv_cache_config = make_kv_cache_config(block_size=block_size, num_blocks=16)
+    manager = make_kv_cache_manager(
+        kv_cache_config,
+        max_model_len=128,
+        enable_caching=False,
+        hash_block_size=block_size,
+        enable_kv_cache_events=False,
+    )
+    req = make_request(
+        "req_remote_recv_no_events",
+        prompt_token_ids=list(range(num_tokens)),
+        block_size=block_size,
+        hash_fn=sha256,
+    )
+
+    manager.emit_remote_recv_block_stored(req, num_tokens)
+    assert manager.take_events() == []
+
+
+def test_update_waiting_for_remote_kv_emits_for_consumer():
+    """Scheduler success path calls emit_remote_recv_block_stored on Decode."""
+    from unittest.mock import MagicMock
+
+    scheduler = object.__new__(Scheduler)
+    scheduler.connector = MagicMock()
+    scheduler.needs_kv_cache_zeroing = False
+    scheduler.kv_cache_manager = MagicMock()
+    scheduler.failed_recving_kv_req_ids = set()
+    scheduler.finished_recving_kv_req_ids = {"req-1"}
+    scheduler.vllm_config = SimpleNamespace(
+        kv_transfer_config=SimpleNamespace(is_kv_consumer=True)
+    )
+
+    request = MagicMock()
+    request.request_id = "req-1"
+    request.num_computed_tokens = 12
+    request.num_tokens = 16
+
+    scheduler._update_waiting_for_remote_kv(request)
+
+    scheduler.kv_cache_manager.cache_blocks.assert_called_once_with(request, 12)
+    scheduler.kv_cache_manager.emit_remote_recv_block_stored.assert_called_once_with(
+        request, 12
+    )
+    assert "req-1" not in scheduler.finished_recving_kv_req_ids
+
+
+def test_update_waiting_for_remote_kv_skips_emit_for_producer():
+    from unittest.mock import MagicMock
+
+    scheduler = object.__new__(Scheduler)
+    scheduler.connector = MagicMock()
+    scheduler.needs_kv_cache_zeroing = False
+    scheduler.kv_cache_manager = MagicMock()
+    scheduler.failed_recving_kv_req_ids = set()
+    scheduler.finished_recving_kv_req_ids = {"req-1"}
+    scheduler.vllm_config = SimpleNamespace(
+        kv_transfer_config=SimpleNamespace(is_kv_consumer=False)
+    )
+
+    request = MagicMock()
+    request.request_id = "req-1"
+    request.num_computed_tokens = 12
+    request.num_tokens = 16
+
+    scheduler._update_waiting_for_remote_kv(request)
+
+    scheduler.kv_cache_manager.cache_blocks.assert_called_once_with(request, 12)
+    scheduler.kv_cache_manager.emit_remote_recv_block_stored.assert_not_called()
+
+
+def test_free_blocks_emits_block_inactive_when_enabled():
+    """BlockInactive is emitted on ref_cnt→0 when enable_block_inactive_events."""
+    block_size = 4
+    pool = BlockPool(
+        num_gpu_blocks=8,
+        enable_caching=True,
+        hash_block_size=block_size,
+        enable_kv_cache_events=True,
+        enable_block_inactive_events=True,
+    )
+    req = make_request(
+        "req_inactive",
+        prompt_token_ids=list(range(block_size)),
+        block_size=block_size,
+        hash_fn=sha256,
+    )
+    blocks = pool.get_new_blocks(1)
+    pool.cache_full_blocks(
+        request=req,
+        blocks=blocks,
+        num_cached_blocks=0,
+        num_full_blocks=1,
+        block_size=block_size,
+        kv_cache_group_id=0,
+    )
+    pool.take_events()  # drain BlockStored
+
+    pool.free_blocks(blocks)
+    events = pool.take_events()
+    assert len(events) == 1
+    assert isinstance(events[0], BlockInactive)
+    assert events[0].medium == MEDIUM_GPU
+
+
+def test_free_blocks_skips_block_inactive_when_disabled():
+    """Prefill-style gate: Stored still works; Inactive is suppressed."""
+    block_size = 4
+    pool = BlockPool(
+        num_gpu_blocks=8,
+        enable_caching=True,
+        hash_block_size=block_size,
+        enable_kv_cache_events=True,
+        enable_block_inactive_events=False,
+    )
+    req = make_request(
+        "req_no_inactive",
+        prompt_token_ids=list(range(block_size)),
+        block_size=block_size,
+        hash_fn=sha256,
+    )
+    blocks = pool.get_new_blocks(1)
+    pool.cache_full_blocks(
+        request=req,
+        blocks=blocks,
+        num_cached_blocks=0,
+        num_full_blocks=1,
+        block_size=block_size,
+        kv_cache_group_id=0,
+    )
+    stored = pool.take_events()
+    assert any(isinstance(e, BlockStored) for e in stored)
+
+    pool.free_blocks(blocks)
+    assert pool.take_events() == []
+
+
+def test_kv_transfer_should_emit_block_inactive():
+    from vllm.config.kv_transfer import KVTransferConfig
+
+    assert KVTransferConfig(kv_role="kv_producer").should_emit_block_inactive is False
+    assert KVTransferConfig(kv_role="kv_consumer").should_emit_block_inactive is True
+    assert KVTransferConfig(kv_role="kv_both").should_emit_block_inactive is True
+    assert KVTransferConfig().should_emit_block_inactive is True
+
+
 def test_eagle_enabled_removes_last_block():
     """Verify Eagle does NOT remove blocks when request
     length is divisible by block size."""

vllm/config/kv_transfer.py

@@ -117,5 +117,15 @@ def is_kv_producer(self) -> bool:
     def is_kv_consumer(self) -> bool:
         return self.kv_connector is not None and self.kv_role in get_args(KVConsumer)
 
+    @property
+    def should_emit_block_inactive(self) -> bool:
+        """Whether this instance should emit BlockInactive KV events.
+
+        Prefill-only producers (``kv_role='kv_producer'``) skip BlockInactive
+        so decode affinity load signals are not driven by prefill frees.
+        Decode (``kv_consumer``), union (``kv_both``), and non-PD setups emit.
+        """
+        return self.kv_role != "kv_producer"
+
     def get_from_extra_config(self, key, default) -> Any:
         return self.kv_connector_extra_config.get(key, default)

vllm/distributed/kv_events.py

@@ -112,12 +112,39 @@ def __hash__(self) -> int:
         )
 
 
+class BlockInactive(KVCacheEvent):
+    """Emitted when a block's ref_cnt drops to zero.
+
+    Distinct from BlockRemoved:
+    - BlockRemoved: the block is evicted/removed from GPU memory.
+    - BlockInactive: the block still exists in GPU memory but has zero active
+      references (all requests referencing it have completed).
+
+    This event is only emitted for GPU blocks (medium=MEDIUM_GPU).
+    """
+
+    block_hashes: list[ExternalBlockHash]
+    medium: str | None
+    group_idx: int | None = None
+    locality: str | None = None
+
+    def __hash__(self) -> int:
+        return hash(
+            (
+                tuple(self.block_hashes),
+                self.medium,
+                self.group_idx,
+                self.locality,
+            )
+        )
+
+
 class AllBlocksCleared(KVCacheEvent):
     pass
 
 
 class KVEventBatch(EventBatch):
-    events: list[BlockStored | BlockRemoved | AllBlocksCleared]
+    events: list[BlockStored | BlockRemoved | BlockInactive | AllBlocksCleared]
 
 
 class KVEventAggregator:

vllm/v1/core/block_pool.py

@@ -6,6 +6,7 @@
 from vllm.distributed.kv_events import (
     MEDIUM_GPU,
     AllBlocksCleared,
+    BlockInactive,
     BlockRemoved,
     BlockStored,
     KVCacheEvent,
@@ -156,6 +157,10 @@ class BlockPool:
             where different KV cache groups have different block sizes, the
             actual block size can be a multiple of hash_block_size.
         enable_kv_cache_events: Whether to enable kv cache events.
+        enable_block_inactive_events: Whether to emit BlockInactive when a
+            block's ref_cnt drops to 0. Prefill-only (kv_producer) instances
+            should disable this so decode load signals are not driven by
+            prefill free_blocks; BlockStored/BlockRemoved are unaffected.
         metrics_collector: Optional metrics collector for tracking block residency.
     """
 
@@ -166,6 +171,7 @@ def __init__(
         hash_block_size: int,
         enable_kv_cache_events: bool = False,
         metrics_collector: KVCacheMetricsCollector | None = None,
+        enable_block_inactive_events: bool = True,
     ):
         assert isinstance(num_gpu_blocks, int) and num_gpu_blocks > 0
         self.num_gpu_blocks = num_gpu_blocks
@@ -191,6 +197,9 @@ def __init__(
         self.null_block.is_null = True
 
         self.enable_kv_cache_events = enable_kv_cache_events
+        # Gated separately from enable_kv_cache_events so prefill can still
+        # publish Stored/Removed while omitting Inactive.
+        self.enable_block_inactive_events = enable_block_inactive_events
         self.kv_event_queue: list[KVCacheEvent] = []
 
         self.metrics_collector = metrics_collector
@@ -727,16 +736,38 @@ def free_blocks(self, ordered_blocks: Iterable[KVCacheBlock]) -> None:
         # Identify blocks with hash (LRU cache) and without it (never match APC)
         blocks_with_hash = []
         blocks_without_hash = []
+        inactive_block_hashes: list[ExternalBlockHash] = []
         for block in ordered_blocks:
             block.ref_cnt -= 1
             if block.ref_cnt == 0 and not block.is_null:
+                # Emit BlockInactive for GPU blocks that had a hash
+                # (prefix-cached blocks) when ref_cnt drops to zero.
+                # Prefill-only (kv_producer) disables enable_block_inactive_events.
+                if (
+                    self.enable_kv_cache_events
+                    and self.enable_block_inactive_events
+                    and block.block_hash is not None
+                ):
+                    inactive_block_hashes.append(
+                        maybe_convert_block_hash(
+                            get_block_hash(block.block_hash)
+                        )
+                    )
                 # When caching is disabled we always append for better
                 # GPU cache locality from reusing recently used blocks
                 if block.block_hash is None and self.enable_caching:
                     blocks_without_hash.append(block)
                 else:
                     blocks_with_hash.append(block)
 
+        if inactive_block_hashes:
+            self.kv_event_queue.append(
+                BlockInactive(
+                    block_hashes=inactive_block_hashes,
+                    medium=MEDIUM_GPU,
+                )
+            )
+
         # Blocks without hash get evicted first - prepend them last to the tail
         self.free_block_queue.prepend_n(blocks_without_hash)
         self.free_block_queue.append_n(blocks_with_hash)