Skip to content
8 changes: 6 additions & 2 deletions lib/bindings/python/tests/test_http_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -187,7 +187,11 @@ async def test_chat_completion_http_error(http_server, msg_to_code: tuple[str, i
async with session.post(url, json=data) as response:
assert response.status == msg_to_code[1]
error_json = await response.json()
# Error responses are sanitized: the client receives a static
# message and never the raw backend error text. The underlying
# detail is logged server-side instead.
assert "internal server error" in str(error_json).lower()
if msg_to_code[0] == MSG_CONTAINS_ERROR:
assert MSG_CONTAINS_ERROR in str(error_json)
assert MSG_CONTAINS_ERROR not in str(error_json)
elif msg_to_code[0] == MSG_CONTAINS_INTERNAL_ERROR:
assert "simulated internal error" in str(error_json).lower()
assert "simulated internal error" not in str(error_json).lower()
77 changes: 54 additions & 23 deletions lib/llm/src/http/service/anthropic.rs
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,7 @@ use crate::request_template::{RequestTemplate, resolve_request_model};
use crate::types::Annotated;

// Re-use helpers from the openai module (sibling under service/)
use super::error::SanitizedError;
use super::metadata::extract_metadata_from_http;
use super::openai::{get_body_limit, get_or_create_request_id};

Expand Down Expand Up @@ -178,10 +179,9 @@ async fn handler_anthropic_messages(
tokio::spawn(anthropic_messages(state, template, request, stream_handle).in_current_span())
.await
.map_err(|e| {
anthropic_error(
StatusCode::INTERNAL_SERVER_ERROR,
"api_error",
&format!("Failed to await messages task: {:?}", e),
anthropic_sanitized_error_with_details(
SanitizedError::Internal,
format!("Failed to await Anthropic messages task: {e:?}"),
)
})?;

Expand Down Expand Up @@ -333,17 +333,15 @@ async fn anthropic_messages(
// Check for cancelled request (client disconnected before response was sent)
if super::metrics::request_was_cancelled(e.as_ref()) {
inflight_guard.mark_error(super::metrics::ErrorType::Cancelled);
return anthropic_error(
StatusCode::from_u16(499).unwrap(),
"request_cancelled",
&format!("Request cancelled: {}", e),
return anthropic_sanitized_error_with_details(
SanitizedError::Cancelled,
format!("{e:#}"),
);
}
inflight_guard.mark_error(super::metrics::ErrorType::Internal);
anthropic_error(
StatusCode::INTERNAL_SERVER_ERROR,
"api_error",
&format!("Failed to generate completions: {}", e),
anthropic_sanitized_error_with_details(
SanitizedError::Internal,
format!("Failed to generate Anthropic completions: {e}"),
)
})?;

Expand Down Expand Up @@ -446,12 +444,20 @@ async fn anthropic_messages(
// Check first event for backend errors using the openai helper
let stream_with_check = super::openai::check_for_backend_error(engine_stream)
.await
.map_err(|(status, json_err)| {
tracing::error!(request_id, %status, ?json_err, "Backend error detected");
anthropic_error(
StatusCode::INTERNAL_SERVER_ERROR,
"api_error",
"Backend error during generation",
.map_err(|(status, _json_err)| {
// check_for_backend_error has already sanitized the body and
// logged the backend detail; we just preserve its status when
// re-wrapping in Anthropic format.
let err = if status.as_u16() == 499 {
SanitizedError::Cancelled
} else if status.is_server_error() {
SanitizedError::PreserveServerError(status)
} else {
Comment thread
peilii marked this conversation as resolved.
Resolved
SanitizedError::Internal
};
anthropic_sanitized_error_with_details(
err,
format!("backend error event (status {})", status.as_u16()),
)
})?;

Expand All @@ -468,11 +474,9 @@ async fn anthropic_messages(
NvCreateChatCompletionResponse::from_annotated_stream(stream, parsing_options.clone())
.await
.map_err(|e| {
tracing::error!(request_id, "Failed to fold messages stream: {:?}", e);
anthropic_error(
StatusCode::INTERNAL_SERVER_ERROR,
"api_error",
&format!("Failed to fold messages stream: {}", e),
anthropic_sanitized_error_with_details(
SanitizedError::Internal,
format!("Failed to fold messages stream: {e:?}"),
)
})?;

Expand Down Expand Up @@ -734,6 +738,33 @@ fn estimate_input_tokens(req: &AnthropicCreateMessageRequest) -> u32 {
count_req.estimate_tokens()
}

/// Build an Anthropic-formatted error response from a canonical
/// [`SanitizedError`] variant. The status, public message, and Anthropic
/// `error_type` all come from the variant; `details` are logged
/// server-side but never reach the client.
fn anthropic_sanitized_error_with_details(
err: SanitizedError,
details: impl std::fmt::Display,
) -> Response {
let status = err.status();
if err.log_as_error() {
tracing::error!(status = %status, "Anthropic {err}: {details}");
} else {
tracing::debug!(status = %status, "Anthropic {err}: {details}");
}
(
status,
Json(AnthropicErrorResponse {
object_type: "error".to_string(),
error: AnthropicErrorBody {
error_type: err.anthropic_type().to_string(),
message: err.to_string(),
},
}),
)
.into_response()
}

/// Build an Anthropic-formatted error response.
/// Maps HTTP status codes to Anthropic error types following the Anthropic API spec.
fn anthropic_error(status: StatusCode, error_type: &str, message: &str) -> Response {
Expand Down
71 changes: 52 additions & 19 deletions lib/llm/src/http/service/disconnect.rs
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@ use futures::{Stream, StreamExt};
use std::sync::Arc;
use std::time::Duration;

use crate::http::service::error::SanitizedError;
use crate::http::service::metrics::{CancellationLabels, ErrorType, InflightGuard, Metrics};

use dynamo_runtime::config::environment_names::llm::DYN_HTTP_BACKEND_STREAM_TIMEOUT_SECS as BACKEND_STREAM_TIMEOUT_ENV;
Expand Down Expand Up @@ -226,13 +227,17 @@ pub fn monitor_for_disconnects(
// doesn't later record this as ClosedUnexpectedly (which
// would mis-attribute the fault as a client disconnect).
stream_handle.disarm();
// DIS-1768: emit structured OpenAI-style error frame + `data: [DONE]`
// so naive `data:`-line parsers see both the error and a stream terminator.
tracing::error!("Streaming error: {err}");
// Emit a structured OpenAI-style error frame + `data: [DONE]`
// so naive `data:`-line parsers see both the error and a
// stream terminator. Body derived from SanitizedError so
// the sanitized message + status live in one place.
let sanitized = SanitizedError::Internal;
let err_json = serde_json::json!({
"error": {
"message": err.to_string(),
"type": "internal_server_error",
"code": 500,
"message": sanitized.to_string(),
"type": sanitized.openai_type_slug(),
"code": sanitized.status().as_u16(),
}
});
yield Event::default().data(err_json.to_string());
Expand Down Expand Up @@ -508,7 +513,7 @@ mod tests {
}

// ─────────────────────────────────────────────────────────────────────────────
// DIS-1768: mid-stream fault SSE contract
// mid-stream fault SSE contract
//
// When the upstream stream yields `Err(_)` mid-stream — e.g. an upstream
// worker dies and the mpsc channel reports
Expand Down Expand Up @@ -550,9 +555,11 @@ mod tests {
String::from_utf8(body.to_vec()).expect("utf8 body")
}

/// Assert the post-fix SSE fault contract: parsed structured error frame with exact
/// message/type/code, positioned before `[DONE]`, and no bare `event: error` trailer.
fn assert_fault_contract(case: &str, text: &str, expected_message: &str) {
/// Assert the post-fix SSE fault contract: a parsed structured error frame
/// carrying the sanitized static message/type/code, positioned before
/// `[DONE]`, with no bare `event: error` trailer, and crucially with no trace
/// of `leaked_detail` (the raw backend error) anywhere in the body.
fn assert_fault_contract(case: &str, text: &str, leaked_detail: &str) {
let done_pos = text.find("data: [DONE]").unwrap_or_else(|| {
panic!("[{case}] body does not terminate with `data: [DONE]`. Body:\n{text}")
});
Expand Down Expand Up @@ -582,25 +589,32 @@ mod tests {
.get("error")
.and_then(|v| v.as_object())
.unwrap_or_else(|| panic!("[{case}] `error` field is not an object. Body:\n{text}"));
let expected = SanitizedError::Internal;
let expected_message = expected.to_string();
assert_eq!(
error.get("message").and_then(|v| v.as_str()),
Some(expected_message),
"[{case}] structured error `message` mismatch. Body:\n{text}"
Some(expected_message.as_str()),
"[{case}] structured error `message` must be the sanitized static string. Body:\n{text}"
);
assert_eq!(
error.get("type").and_then(|v| v.as_str()),
Some("internal_server_error"),
Some(expected.openai_type_slug()),
"[{case}] structured error `type` mismatch. Body:\n{text}"
);
assert_eq!(
error.get("code").and_then(|v| v.as_i64()),
Some(500),
Some(i64::from(expected.status().as_u16())),
"[{case}] structured error `code` mismatch. Body:\n{text}"
);
assert!(
!text.contains("event: error\n: "),
"[{case}] body contains bare `event: error\\n: <comment>` trailer (pre-fix bug). Body:\n{text}"
);
assert!(
!text.contains(leaked_detail),
"[{case}] SSE body leaked raw backend error detail to the client. \
Expected `{leaked_detail}` to be absent. Body:\n{text}"
);
}

/// Upstream worker killed mid-stream → mpsc channel reports `Disconnected` to the
Expand All @@ -609,12 +623,12 @@ mod tests {
#[serial]
async fn test_simulate_worker_kill_emits_structured_error_and_done() {
let (_metrics, guard, ctx, handle) = setup_test("worker-kill-model", "req-wk", "0");
let expected_message = "Disconnected: Stream ended before generation completed";
let stream = simulate_mid_stream_error(3, expected_message);
let backend_detail = "Disconnected: Stream ended before generation completed";
let stream = simulate_mid_stream_error(3, backend_detail);
let monitored = monitor_for_disconnects(stream, ctx, guard, handle);
let body = collect_sse_body(monitored).await;
cleanup_env();
assert_fault_contract("worker_kill", &body, expected_message);
assert_fault_contract("worker_kill", &body, backend_detail);
}

/// Python chat-processor raises mid-stream → Rust→Python `tx.send()` fails with
Expand All @@ -623,11 +637,30 @@ mod tests {
#[serial]
async fn test_simulate_python_consumer_drop_emits_structured_error_and_done() {
let (_metrics, guard, ctx, handle) = setup_test("py-drop-model", "req-py", "0");
let expected_message = "Failed to send response: SendError { .. }";
let stream = simulate_mid_stream_error(3, expected_message);
let backend_detail = "Failed to send response: SendError { .. }";
let stream = simulate_mid_stream_error(3, backend_detail);
let monitored = monitor_for_disconnects(stream, ctx, guard, handle);
let body = collect_sse_body(monitored).await;
cleanup_env();
assert_fault_contract("python_consumer_drop", &body, backend_detail);
}

/// A backend error carrying sensitive internals (file paths, panic text,
/// Python exception details) MUST NOT reach the streaming client. The client
/// receives only the sanitized static frame; the detail stays server-side.
#[tokio::test]
#[serial]
async fn test_mid_stream_error_does_not_leak_internal_details() {
let (_metrics, guard, ctx, handle) = setup_test("leak-model", "req-leak", "0");
let backend_detail = "panicked at '/opt/dynamo/lib/python3.12/site-packages/engine/worker.py:512: ValueError: secret tensor shape mismatch'";
let stream = simulate_mid_stream_error(2, backend_detail);
let monitored = monitor_for_disconnects(stream, ctx, guard, handle);
let body = collect_sse_body(monitored).await;
cleanup_env();
assert_fault_contract("python_consumer_drop", &body, expected_message);
assert_fault_contract("internal_detail_leak", &body, backend_detail);
// Spot-check the most damaging fragments explicitly.
assert!(!body.contains("site-packages"), "leaked a filesystem path");
assert!(!body.contains("panicked at"), "leaked panic text");
assert!(!body.contains("ValueError"), "leaked exception type");
}
}
82 changes: 82 additions & 0 deletions lib/llm/src/http/service/error.rs
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
// SPDX-FileCopyrightText: Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0

use axum::http::StatusCode;
use thiserror::Error;

/// Implementation of the Completion Engines served by the HTTP service should
Expand All @@ -12,3 +13,84 @@ pub struct HttpError {
pub code: u16,
pub message: String,
}

/// Admission-control rejection hint appended to overload responses.
pub const ADMISSION_CONTROL_REJECTION_HINT: &str =
"If this rejection is not intended, consider passing --admission-control none to the frontend.";

/// Canonical sanitized error responses returned at the HTTP boundary.
///
/// Each variant fixes the `(status, public message, protocol error_type)`
/// triple so call sites stop duplicating literals. The protocol-specific
/// mappings (OpenAI `error_type` string, Anthropic `error_type`) and the
/// `Display` impl that produces the user-safe message all live on this
/// enum — clients see exactly what the enum says, never a backend error
/// chain, file path, or panic stack.
#[derive(Debug, Clone, Copy)]
pub enum SanitizedError {
/// 499 Client Closed Request.
Cancelled,
/// 503 Service Unavailable, with admission-control hint.
Comment thread
peilii marked this conversation as resolved.
Resolved
Overloaded,
/// 500 Internal Server Error.
Internal,
/// Preserve a backend-reported 5xx status code while replacing the
/// body with the generic internal-error message. Clients still see
/// the original status (so 503 retry semantics survive); only the
/// payload is sanitized.
PreserveServerError(StatusCode),
}

impl SanitizedError {
pub fn status(self) -> StatusCode {
match self {
// 499 is not IANA-registered but is widely used (nginx).
SanitizedError::Cancelled => StatusCode::from_u16(499).unwrap(),
SanitizedError::Overloaded => StatusCode::SERVICE_UNAVAILABLE,
SanitizedError::Internal => StatusCode::INTERNAL_SERVER_ERROR,
SanitizedError::PreserveServerError(code) => code,
}
}

/// Anthropic `error.type` for this category.
pub fn anthropic_type(self) -> &'static str {
match self {
SanitizedError::Cancelled => "request_cancelled",
SanitizedError::Overloaded => "overloaded_error",
SanitizedError::Internal | SanitizedError::PreserveServerError(_) => "api_error",
}
}

/// OpenAI-style snake_case `type` field used in inline error frames.
pub fn openai_type_slug(self) -> &'static str {
match self {
SanitizedError::Cancelled => "request_cancelled",
SanitizedError::Overloaded => "service_unavailable",
SanitizedError::Internal | SanitizedError::PreserveServerError(_) => {
"internal_server_error"
}
}
}

/// Whether to log this category at `error!` (true) or `debug!` (false).
/// Cancellations are client-driven and routinely fire on disconnect, so
/// they stay at debug to avoid drowning real errors.
pub fn log_as_error(self) -> bool {
!matches!(self, SanitizedError::Cancelled)
}
}

impl std::fmt::Display for SanitizedError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
SanitizedError::Cancelled => f.write_str("Request cancelled"),
SanitizedError::Overloaded => write!(
f,
"Service temporarily overloaded. {ADMISSION_CONTROL_REJECTION_HINT}"
),
SanitizedError::Internal | SanitizedError::PreserveServerError(_) => {
f.write_str("Internal server error")
}
}
}
}
Loading
Loading