refactor error responses into shared utils::response helpers (#33894)

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.

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<AppState> {
@@ -40,6 +40,12 @@ pub(super) fn routes() -> Router<AppState> {
         .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.

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<String>) -> 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<String>,
-    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<String>) -> 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>) -> 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<String>, 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<EgressItem>,
     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,

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;
             }

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<Str
         return Err(openai_error(
             StatusCode::SERVICE_UNAVAILABLE,
             "service unavailable",
+            false,
         ));
     };
     match rx.recv().await {
@@ -242,10 +256,11 @@ async fn decode_prompt_echo(state: &AppState, token_ids: TokenIds) -> Result<Str
             openai_error(
                 StatusCode::INTERNAL_SERVER_ERROR,
                 "detokenized prompt is not valid UTF-8",
+                false,
             )
         }),
         Some(EgressItem::Error(crate::error::Error::Validation(message))) => {
-            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<Str
             Err(openai_error(
                 status,
                 format!("failed to decode prompt for echo: {error}"),
+                false,
             ))
         }
         Some(_) | None => 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;
             }
 

rust/sglang-server/src/api_server/openai/models.rs

@@ -28,6 +28,7 @@ async fn retrieve_model(State(state): State<AppState>, Path(model): Path<String>
         return openai_error(
             StatusCode::NOT_FOUND,
             format!("The model `{model}` does not exist"),
+            false,
         );
     }
     Json(model_card(&state)).into_response()

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()));

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<EgressItem>), 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}"
-        );
-    }
-}

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;

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}"
+        );
+    }
+}