From 12b30c286e17f69dc896703aa789e279774e387e Mon Sep 17 00:00:00 2001
From: Chang Su <8605658+CatherineSue@users.noreply.github.com>
Date: Wed, 2 Sep 2026 01:21:16 -0700
Subject: [PATCH 01/11] feat(vllm-grpc): Redis media-processing sidecar backend
 and reference sidecar

Add SMG_VLLM_MM_PROCESSOR=redis: the servicer hands each media_refs request
to a sidecar over Redis lists and rebuilds the sidecar's full-tensor result
into the same pre-rendered engine input the preprocessed path uses, so
vLLM's processor caches are never relied on (no inject_into_mm_cache, no
data=None).

Protocol (engine-free, smg_grpc_servicer/mm_sidecar_protocol.py): keys under
smg:mm:v1:{namespace} with namespace derived from (model, vLLM version,
dtype); a hello HASH the sidecar refreshes every 5 s (TTL 15 s) carrying its
fingerprint and accepted URL schemes; a jobs LIST (LLEN fast-fail above
SMG_VLLM_MM_SIDECAR_MAX_QUEUE); per-attempt res:{job_id} LISTs with a
120 s TTL. Jobs carry unexpanded prompt ids, URLs and a deadline; results
carry expanded ids, vLLM's mm_hashes, placeholders with is_embed masks and
MultiModalKwargsItem blobs from vLLM's MsgpackEncoder. Client-caused
failure codes map to INVALID_ARGUMENT, everything else to UNAVAILABLE.

The worker advertises mm_processor=redis only while probe() finds a live
hello whose fingerprint matches, and rejects results whose fingerprint
differs. The reference sidecar (python -m smg_grpc_servicer.vllm.mm_sidecar)
builds a GPU-free renderer the way vllm launch render does, fetches with
vLLM's MediaConnector and runs process_for_engine_async(skip_mm_cache=True).

Signed-off-by: Chang Su <8605658+CatherineSue@users.noreply.github.com>
---
 grpc_servicer/README.md                       |  20 ++
 grpc_servicer/pyproject.toml                  |   4 +-
 .../smg_grpc_servicer/mm_sidecar_protocol.py  | 187 ++++++++++++
 .../smg_grpc_servicer/vllm/mm_processor.py    | 252 ++++++++++++++-
 .../smg_grpc_servicer/vllm/mm_sidecar.py      | 288 ++++++++++++++++++
 .../smg_grpc_servicer/vllm/servicer.py        |   1 +
 grpc_servicer/tests/test_vllm_mm_processor.py |   5 -
 .../tests/test_vllm_mm_sidecar_protocol.py    | 111 +++++++
 .../tests/test_vllm_redis_media_processor.py  | 235 ++++++++++++++
 9 files changed, 1096 insertions(+), 7 deletions(-)
 create mode 100644 grpc_servicer/smg_grpc_servicer/mm_sidecar_protocol.py
 create mode 100644 grpc_servicer/smg_grpc_servicer/vllm/mm_sidecar.py
 create mode 100644 grpc_servicer/tests/test_vllm_mm_sidecar_protocol.py
 create mode 100644 grpc_servicer/tests/test_vllm_redis_media_processor.py

diff --git a/grpc_servicer/README.md b/grpc_servicer/README.md
index 9661f72cf3..95cb9497f3 100644
--- a/grpc_servicer/README.md
+++ b/grpc_servicer/README.md
@@ -70,6 +70,26 @@ e2e lanes run the multimodal suites in this mode, and
 `crates/multimodal/scripts/check_worker_anchor_parity.py` checks that a spec's
 anchor is the token vLLM expands.
 
+To move fetching and processing out of the vLLM process, run the GPU-free
+sidecar next to a private Redis and point the worker at it
+(`pip install smg-grpc-servicer[vllm,vllm-redis]`):
+
+```bash
+python -m smg_grpc_servicer.vllm.mm_sidecar --model Qwen/Qwen3-VL-8B-Instruct \
+    --redis-url redis://127.0.0.1:6379/0 --allowed-media-domains example.com
+SMG_VLLM_MM_PROCESSOR=redis SMG_VLLM_MM_REDIS_URL=redis://127.0.0.1:6379/0 \
+    vllm serve Qwen/Qwen3-VL-8B-Instruct --grpc
+```
+
+The sidecar and the worker must run the same model, vLLM version and dtype:
+the worker advertises `mm_processor=redis` only while a sidecar with a
+matching fingerprint keeps its `hello` key alive, and rejects results that
+disagree. Jobs and results travel over Redis lists under
+`smg:mm:v1:{namespace}`; results carry full tensors keyed by a per-attempt job
+id and expire after 120 s. Knobs: `SMG_VLLM_MM_SIDECAR_TIMEOUT_MS` (30000),
+`SMG_VLLM_MM_SIDECAR_MAX_QUEUE` (256, fail fast when the queue is deeper),
+`SMG_VLLM_MM_SIDECAR_NAMESPACE` (override the derived namespace).
+
 ### MLX
 
 ```bash
diff --git a/grpc_servicer/pyproject.toml b/grpc_servicer/pyproject.toml
index 97177ce255..98f48d9817 100644
--- a/grpc_servicer/pyproject.toml
+++ b/grpc_servicer/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "smg-grpc-servicer"
-version = "0.10.0"
+version = "0.11.0"
 description = "SMG gRPC servicer implementations for LLM inference engines (vLLM, MLX, TokenSpeed, SGLang)"
 requires-python = ">=3.10"
 dependencies = [
@@ -31,6 +31,8 @@ classifiers = [
 
 [project.optional-dependencies]
 vllm = ["vllm>=0.19.0", "pyzmq>=25.0.0", "msgspec>=0.18.0"]
+# Redis media-processing sidecar backend for vLLM (SMG_VLLM_MM_PROCESSOR=redis).
+vllm-redis = ["redis>=5.0", "msgspec>=0.18.0"]
 # 0.5.18 dropped the server_args parameter from
 # create_load_snapshot_reader(); the floor tracks that signature, which the
 # request manager calls unconditionally. (Earlier floors recorded the shmem
diff --git a/grpc_servicer/smg_grpc_servicer/mm_sidecar_protocol.py b/grpc_servicer/smg_grpc_servicer/mm_sidecar_protocol.py
new file mode 100644
index 0000000000..471d3d4b4a
--- /dev/null
+++ b/grpc_servicer/smg_grpc_servicer/mm_sidecar_protocol.py
@@ -0,0 +1,187 @@
+"""Servicer <-> sidecar protocol for worker-side media processing over Redis.
+
+Engine-free: no vLLM import. Redis is a job transport plus the sidecar's own
+cache; the servicer always receives full tensors, so nothing here has to stay
+consistent with vLLM's processor caches.
+
+Keys under ``smg:mm:v1:{namespace}``:
+- ``:hello``          HASH, refreshed by the sidecar (fingerprint + schemes)
+- ``:jobs``           LIST, servicer LPUSH / sidecar BRPOP
+- ``:res:{job_id}``   LIST, sidecar LPUSH + EXPIRE / servicer BRPOP
+"""
+
+from __future__ import annotations
+
+import hashlib
+from collections.abc import Mapping
+
+import msgspec
+
+SCHEMA_VERSION = 1
+KEY_ROOT = "smg:mm:v1"
+
+ENV_REDIS_URL = "SMG_VLLM_MM_REDIS_URL"
+ENV_TIMEOUT_MS = "SMG_VLLM_MM_SIDECAR_TIMEOUT_MS"
+ENV_MAX_QUEUE = "SMG_VLLM_MM_SIDECAR_MAX_QUEUE"
+ENV_NAMESPACE = "SMG_VLLM_MM_SIDECAR_NAMESPACE"
+
+DEFAULT_REDIS_URL = "redis://127.0.0.1:6379/0"
+DEFAULT_TIMEOUT_MS = 30_000
+DEFAULT_MAX_QUEUE = 256
+
+HELLO_TTL_S = 15
+HELLO_REFRESH_S = 5
+RESULT_TTL_S = 120
+
+# Client-caused failures become INVALID_ARGUMENT (terminal 400).
+CLIENT_ERROR_CODES = frozenset(
+    {
+        "fetch_failed",
+        "domain_not_allowed",
+        "scheme_not_accepted",
+        "limit_exceeded",
+        "decode_failed",
+        "placeholder_mismatch",
+    }
+)
+# Everything else is UNAVAILABLE (503) so the router retries elsewhere.
+RETRYABLE_ERROR_CODES = frozenset({"expired", "processor_error", "fingerprint_mismatch"})
+
+
+class Fingerprint(msgspec.Struct, frozen=True):
+    """What must match between a worker and the sidecar processing for it."""
+
+    model: str
+    vllm_version: str
+    dtype: str
+    video_backend: str
+    media_io_kwargs: str
+    mm_processor_kwargs: str
+
+    def namespace(self) -> str:
+        digest = hashlib.sha256(
+            f"{self.model}|{self.vllm_version}|{self.dtype}".encode()
+        ).hexdigest()
+        return digest[:16]
+
+    def mismatches(self, other: Fingerprint) -> list[str]:
+        return [
+            name for name in self.__struct_fields__ if getattr(self, name) != getattr(other, name)
+        ]
+
+    def to_hello(self) -> dict[str, str]:
+        return {name: getattr(self, name) for name in self.__struct_fields__}
+
+    @classmethod
+    def from_hello(cls, mapping: Mapping[bytes | str, bytes | str]) -> Fingerprint | None:
+        decoded = {_text(key): _text(value) for key, value in mapping.items()}
+        try:
+            return cls(**{name: decoded[name] for name in cls.__struct_fields__})
+        except KeyError:
+            return None
+
+
+def _text(value: bytes | str) -> str:
+    return value.decode("utf-8", errors="replace") if isinstance(value, bytes) else value
+
+
+def hello_schemes(mapping: Mapping[bytes | str, bytes | str]) -> str:
+    for key, value in mapping.items():
+        if _text(key) == "schemes":
+            return _text(value)
+    return ""
+
+
+def resolve_namespace(fingerprint: Fingerprint, override: str | None) -> str:
+    override = (override or "").strip()
+    return override if override else fingerprint.namespace()
+
+
+class Keys(msgspec.Struct, frozen=True):
+    prefix: str
+
+    @classmethod
+    def for_namespace(cls, namespace: str) -> Keys:
+        return cls(prefix=f"{KEY_ROOT}:{namespace}")
+
+    @property
+    def hello(self) -> str:
+        return f"{self.prefix}:hello"
+
+    @property
+    def jobs(self) -> str:
+        return f"{self.prefix}:jobs"
+
+    def result(self, job_id: str) -> str:
+        return f"{self.prefix}:res:{job_id}"
+
+
+class JobItem(msgspec.Struct):
+    modality: str
+    url: str
+
+
+class Job(msgspec.Struct):
+    v: int
+    job_id: str
+    request_id: str
+    fingerprint: Fingerprint
+    prompt_token_ids: list[int]
+    prompt: str | None
+    items: list[JobItem]
+    enqueued_ms: int
+    deadline_ms: int
+
+
+class Placeholder(msgspec.Struct):
+    offset: int
+    length: int
+    is_embed: list[bool] | None = None
+
+
+class Timing(msgspec.Struct):
+    queue_ms: int = 0
+    fetch_ms: int = 0
+    process_ms: int = 0
+
+
+class JobResult(msgspec.Struct):
+    v: int
+    job_id: str
+    ok: bool
+    fingerprint: Fingerprint | None = None
+    prompt_token_ids: list[int] = msgspec.field(default_factory=list)
+    mm_hashes: dict[str, list[str]] = msgspec.field(default_factory=dict)
+    mm_placeholders: dict[str, list[Placeholder]] = msgspec.field(default_factory=dict)
+    # Per item: one vLLM MsgpackEncoder buffer of a MultiModalKwargsItem.
+    mm_kwargs: dict[str, list[bytes]] = msgspec.field(default_factory=dict)
+    code: str = ""
+    message: str = ""
+    timing: Timing | None = None
+    cache: str = ""
+
+
+def failure(job_id: str, code: str, message: str) -> JobResult:
+    return JobResult(v=SCHEMA_VERSION, job_id=job_id, ok=False, code=code, message=message)
+
+
+_job_encoder = msgspec.msgpack.Encoder()
+_job_decoder = msgspec.msgpack.Decoder(type=Job)
+_result_encoder = msgspec.msgpack.Encoder()
+_result_decoder = msgspec.msgpack.Decoder(type=JobResult)
+
+
+def encode_job(job: Job) -> bytes:
+    return _job_encoder.encode(job)
+
+
+def decode_job(raw: bytes) -> Job:
+    return _job_decoder.decode(raw)
+
+
+def encode_result(result: JobResult) -> bytes:
+    return _result_encoder.encode(result)
+
+
+def decode_result(raw: bytes) -> JobResult:
+    return _result_decoder.decode(raw)
diff --git a/grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py b/grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py
index f14efc1b12..8890ca0a64 100644
--- a/grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py
+++ b/grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py
@@ -10,11 +10,35 @@
 
 import asyncio
 import inspect
+import json
 import logging
 import os
+import time
+import uuid
 from collections.abc import Awaitable, Mapping, Sequence
 from typing import Any
 
+from smg_grpc_servicer.mm_sidecar_protocol import (
+    CLIENT_ERROR_CODES,
+    DEFAULT_MAX_QUEUE,
+    DEFAULT_REDIS_URL,
+    DEFAULT_TIMEOUT_MS,
+    ENV_MAX_QUEUE,
+    ENV_NAMESPACE,
+    ENV_REDIS_URL,
+    ENV_TIMEOUT_MS,
+    SCHEMA_VERSION,
+    Fingerprint,
+    Job,
+    JobItem,
+    JobResult,
+    Keys,
+    decode_result,
+    encode_job,
+    hello_schemes,
+    resolve_namespace,
+)
+
 logger = logging.getLogger(__name__)
 
 ENV_PROCESSOR = "SMG_VLLM_MM_PROCESSOR"
@@ -173,6 +197,8 @@ async def process(
         prompt_text: str | None,
         items: Sequence[Any],
         arrival_time: float,
+        *,
+        request_id: str = "",
     ):
         enforce_item_count(items, self._max_items)
         enforce_item_bytes(items, self._max_item_bytes)
@@ -215,6 +241,221 @@ async def _fetch(self, index: int, item):
         raise ValueError(f"unsupported media modality {item.modality!r}")
 
 
+def engine_fingerprint(engine) -> Fingerprint:
+    """What the sidecar must match to process media for this engine."""
+    import vllm
+    from vllm import envs
+
+    model_config = engine.model_config
+    mm_config = model_config.get_multimodal_config()
+    return Fingerprint(
+        model=model_config.model,
+        vllm_version=vllm.__version__,
+        dtype=str(model_config.dtype),
+        video_backend=envs.VLLM_VIDEO_LOADER_BACKEND,
+        media_io_kwargs=json.dumps(mm_config.media_io_kwargs or {}, sort_keys=True, default=str),
+        mm_processor_kwargs=json.dumps(
+            mm_config.mm_processor_kwargs or {}, sort_keys=True, default=str
+        ),
+    )
+
+
+def _cast_floats(data, dtype):
+    """Cast floating tensors to the model dtype, as the HF processor path does."""
+    import torch
+
+    if torch.is_tensor(data):
+        return data.to(dtype=dtype) if data.is_floating_point() else data
+    if isinstance(data, list):
+        return [_cast_floats(part, dtype) for part in data]
+    return data
+
+
+def _parse_schemes(value: str) -> set[str]:
+    return {scheme.strip().lower() for scheme in value.split(",") if scheme.strip()}
+
+
+class RedisMediaProcessor:
+    """Hand jobs to a media-processing sidecar over Redis lists.
+
+    The sidecar fetches and runs vLLM's processor; the result carries full
+    tensors, rebuilt here into the same pre-rendered engine input the
+    preprocessed path uses.
+    """
+
+    name = MODE_REDIS
+
+    def __init__(
+        self,
+        engine,
+        fingerprint: Fingerprint,
+        *,
+        redis_url: str = DEFAULT_REDIS_URL,
+        timeout_ms: int = DEFAULT_TIMEOUT_MS,
+        max_queue: int = DEFAULT_MAX_QUEUE,
+        namespace: str | None = None,
+        max_item_bytes: int = DEFAULT_MAX_ITEM_BYTES,
+        max_inflight: int = DEFAULT_MAX_INFLIGHT,
+        client=None,
+    ) -> None:
+        self._engine = engine
+        self._fingerprint = fingerprint
+        self._timeout_ms = timeout_ms
+        self._max_queue = max_queue
+        self._max_item_bytes = max_item_bytes
+        self.max_inflight = max_inflight
+        self._keys = Keys.for_namespace(resolve_namespace(fingerprint, namespace))
+        self._client = client if client is not None else _redis_client(redis_url)
+        self._probe_logged = False
+        # Until the sidecar says otherwise, assume the default fetch schemes.
+        self.schemes = "http,https,data"
+        self.accepted_schemes = _parse_schemes(self.schemes)
+        logger.info("Redis media sidecar keys under %s", self._keys.prefix)
+
+    async def probe(self) -> bool:
+        """Whether a sidecar with a matching fingerprint is alive."""
+        try:
+            hello = await self._client.hgetall(self._keys.hello)
+        except Exception as e:  # noqa: BLE001 - any transport failure means "not advertised"
+            self._log_probe_once("redis unreachable: %s", e)
+            return False
+        remote = Fingerprint.from_hello(hello) if hello else None
+        if remote is None:
+            self._log_probe_once("no sidecar hello at %s", self._keys.hello)
+            return False
+        mismatches = self._fingerprint.mismatches(remote)
+        if mismatches:
+            self._log_probe_once(
+                "sidecar fingerprint mismatch on %s (worker=%s sidecar=%s)",
+                ",".join(mismatches),
+                self._fingerprint.to_hello(),
+                remote.to_hello(),
+            )
+            return False
+        schemes = hello_schemes(hello)
+        if schemes:
+            self.schemes = schemes
+            self.accepted_schemes = _parse_schemes(schemes)
+        self._probe_logged = False
+        return True
+
+    def _log_probe_once(self, message: str, *args) -> None:
+        if not self._probe_logged:
+            logger.error("Media sidecar not advertised: " + message, *args)
+            self._probe_logged = True
+
+    async def process(
+        self,
+        prompt_token_ids: list[int],
+        prompt_text: str | None,
+        items: Sequence[Any],
+        arrival_time: float,
+        *,
+        request_id: str = "",
+    ):
+        enforce_item_bytes(items, self._max_item_bytes)
+        now_ms = int(time.time() * 1000)
+        job = Job(
+            v=SCHEMA_VERSION,
+            job_id=uuid.uuid4().hex,
+            request_id=request_id,
+            fingerprint=self._fingerprint,
+            prompt_token_ids=list(prompt_token_ids),
+            prompt=prompt_text,
+            items=[JobItem(modality=item.modality, url=item.url) for item in items],
+            enqueued_ms=now_ms,
+            deadline_ms=now_ms + self._timeout_ms,
+        )
+        result = await self._submit_and_wait(job)
+        return self._rebuild(result, prompt_text, arrival_time)
+
+    async def _submit_and_wait(self, job: Job) -> JobResult:
+        """Transport only: queue the job and wait for its result."""
+        try:
+            depth = await self._client.llen(self._keys.jobs)
+            if depth >= self._max_queue:
+                raise MmProcessorUnavailable(
+                    f"sidecar_overloaded: {depth} jobs queued (cap {self._max_queue})"
+                )
+            await self._client.lpush(self._keys.jobs, encode_job(job))
+            popped = await self._client.brpop(
+                self._keys.result(job.job_id), timeout=self._timeout_ms / 1000
+            )
+        except MmProcessorUnavailable:
+            raise
+        except Exception as e:  # noqa: BLE001 - redis transport failures are retryable
+            raise MmProcessorUnavailable(f"sidecar_unavailable: {e}") from e
+        if popped is None:
+            raise MmProcessorUnavailable(
+                f"sidecar_timeout: no result for job {job.job_id} within {self._timeout_ms} ms"
+            )
+        _, raw = popped
+        result = decode_result(raw)
+        if result.job_id != job.job_id:
+            raise MmProcessorUnavailable(
+                f"sidecar_protocol: result for job {result.job_id} on key of {job.job_id}"
+            )
+        if not result.ok:
+            if result.code in CLIENT_ERROR_CODES:
+                raise ValueError(f"media processing failed ({result.code}): {result.message}")
+            raise MmProcessorUnavailable(f"{result.code or 'processor_error'}: {result.message}")
+        if result.fingerprint is not None:
+            mismatches = self._fingerprint.mismatches(result.fingerprint)
+            if mismatches:
+                raise MmProcessorUnavailable(f"fingerprint_mismatch: {','.join(mismatches)}")
+        return result
+
+    def _rebuild(self, result: JobResult, prompt_text: str | None, arrival_time: float):
+        """Turn a sidecar result into the pre-rendered engine input."""
+        import torch
+        from vllm.inputs.engine import mm_input
+        from vllm.multimodal.inputs import (
+            MultiModalKwargsItem,
+            MultiModalKwargsItems,
+            PlaceholderRange,
+        )
+        from vllm.v1.serial_utils import MsgpackDecoder
+
+        decoder = MsgpackDecoder(t=MultiModalKwargsItem)
+        dtype = self._engine.model_config.dtype
+        mm_kwargs: dict[str, list[MultiModalKwargsItem]] = {}
+        for modality, blobs in result.mm_kwargs.items():
+            items = []
+            for blob in blobs:
+                item = decoder.decode(blob)
+                for elem in item.values():
+                    elem.data = _cast_floats(elem.data, dtype)
+                items.append(item)
+            mm_kwargs[modality] = items
+
+        mm_placeholders = {
+            modality: [
+                PlaceholderRange(
+                    offset=p.offset,
+                    length=p.length,
+                    is_embed=torch.tensor(p.is_embed, dtype=torch.bool) if p.is_embed else None,
+                )
+                for p in ranges
+            ]
+            for modality, ranges in result.mm_placeholders.items()
+        }
+        prompt = mm_input(
+            prompt_token_ids=list(result.prompt_token_ids),
+            mm_kwargs=MultiModalKwargsItems(mm_kwargs),
+            mm_hashes={modality: list(hashes) for modality, hashes in result.mm_hashes.items()},
+            mm_placeholders=mm_placeholders,
+            prompt=prompt_text,
+        )
+        prompt["arrival_time"] = arrival_time
+        return prompt
+
+
+def _redis_client(redis_url: str):
+    import redis.asyncio as redis_asyncio
+
+    return redis_asyncio.from_url(redis_url, decode_responses=False, socket_connect_timeout=1.0)
+
+
 def build_mm_processor(engine, *, env: Mapping[str, str] = os.environ):
     """Construct the configured backend, or None when worker-side processing is off."""
     mode = resolve_mm_processor_mode(env)
@@ -231,4 +472,13 @@ def build_mm_processor(engine, *, env: Mapping[str, str] = os.environ):
         return InProcessMediaProcessor(
             engine, max_item_bytes=max_item_bytes, max_inflight=max_inflight, max_items=max_items
         )
-    raise ValueError(f"{ENV_PROCESSOR}={mode} is not available in this servicer build")
+    return RedisMediaProcessor(
+        engine,
+        engine_fingerprint(engine),
+        redis_url=env.get(ENV_REDIS_URL) or DEFAULT_REDIS_URL,
+        timeout_ms=env_int(env, ENV_TIMEOUT_MS, DEFAULT_TIMEOUT_MS),
+        max_queue=env_int(env, ENV_MAX_QUEUE, DEFAULT_MAX_QUEUE),
+        namespace=env.get(ENV_NAMESPACE),
+        max_item_bytes=max_item_bytes,
+        max_inflight=max_inflight,
+    )
diff --git a/grpc_servicer/smg_grpc_servicer/vllm/mm_sidecar.py b/grpc_servicer/smg_grpc_servicer/vllm/mm_sidecar.py
new file mode 100644
index 0000000000..0166fef103
--- /dev/null
+++ b/grpc_servicer/smg_grpc_servicer/vllm/mm_sidecar.py
@@ -0,0 +1,288 @@
+"""Reference media-processing sidecar for vLLM gRPC workers (GPU-free).
+
+Pops jobs from Redis, fetches the media with vLLM's MediaConnector, runs vLLM's
+own multimodal processor over the unexpanded prompt, and pushes full-tensor
+results back. Built the way ``vllm launch render`` builds its renderer, so the
+tensors match what the worker would have produced itself.
+
+Usage:
+    python -m smg_grpc_servicer.vllm.mm_sidecar --model <model> \\
+        --redis-url redis://127.0.0.1:6379/0 [--namespace NS] [--concurrency N] \\
+        [AsyncEngineArgs flags: --dtype --allowed-media-domains --media-io-kwargs ...]
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import logging
+import time
+
+from smg_grpc_servicer.mm_sidecar_protocol import (
+    DEFAULT_REDIS_URL,
+    HELLO_REFRESH_S,
+    HELLO_TTL_S,
+    RESULT_TTL_S,
+    SCHEMA_VERSION,
+    Fingerprint,
+    Job,
+    JobResult,
+    Keys,
+    Placeholder,
+    Timing,
+    decode_job,
+    encode_result,
+    failure,
+    resolve_namespace,
+)
+
+logger = logging.getLogger("smg_grpc_servicer.vllm.mm_sidecar")
+
+
+def build_config(args: argparse.Namespace):
+    """vLLM config for preprocessing only: no quantized kernels, no KV cache."""
+    from vllm import envs
+    from vllm.config import VllmConfig
+    from vllm.engine.arg_utils import AsyncEngineArgs
+
+    engine_args = AsyncEngineArgs.from_cli_args(args)
+    model_config = engine_args.create_model_config()
+    model_config.quantization = None
+    envs.VLLM_CPU_KVCACHE_SPACE = 0
+    return VllmConfig(model_config=model_config)
+
+
+def config_fingerprint(vllm_config) -> Fingerprint:
+    import vllm
+    from vllm import envs
+
+    model_config = vllm_config.model_config
+    mm_config = model_config.get_multimodal_config()
+    return Fingerprint(
+        model=model_config.model,
+        vllm_version=vllm.__version__,
+        dtype=str(model_config.dtype),
+        video_backend=envs.VLLM_VIDEO_LOADER_BACKEND,
+        media_io_kwargs=json.dumps(mm_config.media_io_kwargs or {}, sort_keys=True, default=str),
+        mm_processor_kwargs=json.dumps(
+            mm_config.mm_processor_kwargs or {}, sort_keys=True, default=str
+        ),
+    )
+
+
+def _url_scheme(url: str) -> str:
+    scheme, sep, _ = url.partition(":")
+    return scheme.lower() if sep else ""
+
+
+class Sidecar:
+    def __init__(self, vllm_config, renderer, client, *, namespace: str | None, concurrency: int):
+        from vllm import envs
+        from vllm.multimodal.media.connector import MEDIA_CONNECTOR_REGISTRY
+        from vllm.transformers_utils.processor import get_video_processor_cls_name
+
+        from smg_grpc_servicer.vllm.media_refs import advertised_schemes, parse_scheme_list
+
+        model_config = vllm_config.model_config
+        mm_config = model_config.get_multimodal_config()
+        self._renderer = renderer
+        self._client = client
+        self._concurrency = max(1, concurrency)
+        self._fingerprint = config_fingerprint(vllm_config)
+        self._keys = Keys.for_namespace(resolve_namespace(self._fingerprint, namespace))
+        self._connector = MEDIA_CONNECTOR_REGISTRY.load(
+            envs.VLLM_MEDIA_CONNECTOR,
+            media_io_kwargs=mm_config.media_io_kwargs,
+            allowed_local_media_path=model_config.allowed_local_media_path,
+            allowed_media_domains=model_config.allowed_media_domains,
+        )
+        self._video_processor = get_video_processor_cls_name(model_config)
+        self._schemes = advertised_schemes(model_config.allowed_local_media_path)
+        self._accepted = parse_scheme_list(self._schemes)
+        self._started_at = time.time()
+        if not model_config.allowed_media_domains:
+            logger.warning(
+                "no --allowed-media-domains: this sidecar will fetch media from any host"
+            )
+
+    async def run(self) -> None:
+        logger.info(
+            "media sidecar serving %s under %s (concurrency=%d)",
+            self._fingerprint.model,
+            self._keys.prefix,
+            self._concurrency,
+        )
+        tasks = [asyncio.create_task(self._heartbeat())]
+        tasks += [asyncio.create_task(self._worker(i)) for i in range(self._concurrency)]
+        try:
+            await asyncio.gather(*tasks)
+        finally:
+            for task in tasks:
+                task.cancel()
+
+    async def _heartbeat(self) -> None:
+        mapping = {
+            **self._fingerprint.to_hello(),
+            "schema": str(SCHEMA_VERSION),
+            "schemes": self._schemes,
+            "started_at": str(int(self._started_at)),
+        }
+        while True:
+            try:
+                await self._client.hset(self._keys.hello, mapping=mapping)
+                await self._client.expire(self._keys.hello, HELLO_TTL_S)
+            except Exception as e:  # noqa: BLE001 - keep advertising through redis blips
+                logger.warning("hello refresh failed: %s", e)
+            await asyncio.sleep(HELLO_REFRESH_S)
+
+    async def _worker(self, index: int) -> None:
+        while True:
+            try:
+                popped = await self._client.brpop(self._keys.jobs, timeout=5)
+            except Exception as e:  # noqa: BLE001 - reconnect on the next iteration
+                logger.warning("worker %d: brpop failed: %s", index, e)
+                await asyncio.sleep(1)
+                continue
+            if popped is None:
+                continue
+            _, raw = popped
+            try:
+                job = decode_job(raw)
+            except Exception as e:  # noqa: BLE001 - a malformed job has no result key to answer on
+                logger.error("worker %d: undecodable job dropped: %s", index, e)
+                continue
+            result = await self.handle(job)
+            try:
+                key = self._keys.result(job.job_id)
+                await self._client.lpush(key, encode_result(result))
+                await self._client.expire(key, RESULT_TTL_S)
+            except Exception as e:  # noqa: BLE001 - the servicer times out and retries
+                logger.warning("worker %d: result push failed for %s: %s", index, job.job_id, e)
+
+    async def handle(self, job: Job) -> JobResult:
+        started = time.time()
+        now_ms = int(started * 1000)
+        if now_ms > job.deadline_ms:
+            return failure(job.job_id, "expired", "job deadline passed before processing")
+        mismatches = self._fingerprint.mismatches(job.fingerprint)
+        if mismatches:
+            return failure(job.job_id, "fingerprint_mismatch", ",".join(mismatches))
+
+        multi_modal_data: dict[str, list] = {}
+        for index, item in enumerate(job.items):
+            scheme = _url_scheme(item.url)
+            if scheme not in self._accepted:
+                return failure(
+                    job.job_id,
+                    "scheme_not_accepted",
+                    f"item {index}: scheme '{scheme or 'none'}' not in {self._schemes}",
+                )
+            try:
+                media = await self._fetch(item.modality, item.url)
+            except ValueError as e:
+                code = "domain_not_allowed" if "domain" in str(e).lower() else "fetch_failed"
+                return failure(job.job_id, code, f"item {index}: {e}")
+            except Exception as e:  # noqa: BLE001 - network/decoder errors are client-visible
+                return failure(job.job_id, "fetch_failed", f"item {index}: {e}")
+            multi_modal_data.setdefault(item.modality, []).append(media)
+        fetched = time.time()
+
+        from vllm import TokensPrompt
+        from vllm.v1.serial_utils import MsgpackEncoder
+
+        prompt = TokensPrompt(
+            prompt_token_ids=list(job.prompt_token_ids), multi_modal_data=multi_modal_data
+        )
+        if job.prompt:
+            prompt["prompt"] = job.prompt
+        try:
+            engine_input = await self._renderer.process_for_engine_async(
+                prompt, arrival_time=0.0, skip_mm_cache=True
+            )
+        except RuntimeError as e:
+            return failure(job.job_id, "placeholder_mismatch", str(e))
+        except ValueError as e:
+            code = "limit_exceeded" if "at most" in str(e).lower() else "decode_failed"
+            return failure(job.job_id, code, str(e))
+        except Exception as e:  # noqa: BLE001 - processor bugs are retryable elsewhere
+            logger.exception("processing failed for job %s", job.job_id)
+            return failure(job.job_id, "processor_error", str(e))
+        processed = time.time()
+
+        encoder = MsgpackEncoder(size_threshold=2**62)
+        mm_kwargs: dict[str, list[bytes]] = {}
+        for modality, items in engine_input["mm_kwargs"].items():
+            blobs = []
+            for item in items:
+                bufs = encoder.encode(item)
+                if len(bufs) != 1:
+                    return failure(job.job_id, "processor_error", "tensor not inlined")
+                blobs.append(bytes(bufs[0]))
+            mm_kwargs[modality] = blobs
+        mm_placeholders = {
+            modality: [
+                Placeholder(
+                    offset=r.offset,
+                    length=r.length,
+                    is_embed=r.is_embed.tolist() if r.is_embed is not None else None,
+                )
+                for r in ranges
+            ]
+            for modality, ranges in engine_input["mm_placeholders"].items()
+        }
+        return JobResult(
+            v=SCHEMA_VERSION,
+            job_id=job.job_id,
+            ok=True,
+            fingerprint=self._fingerprint,
+            prompt_token_ids=list(engine_input["prompt_token_ids"]),
+            mm_hashes={m: list(h) for m, h in engine_input["mm_hashes"].items()},
+            mm_placeholders=mm_placeholders,
+            mm_kwargs=mm_kwargs,
+            timing=Timing(
+                queue_ms=max(0, now_ms - job.enqueued_ms),
+                fetch_ms=int((fetched - started) * 1000),
+                process_ms=int((processed - fetched) * 1000),
+            ),
+        )
+
+    async def _fetch(self, modality: str, url: str):
+        if modality == "image":
+            return await self._connector.fetch_image_async(url)
+        if modality == "video":
+            return await self._connector.fetch_video_async(
+                url, video_processor=self._video_processor
+            )
+        raise ValueError(f"unsupported media modality {modality!r}")
+
+
+async def serve(args: argparse.Namespace) -> None:
+    import redis.asyncio as redis_asyncio
+    from vllm.renderers.registry import renderer_from_config
+
+    vllm_config = build_config(args)
+    renderer = renderer_from_config(vllm_config)
+    client = redis_asyncio.from_url(args.redis_url, decode_responses=False)
+    sidecar = Sidecar(
+        vllm_config, renderer, client, namespace=args.namespace, concurrency=args.concurrency
+    )
+    await sidecar.run()
+
+
+def main() -> None:
+    from vllm.engine.arg_utils import AsyncEngineArgs
+    from vllm.utils.argparse_utils import FlexibleArgumentParser
+
+    logging.basicConfig(level=logging.INFO)
+    parser = FlexibleArgumentParser(description="smg media-processing sidecar for vLLM")
+    parser.add_argument("--redis-url", default=DEFAULT_REDIS_URL)
+    parser.add_argument("--namespace", default=None, help="override the derived key namespace")
+    parser.add_argument("--concurrency", type=int, default=2)
+    parser = AsyncEngineArgs.add_cli_args(parser)
+    args = parser.parse_args()
+    asyncio.run(serve(args))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/grpc_servicer/smg_grpc_servicer/vllm/servicer.py b/grpc_servicer/smg_grpc_servicer/vllm/servicer.py
index 4fac4659f8..a9adfec4ea 100755
--- a/grpc_servicer/smg_grpc_servicer/vllm/servicer.py
+++ b/grpc_servicer/smg_grpc_servicer/vllm/servicer.py
@@ -249,6 +249,7 @@ async def Generate(
                         request.tokenized.original_text or None,
                         items,
                         arrival_time,
+                        request_id=request_id,
                     )
             elif has_preprocessed_mm and input_type == "tokenized":
                 # A pixel-less payload (PD decode leg) is only decodable with
diff --git a/grpc_servicer/tests/test_vllm_mm_processor.py b/grpc_servicer/tests/test_vllm_mm_processor.py
index b9304fb381..869407b245 100644
--- a/grpc_servicer/tests/test_vllm_mm_processor.py
+++ b/grpc_servicer/tests/test_vllm_mm_processor.py
@@ -105,11 +105,6 @@ def test_non_multimodal_model_returns_none(self):
         env = {"SMG_VLLM_MM_PROCESSOR": "inprocess"}
         assert mm_processor.build_mm_processor(self._Engine(multimodal=False), env=env) is None
 
-    def test_redis_not_available_yet(self):
-        env = {"SMG_VLLM_MM_PROCESSOR": "redis"}
-        with pytest.raises(ValueError, match="not available"):
-            mm_processor.build_mm_processor(self._Engine(), env=env)
-
     def test_invalid_item_cap_is_rejected_before_construction(self):
         env = {"SMG_VLLM_MM_PROCESSOR": "inprocess", "SMG_VLLM_MM_MAX_ITEM_BYTES": "0"}
         with pytest.raises(ValueError, match="SMG_VLLM_MM_MAX_ITEM_BYTES"):
diff --git a/grpc_servicer/tests/test_vllm_mm_sidecar_protocol.py b/grpc_servicer/tests/test_vllm_mm_sidecar_protocol.py
new file mode 100644
index 0000000000..19d4a449a0
--- /dev/null
+++ b/grpc_servicer/tests/test_vllm_mm_sidecar_protocol.py
@@ -0,0 +1,111 @@
+"""Unit tests for the Redis media-sidecar protocol (engine-free, needs msgspec).
+
+Run with: pytest grpc_servicer/tests/test_vllm_mm_sidecar_protocol.py
+"""
+
+import pytest
+
+pytest.importorskip("msgspec")
+from smg_grpc_servicer import mm_sidecar_protocol as proto  # noqa: E402
+
+
+def fingerprint(**overrides) -> proto.Fingerprint:
+    values = {
+        "model": "Qwen/Qwen3-VL-8B-Instruct",
+        "vllm_version": "0.27.1",
+        "dtype": "torch.bfloat16",
+        "video_backend": "opencv",
+        "media_io_kwargs": "{}",
+        "mm_processor_kwargs": "{}",
+    }
+    values.update(overrides)
+    return proto.Fingerprint(**values)
+
+
+class TestFingerprint:
+    def test_namespace_is_deterministic_and_dtype_sensitive(self):
+        a = fingerprint().namespace()
+        assert a == fingerprint().namespace()
+        assert len(a) == 16
+        assert a != fingerprint(dtype="torch.float16").namespace()
+        # Kwargs are not part of the namespace; they are checked by fingerprint.
+        assert a == fingerprint(media_io_kwargs='{"video":{"num_frames":8}}').namespace()
+
+    def test_override_wins(self):
+        assert proto.resolve_namespace(fingerprint(), " prod-a ") == "prod-a"
+        assert proto.resolve_namespace(fingerprint(), "") == fingerprint().namespace()
+
+    def test_mismatches_name_fields(self):
+        assert fingerprint().mismatches(fingerprint()) == []
+        assert fingerprint().mismatches(fingerprint(vllm_version="0.28.0", dtype="x")) == [
+            "vllm_version",
+            "dtype",
+        ]
+
+    def test_hello_roundtrip_with_bytes_mapping(self):
+        hello = {k.encode(): v.encode() for k, v in fingerprint().to_hello().items()}
+        hello[b"schemes"] = b"http,https,data"
+        hello[b"schema"] = b"1"
+        assert proto.Fingerprint.from_hello(hello) == fingerprint()
+        assert proto.hello_schemes(hello) == "http,https,data"
+
+    def test_hello_missing_field_is_none(self):
+        hello = {b"model": b"m"}
+        assert proto.Fingerprint.from_hello(hello) is None
+        assert proto.hello_schemes(hello) == ""
+
+
+class TestKeys:
+    def test_key_layout(self):
+        keys = proto.Keys.for_namespace("abc")
+        assert keys.hello == "smg:mm:v1:abc:hello"
+        assert keys.jobs == "smg:mm:v1:abc:jobs"
+        assert keys.result("j1") == "smg:mm:v1:abc:res:j1"
+
+
+class TestMessages:
+    def test_job_roundtrip(self):
+        job = proto.Job(
+            v=proto.SCHEMA_VERSION,
+            job_id="j1",
+            request_id="r1",
+            fingerprint=fingerprint(),
+            prompt_token_ids=[1, 2, 3],
+            prompt="hi <|image_pad|>",
+            items=[proto.JobItem(modality="image", url="https://a/1.png")],
+            enqueued_ms=100,
+            deadline_ms=30100,
+        )
+        assert proto.decode_job(proto.encode_job(job)) == job
+
+    def test_result_roundtrip_keeps_blobs_and_masks(self):
+        result = proto.JobResult(
+            v=proto.SCHEMA_VERSION,
+            job_id="j1",
+            ok=True,
+            fingerprint=fingerprint(),
+            prompt_token_ids=[1, 5, 5, 5, 3],
+            mm_hashes={"image": ["abc"]},
+            mm_placeholders={
+                "image": [proto.Placeholder(offset=1, length=3, is_embed=[True, False, True])]
+            },
+            mm_kwargs={"image": [b"\x93\x01\x02\x03"]},
+            timing=proto.Timing(queue_ms=1, fetch_ms=2, process_ms=3),
+            cache="miss",
+        )
+        parsed = proto.decode_result(proto.encode_result(result))
+        assert parsed == result
+        assert parsed.mm_kwargs["image"][0] == b"\x93\x01\x02\x03"
+        assert parsed.mm_placeholders["image"][0].is_embed == [True, False, True]
+
+    def test_failure_defaults(self):
+        result = proto.failure("j2", "expired", "too late")
+        assert not result.ok
+        assert result.code == "expired"
+        assert result.mm_kwargs == {}
+        assert proto.decode_result(proto.encode_result(result)) == result
+
+    def test_error_code_sets_are_disjoint(self):
+        assert not (proto.CLIENT_ERROR_CODES & proto.RETRYABLE_ERROR_CODES)
+        assert "placeholder_mismatch" in proto.CLIENT_ERROR_CODES
+        assert "fingerprint_mismatch" in proto.RETRYABLE_ERROR_CODES
diff --git a/grpc_servicer/tests/test_vllm_redis_media_processor.py b/grpc_servicer/tests/test_vllm_redis_media_processor.py
new file mode 100644
index 0000000000..21463ab162
--- /dev/null
+++ b/grpc_servicer/tests/test_vllm_redis_media_processor.py
@@ -0,0 +1,235 @@
+"""Unit tests for RedisMediaProcessor transport semantics (engine-free, fake Redis).
+
+Run with: pytest grpc_servicer/tests/test_vllm_redis_media_processor.py
+"""
+
+import asyncio
+import importlib.util
+import sys
+from dataclasses import dataclass
+from pathlib import Path
+
+import pytest
+
+pytest.importorskip("msgspec")
+from smg_grpc_servicer import mm_sidecar_protocol as proto  # noqa: E402
+
+# Import the module directly to avoid pulling vllm via the package __init__
+_MODULE_PATH = Path(__file__).parents[1] / "smg_grpc_servicer" / "vllm" / "mm_processor.py"
+_spec = importlib.util.spec_from_file_location("mm_processor", _MODULE_PATH)
+mm_processor = importlib.util.module_from_spec(_spec)
+sys.modules[_spec.name] = mm_processor
+_spec.loader.exec_module(mm_processor)
+
+
+@dataclass
+class _Item:
+    modality: str
+    url: str
+
+
+class _ModelConfig:
+    dtype = "bf16"
+    is_multimodal_model = True
+
+
+class _Engine:
+    model_config = _ModelConfig()
+
+
+class FakeRedis:
+    """Just enough of redis.asyncio for the servicer side of the protocol."""
+
+    def __init__(self, *, hello=None, queue_depth=0, responder=None, fail=None):
+        self.hello = hello or {}
+        self.queue_depth = queue_depth
+        self.responder = responder
+        self.fail = fail
+        self.pushed = []
+        self.expired = {}
+
+    async def hgetall(self, key):
+        if self.fail:
+            raise self.fail
+        return dict(self.hello)
+
+    async def llen(self, key):
+        if self.fail:
+            raise self.fail
+        return self.queue_depth
+
+    async def lpush(self, key, value):
+        self.pushed.append((key, value))
+
+    async def brpop(self, key, timeout=0):
+        if self.responder is None:
+            return None
+        job = proto.decode_job(self.pushed[-1][1])
+        result = self.responder(job)
+        return (key.encode(), proto.encode_result(result))
+
+    async def expire(self, key, seconds):
+        self.expired[key] = seconds
+
+
+def fingerprint(**overrides):
+    values = {
+        "model": "m",
+        "vllm_version": "0.27.1",
+        "dtype": "torch.bfloat16",
+        "video_backend": "opencv",
+        "media_io_kwargs": "{}",
+        "mm_processor_kwargs": "{}",
+    }
+    values.update(overrides)
+    return proto.Fingerprint(**values)
+
+
+def processor(client, **kwargs):
+    return mm_processor.RedisMediaProcessor(
+        _Engine(), fingerprint(), client=client, timeout_ms=50, max_queue=4, **kwargs
+    )
+
+
+def run(coro):
+    return asyncio.new_event_loop().run_until_complete(coro)
+
+
+def ok_result(job, **extra):
+    return proto.JobResult(
+        v=proto.SCHEMA_VERSION,
+        job_id=job.job_id,
+        ok=True,
+        fingerprint=extra.pop("fingerprint", fingerprint()),
+        prompt_token_ids=[1, 5, 5, 3],
+        mm_hashes={"image": ["h1"]},
+        mm_placeholders={"image": [proto.Placeholder(offset=1, length=2)]},
+        mm_kwargs={"image": [b"\x80"]},
+        **extra,
+    )
+
+
+class TestProbe:
+    def test_no_hello_is_not_advertised(self):
+        assert run(processor(FakeRedis()).probe()) is False
+
+    def test_redis_error_is_not_advertised(self):
+        assert run(processor(FakeRedis(fail=ConnectionError("down"))).probe()) is False
+
+    def test_fingerprint_mismatch_is_not_advertised(self):
+        hello = {k.encode(): v.encode() for k, v in fingerprint(dtype="x").to_hello().items()}
+        assert run(processor(FakeRedis(hello=hello)).probe()) is False
+
+    def test_matching_hello_advertises_and_adopts_schemes(self):
+        hello = {k.encode(): v.encode() for k, v in fingerprint().to_hello().items()}
+        hello[b"schemes"] = b"http,https,data,file"
+        p = processor(FakeRedis(hello=hello))
+        assert run(p.probe()) is True
+        assert p.schemes == "http,https,data,file"
+        assert "file" in p.accepted_schemes
+
+
+class TestSubmitAndWait:
+    def job(self, job_id="j1"):
+        return proto.Job(
+            v=proto.SCHEMA_VERSION,
+            job_id=job_id,
+            request_id="r1",
+            fingerprint=fingerprint(),
+            prompt_token_ids=[1, 2, 3],
+            prompt=None,
+            items=[proto.JobItem(modality="image", url="https://a/1.png")],
+            enqueued_ms=0,
+            deadline_ms=50,
+        )
+
+    def test_queue_over_cap_fails_fast(self):
+        client = FakeRedis(queue_depth=4)
+        with pytest.raises(mm_processor.MmProcessorUnavailable, match="sidecar_overloaded"):
+            run(processor(client)._submit_and_wait(self.job()))
+        assert client.pushed == []
+
+    def test_timeout_is_retryable(self):
+        client = FakeRedis()
+        with pytest.raises(mm_processor.MmProcessorUnavailable, match="sidecar_timeout"):
+            run(processor(client)._submit_and_wait(self.job()))
+        key, raw = client.pushed[0]
+        assert key.endswith(":jobs")
+        assert proto.decode_job(raw).job_id == "j1"
+
+    def test_transport_error_is_retryable(self):
+        client = FakeRedis(fail=ConnectionError("refused"))
+        with pytest.raises(mm_processor.MmProcessorUnavailable, match="sidecar_unavailable"):
+            run(processor(client)._submit_and_wait(self.job()))
+
+    def test_client_error_codes_become_value_errors(self):
+        client = FakeRedis(
+            responder=lambda job: proto.failure(job.job_id, "domain_not_allowed", "x")
+        )
+        with pytest.raises(ValueError, match="domain_not_allowed"):
+            run(processor(client)._submit_and_wait(self.job()))
+
+    def test_retryable_error_codes_stay_unavailable(self):
+        client = FakeRedis(responder=lambda job: proto.failure(job.job_id, "expired", "late"))
+        with pytest.raises(mm_processor.MmProcessorUnavailable, match="expired"):
+            run(processor(client)._submit_and_wait(self.job()))
+
+    def test_result_fingerprint_mismatch_is_unavailable(self):
+        client = FakeRedis(
+            responder=lambda job: ok_result(job, fingerprint=fingerprint(vllm_version="0.1"))
+        )
+        with pytest.raises(mm_processor.MmProcessorUnavailable, match="fingerprint_mismatch"):
+            run(processor(client)._submit_and_wait(self.job()))
+
+    def test_success_returns_full_result(self):
+        client = FakeRedis(responder=ok_result)
+        result = run(processor(client)._submit_and_wait(self.job()))
+        assert result.ok
+        assert result.prompt_token_ids == [1, 5, 5, 3]
+        assert result.mm_kwargs["image"] == [b"\x80"]
+
+
+class TestProcess:
+    def test_process_mints_a_fresh_job_per_attempt(self, monkeypatch):
+        seen = []
+
+        def responder(job):
+            seen.append(job)
+            return ok_result(job)
+
+        client = FakeRedis(responder=responder)
+        p = processor(client)
+        monkeypatch.setattr(p, "_rebuild", lambda result, text, arrival: ("built", result.job_id))
+        items = [_Item("image", "https://a/1.png")]
+        first = run(p.process([1, 2, 3], "text", items, 1.0, request_id="req"))
+        second = run(p.process([1, 2, 3], "text", items, 2.0, request_id="req"))
+        assert first[0] == "built" and second[0] == "built"
+        assert first[1] != second[1], "each attempt gets its own job id"
+        assert [job.request_id for job in seen] == ["req", "req"]
+        assert seen[0].items[0].url == "https://a/1.png"
+        assert seen[0].deadline_ms - seen[0].enqueued_ms == 50
+
+    def test_process_caps_inline_payloads_before_queueing(self):
+        client = FakeRedis(responder=ok_result)
+        p = processor(client, max_item_bytes=4)
+        items = [_Item("image", "data:image/png;base64,AAAAAAAAAAAA")]
+        with pytest.raises(ValueError, match="above the 4-byte cap"):
+            run(p.process([1], None, items, 0.0))
+        assert client.pushed == []
+
+
+class TestBuild:
+    def test_redis_mode_builds_processor(self, monkeypatch):
+        monkeypatch.setattr(mm_processor, "engine_fingerprint", lambda engine: fingerprint())
+        monkeypatch.setattr(mm_processor, "_redis_client", lambda url: FakeRedis())
+        env = {
+            "SMG_VLLM_MM_PROCESSOR": "redis",
+            "SMG_VLLM_MM_SIDECAR_TIMEOUT_MS": "1000",
+            "SMG_VLLM_MM_SIDECAR_MAX_QUEUE": "8",
+            "SMG_VLLM_MM_SIDECAR_NAMESPACE": "ns-1",
+        }
+        p = mm_processor.build_mm_processor(_Engine(), env=env)
+        assert p.name == "redis"
+        assert p._timeout_ms == 1000
+        assert p._max_queue == 8
+        assert p._keys.prefix == "smg:mm:v1:ns-1"

From ac799c490d83bfbf2dd2a2c98c9fc060ce62f965 Mon Sep 17 00:00:00 2001
From: Chang Su <8605658+CatherineSue@users.noreply.github.com>
Date: Mon, 14 Sep 2026 15:21:35 -0700
Subject: [PATCH 02/11] fix(vllm-grpc): harden the media sidecar loop and
 namespace the sidecar by fingerprint

Review fixes for the Redis media-processing backend:

- The sidecar worker loop guards handle(): a failing job is answered with
  processor_error and the loop keeps serving instead of taking the whole
  sidecar down through the fail-fast gather.
- Hello refresh and result push are single MULTI/EXEC transactions, so a
  crash between the write and its EXPIRE can no longer leave a TTL-less
  hello advertising a dead sidecar, or an unbounded result key.
- The key namespace hashes every fingerprint field, so fleets that differ
  in video backend or media/processor kwargs never share a hello key or a
  jobs list.
- Fetch and processing failures are classified by exception type in two
  pure helpers; the two remaining message substrings are pinned by tests.
- The sidecar reuses media_refs.url_scheme, and the Redis backend reuses
  BASE_SCHEMES/parse_scheme_list, so both ends classify URLs identically.
- A missing redis client fails with the install hint for the vllm-redis
  extra instead of a bare ModuleNotFoundError.
- First unit tests for the sidecar loop (guard, transactions, heartbeat).

Signed-off-by: Chang Su <8605658+CatherineSue@users.noreply.github.com>
---
 grpc_servicer/README.md                       |   5 +-
 .../smg_grpc_servicer/mm_sidecar_protocol.py  |   7 +-
 .../smg_grpc_servicer/vllm/mm_processor.py    |  25 ++-
 .../smg_grpc_servicer/vllm/mm_sidecar.py      |  78 ++++---
 grpc_servicer/tests/test_vllm_mm_sidecar.py   | 194 ++++++++++++++++++
 .../tests/test_vllm_mm_sidecar_protocol.py    |   8 +-
 .../tests/test_vllm_redis_media_processor.py  |   9 +
 7 files changed, 276 insertions(+), 50 deletions(-)
 create mode 100644 grpc_servicer/tests/test_vllm_mm_sidecar.py

diff --git a/grpc_servicer/README.md b/grpc_servicer/README.md
index 95cb9497f3..17410b8b0a 100644
--- a/grpc_servicer/README.md
+++ b/grpc_servicer/README.md
@@ -81,8 +81,9 @@ SMG_VLLM_MM_PROCESSOR=redis SMG_VLLM_MM_REDIS_URL=redis://127.0.0.1:6379/0 \
     vllm serve Qwen/Qwen3-VL-8B-Instruct --grpc
 ```
 
-The sidecar and the worker must run the same model, vLLM version and dtype:
-the worker advertises `mm_processor=redis` only while a sidecar with a
+The sidecar and the worker must agree on model, vLLM version, dtype, video
+backend and media/processor kwargs (the key namespace is derived from all of
+them): the worker advertises `mm_processor=redis` only while a sidecar with a
 matching fingerprint keeps its `hello` key alive, and rejects results that
 disagree. Jobs and results travel over Redis lists under
 `smg:mm:v1:{namespace}`; results carry full tensors keyed by a per-attempt job
diff --git a/grpc_servicer/smg_grpc_servicer/mm_sidecar_protocol.py b/grpc_servicer/smg_grpc_servicer/mm_sidecar_protocol.py
index 471d3d4b4a..a5822a78b5 100644
--- a/grpc_servicer/smg_grpc_servicer/mm_sidecar_protocol.py
+++ b/grpc_servicer/smg_grpc_servicer/mm_sidecar_protocol.py
@@ -59,10 +59,9 @@ class Fingerprint(msgspec.Struct, frozen=True):
     mm_processor_kwargs: str
 
     def namespace(self) -> str:
-        digest = hashlib.sha256(
-            f"{self.model}|{self.vllm_version}|{self.dtype}".encode()
-        ).hexdigest()
-        return digest[:16]
+        """Key namespace: fleets that would fail `mismatches` never share keys."""
+        joined = "|".join(getattr(self, name) for name in self.__struct_fields__)
+        return hashlib.sha256(joined.encode()).hexdigest()[:16]
 
     def mismatches(self, other: Fingerprint) -> list[str]:
         return [
diff --git a/grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py b/grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py
index 8890ca0a64..959bda4c8d 100644
--- a/grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py
+++ b/grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py
@@ -38,6 +38,11 @@
     hello_schemes,
     resolve_namespace,
 )
+from smg_grpc_servicer.vllm.media_refs import (
+    BASE_SCHEMES,
+    advertised_schemes,
+    parse_scheme_list,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -162,8 +167,6 @@ def __init__(
         from vllm.multimodal.media.connector import MEDIA_CONNECTOR_REGISTRY
         from vllm.transformers_utils.processor import get_video_processor_cls_name
 
-        from smg_grpc_servicer.vllm.media_refs import advertised_schemes, parse_scheme_list
-
         model_config = engine.model_config
         mm_config = model_config.get_multimodal_config()
         self._engine = engine
@@ -271,10 +274,6 @@ def _cast_floats(data, dtype):
     return data
 
 
-def _parse_schemes(value: str) -> set[str]:
-    return {scheme.strip().lower() for scheme in value.split(",") if scheme.strip()}
-
-
 class RedisMediaProcessor:
     """Hand jobs to a media-processing sidecar over Redis lists.
 
@@ -308,8 +307,8 @@ def __init__(
         self._client = client if client is not None else _redis_client(redis_url)
         self._probe_logged = False
         # Until the sidecar says otherwise, assume the default fetch schemes.
-        self.schemes = "http,https,data"
-        self.accepted_schemes = _parse_schemes(self.schemes)
+        self.schemes = ",".join(BASE_SCHEMES)
+        self.accepted_schemes = parse_scheme_list(self.schemes)
         logger.info("Redis media sidecar keys under %s", self._keys.prefix)
 
     async def probe(self) -> bool:
@@ -335,7 +334,7 @@ async def probe(self) -> bool:
         schemes = hello_schemes(hello)
         if schemes:
             self.schemes = schemes
-            self.accepted_schemes = _parse_schemes(schemes)
+            self.accepted_schemes = parse_scheme_list(schemes)
         self._probe_logged = False
         return True
 
@@ -451,7 +450,13 @@ def _rebuild(self, result: JobResult, prompt_text: str | None, arrival_time: flo
 
 
 def _redis_client(redis_url: str):
-    import redis.asyncio as redis_asyncio
+    try:
+        import redis.asyncio as redis_asyncio
+    except ImportError as e:
+        raise ValueError(
+            f"{ENV_PROCESSOR}=redis requires the redis client: "
+            "pip install smg-grpc-servicer[vllm,vllm-redis]"
+        ) from e
 
     return redis_asyncio.from_url(redis_url, decode_responses=False, socket_connect_timeout=1.0)
 
diff --git a/grpc_servicer/smg_grpc_servicer/vllm/mm_sidecar.py b/grpc_servicer/smg_grpc_servicer/vllm/mm_sidecar.py
index 0166fef103..aadf26f604 100644
--- a/grpc_servicer/smg_grpc_servicer/vllm/mm_sidecar.py
+++ b/grpc_servicer/smg_grpc_servicer/vllm/mm_sidecar.py
@@ -36,6 +36,7 @@
     failure,
     resolve_namespace,
 )
+from smg_grpc_servicer.vllm.media_refs import advertised_schemes, parse_scheme_list, url_scheme
 
 logger = logging.getLogger("smg_grpc_servicer.vllm.mm_sidecar")
 
@@ -71,22 +72,37 @@ def config_fingerprint(vllm_config) -> Fingerprint:
     )
 
 
-def _url_scheme(url: str) -> str:
-    scheme, sep, _ = url.partition(":")
-    return scheme.lower() if sep else ""
+# Codes come from exception types where vLLM offers one. The two substring
+# checks pin message wording from vLLM 0.27: connector.py's allowed-domain
+# rejection ("... domain ...") and processing/context.py's per-prompt limit
+# ("At most N ..."); test_vllm_mm_sidecar.py holds the wording.
+def classify_fetch_error(exc: BaseException) -> str:
+    if isinstance(exc, ValueError) and "domain" in str(exc).lower():
+        return "domain_not_allowed"
+    return "fetch_failed"
+
+
+def classify_process_error(exc: BaseException) -> str:
+    if isinstance(exc, RuntimeError):
+        return "placeholder_mismatch"
+    if isinstance(exc, ValueError):
+        return "limit_exceeded" if "at most" in str(exc).lower() else "decode_failed"
+    return "processor_error"
 
 
 class Sidecar:
     def __init__(self, vllm_config, renderer, client, *, namespace: str | None, concurrency: int):
-        from vllm import envs
+        from vllm import TokensPrompt, envs
         from vllm.multimodal.media.connector import MEDIA_CONNECTOR_REGISTRY
         from vllm.transformers_utils.processor import get_video_processor_cls_name
-
-        from smg_grpc_servicer.vllm.media_refs import advertised_schemes, parse_scheme_list
+        from vllm.v1.serial_utils import MsgpackEncoder
 
         model_config = vllm_config.model_config
         mm_config = model_config.get_multimodal_config()
         self._renderer = renderer
+        self._tokens_prompt = TokensPrompt
+        # All tensors inline: one buffer per kwargs item.
+        self._encoder = MsgpackEncoder(size_threshold=2**62)
         self._client = client
         self._concurrency = max(1, concurrency)
         self._fingerprint = config_fingerprint(vllm_config)
@@ -108,10 +124,10 @@ def __init__(self, vllm_config, renderer, client, *, namespace: str | None, conc
 
     async def run(self) -> None:
         logger.info(
-            "media sidecar serving %s under %s (concurrency=%d)",
-            self._fingerprint.model,
+            "media sidecar serving under %s (concurrency=%d) fingerprint=%s",
             self._keys.prefix,
             self._concurrency,
+            self._fingerprint.to_hello(),
         )
         tasks = [asyncio.create_task(self._heartbeat())]
         tasks += [asyncio.create_task(self._worker(i)) for i in range(self._concurrency)]
@@ -130,8 +146,11 @@ async def _heartbeat(self) -> None:
         }
         while True:
             try:
-                await self._client.hset(self._keys.hello, mapping=mapping)
-                await self._client.expire(self._keys.hello, HELLO_TTL_S)
+                # One transaction: a hello can never outlive its TTL.
+                pipe = self._client.pipeline(transaction=True)
+                pipe.hset(self._keys.hello, mapping=mapping)
+                pipe.expire(self._keys.hello, HELLO_TTL_S)
+                await pipe.execute()
             except Exception as e:  # noqa: BLE001 - keep advertising through redis blips
                 logger.warning("hello refresh failed: %s", e)
             await asyncio.sleep(HELLO_REFRESH_S)
@@ -152,11 +171,17 @@ async def _worker(self, index: int) -> None:
             except Exception as e:  # noqa: BLE001 - a malformed job has no result key to answer on
                 logger.error("worker %d: undecodable job dropped: %s", index, e)
                 continue
-            result = await self.handle(job)
+            try:
+                result = await self.handle(job)
+            except Exception as e:  # noqa: BLE001 - one bad job must not kill the worker
+                logger.exception("worker %d: job %s failed", index, job.job_id)
+                result = failure(job.job_id, "processor_error", repr(e))
             try:
                 key = self._keys.result(job.job_id)
-                await self._client.lpush(key, encode_result(result))
-                await self._client.expire(key, RESULT_TTL_S)
+                pipe = self._client.pipeline(transaction=True)
+                pipe.lpush(key, encode_result(result))
+                pipe.expire(key, RESULT_TTL_S)
+                await pipe.execute()
             except Exception as e:  # noqa: BLE001 - the servicer times out and retries
                 logger.warning("worker %d: result push failed for %s: %s", index, job.job_id, e)
 
@@ -171,7 +196,7 @@ async def handle(self, job: Job) -> JobResult:
 
         multi_modal_data: dict[str, list] = {}
         for index, item in enumerate(job.items):
-            scheme = _url_scheme(item.url)
+            scheme = url_scheme(item.url)
             if scheme not in self._accepted:
                 return failure(
                     job.job_id,
@@ -180,18 +205,12 @@ async def handle(self, job: Job) -> JobResult:
                 )
             try:
                 media = await self._fetch(item.modality, item.url)
-            except ValueError as e:
-                code = "domain_not_allowed" if "domain" in str(e).lower() else "fetch_failed"
-                return failure(job.job_id, code, f"item {index}: {e}")
-            except Exception as e:  # noqa: BLE001 - network/decoder errors are client-visible
-                return failure(job.job_id, "fetch_failed", f"item {index}: {e}")
+            except Exception as e:  # noqa: BLE001 - fetch failures are client-visible codes
+                return failure(job.job_id, classify_fetch_error(e), f"item {index}: {e}")
             multi_modal_data.setdefault(item.modality, []).append(media)
         fetched = time.time()
 
-        from vllm import TokensPrompt
-        from vllm.v1.serial_utils import MsgpackEncoder
-
-        prompt = TokensPrompt(
+        prompt = self._tokens_prompt(
             prompt_token_ids=list(job.prompt_token_ids), multi_modal_data=multi_modal_data
         )
         if job.prompt:
@@ -200,17 +219,14 @@ async def handle(self, job: Job) -> JobResult:
             engine_input = await self._renderer.process_for_engine_async(
                 prompt, arrival_time=0.0, skip_mm_cache=True
             )
-        except RuntimeError as e:
-            return failure(job.job_id, "placeholder_mismatch", str(e))
-        except ValueError as e:
-            code = "limit_exceeded" if "at most" in str(e).lower() else "decode_failed"
+        except Exception as e:  # noqa: BLE001 - classified into a result code
+            code = classify_process_error(e)
+            if code == "processor_error":
+                logger.exception("processing failed for job %s", job.job_id)
             return failure(job.job_id, code, str(e))
-        except Exception as e:  # noqa: BLE001 - processor bugs are retryable elsewhere
-            logger.exception("processing failed for job %s", job.job_id)
-            return failure(job.job_id, "processor_error", str(e))
         processed = time.time()
 
-        encoder = MsgpackEncoder(size_threshold=2**62)
+        encoder = self._encoder
         mm_kwargs: dict[str, list[bytes]] = {}
         for modality, items in engine_input["mm_kwargs"].items():
             blobs = []
diff --git a/grpc_servicer/tests/test_vllm_mm_sidecar.py b/grpc_servicer/tests/test_vllm_mm_sidecar.py
new file mode 100644
index 0000000000..c6f04ed5cb
--- /dev/null
+++ b/grpc_servicer/tests/test_vllm_mm_sidecar.py
@@ -0,0 +1,194 @@
+"""Unit tests for the reference media sidecar loop (engine-free, fake Redis).
+
+Run with: pytest grpc_servicer/tests/test_vllm_mm_sidecar.py
+"""
+
+import asyncio
+import time
+
+import pytest
+
+pytest.importorskip("msgspec")
+from smg_grpc_servicer import mm_sidecar_protocol as proto  # noqa: E402
+from smg_grpc_servicer.vllm import mm_sidecar  # noqa: E402
+
+
+class FakePipeline:
+    def __init__(self, store):
+        self.store = store
+        self.ops = []
+
+    def hset(self, key, mapping=None):
+        self.ops.append(("hset", key, dict(mapping or {})))
+
+    def lpush(self, key, value):
+        self.ops.append(("lpush", key, value))
+
+    def expire(self, key, seconds):
+        self.ops.append(("expire", key, seconds))
+
+    async def execute(self):
+        self.store.executed.append(list(self.ops))
+        if self.store.execute_fail is not None:
+            raise self.store.execute_fail
+        return [True] * len(self.ops)
+
+
+class FakeRedis:
+    """Records transactions; brpop yields queued jobs then stops the loop."""
+
+    def __init__(self, jobs=()):
+        self.jobs = list(jobs)
+        self.executed = []
+        self.execute_fail = None
+
+    def pipeline(self, transaction=True):
+        assert transaction
+        return FakePipeline(self)
+
+    async def brpop(self, key, timeout=0):
+        if self.jobs:
+            return (key.encode(), proto.encode_job(self.jobs.pop(0)))
+        raise asyncio.CancelledError
+
+
+def fingerprint():
+    return proto.Fingerprint(
+        model="m",
+        vllm_version="0.27.1",
+        dtype="torch.bfloat16",
+        video_backend="opencv",
+        media_io_kwargs="{}",
+        mm_processor_kwargs="{}",
+    )
+
+
+def sidecar(client):
+    s = mm_sidecar.Sidecar.__new__(mm_sidecar.Sidecar)
+    s._client = client
+    s._fingerprint = fingerprint()
+    s._keys = proto.Keys.for_namespace("t")
+    s._schemes = "http,https,data"
+    s._accepted = {"http", "https", "data"}
+    s._started_at = time.time()
+    s._concurrency = 1
+    return s
+
+
+def job(job_id="j1"):
+    return proto.Job(
+        v=proto.SCHEMA_VERSION,
+        job_id=job_id,
+        request_id="r1",
+        fingerprint=fingerprint(),
+        prompt_token_ids=[1, 2, 3],
+        prompt=None,
+        items=[proto.JobItem(modality="image", url="https://a/1.png")],
+        enqueued_ms=0,
+        deadline_ms=int(time.time() * 1000) + 60_000,
+    )
+
+
+def run(coro):
+    return asyncio.new_event_loop().run_until_complete(coro)
+
+
+class TestClassification:
+    def test_fetch_errors(self):
+        assert mm_sidecar.classify_fetch_error(ValueError("domain not allowed")) == (
+            "domain_not_allowed"
+        )
+        assert mm_sidecar.classify_fetch_error(ValueError("bad base64")) == "fetch_failed"
+        assert mm_sidecar.classify_fetch_error(TimeoutError("slow")) == "fetch_failed"
+
+    def test_process_errors(self):
+        assert mm_sidecar.classify_process_error(RuntimeError("placeholder")) == (
+            "placeholder_mismatch"
+        )
+        # vLLM 0.27 wording: "At most 1 image(s) may be provided in one prompt."
+        assert (
+            mm_sidecar.classify_process_error(
+                ValueError("At most 1 image(s) may be provided in one prompt.")
+            )
+            == "limit_exceeded"
+        )
+        assert mm_sidecar.classify_process_error(ValueError("cannot identify image")) == (
+            "decode_failed"
+        )
+        assert mm_sidecar.classify_process_error(KeyError("mm_kwargs")) == "processor_error"
+
+
+class TestWorkerLoop:
+    def test_a_failing_handle_answers_and_keeps_the_loop_alive(self, monkeypatch):
+        client = FakeRedis(jobs=[job("j1"), job("j2")])
+        s = sidecar(client)
+        calls = []
+
+        async def handle(j):
+            calls.append(j.job_id)
+            if j.job_id == "j1":
+                raise KeyError("mm_kwargs")
+            return proto.JobResult(v=1, job_id=j.job_id, ok=True)
+
+        monkeypatch.setattr(s, "handle", handle)
+        with pytest.raises(asyncio.CancelledError):
+            run(s._worker(0))
+        assert calls == ["j1", "j2"], "the second job was still served"
+        first, second = client.executed
+        assert first[0][0] == "lpush" and first[0][1] == s._keys.result("j1")
+        answered = proto.decode_result(first[0][2])
+        assert not answered.ok and answered.code == "processor_error"
+        assert first[1] == ("expire", s._keys.result("j1"), proto.RESULT_TTL_S)
+        assert proto.decode_result(second[0][2]).ok
+
+    def test_result_push_is_one_transaction(self, monkeypatch):
+        client = FakeRedis(jobs=[job("j3")])
+        s = sidecar(client)
+
+        async def handle(j):
+            return proto.JobResult(v=1, job_id=j.job_id, ok=True)
+
+        monkeypatch.setattr(s, "handle", handle)
+        with pytest.raises(asyncio.CancelledError):
+            run(s._worker(0))
+        (ops,) = client.executed
+        assert [op[0] for op in ops] == ["lpush", "expire"]
+
+
+class TestHeartbeat:
+    def test_hello_and_ttl_are_one_transaction(self, monkeypatch):
+        client = FakeRedis()
+        s = sidecar(client)
+        monkeypatch.setattr(mm_sidecar, "HELLO_REFRESH_S", 0)
+
+        async def stop_after_two():
+            task = asyncio.ensure_future(s._heartbeat())
+            while len(client.executed) < 2:
+                await asyncio.sleep(0)
+            task.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await task
+
+        run(stop_after_two())
+        ops = client.executed[0]
+        assert ops[0][0] == "hset" and ops[0][1] == s._keys.hello
+        assert ops[0][2]["schemes"] == "http,https,data"
+        assert ops[0][2]["model"] == "m"
+        assert ops[1] == ("expire", s._keys.hello, proto.HELLO_TTL_S)
+
+    def test_heartbeat_survives_a_failed_refresh(self, monkeypatch):
+        client = FakeRedis()
+        client.execute_fail = ConnectionError("redis down")
+        s = sidecar(client)
+        monkeypatch.setattr(mm_sidecar, "HELLO_REFRESH_S", 0)
+
+        async def stop_after_two():
+            task = asyncio.ensure_future(s._heartbeat())
+            while len(client.executed) < 2:
+                await asyncio.sleep(0)
+            task.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await task
+
+        run(stop_after_two())
+        assert len(client.executed) >= 2
diff --git a/grpc_servicer/tests/test_vllm_mm_sidecar_protocol.py b/grpc_servicer/tests/test_vllm_mm_sidecar_protocol.py
index 19d4a449a0..81f785401a 100644
--- a/grpc_servicer/tests/test_vllm_mm_sidecar_protocol.py
+++ b/grpc_servicer/tests/test_vllm_mm_sidecar_protocol.py
@@ -23,13 +23,15 @@ def fingerprint(**overrides) -> proto.Fingerprint:
 
 
 class TestFingerprint:
-    def test_namespace_is_deterministic_and_dtype_sensitive(self):
+    def test_namespace_covers_every_fingerprint_field(self):
         a = fingerprint().namespace()
         assert a == fingerprint().namespace()
         assert len(a) == 16
+        # Fleets that would fail the fingerprint check never share keys.
         assert a != fingerprint(dtype="torch.float16").namespace()
-        # Kwargs are not part of the namespace; they are checked by fingerprint.
-        assert a == fingerprint(media_io_kwargs='{"video":{"num_frames":8}}').namespace()
+        assert a != fingerprint(video_backend="decord").namespace()
+        assert a != fingerprint(media_io_kwargs='{"video":{"num_frames":8}}').namespace()
+        assert a != fingerprint(mm_processor_kwargs='{"max_pixels":1}').namespace()
 
     def test_override_wins(self):
         assert proto.resolve_namespace(fingerprint(), " prod-a ") == "prod-a"
diff --git a/grpc_servicer/tests/test_vllm_redis_media_processor.py b/grpc_servicer/tests/test_vllm_redis_media_processor.py
index 21463ab162..9d0fc460d3 100644
--- a/grpc_servicer/tests/test_vllm_redis_media_processor.py
+++ b/grpc_servicer/tests/test_vllm_redis_media_processor.py
@@ -31,6 +31,7 @@ class _Item:
 class _ModelConfig:
     dtype = "bf16"
     is_multimodal_model = True
+    allowed_local_media_path = ""
 
 
 class _Engine:
@@ -218,6 +219,14 @@ def test_process_caps_inline_payloads_before_queueing(self):
         assert client.pushed == []
 
 
+class TestRedisClientImport:
+    def test_missing_client_names_the_extra(self, monkeypatch):
+        monkeypatch.setitem(sys.modules, "redis", None)
+        monkeypatch.setitem(sys.modules, "redis.asyncio", None)
+        with pytest.raises(ValueError, match=r"vllm,vllm-redis"):
+            mm_processor._redis_client("redis://127.0.0.1:6379/0")
+
+
 class TestBuild:
     def test_redis_mode_builds_processor(self, monkeypatch):
         monkeypatch.setattr(mm_processor, "engine_fingerprint", lambda engine: fingerprint())

From a945849118d68cbedbf34f91e064abc74aa9fee2 Mon Sep 17 00:00:00 2001
From: Chang Su <8605658+CatherineSue@users.noreply.github.com>
Date: Mon, 14 Sep 2026 15:44:38 -0700
Subject: [PATCH 03/11] fix(vllm-grpc): fingerprint the per-prompt media limit,
 harden sidecar result decoding

- Fingerprint gains limit_per_prompt: on the redis path the sidecar's limit is
  the only one applied, so a worker/sidecar skew is now a startup-visible
  "not advertised" instead of a runtime 400 storm
- an undecodable or foreign-schema result is retryable (UNAVAILABLE), not an
  INTERNAL abort
- the no-hello probe message carries the worker fingerprint so a namespace
  disagreement is diffable against the sidecar's startup line
- the redis backend enforces SMG_VLLM_MM_MAX_ITEMS before queueing

Signed-off-by: Chang Su <8605658+CatherineSue@users.noreply.github.com>
---
 grpc_servicer/README.md                       |  9 ++++---
 .../smg_grpc_servicer/mm_sidecar_protocol.py  |  2 ++
 .../smg_grpc_servicer/vllm/mm_processor.py    | 22 ++++++++++++++--
 .../smg_grpc_servicer/vllm/mm_sidecar.py      |  1 +
 grpc_servicer/tests/test_vllm_mm_sidecar.py   |  1 +
 .../tests/test_vllm_mm_sidecar_protocol.py    |  2 ++
 .../tests/test_vllm_redis_media_processor.py  | 26 +++++++++++++++++--
 7 files changed, 55 insertions(+), 8 deletions(-)

diff --git a/grpc_servicer/README.md b/grpc_servicer/README.md
index 17410b8b0a..1ae4233698 100644
--- a/grpc_servicer/README.md
+++ b/grpc_servicer/README.md
@@ -82,10 +82,11 @@ SMG_VLLM_MM_PROCESSOR=redis SMG_VLLM_MM_REDIS_URL=redis://127.0.0.1:6379/0 \
 ```
 
 The sidecar and the worker must agree on model, vLLM version, dtype, video
-backend and media/processor kwargs (the key namespace is derived from all of
-them): the worker advertises `mm_processor=redis` only while a sidecar with a
-matching fingerprint keeps its `hello` key alive, and rejects results that
-disagree. Jobs and results travel over Redis lists under
+backend, media/processor kwargs and `--limit-mm-per-prompt` (pass the flag to
+both processes; the sidecar's limit is the one that applies, and the key
+namespace is derived from all of these): the worker advertises
+`mm_processor=redis` only while a sidecar with a matching fingerprint keeps its
+`hello` key alive, and rejects results that disagree. Jobs and results travel over Redis lists under
 `smg:mm:v1:{namespace}`; results carry full tensors keyed by a per-attempt job
 id and expire after 120 s. Knobs: `SMG_VLLM_MM_SIDECAR_TIMEOUT_MS` (30000),
 `SMG_VLLM_MM_SIDECAR_MAX_QUEUE` (256, fail fast when the queue is deeper),
diff --git a/grpc_servicer/smg_grpc_servicer/mm_sidecar_protocol.py b/grpc_servicer/smg_grpc_servicer/mm_sidecar_protocol.py
index a5822a78b5..b0fc61b836 100644
--- a/grpc_servicer/smg_grpc_servicer/mm_sidecar_protocol.py
+++ b/grpc_servicer/smg_grpc_servicer/mm_sidecar_protocol.py
@@ -57,6 +57,8 @@ class Fingerprint(msgspec.Struct, frozen=True):
     video_backend: str
     media_io_kwargs: str
     mm_processor_kwargs: str
+    # The sidecar's limit is the only one applied on the redis path.
+    limit_per_prompt: str
 
     def namespace(self) -> str:
         """Key namespace: fleets that would fail `mismatches` never share keys."""
diff --git a/grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py b/grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py
index 959bda4c8d..86942cc2f3 100644
--- a/grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py
+++ b/grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py
@@ -260,6 +260,7 @@ def engine_fingerprint(engine) -> Fingerprint:
         mm_processor_kwargs=json.dumps(
             mm_config.mm_processor_kwargs or {}, sort_keys=True, default=str
         ),
+        limit_per_prompt=json.dumps(mm_config.limit_per_prompt or {}, sort_keys=True, default=str),
     )
 
 
@@ -295,6 +296,7 @@ def __init__(
         namespace: str | None = None,
         max_item_bytes: int = DEFAULT_MAX_ITEM_BYTES,
         max_inflight: int = DEFAULT_MAX_INFLIGHT,
+        max_items: int = DEFAULT_MAX_ITEMS,
         client=None,
     ) -> None:
         self._engine = engine
@@ -302,6 +304,7 @@ def __init__(
         self._timeout_ms = timeout_ms
         self._max_queue = max_queue
         self._max_item_bytes = max_item_bytes
+        self._max_items = max_items
         self.max_inflight = max_inflight
         self._keys = Keys.for_namespace(resolve_namespace(fingerprint, namespace))
         self._client = client if client is not None else _redis_client(redis_url)
@@ -320,7 +323,13 @@ async def probe(self) -> bool:
             return False
         remote = Fingerprint.from_hello(hello) if hello else None
         if remote is None:
-            self._log_probe_once("no sidecar hello at %s", self._keys.hello)
+            # A sidecar built for any other fingerprint lives under another
+            # namespace, so this is the message every disagreement produces.
+            self._log_probe_once(
+                "no sidecar hello at %s (worker fingerprint=%s)",
+                self._keys.hello,
+                self._fingerprint.to_hello(),
+            )
             return False
         mismatches = self._fingerprint.mismatches(remote)
         if mismatches:
@@ -352,6 +361,7 @@ async def process(
         *,
         request_id: str = "",
     ):
+        enforce_item_count(items, self._max_items)
         enforce_item_bytes(items, self._max_item_bytes)
         now_ms = int(time.time() * 1000)
         job = Job(
@@ -389,7 +399,14 @@ async def _submit_and_wait(self, job: Job) -> JobResult:
                 f"sidecar_timeout: no result for job {job.job_id} within {self._timeout_ms} ms"
             )
         _, raw = popped
-        result = decode_result(raw)
+        try:
+            result = decode_result(raw)
+        except Exception as e:  # noqa: BLE001 - an undecodable result means "try another worker"
+            raise MmProcessorUnavailable(f"sidecar_protocol: undecodable result: {e}") from e
+        if result.v != SCHEMA_VERSION:
+            raise MmProcessorUnavailable(
+                f"sidecar_protocol: result schema v{result.v}, worker speaks v{SCHEMA_VERSION}"
+            )
         if result.job_id != job.job_id:
             raise MmProcessorUnavailable(
                 f"sidecar_protocol: result for job {result.job_id} on key of {job.job_id}"
@@ -486,4 +503,5 @@ def build_mm_processor(engine, *, env: Mapping[str, str] = os.environ):
         namespace=env.get(ENV_NAMESPACE),
         max_item_bytes=max_item_bytes,
         max_inflight=max_inflight,
+        max_items=max_items,
     )
diff --git a/grpc_servicer/smg_grpc_servicer/vllm/mm_sidecar.py b/grpc_servicer/smg_grpc_servicer/vllm/mm_sidecar.py
index aadf26f604..4ca93fe89b 100644
--- a/grpc_servicer/smg_grpc_servicer/vllm/mm_sidecar.py
+++ b/grpc_servicer/smg_grpc_servicer/vllm/mm_sidecar.py
@@ -69,6 +69,7 @@ def config_fingerprint(vllm_config) -> Fingerprint:
         mm_processor_kwargs=json.dumps(
             mm_config.mm_processor_kwargs or {}, sort_keys=True, default=str
         ),
+        limit_per_prompt=json.dumps(mm_config.limit_per_prompt or {}, sort_keys=True, default=str),
     )
 
 
diff --git a/grpc_servicer/tests/test_vllm_mm_sidecar.py b/grpc_servicer/tests/test_vllm_mm_sidecar.py
index c6f04ed5cb..e3b2b48855 100644
--- a/grpc_servicer/tests/test_vllm_mm_sidecar.py
+++ b/grpc_servicer/tests/test_vllm_mm_sidecar.py
@@ -60,6 +60,7 @@ def fingerprint():
         video_backend="opencv",
         media_io_kwargs="{}",
         mm_processor_kwargs="{}",
+        limit_per_prompt="{}",
     )
 
 
diff --git a/grpc_servicer/tests/test_vllm_mm_sidecar_protocol.py b/grpc_servicer/tests/test_vllm_mm_sidecar_protocol.py
index 81f785401a..738718905c 100644
--- a/grpc_servicer/tests/test_vllm_mm_sidecar_protocol.py
+++ b/grpc_servicer/tests/test_vllm_mm_sidecar_protocol.py
@@ -17,6 +17,7 @@ def fingerprint(**overrides) -> proto.Fingerprint:
         "video_backend": "opencv",
         "media_io_kwargs": "{}",
         "mm_processor_kwargs": "{}",
+        "limit_per_prompt": "{}",
     }
     values.update(overrides)
     return proto.Fingerprint(**values)
@@ -32,6 +33,7 @@ def test_namespace_covers_every_fingerprint_field(self):
         assert a != fingerprint(video_backend="decord").namespace()
         assert a != fingerprint(media_io_kwargs='{"video":{"num_frames":8}}').namespace()
         assert a != fingerprint(mm_processor_kwargs='{"max_pixels":1}').namespace()
+        assert a != fingerprint(limit_per_prompt='{"image":8}').namespace()
 
     def test_override_wins(self):
         assert proto.resolve_namespace(fingerprint(), " prod-a ") == "prod-a"
diff --git a/grpc_servicer/tests/test_vllm_redis_media_processor.py b/grpc_servicer/tests/test_vllm_redis_media_processor.py
index 9d0fc460d3..fdcd4cef56 100644
--- a/grpc_servicer/tests/test_vllm_redis_media_processor.py
+++ b/grpc_servicer/tests/test_vllm_redis_media_processor.py
@@ -67,7 +67,8 @@ async def brpop(self, key, timeout=0):
             return None
         job = proto.decode_job(self.pushed[-1][1])
         result = self.responder(job)
-        return (key.encode(), proto.encode_result(result))
+        payload = result if isinstance(result, bytes) else proto.encode_result(result)
+        return (key.encode(), payload)
 
     async def expire(self, key, seconds):
         self.expired[key] = seconds
@@ -81,6 +82,7 @@ def fingerprint(**overrides):
         "video_backend": "opencv",
         "media_io_kwargs": "{}",
         "mm_processor_kwargs": "{}",
+        "limit_per_prompt": "{}",
     }
     values.update(overrides)
     return proto.Fingerprint(**values)
@@ -98,7 +100,7 @@ def run(coro):
 
 def ok_result(job, **extra):
     return proto.JobResult(
-        v=proto.SCHEMA_VERSION,
+        v=extra.pop("v", proto.SCHEMA_VERSION),
         job_id=job.job_id,
         ok=True,
         fingerprint=extra.pop("fingerprint", fingerprint()),
@@ -175,6 +177,16 @@ def test_retryable_error_codes_stay_unavailable(self):
         with pytest.raises(mm_processor.MmProcessorUnavailable, match="expired"):
             run(processor(client)._submit_and_wait(self.job()))
 
+    def test_undecodable_result_is_retryable(self):
+        client = FakeRedis(responder=lambda job: b"\xc1not-a-result")
+        with pytest.raises(mm_processor.MmProcessorUnavailable, match="undecodable result"):
+            run(processor(client)._submit_and_wait(self.job()))
+
+    def test_result_schema_version_mismatch_is_retryable(self):
+        client = FakeRedis(responder=lambda job: ok_result(job, v=proto.SCHEMA_VERSION + 1))
+        with pytest.raises(mm_processor.MmProcessorUnavailable, match="result schema v2"):
+            run(processor(client)._submit_and_wait(self.job()))
+
     def test_result_fingerprint_mismatch_is_unavailable(self):
         client = FakeRedis(
             responder=lambda job: ok_result(job, fingerprint=fingerprint(vllm_version="0.1"))
@@ -219,6 +231,16 @@ def test_process_caps_inline_payloads_before_queueing(self):
         assert client.pushed == []
 
 
+class TestItemCap:
+    def test_process_caps_item_count_before_queueing(self):
+        client = FakeRedis(responder=ok_result)
+        p = processor(client, max_items=1)
+        items = [_Item("image", "https://a/1.png"), _Item("image", "https://a/2.png")]
+        with pytest.raises(ValueError, match="SMG_VLLM_MM_MAX_ITEMS"):
+            run(p.process([1, 2, 3], None, items, 0.0))
+        assert client.pushed == []
+
+
 class TestRedisClientImport:
     def test_missing_client_names_the_extra(self, monkeypatch):
         monkeypatch.setitem(sys.modules, "redis", None)

From 13b502614cc75051aa831ffb02301774a40a73fa Mon Sep 17 00:00:00 2001
From: Chang Su <8605658+CatherineSue@users.noreply.github.com>
Date: Mon, 14 Sep 2026 15:58:38 -0700
Subject: [PATCH 04/11] fix(vllm-grpc): one fingerprint derivation, resolved
 media limits, protocol check in probe

- engine_fingerprint and the sidecar's config_fingerprint share
  fingerprint_from_model_config, pinned by a test feeding one fake config
  through both entry points
- limit_per_prompt is resolved per modality through vLLM's accessor before
  hashing, so equivalent spellings share a namespace
- probe() rejects a hello whose protocol schema is not the worker's, so
  protocol skew is a startup-visible "not advertised" instead of a
  per-request UNAVAILABLE

Signed-off-by: Chang Su <8605658+CatherineSue@users.noreply.github.com>
---
 grpc_servicer/README.md                       |  3 +-
 .../smg_grpc_servicer/mm_sidecar_protocol.py  |  8 ++-
 .../smg_grpc_servicer/vllm/mm_processor.py    | 28 ++++++++--
 .../smg_grpc_servicer/vllm/mm_sidecar.py      | 19 +------
 grpc_servicer/tests/test_vllm_mm_sidecar.py   | 56 ++++++++++++++++++-
 .../tests/test_vllm_redis_media_processor.py  | 11 ++++
 6 files changed, 100 insertions(+), 25 deletions(-)

diff --git a/grpc_servicer/README.md b/grpc_servicer/README.md
index 1ae4233698..e5879b5579 100644
--- a/grpc_servicer/README.md
+++ b/grpc_servicer/README.md
@@ -83,7 +83,8 @@ SMG_VLLM_MM_PROCESSOR=redis SMG_VLLM_MM_REDIS_URL=redis://127.0.0.1:6379/0 \
 
 The sidecar and the worker must agree on model, vLLM version, dtype, video
 backend, media/processor kwargs and `--limit-mm-per-prompt` (pass the flag to
-both processes; the sidecar's limit is the one that applies, and the key
+both processes; the sidecar's limit is the one that applies, the limit is
+resolved per modality before hashing so equivalent spellings match, and the key
 namespace is derived from all of these): the worker advertises
 `mm_processor=redis` only while a sidecar with a matching fingerprint keeps its
 `hello` key alive, and rejects results that disagree. Jobs and results travel over Redis lists under
diff --git a/grpc_servicer/smg_grpc_servicer/mm_sidecar_protocol.py b/grpc_servicer/smg_grpc_servicer/mm_sidecar_protocol.py
index b0fc61b836..f5288206d8 100644
--- a/grpc_servicer/smg_grpc_servicer/mm_sidecar_protocol.py
+++ b/grpc_servicer/smg_grpc_servicer/mm_sidecar_protocol.py
@@ -86,13 +86,17 @@ def _text(value: bytes | str) -> str:
     return value.decode("utf-8", errors="replace") if isinstance(value, bytes) else value
 
 
-def hello_schemes(mapping: Mapping[bytes | str, bytes | str]) -> str:
+def hello_field(mapping: Mapping[bytes | str, bytes | str], name: str) -> str:
     for key, value in mapping.items():
-        if _text(key) == "schemes":
+        if _text(key) == name:
             return _text(value)
     return ""
 
 
+def hello_schemes(mapping: Mapping[bytes | str, bytes | str]) -> str:
+    return hello_field(mapping, "schemes")
+
+
 def resolve_namespace(fingerprint: Fingerprint, override: str | None) -> str:
     override = (override or "").strip()
     return override if override else fingerprint.namespace()
diff --git a/grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py b/grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py
index 86942cc2f3..92e8b198fe 100644
--- a/grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py
+++ b/grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py
@@ -35,6 +35,7 @@
     Keys,
     decode_result,
     encode_job,
+    hello_field,
     hello_schemes,
     resolve_namespace,
 )
@@ -244,12 +245,20 @@ async def _fetch(self, index: int, item):
         raise ValueError(f"unsupported media modality {item.modality!r}")
 
 
-def engine_fingerprint(engine) -> Fingerprint:
-    """What the sidecar must match to process media for this engine."""
+MM_LIMIT_MODALITIES = ("image", "video", "audio")
+
+
+def resolved_mm_limits(mm_config) -> dict[str, int]:
+    """Per-modality prompt limits as vLLM resolves them (999 when unset), so
+    equivalent spellings of --limit-mm-per-prompt match."""
+    return {modality: mm_config.get_limit_per_prompt(modality) for modality in MM_LIMIT_MODALITIES}
+
+
+def fingerprint_from_model_config(model_config) -> Fingerprint:
+    """The one derivation the worker and the sidecar share."""
     import vllm
     from vllm import envs
 
-    model_config = engine.model_config
     mm_config = model_config.get_multimodal_config()
     return Fingerprint(
         model=model_config.model,
@@ -260,10 +269,15 @@ def engine_fingerprint(engine) -> Fingerprint:
         mm_processor_kwargs=json.dumps(
             mm_config.mm_processor_kwargs or {}, sort_keys=True, default=str
         ),
-        limit_per_prompt=json.dumps(mm_config.limit_per_prompt or {}, sort_keys=True, default=str),
+        limit_per_prompt=json.dumps(resolved_mm_limits(mm_config), sort_keys=True),
     )
 
 
+def engine_fingerprint(engine) -> Fingerprint:
+    """What the sidecar must match to process media for this engine."""
+    return fingerprint_from_model_config(engine.model_config)
+
+
 def _cast_floats(data, dtype):
     """Cast floating tensors to the model dtype, as the HF processor path does."""
     import torch
@@ -321,6 +335,12 @@ async def probe(self) -> bool:
         except Exception as e:  # noqa: BLE001 - any transport failure means "not advertised"
             self._log_probe_once("redis unreachable: %s", e)
             return False
+        schema = hello_field(hello, "schema") if hello else ""
+        if hello and schema != str(SCHEMA_VERSION):
+            self._log_probe_once(
+                "sidecar speaks protocol v%s, worker speaks v%s", schema or "?", SCHEMA_VERSION
+            )
+            return False
         remote = Fingerprint.from_hello(hello) if hello else None
         if remote is None:
             # A sidecar built for any other fingerprint lives under another
diff --git a/grpc_servicer/smg_grpc_servicer/vllm/mm_sidecar.py b/grpc_servicer/smg_grpc_servicer/vllm/mm_sidecar.py
index 4ca93fe89b..a01448a6b4 100644
--- a/grpc_servicer/smg_grpc_servicer/vllm/mm_sidecar.py
+++ b/grpc_servicer/smg_grpc_servicer/vllm/mm_sidecar.py
@@ -15,7 +15,6 @@
 
 import argparse
 import asyncio
-import json
 import logging
 import time
 
@@ -37,6 +36,7 @@
     resolve_namespace,
 )
 from smg_grpc_servicer.vllm.media_refs import advertised_schemes, parse_scheme_list, url_scheme
+from smg_grpc_servicer.vllm.mm_processor import fingerprint_from_model_config
 
 logger = logging.getLogger("smg_grpc_servicer.vllm.mm_sidecar")
 
@@ -55,22 +55,7 @@ def build_config(args: argparse.Namespace):
 
 
 def config_fingerprint(vllm_config) -> Fingerprint:
-    import vllm
-    from vllm import envs
-
-    model_config = vllm_config.model_config
-    mm_config = model_config.get_multimodal_config()
-    return Fingerprint(
-        model=model_config.model,
-        vllm_version=vllm.__version__,
-        dtype=str(model_config.dtype),
-        video_backend=envs.VLLM_VIDEO_LOADER_BACKEND,
-        media_io_kwargs=json.dumps(mm_config.media_io_kwargs or {}, sort_keys=True, default=str),
-        mm_processor_kwargs=json.dumps(
-            mm_config.mm_processor_kwargs or {}, sort_keys=True, default=str
-        ),
-        limit_per_prompt=json.dumps(mm_config.limit_per_prompt or {}, sort_keys=True, default=str),
-    )
+    return fingerprint_from_model_config(vllm_config.model_config)
 
 
 # Codes come from exception types where vLLM offers one. The two substring
diff --git a/grpc_servicer/tests/test_vllm_mm_sidecar.py b/grpc_servicer/tests/test_vllm_mm_sidecar.py
index e3b2b48855..3f577a1560 100644
--- a/grpc_servicer/tests/test_vllm_mm_sidecar.py
+++ b/grpc_servicer/tests/test_vllm_mm_sidecar.py
@@ -4,13 +4,15 @@
 """
 
 import asyncio
+import sys
 import time
+import types
 
 import pytest
 
 pytest.importorskip("msgspec")
 from smg_grpc_servicer import mm_sidecar_protocol as proto  # noqa: E402
-from smg_grpc_servicer.vllm import mm_sidecar  # noqa: E402
+from smg_grpc_servicer.vllm import mm_processor, mm_sidecar  # noqa: E402
 
 
 class FakePipeline:
@@ -64,6 +66,58 @@ def fingerprint():
     )
 
 
+class _MmConfig:
+    def __init__(self, limit_per_prompt):
+        self.limit_per_prompt = limit_per_prompt
+        self.media_io_kwargs = {"video": {"num_frames": 8}}
+        self.mm_processor_kwargs = None
+
+    def get_limit_per_prompt(self, modality):
+        # Mirrors vLLM: 999 when unset, otherwise the validated options' count.
+        value = self.limit_per_prompt.get(modality, 999)
+        return value["count"] if isinstance(value, dict) else value
+
+
+class _ModelConfig:
+    def __init__(self, limit_per_prompt):
+        self.model = "m"
+        self.dtype = "torch.bfloat16"
+        self._mm = _MmConfig(limit_per_prompt)
+
+    def get_multimodal_config(self):
+        return self._mm
+
+
+def _fake_vllm(monkeypatch):
+    fake = types.ModuleType("vllm")
+    fake.__version__ = "0.27.1"
+    fake.envs = types.SimpleNamespace(VLLM_VIDEO_LOADER_BACKEND="opencv")
+    monkeypatch.setitem(sys.modules, "vllm", fake)
+
+
+class TestFingerprintDerivation:
+    def test_worker_and_sidecar_derive_the_same_fingerprint(self, monkeypatch):
+        _fake_vllm(monkeypatch)
+        engine = types.SimpleNamespace(model_config=_ModelConfig({"image": 8}))
+        vllm_config = types.SimpleNamespace(model_config=_ModelConfig({"image": 8}))
+        worker = mm_processor.engine_fingerprint(engine)
+        assert worker == mm_sidecar.config_fingerprint(vllm_config)
+        assert worker.limit_per_prompt == '{"audio": 999, "image": 8, "video": 999}'
+
+    def test_equivalent_limit_spellings_share_a_namespace(self, monkeypatch):
+        _fake_vllm(monkeypatch)
+        explicit = mm_processor.fingerprint_from_model_config(_ModelConfig({"image": 999}))
+        implicit = mm_processor.fingerprint_from_model_config(_ModelConfig({}))
+        nested = mm_processor.fingerprint_from_model_config(
+            _ModelConfig({"image": {"count": 999, "max_pixels": 4}})
+        )
+        assert explicit == implicit == nested
+        assert (
+            explicit.namespace()
+            != mm_processor.fingerprint_from_model_config(_ModelConfig({"image": 8})).namespace()
+        )
+
+
 def sidecar(client):
     s = mm_sidecar.Sidecar.__new__(mm_sidecar.Sidecar)
     s._client = client
diff --git a/grpc_servicer/tests/test_vllm_redis_media_processor.py b/grpc_servicer/tests/test_vllm_redis_media_processor.py
index fdcd4cef56..9a82e94ca4 100644
--- a/grpc_servicer/tests/test_vllm_redis_media_processor.py
+++ b/grpc_servicer/tests/test_vllm_redis_media_processor.py
@@ -121,10 +121,21 @@ def test_redis_error_is_not_advertised(self):
 
     def test_fingerprint_mismatch_is_not_advertised(self):
         hello = {k.encode(): v.encode() for k, v in fingerprint(dtype="x").to_hello().items()}
+        hello[b"schema"] = str(proto.SCHEMA_VERSION).encode()
+        assert run(processor(FakeRedis(hello=hello)).probe()) is False
+
+    def test_protocol_schema_skew_is_not_advertised(self):
+        hello = {k.encode(): v.encode() for k, v in fingerprint().to_hello().items()}
+        hello[b"schema"] = str(proto.SCHEMA_VERSION + 1).encode()
+        assert run(processor(FakeRedis(hello=hello)).probe()) is False
+
+    def test_hello_without_schema_is_not_advertised(self):
+        hello = {k.encode(): v.encode() for k, v in fingerprint().to_hello().items()}
         assert run(processor(FakeRedis(hello=hello)).probe()) is False
 
     def test_matching_hello_advertises_and_adopts_schemes(self):
         hello = {k.encode(): v.encode() for k, v in fingerprint().to_hello().items()}
+        hello[b"schema"] = str(proto.SCHEMA_VERSION).encode()
         hello[b"schemes"] = b"http,https,data,file"
         p = processor(FakeRedis(hello=hello))
         assert run(p.probe()) is True

From 9ea2d3df3a155f07cbcb056662fe5e78c5fd1540 Mon Sep 17 00:00:00 2001
From: Chang Su <8605658+CatherineSue@users.noreply.github.com>
Date: Mon, 14 Sep 2026 16:06:50 -0700
Subject: [PATCH 05/11] fix(vllm-grpc): fingerprint validated media limits with
 their sibling options

resolved_mm_limits carries the effective count from get_limit_per_prompt
plus every non-default sibling option, over the standard modalities and
any configured key, so equivalent --limit-mm-per-prompt spellings still
match while a skewed num_frames/width or an extra modality fails closed.
A vllm-gated test pins the derivation against the real MultiModalConfig.

Signed-off-by: Chang Su <8605658+CatherineSue@users.noreply.github.com>
---
 .../smg_grpc_servicer/vllm/mm_processor.py    | 31 ++++++--
 grpc_servicer/tests/test_vllm_mm_sidecar.py   | 72 +++++++++++++++----
 2 files changed, 85 insertions(+), 18 deletions(-)

diff --git a/grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py b/grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py
index 92e8b198fe..0caf5bc16f 100644
--- a/grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py
+++ b/grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py
@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 import asyncio
+import dataclasses
 import inspect
 import json
 import logging
@@ -248,10 +249,30 @@ async def _fetch(self, index: int, item):
 MM_LIMIT_MODALITIES = ("image", "video", "audio")
 
 
-def resolved_mm_limits(mm_config) -> dict[str, int]:
-    """Per-modality prompt limits as vLLM resolves them (999 when unset), so
-    equivalent spellings of --limit-mm-per-prompt match."""
-    return {modality: mm_config.get_limit_per_prompt(modality) for modality in MM_LIMIT_MODALITIES}
+def _limit_options(options) -> dict[str, Any]:
+    """A validated per-modality limit as a comparable dict, unset fields dropped."""
+    if dataclasses.is_dataclass(options) and not isinstance(options, type):
+        data = dataclasses.asdict(options)
+    elif isinstance(options, Mapping):
+        data = dict(options)
+    else:  # legacy count-only form
+        data = {"count": options}
+    return {key: value for key, value in data.items() if value is not None}
+
+
+def resolved_mm_limits(mm_config) -> dict[str, dict[str, Any]]:
+    """Per-modality prompt limits as vLLM validated them: the effective count
+    (get_limit_per_prompt, 999 when unset) plus any sibling options, over the
+    standard modalities and every configured key, so equivalent spellings of
+    --limit-mm-per-prompt match and a real skew still fails closed."""
+    configured = getattr(mm_config, "limit_per_prompt", None) or {}
+    limits: dict[str, dict[str, Any]] = {}
+    for modality in sorted({*MM_LIMIT_MODALITIES, *configured}):
+        options = configured.get(modality)
+        extra = _limit_options(options) if options is not None else {}
+        extra.pop("count", None)
+        limits[modality] = {"count": mm_config.get_limit_per_prompt(modality), **extra}
+    return limits
 
 
 def fingerprint_from_model_config(model_config) -> Fingerprint:
@@ -269,7 +290,7 @@ def fingerprint_from_model_config(model_config) -> Fingerprint:
         mm_processor_kwargs=json.dumps(
             mm_config.mm_processor_kwargs or {}, sort_keys=True, default=str
         ),
-        limit_per_prompt=json.dumps(resolved_mm_limits(mm_config), sort_keys=True),
+        limit_per_prompt=json.dumps(resolved_mm_limits(mm_config), sort_keys=True, default=str),
     )
 
 
diff --git a/grpc_servicer/tests/test_vllm_mm_sidecar.py b/grpc_servicer/tests/test_vllm_mm_sidecar.py
index 3f577a1560..14693b7dbd 100644
--- a/grpc_servicer/tests/test_vllm_mm_sidecar.py
+++ b/grpc_servicer/tests/test_vllm_mm_sidecar.py
@@ -7,6 +7,7 @@
 import sys
 import time
 import types
+from dataclasses import dataclass
 
 import pytest
 
@@ -66,6 +67,23 @@ def fingerprint():
     )
 
 
+# Shaped like vLLM's validated limit_per_prompt values (pydantic dataclasses).
+@dataclass
+class _BaseOptions:
+    count: int = 999
+
+
+@dataclass
+class _ImageOptions(_BaseOptions):
+    width: int | None = None
+    height: int | None = None
+
+
+@dataclass
+class _VideoOptions(_BaseOptions):
+    num_frames: int | None = None
+
+
 class _MmConfig:
     def __init__(self, limit_per_prompt):
         self.limit_per_prompt = limit_per_prompt
@@ -73,9 +91,8 @@ def __init__(self, limit_per_prompt):
         self.mm_processor_kwargs = None
 
     def get_limit_per_prompt(self, modality):
-        # Mirrors vLLM: 999 when unset, otherwise the validated options' count.
-        value = self.limit_per_prompt.get(modality, 999)
-        return value["count"] if isinstance(value, dict) else value
+        options = self.limit_per_prompt.get(modality)
+        return 999 if options is None else options.count
 
 
 class _ModelConfig:
@@ -98,24 +115,53 @@ def _fake_vllm(monkeypatch):
 class TestFingerprintDerivation:
     def test_worker_and_sidecar_derive_the_same_fingerprint(self, monkeypatch):
         _fake_vllm(monkeypatch)
-        engine = types.SimpleNamespace(model_config=_ModelConfig({"image": 8}))
-        vllm_config = types.SimpleNamespace(model_config=_ModelConfig({"image": 8}))
+        limits = {"image": _ImageOptions(count=8)}
+        engine = types.SimpleNamespace(model_config=_ModelConfig(limits))
+        vllm_config = types.SimpleNamespace(model_config=_ModelConfig(limits))
         worker = mm_processor.engine_fingerprint(engine)
         assert worker == mm_sidecar.config_fingerprint(vllm_config)
-        assert worker.limit_per_prompt == '{"audio": 999, "image": 8, "video": 999}'
+        assert worker.limit_per_prompt == (
+            '{"audio": {"count": 999}, "image": {"count": 8}, "video": {"count": 999}}'
+        )
 
     def test_equivalent_limit_spellings_share_a_namespace(self, monkeypatch):
         _fake_vllm(monkeypatch)
-        explicit = mm_processor.fingerprint_from_model_config(_ModelConfig({"image": 999}))
+        explicit = mm_processor.fingerprint_from_model_config(
+            _ModelConfig({"image": _ImageOptions(count=999)})
+        )
         implicit = mm_processor.fingerprint_from_model_config(_ModelConfig({}))
-        nested = mm_processor.fingerprint_from_model_config(
-            _ModelConfig({"image": {"count": 999, "max_pixels": 4}})
+        assert explicit == implicit
+        skewed = mm_processor.fingerprint_from_model_config(
+            _ModelConfig({"image": _ImageOptions(count=8)})
         )
-        assert explicit == implicit == nested
-        assert (
-            explicit.namespace()
-            != mm_processor.fingerprint_from_model_config(_ModelConfig({"image": 8})).namespace()
+        assert explicit.namespace() != skewed.namespace()
+
+    def test_sibling_options_and_extra_modalities_stay_in_the_fingerprint(self):
+        limits = {
+            "video": _VideoOptions(count=2, num_frames=16),
+            "image": _ImageOptions(count=8, width=4),
+            "point_cloud": _BaseOptions(count=1),
+        }
+        assert mm_processor.resolved_mm_limits(_MmConfig(limits)) == {
+            "audio": {"count": 999},
+            "image": {"count": 8, "width": 4},
+            "point_cloud": {"count": 1},
+            "video": {"count": 2, "num_frames": 16},
+        }
+
+    def test_resolved_limits_match_vllm(self):
+        pytest.importorskip("vllm")
+        from vllm.config import MultiModalConfig
+
+        unset = mm_processor.resolved_mm_limits(MultiModalConfig())
+        assert unset == {"audio": {"count": 999}, "image": {"count": 999}, "video": {"count": 999}}
+        explicit = MultiModalConfig(limit_per_prompt={"image": 999})
+        assert mm_processor.resolved_mm_limits(explicit) == unset
+        resolved = mm_processor.resolved_mm_limits(
+            MultiModalConfig(limit_per_prompt={"image": 8, "video": {"count": 2, "num_frames": 16}})
         )
+        assert resolved["image"] == {"count": 8}
+        assert resolved["video"] == {"count": 2, "num_frames": 16}
 
 
 def sidecar(client):

From 9ca1aed2953ec908c1e485a07899b03cb0dd677d Mon Sep 17 00:00:00 2001
From: Chang Su <8605658+CatherineSue@users.noreply.github.com>
Date: Mon, 14 Sep 2026 16:16:14 -0700
Subject: [PATCH 06/11] fix(vllm-grpc): normalize spelled-out default media
 limits, hash unknown option shapes

resolved_mm_limits records the unset default once and only the configured
modalities whose effective count or sibling options differ from it, so an
explicitly spelled default matches an omitted flag for every modality.
Unknown option shapes are hashed by their attributes (or rejected), never
reduced to a count; a pydantic-dataclass test runs the validated-value path
without vLLM.

Signed-off-by: Chang Su <8605658+CatherineSue@users.noreply.github.com>
---
 .../smg_grpc_servicer/vllm/mm_processor.py    | 39 +++++++----
 grpc_servicer/tests/test_vllm_mm_sidecar.py   | 67 +++++++++++++++----
 2 files changed, 81 insertions(+), 25 deletions(-)

diff --git a/grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py b/grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py
index 0caf5bc16f..2549319fb2 100644
--- a/grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py
+++ b/grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py
@@ -246,32 +246,47 @@ async def _fetch(self, index: int, item):
         raise ValueError(f"unsupported media modality {item.modality!r}")
 
 
-MM_LIMIT_MODALITIES = ("image", "video", "audio")
+# A modality name vLLM never configures: its resolved count is the unset default.
+_UNSET_MODALITY = "__unset__"
 
 
 def _limit_options(options) -> dict[str, Any]:
-    """A validated per-modality limit as a comparable dict, unset fields dropped."""
+    """A validated per-modality limit as a comparable dict, unset fields dropped.
+
+    Unknown shapes are hashed by their attributes, never reduced to a count:
+    the two ends must fail closed rather than falsely match.
+    """
     if dataclasses.is_dataclass(options) and not isinstance(options, type):
         data = dataclasses.asdict(options)
     elif isinstance(options, Mapping):
         data = dict(options)
-    else:  # legacy count-only form
+    elif isinstance(options, int):  # legacy count-only form
         data = {"count": options}
+    elif hasattr(options, "model_dump"):  # a pydantic model
+        data = dict(options.model_dump())
+    elif hasattr(options, "__dict__"):
+        logger.warning("limit_per_prompt value of type %s: hashing its attributes", type(options))
+        data = dict(vars(options))
+    else:
+        raise TypeError(f"unsupported limit_per_prompt value: {options!r}")
     return {key: value for key, value in data.items() if value is not None}
 
 
 def resolved_mm_limits(mm_config) -> dict[str, dict[str, Any]]:
-    """Per-modality prompt limits as vLLM validated them: the effective count
-    (get_limit_per_prompt, 999 when unset) plus any sibling options, over the
-    standard modalities and every configured key, so equivalent spellings of
-    --limit-mm-per-prompt match and a real skew still fails closed."""
+    """Per-modality prompt limits as vLLM validated them: the unset default under
+    "*", then each configured modality's effective count (get_limit_per_prompt)
+    plus its sibling options, omitting entries that only spell out the default,
+    so equivalent --limit-mm-per-prompt spellings match and a real skew fails
+    closed."""
+    unset = {"count": mm_config.get_limit_per_prompt(_UNSET_MODALITY)}
+    limits: dict[str, dict[str, Any]] = {"*": unset}
     configured = getattr(mm_config, "limit_per_prompt", None) or {}
-    limits: dict[str, dict[str, Any]] = {}
-    for modality in sorted({*MM_LIMIT_MODALITIES, *configured}):
-        options = configured.get(modality)
-        extra = _limit_options(options) if options is not None else {}
+    for modality in sorted(configured):
+        extra = _limit_options(configured[modality])
         extra.pop("count", None)
-        limits[modality] = {"count": mm_config.get_limit_per_prompt(modality), **extra}
+        entry = {"count": mm_config.get_limit_per_prompt(modality), **extra}
+        if entry != unset:
+            limits[modality] = entry
     return limits
 
 
diff --git a/grpc_servicer/tests/test_vllm_mm_sidecar.py b/grpc_servicer/tests/test_vllm_mm_sidecar.py
index 14693b7dbd..dd4d1134b6 100644
--- a/grpc_servicer/tests/test_vllm_mm_sidecar.py
+++ b/grpc_servicer/tests/test_vllm_mm_sidecar.py
@@ -120,21 +120,24 @@ def test_worker_and_sidecar_derive_the_same_fingerprint(self, monkeypatch):
         vllm_config = types.SimpleNamespace(model_config=_ModelConfig(limits))
         worker = mm_processor.engine_fingerprint(engine)
         assert worker == mm_sidecar.config_fingerprint(vllm_config)
-        assert worker.limit_per_prompt == (
-            '{"audio": {"count": 999}, "image": {"count": 8}, "video": {"count": 999}}'
-        )
+        assert worker.limit_per_prompt == '{"*": {"count": 999}, "image": {"count": 8}}'
 
     def test_equivalent_limit_spellings_share_a_namespace(self, monkeypatch):
         _fake_vllm(monkeypatch)
-        explicit = mm_processor.fingerprint_from_model_config(
-            _ModelConfig({"image": _ImageOptions(count=999)})
-        )
         implicit = mm_processor.fingerprint_from_model_config(_ModelConfig({}))
-        assert explicit == implicit
-        skewed = mm_processor.fingerprint_from_model_config(
-            _ModelConfig({"image": _ImageOptions(count=8)})
-        )
-        assert explicit.namespace() != skewed.namespace()
+        for spelled_out in (
+            {"image": _ImageOptions(count=999)},
+            {"point_cloud": _BaseOptions(count=999)},
+            {"video": _VideoOptions(count=999), "audio": _BaseOptions()},
+        ):
+            assert mm_processor.fingerprint_from_model_config(_ModelConfig(spelled_out)) == implicit
+        for skewed in (
+            {"image": _ImageOptions(count=8)},
+            {"image": _ImageOptions(count=999, width=4)},
+            {"point_cloud": _BaseOptions(count=2)},
+        ):
+            fp = mm_processor.fingerprint_from_model_config(_ModelConfig(skewed))
+            assert fp.namespace() != implicit.namespace(), skewed
 
     def test_sibling_options_and_extra_modalities_stay_in_the_fingerprint(self):
         limits = {
@@ -143,18 +146,56 @@ def test_sibling_options_and_extra_modalities_stay_in_the_fingerprint(self):
             "point_cloud": _BaseOptions(count=1),
         }
         assert mm_processor.resolved_mm_limits(_MmConfig(limits)) == {
-            "audio": {"count": 999},
+            "*": {"count": 999},
             "image": {"count": 8, "width": 4},
             "point_cloud": {"count": 1},
             "video": {"count": 2, "num_frames": 16},
         }
 
+    def test_unknown_limit_shapes_are_loud_not_lossy(self):
+        class _Model:
+            def __init__(self):
+                self.count = 2
+                self.num_frames = 16
+
+            def model_dump(self):
+                return {"count": self.count, "num_frames": self.num_frames}
+
+        assert mm_processor._limit_options(_Model()) == {"count": 2, "num_frames": 16}
+        plain = types.SimpleNamespace(count=2, num_frames=16, width=None)
+        assert mm_processor._limit_options(plain) == {"count": 2, "num_frames": 16}
+        assert mm_processor._limit_options(7) == {"count": 7}
+        assert mm_processor._limit_options({"count": 3, "length": 9}) == {"count": 3, "length": 9}
+        with pytest.raises(TypeError, match="unsupported limit_per_prompt value"):
+            mm_processor._limit_options(object())
+
+    def test_pydantic_dataclass_options_resolve_like_vllm(self):
+        # vLLM's *DummyOptions are pydantic dataclasses; mirror the decorator so the
+        # is_dataclass/asdict path runs without vLLM installed.
+        pydantic = pytest.importorskip("pydantic")
+        from pydantic.dataclasses import dataclass as pydantic_dataclass
+
+        @pydantic_dataclass
+        class _Base:
+            count: int = pydantic.Field(999, ge=0)
+
+        @pydantic_dataclass(config=pydantic.ConfigDict(extra="forbid"))
+        class _Video(_Base):
+            num_frames: int | None = pydantic.Field(None, gt=0)
+            width: int | None = pydantic.Field(None, gt=0)
+
+        cfg = _MmConfig({"video": _Video(count=2, num_frames=16), "image": _Base(count=999)})
+        assert mm_processor.resolved_mm_limits(cfg) == {
+            "*": {"count": 999},
+            "video": {"count": 2, "num_frames": 16},
+        }
+
     def test_resolved_limits_match_vllm(self):
         pytest.importorskip("vllm")
         from vllm.config import MultiModalConfig
 
         unset = mm_processor.resolved_mm_limits(MultiModalConfig())
-        assert unset == {"audio": {"count": 999}, "image": {"count": 999}, "video": {"count": 999}}
+        assert unset == {"*": {"count": 999}}
         explicit = MultiModalConfig(limit_per_prompt={"image": 999})
         assert mm_processor.resolved_mm_limits(explicit) == unset
         resolved = mm_processor.resolved_mm_limits(

From 05e3a7807abf2a8af39270eff908fdc444aaf3b6 Mon Sep 17 00:00:00 2001
From: Chang Su <8605658+CatherineSue@users.noreply.github.com>
Date: Mon, 14 Sep 2026 16:23:48 -0700
Subject: [PATCH 07/11] test(vllm-grpc): pin the uniform unset media limit the
 fingerprint relies on

The vLLM-gated test now asserts that every standard modality and an
unknown key resolve to the same unset default, and that spelling that
default out for any modality matches omitting it.

Signed-off-by: Chang Su <8605658+CatherineSue@users.noreply.github.com>
---
 grpc_servicer/tests/test_vllm_mm_sidecar.py | 17 ++++++++++++++---
 1 file changed, 14 insertions(+), 3 deletions(-)

diff --git a/grpc_servicer/tests/test_vllm_mm_sidecar.py b/grpc_servicer/tests/test_vllm_mm_sidecar.py
index dd4d1134b6..978d1120a1 100644
--- a/grpc_servicer/tests/test_vllm_mm_sidecar.py
+++ b/grpc_servicer/tests/test_vllm_mm_sidecar.py
@@ -194,10 +194,21 @@ def test_resolved_limits_match_vllm(self):
         pytest.importorskip("vllm")
         from vllm.config import MultiModalConfig
 
-        unset = mm_processor.resolved_mm_limits(MultiModalConfig())
+        # The derivation drops entries equal to one probed unset default, which
+        # is only right while that default is uniform across the options classes
+        # and an unknown key resolves to it instead of raising.
+        cfg = MultiModalConfig()
+        modalities = ("image", "video", "audio")
+        assert {m: cfg.get_limit_per_prompt(m) for m in modalities} == dict.fromkeys(
+            modalities, 999
+        )
+        assert cfg.get_limit_per_prompt(mm_processor._UNSET_MODALITY) == 999
+        unset = mm_processor.resolved_mm_limits(cfg)
         assert unset == {"*": {"count": 999}}
-        explicit = MultiModalConfig(limit_per_prompt={"image": 999})
-        assert mm_processor.resolved_mm_limits(explicit) == unset
+        for modality in modalities:
+            for spelled_out in (999, {}):
+                explicit = MultiModalConfig(limit_per_prompt={modality: spelled_out})
+                assert mm_processor.resolved_mm_limits(explicit) == unset, (modality, spelled_out)
         resolved = mm_processor.resolved_mm_limits(
             MultiModalConfig(limit_per_prompt={"image": 8, "video": {"count": 2, "num_frames": 16}})
         )

From 6989304c8df830f800147b1c94f6038a3023307c Mon Sep 17 00:00:00 2001
From: Chang Su <8605658+CatherineSue@users.noreply.github.com>
Date: Mon, 14 Sep 2026 16:29:26 -0700
Subject: [PATCH 08/11] ci: run the servicer's vLLM-gated unit tests in the
 vLLM e2e lanes

The CPU unit-test job has no vLLM, so every importorskip("vllm") test in
grpc_servicer/tests was a permanent skip; the vLLM e2e job installs vLLM
and the servicer from source, so it runs the suite once before the e2e
cases.

Signed-off-by: Chang Su <8605658+CatherineSue@users.noreply.github.com>
---
 .github/workflows/e2e-gpu-job.yml | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/.github/workflows/e2e-gpu-job.yml b/.github/workflows/e2e-gpu-job.yml
index 0e003efdd8..0250fcaba7 100644
--- a/.github/workflows/e2e-gpu-job.yml
+++ b/.github/workflows/e2e-gpu-job.yml
@@ -174,6 +174,12 @@ jobs:
           pip install wheel/*.whl
           bash scripts/ci_install_e2e_deps.sh ${{ inputs.extra_deps }}
 
+      # The servicer's vLLM-gated unit tests (importorskip("vllm")) only meet a
+      # real vLLM here; the CPU unit-test job skips them.
+      - name: Run vLLM-gated servicer unit tests
+        if: inputs.engine == 'vllm'
+        run: pytest -q grpc_servicer/tests -p no:cacheprovider
+
       - name: Setup Oracle
         if: inputs.setup_agentic_deps
         run: |

From b8b8a5fb983fcb996b6bd62a754bbb1457bb1f29 Mon Sep 17 00:00:00 2001
From: Chang Su <8605658+CatherineSue@users.noreply.github.com>
Date: Mon, 14 Sep 2026 16:37:13 -0700
Subject: [PATCH 09/11] ci: run the servicer suite in one vLLM lane, with an
 import canary

The step is opted into by the vLLM row of e2e-1gpu-chat instead of every
vLLM lane, checks the vllm and msgspec import gates first so the gated
tests cannot silently skip, and prints skip reasons.

Signed-off-by: Chang Su <8605658+CatherineSue@users.noreply.github.com>
---
 .github/workflows/e2e-gpu-job.yml  | 13 ++++++++++---
 .github/workflows/pr-test-rust.yml |  1 +
 2 files changed, 11 insertions(+), 3 deletions(-)

diff --git a/.github/workflows/e2e-gpu-job.yml b/.github/workflows/e2e-gpu-job.yml
index 0250fcaba7..2f3fdae268 100644
--- a/.github/workflows/e2e-gpu-job.yml
+++ b/.github/workflows/e2e-gpu-job.yml
@@ -89,6 +89,11 @@ on:
         type: string
         default: ""
         description: "Prebuilt TokenSpeed carrier image to extract onto the runner (resolved by detect-changes); empty = build from source"
+      servicer_unit_tests:
+        required: false
+        type: boolean
+        default: false
+        description: "Also run grpc_servicer/tests here, where the importorskip('vllm') cases meet a real vLLM; set on one vLLM lane"
 
 jobs:
   run:
@@ -175,10 +180,12 @@ jobs:
           bash scripts/ci_install_e2e_deps.sh ${{ inputs.extra_deps }}
 
       # The servicer's vLLM-gated unit tests (importorskip("vllm")) only meet a
-      # real vLLM here; the CPU unit-test job skips them.
+      # real vLLM here; the CPU unit-test job skips them. One lane opts in.
       - name: Run vLLM-gated servicer unit tests
-        if: inputs.engine == 'vllm'
-        run: pytest -q grpc_servicer/tests -p no:cacheprovider
+        if: inputs.servicer_unit_tests
+        run: |
+          python3 -c "import vllm, msgspec"  # the gates these tests importorskip on
+          pytest -q grpc_servicer/tests -p no:cacheprovider -rs
 
       - name: Setup Oracle
         if: inputs.setup_agentic_deps
diff --git a/.github/workflows/pr-test-rust.yml b/.github/workflows/pr-test-rust.yml
index 1999f27beb..482e9f2aed 100644
--- a/.github/workflows/pr-test-rust.yml
+++ b/.github/workflows/pr-test-rust.yml
@@ -673,6 +673,7 @@ jobs:
       test_dirs: ${{ matrix.test_dirs || 'e2e_test/chat_completions' }}
       min_selected: ${{ matrix.min_selected }}
       tokenspeed_prebuilt_image: ${{ matrix.engine == 'tokenspeed' && needs.detect-changes.outputs.tokenspeed-image || '' }}
+      servicer_unit_tests: ${{ matrix.engine == 'vllm' }}
     secrets: inherit
 
   # Worker-side media processing: the vLLM gRPC worker fetches and

From b9ba83a96cb0323faba59a3004157bd5f4df7ca8 Mon Sep 17 00:00:00 2001
From: Chang Su <8605658+CatherineSue@users.noreply.github.com>
Date: Mon, 14 Sep 2026 16:44:46 -0700
Subject: [PATCH 10/11] ci: install msgspec for the servicer unit tests and
 show skip reasons

The CPU job never had msgspec, so the sidecar protocol and fake-redis
test modules skipped silently; the vLLM lane keeps only the
importorskip("vllm") cases to itself.

Signed-off-by: Chang Su <8605658+CatherineSue@users.noreply.github.com>
---
 .github/workflows/pr-test-rust.yml | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/.github/workflows/pr-test-rust.yml b/.github/workflows/pr-test-rust.yml
index 482e9f2aed..ec886cd2fe 100644
--- a/.github/workflows/pr-test-rust.yml
+++ b/.github/workflows/pr-test-rust.yml
@@ -257,8 +257,10 @@ jobs:
           rm -f crates/grpc_client/python/smg_grpc_proto/proto
           mkdir -p crates/grpc_client/python/smg_grpc_proto/proto
           cp crates/grpc_client/proto/*.proto crates/grpc_client/python/smg_grpc_proto/proto/
-          pip install ./crates/grpc_client/python
-          pytest -q grpc_servicer/tests
+          # msgspec gates the sidecar protocol and fake-redis tests; only the
+          # importorskip("vllm") cases stay skipped here (see the vLLM e2e lane).
+          pip install ./crates/grpc_client/python msgspec
+          pytest -q grpc_servicer/tests -rs
 
       # CPU-only unit tests for the e2e harness itself (connection-mode parsing,
       # ZMQ/EPD command builders, ZMQ-lane collection filter). --noconftest keeps

From 95bb9099c8cfb16e8f1ca1a529e8ccafb69add43 Mon Sep 17 00:00:00 2001
From: Chang Su <8605658+CatherineSue@users.noreply.github.com>
Date: Mon, 14 Sep 2026 16:50:16 -0700
Subject: [PATCH 11/11] ci: state which servicer test gates the CPU job still
 skips, install pydantic

pyzmq would un-skip test_tokenspeed_dp_rank_pin.py, which then imports the
TokenSpeed servicer and fails collection without torch, so the zmq-gated
modules stay skipped and the comment says so; pydantic lets this PR's
limit-options test run here.

Signed-off-by: Chang Su <8605658+CatherineSue@users.noreply.github.com>
---
 .github/workflows/pr-test-rust.yml | 8 +++++---
 1 file changed, 5 insertions(+), 3 deletions(-)

diff --git a/.github/workflows/pr-test-rust.yml b/.github/workflows/pr-test-rust.yml
index ec886cd2fe..cfcad09ed1 100644
--- a/.github/workflows/pr-test-rust.yml
+++ b/.github/workflows/pr-test-rust.yml
@@ -257,9 +257,11 @@ jobs:
           rm -f crates/grpc_client/python/smg_grpc_proto/proto
           mkdir -p crates/grpc_client/python/smg_grpc_proto/proto
           cp crates/grpc_client/proto/*.proto crates/grpc_client/python/smg_grpc_proto/proto/
-          # msgspec gates the sidecar protocol and fake-redis tests; only the
-          # importorskip("vllm") cases stay skipped here (see the vLLM e2e lane).
-          pip install ./crates/grpc_client/python msgspec
+          # msgspec and pydantic gate the sidecar-protocol, fake-redis and
+          # limit-options tests. Modules gated on zmq or vllm still skip here
+          # (-rs lists them): the zmq ones import engine servicers that need
+          # torch, and the vllm cases run on the vLLM e2e lane.
+          pip install ./crates/grpc_client/python msgspec pydantic
           pytest -q grpc_servicer/tests -rs
 
       # CPU-only unit tests for the e2e harness itself (connection-mode parsing,