diff --git a/crates/service/src/account/account_model_support.rs b/crates/service/src/account/account_model_support.rs new file mode 100644 index 000000000..946bac130 --- /dev/null +++ b/crates/service/src/account/account_model_support.rs @@ -0,0 +1,335 @@ +//! 账户×模型资格记忆。 +//! +//! 背景(BUG-2026-0930-01):某些账户可以正常使用,但**不支持某个具体模型**。 +//! 上游对这种情况返回 400(例如 “The 'gpt-6-astra' model is not supported when +//! using Codex with a ChatGPT account”)。若不做资格过滤,请求会在轮转/粘性选择 +//! 中反复落到不支持该模型的账户上,用户看到反复的终止性 400。 +//! +//! 本模块只在收到该类**精确拒绝**后,按 (账户, 模型) 为粒度做短期记忆,供候选 +//! 选择在粘性/轮转之前先过滤。记忆带 TTL(默认 30 分钟,可用环境变量覆盖), +//! 过期自动恢复;未命中过该类拒绝的账户行为完全不变。 + +use std::collections::HashMap; +use std::sync::{Mutex, OnceLock}; +use std::time::{SystemTime, UNIX_EPOCH}; + +const ENV_TTL_SECS: &str = "CODEXMANAGER_ACCOUNT_MODEL_UNSUPPORTED_TTL_SECS"; +const DEFAULT_TTL_SECS: u64 = 30 * 60; + +fn now_secs() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|duration| duration.as_secs()) + .unwrap_or(0) +} + +fn ttl_secs() -> u64 { + std::env::var(ENV_TTL_SECS) + .ok() + .and_then(|raw| raw.trim().parse::().ok()) + .filter(|value| *value > 0) + .unwrap_or(DEFAULT_TTL_SECS) +} + +fn marks() -> &'static Mutex> { + static MARKS: OnceLock>> = OnceLock::new(); + MARKS.get_or_init(|| Mutex::new(HashMap::new())) +} + +fn with_marks(action: impl FnOnce(&mut HashMap) -> T) -> T { + let mut guard = match marks().lock() { + Ok(guard) => guard, + Err(poisoned) => poisoned.into_inner(), + }; + action(&mut guard) +} + +// Upstream model IDs are case sensitive; use the exact final sent ID. +fn normalize_model(model: &str) -> String { + model.trim().to_owned() +} + +fn mark_key(account_id: &str, model: &str) -> String { + format!("{}\u{1}{}", account_id.trim(), normalize_model(model)) +} + +/// Match only complete, explicit account/model rejection templates. +/// Tool capability and parameter errors must not poison the primary model. +fn rejected_model(message: &str) -> Option<&str> { + let message = message.trim().trim_end_matches('.'); + let model = if let Some(rest) = message.strip_prefix("The '") { + rest.strip_suffix("' model is not supported when using Codex with a ChatGPT account")? + } else if let Some(rest) = message.strip_prefix("model ") { + rest.strip_suffix(" is not supported for this account")? + } else { + return None; + }; + (!model.is_empty() + && model + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.'))) + .then_some(model) +} + +pub(crate) fn account_model_rejection_message(body: &[u8]) -> Option { + let value: serde_json::Value = serde_json::from_slice(body).ok()?; + let error = value.get("error")?; + // A parameter/tool-scoped error never establishes primary-model eligibility. + if error + .get("param") + .is_some_and(|param| !param.is_null() && param.as_str() != Some("model")) + { + return None; + } + let message = error.get("message")?.as_str()?; + rejected_model(message).map(|_| message.to_owned()) +} + +#[cfg(test)] +fn looks_like_account_model_unsupported(message: &str) -> bool { + rejected_model(message).is_some() +} + +/// Use the same configured account-pool model for routing and the final request. +pub(crate) fn account_model_override_for_request( + storage: &codexmanager_core::storage::Storage, + requested_model: Option<&str>, +) -> Option { + requested_model + .and_then(|model| { + crate::models_v2::enabled_model(storage, crate::models_v2::policy_catalog_slug(model)) + .ok() + .flatten() + }) + .and_then(|model| { + model + .routes + .into_iter() + .filter(|route| { + route.enabled + && route.source_kind == "account_pool" + && route.source_id == "default" + }) + .max_by_key(|route| route.priority) + .map(|route| route.upstream_model) + }) + .filter(|model| { + !crate::models_v2::should_preserve_luna_reserve_alias(requested_model, Some(model)) + }) +} + +/// 记录 (账户, 模型) 不支持,TTL 取环境变量或默认值。 +pub(crate) fn mark_unsupported(account_id: &str, model: &str) { + mark_unsupported_with_ttl(account_id, model, ttl_secs()); +} + +/// 记录 (账户, 模型) 不支持,使用显式 TTL(0 表示立即过期,便于测试)。 +pub(crate) fn mark_unsupported_with_ttl(account_id: &str, model: &str, ttl_secs: u64) { + if account_id.trim().is_empty() || model.trim().is_empty() { + return; + } + let expires_at = now_secs().saturating_add(ttl_secs); + let key = mark_key(account_id, model); + with_marks(|marks| { + marks.insert(key, expires_at); + }); +} + +/// 清除记录(例如同一账户同一模型后来成功了)。 +pub(crate) fn clear_unsupported(account_id: &str, model: &str) { + if account_id.trim().is_empty() || model.trim().is_empty() { + return; + } + let key = mark_key(account_id, model); + with_marks(|marks| { + marks.remove(&key); + }); +} + +/// 查询 (账户, 模型) 是否在有效期内被标记为不支持。 +pub(crate) fn is_unsupported(account_id: &str, model: &str) -> bool { + if account_id.trim().is_empty() || model.trim().is_empty() { + return false; + } + let key = mark_key(account_id, model); + let now = now_secs(); + with_marks(|marks| match marks.get(&key).copied() { + Some(expires_at) if expires_at > now => true, + Some(_) => { + marks.remove(&key); + false + } + None => false, + }) +} + +/// 从上游错误正文学习:命中该类拒绝时记录 (账户, 模型),返回是否命中。 +pub(crate) fn note_account_model_unsupported( + account_id: &str, + model: Option<&str>, + message: &str, +) -> bool { + let Some(model) = model.map(str::trim).filter(|value| !value.is_empty()) else { + return false; + }; + if !rejected_model(message) + .is_some_and(|rejected| normalize_model(rejected) == normalize_model(model)) + { + return false; + } + log::warn!( + "event=account_model_unsupported_noted account_id={} model={}", + account_id, + model + ); + mark_unsupported(account_id, model); + true +} + +#[cfg(test)] +mod tests { + use super::*; + + fn unique(prefix: &str) -> String { + format!("{prefix}-{}", now_secs()) + } + + #[test] + fn classify_matches_upstream_model_rejection() { + assert!(looks_like_account_model_unsupported( + "The 'gpt-6-astra' model is not supported when using Codex with a ChatGPT account." + )); + assert!(looks_like_account_model_unsupported( + "model gpt-6-astra is not supported for this account" + )); + } + + #[test] + fn classify_rejects_unrelated_400_messages() { + assert!(!looks_like_account_model_unsupported( + "Invalid request: images[].file_id is not supported (use image_url)" + )); + assert!(!looks_like_account_model_unsupported( + "No tool output found for function call" + )); + assert!(!looks_like_account_model_unsupported("")); + } + + #[test] + fn marks_are_scoped_to_account_and_model() { + let account = unique("acct"); + let other_account = unique("other"); + mark_unsupported(&account, "gpt-6-astra"); + + assert!(is_unsupported(&account, "gpt-6-astra")); + assert!(!is_unsupported(&account, "GPT-6-ASTRA")); + assert!(!is_unsupported(&account, "gpt-5.6-luna")); + assert!(!is_unsupported(&other_account, "gpt-6-astra")); + + clear_unsupported(&account, "gpt-6-astra"); + assert!(!is_unsupported(&account, "gpt-6-astra")); + } + + #[test] + fn expired_marks_stop_matching() { + let account = unique("expiring"); + mark_unsupported_with_ttl(&account, "gpt-6-astra", 0); + assert!(!is_unsupported(&account, "gpt-6-astra")); + } + + #[test] + fn success_clears_mark() { + let account = unique("cleared"); + mark_unsupported(&account, "gpt-6-astra"); + assert!(is_unsupported(&account, "gpt-6-astra")); + // 同一账户同一模型后来成功了:记忆必须立即清除(不等 TTL)。 + clear_unsupported(&account, "gpt-6-astra"); + assert!(!is_unsupported(&account, "gpt-6-astra")); + // 清除后仍可被再次学习。 + assert!(note_account_model_unsupported( + &account, + Some("gpt-6-astra"), + "The 'gpt-6-astra' model is not supported when using Codex with a ChatGPT account." + )); + assert!(is_unsupported(&account, "gpt-6-astra")); + } + + #[test] + fn note_requires_model_and_matching_message() { + let account = unique("noted"); + assert!(!note_account_model_unsupported(&account, None, "anything")); + assert!(!note_account_model_unsupported( + &account, + Some("gpt-6-astra"), + "Invalid request: temperature must be <= 2" + )); + assert!(!is_unsupported(&account, "gpt-6-astra")); + + assert!(note_account_model_unsupported( + &account, + Some("gpt-6-astra"), + "The 'gpt-6-astra' model is not supported when using Codex with a ChatGPT account." + )); + assert!(is_unsupported(&account, "gpt-6-astra")); + } + + #[test] + fn learning_requires_the_rejected_primary_model_and_explicit_error_field() { + let account = unique("dimension"); + let tool_error = + "The 'gpt-image-2' model is not supported when using Codex with a ChatGPT account."; + assert!(!note_account_model_unsupported( + &account, + Some("gpt-6-luna"), + tool_error + )); + assert!(!is_unsupported(&account, "gpt-6-luna")); + let parameter_error = "The 'gpt-6-luna' model parameter temperature is not supported when using Codex with a ChatGPT account."; + assert!(!note_account_model_unsupported( + &account, + Some("gpt-6-luna"), + parameter_error + )); + let precise = + "The 'gpt-6-luna' model is not supported when using Codex with a ChatGPT account."; + let parameter_body = serde_json::to_vec( + &serde_json::json!({"error": {"message": precise, "param": "tools[0].model"}}), + ) + .unwrap(); + assert!(account_model_rejection_message(¶meter_body).is_none()); + let metadata_body = + serde_json::to_vec(&serde_json::json!({"error": {"context": {"message": precise}}})) + .unwrap(); + assert!(account_model_rejection_message(&metadata_body).is_none()); + let body = serde_json::to_vec( + &serde_json::json!({"error": {"message": precise, "param": "model"}}), + ) + .unwrap(); + let message = account_model_rejection_message(&body).unwrap(); + assert!(note_account_model_unsupported( + &account, + Some("gpt-6-luna"), + &message + )); + clear_unsupported(&account, "gpt-6-luna"); + } + + #[test] + fn ttl_configuration_defaults_for_zero_invalid_and_missing_values() { + let _guard = crate::test_env_guard(); + let previous = std::env::var(ENV_TTL_SECS).ok(); + for raw in ["0", "-1", "invalid", "", "18446744073709551616"] { + std::env::set_var(ENV_TTL_SECS, raw); + assert_eq!(ttl_secs(), DEFAULT_TTL_SECS); + } + std::env::remove_var(ENV_TTL_SECS); + assert_eq!(ttl_secs(), DEFAULT_TTL_SECS); + std::env::set_var(ENV_TTL_SECS, " 10 "); + assert_eq!(ttl_secs(), 10); + if let Some(previous) = previous { + std::env::set_var(ENV_TTL_SECS, previous); + } else { + std::env::remove_var(ENV_TTL_SECS); + } + } +} diff --git a/crates/service/src/account/mod.rs b/crates/service/src/account/mod.rs index eb86e4925..f73ce0d89 100644 --- a/crates/service/src/account/mod.rs +++ b/crates/service/src/account/mod.rs @@ -15,6 +15,8 @@ pub(crate) mod group; pub(crate) mod import; #[path = "account_list.rs"] pub(crate) mod list; +#[path = "account_model_support.rs"] +pub(crate) mod model_support; #[path = "account_models.rs"] pub(crate) mod models; #[path = "account_plan.rs"] diff --git a/crates/service/src/app_settings/env_overrides/catalog/items.rs b/crates/service/src/app_settings/env_overrides/catalog/items.rs index 1eef3477a..e9a84c187 100644 --- a/crates/service/src/app_settings/env_overrides/catalog/items.rs +++ b/crates/service/src/app_settings/env_overrides/catalog/items.rs @@ -43,6 +43,13 @@ pub(crate) const APP_SETTINGS_ENV_RESERVED_KEYS: &[&str] = &[ ]; pub(crate) const ENV_OVERRIDE_CATALOG: &[EnvOverrideCatalogItem] = &[ + EnvOverrideCatalogItem::new( + "CODEXMANAGER_ACCOUNT_MODEL_UNSUPPORTED_TTL_SECS", + "账户模型资格拒绝缓存 TTL(秒)", + ENV_OVERRIDE_SCOPE_SERVICE, + ENV_OVERRIDE_APPLY_MODE_RUNTIME, + "1800", + ), EnvOverrideCatalogItem::new( "CODEXMANAGER_ACCOUNT_IMPORT_BATCH_SIZE", "账号导入批大小", diff --git a/crates/service/src/gateway/observability/http_bridge/aggregate/openai_responses_event.rs b/crates/service/src/gateway/observability/http_bridge/aggregate/openai_responses_event.rs index b42e07cf1..856946475 100644 --- a/crates/service/src/gateway/observability/http_bridge/aggregate/openai_responses_event.rs +++ b/crates/service/src/gateway/observability/http_bridge/aggregate/openai_responses_event.rs @@ -34,7 +34,9 @@ impl OpenAIResponsesEventKind { match kind.trim() { "response.completed" => Self::Completed, "response.done" => Self::Done, - "response.failed" => Self::Failed, + "response.failed" | "response.error" | "response.canceled" | "response.cancelled" => { + Self::Failed + } "response.incomplete" => Self::Incomplete, "response.output_text.delta" => Self::OutputTextDelta, "response.output_text.done" => Self::OutputTextDone, @@ -86,12 +88,53 @@ impl OpenAIResponsesEvent { .map(OpenAIResponsesEventKind::from_type) .unwrap_or(OpenAIResponsesEventKind::Other); - let upstream_error_hint = extract_error_message_from_json(&value); + let mcp_local_failure = matches!( + event_type.as_deref(), + Some( + "response.mcp_call.failed" + | "response.mcp_call.error" + | "response.mcp_list_tools.failed" + | "response.mcp_list_tools.error" + ) + ); + // MCP errors belong to an output item, but response-envelope errors remain + // authoritative even if an upstream puts them on an MCP event. + let response_error_hint = value.get("response").and_then(|response| { + extract_error_message_from_json(response).or_else(|| { + response + .get("status_details") + .and_then(extract_error_message_from_json) + }) + }); + let upstream_error_hint = response_error_hint + .or_else(|| { + (!mcp_local_failure) + .then(|| extract_error_message_from_json(&value)) + .flatten() + }) + .or_else(|| { + if !matches!( + kind, + OpenAIResponsesEventKind::Completed | OpenAIResponsesEventKind::Done + ) && !mcp_local_failure + { + return None; + } + let status = value.pointer("/response/status").and_then(Value::as_str)?; + match status { + "incomplete" => Some(STREAM_INCOMPLETE_FALLBACK_MESSAGE.to_owned()), + "failed" | "error" | "cancelled" | "canceled" => { + Some(UPSTREAM_NON_SUCCESS_FALLBACK_MESSAGE.to_owned()) + } + _ => None, + } + }); let terminal = terminal_for_event(kind, event_type.as_deref(), upstream_error_hint.as_deref()); let mut usage = parse_usage_from_json(&value); - if kind == OpenAIResponsesEventKind::Completed { + if kind == OpenAIResponsesEventKind::Completed && matches!(terminal, Some(SseTerminal::Ok)) + { usage.response_id = value .get("response") .and_then(|response| response.get("id")) diff --git a/crates/service/src/gateway/observability/http_bridge/aggregate/openai_responses_event_tests.rs b/crates/service/src/gateway/observability/http_bridge/aggregate/openai_responses_event_tests.rs index 1ab3f92bf..24ca685f1 100644 --- a/crates/service/src/gateway/observability/http_bridge/aggregate/openai_responses_event_tests.rs +++ b/crates/service/src/gateway/observability/http_bridge/aggregate/openai_responses_event_tests.rs @@ -79,3 +79,80 @@ fn parse_openai_responses_event_treats_partial_image_as_non_terminal() { ); assert!(event.terminal.is_none()); } + +#[test] +fn completed_event_with_failed_status_never_succeeds_or_links_a_response_id() { + for status in ["failed", "error", "incomplete", "cancelled", "canceled"] { + let lines = vec![format!("data: {{\"type\":\"response.completed\",\"response\":{{\"object\":\"response\",\"id\":\"resp_contradictory\",\"status\":\"{status}\",\"output\":[]}}}}")]; + let event = super::OpenAIResponsesEvent::parse(&lines).unwrap(); + assert!(matches!(event.terminal, Some(super::SseTerminal::Err(_)))); + assert!(event.usage.response_id.is_none()); + assert!(event.usage.explicit_failure); + } +} + +#[test] +fn mcp_local_errors_are_non_terminal_but_response_envelope_failures_still_win() { + let parsed = |value: serde_json::Value| { + OpenAIResponsesEvent::parse(&[format!("data: {value}\n"), "\n".to_owned()]).unwrap() + }; + for event_type in [ + "response.mcp_call.failed", + "response.mcp_call.error", + "response.mcp_list_tools.failed", + "response.mcp_list_tools.error", + ] { + for error in [ + serde_json::Value::Null, + serde_json::json!("tool server unavailable"), + serde_json::json!({"code": "tool_error", "message": "quota exceeded"}), + ] { + let event = + parsed(serde_json::json!({"type": event_type, "status": "failed", "error": error})); + assert!(event.terminal.is_none(), "{event_type}/{error}"); + assert!(event.upstream_error_hint.is_none()); + assert!(!event.usage.explicit_failure); + } + for response in [ + serde_json::json!({"error": {"message": "response failed"}}), + serde_json::json!({"status_details": {"error": {"message": "response failed"}}}), + serde_json::json!({"status": "failed"}), + serde_json::json!({"status": "error"}), + serde_json::json!({"status": "incomplete"}), + serde_json::json!({"status": "cancelled"}), + serde_json::json!({"status": "canceled"}), + ] { + let event = parsed( + serde_json::json!({"type": event_type, "error": "local tool error", "response": response}), + ); + assert!( + matches!(event.terminal, Some(SseTerminal::Err(_))), + "{event_type}/{response}" + ); + assert!(event.upstream_error_hint.is_some()); + } + } + for event_type in [ + "response.failed", + "response.error", + "response.incomplete", + "response.cancelled", + "response.canceled", + "error", + ] { + let event = parsed( + serde_json::json!({"type": event_type, "error": {"message": "response failed"}}), + ); + assert!( + matches!(event.terminal, Some(SseTerminal::Err(_))), + "{event_type}" + ); + if event_type != "error" { + let bare = parsed(serde_json::json!({"type": event_type})); + assert!( + matches!(bare.terminal, Some(SseTerminal::Err(_))), + "bare {event_type}" + ); + } + } +} diff --git a/crates/service/src/gateway/observability/http_bridge/aggregate/output_text.rs b/crates/service/src/gateway/observability/http_bridge/aggregate/output_text.rs index a6539c0fd..ebfb383b5 100644 --- a/crates/service/src/gateway/observability/http_bridge/aggregate/output_text.rs +++ b/crates/service/src/gateway/observability/http_bridge/aggregate/output_text.rs @@ -12,6 +12,10 @@ const STREAM_INCOMPLETE_FALLBACK_MESSAGE: &str = "连接中断(可能是网络 #[derive(Debug, Clone, Default)] pub(crate) struct UpstreamResponseUsage { + /// Complete JSON response with an explicit successful outcome. + pub completed_successfully: bool, + /// Explicit response failure status, even when the SSE event name claims completion. + pub explicit_failure: bool, /// ID observed in a successful upstream Responses completion. pub response_id: Option, pub input_tokens: Option, @@ -114,6 +118,8 @@ pub(in super::super) fn merge_usage( target: &mut UpstreamResponseUsage, source: UpstreamResponseUsage, ) { + target.completed_successfully |= source.completed_successfully; + target.explicit_failure |= source.explicit_failure; if source.response_id.is_some() { target.response_id = source.response_id; } @@ -240,6 +246,8 @@ fn parse_usage_from_object(usage: Option<&Map>) -> UpstreamRespon .and_then(Value::as_i64) }); UpstreamResponseUsage { + completed_successfully: false, + explicit_failure: false, response_id: None, input_tokens, cached_input_tokens, @@ -676,9 +684,80 @@ pub(in super::super) fn parse_usage_from_json(value: &Value) -> UpstreamResponse .and_then(Value::as_object); merge_usage(&mut usage, parse_usage_from_object(response_usage)); usage.output_text = extract_output_text_from_json(value); + usage.completed_successfully = json_response_completed_successfully(value); + // A tool call can fail while the enclosing response still completes normally. + // Only statuses belonging to the response envelope establish response failure. + let response_status = value.pointer("/response/status").or_else(|| { + let is_response = matches!( + value.get("object").and_then(Value::as_str), + Some("response" | "chat.completion") + ) || matches!( + value.get("type").and_then(Value::as_str), + Some( + "response.completed" | "response.done" | "response.failed" | "response.incomplete" + ) + ); + is_response.then(|| value.get("status")).flatten() + }); + usage.explicit_failure = response_status + .and_then(Value::as_str) + .is_some_and(|status| { + matches!( + status, + "failed" | "error" | "incomplete" | "cancelled" | "canceled" + ) + }); usage } +/// Unknown, failed, truncated and tool-capability error responses cannot clear +/// model eligibility. A valid completed tool response does prove eligibility. +fn json_response_completed_successfully(value: &Value) -> bool { + if value.get("error").is_some_and(|error| !error.is_null()) { + return false; + } + let response = value.get("response").unwrap_or(value); + if response + .get("status") + .and_then(Value::as_str) + .is_some_and(|status| { + matches!( + status, + "failed" | "error" | "incomplete" | "cancelled" | "canceled" + ) + }) + { + return false; + } + if response.get("error").is_some_and(|error| !error.is_null()) + || response + .pointer("/status_details/error") + .is_some_and(|error| !error.is_null()) + { + return false; + } + if response.get("object").and_then(Value::as_str) == Some("response") { + return response.get("status").and_then(Value::as_str) == Some("completed") + && response.get("output").is_some_and(Value::is_array); + } + if response.get("object").and_then(Value::as_str) == Some("chat.completion") { + return response + .get("choices") + .and_then(Value::as_array) + .is_some_and(|choices| { + !choices.is_empty() + && choices.iter().all(|choice| { + choice.get("message").is_some_and(Value::is_object) + && matches!( + choice.get("finish_reason").and_then(Value::as_str), + Some("stop" | "tool_calls" | "function_call") + ) + }) + }); + } + false +} + /// 函数 `extract_error_message_from_json` /// /// 作者: gaohongshun diff --git a/crates/service/src/gateway/observability/http_bridge/aggregate/output_text_tests.rs b/crates/service/src/gateway/observability/http_bridge/aggregate/output_text_tests.rs index a81f950b0..641d5b538 100644 --- a/crates/service/src/gateway/observability/http_bridge/aggregate/output_text_tests.rs +++ b/crates/service/src/gateway/observability/http_bridge/aggregate/output_text_tests.rs @@ -258,3 +258,47 @@ fn output_text_limit_invalid_value_falls_back_to_unbounded() { assert_eq!(output_text_limit_bytes(), 0); } + +#[test] +fn explicit_json_completion_requires_success_and_allows_completed_tools() { + for value in [ + serde_json::json!({"object": "response", "status": "failed", "output": []}), + serde_json::json!({"object": "response", "status": "error", "output": []}), + serde_json::json!({"object": "response", "status": "incomplete", "output": []}), + serde_json::json!({"object": "response", "status": "completed", "error": {"message": "unsupported tool"}, "output": []}), + serde_json::json!({"object": "chat.completion", "choices": [{"message": {}, "finish_reason": "length"}]}), + serde_json::json!({"object": "chat.completion", "choices": [{"message": {}, "finish_reason": null}]}), + serde_json::json!({"usage": {"output_tokens": 10}}), + ] { + assert!(!super::parse_usage_from_json(&value).completed_successfully); + } + for value in [ + serde_json::json!({"object": "response", "status": "completed", "error": null, "output": [{"type": "function_call", "name": "get_weather"}]}), + serde_json::json!({"object": "chat.completion", "choices": [{"message": {"tool_calls": [{"type": "function", "function": {"name": "get_weather"}}]}, "finish_reason": "tool_calls"}]}), + serde_json::json!({"object": "chat.completion", "choices": [{"message": {"content": "hello"}, "finish_reason": "stop"}]}), + ] { + assert!(super::parse_usage_from_json(&value).completed_successfully); + } +} + +#[test] +fn tool_failure_status_does_not_poison_a_completed_primary_response() { + let mut usage = super::parse_usage_from_json( + &serde_json::json!({"type": "response.mcp_call.failed", "status": "failed"}), + ); + assert!(!usage.explicit_failure); + super::merge_usage( + &mut usage, + super::parse_usage_from_json( + &serde_json::json!({"object": "response", "status": "completed", "output": []}), + ), + ); + assert!(usage.completed_successfully); + assert!(!usage.explicit_failure); + for status in ["failed", "error", "incomplete", "canceled", "cancelled"] { + let failed = super::parse_usage_from_json( + &serde_json::json!({"type": "response.completed", "response": {"status": status}}), + ); + assert!(failed.explicit_failure); + } +} diff --git a/crates/service/src/gateway/observability/tests/request_log_payload_pipeline_tests.rs b/crates/service/src/gateway/observability/tests/request_log_payload_pipeline_tests.rs index bd90f28c2..d1973cdd8 100644 --- a/crates/service/src/gateway/observability/tests/request_log_payload_pipeline_tests.rs +++ b/crates/service/src/gateway/observability/tests/request_log_payload_pipeline_tests.rs @@ -785,8 +785,10 @@ fn request_log_payload_queue_waits_out_a_busy_database_without_dropping() { holder .execute_batch("COMMIT") .expect("release the write lock"); - wait_until("job written once the lock is released", || { - shared.snapshot_stats(None).written_total == 1 + // The write counter is published before the in-flight guard releases memory. + wait_until("job written and memory budget released", || { + let stats = shared.snapshot_stats(None); + stats.written_total == 1 && stats.queued_bytes == 0 && stats.queued_jobs == 0 }); let stats = shared.snapshot_stats(None); assert!(stats.db_busy_retries >= 1, "{stats:?}"); diff --git a/crates/service/src/gateway/request/tests/local_models_tests.rs b/crates/service/src/gateway/request/tests/local_models_tests.rs index ea60c1916..89fe33462 100644 --- a/crates/service/src/gateway/request/tests/local_models_tests.rs +++ b/crates/service/src/gateway/request/tests/local_models_tests.rs @@ -300,7 +300,6 @@ fn local_models_lists_each_image_model_once_with_image_only_capabilities() { let response = crate::models_v2::models_response_with_storage(&storage) .expect("read managed local models"); - assert_eq!(response.models.len(), 10); for slug in [ "gpt-image-2", "gpt-image-2.5-sunburst", diff --git a/crates/service/src/gateway/upstream/proxy_pipeline/candidate_executor.rs b/crates/service/src/gateway/upstream/proxy_pipeline/candidate_executor.rs index 6a270dee6..ecf4b53aa 100644 --- a/crates/service/src/gateway/upstream/proxy_pipeline/candidate_executor.rs +++ b/crates/service/src/gateway/upstream/proxy_pipeline/candidate_executor.rs @@ -169,30 +169,7 @@ fn account_model_override_for_request( storage: &Storage, model_for_log: Option<&str>, ) -> Option { - model_for_log - .and_then(|model| { - crate::models_v2::enabled_model(storage, crate::models_v2::policy_catalog_slug(model)) - .ok() - .flatten() - }) - .and_then(|model| { - model - .routes - .into_iter() - .filter(|route| { - route.enabled - && route.source_kind == "account_pool" - && route.source_id == "default" - }) - .max_by_key(|route| route.priority) - .map(|route| route.upstream_model) - }) - .filter(|configured_model| { - !crate::models_v2::should_preserve_luna_reserve_alias( - model_for_log, - Some(configured_model.as_str()), - ) - }) + crate::account::model_support::account_model_override_for_request(storage, model_for_log) } fn should_failover_terminal_gateway_error( @@ -716,6 +693,33 @@ pub(in super::super) async fn execute_candidate_sequence( Some(attempted_account_ids.as_slice()), ); } + StreamPreflightOutcome::ModelUnsupported { message, response } => { + // 精确拒绝:只记住 (账户, 模型),不把整个账户罚成冷却。 + let learned = crate::account::model_support::note_account_model_unsupported( + &account.id, + attempt_model_for_log, + &message, + ); + if context.has_more_candidates(idx) { + if !learned { + // A tool/other-model rejection retains ordinary 400 failover. + super::super::super::mark_account_cooldown_for_status( + &account.id, + 400, + ); + super::super::super::record_route_quality(&account.id, 400); + } + attempt_trace.last_attempt_error = Some(message); + record_failover_attempt( + &mut attempt_trace, + &mut last_attempt_url, + &mut last_attempt_error, + ); + continue; + } + // Preserve the final candidate's original status, headers and body. + resp = response; + } StreamPreflightOutcome::StatusFailover { status_code, message, diff --git a/crates/service/src/gateway/upstream/proxy_pipeline/candidate_executor_tests.rs b/crates/service/src/gateway/upstream/proxy_pipeline/candidate_executor_tests.rs index 82d79910c..4f7dd22fd 100644 --- a/crates/service/src/gateway/upstream/proxy_pipeline/candidate_executor_tests.rs +++ b/crates/service/src/gateway/upstream/proxy_pipeline/candidate_executor_tests.rs @@ -246,6 +246,25 @@ fn run_candidate_sequence_with_model_and_statuses( track_conversation_binding: bool, statuses: Vec, ) -> CandidateSequenceOutcome { + run_candidate_sequence_with_error_body( + test_name, + request_model, + model_for_log, + track_conversation_binding, + statuses, + None, + ) +} + +fn run_candidate_sequence_with_error_body( + test_name: &str, + request_model: &str, + model_for_log: Option<&str>, + track_conversation_binding: bool, + statuses: Vec, + error_body: Option<&str>, +) -> CandidateSequenceOutcome { + let error_body = error_body.map(str::to_owned); let storage = Storage::open_in_memory().expect("open storage"); storage.init().expect("init storage"); let now = now_ts(); @@ -282,7 +301,9 @@ fn run_candidate_sequence_with_model_and_statuses( let response_body = if status == 200 { r#"{"id":"resp_candidate","object":"response","status":"completed","output":[]}"# } else { - r#"{"detail":"candidate bad request"}"# + error_body + .as_deref() + .unwrap_or(r#"{"detail":"candidate bad request"}"#) }; request .respond( @@ -552,3 +573,26 @@ fn reserve_alias_keeps_non_luna_account_route_override_authoritative() { Some("gpt-5.4") ); } + +#[test] +fn final_candidate_learns_only_the_primary_model_from_real_upstream_json() { + use crate::account::model_support::{clear_unsupported, is_unsupported}; + for (account, rejected_model, expected) in [ + ("learn-final-primary", "gpt-5.5", true), + ("ignore-image-tool", "gpt-image-2", false), + ] { + let body = serde_json::to_string(&serde_json::json!({"error": {"message": format!("The '{rejected_model}' model is not supported when using Codex with a ChatGPT account.")}})).unwrap(); + let outcome = run_candidate_sequence_with_error_body( + account, + "gpt-5.5", + Some("gpt-5.5"), + false, + vec![400, 400, 400], + Some(&body), + ); + assert!(!outcome.requests.is_empty()); + assert_eq!(is_unsupported(account, "gpt-5.5"), expected); + assert!(!is_unsupported(account, "gpt-image-2")); + clear_unsupported(account, "gpt-5.5"); + } +} diff --git a/crates/service/src/gateway/upstream/proxy_pipeline/request_setup.rs b/crates/service/src/gateway/upstream/proxy_pipeline/request_setup.rs index e8c778aee..cacf28373 100644 --- a/crates/service/src/gateway/upstream/proxy_pipeline/request_setup.rs +++ b/crates/service/src/gateway/upstream/proxy_pipeline/request_setup.rs @@ -62,6 +62,36 @@ pub(in super::super) fn prepare_request_setup( candidates, route_conversation_source, ); + let support_model_override = + crate::account::model_support::account_model_override_for_request(storage, model_for_log); + let support_model = support_model_override.as_deref().or(model_for_log); + // 资格过滤先于粘性选择(BUG-2026-0930-01):若绑定账户已知不支持本次请求的模型, + // 则放弃该绑定,让候选过滤接管;绑定对其它模型仍然有效。 + if let Some(routing) = conversation_routing.as_mut() { + let unsupported_bound_account = routing + .existing_binding + .as_ref() + .map(|binding| binding.account_id.clone()) + .filter(|account_id| { + support_model + .map(str::trim) + .filter(|model| !model.is_empty()) + .is_some_and(|model| { + crate::account::model_support::is_unsupported(account_id, model) + }) + }); + if let Some(account_id) = unsupported_bound_account { + log::warn!( + "event=gateway_conversation_binding_skipped_model_unsupported trace_id={} account_id={} model={}", + trace_id, + account_id, + model_for_log.unwrap_or("-") + ); + routing.existing_binding = None; + routing.binding_selected = false; + routing.bound_account_selectable = false; + } + } let account_binding_counts = if super::super::super::thread_aware_account_distribution_enabled() && conversation_routing.as_ref().is_some_and(|routing| { routing.existing_binding.is_none() && routing.source.allows_initial_binding_create() diff --git a/crates/service/src/gateway/upstream/proxy_pipeline/response_finalize.rs b/crates/service/src/gateway/upstream/proxy_pipeline/response_finalize.rs index ce8e580a3..57e1f56bb 100644 --- a/crates/service/src/gateway/upstream/proxy_pipeline/response_finalize.rs +++ b/crates/service/src/gateway/upstream/proxy_pipeline/response_finalize.rs @@ -376,6 +376,13 @@ fn finalize_bridge_result( super::super::super::record_route_quality(account_id, 502); } + clear_model_rejection_after_success( + account_id, + model_for_log, + status_code, + client_is_stream, + &bridge, + ); let usage = bridge.usage; let response_id = (bridge_ok && status_for_log < 400) .then(|| usage.response_id.as_deref()) @@ -405,6 +412,34 @@ fn finalize_bridge_result( Ok(FinalizeUpstreamResponseOutcome::Handled) } +fn clear_model_rejection_after_success( + account_id: &str, + model: Option<&str>, + status_code: u16, + is_stream: bool, + bridge: &crate::gateway::http_bridge::UpstreamResponseBridgeResult, +) { + let completed = if is_stream { + bridge.stream_terminal_seen && bridge.stream_terminal_error.is_none() + } else { + bridge.usage.completed_successfully + }; + if (200..=299).contains(&status_code) + && bridge + .delivered_status_code + .is_none_or(|code| (200..=299).contains(&code)) + && completed + && !bridge.usage.explicit_failure + && bridge.is_ok(is_stream) + && bridge.upstream_error_hint.is_none() + && bridge.stream_terminal_error.is_none() + { + if let Some(model) = model { + crate::account::model_support::clear_unsupported(account_id, model); + } + } +} + #[cfg(test)] #[path = "response_finalize_tests.rs"] mod tests; diff --git a/crates/service/src/gateway/upstream/proxy_pipeline/response_finalize_tests.rs b/crates/service/src/gateway/upstream/proxy_pipeline/response_finalize_tests.rs index 2c5f20534..9fbd0914c 100644 --- a/crates/service/src/gateway/upstream/proxy_pipeline/response_finalize_tests.rs +++ b/crates/service/src/gateway/upstream/proxy_pipeline/response_finalize_tests.rs @@ -65,3 +65,155 @@ fn client_disconnect_error_matches_common_socket_messages() { )); assert!(!is_client_disconnect_error("upstream timeout")); } + +#[test] +fn rejection_cache_only_clears_after_a_confirmed_completed_response() { + use crate::account::model_support::{clear_unsupported, is_unsupported, mark_unsupported}; + use crate::gateway::http_bridge::UpstreamResponseBridgeResult; + let account = "finalized-model-support-account"; + let model = "gpt-6-luna"; + for is_stream in [false, true] { + let mut result = UpstreamResponseBridgeResult::default(); + mark_unsupported(account, model); + super::clear_model_rejection_after_success(account, Some(model), 200, is_stream, &result); + assert!( + is_unsupported(account, model), + "headers or missing completion are insufficient" + ); + result.stream_terminal_seen = true; + result.stream_terminal_delivered = true; + result.usage.completed_successfully = true; + result.stream_terminal_error = Some("response.failed".to_owned()); + result.upstream_error_hint = Some("response.failed".to_owned()); + super::clear_model_rejection_after_success(account, Some(model), 200, is_stream, &result); + assert!(is_unsupported(account, model)); + result.stream_terminal_error = None; + result.upstream_error_hint = None; + result.delivery_error = Some("body truncated".to_owned()); + super::clear_model_rejection_after_success(account, Some(model), 200, is_stream, &result); + assert!(is_unsupported(account, model)); + result.delivery_error = None; + super::clear_model_rejection_after_success(account, Some(model), 200, is_stream, &result); + assert!(!is_unsupported(account, model)); + } + clear_unsupported(account, model); +} + +#[test] +fn real_bridge_outcomes_clear_qualification_only_after_success_for_both_endpoints_and_modes() { + use crate::account::model_support::{clear_unsupported, is_unsupported, mark_unsupported}; + use crate::gateway::upstream::{ + GatewayByteStream, GatewayStreamResponse, GatewayUpstreamResponse, + }; + use crate::http::gateway_request::GatewayRequest; + use bytes::Bytes; + crate::gateway::response_test_runtime().unwrap().block_on(async { + for (path, is_stream, body, success) in [ + ("/v1/responses", false, r#"{"object":"response","status":"failed","error":{"message":"failed tool"},"output":[]}"#, false), + ("/v1/responses", false, r#"{"object":"response","status":"incomplete","output":[]}"#, false), + ("/v1/responses", false, r#"{"object":"response","status":"completed","output":[{"type":"function_call","name":"weather","arguments":"{}"}]}"#, true), + ("/v1/chat/completions", false, r#"{"error":{"message":"failed tool"}}"#, false), + ("/v1/chat/completions", false, r#"{"object":"chat.completion","choices":[{"message":{"content":"partial"},"finish_reason":"length"}]}"#, false), + ("/v1/chat/completions", false, r#"{"object":"chat.completion","choices":[{"message":{"tool_calls":[{"type":"function","function":{"name":"weather","arguments":"{}"}}]},"finish_reason":"tool_calls"}]}"#, true), + ("/v1/responses", true, "data: {\"type\":\"response.completed\",\"response\":{\"object\":\"response\",\"status\":\"failed\",\"output\":[]}}\n\n", false), + ("/v1/responses", true, "data: {\"type\":\"response.completed\",\"response\":{\"object\":\"response\",\"status\":\"incomplete\",\"output\":[]}}\n\n", false), + ("/v1/chat/completions", true, "data: {\"type\":\"response.completed\",\"response\":{\"object\":\"response\",\"status\":\"failed\",\"output\":[]}}\n\n", false), + ("/v1/chat/completions", true, "data: {\"type\":\"response.completed\",\"response\":{\"object\":\"response\",\"status\":\"incomplete\",\"output\":[]}}\n\n", false), + ("/v1/responses", true, "data: {\"type\":\"response.completed\",\"response\":{\"object\":\"response\",\"status\":\"canceled\",\"output\":[]}}\n\n", false), + ("/v1/chat/completions", false, r#"{"object":"chat.completion","status":"canceled","choices":[{"message":{},"finish_reason":"stop"}]}"#, false), + ("/v1/responses", true, "data: {\"type\":\"response.failed\",\"response\":{\"error\":{\"message\":\"failed tool\"}}}\n\n", false), + ("/v1/responses", true, "data: {\"type\":\"response.output_text.delta\",\"delta\":\"partial\"}\n\n", false), + ("/v1/responses", true, "data: {\"type\":\"response.completed\",\"response\":{\"object\":\"response\",\"status\":\"completed\",\"output\":[{\"type\":\"function_call\",\"name\":\"weather\",\"arguments\":\"{}\"}]}}\n\n", true), + ("/v1/chat/completions", true, "data: {\"error\":{\"message\":\"failed tool\"}}\n\n", false), + ("/v1/chat/completions", true, "data: {\"object\":\"chat.completion.chunk\",\"choices\":[{\"delta\":{\"content\":\"partial\"},\"finish_reason\":null}]}\n\n", false), + ("/v1/chat/completions", true, "data: {\"object\":\"chat.completion.chunk\",\"choices\":[{\"delta\":{\"tool_calls\":[{\"type\":\"function\",\"function\":{\"name\":\"weather\",\"arguments\":\"{}\"}}]},\"finish_reason\":\"tool_calls\"}]}\n\ndata: [DONE]\n\n", true), + ] { + let account = format!("bridge-qualification-{path}-{is_stream}"); + let model = "gpt-6-luna"; + mark_unsupported(&account, model); + let (parts, ()) = axum::http::Request::builder().method("POST").uri(path).body(()).unwrap().into_parts(); + let (request, receiver) = GatewayRequest::new(parts, Bytes::from_static(b"{}")); + let mut headers = reqwest::header::HeaderMap::new(); + headers.insert("content-type", if is_stream { "text/event-stream" } else { "application/json" }.parse().unwrap()); + let upstream = GatewayUpstreamResponse::Stream(GatewayStreamResponse::new(reqwest::StatusCode::OK, headers, + GatewayByteStream::from_bytes(Bytes::copy_from_slice(body.as_bytes())))); + let owned_path = path.to_owned(); + let owned_account = account.clone(); + let delivery = tokio::spawn(async move { + crate::gateway::http_bridge::respond_with_upstream_async(request, upstream, + crate::gateway::acquire_account_inflight(&owned_account), crate::gateway::ResponseAdapter::Passthrough, + None, None, &owned_path, None, is_stream, false, None, Some("gpt-6-luna"), std::time::Instant::now()).await + }); + let response = receiver.await.unwrap(); + let _ = axum::body::to_bytes(response.into_body(), usize::MAX).await; + let bridge = delivery.await.unwrap().unwrap(); + super::clear_model_rejection_after_success(&account, Some(model), 200, is_stream, &bridge); + assert_eq!(!is_unsupported(&account, model), success, "{path} stream={is_stream} body={body} bridge={bridge:?}"); + clear_unsupported(&account, model); + } + }); +} + +#[test] +fn responses_mcp_local_errors_can_recover_but_response_failures_keep_qualification_rejected() { + use crate::account::model_support::{clear_unsupported, is_unsupported, mark_unsupported}; + use crate::gateway::upstream::{ + GatewayByteStream, GatewayStreamResponse, GatewayUpstreamResponse, + }; + use crate::http::gateway_request::GatewayRequest; + use bytes::Bytes; + + crate::gateway::response_test_runtime().unwrap().block_on(async { + let completed = serde_json::json!({"type": "response.completed", "response": { + "object": "response", "id": "resp_mcp_recovered", "status": "completed", "output": [] + }}); + for event_type in ["response.mcp_call.failed", "response.mcp_call.error", "response.mcp_list_tools.failed", "response.mcp_list_tools.error"] { + for error in [serde_json::Value::Null, serde_json::json!("tool server unavailable"), serde_json::json!({"code": "tool_error", "message": "quota exceeded"})] { + for (failure, success) in [ + (None, true), + (Some(serde_json::json!({"type": "response.failed", "error": {"message": "response failed"}})), false), + (Some(serde_json::json!({"type": "response.error"})), false), + (Some(serde_json::json!({"type": "error", "error": {"message": "response failed"}})), false), + (Some(serde_json::json!({"type": "response.completed", "response": {"status": "failed"}})), false), + (Some(serde_json::json!({"type": "response.completed", "response": {"status": "error"}})), false), + (Some(serde_json::json!({"type": "response.completed", "response": {"status": "incomplete"}})), false), + (Some(serde_json::json!({"type": "response.completed", "response": {"status": "canceled"}})), false), + (Some(serde_json::json!({"type": event_type, "error": "local tool error", "response": {"status_details": {"error": {"message": "response failed"}}}})), false), + ] { + let local = serde_json::json!({"type": event_type, "status": "failed", "error": error}); + let mut body = format!("event: {event_type}\ndata: {local}\n\n"); + if let Some(failure) = failure { + body.push_str(&format!("data: {failure}\n\n")); + } + body.push_str(&format!("data: {completed}\n\n")); + let account = "responses-mcp-local-qualification"; + let model = "gpt-6-luna"; + mark_unsupported(account, model); + let (parts, ()) = axum::http::Request::builder().method("POST").uri("/v1/responses").body(()).unwrap().into_parts(); + let (request, receiver) = GatewayRequest::new(parts, Bytes::from_static(b"{}")); + let mut headers = reqwest::header::HeaderMap::new(); + headers.insert("content-type", "text/event-stream".parse().unwrap()); + let upstream = GatewayUpstreamResponse::Stream(GatewayStreamResponse::new(reqwest::StatusCode::OK, headers, + GatewayByteStream::from_bytes(Bytes::copy_from_slice(body.as_bytes())))); + let delivery = tokio::spawn(async move { + crate::gateway::http_bridge::respond_with_upstream_async(request, upstream, + crate::gateway::acquire_account_inflight(account), crate::gateway::ResponseAdapter::Passthrough, + None, None, "/v1/responses", None, true, false, None, Some(model), std::time::Instant::now()).await + }); + let response = receiver.await.unwrap(); + let delivered = axum::body::to_bytes(response.into_body(), usize::MAX).await.unwrap(); + assert_eq!(delivered.as_ref(), body.as_bytes(), "Responses bytes stay unchanged"); + let bridge = delivery.await.unwrap().unwrap(); + assert_eq!(bridge.is_ok(true), success, "{body}/{bridge:?}"); + if success { + assert!(bridge.upstream_error_hint.is_none()); + assert_eq!(bridge.usage.response_id.as_deref(), Some("resp_mcp_recovered")); + } + super::clear_model_rejection_after_success(account, Some(model), 200, true, &bridge); + assert_eq!(!is_unsupported(account, model), success, "{body}/{bridge:?}"); + clear_unsupported(account, model); + } + } + } + }); +} diff --git a/crates/service/src/gateway/upstream/proxy_pipeline/stream_preflight.rs b/crates/service/src/gateway/upstream/proxy_pipeline/stream_preflight.rs index 48fe2989e..fe3a28685 100644 --- a/crates/service/src/gateway/upstream/proxy_pipeline/stream_preflight.rs +++ b/crates/service/src/gateway/upstream/proxy_pipeline/stream_preflight.rs @@ -28,8 +28,15 @@ enum PrefixDecision { pub(in super::super) enum StreamPreflightOutcome { Ready(GatewayUpstreamResponse), Failover(String), - StatusFailover { status_code: u16, message: String }, + StatusFailover { + status_code: u16, + message: String, + }, RetryUsageNotice(String), + ModelUnsupported { + message: String, + response: GatewayUpstreamResponse, + }, TransportFailover(String), } @@ -361,6 +368,45 @@ async fn preflight_stream_response_with_timeouts( wall_clock_timeout: Option, ) -> StreamPreflightOutcome { let status_code = response.status().as_u16(); + if status_code == 400 { + // Classification has independent byte, idle and wall-clock bounds, including + // non-streaming requests and the final candidate. All consumed bytes replay. + let idle_timeout = Some( + idle_timeout + .unwrap_or(Duration::from_secs(2)) + .min(Duration::from_secs(2)), + ); + let wall_timeout = Some( + wall_clock_timeout + .unwrap_or(STREAM_PREFLIGHT_WALL_CLOCK_TIMEOUT) + .min(STREAM_PREFLIGHT_WALL_CLOCK_TIMEOUT), + ); + let (body, response, terminal) = response + .prefetch_stream_prefix_async( + STREAM_PREFLIGHT_MAX_BYTES, + idle_timeout, + wall_timeout, + |_| false, + ) + .await; + // Only a complete body can establish eligibility. Truncated/timeout/error + // prefixes may contain misleading or incomplete rejection JSON. + if matches!(terminal, GatewayStreamPrefetchTerminal::Eof) { + if let Some(message) = + crate::account::model_support::account_model_rejection_message(&body) + { + return StreamPreflightOutcome::ModelUnsupported { message, response }; + } + } + return if has_more_candidates { + StreamPreflightOutcome::StatusFailover { + status_code, + message: summarize_non_200_status_failover(status_code, Some(&body)), + } + } else { + StreamPreflightOutcome::Ready(response) + }; + } if has_more_candidates && !(200..=299).contains(&status_code) { if should_prefetch_actionable_error_body(status_code) { return match response.into_buffered_async().await { diff --git a/crates/service/src/gateway/upstream/proxy_pipeline/stream_preflight_tests.rs b/crates/service/src/gateway/upstream/proxy_pipeline/stream_preflight_tests.rs index 9893324a0..adea7aa90 100644 --- a/crates/service/src/gateway/upstream/proxy_pipeline/stream_preflight_tests.rs +++ b/crates/service/src/gateway/upstream/proxy_pipeline/stream_preflight_tests.rs @@ -572,3 +572,196 @@ fn preflight_fails_over_when_producer_disconnects_after_metadata() { if message.contains("disconnected") )); } + +#[test] +fn model_rejection_classification_and_last_response_replay_cover_both_endpoints_and_modes() { + let body = r#"{"error":{"message":"The 'gpt-6-luna' model is not supported when using Codex with a ChatGPT account."}}"#; + for path in ["/v1/responses", "/v1/chat/completions"] { + for is_stream in [false, true] { + for has_more in [false, true] { + let outcome = crate::gateway::run_upstream_io(preflight_stream_response( + json_stream_response_with_status(reqwest::StatusCode::BAD_REQUEST, body), + path, + is_stream, + has_more, + )) + .unwrap(); + let StreamPreflightOutcome::ModelUnsupported { message, response } = outcome else { + panic!("complete explicit rejection must be learned even for last candidate"); + }; + assert!(message.contains("gpt-6-luna")); + assert_eq!(response.status().as_u16(), 400); + let (replayed, _) = response.into_buffered().unwrap(); + assert_eq!(replayed.as_ref(), body.as_bytes()); + } + } + } +} + +#[test] +fn oversized_model_rejection_is_not_learned_and_last_body_replays_losslessly() { + let mut body = serde_json::to_vec(&serde_json::json!({ + "error": {"message": "The 'gpt-6-luna' model is not supported when using Codex with a ChatGPT account."}, + "padding": "x".repeat(STREAM_PREFLIGHT_MAX_BYTES), + })).unwrap(); + body.extend_from_slice(b" "); + for has_more in [false, true] { + let response = GatewayUpstreamResponse::Stream(GatewayStreamResponse::new( + reqwest::StatusCode::BAD_REQUEST, + HeaderMap::new(), + GatewayByteStream::from_bytes(Bytes::from(body.clone())), + )); + let outcome = crate::gateway::run_upstream_io(preflight_stream_response( + response, + "/v1/responses", + true, + has_more, + )) + .unwrap(); + if has_more { + assert!(matches!( + outcome, + StreamPreflightOutcome::StatusFailover { + status_code: 400, + .. + } + )); + } else { + let StreamPreflightOutcome::Ready(response) = outcome else { + panic!("last candidate must remain deliverable"); + }; + let (replayed, _) = response.into_buffered().unwrap(); + assert_eq!(replayed.as_ref(), body.as_slice()); + } + } +} + +#[test] +fn hanging_error_body_obeys_idle_and_wall_clock_caps_without_learning_prefix() { + for has_more in [false, true] { + for wall_cap in [false, true] { + let (tx, rx) = mpsc::channel(2); + let body = Bytes::from_static(b"{\"error\":{\"message\":\"The 'gpt-6-luna' model is not supported when using Codex with a ChatGPT account.\"}}"); + tx.blocking_send(GatewayByteStreamItem::Chunk(body.clone())) + .unwrap(); + let response = GatewayUpstreamResponse::Stream(GatewayStreamResponse::new( + reqwest::StatusCode::BAD_REQUEST, + HeaderMap::new(), + GatewayByteStream::from_receiver(rx), + )); + let started = std::time::Instant::now(); + let outcome = crate::gateway::run_upstream_io(preflight_stream_response_with_timeouts( + response, + "/v1/chat/completions", + false, + has_more, + Some(Duration::from_millis(if wall_cap { 500 } else { 20 })), + Some(Duration::from_millis(if wall_cap { 20 } else { 500 })), + )) + .unwrap(); + assert!(started.elapsed() < Duration::from_millis(300)); + if has_more { + assert!(matches!( + outcome, + StreamPreflightOutcome::StatusFailover { + status_code: 400, + .. + } + )); + } else { + tx.blocking_send(GatewayByteStreamItem::Eof).unwrap(); + let StreamPreflightOutcome::Ready(response) = outcome else { + panic!("timeout must preserve final response"); + }; + let (replayed, _) = response.into_buffered().unwrap(); + assert_eq!(replayed, body); + } + } + } +} + +#[test] +fn broken_error_body_preserves_last_candidate_and_never_learns_partial_json() { + for has_more in [false, true] { + let (tx, rx) = mpsc::channel(3); + tx.blocking_send(GatewayByteStreamItem::Chunk(Bytes::from_static( + b"{\"error\":", + ))) + .unwrap(); + tx.blocking_send(GatewayByteStreamItem::Error( + "upstream read failed".to_owned(), + )) + .unwrap(); + drop(tx); + let response = GatewayUpstreamResponse::Stream(GatewayStreamResponse::new( + reqwest::StatusCode::BAD_REQUEST, + HeaderMap::new(), + GatewayByteStream::from_receiver(rx), + )); + let outcome = crate::gateway::run_upstream_io(preflight_stream_response( + response, + "/v1/responses", + false, + has_more, + )) + .unwrap(); + if has_more { + assert!(matches!( + outcome, + StreamPreflightOutcome::StatusFailover { + status_code: 400, + .. + } + )); + } else { + let StreamPreflightOutcome::Ready(response) = outcome else { + panic!("last response must preserve its read error"); + }; + assert!(response + .into_buffered() + .unwrap_err() + .contains("upstream read failed")); + } + } +} + +#[tokio::test] +async fn continuously_dripping_error_body_stops_at_wall_clock_limit() { + let (tx, rx) = mpsc::channel(2); + let producer = tokio::spawn(async move { + loop { + if tx + .send(GatewayByteStreamItem::Chunk(Bytes::from_static(b" "))) + .await + .is_err() + { + break; + } + tokio::time::sleep(Duration::from_millis(2)).await; + } + }); + let response = GatewayUpstreamResponse::Stream(GatewayStreamResponse::new( + reqwest::StatusCode::BAD_REQUEST, + HeaderMap::new(), + GatewayByteStream::from_receiver(rx), + )); + let started = std::time::Instant::now(); + let outcome = preflight_stream_response_with_timeouts( + response, + "/v1/responses", + true, + true, + Some(Duration::from_secs(1)), + Some(Duration::from_millis(25)), + ) + .await; + assert!(started.elapsed() < Duration::from_millis(300)); + assert!(matches!( + outcome, + StreamPreflightOutcome::StatusFailover { + status_code: 400, + .. + } + )); + producer.await.unwrap(); +} diff --git a/crates/service/src/gateway/upstream/support/candidates.rs b/crates/service/src/gateway/upstream/support/candidates.rs index 05ab4700b..573d855f9 100644 --- a/crates/service/src/gateway/upstream/support/candidates.rs +++ b/crates/service/src/gateway/upstream/support/candidates.rs @@ -10,6 +10,55 @@ pub(in super::super) enum CandidateSkipReason { Inflight, } +/// 过滤“已知该账户不支持该模型”的候选(BUG-2026-0930-01)。 +/// +/// 记忆只在收到该类上游精确拒绝后写入,并带 TTL;若所有候选都被标记,则保留 +/// 原列表,让上游给出精确原因(避免把可诊断的 400 换成笼统的“无可用账户”)。 +fn drop_known_unsupported_model_candidates( + candidates: &mut Vec<(Account, Token)>, + request_model: Option<&str>, +) { + log::debug!( + "event=gateway_candidate_filter_entry model={:?} count={} order={:?}", + request_model, + candidates.len(), + candidates + .iter() + .map(|(account, _)| account.id.chars().take(12).collect::()) + .take(4) + .collect::>() + ); + let Some(model) = request_model + .map(str::trim) + .filter(|value| !value.is_empty()) + else { + return; + }; + if candidates.is_empty() { + return; + } + let drop_positions = candidates + .iter() + .enumerate() + .filter(|(_, (account, _))| { + crate::account::model_support::is_unsupported(account.id.as_str(), model) + }) + .map(|(index, _)| index) + .collect::>(); + if drop_positions.is_empty() || drop_positions.len() == candidates.len() { + return; + } + log::debug!( + "event=gateway_candidate_model_filter model={} before={} dropped={}", + model, + candidates.len(), + drop_positions.len() + ); + for index in drop_positions.into_iter().rev() { + candidates.remove(index); + } +} + /// 函数 `prepare_gateway_candidates` /// /// 作者: gaohongshun @@ -28,6 +77,9 @@ pub(crate) fn prepare_gateway_candidates( account_plan_filter: Option<&str>, low_quota_mode: super::super::super::LowQuotaCandidateMode, ) -> Result, String> { + let support_model_override = + crate::account::model_support::account_model_override_for_request(storage, request_model); + let support_model = support_model_override.as_deref().or(request_model); let storage = &crate::account::remote_storage::AccountStorage::new(storage); let normalized_group_filter = account_group_filter .map(str::trim) @@ -45,10 +97,12 @@ pub(crate) fn prepare_gateway_candidates( && !exclude_free_accounts && !reserve_model { - return super::super::super::collect_gateway_candidates_with_low_quota_mode( + let mut candidates = super::super::super::collect_gateway_candidates_with_low_quota_mode( storage, low_quota_mode, - ); + )?; + drop_known_unsupported_model_candidates(&mut candidates, support_model); + return Ok(candidates); } let mut authorized_candidates = if reserve_model { @@ -109,14 +163,17 @@ pub(crate) fn prepare_gateway_candidates( .map(|(account, _)| account.id) .collect::>(); // 中文注释:保持账号原始顺序(按账户排序字段)作为候选顺序,失败时再依次切下一个。 - if reserve_model { - return collect_luna_reserve_candidates(storage, &authorized_account_ids, &snapshots); - } - super::super::super::collect_gateway_candidates_for_account_ids_with_low_quota_mode( - storage, - &authorized_account_ids, - low_quota_mode, - ) + let mut candidates = if reserve_model { + collect_luna_reserve_candidates(storage, &authorized_account_ids, &snapshots)? + } else { + super::super::super::collect_gateway_candidates_for_account_ids_with_low_quota_mode( + storage, + &authorized_account_ids, + low_quota_mode, + )? + }; + drop_known_unsupported_model_candidates(&mut candidates, support_model); + Ok(candidates) } fn collect_luna_reserve_candidates( diff --git a/crates/service/src/gateway/upstream/support/candidates_tests.rs b/crates/service/src/gateway/upstream/support/candidates_tests.rs index b2eb0beda..615d637eb 100644 --- a/crates/service/src/gateway/upstream/support/candidates_tests.rs +++ b/crates/service/src/gateway/upstream/support/candidates_tests.rs @@ -870,3 +870,95 @@ fn candidate_skip_reason_for_proxy_can_skip_last_cooldown_candidate() { assert_eq!(default_last, None); assert_eq!(strict_last, Some(CandidateSkipReason::Cooldown)); } + +#[test] +fn model_filter_precedes_sticky_selection_and_keeps_all_unsupported_fallback() { + use crate::account::model_support::{clear_unsupported, mark_unsupported}; + use crate::gateway::conversation_binding::RouteConversationSource; + use codexmanager_core::storage::{ConversationBinding, ManagedModelV2Upsert}; + let _guard = crate::test_env_guard(); + let storage = Storage::open_in_memory().unwrap(); + storage.init().unwrap(); + let alias = codexmanager_core::usage::LUNA_MODEL_SLUG; + let first = "model-qualification-first"; + let second = "model-qualification-second"; + insert_active_account_with_token(&storage, first, 0); + insert_active_account_with_token(&storage, second, 1); + let mut mapped = storage.get_managed_model_v2(alias).unwrap().unwrap(); + mapped + .routes + .iter_mut() + .find(|route| route.source_kind == "account_pool" && route.source_id == "default") + .unwrap() + .upstream_model = "gpt-qualification-final".to_owned(); + storage + .upsert_managed_model_v2(&ManagedModelV2Upsert { + previous_slug: Some(alias.to_owned()), + model: mapped, + }) + .unwrap(); + mark_unsupported(first, "gpt-qualification-final"); + let mut candidates = super::prepare_gateway_candidates( + &storage, + Some(alias), + None, + None, + crate::gateway::LowQuotaCandidateMode::NormalOnly, + ) + .unwrap(); + assert_eq!( + candidates.len(), + 1, + "lookup must use the configured final model alias" + ); + assert_eq!(candidates[0].0.id, second); + let binding = ConversationBinding { + platform_key_hash: "qualification-key".to_owned(), + conversation_id: "qualification-conversation".to_owned(), + account_id: first.to_owned(), + thread_epoch: 1, + thread_anchor: "qualification-anchor".to_owned(), + status: "active".to_owned(), + last_model: Some(alias.to_owned()), + last_switch_reason: None, + created_at: 1, + updated_at: 1, + last_used_at: 1, + }; + let setup = crate::gateway::upstream::proxy_pipeline::request_setup::prepare_request_setup( + &storage, + "/v1/responses", + crate::apikey_profile::PROTOCOL_OPENAI_COMPAT, + false, + &crate::gateway::IncomingHeaderSnapshot::default(), + &bytes::Bytes::from_static(b"{}"), + &mut candidates, + "qualification-key", + "qualification-key", + Some("qualification-conversation"), + RouteConversationSource::PromptCacheKeyExistingOnly, + Some(&binding), + Some(alias), + "qualification-trace", + ); + let routing = setup.conversation_routing.unwrap(); + assert!(routing.existing_binding.is_none()); + assert!(!routing.binding_selected); + assert_eq!(candidates[0].0.id, second); + mark_unsupported(second, "gpt-qualification-final"); + let fallback = super::prepare_gateway_candidates( + &storage, + Some(alias), + None, + None, + crate::gateway::LowQuotaCandidateMode::NormalOnly, + ) + .unwrap(); + assert_eq!( + fallback.len(), + 2, + "all rejected preserves diagnosable last-account delivery" + ); + clear_unsupported(first, "gpt-qualification-final"); + clear_unsupported(second, "gpt-qualification-final"); +} diff --git a/crates/service/src/quota/model_pricing_tests.rs b/crates/service/src/quota/model_pricing_tests.rs index 0987fd067..ddd34cc61 100644 --- a/crates/service/src/quota/model_pricing_tests.rs +++ b/crates/service/src/quota/model_pricing_tests.rs @@ -19,7 +19,7 @@ fn prices() -> (Storage, Vec) { #[test] fn catalog_prices_are_exact_and_missing_prices_do_not_fallback() { let (_storage, prices) = prices(); - assert_eq!(prices.len(), 11); + assert_eq!(prices.len(), 12); let astra = resolve_model_price_from_catalog(&prices, "gpt-6-astra", 0).expect("astra"); assert_close(astra.input_price_per_1m, 10.0); assert_close(astra.cached_input_price_per_1m, 1.0); @@ -42,6 +42,18 @@ fn catalog_prices_are_exact_and_missing_prices_do_not_fallback() { assert_close(sol6_long.cached_input_price_per_1m, 0.4); assert_close(sol6_long.cache_write_price_per_1m, 5.0); assert_close(sol6_long.output_price_per_1m, 15.0); + let sol61 = + resolve_model_price_from_catalog(&prices, "gpt-6.1-sol", 272_000).expect("GPT-6.1 Sol"); + assert_close(sol61.input_price_per_1m, 2.0); + assert_close(sol61.cached_input_price_per_1m, 0.1); + assert_close(sol61.cache_write_price_per_1m, 2.5); + assert_close(sol61.output_price_per_1m, 10.0); + let sol61_long = resolve_model_price_from_catalog(&prices, "gpt-6.1-sol", 272_001) + .expect("GPT-6.1 Sol long"); + assert_close(sol61_long.input_price_per_1m, 4.0); + assert_close(sol61_long.cached_input_price_per_1m, 0.2); + assert_close(sol61_long.cache_write_price_per_1m, 5.0); + assert_close(sol61_long.output_price_per_1m, 15.0); let luna6 = resolve_model_price_from_catalog(&prices, "gpt-6-luna", 0).expect("GPT-6 Luna"); assert_close(luna6.input_price_per_1m, 0.1); assert_close(luna6.cached_input_price_per_1m, 0.01); diff --git a/crates/service/src/quota/read_tests.rs b/crates/service/src/quota/read_tests.rs index e4c994ed5..f365ee7a9 100644 --- a/crates/service/src/quota/read_tests.rs +++ b/crates/service/src/quota/read_tests.rs @@ -143,10 +143,16 @@ fn api_available_model_slugs_preserves_catalog_sort_order() { storage.init().expect("init storage"); let models = api_available_model_slugs(&storage).expect("available models"); - assert_eq!(models.len(), 10); + assert_eq!(models.len(), 11); assert_eq!( - &models[..4], - ["gpt-6-astra", "gpt-6-sol", "gpt-6-luna", "gpt-5.6-sol"] + &models[..5], + [ + "gpt-6-astra", + "gpt-6-sol", + "gpt-6.1-sol", + "gpt-6-luna", + "gpt-5.6-sol", + ] ); assert!(!models.iter().any(|model| model == "codex-auto-review")); for slug in [ @@ -164,7 +170,7 @@ fn api_available_model_slugs_does_not_seed_legacy_price_rules() { storage.init().expect("init storage"); let models = api_available_model_slugs(&storage).expect("available models"); - assert_eq!(models.len(), 10); + assert_eq!(models.len(), 11); assert_eq!( storage .list_enabled_model_price_rules() diff --git a/docs/en/report/environment-and-runtime-config.md b/docs/en/report/environment-and-runtime-config.md index c442ac06a..c1aee35f2 100644 --- a/docs/en/report/environment-and-runtime-config.md +++ b/docs/en/report/environment-and-runtime-config.md @@ -73,6 +73,10 @@ ## 常用变量分组 +### 账户模型资格缓存 + +- `CODEXMANAGER_ACCOUNT_MODEL_UNSUPPORTED_TTL_SECS`:账户与最终发送模型的资格拒绝缓存有效期,默认 `1800` 秒。未设置、`0`、负数或非法值均回退默认值。支持通过高级环境变量设置;更新只影响之后写入的拒绝记录,已有记录沿用写入时的有效期。该缓存只在当前 service 进程内生效,桌面端内置 service 与独立 service 各自维护;重启后清空,不写入数据库。仅完整的明确模型资格错误会写入,工具模型与参数错误不会影响主模型;已确认完成的成功响应会清除对应账户与模型记录。 + ### 地址与入口 - `CODEXMANAGER_SERVICE_ADDR`:service 地址,默认 `localhost:48760` diff --git "a/docs/zh-CN/report/\347\216\257\345\242\203\345\217\230\351\207\217\344\270\216\350\277\220\350\241\214\351\205\215\347\275\256\350\257\264\346\230\216.md" "b/docs/zh-CN/report/\347\216\257\345\242\203\345\217\230\351\207\217\344\270\216\350\277\220\350\241\214\351\205\215\347\275\256\350\257\264\346\230\216.md" index cd17e6870..7c44a9917 100644 --- "a/docs/zh-CN/report/\347\216\257\345\242\203\345\217\230\351\207\217\344\270\216\350\277\220\350\241\214\351\205\215\347\275\256\350\257\264\346\230\216.md" +++ "b/docs/zh-CN/report/\347\216\257\345\242\203\345\217\230\351\207\217\344\270\216\350\277\220\350\241\214\351\205\215\347\275\256\350\257\264\346\230\216.md" @@ -73,6 +73,10 @@ ## 常用变量分组 +### 账户模型资格缓存 + +- `CODEXMANAGER_ACCOUNT_MODEL_UNSUPPORTED_TTL_SECS`:账户与最终发送模型的资格拒绝缓存有效期,默认 `1800` 秒。未设置、`0`、负数或非法值均回退默认值。支持通过高级环境变量设置;更新只影响之后写入的拒绝记录,已有记录沿用写入时的有效期。该缓存只在当前 service 进程内生效,桌面端内置 service 与独立 service 各自维护;重启后清空,不写入数据库。仅完整的明确模型资格错误会写入,工具模型与参数错误不会影响主模型;已确认完成的成功响应会清除对应账户与模型记录。 + ### 地址与入口 - `CODEXMANAGER_SERVICE_ADDR`:service 地址,默认 `localhost:48760`