diff --git a/lib/components/fabro-llm/src/codec/mod.rs b/lib/components/fabro-llm/src/codec/mod.rs index 97bf842b9..13f785ff7 100644 --- a/lib/components/fabro-llm/src/codec/mod.rs +++ b/lib/components/fabro-llm/src/codec/mod.rs @@ -194,9 +194,9 @@ pub(crate) trait Codec: Send + Sync { /// Map a non-2xx response to an `Error`. `retry_after` is the /// transport-parsed `retry-after` header value in seconds (header parsing /// is the transport's job, like `rate_limit` on the decode methods). - /// Default = shared HTTP-status mapping, which openai_compatible and - /// anthropic use as-is; a codec overrides when its dialect's error bodies - /// need more (e.g. gemini's gRPC status). + /// Default = shared HTTP-status mapping, which anthropic uses as-is. A + /// codec overrides when its dialect's error bodies need more (for example, + /// OpenRouter's typed metadata or Gemini's gRPC status). fn decode_error( &self, status: u16, diff --git a/lib/components/fabro-llm/src/codec/openai_compatible/error.rs b/lib/components/fabro-llm/src/codec/openai_compatible/error.rs new file mode 100644 index 000000000..941a887a9 --- /dev/null +++ b/lib/components/fabro-llm/src/codec/openai_compatible/error.rs @@ -0,0 +1,420 @@ +//! Provider error normalization for the Chat Completions wire dialect. + +use crate::codec; +use crate::error::{self as llm_error, Error, ProviderErrorDetail, ProviderErrorKind}; + +const ROUTING_EXHAUSTED_MESSAGE: &str = "Upstream provider routing was exhausted"; + +/// Structured fields used to classify an OpenAI-compatible provider error. +/// +/// OpenRouter's current contract exposes `metadata.error_type`. Older +/// responses can instead expose a list of failed provider attempts. Keep that +/// compatibility input private to the wire boundary so retry code only sees +/// Fabro's normal provider error taxonomy. +struct ParsedProviderError { + message: String, + canonical_code: Option, + native_code: Option, + provider_code: Option, + status_code: Option, + previous_attempts: Option>, + raw: serde_json::Value, +} + +impl ParsedProviderError { + fn from_raw(raw: serde_json::Value, status_code: Option) -> Self { + let error = &raw["error"]; + let metadata = error.get("metadata"); + + let message = error + .get("message") + .and_then(serde_json::Value::as_str) + .filter(|message| !message.is_empty()) + .unwrap_or("Provider returned error") + .to_string(); + let canonical_code = metadata + .and_then(|metadata| metadata.get("error_type")) + .and_then(nonempty_string); + let native_code = error.get("type").and_then(nonempty_string).or_else(|| { + error + .get("code") + .filter(|code| code.is_string()) + .and_then(nonempty_string) + }); + let provider_code = metadata + .and_then(|metadata| metadata.get("provider_code")) + .and_then(scalar_string); + let status_code = status_code.or_else(|| error.get("code").and_then(status_from_value)); + let previous_attempts = metadata + .and_then(|metadata| metadata.get("previous_errors")) + .and_then(previous_attempt_statuses); + + Self { + message, + canonical_code, + native_code, + provider_code, + status_code, + previous_attempts, + raw, + } + } + + fn error_code(&self) -> Option { + self.canonical_code + .clone() + .or_else(|| self.native_code.clone()) + .or_else(|| self.provider_code.clone()) + } + + fn routing_was_exhausted(&self) -> bool { + self.status_code == Some(400) + && self.canonical_code.is_none() + && self.previous_attempts.as_ref().is_some_and(|attempts| { + !attempts.is_empty() && attempts.iter().copied().all(is_transient_status_code) + }) + } + + fn into_error(self, provider: &str, retry_after: Option) -> Error { + if let Some(kind) = self + .canonical_code + .as_deref() + .and_then(kind_from_canonical_code) + { + return self.provider_error(kind, provider, retry_after); + } + + // Do not let legacy routing metadata override a status with a clear + // meaning of its own. + if self.status_code.is_some_and(is_unambiguous_status_code) { + return self.status_error(provider, retry_after); + } + + if self.routing_was_exhausted() { + return self.provider_error_with_message( + ProviderErrorKind::Server, + ROUTING_EXHAUSTED_MESSAGE, + provider, + retry_after, + ); + } + + if self.status_code.is_some() { + self.status_error(provider, retry_after) + } else { + // An in-band error with no status or canonical code happened after + // the HTTP stream opened. Treat that unknown provider failure as + // transient, while retaining all provider detail in `raw`. + self.provider_error(ProviderErrorKind::Server, provider, retry_after) + } + } + + fn status_error(self, provider: &str, retry_after: Option) -> Error { + let Some(status_code) = self.status_code else { + return self.provider_error(ProviderErrorKind::Server, provider, retry_after); + }; + let error_code = self.error_code(); + llm_error::error_from_status_code( + status_code, + self.message, + provider.to_string(), + error_code, + Some(self.raw), + retry_after, + ) + } + + fn provider_error( + self, + kind: ProviderErrorKind, + provider: &str, + retry_after: Option, + ) -> Error { + self.build_provider_error(kind, None, provider, retry_after) + } + + fn provider_error_with_message( + self, + kind: ProviderErrorKind, + message: &str, + provider: &str, + retry_after: Option, + ) -> Error { + self.build_provider_error(kind, Some(message.to_string()), provider, retry_after) + } + + fn build_provider_error( + self, + kind: ProviderErrorKind, + message: Option, + provider: &str, + retry_after: Option, + ) -> Error { + let error_code = self.error_code(); + Error::Provider { + kind, + detail: Box::new(ProviderErrorDetail { + message: message.unwrap_or(self.message), + provider: provider.to_string(), + status_code: self.status_code, + error_code, + retry_after, + raw: Some(self.raw), + }), + } + } +} + +pub(super) fn decode_http( + status: u16, + body: &str, + provider: &str, + retry_after: Option, +) -> Error { + let (message, code, raw) = codec::parse_error_body(body, "type"); + let Some(raw) = raw else { + return llm_error::error_from_status_code( + status, + message, + provider.to_string(), + code, + None, + retry_after, + ); + }; + if raw.get("error").is_none() { + return llm_error::error_from_status_code( + status, + message, + provider.to_string(), + code, + Some(raw), + retry_after, + ); + } + let parsed = ParsedProviderError::from_raw(raw, Some(status)); + + parsed.into_error(provider, retry_after) +} + +pub(super) fn decode_stream(raw: serde_json::Value, provider: &str) -> Error { + ParsedProviderError::from_raw(raw, None).into_error(provider, None) +} + +fn kind_from_canonical_code(code: &str) -> Option { + let kind = match code { + "authentication" => ProviderErrorKind::Authentication, + "permission_denied" => ProviderErrorKind::AccessDenied, + "not_found" => ProviderErrorKind::NotFound, + "payment_required" => ProviderErrorKind::QuotaExceeded, + "rate_limit_exceeded" => ProviderErrorKind::RateLimit, + "provider_overloaded" | "provider_unavailable" | "server" | "timeout" | "unmapped" => { + ProviderErrorKind::Server + } + "context_length_exceeded" + | "max_tokens_exceeded" + | "token_limit_exceeded" + | "string_too_long" => ProviderErrorKind::ContextLength, + "content_policy_violation" | "refusal" => ProviderErrorKind::ContentFilter, + "invalid_request" + | "invalid_prompt" + | "precondition_failed" + | "payload_too_large" + | "unprocessable" + | "invalid_image" + | "image_too_large" + | "image_too_small" + | "unsupported_image_format" + | "image_not_found" + | "image_download_failed" => ProviderErrorKind::InvalidRequest, + _ => return None, + }; + Some(kind) +} + +fn previous_attempt_statuses(value: &serde_json::Value) -> Option> { + let attempts = value.as_array()?; + if attempts.is_empty() { + return None; + } + attempts + .iter() + .map(|attempt| attempt.get("code").and_then(status_from_value)) + .collect() +} + +fn status_from_value(value: &serde_json::Value) -> Option { + u16::try_from(value.as_u64()?).ok() +} + +fn nonempty_string(value: &serde_json::Value) -> Option { + value + .as_str() + .filter(|value| !value.is_empty()) + .map(str::to_string) +} + +fn scalar_string(value: &serde_json::Value) -> Option { + nonempty_string(value).or_else(|| { + value + .as_i64() + .map(|value| value.to_string()) + .or_else(|| value.as_u64().map(|value| value.to_string())) + }) +} + +const fn is_unambiguous_status_code(status: u16) -> bool { + matches!(status, 401 | 403 | 404 | 408 | 413 | 429 | 500..=599) +} + +const fn is_transient_status_code(status: u16) -> bool { + matches!(status, 408 | 429 | 500..=599) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn exhausted_transient_routes_are_retryable() { + let body = r#"{ + "error": { + "message": "Provider returned error", + "code": 400, + "metadata": { + "raw": "{\"error\":{\"message\":\"Failed to start generation: no model registered\",\"type\":\"invalid_request_error\"}}", + "provider_name": "Together", + "is_byok": false, + "previous_errors": [ + {"code": 429, "provider_name": "Fireworks"}, + {"code": 429, "provider_name": "DigitalOcean"}, + {"code": 429, "provider_name": "BaseTen"} + ] + } + }, + "user_id": "org_test" + }"#; + + let error = decode_http(400, body, "openrouter", None); + assert!(error.retryable()); + assert!(error.failover_eligible()); + match error { + Error::Provider { kind, detail } => { + assert_eq!(kind, ProviderErrorKind::Server); + assert_eq!(detail.message, ROUTING_EXHAUSTED_MESSAGE); + assert_eq!(detail.provider, "openrouter"); + assert_eq!(detail.status_code, Some(400)); + assert_eq!(detail.error_code, None); + assert_eq!(detail.retry_after, None); + assert_eq!(detail.raw, serde_json::from_str(body).ok()); + } + other => panic!("expected provider error, got {other:?}"), + } + } + + #[test] + fn canonical_provider_unavailable_overrides_ambiguous_status() { + let body = r#"{ + "error": { + "code": 400, + "message": "Provider returned an invalid response", + "metadata": { + "error_type": "provider_unavailable", + "provider_code": "no_endpoint" + } + } + }"#; + + let error = decode_http(400, body, "openrouter", None); + assert!(error.retryable()); + match error { + Error::Provider { kind, detail } => { + assert_eq!(kind, ProviderErrorKind::Server); + assert_eq!(detail.error_code.as_deref(), Some("provider_unavailable")); + } + other => panic!("expected provider error, got {other:?}"), + } + } + + #[test] + fn canonical_rate_limit_preserves_provider_detail() { + let body = r#"{ + "error": { + "code": 400, + "message": "Rate limit exceeded", + "metadata": {"error_type": "rate_limit_exceeded"} + } + }"#; + + let error = decode_http(400, body, "openrouter", Some(2.5)); + assert!(error.retryable()); + match error { + Error::Provider { kind, detail } => { + assert_eq!(kind, ProviderErrorKind::RateLimit); + assert_eq!(detail.provider, "openrouter"); + assert_eq!(detail.status_code, Some(400)); + assert_eq!(detail.retry_after, Some(2.5)); + assert_eq!(detail.error_code.as_deref(), Some("rate_limit_exceeded")); + assert_eq!(detail.raw, serde_json::from_str(body).ok()); + } + other => panic!("expected provider error, got {other:?}"), + } + } + + #[test] + fn plain_bad_request_remains_non_retryable() { + let body = r#"{"error":{"message":"Bad request","type":"invalid_request_error"}}"#; + + let error = decode_http(400, body, "openrouter", None); + assert!(!error.retryable()); + assert!(matches!(error, Error::Provider { + kind: ProviderErrorKind::InvalidRequest, + .. + })); + } + + #[test] + fn unambiguous_status_precedes_legacy_route_history() { + let body = serde_json::json!({ + "error": { + "message": "Invalid API key", + "code": 401, + "metadata": { + "previous_errors": [{"code": 429}, {"code": 503}] + } + } + }) + .to_string(); + + let error = decode_http(401, &body, "openrouter", None); + assert!(!error.retryable()); + assert!(matches!(error, Error::Provider { + kind: ProviderErrorKind::Authentication, + .. + })); + } + + #[test] + fn malformed_or_deterministic_previous_attempts_do_not_change_bad_request() { + for previous_errors in [ + serde_json::json!([{"code": 429}, {"message": "missing code"}]), + serde_json::json!([{"code": 429}, {"code": 400}]), + ] { + let body = serde_json::json!({ + "error": { + "message": "Bad request", + "code": 400, + "metadata": {"previous_errors": previous_errors} + } + }) + .to_string(); + + let error = decode_http(400, &body, "openrouter", None); + assert!(!error.retryable()); + assert!(matches!(error, Error::Provider { + kind: ProviderErrorKind::InvalidRequest, + .. + })); + } + } +} diff --git a/lib/components/fabro-llm/src/codec/openai_compatible/mod.rs b/lib/components/fabro-llm/src/codec/openai_compatible/mod.rs index 42a2fde09..8fbb37526 100644 --- a/lib/components/fabro-llm/src/codec/openai_compatible/mod.rs +++ b/lib/components/fabro-llm/src/codec/openai_compatible/mod.rs @@ -2,10 +2,11 @@ //! //! Serves every "OpenAI-compatible" route (kimi, zai, minimax, venice, //! inception, ollama, litellm, …). Pure translation: no HTTP, auth, or base -//! URL — the adapter shell owns those. Count-tokens and error mapping use the -//! `Codec` trait defaults (this dialect has no count route and uses the shared -//! HTTP-status error mapping). +//! URL — the adapter shell owns those. This dialect has no count route. Error +//! mapping also understands the typed provider errors and routing-attempt +//! metadata returned by OpenRouter. +mod error; mod request; mod response; mod stream; @@ -40,4 +41,14 @@ impl Codec for OpenAiCompatible { ) -> Box { Box::new(stream::StreamState::new(ctx, rate_limit)) } + + fn decode_error( + &self, + status: u16, + body: &str, + ctx: &CodecCtx<'_>, + retry_after: Option, + ) -> Error { + error::decode_http(status, body, ctx.provider_name, retry_after) + } } diff --git a/lib/components/fabro-llm/src/codec/openai_compatible/stream.rs b/lib/components/fabro-llm/src/codec/openai_compatible/stream.rs index f389ee086..722c7996b 100644 --- a/lib/components/fabro-llm/src/codec/openai_compatible/stream.rs +++ b/lib/components/fabro-llm/src/codec/openai_compatible/stream.rs @@ -3,6 +3,7 @@ //! Byte reading and `data:` framing live in the transport; this decoder is fed //! already-stripped payloads (including the `[DONE]` sentinel) via `on_event`. +use super::error; use super::translate::{map_finish_reason, parse_tool_arguments}; use super::wire::{AccumulatedToolCall, ReasoningDetails, StreamChunk}; use crate::codec::{CodecCtx, RawEvent, StreamDecoder}; @@ -258,6 +259,12 @@ impl StreamDecoder for StreamState { let chunk: StreamChunk = serde_json::from_str(ev.data) .map_err(|e| Error::stream_error(format!("failed to parse SSE chunk: {e}"), e))?; + if chunk.error.is_some() { + let raw = serde_json::from_str(ev.data) + .map_err(|e| Error::stream_error(format!("failed to parse SSE error: {e}"), e))?; + return Err(error::decode_stream(raw, &self.provider_name)); + } + Ok(self.process_chunk(chunk).unwrap_or_default()) } @@ -277,6 +284,7 @@ impl StreamDecoder for StreamState { mod tests { use super::*; use crate::codec::CodecParams; + use crate::error::ProviderErrorKind; use crate::types::Request; /// Build a decoder through `StreamState::new` (with a minimal request) for @@ -362,6 +370,43 @@ mod tests { assert_eq!(choices[0].finish_reason.as_deref(), Some("stop")); } + #[test] + fn in_band_openrouter_error_uses_provider_error_classifier() { + let mut state = test_state("openrouter", "openai/gpt-4o"); + let data = r#"{ + "id": "gen-test", + "object": "chat.completion.chunk", + "model": "openai/gpt-4o", + "provider": "OpenAI", + "error": { + "code": 502, + "message": "Provider disconnected mid-stream", + "metadata": {"error_type": "provider_unavailable"} + }, + "choices": [{ + "index": 0, + "delta": {"content": ""}, + "finish_reason": "error" + }] + }"#; + + let error = state + .on_event(RawEvent { event: None, data }) + .expect_err("an in-band provider error should fail the stream"); + + assert!(error.retryable()); + match error { + Error::Provider { kind, detail } => { + assert_eq!(kind, ProviderErrorKind::Server); + assert_eq!(detail.provider, "openrouter"); + assert_eq!(detail.status_code, Some(502)); + assert_eq!(detail.error_code.as_deref(), Some("provider_unavailable")); + assert_eq!(detail.raw, serde_json::from_str(data).ok()); + } + other => panic!("expected provider error, got {other:?}"), + } + } + #[test] fn process_text_chunks() { let mut state = test_state("test", "model"); diff --git a/lib/components/fabro-llm/src/codec/openai_compatible/wire.rs b/lib/components/fabro-llm/src/codec/openai_compatible/wire.rs index 3e1de949d..aae3b757b 100644 --- a/lib/components/fabro-llm/src/codec/openai_compatible/wire.rs +++ b/lib/components/fabro-llm/src/codec/openai_compatible/wire.rs @@ -364,6 +364,9 @@ pub(super) struct StreamChunk { pub model: Option, pub choices: Option>, pub usage: Option, + /// Top-level in-band provider error. OpenRouter includes this after the + /// HTTP 200 response has already committed. + pub error: Option, } #[derive(serde::Deserialize)]