diff --git a/CHANGELOG.md b/CHANGELOG.md index 1f8af0ce..17f5f09f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,8 @@ All notable changes to Agentic API are documented here. - Bounded WebSocket queues, generated response data, gateway tool results, and MCP discovery and HTTP/SSE payloads so concurrent response streams cannot grow memory without limit. +- Preserved upstream error metadata (`retry-after`, request IDs, rate-limit headers) and the upstream content type on + the non-streaming Responses executor path, sharing one upstream-error adapter with the Messages handler (#250). - Rejected split-execution responses with missing, reused, or unstable tool call IDs before persistence, keeping the reserved response ID available for a corrected retry. - Hardened split execution with atomic duplicate persistence, strict relayed-response validation, independent secret diff --git a/crates/agentic-server/src/handler/common.rs b/crates/agentic-server/src/handler/common.rs index 2c0d0736..c2936d71 100644 --- a/crates/agentic-server/src/handler/common.rs +++ b/crates/agentic-server/src/handler/common.rs @@ -25,13 +25,32 @@ pub fn convert_response(resp: ProxyResponse) -> Response { } } +/// Forward an upstream error verbatim: its status, body, and processed metadata headers. +/// +/// The headers were already filtered by `processed_response_headers` when the inference +/// call captured them, so hop-by-hop and stale representation headers are gone while +/// `retry-after`, request IDs, and rate-limit metadata remain. `content-type` is only +/// defaulted to JSON when the upstream did not label its body, so a plain-text error is +/// not relabelled as JSON. +pub fn upstream_error_response(status: StatusCode, body: String, mut headers: HeaderMap) -> Response { + headers + .entry(http::header::CONTENT_TYPE) + .or_insert(http::HeaderValue::from_static("application/json")); + convert_response(ProxyResponse { + status, + headers, + body: ProxyBody::Full(Bytes::from(body)), + }) +} + /// # Panics /// Panics if the response builder produces an invalid response (unreachable in practice). pub fn executor_error_response(err: ExecutorError) -> Response { - let status = err.http_status(); - if !matches!(err, ExecutorError::LLMRequest { .. }) { - warn!("executor error ({status}): {err}"); + if let ExecutorError::LLMRequest { status, body, headers } = err { + return upstream_error_response(status, body, headers); } + let status = err.http_status(); + warn!("executor error ({status}): {err}"); Response::builder() .status(status) .header("Content-Type", "application/json") diff --git a/crates/agentic-server/src/handler/http/messages.rs b/crates/agentic-server/src/handler/http/messages.rs index 42ac6d9f..cd471ded 100644 --- a/crates/agentic-server/src/handler/http/messages.rs +++ b/crates/agentic-server/src/handler/http/messages.rs @@ -11,13 +11,14 @@ use agentic_core::executor::{ validate_native_web_search_request, }; use agentic_core::proxy::{ - ProxyAuth, ProxyBody, ProxyRequest, ProxyResponse, error_response_for_auth, proxy_request_with_path, - upstream_request_headers, + ProxyAuth, ProxyRequest, error_response_for_auth, proxy_request_with_path, upstream_request_headers, }; use agentic_core::tool::ToolRegistry; use agentic_core::types::messages::{MessagesRequest, has_gateway_tool, registry_tools}; -use super::super::common::{convert_response, read_bytes_with_auth, sse_response_with_headers}; +use super::super::common::{ + convert_response, read_bytes_with_auth, sse_response_with_headers, upstream_error_response, +}; use crate::app::AppState; async fn proxy_messages( @@ -44,20 +45,8 @@ async fn proxy_messages( /// Preserve upstream Messages errors verbatim; render local executor failures /// as an Anthropic error envelope, consistent with the proxy path (E14). fn messages_error_response(err: ExecutorError) -> Response { - if let ExecutorError::LLMRequest { - status, - body, - mut headers, - } = err - { - headers - .entry(http::header::CONTENT_TYPE) - .or_insert(http::HeaderValue::from_static("application/json")); - return convert_response(ProxyResponse { - status, - headers, - body: ProxyBody::Full(Bytes::from(body)), - }); + if let ExecutorError::LLMRequest { status, body, headers } = err { + return upstream_error_response(status, body, headers); } convert_response(error_response_for_auth( err.http_status(), diff --git a/crates/agentic-server/tests/responses_error_headers_test.rs b/crates/agentic-server/tests/responses_error_headers_test.rs new file mode 100644 index 00000000..3efcac41 --- /dev/null +++ b/crates/agentic-server/tests/responses_error_headers_test.rs @@ -0,0 +1,171 @@ +//! Upstream error metadata must survive the non-streaming Responses executor path. +//! +//! Regression coverage for : when a +//! request is routed through the in-process executor (for example `store: true`) and the +//! upstream returns a non-2xx response, the gateway must preserve the upstream status, body, +//! and processed metadata headers (`retry-after`, request IDs, rate-limit headers) and must not +//! relabel a non-JSON error body as `application/json`. + +// `common` is compiled into every test binary; this one never calls `spawn_mock_llm`, +// so silence the resulting dead-code warning (same as the other tests that skip it). +#[allow(dead_code)] +mod common; + +use axum::Router; +use axum::response::{IntoResponse, Response}; +use axum::routing::post; +use http::{HeaderMap, StatusCode}; +use tokio::net::TcpListener; + +use common::{spawn_gateway, test_config, test_state}; + +const NON_STREAMING_STORED_REQUEST: &str = r#"{"model":"test","input":"hi","store":true,"stream":false}"#; +const NON_STREAMING_PROXIED_REQUEST: &str = r#"{"model":"test","input":"hi","store":false,"stream":false}"#; + +/// Spawn a mock upstream whose `POST /v1/responses` always answers with the given error. +async fn spawn_error_upstream( + status: StatusCode, + content_type: &'static str, + body: &'static str, + extra_headers: HeaderMap, +) -> (String, tokio::task::JoinHandle<()>) { + let app = Router::new().route( + "/v1/responses", + post(move || { + let extra_headers = extra_headers.clone(); + async move { + let mut response = Response::builder() + .status(status) + .header("content-type", content_type) + .body(axum::body::Body::from(body)) + .unwrap(); + response.headers_mut().extend(extra_headers); + response.into_response() + } + }), + ); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let handle = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + (format!("http://{addr}"), handle) +} + +fn rate_limit_headers() -> HeaderMap { + let mut headers = HeaderMap::new(); + headers.insert("retry-after", "7".parse().unwrap()); + headers.insert("x-request-id", "req_example".parse().unwrap()); + headers.insert("x-ratelimit-remaining-requests", "0".parse().unwrap()); + headers +} + +async fn post_responses(gateway_url: &str, body: &'static str) -> reqwest::Response { + reqwest::Client::new() + .post(format!("{gateway_url}/v1/responses")) + .header("content-type", "application/json") + .body(body) + .send() + .await + .unwrap() +} + +#[tokio::test] +async fn executor_path_preserves_upstream_json_error_metadata() { + let upstream_body = r#"{"error":{"message":"rate limited","type":"rate_limit_error"}}"#; + let (llm_url, _upstream) = spawn_error_upstream( + StatusCode::TOO_MANY_REQUESTS, + "application/json", + upstream_body, + rate_limit_headers(), + ) + .await; + let (gateway_url, _gateway) = spawn_gateway(test_state(&test_config(&llm_url))).await; + + let response = post_responses(&gateway_url, NON_STREAMING_STORED_REQUEST).await; + + assert_eq!(response.status(), StatusCode::TOO_MANY_REQUESTS); + assert_eq!(response.headers()["retry-after"], "7"); + assert_eq!(response.headers()["x-request-id"], "req_example"); + assert_eq!(response.headers()["x-ratelimit-remaining-requests"], "0"); + assert_eq!(response.headers()["content-type"], "application/json"); + assert_eq!(response.text().await.unwrap(), upstream_body); +} + +#[tokio::test] +async fn executor_path_preserves_upstream_plain_text_error_content_type() { + let upstream_body = "rate limited"; + let (llm_url, _upstream) = spawn_error_upstream( + StatusCode::TOO_MANY_REQUESTS, + "text/plain", + upstream_body, + rate_limit_headers(), + ) + .await; + let (gateway_url, _gateway) = spawn_gateway(test_state(&test_config(&llm_url))).await; + + let response = post_responses(&gateway_url, NON_STREAMING_STORED_REQUEST).await; + + assert_eq!(response.status(), StatusCode::TOO_MANY_REQUESTS); + assert_eq!(response.headers()["retry-after"], "7"); + assert_eq!(response.headers()["x-request-id"], "req_example"); + assert_eq!(response.headers()["content-type"], "text/plain"); + assert_eq!(response.text().await.unwrap(), upstream_body); +} + +#[tokio::test] +async fn executor_path_still_filters_connection_nominated_headers() { + let mut headers = rate_limit_headers(); + headers.insert("connection", "x-upstream-hop".parse().unwrap()); + headers.insert("x-upstream-hop", "1".parse().unwrap()); + let (llm_url, _upstream) = + spawn_error_upstream(StatusCode::TOO_MANY_REQUESTS, "text/plain", "rate limited", headers).await; + let (gateway_url, _gateway) = spawn_gateway(test_state(&test_config(&llm_url))).await; + + let response = post_responses(&gateway_url, NON_STREAMING_STORED_REQUEST).await; + + assert_eq!(response.status(), StatusCode::TOO_MANY_REQUESTS); + assert_eq!(response.headers()["retry-after"], "7"); + assert!(response.headers().get("x-upstream-hop").is_none()); +} + +/// Control: the stateless proxy path already preserved these headers and must keep doing so. +#[tokio::test] +async fn proxy_path_preserves_upstream_error_metadata() { + let upstream_body = "rate limited"; + let (llm_url, _upstream) = spawn_error_upstream( + StatusCode::TOO_MANY_REQUESTS, + "text/plain", + upstream_body, + rate_limit_headers(), + ) + .await; + let (gateway_url, _gateway) = spawn_gateway(test_state(&test_config(&llm_url))).await; + + let response = post_responses(&gateway_url, NON_STREAMING_PROXIED_REQUEST).await; + + assert_eq!(response.status(), StatusCode::TOO_MANY_REQUESTS); + assert_eq!(response.headers()["retry-after"], "7"); + assert_eq!(response.headers()["x-request-id"], "req_example"); + assert_eq!(response.headers()["content-type"], "text/plain"); + assert_eq!(response.text().await.unwrap(), upstream_body); +} + +/// Control: gateway-originated errors keep their JSON envelope and content type. +#[tokio::test] +async fn malformed_client_json_still_returns_json_error_envelope() { + let (llm_url, _upstream) = spawn_error_upstream( + StatusCode::TOO_MANY_REQUESTS, + "text/plain", + "unreachable", + rate_limit_headers(), + ) + .await; + let (gateway_url, _gateway) = spawn_gateway(test_state(&test_config(&llm_url))).await; + + let response = post_responses(&gateway_url, r#"{"model":"test","input":"hi","store":true"#).await; + + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + assert_eq!(response.headers()["content-type"], "application/json"); + assert!(response.headers().get("retry-after").is_none()); + let body: serde_json::Value = response.json().await.unwrap(); + assert_eq!(body["error"]["type"], "invalid_request_error"); +}