From bf2038fd433b575aec134bfa952ab0b92b108612 Mon Sep 17 00:00:00 2001 From: Bob Lee Date: Mon, 10 Aug 2026 19:05:33 -0700 Subject: [PATCH 1/2] fix(ai): retry every provider error --- src/crates/adapters/ai-adapters/src/client.rs | 193 ++++++---------- .../adapters/ai-adapters/src/client/sse.rs | 211 +++--------------- 2 files changed, 95 insertions(+), 309 deletions(-) diff --git a/src/crates/adapters/ai-adapters/src/client.rs b/src/crates/adapters/ai-adapters/src/client.rs index 53bdaa94c..951e29d75 100644 --- a/src/crates/adapters/ai-adapters/src/client.rs +++ b/src/crates/adapters/ai-adapters/src/client.rs @@ -323,19 +323,19 @@ impl AIClient { .await; return Ok(response); } - Err(error) - if attempt < max_attempts - 1 - && is_transient_stream_error(&error.to_string()) => - { + Err(error) => { fail_aggregated_trace( trace.as_ref(), trace_handle.as_ref(), &error.to_string(), ) .await; + if attempt == max_attempts - 1 { + return Err(error); + } let delay_ms = send_message_retry_delay_ms(attempt, &error.to_string()); warn!( - "Retrying aggregated AI stream after transient error: attempt={}/{}, delay_ms={}, error={}", + "Retrying aggregated AI stream after error: attempt={}/{}, delay_ms={}, error={}", attempt + 1, max_attempts, delay_ms, @@ -343,15 +343,6 @@ impl AIClient { ); tokio::time::sleep(Duration::from_millis(delay_ms)).await; } - Err(error) => { - fail_aggregated_trace( - trace.as_ref(), - trace_handle.as_ref(), - &error.to_string(), - ) - .await; - return Err(error); - } } } @@ -447,92 +438,6 @@ fn send_message_retry_delay_ms(attempt_index: usize, error_message: &str) -> u64 } } -fn is_transient_stream_error(error_message: &str) -> bool { - let msg = error_message.to_lowercase(); - - let non_retryable_keywords = [ - "invalid api key", - "unauthorized", - "forbidden", - "model not found", - "unsupported model", - "invalid request", - "bad request", - "prompt is too long", - "content policy", - "proxy authentication required", - "provider quota", - "provider billing", - "insufficient_quota", - "insufficient quota", - "insufficient balance", - "not_enough_balance", - "not enough balance", - "余额不足", - "无可用资源包", - "账户已欠费", - "code=1113", - "\"code\":\"1113\"", - "client error 400", - "client error 401", - "client error 402", - "client error 403", - "client error 404", - "client error 413", - "client error 422", - "sse parsing error", - "schema error", - "unknown api format", - ]; - - if non_retryable_keywords.iter().any(|k| msg.contains(k)) { - return false; - } - - [ - "transport error", - "error decoding response body", - "stream closed before response completed", - "stream processing error", - "sse stream error", - "sse error", - "sse timeout", - "stream data timeout", - "timeout", - "request timeout", - "deadline exceeded", - "connection reset", - "connection closed", - "broken pipe", - "unexpected eof", - "connection refused", - "socket closed", - "temporarily unavailable", - "service unavailable", - "bad gateway", - "gateway timeout", - "overloaded", - "proxy", - "tunnel", - "dns", - "network", - "econnreset", - "econnrefused", - "etimedout", - "rate limit", - "too many requests", - "408", - "409", - "425", - "429", - "502", - "503", - "504", - ] - .iter() - .any(|k| msg.contains(k)) -} - async fn complete_aggregated_trace( trace_config: Option<&ModelExchangeTraceConfig>, trace_handle: Option<&ModelExchangeRequestTraceHandle>, @@ -584,11 +489,43 @@ fn gemini_response_to_trace(response: &GeminiResponse) -> ModelExchangeResponseT #[cfg(test)] mod tests { - use super::{is_transient_stream_error, send_message_retry_delay_ms, AIClient}; + use super::{send_message_retry_delay_ms, AIClient}; use crate::providers::{anthropic, gemini, gemini::GeminiMessageConverter, openai}; use crate::types::{AIConfig, ToolDefinition}; use crate::types::{ReasoningPresetAction, ReasoningPresetDescriptor}; + use axum::extract::State; + use axum::http::header::CONTENT_TYPE; + use axum::response::IntoResponse; + use axum::routing::post; + use axum::Router; use serde_json::{json, Value}; + use std::sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, + }; + + #[derive(Clone)] + struct StreamRetryFixtureState { + attempts: Arc, + } + + async fn malformed_stream_then_success( + State(state): State, + ) -> impl IntoResponse { + let payload = if state.attempts.fetch_add(1, Ordering::SeqCst) == 0 { + "data: not-json\n\n" + } else { + concat!( + "data: {\"id\":\"chatcmpl_test\",\"object\":\"chat.completion.chunk\",", + "\"created\":1,\"model\":\"test-model\",\"choices\":[{\"index\":0,", + "\"delta\":{\"content\":\"Recovered\"},\"finish_reason\":\"stop\"}],", + "\"usage\":null}\n\n", + "data: [DONE]\n\n" + ) + }; + + ([(CONTENT_TYPE, "text/event-stream")], payload) + } fn make_test_client(format: &str, custom_request_body: Option) -> AIClient { AIClient::new(AIConfig { @@ -2255,20 +2192,32 @@ mod tests { assert_eq!(request.timeout(), None); } - #[test] - fn aggregated_send_message_retries_transient_stream_errors() { - for msg in [ - "SSE Error: stream closed before response completed", - "Transport Error: error decoding response body", - "Anthropic API is temporarily overloaded", - "Gemini SSE stream timeout after 60s", - "OpenAI Streaming API error 503: service unavailable", - ] { - assert!( - is_transient_stream_error(msg), - "expected transient stream error: {msg}" - ); - } + #[tokio::test] + async fn aggregated_send_message_retries_every_stream_error() { + let attempts = Arc::new(AtomicUsize::new(0)); + let app = Router::new() + .route("/chat/completions", post(malformed_stream_then_success)) + .with_state(StreamRetryFixtureState { + attempts: Arc::clone(&attempts), + }); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind stream retry fixture"); + let address = listener.local_addr().expect("stream retry fixture address"); + let server_task = tokio::spawn(async move { + axum::serve(listener, app) + .await + .expect("stream retry fixture should run"); + }); + let mut client = make_test_client("openai", None); + client.config.request_url = format!("http://{address}/chat/completions"); + + let result = client.send_test_message(Vec::new(), None, 2).await; + + server_task.abort(); + let response = result.expect("the second stream attempt should succeed"); + assert_eq!(response.text, "Recovered"); + assert_eq!(attempts.load(Ordering::SeqCst), 2); } #[test] @@ -2292,18 +2241,4 @@ mod tests { ); assert_eq!(send_message_retry_delay_ms(5, "too many requests"), 60_000); } - - #[test] - fn aggregated_send_message_does_not_retry_permanent_errors() { - for msg in [ - "OpenAI Streaming API client error 401: unauthorized", - "SSE Parsing Error: missing field choices", - "Provider error: provider=glm, code=1113, message=余额不足或无可用资源包", - ] { - assert!( - !is_transient_stream_error(msg), - "expected permanent stream error: {msg}" - ); - } - } } diff --git a/src/crates/adapters/ai-adapters/src/client/sse.rs b/src/crates/adapters/ai-adapters/src/client/sse.rs index 33825ef41..91773cea9 100644 --- a/src/crates/adapters/ai-adapters/src/client/sse.rs +++ b/src/crates/adapters/ai-adapters/src/client/sse.rs @@ -3,7 +3,7 @@ use crate::client::StreamResponse; use crate::stream::UnifiedResponse; use crate::trace::{ModelExchangeRequestAttempt, ModelExchangeTraceConfig}; use anyhow::{anyhow, Result}; -use bitfun_core_types::errors::{AiProviderError, ErrorCategory}; +use bitfun_core_types::errors::AiProviderError; use chrono::{DateTime, Utc}; use futures::Stream; use log::{debug, error, warn}; @@ -101,81 +101,6 @@ fn format_transport_error(label: &str, error: &reqwest::Error) -> String { message } -fn is_retryable_http_status(status: StatusCode) -> bool { - status.is_server_error() || matches!(status.as_u16(), 408 | 409 | 425 | 429) -} - -fn is_retryable_error_category(category: &ErrorCategory) -> bool { - matches!( - category, - ErrorCategory::Network - | ErrorCategory::RateLimit - | ErrorCategory::Timeout - | ErrorCategory::ProviderUnavailable - ) -} - -fn provider_error_message(body: &str) -> Option { - let value: serde_json::Value = serde_json::from_str(body).ok()?; - let error = value.get("error").unwrap_or(&value); - error - .get("message") - .or_else(|| error.get("error")) - .and_then(serde_json::Value::as_str) - .map(str::to_string) -} - -/// Some OpenAI-compatible gateways occasionally lose their resolved model -/// while routing an otherwise valid request, then report `model=None` or -/// `model=null` as a 400. Treat that response as transient only when it -/// contradicts the concrete model BitFun actually sent. A genuine invalid -/// model name still remains a non-retryable client error. -fn provider_lost_request_model( - status: StatusCode, - request_body: &serde_json::Value, - error_text: &str, -) -> bool { - if status != StatusCode::BAD_REQUEST { - return false; - } - - let has_concrete_model = request_body - .get("model") - .and_then(serde_json::Value::as_str) - .map(str::trim) - .filter(|model| { - !model.is_empty() - && !model.eq_ignore_ascii_case("none") - && !model.eq_ignore_ascii_case("null") - }) - .is_some(); - if !has_concrete_model { - return false; - } - - let Some(message) = provider_error_message(error_text) else { - return false; - }; - let message = message.to_ascii_lowercase(); - let reports_missing_model = ["model=none", "model = none", "model=null", "model = null"] - .iter() - .any(|marker| message.contains(marker)); - - reports_missing_model - && (message.contains("invalid model") || message.contains("missing model")) -} - -fn is_retryable_http_failure( - status: StatusCode, - request_body: &serde_json::Value, - error_text: &str, - provider_error: &AiProviderError, -) -> bool { - is_retryable_http_status(status) - || is_retryable_error_category(&provider_error.category) - || provider_lost_request_model(status, request_body, error_text) -} - fn provider_error_code(body: &str) -> Option { let value: serde_json::Value = serde_json::from_str(body).ok()?; let error = value.get("error").unwrap_or(&value); @@ -358,33 +283,13 @@ where .text() .await .unwrap_or_else(|e| format!("Failed to read error response: {}", e)); - let error_kind = - if status.is_client_error() && !is_retryable_http_status(status) { - "client error" - } else { - "error" - }; + let error_kind = if status.is_client_error() { + "client error" + } else { + "error" + }; let provider_error = http_provider_error(label, status, &error_text, error_kind); - let retryable = is_retryable_http_failure( - status, - request_body, - &error_text, - &provider_error, - ); - if provider_error.category == ErrorCategory::ContextOverflow || !retryable { - if let Some(trace) = trace.as_ref() { - trace - .sink - .request_attempt_failed( - trace_handle.as_ref(), - &provider_error.to_string(), - ) - .await; - } - error!("{}", provider_error); - return Err(anyhow!(provider_error)); - } let error = anyhow!(provider_error); warn!( "{} request failed: {}ms, transport_attempt {}/{}, error: {}", @@ -525,6 +430,7 @@ mod tests { use axum::response::IntoResponse; use axum::routing::post; use axum::{Json, Router}; + use bitfun_core_types::errors::ErrorCategory; use reqwest::header::HeaderValue; use std::sync::{ atomic::{AtomicBool, AtomicUsize, Ordering}, @@ -536,26 +442,35 @@ mod tests { attempts: Arc, } - async fn invalid_model_then_success( + async fn bad_requests_then_success( State(state): State, Json(body): Json, ) -> impl IntoResponse { assert_eq!(body["model"], "configured-model"); - if state.attempts.fetch_add(1, Ordering::SeqCst) == 0 { - ( + match state.attempts.fetch_add(1, Ordering::SeqCst) { + 0 => ( StatusCode::BAD_REQUEST, Json(serde_json::json!({ "error": { - "message": "/chat/completions: Invalid model name passed in model=None. Call `/v1/models` to view available models for your key.", - "type": "None", - "param": "None", - "code": "400" + "message": "Invalid temperature value", + "type": "invalid_request_error", + "code": "invalid_parameter" } })), ) - .into_response() - } else { - StatusCode::OK.into_response() + .into_response(), + 1 => ( + StatusCode::BAD_REQUEST, + Json(serde_json::json!({ + "error": { + "message": "Maximum context length exceeded", + "type": "invalid_request_error", + "code": "context_length_exceeded" + } + })), + ) + .into_response(), + _ => StatusCode::OK.into_response(), } } @@ -632,75 +547,11 @@ mod tests { assert!(observed_cancel.load(Ordering::SeqCst)); } - #[test] - fn retryable_http_statuses_include_rate_limit_and_server_errors() { - assert!(is_retryable_http_status(StatusCode::TOO_MANY_REQUESTS)); - assert!(is_retryable_http_status(StatusCode::REQUEST_TIMEOUT)); - assert!(is_retryable_http_status(StatusCode::INTERNAL_SERVER_ERROR)); - assert!(is_retryable_http_status(StatusCode::BAD_GATEWAY)); - - assert!(!is_retryable_http_status(StatusCode::UNAUTHORIZED)); - assert!(!is_retryable_http_status(StatusCode::BAD_REQUEST)); - assert!(!is_retryable_http_status(StatusCode::NOT_FOUND)); - } - - #[test] - fn provider_model_loss_requires_a_contradictory_concrete_request_model() { - let error_text = serde_json::json!({ - "error": { - "message": "Invalid model name passed in model=None", - "code": "400" - } - }) - .to_string(); - - assert!(provider_lost_request_model( - StatusCode::BAD_REQUEST, - &serde_json::json!({"model": "configured-model"}), - &error_text, - )); - assert!(!provider_lost_request_model( - StatusCode::BAD_REQUEST, - &serde_json::json!({"model": "None"}), - &error_text, - )); - assert!(!provider_lost_request_model( - StatusCode::BAD_REQUEST, - &serde_json::json!({}), - &error_text, - )); - assert!(!provider_lost_request_model( - StatusCode::NOT_FOUND, - &serde_json::json!({"model": "configured-model"}), - &error_text, - )); - } - - #[test] - fn structured_transient_code_overrides_a_nonstandard_client_status() { - let error_text = - r#"{"error":{"code":"server_error","message":"temporary routing failure"}}"#; - let provider_error = http_provider_error( - "OpenAI Streaming API", - StatusCode::BAD_REQUEST, - error_text, - "client error", - ); - - assert_eq!(provider_error.category, ErrorCategory::ProviderUnavailable); - assert!(is_retryable_http_failure( - StatusCode::BAD_REQUEST, - &serde_json::json!({"model": "configured-model"}), - error_text, - &provider_error, - )); - } - #[tokio::test] - async fn contradictory_missing_model_response_uses_existing_retry_loop() { + async fn every_bad_request_uses_existing_retry_loop() { let attempts = Arc::new(AtomicUsize::new(0)); let app = Router::new() - .route("/chat/completions", post(invalid_model_then_success)) + .route("/chat/completions", post(bad_requests_then_success)) .with_state(RetryFixtureState { attempts: Arc::clone(&attempts), }); @@ -721,7 +572,7 @@ mod tests { "OpenAI Streaming API", &url, &request_body, - 2, + 3, None, None, || client.post(&url), @@ -734,9 +585,9 @@ mod tests { server_task.abort(); assert!( result.is_ok(), - "the second transport attempt should succeed" + "ordinary and context-overflow 400 responses should both retry" ); - assert_eq!(attempts.load(Ordering::SeqCst), 2); + assert_eq!(attempts.load(Ordering::SeqCst), 3); } #[test] From 5a9479fe0cafa585f16966fc58f222f7931efe23 Mon Sep 17 00:00:00 2001 From: Bob Lee Date: Mon, 10 Aug 2026 19:18:34 -0700 Subject: [PATCH 2/2] test(cli): cover exhaustive provider retries --- .../cli/tests/cli_command_contracts/exec_cli_contracts.rs | 8 ++++---- src/apps/cli/tests/support/mod.rs | 2 +- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/src/apps/cli/tests/cli_command_contracts/exec_cli_contracts.rs b/src/apps/cli/tests/cli_command_contracts/exec_cli_contracts.rs index 48862de23..1f3f9defe 100644 --- a/src/apps/cli/tests/cli_command_contracts/exec_cli_contracts.rs +++ b/src/apps/cli/tests/cli_command_contracts/exec_cli_contracts.rs @@ -380,7 +380,7 @@ fn stream_json_provider_http_403_emits_one_error_terminal() { "stream-json", ]); let output = command_output_with_timeout(&mut command, std::time::Duration::from_secs(30)); - server.assert_chat_completion_requests(1); + server.assert_chat_completion_requests(10); let stdout = stdout(&output); assert!(!output.status.success(), "{stdout}"); @@ -443,7 +443,7 @@ fn stream_json_provider_and_patch_failures_publish_one_final_classification() { &output_target, ]); let output = command_output_with_timeout(&mut command, std::time::Duration::from_secs(30)); - server.assert_chat_completion_requests(1); + server.assert_chat_completion_requests(10); let stdout = stdout(&output); let stderr = stderr(&output); @@ -481,7 +481,7 @@ fn stream_json_provider_and_patch_failures_publish_one_final_classification() { } #[test] -fn stream_json_disconnect_then_permanent_retry_failure_emits_one_error_terminal() { +fn stream_json_disconnect_then_exhausted_retry_failure_emits_one_error_terminal() { let server = MockOpenAiServer::disconnect_then_http_403(); let environment = CliTestEnvironment::new(); environment.configure_mock_model(server.base_url()); @@ -493,7 +493,7 @@ fn stream_json_disconnect_then_permanent_retry_failure_emits_one_error_terminal( "stream-json", ]); let output = command_output_with_timeout(&mut command, std::time::Duration::from_secs(30)); - server.assert_chat_completion_requests(2); + server.assert_chat_completion_requests(11); let stdout = stdout(&output); assert!(!output.status.success(), "{stdout}"); diff --git a/src/apps/cli/tests/support/mod.rs b/src/apps/cli/tests/support/mod.rs index b8c634b31..0fd491a80 100644 --- a/src/apps/cli/tests/support/mod.rs +++ b/src/apps/cli/tests/support/mod.rs @@ -579,7 +579,7 @@ fn write_http_403(stream: &mut TcpStream, reason: &str) -> std::io::Result<()> { .to_string(); write!( stream, - "HTTP/1.1 403 Forbidden\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + "HTTP/1.1 403 Forbidden\r\nContent-Type: application/json\r\nContent-Length: {}\r\nRetry-After: 1\r\nConnection: close\r\n\r\n{body}", body.len() )?; stream.flush()