fix(ai): retry every provider error

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}");

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()

src/crates/adapters/ai-adapters/src/client.rs

@@ -323,35 +323,26 @@ 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,
                         error
                     );
                     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<AtomicUsize>,
+    }
+
+    async fn malformed_stream_then_success(
+        State(state): State<StreamRetryFixtureState>,
+    ) -> 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<Value>) -> 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}"
-            );
-        }
-    }
 }