feat(vllm-omni): accept application/json on video routes

.github/config/image/vllm-omni/ec2-amzn2023.yml

@@ -25,7 +25,7 @@ build:
   torch_cuda_arch_list: "7.5;8.0;8.6;8.9;9.0;10.0;11.0;12.0+PTX"
   install_kv_connectors: "true"
   dlc_major_version: "1"
-  dlc_minor_version: "3"
+  dlc_minor_version: "4"
   use_sccache: "true"
 
 release:

.github/config/image/vllm-omni/sagemaker-amzn2023.yml

@@ -26,7 +26,7 @@ build:
   torch_cuda_arch_list: "7.5;8.0;8.6;8.9;9.0;10.0;11.0;12.0+PTX"
   install_kv_connectors: "true"
   dlc_major_version: "1"
-  dlc_minor_version: "3"
+  dlc_minor_version: "4"
   use_sccache: "true"
 
 release:

.github/config/model-tests/vllm-omni-model-tests.yml

@@ -104,6 +104,27 @@ smoke-test:
       test_request: 'prompt=a dog running on a beach&num_frames=17&num_inference_steps=4&size=480x320&seed=42'
       validate: "binary_size_gt:1000"
 
+    # Sync route via application/json: the SageMaker routing middleware
+    # (omni_sagemaker_serve.py) converts the JSON object into a
+    # multipart/form-data body before handing it to the upstream
+    # /v1/videos/sync handler, which only accepts form data. Exercises the
+    # JSON content-type path so SageMaker callers can send JSON instead of
+    # building a multipart body by hand.
+    #
+    # sagemaker-only: the JSON->multipart conversion lives in the SageMaker
+    # routing middleware, which is not loaded on the EC2 image (plain
+    # `vllm serve`). On EC2 this JSON body reaches the Form-only upstream
+    # handler and returns 400, so the test is gated to the sagemaker config.
+    - name: "wan2.1-t2v-1.3b-sync-json"
+      s3_model: "wan2.1-t2v-1.3b.tar.gz"
+      fleet: "x86-g6exl-runner"
+      customer_type: "sagemaker"
+      extra_args: ""
+      route: "/v1/videos/sync"
+      content_type: "application/json"
+      test_request: '{"prompt": "a dog running on a beach", "num_frames": 17, "num_inference_steps": 4, "size": "480x320", "seed": 42}'
+      validate: "binary_size_gt:1000"
+
     # Wan2.1-VACE: unified video creation/editing pipeline (WanVACEPipeline,
     # added in vllm-omni #1885). Distinct from WanPipeline T2V — accepts
     # text + optional video/mask/reference image. 1.3B variant fits L40S.

.github/workflows/vllm-omni.tests-model.yml

@@ -59,7 +59,7 @@ jobs:
           uv venv --python 3.12
           source .venv/bin/activate
           uv pip install pyyaml
-          python3 scripts/ci/parse_model_config.py --config .github/config/model-tests/vllm-omni-model-tests.yml --runner-type codebuild-fleet
+          python3 scripts/ci/parse_model_config.py --config .github/config/model-tests/vllm-omni-model-tests.yml --runner-type codebuild-fleet --customer-type "${{ steps.config.outputs.customer-type }}"
 
   smoke-test:
     name: smoke-test (${{ matrix.model.name }})

.github/workflows/vllm.pr-amzn2023.yml

@@ -13,6 +13,8 @@ on:
       - "scripts/docker/common/**"
       - "scripts/docker/telemetry/**"
       - "scripts/docker/vllm/**"
+      # omni_* scripts belong to vllm_omni only
+      - "!scripts/docker/vllm/omni_*"
       - "test/efa/**"
       - "test/sanity/**"
       - "test/security/data/ecr_scan_allowlist/vllm_server/**"

.github/workflows/vllm.pr-ubuntu.yml

@@ -13,6 +13,8 @@ on:
       - "scripts/docker/common/**"
       - "scripts/docker/telemetry/**"
       - "scripts/docker/vllm/**"
+      # omni_* scripts belong to vllm_omni only
+      - "!scripts/docker/vllm/omni_*"
       - "test/efa/**"
       - "test/sanity/**"
       - "test/security/data/ecr_scan_allowlist/vllm/**"

scripts/ci/parse_model_config.py

@@ -62,7 +62,24 @@ def load_yaml(path: str) -> dict:
     return json.loads(result.stdout)
 
 
-def parse_config(config_path: str, section: str, runner_type: str) -> dict[str, str]:
+def matches_customer_type(model: dict, customer_type: str) -> bool:
+    """A model runs on a config unless it pins a different customer_type.
+
+    Models without a ``customer_type`` field run everywhere (backward
+    compatible). A model that pins e.g. ``customer_type: sagemaker`` only runs
+    when the config's customer type matches — used to gate tests for features
+    that exist on only one container variant (e.g. the SageMaker routing
+    middleware, which adds the JSON->multipart video path absent on EC2).
+    """
+    pinned = model.get("customer_type")
+    if not pinned or not customer_type:
+        return True
+    return pinned == customer_type
+
+
+def parse_config(
+    config_path: str, section: str, runner_type: str, customer_type: str = ""
+) -> dict[str, str]:
     cfg = load_yaml(config_path)
 
     s3_prefix = cfg.get("s3_prefix", "")
@@ -73,6 +90,7 @@ def parse_config(config_path: str, section: str, runner_type: str) -> dict[str,
 
     for rt in types:
         models = cfg.get(section, {}).get(rt, []) or []
+        models = [m for m in models if matches_customer_type(m, customer_type)]
         transformed = [transform_model(m, s3_prefix, fixtures_prefix) for m in models]
         key = rt if runner_type == "all" else "matrix"
         results[key] = json.dumps(transformed, separators=(",", ":"))
@@ -91,13 +109,19 @@ def main():
         default="all",
         help="Runner type: all, codebuild-fleet, or runner-scale-sets",
     )
+    parser.add_argument(
+        "--customer-type",
+        default="",
+        help="Config customer type (e.g. ec2, sagemaker). When set, drops models "
+        "that pin a different customer_type. Empty = include all models.",
+    )
     args = parser.parse_args()
 
     if not os.path.isfile(args.config):
         print(f"ERROR: Config file not found: {args.config}", file=sys.stderr)
         sys.exit(1)
 
-    results = parse_config(args.config, args.section, args.runner_type)
+    results = parse_config(args.config, args.section, args.runner_type, args.customer_type)
 
     output_file = os.environ.get("GITHUB_OUTPUT")
     if output_file:

scripts/docker/vllm/omni_sagemaker_serve.py

@@ -8,11 +8,15 @@
 If no route is specified, falls through to vLLM's built-in /invocations
 handler (chat/completion/embed).
 
-Clients targeting routes that require multipart/form-data (e.g. /v1/videos)
-must send the request with ContentType="multipart/form-data; boundary=..."
-and a pre-built multipart body. SageMaker InvokeEndpoint accepts arbitrary
-ContentType values and forwards them to the model server unchanged, so no
-in-middleware conversion is needed.
+Some routes (e.g. /v1/videos, /v1/videos/sync) are implemented upstream as
+multipart/form-data handlers. Clients may target them either with a
+pre-built multipart/form-data body, or — for convenience over SageMaker
+InvokeEndpoint — with a plain application/json object. When the resolved
+route is one of FORM_DATA_ROUTES and the request arrives as application/json,
+this middleware transparently converts the JSON object into a
+multipart/form-data body (each top-level key becomes a form field) so the
+upstream handler receives the encoding it expects. multipart/form-data bodies
+are passed through unchanged.
 
 Usage: vllm serve --omni --middleware omni_sagemaker_serve.SageMakerRouteMiddleware
 
@@ -25,13 +29,19 @@
 no-op and SageMaker /invocations will return 404 for non-default routes.
 """
 
+import json
 import logging
 import re
+import uuid
 
-from starlette.types import ASGIApp, Receive, Scope, Send
+from starlette.types import ASGIApp, Message, Receive, Scope, Send
 
 logger = logging.getLogger("omni_sagemaker")
 
+# Routes whose upstream handlers expect multipart/form-data. A JSON body
+# targeting one of these is converted to multipart before it reaches vLLM.
+FORM_DATA_ROUTES = frozenset({"/v1/videos", "/v1/videos/sync"})
+
 
 def _parse_route(headers: list[tuple[bytes, bytes]]) -> str | None:
     """Extract route=<path> from SageMaker custom attributes header."""
@@ -42,11 +52,79 @@ def _parse_route(headers: list[tuple[bytes, bytes]]) -> str | None:
     return None
 
 
+def _content_type(headers: list[tuple[bytes, bytes]]) -> str:
+    """Return the bare content-type (no parameters), lowercased."""
+    for key, value in headers:
+        if key.lower() == b"content-type":
+            return value.decode().split(";", 1)[0].strip().lower()
+    return ""
+
+
+def _field_value(value: object) -> str:
+    """Render a non-null JSON value as a multipart form field string.
+
+    Scalars map to their natural string form; bool maps to "true"/"false";
+    nested objects/arrays are re-encoded as JSON. None is handled by the
+    caller (_build_multipart drops null keys) and never reaches here.
+    """
+    if isinstance(value, str):
+        return value
+    if isinstance(value, bool):
+        return "true" if value else "false"
+    if isinstance(value, (int, float)):
+        return str(value)
+    return json.dumps(value)
+
+
+def _escape_field_name(name: str) -> str:
+    """Escape a form-field name for the Content-Disposition header.
+
+    RFC 7578 requires CR/LF and double-quote to be escaped inside name="...".
+    Without this a key containing " or a newline would corrupt the header.
+    """
+    return name.replace("\\", "\\\\").replace('"', '\\"').replace("\r", "%0D").replace("\n", "%0A")
+
+
+def _build_multipart(fields: dict, boundary: str) -> bytes:
+    # JSON null has no faithful form-data representation: a form field is
+    # always a present string. Emitting the literal "null" would make the
+    # upstream handler see a present-but-bogus value instead of an absent one,
+    # bypassing its default-on-missing logic. Dropping the key reproduces what
+    # a caller omitting the field would send, which is the closest match.
+    parts = [
+        f'--{boundary}\r\nContent-Disposition: form-data; name="{_escape_field_name(k)}"'
+        f"\r\n\r\n{_field_value(v)}\r\n"
+        for k, v in fields.items()
+        if v is not None
+    ]
+    parts.append(f"--{boundary}--\r\n")
+    return "".join(parts).encode()
+
+
+def _replay(body: bytes) -> Receive:
+    """Build a receive callable that yields `body` as a single http.request."""
+    sent = False
+
+    async def receive() -> Message:
+        nonlocal sent
+        if not sent:
+            sent = True
+            return {"type": "http.request", "body": body, "more_body": False}
+        return {"type": "http.disconnect"}
+
+    return receive
+
+
 class SageMakerRouteMiddleware:
     """ASGI middleware that reroutes /invocations based on CustomAttributes.
 
     Explicit route via header -> rewrites path to that endpoint.
     No route specified -> falls through to vLLM's built-in /invocations handler.
+
+    When the resolved route expects multipart/form-data (FORM_DATA_ROUTES) but
+    the body arrives as application/json, the JSON object is converted to a
+    multipart/form-data body so the upstream handler receives its expected
+    encoding.
     """
 
     def __init__(self, app: ASGIApp) -> None:
@@ -61,4 +139,50 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
                 scope["path"] = route
                 scope["raw_path"] = route.encode()
 
+                if (
+                    route in FORM_DATA_ROUTES
+                    and _content_type(scope.get("headers", [])) == "application/json"
+                ):
+                    scope, receive = await self._json_to_multipart(scope, receive)
+
         await self.app(scope, receive, send)
+
+    async def _json_to_multipart(self, scope: Scope, receive: Receive) -> tuple[Scope, Receive]:
+        """Drain the JSON body, convert it to multipart, and return a new
+        (scope, receive) carrying the rewritten content-type and body.
+
+        On any failure (non-object JSON, parse error) the original body is
+        replayed untouched so the upstream handler produces its normal error
+        response rather than this middleware masking it.
+        """
+        body = b""
+        more_body = True
+        while more_body:
+            message = await receive()
+            body += message.get("body", b"")
+            more_body = message.get("more_body", False)
+
+        try:
+            parsed = json.loads(body)
+            if not isinstance(parsed, dict):
+                raise ValueError("JSON body for a form route must be an object")
+        except (ValueError, json.JSONDecodeError) as exc:
+            logger.warning("Could not convert JSON body to multipart: %s", exc)
+            return scope, _replay(body)
+
+        boundary = uuid.uuid4().hex
+        new_body = _build_multipart(parsed, boundary)
+        content_type = f"multipart/form-data; boundary={boundary}".encode()
+
+        headers = [
+            (k, v)
+            for k, v in scope["headers"]
+            if k.lower() not in (b"content-type", b"content-length")
+        ]
+        headers.append((b"content-type", content_type))
+        headers.append((b"content-length", str(len(new_body)).encode()))
+        scope = dict(scope)
+        scope["headers"] = headers
+
+        logger.info("Converted application/json body to multipart/form-data")
+        return scope, _replay(new_body)

test/security/data/ecr_scan_allowlist/vllm_omni/framework_allowlist.json

@@ -69,6 +69,11 @@
         "reason": "go/stdlib in mooncake/libetcd_wrapper.so, unpatchable without mooncake upgrade",
         "review_by": "2026-08-25"
     },
+    {
+        "vulnerability_id": "CVE-2026-27145",
+        "reason": "go/stdlib in mooncake/libetcd_wrapper.so, unpatchable without mooncake upgrade",
+        "review_by": "2026-08-25"
+    },
     {
         "vulnerability_id": "RUSTSEC-2026-0185",
         "reason": "quinn-proto in uv binary, upstream uv has not released a fix yet",