feat(servicer): take the media-processing settings from launcher flags

crates/grpc_client/proto/vllm_engine.proto

@@ -420,6 +420,11 @@ message GetServerInfoResponse {
   // or "http,https,data,file" (file only with --allowed-local-media-path).
   // Flattened into the `mm_media_ref_schemes` worker label.
   string mm_media_ref_schemes = 17;
+
+  // Where the worker's mm_processor setting came from: "flag" (the launcher's
+  // --mm-processor), "env" (SMG_VLLM_MM_PROCESSOR) or "default"; "" when
+  // mm_processor is empty. Flattened into the `mm_processor_source` label.
+  string mm_processor_source = 18;
 }
 
 // =====================

crates/grpc_client/python/pyproject.toml

@@ -8,7 +8,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "smg-grpc-proto"
-version = "0.4.20"
+version = "0.4.21"
 description = "SMG gRPC proto definitions for vLLM, TRT-LLM, MLX, TokenSpeed, and SGLang"
 requires-python = ">=3.10"
 dependencies = [

grpc_servicer/README.md

@@ -43,22 +43,33 @@ pixel tensors. A vLLM gRPC worker can instead accept media references (URLs)
 and run vLLM's own multimodal processor:
 
 ```bash
-SMG_VLLM_MM_PROCESSOR=inprocess vllm serve Qwen/Qwen3-VL-8B-Instruct --grpc \
+vllm serve Qwen/Qwen3-VL-8B-Instruct --grpc --mm-processor inprocess \
     --allowed-media-domains example.com
 ```
 
+The `--mm-*` flags come from a vLLM launcher that knows them (it hands an
+`MmSettings` to `VllmEngineServicer`); an older launcher, or a flag left out,
+falls back to the matching `SMG_VLLM_MM_*` variable, which logs a deprecation
+line and goes away in the next minor release. The startup line
+`VllmEngineServicer initialized (mm_processor=inprocess, source=flag)` and the
+`mm_processor_source` label name where the value came from.
+
 The worker then advertises `mm_processor=inprocess` and `mm_media_ref_schemes`
 through `GetServerInfo`; a router with media-reference support forwards
 `media_refs` only to workers that advertise, and a router without it ignores the
 labels and keeps sending preprocessed tensors. vLLM's `--allowed-media-domains`,
 `--allowed-local-media-path`, `--media-io-kwargs`, `--limit-mm-per-prompt` and
 `VLLM_*_FETCH_TIMEOUT` govern fetching on the worker; without
 `--allowed-media-domains` the worker fetches from any host the router forwards.
-Related knobs: `SMG_VLLM_MM_MAX_INFLIGHT` (default 64) bounds concurrent media
-jobs; `SMG_VLLM_MM_MAX_ITEMS` (default 16) caps references per request;
-`SMG_VLLM_MM_MAX_ITEM_BYTES` (default 32 MiB) caps inline `data:` payloads;
-`SMG_VLLM_MM_MAX_VIDEO_FRAMES` (default 0, meaning vLLM's own `--media-io-kwargs`
-decide) caps the frames a video is sampled to, so long clips stay bounded.
+Related knobs: `--mm-max-inflight` (`SMG_VLLM_MM_MAX_INFLIGHT`, default 64)
+bounds concurrent media jobs; `--mm-max-items` (`SMG_VLLM_MM_MAX_ITEMS`, unset
+by default) overrides the model's per-modality reference limits;
+`--mm-max-item-bytes` (`SMG_VLLM_MM_MAX_ITEM_BYTES`, default 32 MiB) caps inline
+`data:` payloads; `SMG_VLLM_MM_MAX_VIDEO_FRAMES` (env only for now; default 0,
+meaning vLLM's own `--media-io-kwargs` decide) caps the frames a video is
+sampled to, so long clips stay bounded. All eight flag-backed settings are
+validated when the servicer starts, whatever the processor mode, so a stale
+unreadable value fails loudly.
 
 On the router side, `--mm-processing` selects `auto` (default: forward when
 the model's spec opts in and every registered worker of the model advertises
@@ -84,8 +95,8 @@ sidecar next to a private Redis and point the worker at it
 ```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
+vllm serve Qwen/Qwen3-VL-8B-Instruct --grpc --mm-processor redis \
+    --mm-redis-url redis://127.0.0.1:6379/0
 ```
 
 The sidecar and the worker must agree on model, vLLM version, dtype, video
@@ -96,11 +107,14 @@ 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). On the sidecar,
-`SMG_VLLM_MM_MAX_RESULT_BYTES` (default 512 MiB, lowered to Redis's
-`proto-max-bulk-len` when that is smaller) caps an encoded result and
+id and expire after 120 s. Knobs: `--mm-sidecar-timeout-ms`
+(`SMG_VLLM_MM_SIDECAR_TIMEOUT_MS`, 30000), `--mm-sidecar-max-queue`
+(`SMG_VLLM_MM_SIDECAR_MAX_QUEUE`, 256, fail fast when the queue is deeper),
+`--mm-sidecar-namespace` (`SMG_VLLM_MM_SIDECAR_NAMESPACE`, override the derived
+namespace). The sidecar resolves `--redis-url`, `--namespace` and
+`--mm-sidecar-timeout-ms` the same way, so the two processes cannot disagree.
+On the sidecar, `SMG_VLLM_MM_MAX_RESULT_BYTES` (default 512 MiB, lowered to
+Redis's `proto-max-bulk-len` when that is smaller) caps an encoded result and
 `SMG_VLLM_MM_MAX_VIDEO_FRAMES` caps video sampling as above. A result over the
 cap is answered as a 400 `media_too_large` instead of being pushed, and a result
 Redis refuses is reported to the worker at once; a sidecar timeout is not

grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py

@@ -16,7 +16,7 @@
 import os
 import time
 import uuid
-from collections.abc import Awaitable, Mapping, Sequence
+from collections.abc import Awaitable, Iterable, Mapping, Sequence
 from typing import Any
 
 from smg_grpc_servicer.mm_sidecar_protocol import (
@@ -630,35 +630,165 @@ def _redis_client(redis_url: str):
     )
 
 
-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)
+SOURCE_FLAG = "flag"
+SOURCE_ENV = "env"
+SOURCE_DEFAULT = "default"
+
+# Setting name -> (launcher flag, env var, default). `None` defaults are
+# settings that may legitimately stay unset.
+_MM_SETTING_SPECS: dict[str, tuple[str, str, Any]] = {
+    "processor": ("--mm-processor", ENV_PROCESSOR, MODE_OFF),
+    "max_inflight": ("--mm-max-inflight", ENV_MAX_INFLIGHT, DEFAULT_MAX_INFLIGHT),
+    "max_item_bytes": ("--mm-max-item-bytes", ENV_MAX_ITEM_BYTES, DEFAULT_MAX_ITEM_BYTES),
+    "max_items": ("--mm-max-items", ENV_MAX_ITEMS, None),
+    "redis_url": ("--mm-redis-url", ENV_REDIS_URL, DEFAULT_REDIS_URL),
+    "sidecar_timeout_ms": ("--mm-sidecar-timeout-ms", ENV_TIMEOUT_MS, DEFAULT_TIMEOUT_MS),
+    "sidecar_max_queue": ("--mm-sidecar-max-queue", ENV_MAX_QUEUE, DEFAULT_MAX_QUEUE),
+    "sidecar_namespace": ("--mm-sidecar-namespace", ENV_NAMESPACE, None),
+}
+_MM_INT_SETTINGS = frozenset(
+    {"max_inflight", "max_item_bytes", "max_items", "sidecar_timeout_ms", "sidecar_max_queue"}
+)
+
+
+@dataclasses.dataclass(frozen=True)
+class MmSettings:
+    """The engine-side media settings, as requested or as resolved.
+
+    A field left `None` was not asked for; `resolve` fills it as flag > env >
+    default and records each value's source. `sources` is empty until then.
+    """
+
+    processor: str | None = None
+    max_inflight: int | None = None
+    max_item_bytes: int | None = None
+    max_items: int | None = None
+    redis_url: str | None = None
+    sidecar_timeout_ms: int | None = None
+    sidecar_max_queue: int | None = None
+    sidecar_namespace: str | None = None
+    sources: Mapping[str, str] = dataclasses.field(default_factory=dict, compare=False)
+
+    @classmethod
+    def from_args(cls, args) -> MmSettings:
+        """The `--mm-*` values of a launcher namespace; absent flags ask for nothing."""
+        return cls(**{name: getattr(args, f"mm_{name}", None) for name in _MM_SETTING_SPECS})
+
+    @property
+    def resolved(self) -> bool:
+        """Resolved at least once; `sources` names which settings, so a
+        subset-resolved object is not a full one (see `build_mm_processor`)."""
+        return bool(self.sources)
+
+    @property
+    def source(self) -> str:
+        """Where the processor mode came from."""
+        return self.sources.get("processor", SOURCE_DEFAULT)
+
+    def resolve(
+        self,
+        env: Mapping[str, str] = os.environ,
+        *,
+        only: Iterable[str] | None = None,
+        flags: Mapping[str, str] | None = None,
+    ) -> MmSettings:
+        """Flag > env > default; already resolved settings come back unchanged.
+
+        `only` limits resolution to the named settings (the rest stay unset
+        and unvalidated), for a process that uses a subset; `flags` renames
+        the flag a deprecation line points at, for a parser with its own.
+        """
+        if self.resolved:
+            return self
+        values: dict[str, Any] = {}
+        sources: dict[str, str] = {}
+        wanted = set(_MM_SETTING_SPECS if only is None else only)
+        if unknown := wanted - _MM_SETTING_SPECS.keys():
+            raise ValueError(f"unknown mm settings: {sorted(unknown)}")
+        for name, (flag, env_name, default) in _MM_SETTING_SPECS.items():
+            if name not in wanted:
+                continue
+            flag = (flags or {}).get(name, flag)
+            requested = getattr(self, name)
+            if requested is not None:
+                values[name] = _validate_flag(name, flag, requested)
+                sources[name] = SOURCE_FLAG
+                continue
+            from_env = _read_env(name, env, env_name)
+            if from_env is not None:
+                logger.warning(
+                    "%s is deprecated in favour of %s; env support ends in the next minor release",
+                    env_name,
+                    flag,
+                )
+                values[name] = from_env
+                sources[name] = SOURCE_ENV
+                continue
+            values[name] = default
+            sources[name] = SOURCE_DEFAULT
+        return MmSettings(**values, sources=sources)
+
+
+def _validate_flag(name: str, flag: str, value: Any) -> Any:
+    if name == "processor":
+        mode = str(value).strip().lower()
+        if mode not in VALID_MODES:
+            raise ValueError(f"{flag}={value!r} is not one of {'|'.join(VALID_MODES)}")
+        return mode
+    if name in _MM_INT_SETTINGS:
+        if int(value) <= 0:
+            raise ValueError(f"{flag}={value} must be positive")
+        return int(value)
+    return value
+
+
+def _read_env(name: str, env: Mapping[str, str], env_name: str) -> Any:
+    """The env's value for `name`, validated as before; `None` when unset."""
+    if name == "processor":
+        raw = env.get(env_name)
+        return resolve_mm_processor_mode(env) if raw is not None and raw.strip() else None
+    if name in _MM_INT_SETTINGS:
+        return env_int_opt(env, env_name)
+    raw = env.get(env_name)
+    return raw if raw else None
+
+
+def build_mm_processor(
+    engine, *, env: Mapping[str, str] = os.environ, settings: MmSettings | None = None
+):
+    """Construct the configured backend, or None when worker-side processing is off.
+
+    `settings` are the launcher's flags (already resolved or not); without them
+    everything comes from the env, as before.
+    """
+    resolved = (settings or MmSettings()).resolve(env)
+    if missing := _MM_SETTING_SPECS.keys() - resolved.sources.keys():
+        raise ValueError(f"mm settings resolved without {sorted(missing)}")
+    mode = resolved.processor
     if mode == MODE_OFF:
         return None
     model_config = getattr(engine, "model_config", None)
     if model_config is None or not getattr(model_config, "is_multimodal_model", False):
-        logger.warning("%s=%s ignored: the served model is not multimodal", ENV_PROCESSOR, mode)
+        logger.warning("mm_processor=%s ignored: the served model is not multimodal", mode)
         return None
-    max_item_bytes = env_int(env, ENV_MAX_ITEM_BYTES, DEFAULT_MAX_ITEM_BYTES)
-    max_inflight = env_int(env, ENV_MAX_INFLIGHT, DEFAULT_MAX_INFLIGHT)
-    max_items = env_int_opt(env, ENV_MAX_ITEMS)
+    # The frame budget is not one of the eight launcher flags yet: env only.
     max_video_frames = env_int(env, ENV_MAX_VIDEO_FRAMES, DEFAULT_MAX_VIDEO_FRAMES, minimum=0)
     if mode == MODE_INPROCESS:
         return InProcessMediaProcessor(
             engine,
-            max_item_bytes=max_item_bytes,
-            max_inflight=max_inflight,
-            max_items=max_items,
+            max_item_bytes=resolved.max_item_bytes,
+            max_inflight=resolved.max_inflight,
+            max_items=resolved.max_items,
             max_video_frames=max_video_frames,
         )
     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,
-        max_items=max_items,
+        redis_url=resolved.redis_url,
+        timeout_ms=resolved.sidecar_timeout_ms,
+        max_queue=resolved.sidecar_max_queue,
+        namespace=resolved.sidecar_namespace,
+        max_item_bytes=resolved.max_item_bytes,
+        max_inflight=resolved.max_inflight,
+        max_items=resolved.max_items,
     )