[Bugfix][Frontend] Return 400 for malformed RL dev route bodies (#59236)

Signed-off-by: aoshen02 <aoshen02@users.noreply.github.com>
Co-authored-by: aoshen02 <aoshen02@users.noreply.github.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

tests/entrypoints/unit_tests/test_dev_route_request_bodies.py

@@ -0,0 +1,96 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Malformed bodies on RL dev routes are 400s that never reach the engine."""
+
+import json
+from types import SimpleNamespace
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from vllm.entrypoints.serve.dev.rlhf.api_router import router as rlhf_router
+from vllm.entrypoints.serve.dev.rpc.api_router import router as rpc_router
+from vllm.entrypoints.serve.exception_handling.register import init_exception_handler
+
+pytestmark = pytest.mark.cpu_test
+
+
+class Engine:
+    def __init__(self):
+        self.calls: list[str] = []
+
+    def __getattr__(self, name):
+        async def call(*args, **kwargs):
+            self.calls.append(name)
+
+        return call
+
+
+@pytest.fixture
+def client():
+    app = FastAPI()
+    app.include_router(rlhf_router)
+    app.include_router(rpc_router)
+    init_exception_handler(app)
+    app.state.engine_client = Engine()
+    app.state.args = SimpleNamespace(log_error_stack=False)
+    return TestClient(app)
+
+
+def post(client, path, body):
+    return client.post(
+        path, content=json.dumps(body), headers={"content-type": "application/json"}
+    )
+
+
+@pytest.mark.parametrize(
+    "path",
+    [
+        "/abort_requests",
+        "/init_weight_transfer_engine",
+        "/update_weights",
+        "/collective_rpc",
+    ],
+)
+@pytest.mark.parametrize("body", [[], 1, "x", None])
+def test_non_object_body_is_rejected(client, path, body):
+    assert post(client, path, body).status_code == 400
+    assert client.app.state.engine_client.calls == []
+
+
+@pytest.mark.parametrize(
+    "path,body",
+    [
+        ("/init_weight_transfer_engine", {}),
+        ("/init_weight_transfer_engine", {"init_info": []}),
+        ("/init_weight_transfer_engine", {"init_info": 1}),
+        ("/update_weights", {}),
+        ("/update_weights", {"update_info": 1}),
+        ("/update_weights", {"update_info": [1, 2]}),
+        ("/update_weights", {"update_info": "x"}),
+    ],
+)
+def test_invalid_field_is_rejected_before_the_engine(client, path, body):
+    # Reaching the engine would also abort an in-progress weight update.
+    assert post(client, path, body).status_code == 400
+    assert client.app.state.engine_client.calls == []
+
+
+@pytest.mark.parametrize(
+    "path,body,engine_call",
+    [
+        ("/abort_requests", {"request_ids": ["a"]}, "abort"),
+        (
+            "/init_weight_transfer_engine",
+            {"init_info": {}},
+            "init_weight_transfer_engine",
+        ),
+        ("/update_weights", {"update_info": {"names": []}}, "update_weights"),
+        ("/update_weights", {"update_info": [{}, {}]}, "update_weights"),
+        ("/collective_rpc", {"method": "m"}, "collective_rpc"),
+    ],
+)
+def test_valid_body_reaches_the_engine(client, path, body, engine_call):
+    assert post(client, path, body).status_code == 200
+    assert client.app.state.engine_client.calls == [engine_call]

vllm/entrypoints/serve/dev/rlhf/api_router.py

@@ -3,13 +3,14 @@
 
 import json
 from http import HTTPStatus
-from typing import Annotated
+from typing import Annotated, Any
 
 from fastapi import APIRouter, Body, FastAPI, HTTPException, Query, Request
 from fastapi.responses import JSONResponse
 
 from vllm.distributed.weight_transfer.base import (
     WeightTransferInitRequest,
+    WeightTransferUpdatePayload,
     WeightTransferUpdateRequest,
 )
 from vllm.engine.protocol import EngineClient
@@ -104,6 +105,10 @@ async def abort_requests(raw_request: Request) -> JSONResponse:
         body = await raw_request.json()
     except json.JSONDecodeError as e:
         raise HTTPException(status_code=400, detail="Invalid JSON format") from e  # noqa: B904
+    if not isinstance(body, dict):
+        raise HTTPException(
+            status_code=400, detail="Request body must be a JSON object"
+        )
 
     request_ids = body.get("request_ids")
 
@@ -154,17 +159,10 @@ async def is_paused(raw_request: Request) -> JSONResponse:
 
 
 @router.post("/init_weight_transfer_engine")
-async def init_weight_transfer_engine(raw_request: Request):
-    try:
-        body = await raw_request.json()
-    except json.JSONDecodeError as e:
-        raise HTTPException(status_code=400, detail="Invalid JSON format") from e  # noqa: B904
-    init_info = body.get("init_info")
-    if init_info is None:
-        raise HTTPException(
-            status_code=HTTPStatus.BAD_REQUEST.value,
-            detail="Missing 'init_info' in request body",
-        )
+async def init_weight_transfer_engine(
+    raw_request: Request,
+    init_info: Annotated[dict[str, Any], Body(embed=True)],
+):
     await engine_client(raw_request).init_weight_transfer_engine(
         WeightTransferInitRequest(init_info=init_info)
     )
@@ -184,17 +182,10 @@ async def start_draft_weight_update(raw_request: Request):
 
 
 @router.post("/update_weights")
-async def update_weights(raw_request: Request):
-    try:
-        body = await raw_request.json()
-    except json.JSONDecodeError as e:
-        raise HTTPException(status_code=400, detail="Invalid JSON format") from e  # noqa: B904
-    update_info = body.get("update_info")
-    if update_info is None:
-        raise HTTPException(
-            status_code=HTTPStatus.BAD_REQUEST.value,
-            detail="Missing 'update_info' in request body",
-        )
+async def update_weights(
+    raw_request: Request,
+    update_info: Annotated[WeightTransferUpdatePayload, Body(embed=True)],
+):
     await engine_client(raw_request).update_weights(
         request=WeightTransferUpdateRequest(update_info=update_info)
     )

vllm/entrypoints/serve/dev/rpc/api_router.py

@@ -29,6 +29,11 @@ async def collective_rpc(raw_request: Request):
             status_code=HTTPStatus.BAD_REQUEST.value,
             detail=f"JSON decode error: {e}",
         ) from e
+    if not isinstance(body, dict):
+        raise HTTPException(
+            status_code=HTTPStatus.BAD_REQUEST.value,
+            detail="Request body must be a JSON object",
+        )
     method = body.get("method")
     if method is None:
         raise HTTPException(