feat(vllm-grpc): Redis media-processing sidecar backend and reference sidecar

.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:
@@ -174,6 +179,14 @@ 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. One lane opts in.
+      - name: Run vLLM-gated servicer unit tests
+        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
         run: |

.github/workflows/pr-test-rust.yml

@@ -257,8 +257,12 @@ 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 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,
       # ZMQ/EPD command builders, ZMQ-lane collection filter). --noconftest keeps
@@ -673,6 +677,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

grpc_servicer/README.md

@@ -70,6 +70,29 @@ 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 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, 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
+`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

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

grpc_servicer/smg_grpc_servicer/mm_sidecar_protocol.py

@@ -0,0 +1,192 @@
+"""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
+    # 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."""
+        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 [
+            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_field(mapping: Mapping[bytes | str, bytes | str], name: str) -> str:
+    for key, value in mapping.items():
+        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()
+
+
+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)