diff --git a/rust/sglang-server/src/api_server/frame.rs b/rust/sglang-server/src/api_server/frame.rs index 129f7c619..f58d42427 100644 --- a/rust/sglang-server/src/api_server/frame.rs +++ b/rust/sglang-server/src/api_server/frame.rs @@ -159,12 +159,6 @@ fn hidden_states_rows(vals: &[f32], lens: &[u32]) -> serde_json::Value { serde_json::Value::Array(rows) } -/// The `{ "error": { message, code } }` object every error path emits (an SSE -/// event's data, a unary body, or one entry of a batch array). -pub(super) fn error_value(code: u16, message: &str) -> serde_json::Value { - serde_json::json!({ "error": { "message": message, "code": code } }) -} - /// Format a decoded [`ChunkEvent`] as one SGLang `/generate` frame's JSON. `rid` /// (response `meta_info.id`) is passed as a string; the event's numeric `rid` is /// just the shard routing key. diff --git a/rust/sglang-server/src/api_server/native_api.rs b/rust/sglang-server/src/api_server/native_api.rs index bbe3a92b0..f0db30337 100644 --- a/rust/sglang-server/src/api_server/native_api.rs +++ b/rust/sglang-server/src/api_server/native_api.rs @@ -24,14 +24,14 @@ use tokio::sync::mpsc; use super::AppState; use super::frame::{ - OutputAccumulator, cumulative_frame_string, error_value, frame_value, stream_frame_string, - tag_value, + OutputAccumulator, cumulative_frame_string, frame_value, stream_frame_string, tag_value, }; use super::guard::AbortGuard; -use super::submit::{pre_submit_error, submit}; +use super::submit::submit; use crate::environ::env_bool; use crate::ids::Rid; use crate::message::{EgressItem, GenerateBody, GenerateRequest, RequestKind, SamplingParams}; +use crate::utils::response::{error_response, error_value}; /// The routes this module owns, mounted by `api_server::serve`. pub(super) fn routes() -> Router { @@ -40,6 +40,12 @@ pub(super) fn routes() -> Router { .merge(health_routes()) } +/// native api error response: unary → `code` plus the JSON `body`, +/// streaming → 200 with one SSE error frame + `[DONE]`. +pub(super) fn native_error(code: StatusCode, message: &str, stream: bool) -> Response { + error_response(code, error_value(code.as_u16(), message), stream) +} + /// `/health` + `/health_generate`. Both env knobs are resolved ONCE here, at /// router build (server startup) — changing them on a live process needs a /// restart. The deep-probe handler is built once with @@ -148,7 +154,7 @@ async fn generate( // can only answer unary — as Python's does (FastAPI rejects before its // handler runs). Err(rejection) => { - return pre_submit_error(StatusCode::BAD_REQUEST, &rejection.body_text(), false); + return native_error(StatusCode::BAD_REQUEST, &rejection.body_text(), false); } }; let stream = body.stream; @@ -159,13 +165,13 @@ async fn generate( // The error carries its own status (a bad batch is `Validation` → 400). Err(e) => { let code = StatusCode::from_u16(e.http_status()).unwrap_or(StatusCode::BAD_REQUEST); - return pre_submit_error(code, &e.to_string(), stream); + return native_error(code, &e.to_string(), stream); } }; // Media I/O (URL downloads, file reads) happens here, on the API runtime // — never on the MM worker pool (see `prefetch`). if let Err(e) = super::prefetch::prefetch_all(&mut payloads).await { - return pre_submit_error(StatusCode::BAD_REQUEST, &e, stream); + return native_error(StatusCode::BAD_REQUEST, &e, stream); } if !is_batch { // `into_requests` guarantees exactly one payload for a non-batch body. diff --git a/rust/sglang-server/src/api_server/openai.rs b/rust/sglang-server/src/api_server/openai.rs index f3e59090b..554050371 100644 --- a/rust/sglang-server/src/api_server/openai.rs +++ b/rust/sglang-server/src/api_server/openai.rs @@ -4,11 +4,7 @@ //! request and response primitives. Native [`ChunkEvent`] values remain the one //! backend output type for both unary and streaming responses. -use axum::{ - Json, Router, - http::StatusCode, - response::{IntoResponse, Response}, -}; +use axum::{Router, http::StatusCode, response::Response}; use futures::StreamExt; use tokio::sync::mpsc; @@ -28,6 +24,7 @@ use super::submit::submit; use crate::ids::Rid; use crate::message::{ChunkEvent, EgressItem, GenerateRequest, RequestKind}; use crate::runtime::ServerArgs; +use crate::utils::response::error_response; const MAX_OPENAI_CHOICES: usize = 4096; @@ -86,10 +83,9 @@ fn unix_seconds_u32() -> u32 { u32::try_from(unix_seconds()).unwrap_or(u32::MAX) } -/// The OpenAI `{"error": {...}}` payload: `type` is the SDK-facing error kind -/// (`AuthenticationError` / `InternalServerError` / `BadRequestError`), and -/// `code` carries the HTTP status — the shape Python's OpenAI frontend emits. -fn error_payload(code: StatusCode, message: String) -> serde_json::Value { +/// The OpenAI error payload. +pub(super) fn error_payload(code: StatusCode, message: impl Into) -> serde_json::Value { + let message = message.into(); let error_type = if code == StatusCode::UNAUTHORIZED { "AuthenticationError" } else if code.is_server_error() { @@ -108,36 +104,15 @@ fn error_payload(code: StatusCode, message: String) -> serde_json::Value { }) } -/// Shape a `StatusCode` + message into an OpenAI error response, mirroring -/// `pre_submit_error`'s rule: unary requests get the JSON error with its -/// status; a request whose stream is already committed gets 200 + one SSE -/// error frame + `[DONE]`. -pub(super) fn openai_error_response( - code: StatusCode, - message: impl Into, - stream: bool, -) -> Response { - let body = error_payload(code, message.into()); - if !stream { - return (code, Json(body)).into_response(); - } - super::submit::sse_error_response(body) -} - -/// Unary OpenAI error — the common pre-submit case (Python validates before -/// the stream starts and answers 4xx in JSON even for `stream=true`). -fn openai_error(code: StatusCode, message: impl Into) -> Response { - openai_error_response(code, message, false) -} - -/// The OpenAI error frame payload for errors raised *inside* a committed -/// stream, where only a `data:` frame can be emitted (the status is folded -/// into the body, since the response status is already 200). -pub(super) fn streaming_error(code: u16, message: impl Into) -> String { - let status = StatusCode::from_u16(code).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR); - error_payload(status, message.into()).to_string() +/// Form an OpenAI error response: unary → `code` plus the JSON `body`, +/// streaming → 200 with one SSE error frame + `[DONE]`. +pub(super) fn openai_error(code: StatusCode, message: impl Into, stream: bool) -> Response { + error_response(code, error_payload(code, message), stream) } +/// Drain one submitted request to its terminal output: fold frames, disarm +/// `guard` on a natural terminal, and map errors / validation aborts / +/// truncation to `(status, message)` for the OpenAI error shape. async fn collect_output( mut rx: mpsc::Receiver, guard: &mut AbortGuard, @@ -191,10 +166,10 @@ async fn submit_generation( guard.arm(rid); Ok(rx) } - // Same rule as `pre_submit_error`: a committed stream gets 200 plus an + // Same `error_response` rule: a committed stream gets 200 plus an // SSE error frame + `[DONE]`, not a unary 503 — but with the OpenAI // error shape, since this is the OpenAI frontend. - Err(_) => Err(openai_error_response( + Err(_) => Err(openai_error( StatusCode::SERVICE_UNAVAILABLE, "service unavailable", stream, diff --git a/rust/sglang-server/src/api_server/openai/chat.rs b/rust/sglang-server/src/api_server/openai/chat.rs index 5e8c04ddf..54ee59911 100644 --- a/rust/sglang-server/src/api_server/openai/chat.rs +++ b/rust/sglang-server/src/api_server/openai/chat.rs @@ -33,8 +33,8 @@ use super::tools::{ parse_chat_tool_calls, }; use super::{ - AppState, ChatFormatter, collect_output, contains_media, indexed_egress_stream, openai_error, - streaming_error, submit_generation, unix_seconds_u32, + AppState, ChatFormatter, collect_output, contains_media, error_payload, indexed_egress_stream, + openai_error, submit_generation, unix_seconds_u32, }; use crate::ids::Rid; use crate::message::{ChunkExtras, EgressItem, GenerateRequest, OneOrMany, SamplingParams}; @@ -49,25 +49,29 @@ async fn chat_completions( ) -> Response { let request = match body { Ok(Json(request)) => request, - Err(rejection) => return openai_error(StatusCode::BAD_REQUEST, rejection.body_text()), + Err(rejection) => { + return openai_error(StatusCode::BAD_REQUEST, rejection.body_text(), false); + } }; if request.model != state.server_args.served_model_name { return openai_error( StatusCode::BAD_REQUEST, format!("The model `{}` does not exist", request.model), + false, ); } if request.messages.is_empty() { - return openai_error(StatusCode::BAD_REQUEST, "messages cannot be empty"); + return openai_error(StatusCode::BAD_REQUEST, "messages cannot be empty", false); } if serde_json::to_value(&request.messages).is_ok_and(|messages| contains_media(&messages)) { return openai_error( StatusCode::BAD_REQUEST, "image, audio, video, and file message content is not supported", + false, ); } if request.n == Some(0) { - return openai_error(StatusCode::BAD_REQUEST, "n must be at least 1"); + return openai_error(StatusCode::BAD_REQUEST, "n must be at least 1", false); } #[allow(deprecated)] let max_tokens = request.max_completion_tokens.or(request.max_tokens); @@ -75,6 +79,7 @@ async fn chat_completions( return openai_error( StatusCode::BAD_REQUEST, "max_completion_tokens must be positive", + false, ); } if request.modalities.as_ref().is_some_and(|modalities| { @@ -87,6 +92,7 @@ async fn chat_completions( return openai_error( StatusCode::BAD_REQUEST, "audio, prediction, web search, and multimodal inputs are not supported", + false, ); } #[allow(deprecated)] @@ -94,6 +100,7 @@ async fn chat_completions( return openai_error( StatusCode::BAD_REQUEST, "deprecated function_call/functions are not supported; use tools and tool_choice", + false, ); } @@ -110,6 +117,7 @@ async fn chat_completions( return openai_error( StatusCode::BAD_REQUEST, "tool calls require --tool-call-parser", + false, ); } // Python gates the split on `request.separate_reasoning` (default true); @@ -143,7 +151,9 @@ async fn chat_completions( &state.server_args, ) { Ok(sampling) => sampling, - Err(message) => return openai_error(StatusCode::BAD_REQUEST, message), + Err(message) => { + return openai_error(StatusCode::BAD_REQUEST, message, false); + } }; let stream = request.stream.unwrap_or(false); @@ -244,6 +254,7 @@ pub(super) async fn prepare_chat_request( return Err(openai_error( StatusCode::BAD_REQUEST, "this model has no usable chat template", + false, )); }; // Template stops first, then the request's own — Python @@ -255,6 +266,7 @@ pub(super) async fn prepare_chat_request( openai_error( StatusCode::BAD_REQUEST, format!("chat template render failed: {error}"), + false, ) })?; Ok((request, prompt)) @@ -428,7 +440,9 @@ pub(super) async fn unary_chat( for (index, rid, rx) in submitted { let output = match collect_output(rx, &mut guard, &rid).await { Ok(output) => output, - Err((status, message)) => return openai_error(status, message), + Err((status, message)) => { + return openai_error(status, message, false); + } }; if prompt_tokens == 0 { @@ -559,7 +573,7 @@ pub(super) fn chat_event_stream( id: None, event: None, comment: None, - error: Some(streaming_error(500, "response truncated before completion")), + error: Some(error_payload(StatusCode::INTERNAL_SERVER_ERROR, "response truncated before completion").to_string()), }; continue; }; @@ -576,7 +590,7 @@ pub(super) fn chat_event_stream( id: None, event: None, comment: None, - error: Some(streaming_error(error.http_status(), error.to_string())), + error: Some(error_payload(StatusCode::from_u16(error.http_status()).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR), error.to_string()).to_string()), }; continue; } @@ -592,7 +606,7 @@ pub(super) fn chat_event_stream( id: None, event: None, comment: None, - error: Some(streaming_error(code, message)), + error: Some(error_payload(StatusCode::from_u16(code).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR), message).to_string()), }; continue; } diff --git a/rust/sglang-server/src/api_server/openai/completions.rs b/rust/sglang-server/src/api_server/openai/completions.rs index fcd2be828..7147c685f 100644 --- a/rust/sglang-server/src/api_server/openai/completions.rs +++ b/rust/sglang-server/src/api_server/openai/completions.rs @@ -23,8 +23,8 @@ use tokio::sync::mpsc; use super::super::guard::AbortGuard; use super::super::submit::submit; use super::{ - AppState, MAX_OPENAI_CHOICES, collect_output, indexed_egress_stream, openai_error, - streaming_error, submit_generation, unix_seconds_u32, + AppState, MAX_OPENAI_CHOICES, collect_output, error_payload, indexed_egress_stream, + openai_error, submit_generation, unix_seconds_u32, }; use crate::ids::Rid; use crate::message::{ @@ -64,7 +64,7 @@ async fn completions( let request = match body { Ok(Json(request)) => request, Err(rejection) => { - return openai_error(StatusCode::BAD_REQUEST, rejection.body_text()); + return openai_error(StatusCode::BAD_REQUEST, rejection.body_text(), false); } }; let stream = request.stream.unwrap_or(false); @@ -74,6 +74,7 @@ async fn completions( return openai_error( StatusCode::BAD_REQUEST, format!("The model `{model}` does not exist"), + false, ); } @@ -81,33 +82,44 @@ async fn completions( return openai_error( StatusCode::BAD_REQUEST, "prompt_embeds is not supported by the Rust frontend", + false, ); } if request.suffix.is_some() { return openai_error( StatusCode::BAD_REQUEST, "suffix is not supported by this model", + false, ); } if request.best_of.is_some_and(|best_of| best_of != 1) { return openai_error( StatusCode::BAD_REQUEST, "best_of values greater than 1 are not supported", + false, ); } if request.max_tokens == Some(0) { - return openai_error(StatusCode::BAD_REQUEST, "max_tokens must be positive"); + return openai_error( + StatusCode::BAD_REQUEST, + "max_tokens must be positive", + false, + ); } if request.n == Some(0) { - return openai_error(StatusCode::BAD_REQUEST, "n must be at least 1"); + return openai_error(StatusCode::BAD_REQUEST, "n must be at least 1", false); } let prompts = match completion_prompt_specs(&request.prompt) { Ok(prompts) => prompts, - Err(message) => return openai_error(StatusCode::BAD_REQUEST, message), + Err(message) => { + return openai_error(StatusCode::BAD_REQUEST, &message, false); + } }; let mut sampling = match completion_sampling_params(&request) { Ok(sampling) => sampling, - Err(message) => return openai_error(StatusCode::BAD_REQUEST, message), + Err(message) => { + return openai_error(StatusCode::BAD_REQUEST, &message, false); + } }; if let Err(error) = sampling.normalize( state.server_args.skip_tokenizer_init, @@ -117,7 +129,7 @@ async fn completions( .vocab_size .unwrap_or(u64::MAX), ) { - return openai_error(StatusCode::BAD_REQUEST, error.to_string()); + return openai_error(StatusCode::BAD_REQUEST, error.to_string(), false); } let n = request.n.unwrap_or(1) as usize; @@ -127,6 +139,7 @@ async fn completions( return openai_error( StatusCode::BAD_REQUEST, format!("prompt count times n exceeds the maximum of {MAX_OPENAI_CHOICES}"), + false, ); } }; @@ -235,6 +248,7 @@ async fn decode_prompt_echo(state: &AppState, token_ids: TokenIds) -> Result Result { - Err(openai_error(StatusCode::BAD_REQUEST, message)) + Err(openai_error(StatusCode::BAD_REQUEST, &message, false)) } Some(EgressItem::Error(error)) => { let status = StatusCode::from_u16(error.http_status()) @@ -253,11 +268,13 @@ async fn decode_prompt_echo(state: &AppState, token_ids: TokenIds) -> Result Err(openai_error( StatusCode::INTERNAL_SERVER_ERROR, "failed to decode prompt for echo: reply channel closed", + false, )), } } @@ -358,7 +375,9 @@ pub(super) async fn unary_completion( for choice in submitted { let output = match collect_output(choice.rx, &mut guard, &choice.rid).await { Ok(output) => output, - Err((status, message)) => return openai_error(status, message), + Err((status, message)) => { + return openai_error(status, &message, false); + } }; prompt_tokens @@ -525,7 +544,7 @@ pub(super) fn completion_event_stream( while let Some((index, item)) = events.next().await { let Some(item) = item else { - yield streaming_error(500, "response truncated before completion"); + yield error_payload(StatusCode::INTERNAL_SERVER_ERROR, "response truncated before completion").to_string(); continue; }; let output = match item { @@ -536,7 +555,7 @@ pub(super) fn completion_event_stream( } EgressItem::Error(error) => { guard.disarm(&rids[index]); - yield streaming_error(error.http_status(), error.to_string()); + yield error_payload(StatusCode::from_u16(error.http_status()).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR), error.to_string()).to_string(); continue; } EgressItem::Control(_) | EgressItem::Data(_) => continue, @@ -547,7 +566,7 @@ pub(super) fn completion_event_stream( .as_ref() .and_then(|reason| reason.abort_status()) { - yield streaming_error(code, message); + yield error_payload(StatusCode::from_u16(code).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR), message).to_string(); continue; } diff --git a/rust/sglang-server/src/api_server/openai/models.rs b/rust/sglang-server/src/api_server/openai/models.rs index 22f78f9b9..f8a56b7d1 100644 --- a/rust/sglang-server/src/api_server/openai/models.rs +++ b/rust/sglang-server/src/api_server/openai/models.rs @@ -28,6 +28,7 @@ async fn retrieve_model(State(state): State, Path(model): Path return openai_error( StatusCode::NOT_FOUND, format!("The model `{model}` does not exist"), + false, ); } Json(model_card(&state)).into_response() diff --git a/rust/sglang-server/src/api_server/openai/test_utils.rs b/rust/sglang-server/src/api_server/openai/test_utils.rs index 5ae3170d4..68eeea61f 100644 --- a/rust/sglang-server/src/api_server/openai/test_utils.rs +++ b/rust/sglang-server/src/api_server/openai/test_utils.rs @@ -18,7 +18,7 @@ use axum::response::Response; use serde_json::json; use tower::util::ServiceExt; -use super::routes; +use super::{openai_error, routes}; use crate::ids::Rid; use crate::message::{ChunkEvent, EgressItem}; use crate::runtime::ServerArgs; @@ -157,13 +157,13 @@ pub(super) async fn body_json(response: Response) -> serde_json::Value { serde_json::from_slice(&bytes).unwrap() } -/// The common StatusCode→error helper follows `pre_submit_error`'s shape: +/// The common StatusCode→error helper follows `error_response`'s shape: /// unary requests get the JSON error with its status; a committed stream gets /// 200 + one SSE error frame + `[DONE]`, and the frame carries the OpenAI /// error fields (`type`, `param`, `code`) that the SDKs dispatch on. #[tokio::test] async fn openai_error_response_covers_unary_and_sse() { - let unary = super::openai_error_response(StatusCode::BAD_REQUEST, "bad input", false); + let unary = openai_error(StatusCode::BAD_REQUEST, "bad input", false); assert_eq!(unary.status(), StatusCode::BAD_REQUEST); let value = body_json(unary).await; assert_eq!(value["error"]["message"], "bad input"); @@ -171,7 +171,7 @@ async fn openai_error_response_covers_unary_and_sse() { assert_eq!(value["error"]["code"], 400); assert!(value["error"]["param"].is_null()); - let streamed = super::openai_error_response(StatusCode::BAD_REQUEST, "bad input", true); + let streamed = openai_error(StatusCode::BAD_REQUEST, "bad input", true); assert_eq!(streamed.status(), StatusCode::OK); let bytes = axum::body::to_bytes(streamed.into_body(), 64 * 1024) .await @@ -290,7 +290,7 @@ async fn basic_openai_router_excludes_responses_api() { /// A closed tm inbox with a *streaming* request must answer inside the /// committed stream: 200 + one OpenAI-shaped SSE error frame + `[DONE]` (the -/// same rule `pre_submit_error` applies to the native API), not a unary 503. +/// same `error_response` rule the native API applies), not a unary 503. #[tokio::test] async fn streaming_submit_failure_answers_inside_the_stream() { let app = routes().with_state(app_state(senders_closed())); diff --git a/rust/sglang-server/src/api_server/submit.rs b/rust/sglang-server/src/api_server/submit.rs index 7dabba70b..fa2f23985 100644 --- a/rust/sglang-server/src/api_server/submit.rs +++ b/rust/sglang-server/src/api_server/submit.rs @@ -2,20 +2,10 @@ //! module: mint the client-visible rid (uuid hex, Python-parity), build the //! `Request`, and hand it to the TM with an egress receiver for the response. -use std::convert::Infallible; - -use axum::{ - Json, - http::StatusCode, - response::{ - IntoResponse, Response, - sse::{Event, Sse}, - }, -}; +use axum::{http::StatusCode, response::Response}; use tokio::sync::mpsc; -use super::AppState; -use super::frame::error_value; +use super::{AppState, native_api::native_error}; use crate::fsm::RequestState; use crate::ids::Rid; use crate::message::{EgressItem, EgressSink, Request, RequestKind}; @@ -29,8 +19,7 @@ pub(super) async fn submit( state: &AppState, kind: RequestKind, // `stream`: the client is reading an SSE stream, so it expects 200 plus an - // error frame rather than a 4xx — same rule `pre_submit_error` applies - // everywhere else. + // error frame rather than a 4xx — `utils::response::error_response`'s rule. stream: bool, ) -> Result<(Rid, mpsc::Receiver), Response> { let rid = match &kind { @@ -61,7 +50,7 @@ pub(super) async fn submit( Err(_) => { tracing::error!(%rid, "tm inbox closed; request rejected"); // Return 503 so the client can retry. - Err(pre_submit_error( + Err(native_error( StatusCode::SERVICE_UNAVAILABLE, "service unavailable", stream, @@ -69,68 +58,3 @@ pub(super) async fn submit( } } } - -/// Shape an error that occurs *before* (or instead of) a successful submit into a -/// client response. Two parity points with Python's `generate_request`. The body -/// is the same `{"error": {...}}` object every other path emits — not bare text, -/// which a client parsing JSON chokes on. And a streaming request gets 200 plus -/// one SSE error frame and `[DONE]`, not a 4xx: the client has already committed -/// to reading a stream, and Python answers it inside `stream_results()`. -pub(super) fn pre_submit_error(code: StatusCode, message: &str, stream: bool) -> Response { - let body = error_value(code.as_u16(), message); - if !stream { - return (code, Json(body)).into_response(); - } - sse_error_response(body) -} - -/// A 200 SSE response carrying one error frame + `[DONE]` — how a stream the -/// client is already committed to reading reports a failure. Shared by every -/// endpoint family: the native API via [`pre_submit_error`] and the OpenAI -/// frontend's `openai_error_response`. -pub(super) fn sse_error_response(body: serde_json::Value) -> Response { - let frames = [body.to_string(), "[DONE]".to_string()]; - Sse::new(futures::stream::iter( - frames.map(|data| Ok::<_, Infallible>(Event::default().data(data))), - )) - .into_response() -} - -#[cfg(test)] -mod tests { - use super::*; - - /// Unary pre-submit errors are a 4xx/5xx with a JSON `{"error":...}` body; - /// streaming ones are 200 + an SSE error frame + `[DONE]`, because Python - /// answers from inside `stream_results()` once the stream is committed. - #[tokio::test] - async fn pre_submit_errors_match_python_shape() { - let unary = pre_submit_error(StatusCode::BAD_REQUEST, "bad input", false); - assert_eq!(unary.status(), StatusCode::BAD_REQUEST); - let body = axum::body::to_bytes(unary.into_body(), 64 * 1024) - .await - .unwrap(); - let v: serde_json::Value = serde_json::from_slice(&body).expect("JSON body"); - assert_eq!(v["error"]["message"], "bad input"); - assert_eq!(v["error"]["code"], 400); - - let streamed = pre_submit_error(StatusCode::BAD_REQUEST, "bad input", true); - assert_eq!( - streamed.status(), - StatusCode::OK, - "the stream itself is 200" - ); - let body = axum::body::to_bytes(streamed.into_body(), 64 * 1024) - .await - .unwrap(); - let text = String::from_utf8(body.to_vec()).unwrap(); - assert!( - text.contains(r#""code":400"#), - "carries the status in-band: {text}" - ); - assert!( - text.trim_end().ends_with("data: [DONE]"), - "terminated: {text}" - ); - } -} diff --git a/rust/sglang-server/src/utils.rs b/rust/sglang-server/src/utils.rs index 39d9e0a1e..33bab3e85 100644 --- a/rust/sglang-server/src/utils.rs +++ b/rust/sglang-server/src/utils.rs @@ -1,4 +1,5 @@ //! Shared helpers with no home in a pipeline stage. pub mod regex; +pub mod response; pub mod sock; diff --git a/rust/sglang-server/src/utils/response.rs b/rust/sglang-server/src/utils/response.rs new file mode 100644 index 000000000..994863aa9 --- /dev/null +++ b/rust/sglang-server/src/utils/response.rs @@ -0,0 +1,94 @@ +//! Shared HTTP error-response shaping for the api-server endpoint modules. +//! +//! Two mechanics live here; the WIRE SHAPES stay owned by their endpoints: +//! the native `{"error": {"message", "code"}}` body (Python +//! `http_server.generate_request` parity) built by [`error_value`], and the +//! SSE variant [`sse_error_response`] used by any +//! endpoint family (native and OpenAI alike — the caller supplies its own +//! body shape). The OpenAI error payload and the PD bootstrap registry's +//! plain-text bodies are protocol-owned and deliberately not unified here. + +use std::convert::Infallible; + +use axum::{ + Json, + http::StatusCode, + response::{ + IntoResponse, Response, + sse::{Event, Sse}, + }, +}; + +/// The native error body — the same `{"error": {...}}` object every native +/// path emits, not bare text, which a client parsing JSON chokes on. +pub fn error_value(code: u16, message: &str) -> serde_json::Value { + serde_json::json!({ "error": { "message": message, "code": code } }) +} + +/// Form an error in the shape the client committed to: unary → `code` plus +/// the JSON `body`; streaming → 200 with one SSE error frame + `[DONE]` (the +/// client is already reading a stream — Python answers in-stream too, from +/// `stream_results()`). The `body` is caller-shaped: native [`error_value`] +/// or the OpenAI error payload. +pub fn error_response(code: StatusCode, body: serde_json::Value, stream: bool) -> Response { + if !stream { + return (code, Json(body)).into_response(); + } + sse_error_response(body) +} + +/// A 200 SSE response carrying one error frame + `[DONE]` — how a stream the +/// client is already committed to reading reports a failure. Shared by every +/// endpoint family: the native API and the OpenAI +/// frontend's `openai_error_response`. +pub fn sse_error_response(body: serde_json::Value) -> Response { + let frames = [body.to_string(), "[DONE]".to_string()]; + Sse::new(futures::stream::iter( + frames.map(|data| Ok::<_, Infallible>(Event::default().data(data))), + )) + .into_response() +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Python-parity pins for the native error shapes (`generate_request`): + /// unary errors are a 4xx/5xx with a JSON `{"error": ...}` body; streaming + /// ones are 200 + one SSE error frame + `[DONE]`, because Python answers + /// from inside `stream_results()` once the stream is committed. + #[tokio::test] + async fn error_responses_match_python_shape() { + let unary = error_response( + StatusCode::BAD_REQUEST, + error_value(400, "bad input"), + false, + ); + assert_eq!(unary.status(), StatusCode::BAD_REQUEST); + let body = axum::body::to_bytes(unary.into_body(), 64 * 1024) + .await + .unwrap(); + let v: serde_json::Value = serde_json::from_slice(&body).expect("JSON body"); + assert_eq!(v["error"]["message"], "bad input"); + assert_eq!(v["error"]["code"], 400); + + let streamed = error_response(StatusCode::BAD_REQUEST, error_value(400, "bad input"), true); + assert_eq!( + streamed.status(), + StatusCode::OK, + "the stream itself is 200" + ); + let body = axum::body::to_bytes(streamed.into_body(), 64 * 1024) + .await + .unwrap(); + let text = String::from_utf8(body.to_vec()).unwrap(); + assert!( + text.contains(r#""code":400"#), + "carries the status in-band: {text}" + ); + assert!( + text.trim_end().ends_with("data: [DONE]"), + "terminated: {text}" + ); + } +}