issues/952: fix streaming connection closed before body consumed in orchestrated disaggregated routing

Signed-off-by: shoreline2 <shoreline2@users.noreply.github.com>

Author: shoreline2

src/vllm_router/services/request_service/request.py

@@ -786,7 +786,7 @@ async def route_orchestrated_disaggregated_request(
 
     try:
         # Use the shared aiohttp client from app state
-        client = request.app.state.aiohttp_client_wrapper()
+        client: aiohttp.ClientSession = request.app.state.aiohttp_client_wrapper()
 
         # Send to Prefill
         async with client.post(
@@ -835,57 +835,60 @@ async def route_orchestrated_disaggregated_request(
         decode_api_url = f"{decode_url}{endpoint}"
         logger.info(f"[{request_id}] Sending decode request to {decode_api_url}")
 
-        async with client.post(
+        decode_resp = await client.post(
             decode_api_url,
             json=decode_request,
             headers={
                 "Content-Type": "application/json",
                 "X-Request-Id": request_id,
             },
             timeout=aiohttp.ClientTimeout(total=600),
-        ) as decode_resp:
-            if decode_resp.status != 200:
-                error_text = await decode_resp.text()
-                logger.error(
-                    f"[{request_id}] Decode failed with status {decode_resp.status}: {error_text}"
-                )
-                return JSONResponse(
-                    status_code=decode_resp.status,
-                    content={"error": f"Decode failed: {error_text}"},
-                    headers={"X-Request-Id": request_id},
-                )
+        )
+        if decode_resp.status != 200:
+            error_text = await decode_resp.text()
+            decode_resp.release()
+            logger.error(
+                f"[{request_id}] Decode failed with status {decode_resp.status}: {error_text}"
+            )
+            return JSONResponse(
+                status_code=decode_resp.status,
+                content={"error": f"Decode failed: {error_text}"},
+                headers={"X-Request-Id": request_id},
+            )
 
-            if is_streaming:
-                # For streaming, yield chunks as they arrive (true streaming)
-                async def generate_stream():
-                    try:
-                        async for chunk in decode_resp.content.iter_any():
-                            if chunk:
-                                yield chunk
-                    finally:
-                        curr_time = time.time()
-                        logger.info(
-                            f"[{request_id}] Orchestrated streaming request completed, total time = {curr_time - in_router_time:.4f}s"
-                        )
-
-                return StreamingResponse(
-                    generate_stream(),
-                    media_type="text/event-stream",
-                    headers={"X-Request-Id": request_id},
-                )
-            else:
-                # For non-streaming, read full response
-                response_data = await decode_resp.read()
+        if is_streaming:
+            # For streaming, yield chunks as they arrive (true streaming)
+            async def generate_stream():
+                try:
+                    async for chunk in decode_resp.content.iter_any():
+                        if chunk:
+                            yield chunk
+                finally:
+                    decode_resp.release()
+                    curr_time = time.time()
+                    logger.info(
+                        f"[{request_id}] Orchestrated streaming request completed, total time = {curr_time - in_router_time:.4f}s"
+                    )
 
-                curr_time = time.time()
-                logger.info(
-                    f"[{request_id}] Orchestrated request completed, total time = {curr_time - in_router_time:.4f}s"
-                )
+            return StreamingResponse(
+                generate_stream(),
+                media_type="text/event-stream",
+                headers={"X-Request-Id": request_id},
+            )
+        else:
+            # For non-streaming, read full response
+            response_data = await decode_resp.read()
+            decode_resp.release()
 
-                return JSONResponse(
-                    content=json.loads(response_data),
-                    headers={"X-Request-Id": request_id},
-                )
+            curr_time = time.time()
+            logger.info(
+                f"[{request_id}] Orchestrated request completed, total time = {curr_time - in_router_time:.4f}s"
+            )
+
+            return JSONResponse(
+                content=json.loads(response_data),
+                headers={"X-Request-Id": request_id},
+            )
 
     except aiohttp.ClientError as e:
         logger.error(