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()
11 changes: 7 additions & 4 deletions lib/llm/src/http/service/anthropic.rs
Original file line number Diff line number Diff line change
Expand Up @@ -178,10 +178,11 @@ async fn handler_anthropic_messages(
tokio::spawn(anthropic_messages(state, template, request, stream_handle).in_current_span())
.await
.map_err(|e| {
tracing::error!("Failed to await Anthropic messages task: {e:?}");
anthropic_error(
StatusCode::INTERNAL_SERVER_ERROR,
"api_error",
&format!("Failed to await messages task: {:?}", e),
"Internal server error",
Comment thread
peilii marked this conversation as resolved.
Resolved
)
})?;

Expand Down Expand Up @@ -333,17 +334,19 @@ 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);
tracing::debug!("Anthropic request cancelled before response: {e:#}");
Comment thread
peilii marked this conversation as resolved.
Resolved
return anthropic_error(
StatusCode::from_u16(499).unwrap(),
Comment thread
peilii marked this conversation as resolved.
Resolved
"request_cancelled",
&format!("Request cancelled: {}", e),
"Request cancelled",
);
}
inflight_guard.mark_error(super::metrics::ErrorType::Internal);
tracing::error!("Failed to generate Anthropic completions: {e}");
anthropic_error(
Comment thread
peilii marked this conversation as resolved.
Resolved
StatusCode::INTERNAL_SERVER_ERROR,
"api_error",
&format!("Failed to generate completions: {}", e),
"Internal server error",
)
})?;

Expand Down Expand Up @@ -472,7 +475,7 @@ async fn anthropic_messages(
anthropic_error(
StatusCode::INTERNAL_SERVER_ERROR,
"api_error",
&format!("Failed to fold messages stream: {}", e),
"Internal server error",
)
})?;

Expand Down
62 changes: 47 additions & 15 deletions lib/llm/src/http/service/disconnect.rs
Original file line number Diff line number Diff line change
Expand Up @@ -226,11 +226,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.
// Log the underlying error server-side; the client only
Comment thread
peilii marked this conversation as resolved.
Resolved
// receives a sanitized message so streaming responses don't
// leak internal paths, panic text, or backend exception
// details.
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.
let err_json = serde_json::json!({
"error": {
"message": err.to_string(),
"message": "Internal server error",
"type": "internal_server_error",
"code": 500,
}
Expand Down Expand Up @@ -508,7 +514,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 +556,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 @@ -584,8 +592,8 @@ mod tests {
.unwrap_or_else(|| panic!("[{case}] `error` field is not an object. Body:\n{text}"));
assert_eq!(
error.get("message").and_then(|v| v.as_str()),
Some(expected_message),
"[{case}] structured error `message` mismatch. Body:\n{text}"
Some("Internal server error"),
Comment thread
peilii marked this conversation as resolved.
Resolved
"[{case}] structured error `message` must be the sanitized static string. Body:\n{text}"
);
assert_eq!(
error.get("type").and_then(|v| v.as_str()),
Expand All @@ -601,6 +609,11 @@ mod tests {
!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 +622,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 +636,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");
}
}
Loading
Loading