diff --git a/openai/src/client.rs b/openai/src/client.rs index 948f4d6d..f19feaf2 100644 --- a/openai/src/client.rs +++ b/openai/src/client.rs @@ -1216,28 +1216,23 @@ fn responses_stream_inner( usage_emitted = true; } } - ResponsesStreamEvent::ResponseFailed { error } => { - let msg = error - .and_then(|e| e.message) - .unwrap_or_else(|| "Response failed".to_string()); - if msg.contains("call_id") { - yield Err(OpenAIError::Api(format!( - "{msg}\nSerialized responses request: {serialized_request}" - ))); - } else { - yield Err(OpenAIError::Api(msg)); - } + ResponsesStreamEvent::ResponseFailed { response } => { + let (code, message) = response + .and_then(|response| response.error) + .map_or((None, None), |e| (e.code, e.message)); + yield Err(responses_event_error( + code, + message.unwrap_or_else(|| "Response failed".to_string()), + &serialized_request, + )); return; } - ResponsesStreamEvent::Error { message, .. } => { - let msg = message.unwrap_or_else(|| "Unknown error".to_string()); - if msg.contains("call_id") { - yield Err(OpenAIError::Api(format!( - "{msg}\nSerialized responses request: {serialized_request}" - ))); - } else { - yield Err(OpenAIError::Api(msg)); - } + ResponsesStreamEvent::Error { message, code } => { + yield Err(responses_event_error( + code, + message.unwrap_or_else(|| "Unknown error".to_string()), + &serialized_request, + )); return; } // Ignore other events @@ -1268,6 +1263,26 @@ fn responses_stream_inner( /// Start event for a provider-executed (hosted) tool item, when the output /// item is one we model — the provider runs it, no consumer execution. +/// The error a Responses `error` or `response.failed` event reports, keeping +/// its `code` so the failure classifies (a mid-stream `rate_limit_exceeded` +/// is transient). A `call_id` complaint carries the request that caused it. +fn responses_event_error( + code: Option, + message: String, + serialized_request: &str, +) -> OpenAIError { + let message = if message.contains("call_id") { + format!("{message}\nSerialized responses request: {serialized_request}") + } else { + message + }; + OpenAIError::StreamEvent { + error_type: None, + code, + message, + } +} + fn builtin_call_event(item: &ResponsesOutputItem) -> Option { let (id, tool, detail) = match item { ResponsesOutputItem::WebSearchCall { id, action, .. } => ( @@ -1911,6 +1926,62 @@ mod tests { } } + /// A rate limit reported after the stream started is transient, on every + /// event that can carry it: the Responses `error` and `response.failed` + /// events and the chat-completions error object. + #[tokio::test] + async fn a_rate_limit_reported_mid_stream_is_transient() { + let message = "Reached request rate limit. Your limit will reset in 4 seconds."; + let error = serde_json::json!({"code": "rate_limit_exceeded", "message": message}); + let cases = [ + ( + ApiKind::Responses, + serde_json::json!({"type": "error", "code": "rate_limit_exceeded", "message": message}), + ), + ( + ApiKind::Responses, + serde_json::json!({ + "type": "response.failed", + "response": {"status": "failed", "error": error}, + }), + ), + ( + ApiKind::ChatCompletions, + serde_json::json!({"error": error}), + ), + ]; + for (api, event) in cases { + let url = serve_scripted(vec![vec![ + Step::Send(SSE_HEAD.to_string()), + Step::Send(text_event(api, "hi")), + Step::Send(format!("data: {event}\n\n")), + ]]); + let client = scripted_client(api, url, RetryPolicy::none()); + let stream = client.respond(oneshot("system", "hello")); + futures_lite::pin!(stream); + let first = stream + .next() + .await + .expect("stream yields") + .expect("the first event arrives"); + assert!( + matches!(&first, Event::Text(text) if text == "hi"), + "{event}" + ); + let error = stream + .next() + .await + .expect("stream yields") + .expect_err("the stream reports the rate limit"); + assert_eq!( + error.failure().kind, + aither_core::llm::FailureKind::Transient, + "{event}: {error}" + ); + assert!(error.to_string().contains(message), "{event}: {error}"); + } + } + #[tokio::test] async fn a_models_request_that_never_answers_fails_with_idle_timeout() { let url = serve_scripted(vec![vec![Step::Hold]]); diff --git a/openai/src/error.rs b/openai/src/error.rs index 8575608d..df2930e9 100644 --- a/openai/src/error.rs +++ b/openai/src/error.rs @@ -123,6 +123,7 @@ impl ErrorObject { if is(&["insufficient_quota", "billing_hard_limit_reached"]) { Some(FailureKind::Quota) } else if is(&[ + "rate_limit_exceeded", "server_error", "service_unavailable_error", "overloaded_error", diff --git a/openai/src/response.rs b/openai/src/response.rs index d2592b57..e8d61b87 100644 --- a/openai/src/response.rs +++ b/openai/src/response.rs @@ -31,7 +31,7 @@ pub enum ResponsesStreamEvent { #[serde(rename = "response.failed")] ResponseFailed { #[serde(default)] - error: Option, + response: Option, }, /// Output item added #[serde(rename = "response.output_item.added")] @@ -128,6 +128,9 @@ pub struct ResponsesStreamResponse { pub output: Vec, #[serde(default)] pub usage: Option, + /// Why the response failed, on `response.failed`. + #[serde(default)] + pub error: Option, } #[derive(Debug, Deserialize, Clone, Default)]