[Serve][LLM] Drop LLM-side body-forwarding gate

The expensive part of body forwarding lives in HAProxy + Lua (the
`wait-for-body` round-trip, `tune.bufsize` connection cost, and body
re-emit on the socket). On the LLMRouter side, `await request.body()`
on a Content-Length: 0 request is essentially free (one async tick,
returns b""), so gating it earns nothing. Reverts the LLM-side branch
and keeps the body read unconditional, making this PR Serve-only.

Signed-off-by: Seiji Eicher <seiji@anyscale.com>

Author: eicherseiji

python/ray/llm/_internal/serve/core/ingress/router.py

@@ -5,9 +5,6 @@
 
 from ray import serve
 from ray.serve._private.common import ReplicaID
-from ray.serve._private.constants import (
-    RAY_SERVE_INGRESS_REQUEST_ROUTER_FORWARD_BODY,
-)
 from ray.serve.handle import DeploymentHandle
 
 _BODY_TRUNCATED_HEADER = "x-body-truncated"
@@ -29,6 +26,7 @@
 
 router_app = FastAPI()
 
+
 @serve.ingress(router_app)
 class LLMRouter:
     """Ingress request router for direct streaming.
@@ -94,11 +92,8 @@ async def check_health(self):
 
     @router_app.post("/internal/route")
     async def route(self, request: Request):
-        body = None
-        body_truncated = False
-        if RAY_SERVE_INGRESS_REQUEST_ROUTER_FORWARD_BODY:
-            body = await request.body()
-            body_truncated = _BODY_TRUNCATED_HEADER in request.headers
+        body = await request.body()
+        body_truncated = _BODY_TRUNCATED_HEADER in request.headers
         try:
             host, port, replica_id = self._pick_replica(
                 request_body=body, body_truncated=body_truncated

python/ray/llm/tests/serve/cpu/deployments/routers/test_builder_ingress.py

@@ -438,54 +438,6 @@ def test_direct_streaming_rejects_ingress_config(
         with pytest.raises(ValueError, match=match):
             build_openai_app(LLMServingArgs(llm_configs=[llm_config], **builder_kwargs))
 
-    def test_direct_streaming_rejects_multiple_llm_configs(
-        self, llm_config, disable_placement_bundles, monkeypatch
-    ):
-        monkeypatch.setattr(
-            "ray.llm._internal.serve.core.ingress.builder."
-            "RAY_SERVE_LLM_ENABLE_DIRECT_STREAMING",
-            True,
-        )
-        other_llm_config = LLMConfig(
-            model_loading_config=ModelLoadingConfig(model_id="other-model")
-        )
-
-        with pytest.raises(
-            ValueError,
-            match="currently supports exactly one LLM config",
-        ):
-            build_openai_app(LLMServingArgs(llm_configs=[llm_config, other_llm_config]))
-
-    @pytest.mark.parametrize(
-        ("builder_kwargs", "match"),
-        [
-            (
-                {"ingress_deployment_config": {"num_replicas": 2}},
-                "does not support ingress_deployment_config",
-            ),
-            (
-                {"ingress_cls_config": {"ingress_extra_kwargs": {"key": "value"}}},
-                "does not support ingress_cls_config",
-            ),
-        ],
-    )
-    def test_direct_streaming_rejects_ingress_config(
-        self,
-        llm_config,
-        disable_placement_bundles,
-        monkeypatch,
-        builder_kwargs,
-        match,
-    ):
-        monkeypatch.setattr(
-            "ray.llm._internal.serve.core.ingress.builder."
-            "RAY_SERVE_LLM_ENABLE_DIRECT_STREAMING",
-            True,
-        )
-
-        with pytest.raises(ValueError, match=match):
-            build_openai_app(LLMServingArgs(llm_configs=[llm_config], **builder_kwargs))
-
 
 class TestIngressScaleToZero:
     """Tests for ingress scale-to-zero behavior when all models have min_replicas=0."""

python/ray/serve/_private/controller.py

@@ -1552,73 +1552,6 @@ def _get_target_groups_for_app(
 
         return target_groups
 
-    def _get_target_groups_for_app_with_ingress_request_router(
-        self,
-        app_name: str,
-        route_prefix: str,
-        ingress_request_router_deployment_name: str,
-    ) -> List[TargetGroup]:
-        """Create target groups for ingress bypass mode.
-
-        Ingress request router targets serve /internal/route for Lua
-        routing decisions. Main targets serve data plane traffic via
-        direct ingress.
-        """
-        ingress_request_router_targets = self._get_targets_for_protocol(
-            self._get_running_replica_details_for_deployment(
-                app_name, ingress_request_router_deployment_name
-            ),
-            RequestProtocol.HTTP,
-        )
-
-        backend_deployment_names = (
-            deployment_name
-            for deployment_name in self.application_state_manager.get_deployments(
-                app_name
-            )
-            ingress_request_router_targets = None
-        else:
-            ingress_request_router_targets = self._get_targets_for_protocol(
-                self._get_running_replica_details_for_deployment(
-                    app_name, ingress_request_router_deployment_name
-                ),
-                RequestProtocol.HTTP,
-            )
-
-        target_groups = []
-
-        # Create targets for each protocol
-        http_targets = self._get_targets_for_protocol(
-            replica_details, RequestProtocol.HTTP
-        )
-        if http_targets:
-            target_groups.append(
-                TargetGroup(
-                    protocol=RequestProtocol.HTTP,
-                    route_prefix=route_prefix,
-                    targets=http_targets,
-                    app_name=app_name,
-                    ingress_request_router_targets=ingress_request_router_targets,
-                )
-            )
-
-        # Add gRPC targets if enabled
-        if is_grpc_enabled(self.get_grpc_config()):
-            grpc_targets = self._get_targets_for_protocol(
-                replica_details, RequestProtocol.GRPC
-            )
-            if grpc_targets:
-                target_groups.append(
-                    TargetGroup(
-                        protocol=RequestProtocol.GRPC,
-                        route_prefix=route_prefix,
-                        targets=grpc_targets,
-                        app_name=app_name,
-                    )
-                )
-
-        return target_groups
-
     def _get_target_groups_for_app_with_no_running_replicas(
         self, route_prefix: str, app_name: str
     ) -> List[TargetGroup]: