fix(agent): harden upstream invalid response retries

docs/Agent/LlmGateway.md

@@ -253,6 +253,7 @@ Gateway 错误建议：
 
 ```text
 model.provider_denied
+model.upstream_invalid_response
 model.unsupported_tool_call
 model.invalid_tool_call
 model.request_build_failed
@@ -262,7 +263,7 @@ model.cancelled
 model.native_metadata_lost
 ```
 
-Agent runtime 会按 Profile `run.modelRetry` 重试 `429` rate limit 与 transient transport/provider availability 错误；payload build、policy denied、response decode、tool call id、native metadata 等契约错误不重试。
+Agent runtime 会按 Profile `run.modelRetry` 重试 `429` rate limit 与 transient transport/provider availability 错误。`model.upstream_invalid_response` 专用于上游响应体不可读或不是合法 provider JSON 的暂态异常，可自动重试；payload build、policy denied、provider 明确拒绝、response schema decode、tool call id、native metadata 等本地或 provider 明确契约错误不重试。
 
 `model.native_metadata_lost` 不应静默降级。
 

docs/Agent/RunEventJournal.md

@@ -173,7 +173,7 @@ run_failed
 }
 ```
 
-- `retryable`：宿主可不询问用户、安全地自动重试。仅在 `RateLimited`/`Transient` 等暂态错误上为 `true`。
+- `retryable`：宿主可不询问用户、安全地自动重试。仅在 `RateLimited`/`Transient` 等暂态错误上为 `true`。例如 `model.upstream_invalid_response` 表示上游响应体本应承载 provider JSON 契约但不可读或不是合法 JSON，属于可重试的 transient invalid response；它不应用于 request build、provider 明确拒绝、tool call / native metadata / response schema 等本地契约错误。
 - `userRetryable`：用户可通过 UI 手动重试。`retryable=true` 时一定为 `true`；此外 `model.tool_call_required`、`agent.tool_after_finish`、`agent.max_tool_rounds_exceeded` 等指令漂移类错误也是 `userRetryable=true`，但 **禁止** 自动重试。若 run 已经成功 `workspace.commit` 过 chat，终态会改为 `run_partial_success`，不会复用 `run_failed.userRetryable`，避免用户在已保留输出上直接重试造成重复消息。
 
 `run_partial_success` 是一个独立终态：当一次 run 已经通过 `workspace.commit` 向 chat 发布过至少一条消息，但后续仍因 drift、dispatch、`workspace.finish` / persistent commit 等错误失败时，executor 会保留这些 host-confirmed chat commit，并写 `run_partial_success`。它不是 clean success：错误仍在 payload 中暴露，`retryable` / `userRetryable` 固定为 `false`，宿主 UI 应以 warning 展示“已保留部分结果”。`workspace.finish` 未干净完成时不会提交 persistent state。

src-tauri/src/application/services/agent_runtime_service/error_payload.rs

@@ -199,6 +199,7 @@ mod tests {
             "openai returned status 200 non-JSON body (generate): expected value at line 1 column 1"
         );
         assert_eq!(payload["retryable"], true);
+        assert_eq!(payload["userRetryable"], true);
     }
 
     #[test]

src-tauri/src/infrastructure/apis/http_chat_completion_repository/claude.rs

@@ -11,8 +11,8 @@ use crate::domain::repositories::chat_completion_repository::{
 };
 
 use super::HttpChatCompletionRepository;
-use super::body_preview::read_upstream_json_body;
 use super::normalizers;
+use super::response_body::read_upstream_json_body;
 
 const ANTHROPIC_VERSION: &str = "2023-06-01";
 const ANTHROPIC_BETA_OUTPUT_128K: &str = "output-128k-2025-02-19";

src-tauri/src/infrastructure/apis/http_chat_completion_repository/cohere.rs

@@ -7,7 +7,7 @@ use crate::domain::repositories::chat_completion_repository::{
 };
 
 use super::HttpChatCompletionRepository;
-use super::body_preview::read_upstream_json_body;
+use super::response_body::read_upstream_json_body;
 
 pub(super) async fn list_models(
     repository: &HttpChatCompletionRepository,

src-tauri/src/infrastructure/apis/http_chat_completion_repository/gemini_interactions.rs

@@ -12,8 +12,8 @@ use crate::domain::repositories::chat_completion_repository::{
 };
 
 use super::HttpChatCompletionRepository;
-use super::body_preview::read_upstream_json_body;
 use super::normalizers;
+use super::response_body::read_upstream_json_body;
 
 const GEMINI_API_VERSION: &str = "v1beta";
 

src-tauri/src/infrastructure/apis/http_chat_completion_repository/makersuite.rs

@@ -8,8 +8,8 @@ use crate::domain::repositories::chat_completion_repository::{
 };
 
 use super::HttpChatCompletionRepository;
-use super::body_preview::read_upstream_json_body;
 use super::normalizers;
+use super::response_body::read_upstream_json_body;
 
 const GEMINI_API_VERSION: &str = "v1beta";
 

src-tauri/src/infrastructure/apis/http_chat_completion_repository/mod.rs

@@ -13,7 +13,6 @@ use crate::domain::repositories::chat_completion_repository::{
 };
 use crate::infrastructure::http_client_pool::{HttpClientPool, HttpClientProfile};
 
-mod body_preview;
 mod claude;
 mod cohere;
 mod gemini_interactions;
@@ -21,6 +20,7 @@ mod makersuite;
 mod normalizers;
 mod openai;
 mod openai_responses;
+mod response_body;
 mod vertexai;
 
 #[derive(Debug, Clone, Copy)]

src-tauri/src/infrastructure/apis/http_chat_completion_repository/openai.rs

@@ -7,7 +7,7 @@ use crate::domain::repositories::chat_completion_repository::{
 };
 
 use super::HttpChatCompletionRepository;
-use super::body_preview::read_upstream_json_body;
+use super::response_body::read_upstream_json_body;
 
 pub(super) async fn list_models(
     repository: &HttpChatCompletionRepository,

src-tauri/src/infrastructure/apis/http_chat_completion_repository/openai_responses.rs

@@ -19,11 +19,15 @@ use crate::domain::repositories::chat_completion_repository::{
 };
 
 use super::HttpChatCompletionRepository;
-use super::body_preview::{log_upstream_body_parse_failure, read_upstream_json_body};
 use super::normalizers;
+use super::response_body::{log_upstream_body_parse_failure, read_upstream_json_body};
 
 type ResponsesWsStream = tokio_tungstenite::WebSocketStream<reqwest::Upgraded>;
 
+const OPERATION_GENERATE_WS: &str = "generate_ws";
+const OPERATION_GENERATE_STREAM_WS: &str = "generate_stream_ws";
+const OPERATION_GENERATE_PERSISTENT_WS: &str = "generate_persistent_ws";
+
 #[derive(Default)]
 pub(super) struct ResponsesWsSessionPool {
     sessions: Mutex<HashMap<String, Arc<Mutex<ResponsesWsSession>>>>,
@@ -629,12 +633,17 @@ impl ResponsesWsSession {
 
             match message {
                 Message::Text(text) => {
-                    if let Some(response) = response_from_ws_payload(text.as_str().as_bytes())? {
+                    if let Some(response) = response_from_ws_payload(
+                        text.as_str().as_bytes(),
+                        OPERATION_GENERATE_PERSISTENT_WS,
+                    )? {
                         return Ok(response);
                     }
                 }
                 Message::Binary(bytes) => {
-                    if let Some(response) = response_from_ws_payload(bytes.as_ref())? {
+                    if let Some(response) =
+                        response_from_ws_payload(bytes.as_ref(), OPERATION_GENERATE_PERSISTENT_WS)?
+                    {
                         return Ok(response);
                     }
                 }
@@ -687,12 +696,16 @@ async fn generate_ws(
 
         match message {
             Message::Text(text) => {
-                if let Some(response) = response_from_ws_payload(text.as_str().as_bytes())? {
+                if let Some(response) =
+                    response_from_ws_payload(text.as_str().as_bytes(), OPERATION_GENERATE_WS)?
+                {
                     return Ok(normalizers::normalize_openai_responses_response(response));
                 }
             }
             Message::Binary(bytes) => {
-                if let Some(response) = response_from_ws_payload(bytes.as_ref())? {
+                if let Some(response) =
+                    response_from_ws_payload(bytes.as_ref(), OPERATION_GENERATE_WS)?
+                {
                     return Ok(normalizers::normalize_openai_responses_response(response));
                 }
             }
@@ -824,9 +837,11 @@ fn forward_ws_stream_event(
     sender: &ChatCompletionStreamSender,
     payload: &[u8],
 ) -> Result<(), ResponsesWsStreamError> {
-    let event = parse_ws_event(payload).map_err(|error| ResponsesWsStreamError {
-        error,
-        emitted: state.has_emitted(),
+    let event = parse_ws_event(payload, OPERATION_GENERATE_STREAM_WS).map_err(|error| {
+        ResponsesWsStreamError {
+            error,
+            emitted: state.has_emitted(),
+        }
     })?;
     let event_type = event
         .get("type")
@@ -1070,8 +1085,8 @@ fn provider_session_id(payload: &Value) -> Result<Option<String>, DomainError> {
     Ok(Some(session_id.to_string()))
 }
 
-fn response_from_ws_payload(payload: &[u8]) -> Result<Option<Value>, DomainError> {
-    let event = parse_ws_event(payload)?;
+fn response_from_ws_payload(payload: &[u8], operation: &str) -> Result<Option<Value>, DomainError> {
+    let event = parse_ws_event(payload, operation)?;
     let event_type = event
         .get("type")
         .and_then(Value::as_str)
@@ -1091,18 +1106,18 @@ fn response_from_ws_payload(payload: &[u8]) -> Result<Option<Value>, DomainError
     }
 }
 
-fn parse_ws_event(payload: &[u8]) -> Result<Value, DomainError> {
+fn parse_ws_event(payload: &[u8], operation: &str) -> Result<Value, DomainError> {
     serde_json::from_slice(payload).map_err(|error| {
         log_upstream_body_parse_failure(
             "OpenAI Responses",
-            "generate_stream_ws",
+            operation,
             StatusCode::SWITCHING_PROTOCOLS,
             "application/json",
             payload,
             &error,
         );
         DomainError::transient(format!(
-            "model.upstream_invalid_response: OpenAI Responses WebSocket event is not valid JSON: {error}"
+            "model.upstream_invalid_response: OpenAI Responses WebSocket event is not valid JSON ({operation}): {error}"
         ))
     })
 }
@@ -1255,4 +1270,16 @@ mod tests {
         assert_eq!(error.error.to_string(), "Internal error: unsupported ws");
         assert!(receiver.try_recv().is_err());
     }
+
+    #[test]
+    fn ws_invalid_json_event_is_transient_and_preserves_operation() {
+        let error = parse_ws_event(b"not-json", OPERATION_GENERATE_WS)
+            .expect_err("invalid websocket event should fail");
+
+        assert!(matches!(error, DomainError::Transient(_)));
+        assert_eq!(
+            error.to_string(),
+            "model.upstream_invalid_response: OpenAI Responses WebSocket event is not valid JSON (generate_ws): expected ident at line 1 column 2"
+        );
+    }
 }

src-tauri/src/infrastructure/apis/http_chat_completion_repository/response_body.rs

@@ -46,8 +46,9 @@ pub(super) fn body_preview_string(body: &[u8]) -> String {
 }
 
 /// Read an upstream HTTP response body and parse it as JSON, logging a
-/// detailed diagnostic event on failure. Caller is responsible for ensuring
-/// the response status was 2xx before invoking this helper.
+/// detailed diagnostic event on failure. Callers should classify explicit HTTP
+/// error statuses before invoking this helper; this path is for responses whose
+/// body was expected to carry the provider JSON contract.
 ///
 /// Both body-read failures and JSON-decode failures are reported as
 /// [`DomainError::Transient`] tagged with `model.upstream_invalid_response`,
@@ -74,15 +75,31 @@ pub(super) async fn read_upstream_json_body(
         ))
     })?;
 
-    match serde_json::from_slice::<Value>(&body) {
+    parse_upstream_json_body(
+        provider_name,
+        operation,
+        status,
+        &content_type,
+        body.as_ref(),
+    )
+}
+
+pub(super) fn parse_upstream_json_body(
+    provider_name: &str,
+    operation: &str,
+    status: StatusCode,
+    content_type: &str,
+    body: &[u8],
+) -> Result<Value, DomainError> {
+    match serde_json::from_slice::<Value>(body) {
         Ok(value) => Ok(value),
         Err(error) => {
             log_upstream_body_parse_failure(
                 provider_name,
                 operation,
                 status,
-                &content_type,
-                &body,
+                content_type,
+                body,
                 &error,
             );
             Err(DomainError::transient(format!(
@@ -117,4 +134,36 @@ mod tests {
         let preview = body_preview_string(&body);
         assert!(preview.contains("OK"));
     }
+
+    #[test]
+    fn parse_upstream_json_body_returns_transient_invalid_response_for_non_json_body() {
+        let error = parse_upstream_json_body(
+            "OpenAI",
+            "generate",
+            StatusCode::OK,
+            "text/html",
+            b"<html>gateway challenge</html>",
+        )
+        .expect_err("non-json body should be transient");
+
+        assert!(matches!(error, DomainError::Transient(_)));
+        assert_eq!(
+            error.to_string(),
+            "model.upstream_invalid_response: OpenAI returned status 200 non-JSON body (generate): expected value at line 1 column 1"
+        );
+    }
+
+    #[test]
+    fn parse_upstream_json_body_accepts_valid_json_even_with_imprecise_content_type() {
+        let value = parse_upstream_json_body(
+            "OpenAI",
+            "generate",
+            StatusCode::OK,
+            "text/plain",
+            br#"{"ok":true}"#,
+        )
+        .expect("valid JSON should parse");
+
+        assert_eq!(value["ok"], true);
+    }
 }

src-tauri/src/infrastructure/apis/http_chat_completion_repository/vertexai.rs

@@ -8,8 +8,8 @@ use crate::domain::repositories::chat_completion_repository::{
 };
 
 use super::HttpChatCompletionRepository;
-use super::body_preview::read_upstream_json_body;
 use super::normalizers;
+use super::response_body::read_upstream_json_body;
 
 const PROVIDER_NAME: &str = "Google Vertex AI";