From fe2d795f2deee3c7084df442fa673871176d8748 Mon Sep 17 00:00:00 2001
From: key4ng <rukeyang@gmail.com>
Date: Wed, 10 Jun 2026 15:28:17 -0700
Subject: [PATCH 01/14] feat(grpc): add vLLM KV-event conversion helpers

Signed-off-by: key4ng <rukeyang@gmail.com>
---
 .../smg_grpc_servicer/vllm/kv_events.py       | 177 ++++++++++++++++
 grpc_servicer/tests/test_vllm_kv_events.py    | 190 ++++++++++++++++++
 2 files changed, 367 insertions(+)
 create mode 100644 grpc_servicer/smg_grpc_servicer/vllm/kv_events.py
 create mode 100644 grpc_servicer/tests/test_vllm_kv_events.py

diff --git a/grpc_servicer/smg_grpc_servicer/vllm/kv_events.py b/grpc_servicer/smg_grpc_servicer/vllm/kv_events.py
new file mode 100644
index 0000000000..fcdd603dc8
--- /dev/null
+++ b/grpc_servicer/smg_grpc_servicer/vllm/kv_events.py
@@ -0,0 +1,177 @@
+"""vLLM KV-cache-event → proto conversion and ZMQ streaming.
+
+Engine-free: imports only stdlib + the generated proto. vLLM event objects are
+dispatched by class name (BlockStored / BlockRemoved / AllBlocksCleared) so this
+module needs no vLLM import and is unit-testable without a vLLM install.
+
+Mirrors grpc_servicer/smg_grpc_servicer/sglang/servicer.py:652-779.
+"""
+
+import asyncio
+import logging
+from collections.abc import AsyncIterator, Awaitable, Callable
+
+from smg_grpc_proto.generated import common_pb2
+
+logger = logging.getLogger(__name__)
+
+_U64_MASK = 0xFFFFFFFFFFFFFFFF
+_I64_SIGN_BIT = 0x8000000000000000
+_U64_MODULUS = 0x10000000000000000
+
+
+def to_int64(value: int) -> int:
+    """Reduce an arbitrary-width int (e.g. a vLLM block hash) to a signed int64.
+
+    The proto block_hash field is int64; vLLM hashes can be wider. SMG uses the
+    hash only as a node identity and re-derives a content hash from token_ids
+    (kv_event_monitor.rs:592-597), so a deterministic 64-bit reduction is safe.
+    """
+    masked = value & _U64_MASK
+    if masked >= _I64_SIGN_BIT:
+        masked -= _U64_MODULUS
+    return masked
+
+
+def endpoint_for_rank(endpoint: str, dp_rank: int) -> str:
+    """Resolve a vLLM KV-events PUB endpoint to a connectable SUB address.
+
+    The publisher binds to e.g. ``tcp://*:5557``; subscribers connect to
+    ``127.0.0.1``. For data-parallel deployments each rank publishes on
+    ``base_port + dp_rank``. Non-tcp endpoints (ipc://, inproc://) are returned
+    with the bind wildcard substituted but no port arithmetic.
+    """
+    resolved = endpoint.replace("*", "127.0.0.1")
+    if resolved.startswith("tcp://") and dp_rank:
+        host, _, port = resolved.rpartition(":")
+        return f"{host}:{int(port) + dp_rank}"
+    return resolved
+
+
+def resolve_kv_events_config(engine: object):
+    """Return the vLLM KVEventsConfig iff ZMQ event publishing is enabled, else None.
+
+    Reads ``engine.vllm_config.kv_events_config`` (same access pattern as
+    servicer.py:416 for kv_transfer_config). Uses getattr throughout so it works
+    against any vLLM version and is testable with a fake engine.
+    """
+    vllm_config = getattr(engine, "vllm_config", None)
+    cfg = getattr(vllm_config, "kv_events_config", None)
+    if cfg is None:
+        return None
+    if not getattr(cfg, "enable_kv_cache_events", False):
+        logger.info("vLLM KV cache events not enabled; SubscribeKvEvents disabled")
+        return None
+    if getattr(cfg, "publisher", None) != "zmq":
+        logger.info(
+            "vLLM KV events publisher is %r, not 'zmq'; SubscribeKvEvents disabled",
+            getattr(cfg, "publisher", None),
+        )
+        return None
+    logger.info("vLLM KV events enabled: endpoint=%s", getattr(cfg, "endpoint", "?"))
+    return cfg
+
+
+def convert_event(event: object, event_id: int) -> common_pb2.KvCacheEvent | None:
+    """Convert one decoded vLLM event to a proto KvCacheEvent (or None if unknown)."""
+    name = type(event).__name__
+
+    if name == "BlockStored":
+        block_size = int(event.block_size)
+        blocks = []
+        for i, block_hash in enumerate(event.block_hashes):
+            start = i * block_size
+            end = start + block_size
+            block = common_pb2.KvBlock(
+                block_hash=to_int64(block_hash),
+                token_ids=list(event.token_ids[start:end]),
+                block_size=block_size,
+            )
+            lora_id = getattr(event, "lora_id", None)
+            if lora_id is not None:
+                block.lora_id = to_int64(lora_id)
+            blocks.append(block)
+        stored = common_pb2.KvBlocksStored(blocks=blocks)
+        parent = getattr(event, "parent_block_hash", None)
+        if parent is not None:
+            stored.parent_block_hash = to_int64(parent)
+        return common_pb2.KvCacheEvent(event_id=event_id, stored=stored)
+
+    if name == "BlockRemoved":
+        return common_pb2.KvCacheEvent(
+            event_id=event_id,
+            removed=common_pb2.KvBlocksRemoved(
+                block_hashes=[to_int64(h) for h in event.block_hashes]
+            ),
+        )
+
+    if name == "AllBlocksCleared":
+        return common_pb2.KvCacheEvent(event_id=event_id, cleared=common_pb2.KvCacheCleared())
+
+    logger.debug("Unknown vLLM KV event type %r, skipping", name)
+    return None
+
+
+def convert_batch(
+    raw_batch: object, seq_num: int, event_id_start: int
+) -> tuple[common_pb2.KvEventBatch, int]:
+    """Convert a decoded vLLM KVEventBatch to a proto KvEventBatch.
+
+    Returns the proto batch and the new event-id counter. The counter advances
+    once per input event (even if unconvertible) so ids stay monotonic.
+    """
+    proto = common_pb2.KvEventBatch(sequence_number=seq_num, timestamp=raw_batch.ts)
+    dp_rank = getattr(raw_batch, "data_parallel_rank", None)
+    if dp_rank is not None:
+        proto.dp_rank = dp_rank
+
+    event_id = event_id_start
+    for event in raw_batch.events:
+        event_id += 1
+        proto_event = convert_event(event, event_id)
+        if proto_event is not None:
+            proto.events.append(proto_event)
+    return proto, event_id
+
+
+async def stream_kv_events(
+    sub_socket: object,
+    decode: Callable[[bytes], object],
+    send_initial_metadata: Callable[[], Awaitable[None]],
+    is_cancelled: Callable[[], bool],
+    *,
+    recv_timeout: float = 1.0,
+) -> AsyncIterator[common_pb2.KvEventBatch]:
+    """Core ZMQ→proto streaming loop, decoupled from vLLM and gRPC types.
+
+    Args:
+        sub_socket: a connected ``zmq.asyncio`` SUB socket (duck-typed; only
+            ``recv_multipart()`` is used).
+        decode: bytes → decoded vLLM batch (e.g. ``msgspec.msgpack.Decoder(KVEventBatch).decode``).
+        send_initial_metadata: awaitable called once before the first recv so the
+            gRPC client's ``subscribe_kv_events().await`` resolves promptly.
+        is_cancelled: returns True when the RPC is cancelled; loop then exits.
+        recv_timeout: per-recv timeout so cancellation is observed even when idle.
+
+    Yields proto KvEventBatch using the ZMQ publisher's native sequence numbers.
+    """
+    await send_initial_metadata()
+    event_id = 0
+    while not is_cancelled():
+        try:
+            frames = await asyncio.wait_for(sub_socket.recv_multipart(), timeout=recv_timeout)
+        except TimeoutError:
+            continue
+
+        # ZMQ multipart: [topic, 8-byte big-endian seq, msgpack payload].
+        if len(frames) < 3:
+            continue
+        zmq_seq = int.from_bytes(frames[1], "big")
+        try:
+            raw_batch = decode(frames[2])
+        except Exception as e:  # noqa: BLE001 - one bad frame must not kill the stream
+            logger.warning("Failed to decode vLLM KV event batch: %s", e)
+            continue
+
+        proto_batch, event_id = convert_batch(raw_batch, zmq_seq, event_id)
+        yield proto_batch
diff --git a/grpc_servicer/tests/test_vllm_kv_events.py b/grpc_servicer/tests/test_vllm_kv_events.py
new file mode 100644
index 0000000000..7b1da3aa79
--- /dev/null
+++ b/grpc_servicer/tests/test_vllm_kv_events.py
@@ -0,0 +1,190 @@
+"""Unit tests for vLLM KV-event conversion helpers (engine-free, no vLLM required).
+
+Run with: pytest grpc_servicer/tests/test_vllm_kv_events.py -v
+"""
+
+import importlib.util
+from pathlib import Path
+
+import pytest
+
+pytest.importorskip("smg_grpc_proto")
+
+# Import the module directly to avoid pulling vllm via the package __init__.
+_MODULE_PATH = Path(__file__).parents[1] / "smg_grpc_servicer" / "vllm" / "kv_events.py"
+_spec = importlib.util.spec_from_file_location("vllm_kv_events", _MODULE_PATH)
+kv_events = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(kv_events)
+
+
+# --- Fake vLLM event objects (dispatch is by class name, so names matter) ---
+class BlockStored:
+    def __init__(self, block_hashes, parent_block_hash, token_ids, block_size, lora_id=None):
+        self.block_hashes = block_hashes
+        self.parent_block_hash = parent_block_hash
+        self.token_ids = token_ids
+        self.block_size = block_size
+        self.lora_id = lora_id
+
+
+class BlockRemoved:
+    def __init__(self, block_hashes):
+        self.block_hashes = block_hashes
+
+
+class AllBlocksCleared:
+    pass
+
+
+class KVEventBatch:
+    def __init__(self, ts, events, data_parallel_rank=None):
+        self.ts = ts
+        self.events = events
+        self.data_parallel_rank = data_parallel_rank
+
+
+class TestToInt64:
+    def test_small_positive_unchanged(self):
+        assert kv_events.to_int64(42) == 42
+
+    def test_value_above_i64_max_wraps_to_negative(self):
+        # 2**63 wraps to the minimum signed 64-bit value.
+        assert kv_events.to_int64(2**63) == -(2**63)
+
+    def test_full_256bit_hash_reduced_deterministically(self):
+        h = 0xDEADBEEFCAFEBABE_1122334455667788  # 128-bit
+        out = kv_events.to_int64(h)
+        assert -(2**63) <= out < 2**63
+        assert out == kv_events.to_int64(h)  # deterministic
+        # Equals the low 64 bits, reinterpreted as signed.
+        low = h & 0xFFFFFFFFFFFFFFFF
+        assert out == (low - 2**64 if low >= 2**63 else low)
+
+
+class TestEndpointForRank:
+    def test_bind_star_becomes_loopback(self):
+        assert kv_events.endpoint_for_rank("tcp://*:5557", 0) == "tcp://127.0.0.1:5557"
+
+    def test_rank_offsets_port(self):
+        assert kv_events.endpoint_for_rank("tcp://*:5557", 3) == "tcp://127.0.0.1:5560"
+
+    def test_explicit_host_preserved(self):
+        assert kv_events.endpoint_for_rank("tcp://10.0.0.2:6000", 0) == "tcp://10.0.0.2:6000"
+
+    def test_ipc_endpoint_passthrough(self):
+        assert kv_events.endpoint_for_rank("ipc:///tmp/kv", 0) == "ipc:///tmp/kv"
+
+
+class TestResolveKvEventsConfig:
+    class _Cfg:
+        def __init__(self, enable, publisher, endpoint="tcp://*:5557", topic=""):
+            self.enable_kv_cache_events = enable
+            self.publisher = publisher
+            self.endpoint = endpoint
+            self.topic = topic
+
+    class _Engine:
+        def __init__(self, cfg):
+            self.vllm_config = type("VC", (), {"kv_events_config": cfg})()
+
+    def test_none_when_no_vllm_config(self):
+        engine = type("E", (), {})()  # no vllm_config attr
+        assert kv_events.resolve_kv_events_config(engine) is None
+
+    def test_none_when_disabled(self):
+        engine = self._Engine(self._Cfg(enable=False, publisher="zmq"))
+        assert kv_events.resolve_kv_events_config(engine) is None
+
+    def test_none_when_publisher_not_zmq(self):
+        engine = self._Engine(self._Cfg(enable=True, publisher="null"))
+        assert kv_events.resolve_kv_events_config(engine) is None
+
+    def test_returns_cfg_when_enabled_zmq(self):
+        cfg = self._Cfg(enable=True, publisher="zmq")
+        engine = self._Engine(cfg)
+        assert kv_events.resolve_kv_events_config(engine) is cfg
+
+
+class TestConvertEvent:
+    def test_block_stored_single_block(self):
+        ev = BlockStored(
+            block_hashes=[111], parent_block_hash=None, token_ids=[1, 2, 3, 4], block_size=4
+        )
+        out = kv_events.convert_event(ev, event_id=7)
+        assert out.event_id == 7
+        assert out.WhichOneof("data") == "stored"
+        assert len(out.stored.blocks) == 1
+        assert out.stored.blocks[0].block_hash == 111
+        assert list(out.stored.blocks[0].token_ids) == [1, 2, 3, 4]
+        assert out.stored.blocks[0].block_size == 4
+        assert not out.stored.HasField("parent_block_hash")
+
+    def test_block_stored_multi_block_slices_tokens(self):
+        ev = BlockStored(
+            block_hashes=[10, 20],
+            parent_block_hash=9,
+            token_ids=[1, 2, 3, 4, 5, 6, 7, 8],
+            block_size=4,
+        )
+        out = kv_events.convert_event(ev, event_id=1)
+        assert [b.block_hash for b in out.stored.blocks] == [10, 20]
+        assert list(out.stored.blocks[0].token_ids) == [1, 2, 3, 4]
+        assert list(out.stored.blocks[1].token_ids) == [5, 6, 7, 8]
+        assert out.stored.parent_block_hash == 9
+
+    def test_block_stored_with_lora(self):
+        ev = BlockStored(
+            block_hashes=[1], parent_block_hash=None, token_ids=[1, 2], block_size=2, lora_id=5
+        )
+        out = kv_events.convert_event(ev, event_id=1)
+        assert out.stored.blocks[0].lora_id == 5
+
+    def test_block_removed(self):
+        out = kv_events.convert_event(BlockRemoved(block_hashes=[1, 2, 3]), event_id=2)
+        assert out.WhichOneof("data") == "removed"
+        assert list(out.removed.block_hashes) == [1, 2, 3]
+
+    def test_all_blocks_cleared(self):
+        out = kv_events.convert_event(AllBlocksCleared(), event_id=3)
+        assert out.WhichOneof("data") == "cleared"
+
+    def test_unknown_event_returns_none(self):
+        class Mystery:
+            pass
+
+        assert kv_events.convert_event(Mystery(), event_id=4) is None
+
+
+class TestConvertBatch:
+    def test_seq_timestamp_and_event_ids(self):
+        batch = KVEventBatch(
+            ts=12.5,
+            events=[
+                BlockStored(
+                    block_hashes=[1], parent_block_hash=None, token_ids=[1, 2], block_size=2
+                ),
+                BlockRemoved(block_hashes=[1]),
+            ],
+        )
+        proto, next_id = kv_events.convert_batch(batch, seq_num=99, event_id_start=0)
+        assert proto.sequence_number == 99
+        assert abs(proto.timestamp - 12.5) < 1e-9
+        assert [e.event_id for e in proto.events] == [1, 2]
+        assert next_id == 2
+        assert not proto.HasField("dp_rank")
+
+    def test_dp_rank_set_when_present(self):
+        batch = KVEventBatch(ts=1.0, events=[], data_parallel_rank=2)
+        proto, _ = kv_events.convert_batch(batch, seq_num=1, event_id_start=0)
+        assert proto.dp_rank == 2
+
+    def test_unknown_events_skipped_but_counter_advances(self):
+        class Mystery:
+            pass
+
+        batch = KVEventBatch(ts=1.0, events=[Mystery(), BlockRemoved(block_hashes=[5])])
+        proto, next_id = kv_events.convert_batch(batch, seq_num=1, event_id_start=10)
+        # One event emitted (the BlockRemoved), but the id counter advanced past both.
+        assert len(proto.events) == 1
+        assert proto.events[0].event_id == 12
+        assert next_id == 12

From 131f9b769a23d7f06ed86d176618d376c51b726e Mon Sep 17 00:00:00 2001
From: key4ng <rukeyang@gmail.com>
Date: Wed, 10 Jun 2026 15:38:06 -0700
Subject: [PATCH 02/14] fix(grpc): restore py3.10 timeout catch and harden vLLM
 KV-event helpers

Signed-off-by: key4ng <rukeyang@gmail.com>
---
 .../smg_grpc_servicer/vllm/kv_events.py          | 10 ++++++----
 grpc_servicer/tests/test_vllm_kv_events.py       | 16 ++++++++++++++++
 2 files changed, 22 insertions(+), 4 deletions(-)

diff --git a/grpc_servicer/smg_grpc_servicer/vllm/kv_events.py b/grpc_servicer/smg_grpc_servicer/vllm/kv_events.py
index fcdd603dc8..575d319da8 100644
--- a/grpc_servicer/smg_grpc_servicer/vllm/kv_events.py
+++ b/grpc_servicer/smg_grpc_servicer/vllm/kv_events.py
@@ -43,8 +43,9 @@ def endpoint_for_rank(endpoint: str, dp_rank: int) -> str:
     """
     resolved = endpoint.replace("*", "127.0.0.1")
     if resolved.startswith("tcp://") and dp_rank:
-        host, _, port = resolved.rpartition(":")
-        return f"{host}:{int(port) + dp_rank}"
+        host, sep, port = resolved.rpartition(":")
+        if sep and port.isdigit():
+            return f"{host}:{int(port) + dp_rank}"
     return resolved
 
 
@@ -146,7 +147,8 @@ async def stream_kv_events(
 
     Args:
         sub_socket: a connected ``zmq.asyncio`` SUB socket (duck-typed; only
-            ``recv_multipart()`` is used).
+            ``recv_multipart()`` is used). The caller owns the socket lifecycle
+            (this function never closes it).
         decode: bytes → decoded vLLM batch (e.g. ``msgspec.msgpack.Decoder(KVEventBatch).decode``).
         send_initial_metadata: awaitable called once before the first recv so the
             gRPC client's ``subscribe_kv_events().await`` resolves promptly.
@@ -160,7 +162,7 @@ async def stream_kv_events(
     while not is_cancelled():
         try:
             frames = await asyncio.wait_for(sub_socket.recv_multipart(), timeout=recv_timeout)
-        except TimeoutError:
+        except (TimeoutError, asyncio.TimeoutError):  # noqa: UP041 - distinct classes on py3.10
             continue
 
         # ZMQ multipart: [topic, 8-byte big-endian seq, msgpack payload].
diff --git a/grpc_servicer/tests/test_vllm_kv_events.py b/grpc_servicer/tests/test_vllm_kv_events.py
index 7b1da3aa79..71ca6161b4 100644
--- a/grpc_servicer/tests/test_vllm_kv_events.py
+++ b/grpc_servicer/tests/test_vllm_kv_events.py
@@ -74,6 +74,9 @@ def test_explicit_host_preserved(self):
     def test_ipc_endpoint_passthrough(self):
         assert kv_events.endpoint_for_rank("ipc:///tmp/kv", 0) == "ipc:///tmp/kv"
 
+    def test_portless_tcp_wildcard_no_crash(self):
+        assert kv_events.endpoint_for_rank("tcp://*", 2) == "tcp://127.0.0.1"
+
 
 class TestResolveKvEventsConfig:
     class _Cfg:
@@ -139,6 +142,19 @@ def test_block_stored_with_lora(self):
         out = kv_events.convert_event(ev, event_id=1)
         assert out.stored.blocks[0].lora_id == 5
 
+    def test_block_stored_wide_hashes_reduced_to_int64(self):
+        ev = BlockStored(
+            block_hashes=[2**63],
+            parent_block_hash=2**63,
+            token_ids=[1, 2],
+            block_size=2,
+            lora_id=2**63,
+        )
+        out = kv_events.convert_event(ev, event_id=1)
+        assert out.stored.blocks[0].block_hash == -(2**63)
+        assert out.stored.parent_block_hash == -(2**63)
+        assert out.stored.blocks[0].lora_id == -(2**63)
+
     def test_block_removed(self):
         out = kv_events.convert_event(BlockRemoved(block_hashes=[1, 2, 3]), event_id=2)
         assert out.WhichOneof("data") == "removed"

From 4b890d04ae24972adf2af5741ad32481323e63d9 Mon Sep 17 00:00:00 2001
From: key4ng <rukeyang@gmail.com>
Date: Wed, 10 Jun 2026 15:42:37 -0700
Subject: [PATCH 03/14] test(grpc): integration-test vLLM KV-event ZMQ
 streaming loop

Signed-off-by: key4ng <rukeyang@gmail.com>
---
 .../tests/test_vllm_kv_events_stream.py       | 119 ++++++++++++++++++
 1 file changed, 119 insertions(+)
 create mode 100644 grpc_servicer/tests/test_vllm_kv_events_stream.py

diff --git a/grpc_servicer/tests/test_vllm_kv_events_stream.py b/grpc_servicer/tests/test_vllm_kv_events_stream.py
new file mode 100644
index 0000000000..fc8236f09c
--- /dev/null
+++ b/grpc_servicer/tests/test_vllm_kv_events_stream.py
@@ -0,0 +1,119 @@
+"""Integration test for kv_events.stream_kv_events over a real ZMQ PUB/SUB pair.
+
+No vLLM required: the msgpack decoder is replaced with an injected fake.
+Run with: pytest grpc_servicer/tests/test_vllm_kv_events_stream.py -v
+"""
+
+import asyncio
+import importlib.util
+from pathlib import Path
+
+import pytest
+
+pytest.importorskip("smg_grpc_proto")
+zmq = pytest.importorskip("zmq")
+import zmq.asyncio  # noqa: E402, F811
+
+_MODULE_PATH = Path(__file__).parents[1] / "smg_grpc_servicer" / "vllm" / "kv_events.py"
+_spec = importlib.util.spec_from_file_location("vllm_kv_events", _MODULE_PATH)
+kv_events = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(kv_events)
+
+
+# Class name must be "BlockStored" for kv_events.convert_event dispatch.
+class BlockStored:
+    def __init__(self):
+        self.block_hashes = [1]
+        self.parent_block_hash = None
+        self.token_ids = [1, 2]
+        self.block_size = 2
+        self.lora_id = None
+
+
+class KVEventBatch:
+    def __init__(self, seq):
+        self.ts = float(seq)
+        self.events = [BlockStored()]
+        self.data_parallel_rank = None
+
+
+def _seq_bytes(n: int) -> bytes:
+    return n.to_bytes(8, "big")
+
+
+@pytest.mark.asyncio
+async def test_stream_yields_batches_in_order():
+    ctx = zmq.asyncio.Context.instance()
+    pub = ctx.socket(zmq.PUB)
+    port = pub.bind_to_random_port("tcp://127.0.0.1")
+    sub = ctx.socket(zmq.SUB)
+    sub.subscribe(b"")
+    sub.connect(f"tcp://127.0.0.1:{port}")
+    await asyncio.sleep(0.2)  # allow SUB connection to establish before publishing
+
+    # decode ignores payload bytes and returns a prebuilt fake batch keyed by seq.
+    def fake_decode(payload: bytes):
+        return KVEventBatch(int.from_bytes(payload, "big"))
+
+    collected = []
+
+    async def consume():
+        async for batch in kv_events.stream_kv_events(
+            sub, fake_decode, send_initial_metadata=_noop, is_cancelled=lambda: len(collected) >= 2
+        ):
+            collected.append(batch)
+
+    async def _noop():
+        return None
+
+    consumer = asyncio.create_task(consume())
+    for seq in (1, 2):
+        await pub.send_multipart([b"kv", _seq_bytes(seq), seq.to_bytes(8, "big")])
+        await asyncio.sleep(0.05)
+    await asyncio.wait_for(consumer, timeout=5)
+
+    pub.close(linger=0)
+    sub.close(linger=0)
+
+    assert [b.sequence_number for b in collected] == [1, 2]
+    assert collected[0].events[0].stored.blocks[0].block_hash == 1
+
+
+@pytest.mark.asyncio
+async def test_stream_skips_short_frames_and_decode_errors():
+    ctx = zmq.asyncio.Context.instance()
+    pub = ctx.socket(zmq.PUB)
+    port = pub.bind_to_random_port("tcp://127.0.0.1")
+    sub = ctx.socket(zmq.SUB)
+    sub.subscribe(b"")
+    sub.connect(f"tcp://127.0.0.1:{port}")
+    await asyncio.sleep(0.2)
+
+    def fake_decode(payload: bytes):
+        if payload == b"bad":
+            raise ValueError("boom")
+        return KVEventBatch(7)
+
+    collected = []
+
+    async def _noop():
+        return None
+
+    async def consume():
+        async for batch in kv_events.stream_kv_events(
+            sub, fake_decode, send_initial_metadata=_noop, is_cancelled=lambda: len(collected) >= 1
+        ):
+            collected.append(batch)
+
+    consumer = asyncio.create_task(consume())
+    await pub.send_multipart([b"kv", _seq_bytes(1)])  # short frame -> skipped
+    await asyncio.sleep(0.05)
+    await pub.send_multipart([b"kv", _seq_bytes(2), b"bad"])  # decode error -> skipped
+    await asyncio.sleep(0.05)
+    await pub.send_multipart([b"kv", _seq_bytes(3), b"ok"])  # good -> yielded
+    await asyncio.wait_for(consumer, timeout=5)
+
+    pub.close(linger=0)
+    sub.close(linger=0)
+
+    assert [b.sequence_number for b in collected] == [3]

From 4306d18f4b200093cf16631148e9cbe2937dcb67 Mon Sep 17 00:00:00 2001
From: key4ng <rukeyang@gmail.com>
Date: Wed, 10 Jun 2026 15:57:32 -0700
Subject: [PATCH 04/14] feat(grpc): implement SubscribeKvEvents in vLLM
 servicer

Signed-off-by: key4ng <rukeyang@gmail.com>
---
 .../smg_grpc_servicer/vllm/servicer.py        | 63 +++++++++++++++++++
 grpc_servicer/tests/test_vllm_kv_events.py    | 19 ++++++
 2 files changed, 82 insertions(+)

diff --git a/grpc_servicer/smg_grpc_servicer/vllm/servicer.py b/grpc_servicer/smg_grpc_servicer/vllm/servicer.py
index 4557b6c3d9..387ee73ea3 100755
--- a/grpc_servicer/smg_grpc_servicer/vllm/servicer.py
+++ b/grpc_servicer/smg_grpc_servicer/vllm/servicer.py
@@ -5,6 +5,7 @@
 Implements the VllmEngine gRPC service on top of vLLM's EngineClient.
 """
 
+import asyncio
 import hashlib
 import itertools
 import json
@@ -33,6 +34,11 @@
 from vllm.sampling_params import RequestOutputKind, StructuredOutputsParams
 
 from smg_grpc_servicer.tokenizer_bundle import CHUNK_SIZE, build_tokenizer_zip
+from smg_grpc_servicer.vllm.kv_events import (
+    endpoint_for_rank,
+    resolve_kv_events_config,
+    stream_kv_events,
+)
 from smg_grpc_servicer.vllm.kv_transfer import params_from_request, params_to_response_fields
 
 logger = init_logger(__name__)
@@ -136,6 +142,9 @@ def __init__(self, async_llm: EngineClient, start_time: float):
         """
         self.engine = async_llm
         self.start_time = start_time
+        # Resolve KV-event publishing config from the engine. Non-None only when
+        # vLLM was started with --kv-events-config enabling the ZMQ publisher.
+        self._kv_events_config = resolve_kv_events_config(async_llm)
         logger.info("VllmEngineServicer initialized")
 
     async def Generate(
@@ -993,3 +1002,57 @@ def _complete_response(
                 **stop_kwargs,
             ),
         )
+
+    async def SubscribeKvEvents(
+        self,
+        request: common_pb2.SubscribeKvEventsRequest,
+        context: grpc.aio.ServicerContext,
+    ) -> AsyncIterator[common_pb2.KvEventBatch]:
+        """Bridge vLLM's in-process ZMQ KV cache events to gRPC server-streaming.
+
+        Mirrors sglang/servicer.py:SubscribeKvEvents. Uses the ZMQ publisher's
+        native sequence numbers directly as gRPC sequence numbers.
+        """
+        if self._kv_events_config is None:
+            await context.abort(
+                grpc.StatusCode.UNIMPLEMENTED,
+                "KV cache events not enabled. Start vLLM with "
+                "--kv-events-config "
+                '\'{"enable_kv_cache_events": true, "publisher": "zmq"}\'',
+            )
+            return
+
+        import msgspec
+        import zmq
+        import zmq.asyncio
+        from vllm.distributed.kv_events import KVEventBatch
+
+        config = self._kv_events_config
+
+        # For DP attention each rank publishes on port + rank with independent
+        # sequence counters; subscribing to several on one socket interleaves
+        # them and breaks gap detection. Subscribe to rank 0 only for now.
+        # TODO(phase3): per-rank virtual workers or merged renumbering.
+        pub_endpoint = endpoint_for_rank(config.endpoint, 0)
+
+        zmq_ctx = zmq.asyncio.Context.instance()
+        sub_socket = zmq_ctx.socket(zmq.SUB)
+        sub_socket.subscribe(config.topic.encode("utf-8"))
+        sub_socket.connect(pub_endpoint)
+        logger.info("SubscribeKvEvents: connected to ZMQ endpoint %s", pub_endpoint)
+
+        decoder = msgspec.msgpack.Decoder(KVEventBatch)
+
+        try:
+            async for proto_batch in stream_kv_events(
+                sub_socket,
+                decoder.decode,
+                lambda: context.send_initial_metadata(()),
+                context.cancelled,
+            ):
+                yield proto_batch
+        except asyncio.CancelledError:
+            pass
+        finally:
+            sub_socket.close(linger=0)
+            logger.info("SubscribeKvEvents: stream closed")
diff --git a/grpc_servicer/tests/test_vllm_kv_events.py b/grpc_servicer/tests/test_vllm_kv_events.py
index 71ca6161b4..fc502bb1d2 100644
--- a/grpc_servicer/tests/test_vllm_kv_events.py
+++ b/grpc_servicer/tests/test_vllm_kv_events.py
@@ -204,3 +204,22 @@ class Mystery:
         assert len(proto.events) == 1
         assert proto.events[0].event_id == 12
         assert next_id == 12
+
+
+class TestServicerWiring:
+    def test_servicer_resolves_config_from_engine(self):
+        pytest.importorskip("vllm")
+        from smg_grpc_servicer.vllm.servicer import VllmEngineServicer
+
+        class _Cfg:
+            enable_kv_cache_events = True
+            publisher = "zmq"
+            endpoint = "tcp://*:5557"
+            topic = ""
+
+        class _Engine:
+            vllm_config = type("VC", (), {"kv_events_config": _Cfg()})()
+
+        servicer = VllmEngineServicer(_Engine(), start_time=0.0)
+        assert servicer._kv_events_config is not None
+        assert hasattr(servicer, "SubscribeKvEvents")

From c00baba1dde6e1aee0154291e68acf89a93e3803 Mon Sep 17 00:00:00 2001
From: key4ng <rukeyang@gmail.com>
Date: Wed, 10 Jun 2026 16:07:59 -0700
Subject: [PATCH 05/14] docs(grpc): document vLLM KV-events worker + pin
 zmq/msgspec deps

Signed-off-by: key4ng <rukeyang@gmail.com>
---
 docs/getting-started/kv-events-cache-aware.md | 20 +++++++++++++++++++
 grpc_servicer/pyproject.toml                  |  2 +-
 2 files changed, 21 insertions(+), 1 deletion(-)

diff --git a/docs/getting-started/kv-events-cache-aware.md b/docs/getting-started/kv-events-cache-aware.md
index 305af5e291..6db67681d5 100644
--- a/docs/getting-started/kv-events-cache-aware.md
+++ b/docs/getting-started/kv-events-cache-aware.md
@@ -91,6 +91,26 @@ All fields and defaults match SGLang's `KVEventsConfig` (see `python/sglang/srt/
 
 For data-parallel deployments, the actual TCP port becomes `endpoint_port + dp_rank` (rank 0 keeps the configured port).
 
+### Alternative: launch a vLLM worker
+
+vLLM emits KV cache events the same way (ZMQ publisher); enable them with `--kv-events-config` and run the worker in SMG gRPC mode:
+
+```bash
+pip install "smg-grpc-servicer[vllm]"
+
+vllm serve meta-llama/Llama-3.1-8B-Instruct \
+  --kv-events-config '{"enable_kv_cache_events": true, "publisher": "zmq", "endpoint": "tcp://*:5557", "topic": "kv-events"}'
+# ...launched in SMG gRPC mode, the same way you run vLLM gRPC workers today.
+```
+
+| Field | Why |
+|---|---|
+| `enable_kv_cache_events: true` | vLLM-specific master switch. Without it no events are published even if a publisher is set. |
+| `publisher: "zmq"` | Selects the ZMQ publisher the servicer bridges. |
+| `endpoint` / `topic` | Same meaning as SGLang. For data-parallel, the port is `endpoint_port + dp_rank`; SMG currently consumes rank 0. |
+
+Everything downstream (SMG flags, block-size learning, verification logs) is identical to the SGLang flow — `KvEventMonitor` is engine-agnostic.
+
 ---
 
 ## Step 2 — Launch SMG
diff --git a/grpc_servicer/pyproject.toml b/grpc_servicer/pyproject.toml
index 8b95e1faad..9864fdfe0e 100644
--- a/grpc_servicer/pyproject.toml
+++ b/grpc_servicer/pyproject.toml
@@ -30,7 +30,7 @@ classifiers = [
 ]
 
 [project.optional-dependencies]
-vllm = ["vllm>=0.19.0"]
+vllm = ["vllm>=0.19.0", "pyzmq>=25.0.0", "msgspec>=0.18.0"]
 sglang = ["sglang>=0.5.10"]
 # smg-grpc-proto>=0.4.7 is the first release that ships mlx_engine_pb2;
 # without this floor, installing [mlx] against an older proto build would

From db2652d862289c6476bb90bc9b2450a171183b99 Mon Sep 17 00:00:00 2001
From: key4ng <rukeyang@gmail.com>
Date: Wed, 10 Jun 2026 16:11:05 -0700
Subject: [PATCH 06/14] test(bench): add prefix-heavy KV-events cache-hit
 benchmark

Signed-off-by: key4ng <rukeyang@gmail.com>
---
 scripts/bench_kv_events_cache_hit.py | 116 +++++++++++++++++++++++++++
 1 file changed, 116 insertions(+)
 create mode 100644 scripts/bench_kv_events_cache_hit.py

diff --git a/scripts/bench_kv_events_cache_hit.py b/scripts/bench_kv_events_cache_hit.py
new file mode 100644
index 0000000000..e4c0733164
--- /dev/null
+++ b/scripts/bench_kv_events_cache_hit.py
@@ -0,0 +1,116 @@
+"""Prefix-heavy cache-aware routing benchmark for KV-event-driven routing.
+
+Generates N prefix groups, each a long shared prefix followed by M short unique
+suffixes, and replays them through SMG's OpenAI-compatible endpoint while
+measuring time-to-first-token (TTFT) and end-to-end latency. Run the same
+workload under different SMG --policy settings to compare.
+
+Usage:
+  pip install aiohttp
+  python scripts/bench_kv_events_cache_hit.py \
+    --base-url http://localhost:30000 --model meta-llama/Llama-3.1-8B-Instruct \
+    --prefix-groups 16 --suffixes-per-group 16 --concurrency 8 --label event-driven
+"""
+
+import argparse
+import asyncio
+import json
+import statistics
+import time
+
+import aiohttp
+
+
+def build_requests(prefix_groups: int, suffixes_per_group: int, prefix_tokens: int):
+    """Return a shuffled-by-construction list of (group_id, messages) requests."""
+    requests = []
+    for g in range(prefix_groups):
+        # A long, group-specific shared prefix (~prefix_tokens words).
+        shared = f"System context for group {g}: " + " ".join(
+            f"fact{g}_{i}" for i in range(prefix_tokens)
+        )
+        for s in range(suffixes_per_group):
+            messages = [
+                {"role": "system", "content": shared},
+                {"role": "user", "content": f"Question {s}: summarize in one word."},
+            ]
+            requests.append((g, messages))
+    # Interleave groups so consecutive requests don't trivially hit the same worker.
+    requests.sort(key=lambda gm: (gm[1][1]["content"], gm[0]))
+    return requests
+
+
+async def one_request(session, base_url, model, messages):
+    payload = {
+        "model": model,
+        "messages": messages,
+        "max_tokens": 8,
+        "stream": True,
+        "temperature": 0.0,
+    }
+    start = time.perf_counter()
+    ttft = None
+    async with session.post(f"{base_url}/v1/chat/completions", json=payload) as resp:
+        resp.raise_for_status()
+        async for line in resp.content:
+            if not line or not line.startswith(b"data:"):
+                continue
+            if ttft is None:
+                ttft = time.perf_counter() - start
+            if line.strip() == b"data: [DONE]":
+                break
+    total = time.perf_counter() - start
+    return ttft if ttft is not None else total, total
+
+
+async def run(args):
+    requests = build_requests(args.prefix_groups, args.suffixes_per_group, args.prefix_tokens)
+    sem = asyncio.Semaphore(args.concurrency)
+    ttfts, totals = [], []
+
+    async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=120)) as session:
+
+        async def worker(messages):
+            async with sem:
+                ttft, total = await one_request(session, args.base_url, args.model, messages)
+                ttfts.append(ttft)
+                totals.append(total)
+
+        wall_start = time.perf_counter()
+        await asyncio.gather(*(worker(m) for _, m in requests))
+        wall = time.perf_counter() - wall_start
+
+    def pct(xs, p):
+        return statistics.quantiles(xs, n=100)[p - 1] if len(xs) > 1 else xs[0]
+
+    print(
+        json.dumps(
+            {
+                "label": args.label,
+                "requests": len(requests),
+                "concurrency": args.concurrency,
+                "wall_seconds": round(wall, 3),
+                "throughput_rps": round(len(requests) / wall, 2),
+                "ttft_p50_ms": round(pct(ttfts, 50) * 1000, 1),
+                "ttft_p99_ms": round(pct(ttfts, 99) * 1000, 1),
+                "total_p50_ms": round(pct(totals, 50) * 1000, 1),
+            },
+            indent=2,
+        )
+    )
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--base-url", default="http://localhost:30000")
+    ap.add_argument("--model", required=True)
+    ap.add_argument("--prefix-groups", type=int, default=16)
+    ap.add_argument("--suffixes-per-group", type=int, default=16)
+    ap.add_argument("--prefix-tokens", type=int, default=400)
+    ap.add_argument("--concurrency", type=int, default=8)
+    ap.add_argument("--label", default="run")
+    asyncio.run(run(ap.parse_args()))
+
+
+if __name__ == "__main__":
+    main()

From e263cf4741cc4356bbf7f2f5f424ae86ce31014c Mon Sep 17 00:00:00 2001
From: key4ng <rukeyang@gmail.com>
Date: Wed, 10 Jun 2026 16:22:44 -0700
Subject: [PATCH 07/14] docs(grpc): clarify vLLM gRPC-mode requirement for KV
 events

Signed-off-by: key4ng <rukeyang@gmail.com>
---
 docs/getting-started/kv-events-cache-aware.md | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/docs/getting-started/kv-events-cache-aware.md b/docs/getting-started/kv-events-cache-aware.md
index 6db67681d5..9b64f2ada3 100644
--- a/docs/getting-started/kv-events-cache-aware.md
+++ b/docs/getting-started/kv-events-cache-aware.md
@@ -98,11 +98,14 @@ vLLM emits KV cache events the same way (ZMQ publisher); enable them with `--kv-
 ```bash
 pip install "smg-grpc-servicer[vllm]"
 
+# Launch vLLM as an SMG gRPC worker (see the gRPC Workers guide for the full
+# launch mechanics); the only KV-events-specific addition is --kv-events-config:
 vllm serve meta-llama/Llama-3.1-8B-Instruct \
   --kv-events-config '{"enable_kv_cache_events": true, "publisher": "zmq", "endpoint": "tcp://*:5557", "topic": "kv-events"}'
-# ...launched in SMG gRPC mode, the same way you run vLLM gRPC workers today.
 ```
 
+Event-driven routing requires the worker to run in **SMG gRPC mode** — see [gRPC Workers](grpc-workers.md) for the launch details; the snippet above shows only the KV-events-specific flag you add on top.
+
 | Field | Why |
 |---|---|
 | `enable_kv_cache_events: true` | vLLM-specific master switch. Without it no events are published even if a publisher is set. |

From 7fc376aa88821758534a91c86055555868d1db41 Mon Sep 17 00:00:00 2001
From: key4ng <rukeyang@gmail.com>
Date: Wed, 10 Jun 2026 16:23:29 -0700
Subject: [PATCH 08/14] fix(grpc): accept bytes block hashes in vLLM KV-event
 bridge

Signed-off-by: key4ng <rukeyang@gmail.com>
---
 .../smg_grpc_servicer/vllm/kv_events.py       | 15 +++++++++-----
 grpc_servicer/tests/test_vllm_kv_events.py    | 20 +++++++++++++++++++
 2 files changed, 30 insertions(+), 5 deletions(-)

diff --git a/grpc_servicer/smg_grpc_servicer/vllm/kv_events.py b/grpc_servicer/smg_grpc_servicer/vllm/kv_events.py
index 575d319da8..1589508c7f 100644
--- a/grpc_servicer/smg_grpc_servicer/vllm/kv_events.py
+++ b/grpc_servicer/smg_grpc_servicer/vllm/kv_events.py
@@ -20,13 +20,18 @@
 _U64_MODULUS = 0x10000000000000000
 
 
-def to_int64(value: int) -> int:
-    """Reduce an arbitrary-width int (e.g. a vLLM block hash) to a signed int64.
-
-    The proto block_hash field is int64; vLLM hashes can be wider. SMG uses the
-    hash only as a node identity and re-derives a content hash from token_ids
+def to_int64(value: int | bytes) -> int:
+    """Reduce a vLLM block hash to a signed int64.
+
+    vLLM's ExternalBlockHash is ``bytes | int``: int when
+    VLLM_KV_EVENTS_USE_INT_BLOCK_HASHES is set (the default), else sha256 bytes.
+    Bytes are interpreted big-endian (matching vLLM's own conversion) before the
+    64-bit reduction. The proto block_hash field is int64; SMG uses the hash only
+    as a node identity and re-derives a content hash from token_ids
     (kv_event_monitor.rs:592-597), so a deterministic 64-bit reduction is safe.
     """
+    if isinstance(value, (bytes, bytearray)):
+        value = int.from_bytes(value, "big")
     masked = value & _U64_MASK
     if masked >= _I64_SIGN_BIT:
         masked -= _U64_MODULUS
diff --git a/grpc_servicer/tests/test_vllm_kv_events.py b/grpc_servicer/tests/test_vllm_kv_events.py
index fc502bb1d2..1eb884c54d 100644
--- a/grpc_servicer/tests/test_vllm_kv_events.py
+++ b/grpc_servicer/tests/test_vllm_kv_events.py
@@ -60,6 +60,16 @@ def test_full_256bit_hash_reduced_deterministically(self):
         low = h & 0xFFFFFFFFFFFFFFFF
         assert out == (low - 2**64 if low >= 2**63 else low)
 
+    def test_bytes_hash_interpreted_big_endian(self):
+        assert kv_events.to_int64((1).to_bytes(8, "big")) == 1
+        assert kv_events.to_int64(b"\x00" * 8) == 0
+
+    def test_bytes_hash_wraps_to_int64_like_int_path(self):
+        # bytes path must produce the same identity as the equivalent int.
+        wide = 0xDEADBEEFCAFEBABE
+        assert kv_events.to_int64(wide.to_bytes(8, "big")) == kv_events.to_int64(wide)
+        assert kv_events.to_int64((2**63).to_bytes(8, "big")) == -(2**63)
+
 
 class TestEndpointForRank:
     def test_bind_star_becomes_loopback(self):
@@ -155,6 +165,16 @@ def test_block_stored_wide_hashes_reduced_to_int64(self):
         assert out.stored.parent_block_hash == -(2**63)
         assert out.stored.blocks[0].lora_id == -(2**63)
 
+    def test_block_stored_bytes_block_hash(self):
+        ev = BlockStored(
+            block_hashes=[(5).to_bytes(8, "big")],
+            parent_block_hash=None,
+            token_ids=[1, 2],
+            block_size=2,
+        )
+        out = kv_events.convert_event(ev, event_id=1)
+        assert out.stored.blocks[0].block_hash == 5
+
     def test_block_removed(self):
         out = kv_events.convert_event(BlockRemoved(block_hashes=[1, 2, 3]), event_id=2)
         assert out.WhichOneof("data") == "removed"

From 8e18d2f55ccff6b82f87b8c2b68274af8ca92a23 Mon Sep 17 00:00:00 2001
From: key4ng <rukeyang@gmail.com>
Date: Wed, 10 Jun 2026 20:21:40 -0700
Subject: [PATCH 09/14] fix(grpc): address review feedback on vLLM KV-event
 bridge

- stream_kv_events: poll() before recv_multipart() instead of asyncio.wait_for
  (cancelling a zmq.asyncio recv future can drop a dequeued message)
- endpoint_for_rank: rewrite 0.0.0.0 -> 127.0.0.1 (not connectable on macOS/Windows)
- SubscribeKvEvents: map unexpected errors to INTERNAL + logger.exception (repo
  convention); drop redundant return after context.abort()
- tests: close ZMQ sockets in try/finally; cover 0.0.0.0 endpoint
- bench: guard pct() against empty samples

Signed-off-by: key4ng <rukeyang@gmail.com>
---
 .../smg_grpc_servicer/vllm/kv_events.py       |  26 ++--
 .../smg_grpc_servicer/vllm/servicer.py        |   4 +-
 grpc_servicer/tests/test_vllm_kv_events.py    |   5 +
 .../tests/test_vllm_kv_events_stream.py       | 120 +++++++++---------
 scripts/bench_kv_events_cache_hit.py          |   2 +
 5 files changed, 88 insertions(+), 69 deletions(-)

diff --git a/grpc_servicer/smg_grpc_servicer/vllm/kv_events.py b/grpc_servicer/smg_grpc_servicer/vllm/kv_events.py
index 1589508c7f..b5018410e9 100644
--- a/grpc_servicer/smg_grpc_servicer/vllm/kv_events.py
+++ b/grpc_servicer/smg_grpc_servicer/vllm/kv_events.py
@@ -7,7 +7,6 @@
 Mirrors grpc_servicer/smg_grpc_servicer/sglang/servicer.py:652-779.
 """
 
-import asyncio
 import logging
 from collections.abc import AsyncIterator, Awaitable, Callable
 
@@ -42,11 +41,13 @@ def endpoint_for_rank(endpoint: str, dp_rank: int) -> str:
     """Resolve a vLLM KV-events PUB endpoint to a connectable SUB address.
 
     The publisher binds to e.g. ``tcp://*:5557``; subscribers connect to
-    ``127.0.0.1``. For data-parallel deployments each rank publishes on
-    ``base_port + dp_rank``. Non-tcp endpoints (ipc://, inproc://) are returned
-    with the bind wildcard substituted but no port arithmetic.
+    ``127.0.0.1``. Bind wildcards (``*`` and ``0.0.0.0``) are rewritten to
+    ``127.0.0.1`` since ``0.0.0.0`` is not a connectable address on macOS/Windows.
+    For data-parallel deployments each rank publishes on ``base_port + dp_rank``.
+    Non-tcp endpoints (ipc://, inproc://) are returned with the bind wildcard
+    substituted but no port arithmetic.
     """
-    resolved = endpoint.replace("*", "127.0.0.1")
+    resolved = endpoint.replace("*", "127.0.0.1").replace("0.0.0.0", "127.0.0.1")
     if resolved.startswith("tcp://") and dp_rank:
         host, sep, port = resolved.rpartition(":")
         if sep and port.isdigit():
@@ -152,23 +153,26 @@ async def stream_kv_events(
 
     Args:
         sub_socket: a connected ``zmq.asyncio`` SUB socket (duck-typed; only
-            ``recv_multipart()`` is used). The caller owns the socket lifecycle
-            (this function never closes it).
+            ``poll()`` and ``recv_multipart()`` are used). The caller owns the
+            socket lifecycle (this function never closes it).
         decode: bytes → decoded vLLM batch (e.g. ``msgspec.msgpack.Decoder(KVEventBatch).decode``).
         send_initial_metadata: awaitable called once before the first recv so the
             gRPC client's ``subscribe_kv_events().await`` resolves promptly.
         is_cancelled: returns True when the RPC is cancelled; loop then exits.
-        recv_timeout: per-recv timeout so cancellation is observed even when idle.
+        recv_timeout: poll timeout so cancellation is observed even when idle.
 
     Yields proto KvEventBatch using the ZMQ publisher's native sequence numbers.
     """
     await send_initial_metadata()
     event_id = 0
     while not is_cancelled():
-        try:
-            frames = await asyncio.wait_for(sub_socket.recv_multipart(), timeout=recv_timeout)
-        except (TimeoutError, asyncio.TimeoutError):  # noqa: UP041 - distinct classes on py3.10
+        # poll() is the safe way to bound the wait on a zmq.asyncio socket:
+        # cancelling a recv_multipart() future (e.g. via asyncio.wait_for) does
+        # not cancel the in-flight ZMQ recv and can drop an already-dequeued
+        # message. poll() reports readability without that side effect.
+        if not await sub_socket.poll(timeout=int(recv_timeout * 1000)):
             continue
+        frames = await sub_socket.recv_multipart()
 
         # ZMQ multipart: [topic, 8-byte big-endian seq, msgpack payload].
         if len(frames) < 3:
diff --git a/grpc_servicer/smg_grpc_servicer/vllm/servicer.py b/grpc_servicer/smg_grpc_servicer/vllm/servicer.py
index 387ee73ea3..b7ba672cf2 100755
--- a/grpc_servicer/smg_grpc_servicer/vllm/servicer.py
+++ b/grpc_servicer/smg_grpc_servicer/vllm/servicer.py
@@ -1020,7 +1020,6 @@ async def SubscribeKvEvents(
                 "--kv-events-config "
                 '\'{"enable_kv_cache_events": true, "publisher": "zmq"}\'',
             )
-            return
 
         import msgspec
         import zmq
@@ -1053,6 +1052,9 @@ async def SubscribeKvEvents(
                 yield proto_batch
         except asyncio.CancelledError:
             pass
+        except Exception as e:
+            logger.exception("SubscribeKvEvents failed")
+            await context.abort(grpc.StatusCode.INTERNAL, str(e))
         finally:
             sub_socket.close(linger=0)
             logger.info("SubscribeKvEvents: stream closed")
diff --git a/grpc_servicer/tests/test_vllm_kv_events.py b/grpc_servicer/tests/test_vllm_kv_events.py
index 1eb884c54d..ad8dceed05 100644
--- a/grpc_servicer/tests/test_vllm_kv_events.py
+++ b/grpc_servicer/tests/test_vllm_kv_events.py
@@ -87,6 +87,11 @@ def test_ipc_endpoint_passthrough(self):
     def test_portless_tcp_wildcard_no_crash(self):
         assert kv_events.endpoint_for_rank("tcp://*", 2) == "tcp://127.0.0.1"
 
+    def test_zero_host_becomes_loopback(self):
+        # 0.0.0.0 is not connectable on macOS/Windows; must rewrite to 127.0.0.1.
+        assert kv_events.endpoint_for_rank("tcp://0.0.0.0:5557", 0) == "tcp://127.0.0.1:5557"
+        assert kv_events.endpoint_for_rank("tcp://0.0.0.0:5557", 2) == "tcp://127.0.0.1:5559"
+
 
 class TestResolveKvEventsConfig:
     class _Cfg:
diff --git a/grpc_servicer/tests/test_vllm_kv_events_stream.py b/grpc_servicer/tests/test_vllm_kv_events_stream.py
index fc8236f09c..7f94678a37 100644
--- a/grpc_servicer/tests/test_vllm_kv_events_stream.py
+++ b/grpc_servicer/tests/test_vllm_kv_events_stream.py
@@ -45,35 +45,38 @@ def _seq_bytes(n: int) -> bytes:
 async def test_stream_yields_batches_in_order():
     ctx = zmq.asyncio.Context.instance()
     pub = ctx.socket(zmq.PUB)
-    port = pub.bind_to_random_port("tcp://127.0.0.1")
     sub = ctx.socket(zmq.SUB)
-    sub.subscribe(b"")
-    sub.connect(f"tcp://127.0.0.1:{port}")
-    await asyncio.sleep(0.2)  # allow SUB connection to establish before publishing
-
-    # decode ignores payload bytes and returns a prebuilt fake batch keyed by seq.
-    def fake_decode(payload: bytes):
-        return KVEventBatch(int.from_bytes(payload, "big"))
-
     collected = []
-
-    async def consume():
-        async for batch in kv_events.stream_kv_events(
-            sub, fake_decode, send_initial_metadata=_noop, is_cancelled=lambda: len(collected) >= 2
-        ):
-            collected.append(batch)
-
-    async def _noop():
-        return None
-
-    consumer = asyncio.create_task(consume())
-    for seq in (1, 2):
-        await pub.send_multipart([b"kv", _seq_bytes(seq), seq.to_bytes(8, "big")])
-        await asyncio.sleep(0.05)
-    await asyncio.wait_for(consumer, timeout=5)
-
-    pub.close(linger=0)
-    sub.close(linger=0)
+    try:
+        port = pub.bind_to_random_port("tcp://127.0.0.1")
+        sub.subscribe(b"")
+        sub.connect(f"tcp://127.0.0.1:{port}")
+        await asyncio.sleep(0.2)  # allow SUB connection to establish before publishing
+
+        # decode ignores payload bytes and returns a prebuilt fake batch keyed by seq.
+        def fake_decode(payload: bytes):
+            return KVEventBatch(int.from_bytes(payload, "big"))
+
+        async def _noop():
+            return None
+
+        async def consume():
+            async for batch in kv_events.stream_kv_events(
+                sub,
+                fake_decode,
+                send_initial_metadata=_noop,
+                is_cancelled=lambda: len(collected) >= 2,
+            ):
+                collected.append(batch)
+
+        consumer = asyncio.create_task(consume())
+        for seq in (1, 2):
+            await pub.send_multipart([b"kv", _seq_bytes(seq), seq.to_bytes(8, "big")])
+            await asyncio.sleep(0.05)
+        await asyncio.wait_for(consumer, timeout=5)
+    finally:
+        pub.close(linger=0)
+        sub.close(linger=0)
 
     assert [b.sequence_number for b in collected] == [1, 2]
     assert collected[0].events[0].stored.blocks[0].block_hash == 1
@@ -83,37 +86,40 @@ async def _noop():
 async def test_stream_skips_short_frames_and_decode_errors():
     ctx = zmq.asyncio.Context.instance()
     pub = ctx.socket(zmq.PUB)
-    port = pub.bind_to_random_port("tcp://127.0.0.1")
     sub = ctx.socket(zmq.SUB)
-    sub.subscribe(b"")
-    sub.connect(f"tcp://127.0.0.1:{port}")
-    await asyncio.sleep(0.2)
-
-    def fake_decode(payload: bytes):
-        if payload == b"bad":
-            raise ValueError("boom")
-        return KVEventBatch(7)
-
     collected = []
-
-    async def _noop():
-        return None
-
-    async def consume():
-        async for batch in kv_events.stream_kv_events(
-            sub, fake_decode, send_initial_metadata=_noop, is_cancelled=lambda: len(collected) >= 1
-        ):
-            collected.append(batch)
-
-    consumer = asyncio.create_task(consume())
-    await pub.send_multipart([b"kv", _seq_bytes(1)])  # short frame -> skipped
-    await asyncio.sleep(0.05)
-    await pub.send_multipart([b"kv", _seq_bytes(2), b"bad"])  # decode error -> skipped
-    await asyncio.sleep(0.05)
-    await pub.send_multipart([b"kv", _seq_bytes(3), b"ok"])  # good -> yielded
-    await asyncio.wait_for(consumer, timeout=5)
-
-    pub.close(linger=0)
-    sub.close(linger=0)
+    try:
+        port = pub.bind_to_random_port("tcp://127.0.0.1")
+        sub.subscribe(b"")
+        sub.connect(f"tcp://127.0.0.1:{port}")
+        await asyncio.sleep(0.2)
+
+        def fake_decode(payload: bytes):
+            if payload == b"bad":
+                raise ValueError("boom")
+            return KVEventBatch(7)
+
+        async def _noop():
+            return None
+
+        async def consume():
+            async for batch in kv_events.stream_kv_events(
+                sub,
+                fake_decode,
+                send_initial_metadata=_noop,
+                is_cancelled=lambda: len(collected) >= 1,
+            ):
+                collected.append(batch)
+
+        consumer = asyncio.create_task(consume())
+        await pub.send_multipart([b"kv", _seq_bytes(1)])  # short frame -> skipped
+        await asyncio.sleep(0.05)
+        await pub.send_multipart([b"kv", _seq_bytes(2), b"bad"])  # decode error -> skipped
+        await asyncio.sleep(0.05)
+        await pub.send_multipart([b"kv", _seq_bytes(3), b"ok"])  # good -> yielded
+        await asyncio.wait_for(consumer, timeout=5)
+    finally:
+        pub.close(linger=0)
+        sub.close(linger=0)
 
     assert [b.sequence_number for b in collected] == [3]
diff --git a/scripts/bench_kv_events_cache_hit.py b/scripts/bench_kv_events_cache_hit.py
index e4c0733164..d31aec7ad8 100644
--- a/scripts/bench_kv_events_cache_hit.py
+++ b/scripts/bench_kv_events_cache_hit.py
@@ -81,6 +81,8 @@ async def worker(messages):
         wall = time.perf_counter() - wall_start
 
     def pct(xs, p):
+        if not xs:
+            return 0.0
         return statistics.quantiles(xs, n=100)[p - 1] if len(xs) > 1 else xs[0]
 
     print(

From f8be95c51991eddb1258eb6e5f619e21278de372 Mon Sep 17 00:00:00 2001
From: key4ng <rukeyang@gmail.com>
Date: Wed, 10 Jun 2026 20:21:40 -0700
Subject: [PATCH 10/14] docs(grpc): use vLLM gRPC entrypoint in KV-events guide

Plain 'vllm serve' starts the HTTP server and cannot stream KV events; show
'python -m vllm.entrypoints.grpc_server' (the SMG gRPC worker entrypoint).

Signed-off-by: key4ng <rukeyang@gmail.com>
---
 docs/getting-started/kv-events-cache-aware.md | 11 +++++++----
 1 file changed, 7 insertions(+), 4 deletions(-)

diff --git a/docs/getting-started/kv-events-cache-aware.md b/docs/getting-started/kv-events-cache-aware.md
index 9b64f2ada3..5870230a73 100644
--- a/docs/getting-started/kv-events-cache-aware.md
+++ b/docs/getting-started/kv-events-cache-aware.md
@@ -98,13 +98,16 @@ vLLM emits KV cache events the same way (ZMQ publisher); enable them with `--kv-
 ```bash
 pip install "smg-grpc-servicer[vllm]"
 
-# Launch vLLM as an SMG gRPC worker (see the gRPC Workers guide for the full
-# launch mechanics); the only KV-events-specific addition is --kv-events-config:
-vllm serve meta-llama/Llama-3.1-8B-Instruct \
+# Use vLLM's gRPC entrypoint (NOT `vllm serve`, which starts the HTTP server and
+# cannot stream KV events); add --kv-events-config to turn on event publishing:
+python -m vllm.entrypoints.grpc_server \
+  --model meta-llama/Llama-3.1-8B-Instruct \
+  --host 0.0.0.0 \
+  --port 50051 \
   --kv-events-config '{"enable_kv_cache_events": true, "publisher": "zmq", "endpoint": "tcp://*:5557", "topic": "kv-events"}'
 ```
 
-Event-driven routing requires the worker to run in **SMG gRPC mode** — see [gRPC Workers](grpc-workers.md) for the launch details; the snippet above shows only the KV-events-specific flag you add on top.
+Event-driven routing requires the worker to run in **SMG gRPC mode** — KV events are streamed over the `SubscribeKvEvents` gRPC RPC, so a plain `vllm serve` HTTP worker cannot participate. See [gRPC Workers](grpc-workers.md) for the full launch reference; the snippet above adds only the KV-events-specific `--kv-events-config` flag on top.
 
 | Field | Why |
 |---|---|

From 847fe902c0cc0e64246c4420486188d1b446a170 Mon Sep 17 00:00:00 2001
From: key4ng <rukeyang@gmail.com>
Date: Thu, 11 Jun 2026 15:12:44 -0700
Subject: [PATCH 11/14] chore(grpc): tighten vLLM KV-event comments/docs and
 drop benchmark script

- Trim verbose docstrings/comments in kv_events.py and servicer.py and remove
  cross-references to the SGLang servicer.
- Reword the vLLM docs section to stand on its own.
- Remove scripts/bench_kv_events_cache_hit.py.

Signed-off-by: key4ng <rukeyang@gmail.com>
---
 docs/getting-started/kv-events-cache-aware.md |  11 +-
 .../smg_grpc_servicer/vllm/kv_events.py       |  42 +++----
 .../smg_grpc_servicer/vllm/servicer.py        |   6 +-
 scripts/bench_kv_events_cache_hit.py          | 118 ------------------
 4 files changed, 24 insertions(+), 153 deletions(-)
 delete mode 100644 scripts/bench_kv_events_cache_hit.py

diff --git a/docs/getting-started/kv-events-cache-aware.md b/docs/getting-started/kv-events-cache-aware.md
index 5870230a73..20045cbc4a 100644
--- a/docs/getting-started/kv-events-cache-aware.md
+++ b/docs/getting-started/kv-events-cache-aware.md
@@ -93,13 +93,12 @@ For data-parallel deployments, the actual TCP port becomes `endpoint_port + dp_r
 
 ### Alternative: launch a vLLM worker
 
-vLLM emits KV cache events the same way (ZMQ publisher); enable them with `--kv-events-config` and run the worker in SMG gRPC mode:
+vLLM publishes KV cache events on a ZMQ socket; enable them with `--kv-events-config` and run the worker in SMG gRPC mode:
 
 ```bash
 pip install "smg-grpc-servicer[vllm]"
 
-# Use vLLM's gRPC entrypoint (NOT `vllm serve`, which starts the HTTP server and
-# cannot stream KV events); add --kv-events-config to turn on event publishing:
+# vLLM's gRPC entrypoint (not `vllm serve`, which is HTTP and can't stream KV events):
 python -m vllm.entrypoints.grpc_server \
   --model meta-llama/Llama-3.1-8B-Instruct \
   --host 0.0.0.0 \
@@ -107,15 +106,15 @@ python -m vllm.entrypoints.grpc_server \
   --kv-events-config '{"enable_kv_cache_events": true, "publisher": "zmq", "endpoint": "tcp://*:5557", "topic": "kv-events"}'
 ```
 
-Event-driven routing requires the worker to run in **SMG gRPC mode** — KV events are streamed over the `SubscribeKvEvents` gRPC RPC, so a plain `vllm serve` HTTP worker cannot participate. See [gRPC Workers](grpc-workers.md) for the full launch reference; the snippet above adds only the KV-events-specific `--kv-events-config` flag on top.
+Event-driven routing needs the worker in **SMG gRPC mode** — KV events stream over the `SubscribeKvEvents` RPC, so a plain `vllm serve` HTTP worker can't participate. See [gRPC Workers](grpc-workers.md) for the full launch reference.
 
 | Field | Why |
 |---|---|
 | `enable_kv_cache_events: true` | vLLM-specific master switch. Without it no events are published even if a publisher is set. |
 | `publisher: "zmq"` | Selects the ZMQ publisher the servicer bridges. |
-| `endpoint` / `topic` | Same meaning as SGLang. For data-parallel, the port is `endpoint_port + dp_rank`; SMG currently consumes rank 0. |
+| `endpoint` / `topic` | ZMQ `PUB` address and topic prefix. For data-parallel, the port is `endpoint_port + dp_rank`; SMG currently consumes rank 0. |
 
-Everything downstream (SMG flags, block-size learning, verification logs) is identical to the SGLang flow — `KvEventMonitor` is engine-agnostic.
+Everything downstream — SMG flags, block-size learning, and the verification logs — is unchanged; `KvEventMonitor` consumes the events the same way for any gRPC worker.
 
 ---
 
diff --git a/grpc_servicer/smg_grpc_servicer/vllm/kv_events.py b/grpc_servicer/smg_grpc_servicer/vllm/kv_events.py
index b5018410e9..e758c7f454 100644
--- a/grpc_servicer/smg_grpc_servicer/vllm/kv_events.py
+++ b/grpc_servicer/smg_grpc_servicer/vllm/kv_events.py
@@ -1,10 +1,8 @@
 """vLLM KV-cache-event → proto conversion and ZMQ streaming.
 
-Engine-free: imports only stdlib + the generated proto. vLLM event objects are
-dispatched by class name (BlockStored / BlockRemoved / AllBlocksCleared) so this
-module needs no vLLM import and is unit-testable without a vLLM install.
-
-Mirrors grpc_servicer/smg_grpc_servicer/sglang/servicer.py:652-779.
+Engine-free: imports only stdlib + the generated proto, and dispatches vLLM
+events by class name (BlockStored / BlockRemoved / AllBlocksCleared), so it
+needs no vLLM import and is unit-testable without vLLM installed.
 """
 
 import logging
@@ -20,14 +18,11 @@
 
 
 def to_int64(value: int | bytes) -> int:
-    """Reduce a vLLM block hash to a signed int64.
-
-    vLLM's ExternalBlockHash is ``bytes | int``: int when
-    VLLM_KV_EVENTS_USE_INT_BLOCK_HASHES is set (the default), else sha256 bytes.
-    Bytes are interpreted big-endian (matching vLLM's own conversion) before the
-    64-bit reduction. The proto block_hash field is int64; SMG uses the hash only
-    as a node identity and re-derives a content hash from token_ids
-    (kv_event_monitor.rs:592-597), so a deterministic 64-bit reduction is safe.
+    """Reduce a vLLM block hash to a signed int64 for the proto block_hash field.
+
+    vLLM's block hash is ``int | bytes`` (sha256 bytes when int hashes are
+    disabled); bytes are read big-endian. SMG uses the hash only as a node
+    identity, so the 64-bit reduction is safe as long as it stays deterministic.
     """
     if isinstance(value, (bytes, bytearray)):
         value = int.from_bytes(value, "big")
@@ -40,12 +35,10 @@ def to_int64(value: int | bytes) -> int:
 def endpoint_for_rank(endpoint: str, dp_rank: int) -> str:
     """Resolve a vLLM KV-events PUB endpoint to a connectable SUB address.
 
-    The publisher binds to e.g. ``tcp://*:5557``; subscribers connect to
-    ``127.0.0.1``. Bind wildcards (``*`` and ``0.0.0.0``) are rewritten to
-    ``127.0.0.1`` since ``0.0.0.0`` is not a connectable address on macOS/Windows.
-    For data-parallel deployments each rank publishes on ``base_port + dp_rank``.
-    Non-tcp endpoints (ipc://, inproc://) are returned with the bind wildcard
-    substituted but no port arithmetic.
+    Bind wildcards (``*``, ``0.0.0.0``) are rewritten to ``127.0.0.1`` (the
+    latter is not connectable on macOS/Windows). For data-parallel deployments
+    each rank publishes on ``base_port + dp_rank``; non-tcp endpoints (ipc://,
+    inproc://) get the wildcard substituted but no port arithmetic.
     """
     resolved = endpoint.replace("*", "127.0.0.1").replace("0.0.0.0", "127.0.0.1")
     if resolved.startswith("tcp://") and dp_rank:
@@ -58,9 +51,8 @@ def endpoint_for_rank(endpoint: str, dp_rank: int) -> str:
 def resolve_kv_events_config(engine: object):
     """Return the vLLM KVEventsConfig iff ZMQ event publishing is enabled, else None.
 
-    Reads ``engine.vllm_config.kv_events_config`` (same access pattern as
-    servicer.py:416 for kv_transfer_config). Uses getattr throughout so it works
-    against any vLLM version and is testable with a fake engine.
+    Reads ``engine.vllm_config.kv_events_config`` via getattr so it tolerates any
+    vLLM version and is testable with a fake engine.
     """
     vllm_config = getattr(engine, "vllm_config", None)
     cfg = getattr(vllm_config, "kv_events_config", None)
@@ -166,10 +158,8 @@ async def stream_kv_events(
     await send_initial_metadata()
     event_id = 0
     while not is_cancelled():
-        # poll() is the safe way to bound the wait on a zmq.asyncio socket:
-        # cancelling a recv_multipart() future (e.g. via asyncio.wait_for) does
-        # not cancel the in-flight ZMQ recv and can drop an already-dequeued
-        # message. poll() reports readability without that side effect.
+        # poll() before recv: cancelling a zmq.asyncio recv future does not
+        # cancel the in-flight ZMQ recv and can drop an already-dequeued message.
         if not await sub_socket.poll(timeout=int(recv_timeout * 1000)):
             continue
         frames = await sub_socket.recv_multipart()
diff --git a/grpc_servicer/smg_grpc_servicer/vllm/servicer.py b/grpc_servicer/smg_grpc_servicer/vllm/servicer.py
index b7ba672cf2..d2e4fe22d0 100755
--- a/grpc_servicer/smg_grpc_servicer/vllm/servicer.py
+++ b/grpc_servicer/smg_grpc_servicer/vllm/servicer.py
@@ -1008,10 +1008,10 @@ async def SubscribeKvEvents(
         request: common_pb2.SubscribeKvEventsRequest,
         context: grpc.aio.ServicerContext,
     ) -> AsyncIterator[common_pb2.KvEventBatch]:
-        """Bridge vLLM's in-process ZMQ KV cache events to gRPC server-streaming.
+        """Bridge vLLM's in-process ZMQ KV cache events to a gRPC stream.
 
-        Mirrors sglang/servicer.py:SubscribeKvEvents. Uses the ZMQ publisher's
-        native sequence numbers directly as gRPC sequence numbers.
+        The ZMQ publisher's sequence numbers are used directly as the gRPC
+        batch sequence numbers.
         """
         if self._kv_events_config is None:
             await context.abort(
diff --git a/scripts/bench_kv_events_cache_hit.py b/scripts/bench_kv_events_cache_hit.py
deleted file mode 100644
index d31aec7ad8..0000000000
--- a/scripts/bench_kv_events_cache_hit.py
+++ /dev/null
@@ -1,118 +0,0 @@
-"""Prefix-heavy cache-aware routing benchmark for KV-event-driven routing.
-
-Generates N prefix groups, each a long shared prefix followed by M short unique
-suffixes, and replays them through SMG's OpenAI-compatible endpoint while
-measuring time-to-first-token (TTFT) and end-to-end latency. Run the same
-workload under different SMG --policy settings to compare.
-
-Usage:
-  pip install aiohttp
-  python scripts/bench_kv_events_cache_hit.py \
-    --base-url http://localhost:30000 --model meta-llama/Llama-3.1-8B-Instruct \
-    --prefix-groups 16 --suffixes-per-group 16 --concurrency 8 --label event-driven
-"""
-
-import argparse
-import asyncio
-import json
-import statistics
-import time
-
-import aiohttp
-
-
-def build_requests(prefix_groups: int, suffixes_per_group: int, prefix_tokens: int):
-    """Return a shuffled-by-construction list of (group_id, messages) requests."""
-    requests = []
-    for g in range(prefix_groups):
-        # A long, group-specific shared prefix (~prefix_tokens words).
-        shared = f"System context for group {g}: " + " ".join(
-            f"fact{g}_{i}" for i in range(prefix_tokens)
-        )
-        for s in range(suffixes_per_group):
-            messages = [
-                {"role": "system", "content": shared},
-                {"role": "user", "content": f"Question {s}: summarize in one word."},
-            ]
-            requests.append((g, messages))
-    # Interleave groups so consecutive requests don't trivially hit the same worker.
-    requests.sort(key=lambda gm: (gm[1][1]["content"], gm[0]))
-    return requests
-
-
-async def one_request(session, base_url, model, messages):
-    payload = {
-        "model": model,
-        "messages": messages,
-        "max_tokens": 8,
-        "stream": True,
-        "temperature": 0.0,
-    }
-    start = time.perf_counter()
-    ttft = None
-    async with session.post(f"{base_url}/v1/chat/completions", json=payload) as resp:
-        resp.raise_for_status()
-        async for line in resp.content:
-            if not line or not line.startswith(b"data:"):
-                continue
-            if ttft is None:
-                ttft = time.perf_counter() - start
-            if line.strip() == b"data: [DONE]":
-                break
-    total = time.perf_counter() - start
-    return ttft if ttft is not None else total, total
-
-
-async def run(args):
-    requests = build_requests(args.prefix_groups, args.suffixes_per_group, args.prefix_tokens)
-    sem = asyncio.Semaphore(args.concurrency)
-    ttfts, totals = [], []
-
-    async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=120)) as session:
-
-        async def worker(messages):
-            async with sem:
-                ttft, total = await one_request(session, args.base_url, args.model, messages)
-                ttfts.append(ttft)
-                totals.append(total)
-
-        wall_start = time.perf_counter()
-        await asyncio.gather(*(worker(m) for _, m in requests))
-        wall = time.perf_counter() - wall_start
-
-    def pct(xs, p):
-        if not xs:
-            return 0.0
-        return statistics.quantiles(xs, n=100)[p - 1] if len(xs) > 1 else xs[0]
-
-    print(
-        json.dumps(
-            {
-                "label": args.label,
-                "requests": len(requests),
-                "concurrency": args.concurrency,
-                "wall_seconds": round(wall, 3),
-                "throughput_rps": round(len(requests) / wall, 2),
-                "ttft_p50_ms": round(pct(ttfts, 50) * 1000, 1),
-                "ttft_p99_ms": round(pct(ttfts, 99) * 1000, 1),
-                "total_p50_ms": round(pct(totals, 50) * 1000, 1),
-            },
-            indent=2,
-        )
-    )
-
-
-def main():
-    ap = argparse.ArgumentParser()
-    ap.add_argument("--base-url", default="http://localhost:30000")
-    ap.add_argument("--model", required=True)
-    ap.add_argument("--prefix-groups", type=int, default=16)
-    ap.add_argument("--suffixes-per-group", type=int, default=16)
-    ap.add_argument("--prefix-tokens", type=int, default=400)
-    ap.add_argument("--concurrency", type=int, default=8)
-    ap.add_argument("--label", default="run")
-    asyncio.run(run(ap.parse_args()))
-
-
-if __name__ == "__main__":
-    main()

From 2bbfd54f036b8b6a538b909e67b95c114f4b0ccf Mon Sep 17 00:00:00 2001
From: key4ng <rukeyang@gmail.com>
Date: Thu, 11 Jun 2026 15:21:19 -0700
Subject: [PATCH 12/14] refactor(grpc): move vLLM KV-event imports to module
 top

Per review: hoist msgspec/zmq/KVEventBatch out of SubscribeKvEvents to the
top of servicer.py (the module already imports vllm at import time).

Signed-off-by: key4ng <rukeyang@gmail.com>
---
 grpc_servicer/smg_grpc_servicer/vllm/servicer.py | 9 ++++-----
 1 file changed, 4 insertions(+), 5 deletions(-)

diff --git a/grpc_servicer/smg_grpc_servicer/vllm/servicer.py b/grpc_servicer/smg_grpc_servicer/vllm/servicer.py
index d2e4fe22d0..384bd580ba 100755
--- a/grpc_servicer/smg_grpc_servicer/vllm/servicer.py
+++ b/grpc_servicer/smg_grpc_servicer/vllm/servicer.py
@@ -15,11 +15,15 @@
 from pathlib import Path
 
 import grpc
+import msgspec
 import torch
+import zmq
+import zmq.asyncio
 from smg_grpc_proto import vllm_engine_pb2, vllm_engine_pb2_grpc
 from smg_grpc_proto.generated import common_pb2
 from transformers import BatchFeature
 from vllm import PoolingParams, SamplingParams, TokensPrompt
+from vllm.distributed.kv_events import KVEventBatch
 from vllm.engine.protocol import EngineClient
 from vllm.inputs.engine import MultiModalInput as VllmMultiModalInput
 from vllm.inputs.engine import mm_input, tokens_input
@@ -1021,11 +1025,6 @@ async def SubscribeKvEvents(
                 '\'{"enable_kv_cache_events": true, "publisher": "zmq"}\'',
             )
 
-        import msgspec
-        import zmq
-        import zmq.asyncio
-        from vllm.distributed.kv_events import KVEventBatch
-
         config = self._kv_events_config
 
         # For DP attention each rank publishes on port + rank with independent

From 7521d9cb7970cea04de705dbd0e5d7fcb381fade Mon Sep 17 00:00:00 2001
From: key4ng <rukeyang@gmail.com>
Date: Thu, 11 Jun 2026 15:29:32 -0700
Subject: [PATCH 13/14] docs(grpc): launch vLLM KV-events worker with 'vllm
 serve --grpc'
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Matches grpc_servicer/README.md — the --grpc flag loads smg-grpc-servicer, so the worker serves gRPC (and KV events) rather than HTTP.

Signed-off-by: key4ng <rukeyang@gmail.com>
---
 docs/getting-started/kv-events-cache-aware.md | 11 +++++------
 1 file changed, 5 insertions(+), 6 deletions(-)

diff --git a/docs/getting-started/kv-events-cache-aware.md b/docs/getting-started/kv-events-cache-aware.md
index 20045cbc4a..74c01d684c 100644
--- a/docs/getting-started/kv-events-cache-aware.md
+++ b/docs/getting-started/kv-events-cache-aware.md
@@ -98,15 +98,14 @@ vLLM publishes KV cache events on a ZMQ socket; enable them with `--kv-events-co
 ```bash
 pip install "smg-grpc-servicer[vllm]"
 
-# vLLM's gRPC entrypoint (not `vllm serve`, which is HTTP and can't stream KV events):
-python -m vllm.entrypoints.grpc_server \
-  --model meta-llama/Llama-3.1-8B-Instruct \
-  --host 0.0.0.0 \
-  --port 50051 \
+# `--grpc` runs vLLM in SMG gRPC mode (loads smg-grpc-servicer);
+# --kv-events-config turns on KV-event publishing:
+vllm serve meta-llama/Llama-3.1-8B-Instruct \
+  --grpc \
   --kv-events-config '{"enable_kv_cache_events": true, "publisher": "zmq", "endpoint": "tcp://*:5557", "topic": "kv-events"}'
 ```
 
-Event-driven routing needs the worker in **SMG gRPC mode** — KV events stream over the `SubscribeKvEvents` RPC, so a plain `vllm serve` HTTP worker can't participate. See [gRPC Workers](grpc-workers.md) for the full launch reference.
+Event-driven routing needs the worker in **SMG gRPC mode** (`--grpc`) — KV events stream over the `SubscribeKvEvents` RPC, so an HTTP worker can't participate. See [gRPC Workers](grpc-workers.md) for additional launch flags (host, port, TP size).
 
 | Field | Why |
 |---|---|

From 7854bcb8513b88885d9e9e19ccc3eb89d27841f9 Mon Sep 17 00:00:00 2001
From: key4ng <rukeyang@gmail.com>
Date: Thu, 11 Jun 2026 15:39:29 -0700
Subject: [PATCH 14/14] docs(grpc): note vLLM block-size is learned from events

Clarify SMG learns block size from BlockStored events; --block-size pins a non-default value.

Signed-off-by: key4ng <rukeyang@gmail.com>
---
 docs/getting-started/kv-events-cache-aware.md | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/docs/getting-started/kv-events-cache-aware.md b/docs/getting-started/kv-events-cache-aware.md
index 74c01d684c..f4f9d277b5 100644
--- a/docs/getting-started/kv-events-cache-aware.md
+++ b/docs/getting-started/kv-events-cache-aware.md
@@ -113,6 +113,8 @@ Event-driven routing needs the worker in **SMG gRPC mode** (`--grpc`) — KV eve
 | `publisher: "zmq"` | Selects the ZMQ publisher the servicer bridges. |
 | `endpoint` / `topic` | ZMQ `PUB` address and topic prefix. For data-parallel, the port is `endpoint_port + dp_rank`; SMG currently consumes rank 0. |
 
+SMG learns the block size from the `BlockStored` events themselves, so you needn't set it; pass vLLM's `--block-size N` (the analogue of SGLang's `--page-size`) only to pin a non-default value.
+
 Everything downstream — SMG flags, block-size learning, and the verification logs — is unchanged; `KvEventMonitor` consumes the events the same way for any gRPC worker.
 
 ---