fix(kiro): 修复流响应识别与号池配额刷新 Token 续期问题 (#322)

* fix(kiro): 兼容 application/json 头下的 eventstream 流响应

- stream_pump: 对带 kiro:generateAssistantResponse envelope 的上游响应按流处理，不再仅依赖 text/event-stream 头判断\n- tests: 补充 application/json 头下仍识别为 Kiro stream 的回归测试

* fix(kiro): 修复号池配额刷新误判封禁

apps/aether-gateway/src/execution_runtime/stream_pump.rs

@@ -60,7 +60,7 @@ pub(crate) fn build_direct_execution_frame_stream(
         let mut stream_terminal_observer = StreamingStandardTerminalObserver::default();
         let mut observer_buffered = Vec::new();
 
-        if !response_headers_indicate_sse(&headers) {
+        if !should_treat_upstream_response_as_stream(&headers, &observer_context) {
             let original_headers = headers.clone();
             match buffer_non_sse_upstream_body(response, started_at).await {
                 Ok(buffered) => {
@@ -388,6 +388,20 @@ fn response_headers_indicate_sse(headers: &BTreeMap<String, String>) -> bool {
         .is_some_and(|value| value.to_ascii_lowercase().contains("text/event-stream"))
 }
 
+fn should_treat_upstream_response_as_stream(
+    headers: &BTreeMap<String, String>,
+    report_context: &Value,
+) -> bool {
+    if response_headers_indicate_sse(headers) {
+        return true;
+    }
+
+    report_context
+        .get("envelope_name")
+        .and_then(Value::as_str)
+        .is_some_and(|value| value.eq_ignore_ascii_case(crate::ai_pipeline::KIRO_ENVELOPE_NAME))
+}
+
 async fn buffer_non_sse_upstream_body(
     response: DirectUpstreamResponse,
     started_at: Instant,
@@ -648,7 +662,7 @@ mod tests {
     use serde_json::Value;
     use tokio::sync::watch;
 
-    use super::build_direct_execution_frame_stream;
+    use super::{build_direct_execution_frame_stream, should_treat_upstream_response_as_stream};
     use crate::execution_runtime::transport::{
         execute_stream_plan_via_local_tunnel, DirectSyncExecutionRuntime, DirectUpstreamResponse,
     };
@@ -666,6 +680,19 @@ mod tests {
         }
     }
 
+    #[test]
+    fn treats_kiro_eventstream_envelope_as_stream_even_when_content_type_is_json() {
+        let headers = BTreeMap::from([("content-type".into(), "application/json".into())]);
+        let report_context = serde_json::json!({
+            "envelope_name": "kiro:generateAssistantResponse",
+        });
+
+        assert!(should_treat_upstream_response_as_stream(
+            &headers,
+            &report_context
+        ));
+    }
+
     #[tokio::test]
     async fn direct_execution_frame_stream_reports_ttfb_after_first_upstream_chunk() {
         let listener = crate::test_support::bind_loopback_listener()

apps/aether-gateway/src/handlers/admin/provider/oauth/quota/kiro/mod.rs

@@ -8,15 +8,83 @@ use super::shared::{
     persist_provider_quota_refresh_state, quota_refresh_success_invalid_state,
     ProviderQuotaExecutionOutcome,
 };
-use crate::handlers::admin::request::AdminAppState;
+use crate::handlers::admin::request::{AdminAppState, AdminLocalOAuthRefreshError};
 use crate::GatewayError;
 use aether_contracts::ProxySnapshot;
 use aether_data_contracts::repository::provider_catalog::{
     StoredProviderCatalogEndpoint, StoredProviderCatalogKey, StoredProviderCatalogProvider,
 };
+use aether_provider_transport::kiro::build_kiro_request_auth_from_config;
+use aether_provider_transport::{CachedOAuthEntry, LocalResolvedOAuthRequestAuth};
 use serde_json::json;
 use std::time::{SystemTime, UNIX_EPOCH};
 
+fn kiro_quota_error_is_token_invalid(detail: Option<&str>) -> bool {
+    let Some(detail) = detail else {
+        return false;
+    };
+    let normalized = detail.to_ascii_lowercase();
+    normalized.contains("bearer token invalid")
+        || normalized.contains("bearer token invild")
+        || normalized.contains("bearer token is invalid")
+        || normalized.contains("invalid bearer token")
+        || normalized.contains("token expired")
+        || normalized.contains("token has expired")
+        || normalized.contains("expired token")
+}
+
+fn kiro_quota_error_is_account_banned(detail: Option<&str>) -> bool {
+    let Some(detail) = detail else {
+        return false;
+    };
+    let normalized = detail.to_ascii_lowercase();
+    [
+        "account suspended",
+        "account is suspended",
+        "account banned",
+        "account is banned",
+        "terms of service",
+        "封禁",
+        "封号",
+        "被封",
+        "账户已封禁",
+    ]
+    .iter()
+    .any(|keyword| normalized.contains(keyword))
+}
+
+fn kiro_auth_from_refreshed_entry(
+    entry: &CachedOAuthEntry,
+) -> Option<LocalResolvedOAuthRequestAuth> {
+    if !entry.provider_type.trim().eq_ignore_ascii_case("kiro") {
+        return None;
+    }
+    let auth_config = entry
+        .metadata
+        .as_ref()
+        .and_then(aether_provider_transport::kiro::KiroAuthConfig::from_json_value)?;
+    let auth = build_kiro_request_auth_from_config(auth_config, None)?;
+    Some(LocalResolvedOAuthRequestAuth::Kiro(auth))
+}
+
+fn kiro_quota_refresh_failure_status(err: &AdminLocalOAuthRefreshError) -> Option<u16> {
+    match err {
+        AdminLocalOAuthRefreshError::HttpStatus { status_code, .. } => Some(*status_code),
+        _ => None,
+    }
+}
+
+fn kiro_quota_refresh_failure_message(err: &AdminLocalOAuthRefreshError) -> String {
+    match err {
+        AdminLocalOAuthRefreshError::HttpStatus {
+            status_code,
+            body_excerpt,
+            ..
+        } => format!("Kiro Token 刷新失败 ({status_code}): {body_excerpt}"),
+        _ => format!("Kiro Token 刷新失败: {err}"),
+    }
+}
+
 pub(crate) async fn refresh_kiro_provider_quota_locally(
     state: &AdminAppState<'_>,
     provider: &StoredProviderCatalogProvider,
@@ -46,10 +114,26 @@ pub(crate) async fn refresh_kiro_provider_quota_locally(
             }
         };
 
-        let Some(auth) = state
+        let auth = match state.force_local_oauth_refresh_entry(&transport).await {
+            Ok(Some(entry)) => match kiro_auth_from_refreshed_entry(&entry) {
+                Some(LocalResolvedOAuthRequestAuth::Kiro(auth)) => auth,
+                _ => {
+                    failed_count += 1;
+                    results.push(json!({
+                        "key_id": key.id,
+                        "key_name": key.name,
+                        "status": "error",
+                        "message": "Kiro Token 刷新成功但认证信息解析失败",
+                    }));
+                    continue;
+                }
+            },
+            Ok(None) => match state
                 .resolve_local_oauth_kiro_request_auth(&transport)
                 .await?
-        else {
+            {
+                Some(auth) => auth,
+                None => {
                     failed_count += 1;
                     results.push(json!({
                         "key_id": key.id,
@@ -58,6 +142,24 @@ pub(crate) async fn refresh_kiro_provider_quota_locally(
                         "message": "缺少 Kiro 认证配置 (auth_config)",
                     }));
                     continue;
+                }
+            },
+            Err(err) => {
+                failed_count += 1;
+                let mut payload = serde_json::Map::new();
+                payload.insert("key_id".to_string(), json!(key.id));
+                payload.insert("key_name".to_string(), json!(key.name));
+                payload.insert("status".to_string(), json!("error"));
+                payload.insert(
+                    "message".to_string(),
+                    json!(kiro_quota_refresh_failure_message(&err)),
+                );
+                if let Some(status_code) = kiro_quota_refresh_failure_status(&err) {
+                    payload.insert("status_code".to_string(), json!(status_code));
+                }
+                results.push(serde_json::Value::Object(payload));
+                continue;
+            }
         };
 
         let result =
@@ -152,6 +254,10 @@ pub(crate) async fn refresh_kiro_provider_quota_locally(
                         .clone()
                         .filter(|value| !value.trim().is_empty())
                         .unwrap_or_else(|| format!("HTTP {}", result.status_code));
+                    if kiro_quota_error_is_token_invalid(err_msg.as_deref()) {
+                        oauth_invalid_at_unix_secs = Some(now_unix_secs);
+                        oauth_invalid_reason = Some("Kiro Token 无效或已过期".to_string());
+                    } else if kiro_quota_error_is_account_banned(err_msg.as_deref()) {
                         oauth_invalid_at_unix_secs = Some(now_unix_secs);
                         oauth_invalid_reason = Some(format!("账户已封禁: {reason}"));
                         metadata_update = Some(json!({
@@ -164,6 +270,7 @@ pub(crate) async fn refresh_kiro_provider_quota_locally(
                         }));
                         status = "banned".to_string();
                     }
+                }
                 _ => {}
             }
         }
@@ -227,3 +334,32 @@ pub(crate) async fn refresh_kiro_provider_quota_locally(
         "auto_removed": 0,
     })))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::{kiro_quota_error_is_account_banned, kiro_quota_error_is_token_invalid};
+
+    #[test]
+    fn bearer_token_invalid_is_not_classified_as_banned() {
+        let detail = Some("Bearer token invalid");
+
+        assert!(kiro_quota_error_is_token_invalid(detail));
+        assert!(!kiro_quota_error_is_account_banned(detail));
+    }
+
+    #[test]
+    fn bearer_token_invild_typo_is_not_classified_as_banned() {
+        let detail = Some("bearer token invild");
+
+        assert!(kiro_quota_error_is_token_invalid(detail));
+        assert!(!kiro_quota_error_is_account_banned(detail));
+    }
+
+    #[test]
+    fn explicit_kiro_account_suspension_is_classified_as_banned() {
+        let detail = Some("account suspended due to Terms of Service violation");
+
+        assert!(!kiro_quota_error_is_token_invalid(detail));
+        assert!(kiro_quota_error_is_account_banned(detail));
+    }
+}