diff --git a/proxy/src/proxy.rs b/proxy/src/proxy.rs index 66167530b..5b890f73d 100644 --- a/proxy/src/proxy.rs +++ b/proxy/src/proxy.rs @@ -665,6 +665,62 @@ mod tests { ); } + #[tokio::test] + async fn safe_upstream_errors_reach_local_chat_callers_without_retrying() { + for (status, code) in [ + (StatusCode::BAD_REQUEST, "upstream_invalid_request"), + (StatusCode::BAD_REQUEST, "upstream_context_limit"), + (StatusCode::BAD_GATEWAY, "upstream_provider_error"), + (StatusCode::GATEWAY_TIMEOUT, "upstream_timeout"), + ] { + for stream in [false, true] { + let error_body = Bytes::from( + serde_json::to_vec(&serde_json::json!({ + "status": status.as_u16(), + "message": "Safe upstream error fixture" + })) + .unwrap(), + ); + let transport = Arc::new(MockTransport::new(vec![Ok(raw_response( + status, + &[ + ("content-type", "application/json"), + ("x-opensecret-error-contract", "1"), + ("x-opensecret-error-code", code), + ], + vec![error_body.clone()], + ))])); + let request_body = serde_json::to_vec(&serde_json::json!({ + "model": "fixture-model", "stream": stream + })) + .unwrap(); + let response = mock_app(transport.clone()) + .oneshot( + AxumRequest::builder() + .method(Method::POST) + .uri("/v1/chat/completions") + .header(header::CONTENT_TYPE, "application/json") + .body(Body::from(request_body)) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(response.status(), status); + assert_eq!(response.headers()["x-opensecret-error-contract"], "1"); + assert_eq!(response.headers()["x-opensecret-error-code"], code); + assert!(!response + .headers() + .contains_key("x-opensecret-client-replay")); + assert_eq!( + to_bytes(response.into_body(), 1024).await.unwrap(), + error_body + ); + assert_eq!(transport.take_requests().len(), 1); + } + } + } + #[tokio::test(start_paused = true)] async fn response_start_timeout_is_gateway_timeout() { let mut config = test_config(); diff --git a/sdk/rust/src/client/recovery_tests.rs b/sdk/rust/src/client/recovery_tests.rs index 1538b9247..05cc9e869 100644 --- a/sdk/rust/src/client/recovery_tests.rs +++ b/sdk/rust/src/client/recovery_tests.rs @@ -698,6 +698,82 @@ async fn recovery_does_not_replay_authenticated_logical_errors() { } } +#[tokio::test] +async fn safe_upstream_errors_preserve_the_inference_contract_without_recovery() { + for (status, code) in [ + (400, "upstream_invalid_request"), + (400, "upstream_context_limit"), + (502, "upstream_provider_error"), + (504, "upstream_timeout"), + ] { + for stream in [false, true] { + let server = MockServer::start().await; + let state = TestV2ServerState::new(); + let body = serde_json::json!({ + "status": status, + "message": "Safe upstream error fixture" + }); + state.queue_json_response_with_headers( + status, + body.clone(), + &[ + ("x-opensecret-error-contract", "1"), + ("x-opensecret-error-code", code), + ], + ); + mount_sessions(&server, &state, 1, false).await; + Mock::given(method("POST")) + .and(path("/v2/request")) + .respond_with(state.request_responder()) + .expect(1) + .mount(&server) + .await; + let client = + OpenSecretClient::new_with_api_key(server.uri(), "fixture-key".to_string()) + .unwrap(); + client + .set_tokens( + "fixture-access".to_string(), + Some("fixture-refresh".to_string()), + ) + .unwrap(); + let response = client + .send_inference_request( + HttpRequest::post("/v1/chat/completions") + .header(header::CONTENT_TYPE, "application/json") + .body(Bytes::from( + serde_json::to_vec(&serde_json::json!({ + "model": "fixture-model", "stream": stream + })) + .unwrap(), + )) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status().as_u16(), status); + assert_eq!(response.headers()["x-opensecret-error-contract"], "1"); + assert_eq!(response.headers()["x-opensecret-error-code"], code); + assert!(!response + .headers() + .contains_key("x-opensecret-client-replay")); + assert_eq!( + collect_response_body(response.into_body()).await.unwrap(), + Bytes::from(serde_json::to_vec(&body).unwrap()) + ); + assert_eq!(state.captured_requests().len(), 1); + assert!(client.get_session_id().unwrap().is_some()); + let tokens = client.get_tokens().unwrap().unwrap(); + assert_eq!(tokens.access_token, "fixture-access"); + assert_eq!(tokens.refresh_token.as_deref(), Some("fixture-refresh")); + assert!(!state + .captured_requests() + .iter() + .any(|request| request["target"] == "/refresh")); + } + } +} + #[tokio::test] async fn recovery_does_not_follow_request_redirects() { let server = MockServer::start().await; diff --git a/sdk/rust/src/transport_v2/client.rs b/sdk/rust/src/transport_v2/client.rs index 81baa29bd..06498949a 100644 --- a/sdk/rust/src/transport_v2/client.rs +++ b/sdk/rust/src/transport_v2/client.rs @@ -597,6 +597,24 @@ mod tests { }); } + pub(crate) fn queue_json_response_with_headers( + &self, + status: u16, + body: serde_json::Value, + headers: &[(&str, &str)], + ) { + self.queue_json_response(status, body); + self.responses + .lock() + .unwrap() + .back_mut() + .unwrap() + .headers + .extend(headers.iter().map(|(name, value)| { + LogicalHeader::new((*name).to_string(), (*value).to_string()).unwrap() + })); + } + pub(crate) fn queue_delayed_json_response( &self, status: u16, diff --git a/sdk/src/lib/test/encryptedApi.test.ts b/sdk/src/lib/test/encryptedApi.test.ts index 3b9d4fb29..f67204a25 100644 --- a/sdk/src/lib/test/encryptedApi.test.ts +++ b/sdk/src/lib/test/encryptedApi.test.ts @@ -493,6 +493,50 @@ describe("simplified Transport V2 encrypted API seam", () => { }); } + test("OpenAI calls retain safe upstream messages, statuses, and codes", async () => { + for (const [status, code] of [ + [400, "upstream_invalid_request"], + [400, "upstream_context_limit"], + [502, "upstream_provider_error"], + [504, "upstream_timeout"] + ] as const) { + for (const target of ["/v1/chat/completions", "/v1/responses"]) { + const testHarness = harness(async () => + exchange( + jsonResponse( + { status, message: "Safe upstream error fixture" }, + { + status, + headers: { + "x-opensecret-error-contract": "1", + "x-opensecret-error-code": code + } + } + ) + ) + ); + let failure: unknown; + try { + await openAiAuthenticatedApiCallWithDependencies<{ model: string }, unknown>( + `${appApiUrl}${target}`, + "POST", + { model: "fixture-model" }, + undefined, + "fixture-api-key", + testHarness.dependencies + ); + } catch (error) { + failure = error; + } + expect(failure).toBeInstanceOf(Error); + expect(failure).toMatchObject({ status, message: "Safe upstream error fixture" }); + expect((failure as { headers: Headers }).headers.get("x-opensecret-error-code")).toBe(code); + expect(testHarness.request).toHaveBeenCalledTimes(1); + expect(testHarness.authority).not.toHaveBeenCalled(); + } + } + }); + test("surfaces authenticated logical errors and notifies user auth without retrying", async () => { const logicalError = jsonResponse( { message: "operation denied" }, diff --git a/sdk/src/lib/test/transportV2AuthRuntime.test.ts b/sdk/src/lib/test/transportV2AuthRuntime.test.ts index 1c2c2365e..404d7bbe1 100644 --- a/sdk/src/lib/test/transportV2AuthRuntime.test.ts +++ b/sdk/src/lib/test/transportV2AuthRuntime.test.ts @@ -281,6 +281,46 @@ describe("Transport V2 authentication runtime", () => { expect(harness.request).toHaveBeenCalledTimes(1); }); + test("safe upstream errors do not refresh or clear user credentials", async () => { + const api = apiUrl("upstream-errors"); + const initial = credentialPair("user", "user-1", NOW + 3_600, 7); + const harness = runtimeWith(async () => { + throw new Error("upstream errors must not trigger authentication requests"); + }); + const auth = new TransportV2AuthRuntime({ + runtime: harness.runtime, + nowUnixSeconds: () => NOW + }); + const installed = installTransportV2Credentials(api, "user", initial.access, initial.refresh); + const sent = await auth.authority(api, undefined, "user"); + + for (const [status, code] of [ + [400, "upstream_invalid_request"], + [400, "upstream_context_limit"], + [502, "upstream_provider_error"], + [504, "upstream_timeout"] + ] as const) { + auth.noteResponse( + Response.json( + { status, message: "Safe upstream error fixture" }, + { + status, + headers: { + "x-opensecret-error-contract": "1", + "x-opensecret-error-code": code + } + } + ), + api, + undefined, + "user", + sent + ); + expect(readTransportV2Credentials(api, "user")).toEqual(installed); + } + expect(harness.request).not.toHaveBeenCalled(); + }); + test("keeps an authenticated expiry response and refreshes only for a later request", async () => { const api = apiUrl("post-response"); const initial = credentialPair("user", "user-1", NOW + 3_600, 7); diff --git a/sdk/src/lib/test/transportV2Recovery.test.ts b/sdk/src/lib/test/transportV2Recovery.test.ts index 3d49355e6..f2c1a251c 100644 --- a/sdk/src/lib/test/transportV2Recovery.test.ts +++ b/sdk/src/lib/test/transportV2Recovery.test.ts @@ -66,6 +66,7 @@ async function responseFor( options: { status?: number; logicalHeaders?: { name: string; value: string }[]; + body?: string; truncate?: boolean; corrupt?: boolean; } = {} @@ -75,7 +76,7 @@ async function responseFor( new Uint8Array([1]), utf8(JSON.stringify({ status: options.status ?? 200, headers: options.logicalHeaders ?? [] })) ), - concatBytes(new Uint8Array([2]), utf8("ok")) + concatBytes(new Uint8Array([2]), utf8(options.body ?? "ok")) ]; if (!options.truncate) records.push(new Uint8Array([3])); const frames = await Promise.all( @@ -253,6 +254,45 @@ describe("Transport V2 inference budget across session repair", () => { beforeEach(() => globalThis.sessionStorage.clear()); +describe("Transport V2 safe upstream error contract", () => { + for (const [status, code] of [ + [400, "upstream_invalid_request"], + [400, "upstream_context_limit"], + [502, "upstream_provider_error"], + [504, "upstream_timeout"] + ] as const) { + for (const target of ["/v1/chat/completions", "/v1/responses"]) { + test(`${target} preserves ${code} without session repair or replay`, async () => { + const body = JSON.stringify({ status, message: "Safe upstream error fixture" }); + const fixture = await harness((attempt) => + responseFor(attempt, { + status, + body, + logicalHeaders: [ + { name: "content-type", value: "application/json" }, + { name: CONTRACT, value: "1" }, + { name: CODE, value: code } + ] + }) + ); + const response = await inferenceFetch(fixture.runtime)(`${API_URL}${target}`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "fixture-model", stream: true }) + }); + + expect(response.status).toBe(status); + expect(response.headers.get(CONTRACT)).toBe("1"); + expect(response.headers.get(CODE)).toBe(code); + expect(response.headers.has("x-opensecret-client-replay")).toBe(false); + expect(await response.text()).toBe(body); + expect(fixture.attempts).toHaveLength(1); + expect(fixture.establish).toHaveBeenCalledTimes(1); + }); + } + } +}); + describe("Transport V2 bounded session repair", () => { for (const code of CODES) { test(`repairs ${code} once with fresh transport keys and the exact logical request`, async () => { diff --git a/services/opensecret/docs/transport-v2-protocol.md b/services/opensecret/docs/transport-v2-protocol.md index cfcac086d..3d9b289c6 100644 --- a/services/opensecret/docs/transport-v2-protocol.md +++ b/services/opensecret/docs/transport-v2-protocol.md @@ -453,6 +453,45 @@ do not receive these recovery markers. These outer headers are unauthenticated hints, not evidence the original request never executed; see the retry rules below. +### Completion provider errors + +Before Chat Completions or Responses starts its public response, provider errors +use the existing `{"status":400,"message":"..."}` shape with static messages, +`x-opensecret-error-contract: 1`, and a static `x-opensecret-error-code`. V2 +authenticates the logical status, headers and body; V1 uses its existing HTTP +error representation with the classified status/message. + +| Failure | Public HTTP status | Error code | +| --- | --- | --- | +| Provider rejects a request with 400 or 422 | 400 or 422 respectively | `upstream_invalid_request`, or a recognized detail below | +| Provider rejects payload size with 413 | 413 | `upstream_payload_too_large` | +| Provider 408/504 or response-start timeout | 504 | `upstream_timeout` | +| Other provider HTTP failure, connection/send failure, unreadable or invalid response | 502 | `upstream_provider_error` | +| Provider 429 | 429 | Existing `inference_capacity` | +| Provider 503/529 | 503 | Existing `inference_capacity` | + +For 400/422 only, the bounded diagnostic can select +`upstream_invalid_parameter`, `upstream_unsupported_parameter`, +`upstream_invalid_messages`, or `upstream_context_limit`, each with a locally +written message. Unknown, unreadable or oversized bodies fall back to the status +category. Raw text, parameter names/values, request IDs and provider headers are +never copied. Diagnostics cannot override HTTP categories: HTTP429 remains a +capacity error even with an `invalid_request_error` label; provider 401/403/404 +becomes 502, never a Maple authentication, entitlement or resource error. A +provider rejection need not mean the caller violated Maple's schema. + +Routing, billing, retries and capacity feedback are unchanged. Capacity retains +its existing `Retry-After` and client-replay rules. New `upstream_*` codes grant +no replay permission and request no session recreation or token refresh. + +After streaming starts, Chat emits one encrypted OpenAI-shaped error; Responses +emits encrypted `response.failed` through its existing storage-terminal path, +with the safe code/message and error-contract metadata. Neither changes the +started HTTP status or grants retry permission. Mid-stream timeouts use +`upstream_timeout`; malformed/error payloads and unexpected EOF use +`upstream_provider_error`. Consumer drops, internal build failures, discovery +unavailability and helper-specific failure contracts remain separate. + ### Cryptographic interoperability fixture `crypto::tests::deterministic_key_and_record_vector` fixes the following diff --git a/services/opensecret/src/main.rs b/services/opensecret/src/main.rs index 42f6295ae..b92d1cdf2 100644 --- a/services/opensecret/src/main.rs +++ b/services/opensecret/src/main.rs @@ -389,6 +389,9 @@ pub enum ApiError { client_replay_safe: bool, }, + #[error("{0}")] + InferenceProvider(web::provider_error::PublicProviderError), + #[error("Bad Request")] BadRequest, @@ -473,6 +476,7 @@ impl IntoResponse for ApiError { ApiError::ServiceUnavailable => StatusCode::SERVICE_UNAVAILABLE, ApiError::ImageDescriptionUnavailable => StatusCode::SERVICE_UNAVAILABLE, ApiError::InferenceCapacity { status, .. } => *status, + ApiError::InferenceProvider(error) => error.status(), ApiError::BadRequest => StatusCode::BAD_REQUEST, ApiError::SessionNotFound => StatusCode::BAD_REQUEST, ApiError::Conflict => StatusCode::CONFLICT, @@ -498,6 +502,7 @@ impl IntoResponse for ApiError { ApiError::AccessTokenExpired => Some(ACCESS_TOKEN_EXPIRED_ERROR_CODE), ApiError::ImageDescriptionUnavailable => Some(IMAGE_DESCRIPTION_UNAVAILABLE_ERROR_CODE), ApiError::InferenceCapacity { .. } => Some(INFERENCE_CAPACITY_ERROR_CODE), + ApiError::InferenceProvider(error) => Some(error.code()), _ => None, }; let mut response = ( diff --git a/services/opensecret/src/transport_v2/gateway.rs b/services/opensecret/src/transport_v2/gateway.rs index f50a305ab..bf814caa3 100644 --- a/services/opensecret/src/transport_v2/gateway.rs +++ b/services/opensecret/src/transport_v2/gateway.rs @@ -1441,6 +1441,95 @@ mod tests { } } + #[tokio::test] + async fn provider_error_status_and_details_are_authenticated_without_replay_permission() { + use crate::inference::{AttemptFailure, AttemptFailureKind, AttemptStage, ReplaySafety}; + use crate::web::provider_error::PublicProviderError; + + for (upstream_status, status, code) in [ + (400, 400, "upstream_invalid_request"), + (401, 502, "upstream_provider_error"), + (504, 504, "upstream_timeout"), + ] { + let failure = AttemptFailure::new( + AttemptFailureKind::HttpStatus, + AttemptStage::AwaitingResponse, + ReplaySafety::NotProvenPreAcceptance, + ) + .with_upstream_response( + upstream_status, + None, + Some("private-request-canary".into()), + ); + let error = PublicProviderError::from_failure(&failure).unwrap(); + let test = test_session(14); + let sessions = Arc::new(SessionStore::new(NonZeroUsize::new(2).unwrap())); + sessions.insert(Arc::clone(&test.server)).unwrap(); + let dispatches = Arc::new(AtomicUsize::new(0)); + let calls = dispatches.clone(); + let application = Router::new().route( + "/provider-error", + any(move || { + calls.fetch_add(1, Ordering::SeqCst); + async move { ApiError::InferenceProvider(error).into_response() } + }), + ); + let router = request_router(application, sessions); + let request_id = RequestId::from_bytes([0xb1; 16]); + let record = seal_request(&test.client, request_id, "/provider-error", None); + let response = router + .clone() + .oneshot(outer_request( + test.server.id(), + &test.routing_key, + record.clone(), + )) + .await + .unwrap(); + for name in [ + crate::ERROR_CODE_HEADER, + crate::CLIENT_REPLAY_HEADER, + "retry-after", + ] { + assert!(!response.headers().contains_key(name)); + } + let records = decrypt_records(&test.client, request_id, response).await; + let ResponseRecord::Start(start) = &records[0] else { + panic!("missing response start") + }; + assert_eq!(start.status(), status); + let logical_header = |name: &str| { + start + .headers() + .iter() + .find(|header| header.name() == name) + .map(|header| header.value()) + }; + assert_eq!(logical_header(crate::ERROR_CONTRACT_HEADER), Some("1")); + assert_eq!(logical_header(crate::ERROR_CODE_HEADER), Some(code)); + assert_eq!(logical_header(crate::CLIENT_REPLAY_HEADER), None); + assert_eq!(logical_header("retry-after"), None); + let mut body = Vec::new(); + for record in &records { + if let ResponseRecord::Chunk(bytes) = record { + body.extend_from_slice(bytes); + } + } + assert_eq!( + serde_json::from_slice::(&body).unwrap(), + serde_json::json!({"status": status, "message": error.message()}) + ); + assert!(!String::from_utf8(body).unwrap().contains("private-")); + assert!(matches!(records.last(), Some(ResponseRecord::End))); + let response = router + .oneshot(outer_request(test.server.id(), &test.routing_key, record)) + .await + .unwrap(); + assert_outer_rejection(response, StatusCode::BAD_REQUEST, None); + assert_eq!(dispatches.load(Ordering::SeqCst), 1); + } + } + #[tokio::test] async fn response_records_are_bound_to_the_admitted_session_and_request() { let first = test_session(4); diff --git a/services/opensecret/src/web/mod.rs b/services/opensecret/src/web/mod.rs index 770cbc01d..87496434b 100644 --- a/services/opensecret/src/web/mod.rs +++ b/services/opensecret/src/web/mod.rs @@ -9,6 +9,7 @@ mod openai; pub mod openai_auth; pub mod platform; pub mod protected_routes; +pub mod provider_error; pub mod responses; pub mod web_routes; pub(crate) mod web_safety; diff --git a/services/opensecret/src/web/openai.rs b/services/opensecret/src/web/openai.rs index cdecbd08b..ca5201536 100644 --- a/services/opensecret/src/web/openai.rs +++ b/services/opensecret/src/web/openai.rs @@ -1015,7 +1015,9 @@ fn public_completion_error(error: &ProviderRequestError, failure: &AttemptFailur }; } - ApiError::from(error.clone()) + super::provider_error::PublicProviderError::from_failure(failure) + .map(ApiError::InferenceProvider) + .unwrap_or_else(|| ApiError::from(error.clone())) } fn terminalize_recovered_provider_failures( @@ -2116,7 +2118,7 @@ async fn proxy_openai( "Completion attempt failed: kind={:?}, stage={:?}", failure.kind, failure.stage ); - let error_payload = completion_error_payload(failure.client_message()); + let error_payload = completion_failure_payload(&failure); match encrypt_sse_event(&state, &session_id, &error_payload).await { Ok(event) => yield Ok(event), Err(error) => { @@ -3045,6 +3047,10 @@ pub(crate) async fn finish_started_completion( { Ok(response_json) => response_json, Err(failure) => { + let public_error = + super::provider_error::PublicProviderError::from_failure(&failure) + .map(ApiError::InferenceProvider) + .unwrap_or(ApiError::InternalServerError); let terminal = AttemptTerminal::Failed { attempt: attempt.clone(), failure, @@ -3052,7 +3058,7 @@ pub(crate) async fn finish_started_completion( terminal_guard.record_terminal(&terminal); return Err(CompletionExecutionError::Attempt { terminal, - public_error: ApiError::InternalServerError, + public_error, }); } }; @@ -3510,6 +3516,20 @@ fn completion_error_payload(message: &str) -> Value { }) } +fn completion_failure_payload(failure: &AttemptFailure) -> Value { + match super::provider_error::PublicProviderError::from_failure(failure) { + Some(error) => json!({ + "error": { + "message": error.message(), + "type": if error.status().is_client_error() { "invalid_request_error" } else { "server_error" }, + "param": null, + "code": error.code(), + } + }), + None => completion_error_payload(failure.client_message()), + } +} + fn sse_event_from_encoded_data(event_data: &str) -> Event { Event::default().data(event_data) } @@ -5692,7 +5712,7 @@ mod tests { } #[tokio::test] - async fn upstream_diagnostic_reaches_correlated_terminal_but_not_public_error() { + async fn upstream_diagnostic_projects_static_public_detail_without_provider_text() { let app = Router::new().route("/v1/chat/completions", post(|| async { axum::http::Response::builder().status(400) .body(axum::body::Body::from(r#"{"error":{"code":"unsupported_parameter","type":"invalid_request_error","message":"private-prompt-canary","param":"private-tool-canary"}}"#)) @@ -5710,7 +5730,9 @@ mod tests { Some("unsupported_parameter") ); let public = public_completion_error(&error, &failure); - assert!(matches!(public, ApiError::InternalServerError)); + assert!(matches!(public, ApiError::InferenceProvider(error) + if error.status() == StatusCode::BAD_REQUEST + && error.code() == "upstream_unsupported_parameter")); let router = ProviderRouter::default(); let pinned = pinned_test_completion(); let attempt = pinned @@ -5785,6 +5807,152 @@ mod tests { } } + #[tokio::test] + async fn mock_provider_errors_preserve_safe_status_detail_and_never_allow_replay() { + for (upstream_status, label, public_status, code) in [ + (400, "BadRequestError", 400, "upstream_invalid_request"), + (400, "private-label-canary", 400, "upstream_invalid_request"), + ( + 400, + "message-structure-fixture", + 400, + "upstream_invalid_messages", + ), + ( + 400, + "context_length_exceeded", + 400, + "upstream_context_limit", + ), + (400, "invalid_parameter", 400, "upstream_invalid_parameter"), + ( + 422, + "unsupported_parameter", + 422, + "upstream_unsupported_parameter", + ), + ( + 413, + "private-label-canary", + 413, + "upstream_payload_too_large", + ), + (401, "invalid_api_key", 502, "upstream_provider_error"), + (403, "permission_denied", 502, "upstream_provider_error"), + (404, "model_not_found", 502, "upstream_provider_error"), + (500, "invalid_request_error", 502, "upstream_provider_error"), + (502, "private-label-canary", 502, "upstream_provider_error"), + (408, "private-label-canary", 504, "upstream_timeout"), + (504, "private-label-canary", 504, "upstream_timeout"), + ] { + let calls = Arc::new(AtomicUsize::new(0)); + let mock_calls = calls.clone(); + let app = + Router::new().route( + "/v1/chat/completions", + post(move || { + mock_calls.fetch_add(1, Ordering::SeqCst); + async move { + axum::http::Response::builder().status(upstream_status) + .header(header::RETRY_AFTER, "7") + .header("x-request-id", "private-request-canary") + .header("x-opensecret-client-replay", "safe") + .body(axum::body::Body::from(json!({"error": { + "code": label, "type": "private-type-canary", + "message": if label == "message-structure-fixture" { + "conversation roles must alternate: private-prompt-canary" + } else { "private-prompt-canary" }, + "param": "private-tool-canary" + }}).to_string())).unwrap() + } + }), + ); + let (trace, server) = call_mock_provider(app).await; + server.abort(); + assert_eq!(calls.load(Ordering::SeqCst), 1); + assert!(trace.prior_failures.is_empty()); + let error = trace.result.err().expect("provider rejects the request"); + let failure = attempt_failure_from_provider_error(&error); + assert_eq!(failure.replay_safety, ReplaySafety::NotProvenPreAcceptance); + let pinned = pinned_test_completion(); + let execution_error = CompletionExecutionError::Attempt { + terminal: AttemptTerminal::Failed { + attempt: pinned + .begin_execution() + .begin_attempt(pinned.route.identity()), + failure: failure.clone(), + }, + public_error: public_completion_error(&error, &failure), + }; + let response = execution_error + .into_pre_persistence_api_error() + .into_response(); + assert_eq!(response.status().as_u16(), public_status); + assert_eq!(response.headers()[crate::ERROR_CONTRACT_HEADER], "1"); + assert_eq!(response.headers()[crate::ERROR_CODE_HEADER], code); + for name in [crate::CLIENT_REPLAY_HEADER, "retry-after", "x-request-id"] { + assert!(!response.headers().contains_key(name)); + } + let bytes = axum::body::to_bytes(response.into_body(), 1024) + .await + .unwrap(); + let body: Value = serde_json::from_slice(&bytes).unwrap(); + assert_eq!(body["status"], public_status); + assert!(body["message"] + .as_str() + .unwrap() + .starts_with("Upstream provider")); + assert_eq!(body.as_object().unwrap().len(), 2); // Legacy ErrorResponse shape. + assert!(!String::from_utf8(bytes.to_vec()) + .unwrap() + .contains("private-")); + let payload = completion_failure_payload(&failure); + assert_eq!(payload["error"]["code"], code); + assert_eq!( + payload["error"]["type"], + if public_status < 500 { + "invalid_request_error" + } else { + "server_error" + } + ); + assert!(payload["error"]["param"].is_null()); + assert!(!payload.to_string().contains("private-")); + } + } + + #[test] + fn completion_transport_errors_and_stream_payloads_use_safe_provider_projection() { + for (error, status, code) in [ + ( + ProviderRequestError::Timeout(Duration::from_secs(1)), + 504, + "upstream_timeout", + ), + ( + ProviderRequestError::Connect("private-address-canary".into()), + 502, + "upstream_provider_error", + ), + ( + ProviderRequestError::Send("private-body-canary".into()), + 502, + "upstream_provider_error", + ), + ] { + let failure = attempt_failure_from_provider_error(&error); + let response = public_completion_error(&error, &failure).into_response(); + assert_eq!(response.status().as_u16(), status); + assert_eq!(response.headers()[crate::ERROR_CODE_HEADER], code); + assert!(!response.headers().contains_key(crate::CLIENT_REPLAY_HEADER)); + let payload = completion_failure_payload(&failure); + assert_eq!(payload["error"]["code"], code); + assert_eq!(payload["error"]["type"], "server_error"); + assert!(payload["error"]["param"].is_null()); + assert!(!payload.to_string().contains("private-")); + } + } + #[tokio::test] async fn upstream_capacity_failure_does_not_claim_client_replay_safety() { let provider_error = ProviderRequestError::Upstream(UpstreamProviderError { @@ -6713,7 +6881,7 @@ mod tests { assert_eq!(failure.replay_safety, ReplaySafety::NotProvenPreAcceptance); assert!(matches!( public_completion_error(&generic, &failure), - ApiError::InternalServerError + ApiError::InferenceProvider(error) if error.status() == StatusCode::BAD_GATEWAY )); } diff --git a/services/opensecret/src/web/provider_error.rs b/services/opensecret/src/web/provider_error.rs new file mode 100644 index 000000000..424dd3494 --- /dev/null +++ b/services/opensecret/src/web/provider_error.rs @@ -0,0 +1,155 @@ +//! Client-facing completion errors. Provider text and identifiers never enter +//! this type: even the lossy diagnostic is mapped to a finite local vocabulary. + +use crate::inference::{AttemptFailure, AttemptFailureKind}; +use axum::http::StatusCode; +use std::fmt; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Detail { + InvalidRequest, + InvalidParameter, + UnsupportedParameter, + InvalidMessages, + ContextLimit, + PayloadTooLarge, + ProviderError, + Timeout, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct PublicProviderError { + status: StatusCode, + detail: Detail, +} + +impl PublicProviderError { + pub(crate) fn from_failure(failure: &AttemptFailure) -> Option { + let (status, detail) = match failure.kind { + AttemptFailureKind::HttpStatus => match failure.status { + Some(status @ (400 | 422)) => { + let detail = match failure + .upstream_diagnostic + .as_deref() + .and_then(|diagnostic| diagnostic.summary) + { + Some("upstream rejected a parameter value") => Detail::InvalidParameter, + Some("upstream does not support a request parameter") => { + Detail::UnsupportedParameter + } + Some("upstream rejected the message structure") => Detail::InvalidMessages, + Some("upstream context or token limit exceeded") => Detail::ContextLimit, + _ => Detail::InvalidRequest, + }; + let status = if status == 400 { + StatusCode::BAD_REQUEST + } else { + StatusCode::UNPROCESSABLE_ENTITY + }; + (status, detail) + } + Some(413) => (StatusCode::PAYLOAD_TOO_LARGE, Detail::PayloadTooLarge), + Some(408 | 504) => (StatusCode::GATEWAY_TIMEOUT, Detail::Timeout), + // In particular, an upstream 401/403 is not a Maple login or + // entitlement failure, and a 404 is not a missing user resource. + // The HTTP status wins over contradictory diagnostic labels. + _ => (StatusCode::BAD_GATEWAY, Detail::ProviderError), + }, + AttemptFailureKind::ResponseStartTimeout | AttemptFailureKind::StreamTimeout => { + (StatusCode::GATEWAY_TIMEOUT, Detail::Timeout) + } + AttemptFailureKind::Connect + | AttemptFailureKind::Transport + | AttemptFailureKind::ResponseBody + | AttemptFailureKind::InvalidResponse + | AttemptFailureKind::UpstreamResponseError + | AttemptFailureKind::UpstreamStreamError + | AttemptFailureKind::UnexpectedEof => (StatusCode::BAD_GATEWAY, Detail::ProviderError), + // These retain their existing, separate contracts. In particular, + // a caller losing interest does not prove an upstream failure. + AttemptFailureKind::CapacityRejected + | AttemptFailureKind::ProviderUnavailable + | AttemptFailureKind::RequestBuild + | AttemptFailureKind::ConsumerDropped => return None, + }; + Some(Self { status, detail }) + } + + pub(crate) fn status(self) -> StatusCode { + self.status + } + + pub(crate) fn code(self) -> &'static str { + match self.detail { + Detail::InvalidRequest => "upstream_invalid_request", + Detail::InvalidParameter => "upstream_invalid_parameter", + Detail::UnsupportedParameter => "upstream_unsupported_parameter", + Detail::InvalidMessages => "upstream_invalid_messages", + Detail::ContextLimit => "upstream_context_limit", + Detail::PayloadTooLarge => "upstream_payload_too_large", + Detail::ProviderError => "upstream_provider_error", + Detail::Timeout => "upstream_timeout", + } + } + + pub(crate) fn message(self) -> &'static str { + match self.detail { + Detail::InvalidRequest => "Upstream provider rejected the request. Check that the message format and parameters are supported by this model.", + Detail::InvalidParameter => "Upstream provider rejected a parameter value. Check the parameters supported by this model.", + Detail::UnsupportedParameter => "Upstream provider does not support a request parameter. Check the parameters supported by this model.", + Detail::InvalidMessages => "Upstream provider rejected the message structure. Check the message roles and ordering supported by this model.", + Detail::ContextLimit => "Upstream provider rejected the context or token limit. Reduce the input or requested output tokens.", + Detail::PayloadTooLarge => "Upstream provider rejected the request size. Reduce the request payload.", + Detail::ProviderError => "Upstream provider could not complete the request.", + Detail::Timeout => "Upstream provider timed out while processing the request.", + } + } +} + +impl fmt::Display for PublicProviderError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.message()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::inference::{AttemptStage, ReplaySafety}; + + #[test] + fn routing_build_and_consumer_outcomes_keep_their_own_contracts() { + for kind in [ + AttemptFailureKind::CapacityRejected, + AttemptFailureKind::ProviderUnavailable, + AttemptFailureKind::RequestBuild, + AttemptFailureKind::ConsumerDropped, + ] { + let failure = AttemptFailure::new( + kind, + AttemptStage::BeforeSend, + ReplaySafety::ProvenPreAcceptance, + ); + assert!(PublicProviderError::from_failure(&failure).is_none()); + } + } + + #[test] + fn raw_payload_codes_and_identifiers_cannot_become_public_details() { + let failure = AttemptFailure::new( + AttemptFailureKind::UpstreamStreamError, + AttemptStage::Stream, + ReplaySafety::NotProvenPreAcceptance, + ) + .with_upstream_code(Some("private-prompt-canary".into())) + .with_upstream_response(200, None, Some("private-request-canary".into())); + let error = PublicProviderError::from_failure(&failure).unwrap(); + assert_eq!(error.status(), StatusCode::BAD_GATEWAY); + assert_eq!(error.code(), "upstream_provider_error"); + assert_eq!( + error.message(), + "Upstream provider could not complete the request." + ); + assert!(!format!("{error:?} {error}").contains("private-")); + } +} diff --git a/services/opensecret/src/web/responses/events.rs b/services/opensecret/src/web/responses/events.rs index def4d62b5..60930008b 100644 --- a/services/opensecret/src/web/responses/events.rs +++ b/services/opensecret/src/web/responses/events.rs @@ -309,6 +309,41 @@ mod tests { ); assert!(serialized.get("error").is_none()); assert_eq!(serialized["opensecret"]["error_code"], "inference_capacity"); + + let provider_failure = crate::inference::AttemptFailure::new( + crate::inference::AttemptFailureKind::HttpStatus, + crate::inference::AttemptStage::AwaitingResponse, + crate::inference::ReplaySafety::NotProvenPreAcceptance, + ) + .with_upstream_response(400, None, None); + let public = + crate::web::provider_error::PublicProviderError::from_failure(&provider_failure) + .unwrap(); + let mut provider_failed = failed_payload.clone(); + provider_failed.response.error = Some(ResponseError { + code: public.code().to_string(), + message: public.message().to_string(), + }); + provider_failed.opensecret = Some(OpenSecretResponseError { + error_contract: "1", + error_code: public.code(), + }); + let serialized = serde_json::to_value(&provider_failed).unwrap(); + assert_eq!(serialized["type"], EVENT_RESPONSE_FAILED); + assert_eq!(serialized["response"]["status"], STATUS_FAILED); + assert_eq!( + serialized["response"]["error"]["code"], + "upstream_invalid_request" + ); + assert_eq!(serialized["opensecret"]["error_contract"], "1"); + assert_eq!( + serialized["opensecret"]["error_code"], + "upstream_invalid_request" + ); + assert!(serialized.get("error").is_none()); + // A failed response is already accepted work; it carries no retry or + // replay permission despite its provider-side invalid-request category. + assert_eq!(serialized["opensecret"].as_object().unwrap().len(), 2); assert_eq!( ResponseEvent::Failed(failed_payload).event_type(), EVENT_RESPONSE_FAILED diff --git a/services/opensecret/src/web/responses/handlers.rs b/services/opensecret/src/web/responses/handlers.rs index edb9f5cce..d6f1eb823 100644 --- a/services/opensecret/src/web/responses/handlers.rs +++ b/services/opensecret/src/web/responses/handlers.rs @@ -36,6 +36,7 @@ use crate::{ ResolvedInferenceModel, StartedCompletion, }, openai_auth::AuthMethod, + provider_error::PublicProviderError, responses::{ build_prompt, build_prompt_with_token_reserve, build_usage, constants::*, @@ -302,6 +303,7 @@ impl ResponseExecutionPolicy { pub enum PublicResponseFailure { CapacityRateLimited, CapacityOverloaded, + Provider(PublicProviderError), DeadlineExceeded, Internal, } @@ -309,13 +311,7 @@ pub enum PublicResponseFailure { impl PublicResponseFailure { fn from_completion_error(error: &CompletionExecutionError) -> Self { match error { - CompletionExecutionError::Request(ApiError::InferenceCapacity { status, .. }) => { - if *status == axum::http::StatusCode::TOO_MANY_REQUESTS { - Self::CapacityRateLimited - } else { - Self::CapacityOverloaded - } - } + CompletionExecutionError::Request(error) => Self::from_api_error(error), CompletionExecutionError::Attempt { terminal: AttemptTerminal::Failed { failure, .. }, .. @@ -326,6 +322,22 @@ impl PublicResponseFailure { Self::CapacityOverloaded } } + CompletionExecutionError::Attempt { public_error, .. } => { + Self::from_api_error(public_error) + } + } + } + + fn from_api_error(error: &ApiError) -> Self { + match error { + ApiError::InferenceCapacity { status, .. } => { + if *status == axum::http::StatusCode::TOO_MANY_REQUESTS { + Self::CapacityRateLimited + } else { + Self::CapacityOverloaded + } + } + ApiError::InferenceProvider(error) => Self::Provider(*error), _ => Self::Internal, } } @@ -333,6 +345,7 @@ impl PublicResponseFailure { fn openai_code(self) -> &'static str { match self { Self::CapacityRateLimited => "rate_limit_exceeded", + Self::Provider(error) => error.code(), Self::CapacityOverloaded | Self::DeadlineExceeded | Self::Internal => "server_error", } } @@ -343,17 +356,21 @@ impl PublicResponseFailure { "Inference capacity is temporarily unavailable." } Self::DeadlineExceeded => "The response exceeded its execution deadline.", + Self::Provider(error) => error.message(), Self::Internal => "The response could not be completed.", } } fn contract_metadata(self) -> Option { - matches!(self, Self::CapacityRateLimited | Self::CapacityOverloaded).then_some( - OpenSecretResponseError { - error_contract: ERROR_CONTRACT_VERSION, - error_code: INFERENCE_CAPACITY_ERROR_CODE, - }, - ) + let error_code = match self { + Self::CapacityRateLimited | Self::CapacityOverloaded => INFERENCE_CAPACITY_ERROR_CODE, + Self::Provider(error) => error.code(), + Self::DeadlineExceeded | Self::Internal => return None, + }; + Some(OpenSecretResponseError { + error_contract: ERROR_CONTRACT_VERSION, + error_code, + }) } } @@ -619,7 +636,10 @@ mod tests { model_config::{ModelAliasTargets, ModelPlan}, models::responses::ResponseStatus, provider_registry::{ProviderId, RouteSelectionSource}, - web::openai::{CompletionChunk, CompletionExecutionError, CompletionUsage}, + web::{ + openai::{CompletionChunk, CompletionExecutionError, CompletionUsage}, + provider_error::PublicProviderError, + }, ApiError, ResponseMessageReservation, ResponseMessageReservations, }; use axum::{routing::get, Json, Router}; @@ -1855,7 +1875,7 @@ mod tests { } #[test] - fn only_capacity_failures_expose_opensecret_terminal_metadata() { + fn typed_failures_expose_only_safe_opensecret_terminal_metadata() { let rate_limit = PublicResponseFailure::CapacityRateLimited; assert_eq!(rate_limit.openai_code(), "rate_limit_exceeded"); let metadata = rate_limit.contract_metadata().unwrap(); @@ -1874,6 +1894,93 @@ mod tests { .is_none()); } + #[test] + fn provider_rejections_keep_safe_details_before_and_after_persistence() { + use axum::response::IntoResponse; + + for upstream_status in [400, 502] { + let failure = AttemptFailure::new( + AttemptFailureKind::HttpStatus, + AttemptStage::AwaitingResponse, + ReplaySafety::NotProvenPreAcceptance, + ) + .with_upstream_response(upstream_status, None, None); + let public = PublicProviderError::from_failure(&failure).unwrap(); + let CompletionChunk::Terminal(AttemptTerminal::Completed { attempt, .. }) = + completed_attempt_chunk() + else { + panic!("fixture must contain an attempt"); + }; + let execution_error = CompletionExecutionError::Attempt { + terminal: AttemptTerminal::Failed { attempt, failure }, + public_error: ApiError::InferenceProvider(public), + }; + + // Later model turns have already started the Responses stream: keep + // these details in response.failed, never as a new HTTP response. + let terminal = PublicResponseFailure::from_completion_error(&execution_error); + assert_eq!(terminal, PublicResponseFailure::Provider(public)); + assert_eq!(terminal.message(), public.message()); + assert_eq!(terminal.openai_code(), public.code()); + let metadata = terminal.contract_metadata().unwrap(); + assert_eq!(metadata.error_contract, "1"); + assert_eq!(metadata.error_code, public.code()); + + let response = + responses_pre_persistence_api_error(execution_error, true).into_response(); + assert_eq!(response.status(), public.status()); + assert_eq!(response.headers()[crate::ERROR_CODE_HEADER], public.code()); + assert!(!response.headers().contains_key(crate::CLIENT_REPLAY_HEADER)); + assert!(!response + .headers() + .contains_key(axum::http::header::RETRY_AFTER)); + } + } + + #[tokio::test] + async fn assistant_turn_preserves_safe_provider_failure_without_synthesizing_completion() { + let failure = AttemptFailure::new( + AttemptFailureKind::StreamTimeout, + AttemptStage::Stream, + ReplaySafety::NotProvenPreAcceptance, + ); + let public = PublicProviderError::from_failure(&failure).unwrap(); + let CompletionChunk::Terminal(AttemptTerminal::Completed { attempt, .. }) = + completed_attempt_chunk() + else { + panic!("fixture must contain an attempt"); + }; + let (tx_completion, rx_completion) = mpsc::channel(1); + tx_completion + .send(CompletionChunk::Terminal(AttemptTerminal::Failed { + attempt, + failure, + })) + .await + .unwrap(); + drop(tx_completion); + let (tx_storage, mut rx_storage) = mpsc::channel(1); + let (tx_client, mut rx_client) = mpsc::channel(1); + let error = consume_assistant_turn( + rx_completion, + false, + &tx_client, + &tx_storage, + &mut Some(Uuid::new_v4()), + Uuid::new_v4(), + ) + .await + .expect_err("failed provider stream must stay failed"); + assert_eq!( + PublicResponseFailure::from_api_error(&error), + PublicResponseFailure::Provider(public) + ); + assert_eq!(public.code(), "upstream_timeout"); + // Only the existing storage supervisor may emit the final terminal. + assert!(rx_storage.try_recv().is_err()); + assert!(rx_client.try_recv().is_err()); + } + #[test] fn local_capacity_request_errors_remain_typed_after_persistence() { for (status, expected) in [ @@ -5047,7 +5154,9 @@ async fn consume_assistant_turn( "Received failed inference terminal: kind={:?}, stage={:?}", failure.kind, failure.stage ); - return Err(ApiError::InternalServerError); + return Err(PublicProviderError::from_failure(&failure) + .map(ApiError::InferenceProvider) + .unwrap_or(ApiError::InternalServerError)); } crate::web::openai::CompletionChunk::FullResponse(_) => { error!("Received FullResponse in streaming mode"); @@ -5152,7 +5261,9 @@ async fn setup_completion_processor( .await { Ok(turn) => turn, - Err(_) => return ResponseTerminal::Failed(PublicResponseFailure::Internal), + Err(error) => { + return ResponseTerminal::Failed(PublicResponseFailure::from_api_error(&error)); + } }; match turn {