aither_openai/
error.rs

1use std::fmt;
2use std::time::Duration;
3
4use aither_core::llm::{
5    Classify, Failure, FailureKind,
6    failure::zenwave::{body_failure, failure as http_failure, sse_failure},
7};
8use zenwave::{BodyError, Error as ZenwaveError, sse::ParseError as SseParseError};
9
10/// Errors that can arise when calling the `OpenAI` API.
11#[derive(Debug)]
12pub enum OpenAIError {
13    /// HTTP layer errors.
14    Http(ZenwaveError),
15    /// Response body parsing failures.
16    Body(BodyError),
17    /// SSE parsing failures.
18    Stream(SseParseError),
19    /// JSON serialization/deserialization errors.
20    Json(serde_json::Error),
21    /// Base64 decoding failures (e.g., image payloads).
22    Decode(base64::DecodeError),
23    /// API contract violations or unsupported operations.
24    Api(String),
25    /// Rate limit exceeded (HTTP 429).
26    RateLimit {
27        /// Message from the API.
28        message: String,
29        /// Suggested retry delay from Retry-After header.
30        retry_after: Option<Duration>,
31    },
32    /// The account's quota or billing limit is exhausted (HTTP 429
33    /// `insufficient_quota`).
34    QuotaExceeded {
35        /// Message from the API.
36        message: String,
37    },
38    /// Server error (HTTP 5xx).
39    ServerError {
40        /// HTTP status code.
41        status: u16,
42        /// Error message.
43        message: String,
44        /// Suggested retry delay from Retry-After header.
45        retry_after: Option<Duration>,
46    },
47    /// The API reported an error in an event of an open stream.
48    StreamEvent {
49        /// The error's `type`, such as `server_error`.
50        error_type: Option<String>,
51        /// The error's `code`, such as `insufficient_quota`.
52        code: Option<String>,
53        /// Message from the API.
54        message: String,
55    },
56    /// Request timed out.
57    Timeout,
58}
59
60impl fmt::Display for OpenAIError {
61    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
62        match self {
63            Self::Http(err) => write!(f, "HTTP error: {err}"),
64            Self::Body(err) => write!(f, "Body error: {err}"),
65            Self::Stream(err) => write!(f, "SSE error: {err}"),
66            Self::Json(err) => write!(f, "JSON error: {err}"),
67            Self::Decode(err) => write!(f, "Base64 decode error: {err}"),
68            Self::Api(message) | Self::StreamEvent { message, .. } => write!(f, "{message}"),
69            Self::RateLimit {
70                message,
71                retry_after,
72            } => {
73                write!(f, "Rate limit exceeded: {message}")?;
74                if let Some(delay) = retry_after {
75                    write!(f, " (retry after {}s)", delay.as_secs())?;
76                }
77                Ok(())
78            }
79            Self::QuotaExceeded { message } => write!(f, "Quota exceeded: {message}"),
80            Self::ServerError {
81                status, message, ..
82            } => {
83                write!(f, "Server error {status}: {message}")
84            }
85            Self::Timeout => write!(f, "Request timed out"),
86        }
87    }
88}
89
90impl std::error::Error for OpenAIError {}
91
92/// The `error` object of an `OpenAI` error body, `{"error":{"type":..,"code":..}}`.
93#[derive(Debug, Default, serde::Deserialize)]
94struct ErrorObject {
95    #[serde(rename = "type")]
96    error_type: Option<String>,
97    code: Option<String>,
98    message: Option<String>,
99}
100
101impl ErrorObject {
102    fn from_body(body: &str) -> Option<Self> {
103        #[derive(serde::Deserialize)]
104        struct Body {
105            error: ErrorObject,
106        }
107        serde_json::from_str::<Body>(body).ok().map(|b| b.error)
108    }
109
110    /// The kind this error object forces regardless of status, if any.
111    fn kind(&self) -> Option<FailureKind> {
112        let is = |wanted: &[&str]| {
113            [&self.code, &self.error_type]
114                .into_iter()
115                .flatten()
116                .any(|value| wanted.contains(&value.as_str()))
117        };
118        if is(&["insufficient_quota", "billing_hard_limit_reached"]) {
119            Some(FailureKind::Quota)
120        } else if is(&[
121            "server_error",
122            "service_unavailable_error",
123            "overloaded_error",
124        ]) {
125            Some(FailureKind::Transient)
126        } else {
127            None
128        }
129    }
130}
131
132impl OpenAIError {
133    /// Convert a zenwave error to an `OpenAIError`.
134    pub(crate) fn from_http(err: zenwave::Error) -> Self {
135        let failure = http_failure(&err);
136        if let zenwave::Error::Http { status, .. } = &err {
137            let object = err.response_body().and_then(ErrorObject::from_body);
138            if status.as_u16() == 429 {
139                if failure.kind == FailureKind::Quota
140                    || object.as_ref().and_then(ErrorObject::kind) == Some(FailureKind::Quota)
141                {
142                    return Self::QuotaExceeded {
143                        message: object
144                            .and_then(|o| o.message)
145                            .unwrap_or_else(|| "Quota exceeded".to_string()),
146                    };
147                }
148                return Self::RateLimit {
149                    message: "Rate limit exceeded".to_string(),
150                    retry_after: failure.retry_after,
151                };
152            }
153            if status.as_u16() >= 500 {
154                return Self::ServerError {
155                    status: status.as_u16(),
156                    message: err.to_string(),
157                    retry_after: failure.retry_after,
158                };
159            }
160        }
161        if err.is_timeout() {
162            return Self::Timeout;
163        }
164        Self::Http(err)
165    }
166
167    /// Builds the error for an `error` object that arrived as a stream event.
168    pub(crate) fn from_stream_event(error: &serde_json::Value) -> Self {
169        let text = |key: &str| error.get(key).and_then(|v| v.as_str()).map(str::to_string);
170        Self::StreamEvent {
171            error_type: text("type"),
172            code: text("code"),
173            message: text("message").unwrap_or_else(|| "Unknown API error".to_string()),
174        }
175    }
176}
177
178impl Classify for OpenAIError {
179    fn failure(&self) -> Failure {
180        match self {
181            Self::Http(err) => {
182                let mut failure = http_failure(err);
183                if let Some(kind) = err
184                    .response_body()
185                    .and_then(ErrorObject::from_body)
186                    .and_then(|object| object.kind())
187                {
188                    failure.kind = kind;
189                }
190                failure
191            }
192            Self::Body(err) => body_failure(err),
193            Self::Stream(err) => sse_failure(err),
194            // An EOF while parsing JSON is a body cut off mid-response.
195            Self::Json(err) if err.is_eof() => Failure::network(),
196            Self::Json(_) | Self::Decode(_) | Self::Api(_) => Failure::permanent(),
197            Self::RateLimit { retry_after, .. } | Self::ServerError { retry_after, .. } => {
198                Failure::transient().with_retry_after(*retry_after)
199            }
200            Self::QuotaExceeded { .. } => Failure::quota(),
201            Self::Timeout => Failure::network(),
202            Self::StreamEvent {
203                error_type, code, ..
204            } => {
205                let object = ErrorObject {
206                    error_type: error_type.clone(),
207                    code: code.clone(),
208                    message: None,
209                };
210                Failure::new(object.kind().unwrap_or(FailureKind::Permanent))
211            }
212        }
213    }
214}
215
216impl From<BodyError> for OpenAIError {
217    fn from(value: BodyError) -> Self {
218        Self::Body(value)
219    }
220}
221
222impl From<SseParseError> for OpenAIError {
223    fn from(value: SseParseError) -> Self {
224        Self::Stream(value)
225    }
226}
227
228impl From<serde_json::Error> for OpenAIError {
229    fn from(value: serde_json::Error) -> Self {
230        Self::Json(value)
231    }
232}
233
234impl From<base64::DecodeError> for OpenAIError {
235    fn from(value: base64::DecodeError) -> Self {
236        Self::Decode(value)
237    }
238}
239
240impl From<std::io::Error> for OpenAIError {
241    fn from(value: std::io::Error) -> Self {
242        Self::Api(format!("IO error: {value}"))
243    }
244}
245
246#[cfg(test)]
247mod tests {
248    use std::time::Duration;
249
250    use aither_core::{
251        LanguageModel,
252        llm::{Classify, FailureKind, failure::zenwave::http_error, oneshot},
253    };
254    use futures_lite::StreamExt;
255
256    use super::OpenAIError;
257    use crate::OpenAI;
258
259    fn from_status(status: u16, headers: &[(&str, &str)], body: &str) -> OpenAIError {
260        OpenAIError::from_http(http_error(status, headers, body))
261    }
262
263    fn raw_http(status: u16, headers: &[(&str, &str)], body: &str) -> OpenAIError {
264        OpenAIError::Http(http_error(status, headers, body))
265    }
266
267    #[test]
268    fn server_error_is_transient() {
269        assert_eq!(
270            from_status(500, &[], "").failure().kind,
271            FailureKind::Transient
272        );
273        assert_eq!(
274            raw_http(503, &[], "").failure().kind,
275            FailureKind::Transient
276        );
277    }
278
279    #[test]
280    fn overloaded_is_transient() {
281        assert_eq!(
282            raw_http(529, &[], "").failure().kind,
283            FailureKind::Transient
284        );
285        let event = OpenAIError::StreamEvent {
286            error_type: Some("server_error".into()),
287            code: None,
288            message: "overloaded".into(),
289        };
290        assert_eq!(event.failure().kind, FailureKind::Transient);
291    }
292
293    #[test]
294    fn quota_429_is_quota_and_rate_limit_429_is_transient() {
295        let quota = r#"{"error":{"message":"You exceeded your current quota","type":"insufficient_quota","code":"insufficient_quota"}}"#;
296        assert_eq!(
297            from_status(429, &[], quota).failure().kind,
298            FailureKind::Quota
299        );
300        assert_eq!(raw_http(429, &[], quota).failure().kind, FailureKind::Quota);
301        let rate =
302            r#"{"error":{"message":"Slow down","type":"requests","code":"rate_limit_exceeded"}}"#;
303        assert_eq!(
304            from_status(429, &[], rate).failure().kind,
305            FailureKind::Transient
306        );
307        assert_eq!(
308            raw_http(429, &[], rate).failure().kind,
309            FailureKind::Transient
310        );
311    }
312
313    #[test]
314    fn unauthorized_is_permanent() {
315        assert_eq!(
316            from_status(401, &[], "").failure().kind,
317            FailureKind::Permanent
318        );
319        assert_eq!(
320            raw_http(401, &[], "").failure().kind,
321            FailureKind::Permanent
322        );
323    }
324
325    #[test]
326    fn retry_after_seconds_and_date() {
327        let failure = from_status(429, &[("retry-after", "7")], "").failure();
328        assert_eq!(failure.retry_after, Some(Duration::from_secs(7)));
329        let failure =
330            raw_http(503, &[("retry-after", "Fri, 01 Jan 2100 00:00:00 GMT")], "").failure();
331        assert!(
332            failure
333                .retry_after
334                .is_some_and(|d| d > Duration::from_secs(3600))
335        );
336    }
337
338    #[test]
339    fn truncated_json_is_network() {
340        let error = serde_json::from_str::<serde_json::Value>("{\"a\":").unwrap_err();
341        assert_eq!(
342            OpenAIError::Json(error).failure().kind,
343            FailureKind::Network
344        );
345    }
346
347    #[tokio::test]
348    async fn refused_connection_is_network() {
349        let port = std::net::TcpListener::bind("127.0.0.1:0")
350            .unwrap()
351            .local_addr()
352            .unwrap()
353            .port();
354        let client = OpenAI::builder("key")
355            .base_url(format!("http://127.0.0.1:{port}"))
356            .no_retry()
357            .build();
358        let stream = client.respond(oneshot("system", "hello"));
359        futures_lite::pin!(stream);
360        let error = stream
361            .next()
362            .await
363            .expect("stream yields")
364            .expect_err("connection is refused");
365        assert_eq!(error.failure().kind, FailureKind::Network, "{error}");
366    }
367}
aither_openai/error.rs

aither_openai/
error.rs
