diff --git a/src-tauri/src/proxy/forwarder.rs b/src-tauri/src/proxy/forwarder.rs index a9b470cf6..de6c692d5 100644 --- a/src-tauri/src/proxy/forwarder.rs +++ b/src-tauri/src/proxy/forwarder.rs @@ -11,8 +11,8 @@ use super::{ log_codes::fwd as log_fwd, provider_router::ProviderRouter, providers::{ - gemini_shadow::GeminiShadowStore, get_adapter, AuthInfo, AuthStrategy, ProviderAdapter, - ProviderType, + codex_chat_history::CodexChatHistoryStore, gemini_shadow::GeminiShadowStore, get_adapter, + AuthInfo, AuthStrategy, ProviderAdapter, ProviderType, }, thinking_budget_rectifier::{rectify_thinking_budget, should_rectify_thinking_budget}, thinking_rectifier::{ @@ -92,6 +92,7 @@ pub struct RequestForwarder { status: Arc>, current_providers: Arc>>, gemini_shadow: Arc, + codex_chat_history: Arc, /// 故障转移切换管理器 failover_manager: Arc, /// AppHandle，用于发射事件和更新托盘 @@ -128,6 +129,7 @@ impl RequestForwarder { status: Arc>, current_providers: Arc>>, gemini_shadow: Arc, + codex_chat_history: Arc, failover_manager: Arc, app_handle: Option, current_provider_id_at_start: String, @@ -148,6 +150,7 @@ impl RequestForwarder { status, current_providers, gemini_shadow, + codex_chat_history, failover_manager, app_handle, current_provider_id_at_start, @@ -1146,6 +1149,17 @@ impl RequestForwarder { // 转换请求体（如果需要） let request_body = if codex_responses_to_chat { + let mut mapped_body = mapped_body; + let restored = self + .codex_chat_history + .enrich_request(&mut mapped_body) + .await; + if restored > 0 { + log::debug!( + "[Codex] Restored {restored} cached function call(s) for Chat upstream" + ); + } + super::providers::apply_codex_chat_upstream_model(provider, &mut mapped_body); super::providers::transform_codex_chat::responses_to_chat_completions(mapped_body)? } else if needs_transform { if adapter.name() == "Claude" { @@ -2389,6 +2403,7 @@ mod tests { status: Arc::new(RwLock::new(ProxyStatus::default())), current_providers: Arc::new(RwLock::new(HashMap::new())), gemini_shadow: Arc::new(GeminiShadowStore::new()), + codex_chat_history: Arc::new(CodexChatHistoryStore::default()), failover_manager: Arc::new(FailoverSwitchManager::new(db)), app_handle: None, current_provider_id_at_start: String::new(), diff --git a/src-tauri/src/proxy/handler_context.rs b/src-tauri/src/proxy/handler_context.rs index 987eed406..b1662733f 100644 --- a/src-tauri/src/proxy/handler_context.rs +++ b/src-tauri/src/proxy/handler_context.rs @@ -223,6 +223,7 @@ impl RequestContext { state.status.clone(), state.current_providers.clone(), state.gemini_shadow.clone(), + state.codex_chat_history.clone(), state.failover_manager.clone(), state.app_handle.clone(), self.current_provider_id.clone(), diff --git a/src-tauri/src/proxy/handlers.rs b/src-tauri/src/proxy/handlers.rs index 956fc0bbb..ee66edfa5 100644 --- a/src-tauri/src/proxy/handlers.rs +++ b/src-tauri/src/proxy/handlers.rs @@ -16,7 +16,8 @@ use super::{ }, handler_context::RequestContext, providers::{ - get_adapter, get_claude_api_format, streaming::create_anthropic_sse_stream, + codex_chat_history::record_responses_sse_stream, get_adapter, get_claude_api_format, + streaming::create_anthropic_sse_stream, streaming_codex_chat::create_responses_sse_stream_from_chat, streaming_gemini::create_anthropic_sse_stream_from_gemini, streaming_responses::create_anthropic_sse_stream_from_responses, transform, @@ -701,6 +702,7 @@ async fn handle_codex_chat_to_responses_transform( if is_stream || response.is_sse() { let stream = response.bytes_stream(); let sse_stream = create_responses_sse_stream_from_chat(stream); + let sse_stream = record_responses_sse_stream(sse_stream, state.codex_chat_history.clone()); let usage_collector = if usage_logging_enabled(state) { let state = state.clone(); @@ -786,6 +788,10 @@ async fn handle_codex_chat_to_responses_transform( log::error!("[Codex] Chat → Responses 响应转换失败: {e}"); e })?; + state + .codex_chat_history + .record_response(&responses_response) + .await; if let Some(usage) = TokenUsage::from_codex_response_auto(&responses_response) { let model = responses_response diff --git a/src-tauri/src/proxy/providers/codex.rs b/src-tauri/src/proxy/providers/codex.rs index e5e54288b..9a8e28dfa 100644 --- a/src-tauri/src/proxy/providers/codex.rs +++ b/src-tauri/src/proxy/providers/codex.rs @@ -9,6 +9,7 @@ use super::{AuthInfo, AuthStrategy, ProviderAdapter}; use crate::provider::Provider; use crate::proxy::error::ProxyError; use regex::Regex; +use serde_json::Value as JsonValue; use std::sync::LazyLock; use toml::Value as TomlValue; @@ -20,6 +21,13 @@ static CODEX_CLIENT_REGEX: LazyLock = /// Codex 适配器 pub struct CodexAdapter; +/// Local model written into Codex config.toml for Chat-only upstreams. +/// +/// Codex itself must see a model it can load metadata for; the real provider +/// model is restored inside the proxy immediately before Chat Completions +/// conversion. +pub const CODEX_CHAT_CLIENT_MODEL: &str = "gpt-5.4"; + /// Whether this Codex provider's real upstream should be called through /// OpenAI Chat Completions, even if the local Codex client is talking to CC /// Switch through the Responses API. @@ -82,6 +90,39 @@ pub fn should_convert_codex_responses_to_chat(provider: &Provider, endpoint: &st ) && codex_provider_uses_chat_completions(provider) } +/// Extract the real upstream model configured for a Codex provider. +pub fn codex_provider_upstream_model(provider: &Provider) -> Option { + provider + .settings_config + .get("model") + .and_then(|v| v.as_str()) + .map(str::trim) + .filter(|model| !model.is_empty()) + .map(ToString::to_string) + .or_else(|| { + provider + .settings_config + .get("config") + .and_then(|v| v.as_str()) + .and_then(extract_codex_model_from_toml) + }) +} + +/// For Codex Chat providers, replace the local Codex-safe model with the real +/// upstream model before converting the request to Chat Completions. +pub fn apply_codex_chat_upstream_model( + provider: &Provider, + body: &mut JsonValue, +) -> Option { + if !codex_provider_uses_chat_completions(provider) { + return None; + } + + let upstream_model = codex_provider_upstream_model(provider)?; + body["model"] = JsonValue::String(upstream_model.clone()); + Some(upstream_model) +} + fn is_chat_wire_api(value: &str) -> bool { matches!( value.trim().to_ascii_lowercase().as_str(), @@ -120,6 +161,16 @@ fn extract_codex_wire_api_from_toml(config_text: &str) -> Option { .map(ToString::to_string) } +fn extract_codex_model_from_toml(config_text: &str) -> Option { + let doc = config_text.parse::().ok()?; + + doc.get("model") + .and_then(|v| v.as_str()) + .map(str::trim) + .filter(|model| !model.is_empty()) + .map(ToString::to_string) +} + fn extract_codex_base_url_from_toml(config_text: &str) -> Option { let doc = config_text.parse::().ok()?; @@ -481,4 +532,51 @@ wire_api = "chat" "/responses/compact?stream=true" )); } + + #[test] + fn test_codex_provider_uses_chat_completions_from_meta_api_format_for_responses() { + let mut provider = create_provider(json!({ + "base_url": "https://api.deepseek.com/v1" + })); + provider.meta = Some(crate::provider::ProviderMeta { + api_format: Some("openai_chat".to_string()), + ..Default::default() + }); + + assert!(should_convert_codex_responses_to_chat( + &provider, + "/v1/responses" + )); + } + + #[test] + fn test_apply_codex_chat_upstream_model_uses_provider_config_model() { + let mut provider = create_provider(json!({ + "config": r#" +model_provider = "deepseek" +model = "deepseek-v4-flash" + +[model_providers.deepseek] +name = "DeepSeek" +base_url = "https://api.deepseek.com/v1" +wire_api = "responses" +"# + })); + provider.meta = Some(crate::provider::ProviderMeta { + api_format: Some("openai_chat".to_string()), + ..Default::default() + }); + let mut body = json!({ + "model": CODEX_CHAT_CLIENT_MODEL, + "input": "ping" + }); + + let upstream_model = apply_codex_chat_upstream_model(&provider, &mut body); + + assert_eq!(upstream_model.as_deref(), Some("deepseek-v4-flash")); + assert_eq!( + body.get("model").and_then(|v| v.as_str()), + Some("deepseek-v4-flash") + ); + } } diff --git a/src-tauri/src/proxy/providers/codex_chat_common.rs b/src-tauri/src/proxy/providers/codex_chat_common.rs new file mode 100644 index 000000000..3989de692 --- /dev/null +++ b/src-tauri/src/proxy/providers/codex_chat_common.rs @@ -0,0 +1,172 @@ +use serde_json::{json, Map, Value}; + +const THINK_OPEN_TAG: &str = ""; +const THINK_CLOSE_TAG: &str = ""; + +pub(crate) fn extract_reasoning_field_text(value: &Value) -> Option { + for key in ["reasoning_content", "reasoning"] { + if let Some(text) = value.get(key).and_then(|v| v.as_str()) { + if !text.is_empty() { + return Some(text.to_string()); + } + } + } + + let reasoning = value.get("reasoning")?; + for key in ["content", "text", "summary"] { + if let Some(text) = reasoning.get(key).and_then(|v| v.as_str()) { + if !text.is_empty() { + return Some(text.to_string()); + } + } + } + + None +} + +pub(crate) fn extract_reasoning_summary_text(value: &Value) -> Option { + for key in ["reasoning_content", "content", "text"] { + if let Some(text) = value.get(key).and_then(|v| v.as_str()) { + if !text.is_empty() { + return Some(text.to_string()); + } + } + } + + let summary = value.get("summary")?; + if let Some(text) = summary.as_str() { + return (!text.is_empty()).then(|| text.to_string()); + } + + let parts = summary.as_array()?; + let text = parts + .iter() + .filter_map(|part| { + part.get("text") + .and_then(|v| v.as_str()) + .or_else(|| part.get("content").and_then(|v| v.as_str())) + .or_else(|| part.as_str()) + }) + .filter(|text| !text.is_empty()) + .collect::>() + .join("\n\n"); + + (!text.is_empty()).then_some(text) +} + +pub(crate) fn append_reasoning_content(message: &mut Map, reasoning: &str) -> bool { + let reasoning = reasoning.trim(); + if reasoning.is_empty() { + return false; + } + + match message.get_mut("reasoning_content") { + Some(Value::String(existing)) if !existing.is_empty() => { + existing.push_str("\n\n"); + existing.push_str(reasoning); + } + _ => { + message.insert( + "reasoning_content".to_string(), + Value::String(reasoning.to_string()), + ); + } + } + true +} + +pub(crate) fn attach_reasoning_content_field(item: &mut Value, reasoning: &str) -> bool { + let reasoning = reasoning.trim(); + if reasoning.is_empty() { + return false; + } + + if let Some(obj) = item.as_object_mut() { + obj.insert( + "reasoning_content".to_string(), + Value::String(reasoning.to_string()), + ); + return true; + } + + false +} + +pub(crate) fn attach_optional_reasoning_content_field( + item: &mut Value, + reasoning: Option<&str>, +) -> bool { + let Some(reasoning) = reasoning else { + return false; + }; + attach_reasoning_content_field(item, reasoning) +} + +pub(crate) fn response_function_call_item( + item_id: &str, + status: &str, + call_id: &str, + name: &str, + arguments: &str, + reasoning: Option<&str>, +) -> Value { + let mut item = json!({ + "id": item_id, + "type": "function_call", + "status": status, + "call_id": call_id, + "name": name, + "arguments": arguments + }); + attach_optional_reasoning_content_field(&mut item, reasoning); + item +} + +pub(crate) fn response_item_call_id(item: &Value) -> Option { + item.get("call_id") + .or_else(|| item.get("id")) + .and_then(|value| value.as_str()) + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(ToString::to_string) +} + +pub(crate) fn is_empty_value(value: &Value) -> bool { + match value { + Value::Null => true, + Value::String(value) => value.trim().is_empty(), + Value::Array(value) => value.is_empty(), + Value::Object(value) => value.is_empty(), + _ => false, + } +} + +pub(crate) fn split_leading_think_block(text: &str) -> Option<(String, String)> { + let leading_ws_len = text.len() - text.trim_start().len(); + let after_ws = &text[leading_ws_len..]; + if !after_ws.starts_with(THINK_OPEN_TAG) { + return None; + } + + let body_start = leading_ws_len + THINK_OPEN_TAG.len(); + let close_relative = text[body_start..].find(THINK_CLOSE_TAG)?; + let close_start = body_start + close_relative; + let answer_start = close_start + THINK_CLOSE_TAG.len(); + + Some(( + text[body_start..close_start].trim().to_string(), + strip_think_answer_separator(&text[answer_start..]).to_string(), + )) +} + +pub(crate) fn strip_leading_think_open_tag(text: &str) -> Option { + let leading_ws_len = text.len() - text.trim_start().len(); + let after_ws = &text[leading_ws_len..]; + after_ws + .strip_prefix(THINK_OPEN_TAG) + .map(|value| value.trim().to_string()) +} + +fn strip_think_answer_separator(text: &str) -> &str { + text.trim_start_matches(['\r', '\n', '\t', ' ']) +} diff --git a/src-tauri/src/proxy/providers/codex_chat_history.rs b/src-tauri/src/proxy/providers/codex_chat_history.rs new file mode 100644 index 000000000..9f0dab4ac --- /dev/null +++ b/src-tauri/src/proxy/providers/codex_chat_history.rs @@ -0,0 +1,561 @@ +use super::codex_chat_common::{is_empty_value, response_item_call_id}; +use crate::proxy::sse::{append_utf8_safe, strip_sse_field, take_sse_block}; +use bytes::Bytes; +use futures::{Stream, StreamExt}; +use serde_json::Value; +use std::collections::{HashMap, HashSet, VecDeque}; +use std::sync::Arc; +use tokio::sync::RwLock; + +const MAX_CACHED_RESPONSES: usize = 512; + +#[derive(Debug, Clone, Default)] +struct CachedResponse { + calls_by_id: HashMap, + call_order: Vec, +} + +#[derive(Debug, Default)] +struct CodexChatHistoryInner { + responses: HashMap, + response_order: VecDeque, +} + +#[derive(Debug, Clone, Default)] +struct CachedLookup { + previous: Option, +} + +/// Cross-request history needed when Codex Responses is bridged to Chat +/// Completions. +/// +/// Chat providers such as DeepSeek require an assistant message with the +/// original tool call and its `reasoning_content` immediately before the tool +/// result. Codex often sends follow-up requests as +/// `previous_response_id + function_call_output`, so this store restores the +/// missing function call before the request is converted to Chat messages. +#[derive(Debug, Default)] +pub struct CodexChatHistoryStore { + inner: RwLock, +} + +impl CodexChatHistoryStore { + pub async fn record_response(&self, response: &Value) -> usize { + let Some(response_id) = response + .get("id") + .and_then(|value| value.as_str()) + .filter(|value| !value.is_empty()) + else { + return 0; + }; + + let calls = response + .get("output") + .and_then(|value| value.as_array()) + .map(|items| { + items + .iter() + .filter_map(cached_function_call) + .collect::>() + }) + .unwrap_or_default(); + + if calls.is_empty() { + return 0; + } + + let mut inner = self.inner.write().await; + inner.insert_calls(response_id, calls) + } + + async fn record_function_call(&self, response_id: Option<&str>, item: &Value) -> bool { + let Some(call) = cached_function_call(item) else { + return false; + }; + + let mut inner = self.inner.write().await; + if let Some(response_id) = response_id.filter(|value| !value.is_empty()) { + inner.insert_calls(response_id, vec![call]) > 0 + } else { + false + } + } + + pub async fn enrich_request(&self, body: &mut Value) -> usize { + let previous_response_id = body + .get("previous_response_id") + .and_then(|value| value.as_str()) + .filter(|value| !value.is_empty()); + let lookup = self.lookup(previous_response_id).await; + + let Some(input) = body.get_mut("input") else { + return 0; + }; + + let original_input = std::mem::take(input); + let original_was_object = matches!(&original_input, Value::Object(_)); + let items = match original_input { + Value::Array(items) => items, + Value::Object(object) => vec![Value::Object(object)], + other => { + *input = other; + return 0; + } + }; + + let output_call_ids = items + .iter() + .filter(|item| { + item.get("type").and_then(|value| value.as_str()) == Some("function_call_output") + }) + .filter_map(response_item_call_id) + .collect::>(); + let existing_call_ids = items + .iter() + .filter(|item| { + item.get("type").and_then(|value| value.as_str()) == Some("function_call") + }) + .filter_map(response_item_call_id) + .collect::>(); + + let restore_group = lookup + .previous + .as_ref() + .map(|previous| { + previous + .call_order + .iter() + .filter(|call_id| { + output_call_ids.contains(*call_id) && !existing_call_ids.contains(*call_id) + }) + .filter_map(|call_id| { + previous + .calls_by_id + .get(call_id) + .cloned() + .map(|item| (call_id.clone(), item)) + }) + .collect::>() + }) + .unwrap_or_default(); + + let restore_group_ids = restore_group + .iter() + .map(|(call_id, _)| call_id.clone()) + .collect::>(); + let mut restore_group = Some(restore_group); + let mut seen_call_ids = HashSet::new(); + let mut restored = 0usize; + let mut enriched = 0usize; + let mut new_items = Vec::new(); + + for mut item in items { + match item.get("type").and_then(|value| value.as_str()) { + Some("function_call") => { + if let Some(call_id) = response_item_call_id(&item) { + if let Some(cached) = lookup.call(&call_id) { + if enrich_function_call_reasoning(&mut item, cached) { + enriched += 1; + } + } + seen_call_ids.insert(call_id); + } + new_items.push(item); + } + Some("function_call_output") => { + if let Some(group) = restore_group.take().filter(|group| !group.is_empty()) { + for (call_id, cached_item) in group { + seen_call_ids.insert(call_id); + new_items.push(cached_item); + restored += 1; + } + } + + if let Some(call_id) = response_item_call_id(&item) { + if !seen_call_ids.contains(&call_id) + && !restore_group_ids.contains(&call_id) + { + if let Some(cached) = lookup.call(&call_id).cloned() { + seen_call_ids.insert(call_id); + new_items.push(cached); + restored += 1; + } + } + } + new_items.push(item); + } + _ => new_items.push(item), + } + } + + let changed = restored + enriched; + if changed == 0 && original_was_object && new_items.len() == 1 { + *input = new_items.into_iter().next().unwrap_or(Value::Null); + } else { + *input = Value::Array(new_items); + } + changed + } + + async fn lookup(&self, previous_response_id: Option<&str>) -> CachedLookup { + let inner = self.inner.read().await; + CachedLookup { + previous: previous_response_id.and_then(|id| inner.responses.get(id).cloned()), + } + } +} + +impl CodexChatHistoryInner { + fn insert_calls(&mut self, response_id: &str, calls: Vec<(String, Value)>) -> usize { + if !self.responses.contains_key(response_id) { + self.response_order.push_back(response_id.to_string()); + } + + let cached_response = self.responses.entry(response_id.to_string()).or_default(); + let mut inserted_or_updated = 0usize; + for (call_id, item) in calls { + if !cached_response.calls_by_id.contains_key(&call_id) { + cached_response.call_order.push(call_id.clone()); + } + cached_response + .calls_by_id + .insert(call_id.clone(), item.clone()); + inserted_or_updated += 1; + } + + self.prune(); + inserted_or_updated + } + + fn prune(&mut self) { + while self.response_order.len() > MAX_CACHED_RESPONSES { + let Some(response_id) = self.response_order.pop_front() else { + break; + }; + self.responses.remove(&response_id); + } + } +} + +impl CachedLookup { + fn call(&self, call_id: &str) -> Option<&Value> { + self.previous + .as_ref() + .and_then(|previous| previous.calls_by_id.get(call_id)) + } +} + +pub fn record_responses_sse_stream( + stream: impl Stream> + Send + 'static, + history: Arc, +) -> impl Stream> + Send { + async_stream::stream! { + let mut buffer = String::new(); + let mut utf8_remainder = Vec::new(); + let mut current_response_id: Option = None; + + tokio::pin!(stream); + + while let Some(chunk) = stream.next().await { + match chunk { + Ok(bytes) => { + append_utf8_safe(&mut buffer, &mut utf8_remainder, &bytes); + while let Some(block) = take_sse_block(&mut buffer) { + inspect_sse_block(&block, &mut current_response_id, history.as_ref()).await; + } + yield Ok(bytes); + } + Err(err) => yield Err(err), + } + } + } +} + +async fn inspect_sse_block( + block: &str, + current_response_id: &mut Option, + history: &CodexChatHistoryStore, +) { + if block.trim().is_empty() { + return; + } + + let mut data_parts = Vec::new(); + for line in block.lines() { + if let Some(data) = strip_sse_field(line, "data") { + data_parts.push(data.to_string()); + } + } + + let data = data_parts.join("\n"); + if data.trim().is_empty() || data.trim() == "[DONE]" { + return; + } + + let Ok(value) = serde_json::from_str::(&data) else { + return; + }; + + if let Some(response_id) = value + .pointer("/response/id") + .and_then(|value| value.as_str()) + .filter(|value| !value.is_empty()) + { + *current_response_id = Some(response_id.to_string()); + } + + match value.get("type").and_then(|value| value.as_str()) { + Some("response.output_item.done") => { + if let Some(item) = value.get("item") { + history + .record_function_call(current_response_id.as_deref(), item) + .await; + } + } + Some("response.completed") => { + if let Some(response) = value.get("response") { + history.record_response(response).await; + } + } + _ => {} + } +} + +fn cached_function_call(item: &Value) -> Option<(String, Value)> { + if item.get("type").and_then(|value| value.as_str()) != Some("function_call") { + return None; + } + let call_id = response_item_call_id(item)?; + Some((call_id, item.clone())) +} + +fn enrich_function_call_reasoning(item: &mut Value, cached: &Value) -> bool { + let mut changed = false; + for key in ["reasoning_content", "reasoning"] { + if item.get(key).is_some_and(|value| !is_empty_value(value)) { + continue; + } + let Some(reasoning) = cached.get(key).filter(|value| !is_empty_value(value)) else { + continue; + }; + if let Some(object) = item.as_object_mut() { + object.insert(key.to_string(), reasoning.clone()); + changed = true; + } + } + changed +} + +#[cfg(test)] +mod tests { + use super::*; + use futures::StreamExt; + use serde_json::json; + + #[tokio::test] + async fn enriches_tool_output_with_cached_function_call_from_previous_response() { + let history = CodexChatHistoryStore::default(); + history + .record_response(&json!({ + "id": "resp_1", + "output": [ + { + "type": "function_call", + "call_id": "call_1", + "name": "read_file", + "arguments": "{\"path\":\"README.md\"}", + "reasoning_content": "Need to inspect the file." + } + ] + })) + .await; + + let mut request = json!({ + "previous_response_id": "resp_1", + "input": [ + { + "type": "function_call_output", + "call_id": "call_1", + "output": "ok" + } + ] + }); + + assert_eq!(history.enrich_request(&mut request).await, 1); + let input = request["input"].as_array().unwrap(); + assert_eq!(input[0]["type"], "function_call"); + assert_eq!(input[0]["reasoning_content"], "Need to inspect the file."); + assert_eq!(input[1]["type"], "function_call_output"); + } + + #[tokio::test] + async fn does_not_restore_without_matching_previous_response() { + let history = CodexChatHistoryStore::default(); + history + .record_response(&json!({ + "id": "resp_1", + "output": [ + { + "type": "function_call", + "call_id": "call_1", + "name": "read_file", + "arguments": "{}", + "reasoning_content": "This belongs to another response." + } + ] + })) + .await; + + let mut missing_previous = json!({ + "input": [ + { + "type": "function_call_output", + "call_id": "call_1", + "output": "ok" + } + ] + }); + assert_eq!(history.enrich_request(&mut missing_previous).await, 0); + assert_eq!(missing_previous["input"][0]["type"], "function_call_output"); + + let mut different_previous = json!({ + "previous_response_id": "resp_2", + "input": [ + { + "type": "function_call_output", + "call_id": "call_1", + "output": "ok" + } + ] + }); + assert_eq!(history.enrich_request(&mut different_previous).await, 0); + assert_eq!( + different_previous["input"][0]["type"], + "function_call_output" + ); + } + + #[tokio::test] + async fn enriches_existing_function_call_missing_reasoning() { + let history = CodexChatHistoryStore::default(); + history + .record_response(&json!({ + "id": "resp_1", + "output": [ + { + "type": "function_call", + "call_id": "call_1", + "name": "read_file", + "arguments": "{}", + "reasoning_content": "Need to inspect the file." + } + ] + })) + .await; + + let mut request = json!({ + "previous_response_id": "resp_1", + "input": [ + { + "type": "function_call", + "call_id": "call_1", + "name": "read_file", + "arguments": "{}" + }, + { + "type": "function_call_output", + "call_id": "call_1", + "output": "ok" + } + ] + }); + + assert_eq!(history.enrich_request(&mut request).await, 1); + let input = request["input"].as_array().unwrap(); + assert_eq!(input[0]["reasoning_content"], "Need to inspect the file."); + assert_eq!(input.len(), 2); + } + + #[tokio::test] + async fn restores_parallel_tool_calls_as_one_assistant_group() { + let history = CodexChatHistoryStore::default(); + history + .record_response(&json!({ + "id": "resp_1", + "output": [ + { + "type": "function_call", + "call_id": "call_1", + "name": "first", + "arguments": "{}", + "reasoning_content": "Need both tools." + }, + { + "type": "function_call", + "call_id": "call_2", + "name": "second", + "arguments": "{}", + "reasoning_content": "Need both tools." + } + ] + })) + .await; + + let mut request = json!({ + "previous_response_id": "resp_1", + "input": [ + { + "type": "function_call_output", + "call_id": "call_1", + "output": "one" + }, + { + "type": "function_call_output", + "call_id": "call_2", + "output": "two" + } + ] + }); + + assert_eq!(history.enrich_request(&mut request).await, 2); + let input = request["input"].as_array().unwrap(); + assert_eq!(input[0]["type"], "function_call"); + assert_eq!(input[0]["call_id"], "call_1"); + assert_eq!(input[1]["type"], "function_call"); + assert_eq!(input[1]["call_id"], "call_2"); + assert_eq!(input[2]["type"], "function_call_output"); + assert_eq!(input[3]["type"], "function_call_output"); + } + + #[tokio::test] + async fn records_streamed_function_call_done_items() { + let history = Arc::new(CodexChatHistoryStore::default()); + let stream = futures::stream::iter(vec![ + Ok::<_, std::io::Error>(Bytes::from_static( + b"event: response.created\ndata: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_stream\"}}\n\n", + )), + Ok(Bytes::from_static( + b"event: response.output_item.done\ndata: {\"type\":\"response.output_item.done\",\"item\":{\"type\":\"function_call\",\"call_id\":\"call_1\",\"name\":\"read_file\",\"arguments\":\"{}\",\"reasoning_content\":\"Need a file.\"}}\n\n", + )), + ]); + + let output = record_responses_sse_stream(stream, history.clone()) + .collect::>() + .await; + assert_eq!(output.len(), 2); + + let mut request = json!({ + "previous_response_id": "resp_stream", + "input": [ + { + "type": "function_call_output", + "call_id": "call_1", + "output": "ok" + } + ] + }); + + assert_eq!(history.enrich_request(&mut request).await, 1); + assert_eq!(request["input"][0]["reasoning_content"], "Need a file."); + } +} diff --git a/src-tauri/src/proxy/providers/mod.rs b/src-tauri/src/proxy/providers/mod.rs index 24300286f..7ab568e5f 100644 --- a/src-tauri/src/proxy/providers/mod.rs +++ b/src-tauri/src/proxy/providers/mod.rs @@ -15,6 +15,8 @@ mod adapter; mod auth; mod claude; mod codex; +pub(crate) mod codex_chat_common; +pub mod codex_chat_history; pub mod codex_oauth_auth; pub mod copilot_auth; pub mod copilot_model_map; @@ -42,8 +44,12 @@ pub use claude::{ claude_api_format_needs_transform, get_claude_api_format, transform_claude_request_for_api_format, ClaudeAdapter, }; -pub use codex::should_convert_codex_responses_to_chat; pub use codex::CodexAdapter; +pub use codex::{ + apply_codex_chat_upstream_model, codex_provider_upstream_model, + codex_provider_uses_chat_completions, should_convert_codex_responses_to_chat, + CODEX_CHAT_CLIENT_MODEL, +}; pub use gemini::GeminiAdapter; /// 供应商类型枚举 diff --git a/src-tauri/src/proxy/providers/streaming_codex_chat.rs b/src-tauri/src/proxy/providers/streaming_codex_chat.rs index 99a2d895f..5f1dbbe81 100644 --- a/src-tauri/src/proxy/providers/streaming_codex_chat.rs +++ b/src-tauri/src/proxy/providers/streaming_codex_chat.rs @@ -1,8 +1,13 @@ //! OpenAI Chat Completions SSE → OpenAI Responses SSE conversion. -use super::transform_codex_chat::{ - chat_usage_to_responses_usage, response_id_from_chat_id, response_status_from_finish_reason, - split_leading_think_block, strip_leading_think_open_tag, +use super::{ + codex_chat_common::{ + extract_reasoning_field_text, response_function_call_item, split_leading_think_block, + strip_leading_think_open_tag, + }, + transform_codex_chat::{ + chat_usage_to_responses_usage, response_id_from_chat_id, response_status_from_finish_reason, + }, }; use crate::proxy::sse::{strip_sse_field, take_sse_block}; use bytes::Bytes; @@ -49,6 +54,7 @@ struct ToolCallState { call_id: String, name: String, arguments: String, + reasoning_content: String, added: bool, done: bool, } @@ -133,9 +139,12 @@ impl ChatToResponsesState { if let Some(tool_calls) = delta.get("tool_calls").and_then(|v| v.as_array()) { events.extend(self.flush_inline_think_at_boundary()); + let reasoning_for_tool_call = self.current_reasoning_text(); events.extend(self.finalize_reasoning()); for tool_call in tool_calls { - events.extend(self.push_tool_call_delta(tool_call)); + events.extend( + self.push_tool_call_delta(tool_call, reasoning_for_tool_call.as_deref()), + ); } } } @@ -375,7 +384,11 @@ impl ChatToResponsesState { events } - fn push_tool_call_delta(&mut self, tool_call: &Value) -> Vec { + fn current_reasoning_text(&self) -> Option { + (!self.reasoning.text.trim().is_empty()).then(|| self.reasoning.text.trim().to_string()) + } + + fn push_tool_call_delta(&mut self, tool_call: &Value, reasoning: Option<&str>) -> Vec { let chat_index = tool_call.get("index").and_then(|v| v.as_u64()).unwrap_or(0) as usize; let id_delta = tool_call .get("id") @@ -408,6 +421,12 @@ impl ChatToResponsesState { if !args_delta.is_empty() { state.arguments.push_str(&args_delta); } + if state.reasoning_content.is_empty() { + if let Some(reasoning) = reasoning.map(str::trim).filter(|value| !value.is_empty()) + { + state.reasoning_content = reasoning.to_string(); + } + } if !state.added && (!state.call_id.is_empty() || !state.name.is_empty()) { should_add = true; @@ -434,19 +453,21 @@ impl ChatToResponsesState { state.item_id = format!("fc_{}", state.call_id); item_id = state.item_id.clone(); + let item = response_function_call_item( + &item_id, + "in_progress", + &state.call_id, + &state.name, + "", + Some(&state.reasoning_content), + ); + events.push(sse_event( "response.output_item.added", json!({ "type": "response.output_item.added", "output_index": assigned, - "item": { - "id": item_id, - "type": "function_call", - "status": "in_progress", - "call_id": state.call_id, - "name": state.name, - "arguments": "" - } + "item": item }), )); @@ -643,19 +664,20 @@ impl ChatToResponsesState { } state.output_index = Some(assigned); state.item_id = format!("fc_{}", state.call_id); + let item = response_function_call_item( + &state.item_id, + "in_progress", + &state.call_id, + &state.name, + "", + Some(&state.reasoning_content), + ); add_event = Some(sse_event( "response.output_item.added", json!({ "type": "response.output_item.added", "output_index": assigned, - "item": { - "id": state.item_id, - "type": "function_call", - "status": "in_progress", - "call_id": state.call_id, - "name": state.name, - "arguments": "" - } + "item": item }), )); } @@ -666,14 +688,14 @@ impl ChatToResponsesState { let state = self.tools.get_mut(&key).expect("tool state exists"); let output_index = state.output_index.unwrap_or(0); - let item = json!({ - "id": state.item_id, - "type": "function_call", - "status": "completed", - "call_id": state.call_id, - "name": state.name, - "arguments": state.arguments - }); + let item = response_function_call_item( + &state.item_id, + "completed", + &state.call_id, + &state.name, + &state.arguments, + Some(&state.reasoning_content), + ); state.done = true; self.output_items.push((output_index, item.clone())); @@ -753,24 +775,7 @@ impl ChatToResponsesState { } fn chat_delta_reasoning_text(delta: &Value) -> Option { - for key in ["reasoning_content", "reasoning"] { - if let Some(text) = delta.get(key).and_then(|v| v.as_str()) { - if !text.is_empty() { - return Some(text.to_string()); - } - } - } - - let reasoning = delta.get("reasoning")?; - for key in ["content", "text", "summary"] { - if let Some(text) = reasoning.get(key).and_then(|v| v.as_str()) { - if !text.is_empty() { - return Some(text.to_string()); - } - } - } - - None + extract_reasoning_field_text(delta) } enum ThinkPrefixDecision { @@ -998,6 +1003,21 @@ mod tests { assert!(output.contains("\"call_id\":\"call_1\"")); } + #[tokio::test] + async fn preserves_reasoning_content_on_streamed_tool_call_items() { + let output = collect(vec![ + "data: {\"id\":\"chatcmpl_tool_reasoning\",\"model\":\"deepseek-v4-flash\",\"choices\":[{\"delta\":{\"reasoning_content\":\"Need file.\"}}]}\n\n", + "data: {\"id\":\"chatcmpl_tool_reasoning\",\"model\":\"deepseek-v4-flash\",\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_1\",\"type\":\"function\",\"function\":{\"name\":\"read_file\"}}]}}]}\n\n", + "data: {\"id\":\"chatcmpl_tool_reasoning\",\"model\":\"deepseek-v4-flash\",\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"{\\\"path\\\":\\\"README.md\\\"}\"}}]},\"finish_reason\":\"tool_calls\"}]}\n\n", + "data: [DONE]\n\n", + ]) + .await; + + assert!(output.contains("event: response.output_item.done")); + assert!(output.contains("\"type\":\"function_call\"")); + assert!(output.contains("\"reasoning_content\":\"Need file.\"")); + } + #[tokio::test] async fn stream_error_emits_failed_without_completed() { let upstream = stream::iter(vec![Err::(std::io::Error::other( diff --git a/src-tauri/src/proxy/providers/transform_codex_chat.rs b/src-tauri/src/proxy/providers/transform_codex_chat.rs index dc0ce9eed..4f05a4d33 100644 --- a/src-tauri/src/proxy/providers/transform_codex_chat.rs +++ b/src-tauri/src/proxy/providers/transform_codex_chat.rs @@ -4,6 +4,10 @@ //! Responses API, while the selected upstream provider only exposes an //! OpenAI-compatible Chat Completions endpoint. +use super::codex_chat_common::{ + append_reasoning_content, extract_reasoning_field_text, extract_reasoning_summary_text, + response_function_call_item, split_leading_think_block, +}; use crate::proxy::{error::ProxyError, json_canonical::canonical_json_string}; use serde_json::{json, Value}; @@ -23,9 +27,6 @@ const EXTRA_CHAT_PASSTHROUGH_FIELDS: &[&str] = &[ "top_logprobs", "user", ]; -const THINK_OPEN_TAG: &str = ""; -const THINK_CLOSE_TAG: &str = ""; - /// Convert an OpenAI Responses request into an OpenAI Chat Completions request. pub fn responses_to_chat_completions(body: Value) -> Result { let mut result = json!({}); @@ -122,6 +123,8 @@ fn append_responses_input_as_chat_messages( messages: &mut Vec, ) -> Result<(), ProxyError> { let mut pending_tool_calls = Vec::new(); + let mut pending_reasoning: Option = None; + let mut last_assistant_index: Option = None; match input { Value::String(text) => { @@ -132,16 +135,33 @@ fn append_responses_input_as_chat_messages( } Value::Array(items) => { for item in items { - append_responses_item_as_chat_message(item, messages, &mut pending_tool_calls)?; + append_responses_item_as_chat_message( + item, + messages, + &mut pending_tool_calls, + &mut pending_reasoning, + &mut last_assistant_index, + )?; } } Value::Object(_) => { - append_responses_item_as_chat_message(input, messages, &mut pending_tool_calls)?; + append_responses_item_as_chat_message( + input, + messages, + &mut pending_tool_calls, + &mut pending_reasoning, + &mut last_assistant_index, + )?; } _ => {} } - flush_pending_tool_calls(messages, &mut pending_tool_calls); + flush_pending_tool_calls( + messages, + &mut pending_tool_calls, + &mut pending_reasoning, + &mut last_assistant_index, + ); Ok(()) } @@ -149,14 +169,22 @@ fn append_responses_item_as_chat_message( item: &Value, messages: &mut Vec, pending_tool_calls: &mut Vec, + pending_reasoning: &mut Option, + last_assistant_index: &mut Option, ) -> Result<(), ProxyError> { let item_type = item.get("type").and_then(|v| v.as_str()); match item_type { Some("function_call") => { + append_unique_pending_reasoning(pending_reasoning, responses_item_reasoning_text(item)); pending_tool_calls.push(responses_function_call_to_chat_tool_call(item)); } Some("function_call_output") => { - flush_pending_tool_calls(messages, pending_tool_calls); + flush_pending_tool_calls( + messages, + pending_tool_calls, + pending_reasoning, + last_assistant_index, + ); let call_id = item.get("call_id").and_then(|v| v.as_str()).unwrap_or(""); let output = match item.get("output") { Some(Value::String(s)) => s.clone(), @@ -170,19 +198,37 @@ fn append_responses_item_as_chat_message( })); } Some("reasoning") => { - // Reasoning items are Responses-specific context. Chat-only providers - // cannot consume encrypted reasoning state, so omit it. + let reasoning = responses_reasoning_item_text(item); + let attached_to_previous = pending_tool_calls.is_empty() + && attach_reasoning_to_last_assistant(messages, *last_assistant_index, &reasoning); + if !attached_to_previous { + append_pending_reasoning(pending_reasoning, reasoning); + } } Some("message") | None => { - flush_pending_tool_calls(messages, pending_tool_calls); + flush_pending_tool_calls( + messages, + pending_tool_calls, + pending_reasoning, + last_assistant_index, + ); if item.get("role").is_some() || item.get("content").is_some() { - messages.push(responses_message_item_to_chat_message(item)); + let message = responses_message_item_to_chat_message(item, pending_reasoning); + update_last_assistant_index(messages, &message, last_assistant_index); + messages.push(message); } } _ => { - flush_pending_tool_calls(messages, pending_tool_calls); + flush_pending_tool_calls( + messages, + pending_tool_calls, + pending_reasoning, + last_assistant_index, + ); if item.get("role").is_some() || item.get("content").is_some() { - messages.push(responses_message_item_to_chat_message(item)); + let message = responses_message_item_to_chat_message(item, pending_reasoning); + update_last_assistant_index(messages, &message, last_assistant_index); + messages.push(message); } } } @@ -190,29 +236,178 @@ fn append_responses_item_as_chat_message( Ok(()) } -fn flush_pending_tool_calls(messages: &mut Vec, pending_tool_calls: &mut Vec) { +fn flush_pending_tool_calls( + messages: &mut Vec, + pending_tool_calls: &mut Vec, + pending_reasoning: &mut Option, + last_assistant_index: &mut Option, +) { if pending_tool_calls.is_empty() { return; } - messages.push(json!({ + let mut message = json!({ "role": "assistant", "content": null, "tool_calls": std::mem::take(pending_tool_calls) - })); + }); + attach_pending_reasoning_to_assistant(&mut message, pending_reasoning); + *last_assistant_index = Some(messages.len()); + messages.push(message); } -fn responses_message_item_to_chat_message(item: &Value) -> Value { +fn responses_message_item_to_chat_message( + item: &Value, + pending_reasoning: &mut Option, +) -> Value { let role = item.get("role").and_then(|v| v.as_str()).unwrap_or("user"); + let chat_role = responses_role_to_chat_role(role); let content = item .get("content") - .map(|value| responses_content_to_chat_content(role, value)) + .map(|value| responses_content_to_chat_content(chat_role, value)) .unwrap_or(Value::Null); - json!({ - "role": role, + let mut message = json!({ + "role": chat_role, "content": content - }) + }); + + if chat_role == "assistant" { + append_pending_reasoning(pending_reasoning, responses_message_reasoning_text(item)); + attach_pending_reasoning_to_assistant(&mut message, pending_reasoning); + } else if pending_reasoning.is_some() { + pending_reasoning.take(); + } + + message +} + +fn responses_role_to_chat_role(role: &str) -> &'static str { + match role { + "system" | "developer" => "system", + "assistant" => "assistant", + "tool" => "tool", + "user" | "latest_reminder" => "user", + _ => "user", + } +} + +fn update_last_assistant_index( + messages: &[Value], + message: &Value, + last_assistant_index: &mut Option, +) { + match message.get("role").and_then(|v| v.as_str()) { + Some("assistant") => { + *last_assistant_index = Some(messages.len()); + } + Some("tool") => {} + _ => { + *last_assistant_index = None; + } + } +} + +fn append_pending_reasoning(pending_reasoning: &mut Option, reasoning: Option) { + let Some(reasoning) = reasoning else { + return; + }; + let reasoning = reasoning.trim(); + if reasoning.is_empty() { + return; + } + + match pending_reasoning { + Some(existing) if !existing.is_empty() => { + existing.push_str("\n\n"); + existing.push_str(reasoning); + } + _ => { + *pending_reasoning = Some(reasoning.to_string()); + } + } +} + +fn append_unique_pending_reasoning( + pending_reasoning: &mut Option, + reasoning: Option, +) { + let Some(reasoning) = reasoning else { + return; + }; + let reasoning = reasoning.trim(); + if reasoning.is_empty() { + return; + } + + match pending_reasoning { + Some(existing) if existing.contains(reasoning) => {} + Some(existing) if !existing.is_empty() => { + existing.push_str("\n\n"); + existing.push_str(reasoning); + } + _ => { + *pending_reasoning = Some(reasoning.to_string()); + } + } +} + +fn attach_pending_reasoning_to_assistant( + message: &mut Value, + pending_reasoning: &mut Option, +) { + let Some(reasoning) = pending_reasoning.take() else { + return; + }; + if reasoning.trim().is_empty() { + return; + } + + if let Some(obj) = message.as_object_mut() { + append_reasoning_content(obj, &reasoning); + } +} + +fn attach_reasoning_to_last_assistant( + messages: &mut [Value], + last_assistant_index: Option, + reasoning: &Option, +) -> bool { + let Some(reasoning) = reasoning + .as_deref() + .map(str::trim) + .filter(|s| !s.is_empty()) + else { + return true; + }; + let Some(index) = last_assistant_index else { + return false; + }; + let Some(message) = messages.get_mut(index) else { + return false; + }; + if message.get("role").and_then(|v| v.as_str()) != Some("assistant") { + return false; + } + + if let Some(obj) = message.as_object_mut() { + append_reasoning_content(obj, reasoning); + return true; + } + + false +} + +fn responses_message_reasoning_text(item: &Value) -> Option { + responses_item_reasoning_text(item) +} + +fn responses_item_reasoning_text(item: &Value) -> Option { + extract_reasoning_field_text(item) +} + +fn responses_reasoning_item_text(item: &Value) -> Option { + extract_reasoning_summary_text(item) } fn responses_content_to_chat_content(_role: &str, content: &Value) -> Value { @@ -372,14 +567,20 @@ pub fn chat_completion_to_response(body: Value) -> Result { let created_at = body.get("created").and_then(|v| v.as_u64()).unwrap_or(0); let finish_reason = choice.get("finish_reason").and_then(|v| v.as_str()); + let reasoning = chat_reasoning_text(message); let mut output = Vec::new(); - if let Some(reasoning_item) = chat_reasoning_to_response_output_item(message, &response_id) { + if let Some(reasoning_item) = + chat_reasoning_to_response_output_item(reasoning.as_deref(), &response_id) + { output.push(reasoning_item); } if let Some(message_item) = chat_message_to_response_output_item(message, &response_id) { output.push(message_item); } - output.extend(chat_tool_calls_to_response_output_items(message)); + output.extend(chat_tool_calls_to_response_output_items( + message, + reasoning.as_deref(), + )); let mut response = json!({ "id": response_id, @@ -398,8 +599,11 @@ pub fn chat_completion_to_response(body: Value) -> Result { Ok(response) } -fn chat_reasoning_to_response_output_item(message: &Value, response_id: &str) -> Option { - let reasoning = chat_reasoning_text(message)?; +fn chat_reasoning_to_response_output_item( + reasoning: Option<&str>, + response_id: &str, +) -> Option { + let reasoning = reasoning?; if reasoning.is_empty() { return None; } @@ -415,22 +619,8 @@ fn chat_reasoning_to_response_output_item(message: &Value, response_id: &str) -> } fn chat_reasoning_text(message: &Value) -> Option { - for key in ["reasoning_content", "reasoning"] { - if let Some(text) = message.get(key).and_then(|v| v.as_str()) { - if !text.is_empty() { - return Some(text.to_string()); - } - } - } - - if let Some(reasoning) = message.get("reasoning") { - for key in ["content", "text", "summary"] { - if let Some(text) = reasoning.get(key).and_then(|v| v.as_str()) { - if !text.is_empty() { - return Some(text.to_string()); - } - } - } + if let Some(reasoning) = extract_reasoning_field_text(message) { + return Some(reasoning); } if let Some(content) = message.get("content").and_then(|v| v.as_str()) { @@ -510,51 +700,31 @@ fn chat_message_to_response_output_item(message: &Value, response_id: &str) -> O })) } -pub(crate) fn split_leading_think_block(text: &str) -> Option<(String, String)> { - let leading_ws_len = text.len() - text.trim_start().len(); - let after_ws = &text[leading_ws_len..]; - if !after_ws.starts_with(THINK_OPEN_TAG) { - return None; - } - - let body_start = leading_ws_len + THINK_OPEN_TAG.len(); - let close_relative = text[body_start..].find(THINK_CLOSE_TAG)?; - let close_start = body_start + close_relative; - let answer_start = close_start + THINK_CLOSE_TAG.len(); - - Some(( - text[body_start..close_start].trim().to_string(), - strip_think_answer_separator(&text[answer_start..]).to_string(), - )) -} - -pub(crate) fn strip_leading_think_open_tag(text: &str) -> Option { - let leading_ws_len = text.len() - text.trim_start().len(); - let after_ws = &text[leading_ws_len..]; - after_ws - .strip_prefix(THINK_OPEN_TAG) - .map(|value| value.trim().to_string()) -} - -fn strip_think_answer_separator(text: &str) -> &str { - text.trim_start_matches(['\r', '\n', '\t', ' ']) -} - -fn chat_tool_calls_to_response_output_items(message: &Value) -> Vec { +fn chat_tool_calls_to_response_output_items( + message: &Value, + reasoning: Option<&str>, +) -> Vec { let mut output = Vec::new(); if let Some(tool_calls) = message.get("tool_calls").and_then(|v| v.as_array()) { for (index, tool_call) in tool_calls.iter().enumerate() { - output.push(chat_tool_call_to_response_item(tool_call, index)); + output.push(chat_tool_call_to_response_item(tool_call, index, reasoning)); } } else if let Some(function_call) = message.get("function_call") { - output.push(chat_legacy_function_call_to_response_item(function_call)); + output.push(chat_legacy_function_call_to_response_item( + function_call, + reasoning, + )); } output } -fn chat_tool_call_to_response_item(tool_call: &Value, index: usize) -> Value { +fn chat_tool_call_to_response_item( + tool_call: &Value, + index: usize, + reasoning: Option<&str>, +) -> Value { let call_id = tool_call .get("id") .and_then(|v| v.as_str()) @@ -569,17 +739,14 @@ fn chat_tool_call_to_response_item(tool_call: &Value, index: usize) -> Value { None => "{}".to_string(), }; - json!({ - "id": format!("fc_{call_id}"), - "type": "function_call", - "status": "completed", - "call_id": call_id, - "name": name, - "arguments": arguments - }) + let item_id = format!("fc_{call_id}"); + response_function_call_item(&item_id, "completed", &call_id, name, &arguments, reasoning) } -fn chat_legacy_function_call_to_response_item(function_call: &Value) -> Value { +fn chat_legacy_function_call_to_response_item( + function_call: &Value, + reasoning: Option<&str>, +) -> Value { let call_id = function_call .get("id") .and_then(|v| v.as_str()) @@ -595,14 +762,8 @@ fn chat_legacy_function_call_to_response_item(function_call: &Value) -> Value { None => "{}".to_string(), }; - json!({ - "id": format!("fc_{call_id}"), - "type": "function_call", - "status": "completed", - "call_id": call_id, - "name": name, - "arguments": arguments - }) + let item_id = format!("fc_{call_id}"); + response_function_call_item(&item_id, "completed", call_id, name, &arguments, reasoning) } pub(crate) fn chat_usage_to_responses_usage(usage: Option<&Value>) -> Value { @@ -734,6 +895,236 @@ mod tests { assert_eq!(result["reasoning_effort"], "high"); } + #[test] + fn responses_request_to_chat_normalizes_codex_internal_roles() { + let input = json!({ + "model": "gpt-5.4", + "input": [ + { + "type": "message", + "role": "developer", + "content": [ + {"type": "input_text", "text": "Follow project instructions."} + ] + }, + { + "type": "message", + "role": "latest_reminder", + "content": "Keep the reply brief." + }, + { + "type": "message", + "role": "unknown_codex_role", + "content": "Fallback content." + } + ] + }); + + let result = responses_to_chat_completions(input).unwrap(); + let messages = result["messages"].as_array().unwrap(); + + assert_eq!(messages[0]["role"], "system"); + assert_eq!(messages[0]["content"], "Follow project instructions."); + assert_eq!(messages[1]["role"], "user"); + assert_eq!(messages[1]["content"], "Keep the reply brief."); + assert_eq!(messages[2]["role"], "user"); + assert_eq!(messages[2]["content"], "Fallback content."); + } + + #[test] + fn responses_request_to_chat_passes_reasoning_content_back_to_assistant_message() { + let input = json!({ + "model": "gpt-5.4", + "input": [ + { + "type": "reasoning", + "summary": [ + {"type": "summary_text", "text": "Need to inspect the repo."} + ] + }, + { + "type": "message", + "role": "assistant", + "content": [ + {"type": "output_text", "text": "I will check the files."} + ] + }, + { + "type": "message", + "role": "user", + "content": "Continue" + } + ] + }); + + let result = responses_to_chat_completions(input).unwrap(); + let messages = result["messages"].as_array().unwrap(); + + assert_eq!(messages[0]["role"], "assistant"); + assert_eq!(messages[0]["content"], "I will check the files."); + assert_eq!( + messages[0]["reasoning_content"], + "Need to inspect the repo." + ); + assert_eq!(messages[1]["role"], "user"); + assert!(messages[1].get("reasoning_content").is_none()); + } + + #[test] + fn responses_request_to_chat_attaches_trailing_reasoning_to_previous_assistant() { + let input = json!({ + "model": "gpt-5.4", + "input": [ + { + "type": "message", + "role": "assistant", + "content": "I checked the files." + }, + { + "type": "reasoning", + "summary": [ + {"type": "summary_text", "text": "The answer came from README."} + ] + }, + { + "type": "message", + "role": "user", + "content": "Continue" + } + ] + }); + + let result = responses_to_chat_completions(input).unwrap(); + let messages = result["messages"].as_array().unwrap(); + + assert_eq!(messages[0]["role"], "assistant"); + assert_eq!(messages[0]["content"], "I checked the files."); + assert_eq!( + messages[0]["reasoning_content"], + "The answer came from README." + ); + assert_eq!(messages[1]["role"], "user"); + assert!(messages[1].get("reasoning_content").is_none()); + } + + #[test] + fn responses_request_to_chat_keeps_embedded_assistant_reasoning() { + let input = json!({ + "model": "gpt-5.4", + "input": [ + { + "type": "message", + "role": "assistant", + "reasoning_content": "I need to preserve thinking history.", + "content": "Done." + } + ] + }); + + let result = responses_to_chat_completions(input).unwrap(); + let messages = result["messages"].as_array().unwrap(); + + assert_eq!(messages[0]["role"], "assistant"); + assert_eq!(messages[0]["content"], "Done."); + assert_eq!( + messages[0]["reasoning_content"], + "I need to preserve thinking history." + ); + } + + #[test] + fn responses_request_to_chat_attaches_reasoning_to_tool_call_message() { + let input = json!({ + "model": "gpt-5.4", + "input": [ + { + "type": "reasoning", + "summary": "Need to read a file." + }, + { + "type": "function_call", + "call_id": "call_1", + "name": "read_file", + "arguments": "{\"path\":\"README.md\"}" + }, + { + "type": "function_call_output", + "call_id": "call_1", + "output": "Readme content" + } + ] + }); + + let result = responses_to_chat_completions(input).unwrap(); + let messages = result["messages"].as_array().unwrap(); + + assert_eq!(messages[0]["role"], "assistant"); + assert_eq!(messages[0]["reasoning_content"], "Need to read a file."); + assert_eq!(messages[0]["tool_calls"][0]["id"], "call_1"); + assert_eq!(messages[1]["role"], "tool"); + } + + #[test] + fn responses_request_to_chat_recovers_reasoning_from_function_call_item() { + let input = json!({ + "model": "gpt-5.4", + "input": [ + { + "type": "function_call", + "call_id": "call_1", + "name": "read_file", + "arguments": "{\"path\":\"README.md\"}", + "reasoning_content": "Need to read a file." + }, + { + "type": "function_call_output", + "call_id": "call_1", + "output": "Readme content" + } + ] + }); + + let result = responses_to_chat_completions(input).unwrap(); + let messages = result["messages"].as_array().unwrap(); + + assert_eq!(messages[0]["role"], "assistant"); + assert_eq!(messages[0]["tool_calls"][0]["id"], "call_1"); + assert_eq!(messages[0]["reasoning_content"], "Need to read a file."); + assert_eq!(messages[1]["role"], "tool"); + } + + #[test] + fn responses_request_to_chat_attaches_trailing_reasoning_to_tool_call_message() { + let input = json!({ + "model": "gpt-5.4", + "input": [ + { + "type": "function_call", + "call_id": "call_1", + "name": "read_file", + "arguments": "{\"path\":\"README.md\"}" + }, + { + "type": "function_call_output", + "call_id": "call_1", + "output": "Readme content" + }, + { + "type": "reasoning", + "summary": "Need to read a file." + } + ] + }); + + let result = responses_to_chat_completions(input).unwrap(); + let messages = result["messages"].as_array().unwrap(); + + assert_eq!(messages[0]["role"], "assistant"); + assert_eq!(messages[0]["tool_calls"][0]["id"], "call_1"); + assert_eq!(messages[0]["reasoning_content"], "Need to read a file."); + assert_eq!(messages[1]["role"], "tool"); + } + #[test] fn responses_request_to_chat_keeps_multiple_tool_calls_adjacent_to_outputs() { let input = json!({ @@ -827,6 +1218,10 @@ mod tests { assert_eq!(result["output"][1]["content"][0]["text"], "Let me check."); assert_eq!(result["output"][2]["type"], "function_call"); assert_eq!(result["output"][2]["call_id"], "call_1"); + assert_eq!( + result["output"][2]["reasoning_content"], + "I should check the weather before answering." + ); assert_eq!(result["usage"]["input_tokens"], 10); assert_eq!(result["usage"]["output_tokens"], 5); assert_eq!(result["usage"]["input_tokens_details"]["cached_tokens"], 3); diff --git a/src-tauri/src/proxy/response_processor.rs b/src-tauri/src/proxy/response_processor.rs index c20ff98a4..6ac5e80fa 100644 --- a/src-tauri/src/proxy/response_processor.rs +++ b/src-tauri/src/proxy/response_processor.rs @@ -818,7 +818,9 @@ mod tests { use crate::provider::ProviderMeta; use crate::proxy::failover_switch::FailoverSwitchManager; use crate::proxy::provider_router::ProviderRouter; - use crate::proxy::providers::gemini_shadow::GeminiShadowStore; + use crate::proxy::providers::{ + codex_chat_history::CodexChatHistoryStore, gemini_shadow::GeminiShadowStore, + }; use crate::proxy::types::{ProxyConfig, ProxyStatus}; use rust_decimal::Decimal; use std::collections::HashMap; @@ -940,6 +942,7 @@ mod tests { current_providers: Arc::new(RwLock::new(HashMap::new())), provider_router: Arc::new(ProviderRouter::new(db.clone())), gemini_shadow: Arc::new(GeminiShadowStore::default()), + codex_chat_history: Arc::new(CodexChatHistoryStore::default()), app_handle: None, failover_manager: Arc::new(FailoverSwitchManager::new(db)), } diff --git a/src-tauri/src/proxy/server.rs b/src-tauri/src/proxy/server.rs index 2bc9fe2b9..b4c551e35 100644 --- a/src-tauri/src/proxy/server.rs +++ b/src-tauri/src/proxy/server.rs @@ -9,8 +9,12 @@ //! a direct (non-proxied) CLI request. use super::{ - failover_switch::FailoverSwitchManager, handlers, log_codes::srv as log_srv, - provider_router::ProviderRouter, providers::gemini_shadow::GeminiShadowStore, types::*, + failover_switch::FailoverSwitchManager, + handlers, + log_codes::srv as log_srv, + provider_router::ProviderRouter, + providers::{codex_chat_history::CodexChatHistoryStore, gemini_shadow::GeminiShadowStore}, + types::*, ProxyError, }; use crate::database::Database; @@ -38,6 +42,8 @@ pub struct ProxyState { pub provider_router: Arc, /// Gemini Native shadow state，用于 thoughtSignature / tool call 回放 pub gemini_shadow: Arc, + /// Codex Chat bridge history，用于恢复 previous_response_id 指向的 tool call + pub codex_chat_history: Arc, /// AppHandle，用于发射事件和更新托盘菜单 pub app_handle: Option, /// 故障转移切换管理器 @@ -72,6 +78,7 @@ impl ProxyServer { current_providers: Arc::new(RwLock::new(std::collections::HashMap::new())), provider_router, gemini_shadow: Arc::new(GeminiShadowStore::default()), + codex_chat_history: Arc::new(CodexChatHistoryStore::default()), app_handle, failover_manager, }; diff --git a/src-tauri/src/services/proxy.rs b/src-tauri/src/services/proxy.rs index da4d01c84..434e203bb 100644 --- a/src-tauri/src/services/proxy.rs +++ b/src-tauri/src/services/proxy.rs @@ -283,6 +283,48 @@ impl ProxyService { Ok(()) } + pub async fn sync_codex_live_from_provider_while_proxy_active( + &self, + provider: &Provider, + ) -> Result<(), String> { + let mut effective_settings = match self.read_codex_live() { + Ok(config) => config, + Err(_) => build_effective_settings_with_common_config( + self.db.as_ref(), + &AppType::Codex, + provider, + ) + .map_err(|e| format!("构建 codex 有效配置失败: {e}"))?, + }; + let (_, proxy_codex_base_url) = self.build_proxy_urls().await?; + + if let Some(auth) = effective_settings + .get_mut("auth") + .and_then(|v| v.as_object_mut()) + { + auth.insert("OPENAI_API_KEY".to_string(), json!(PROXY_TOKEN_PLACEHOLDER)); + } else if let Some(root) = effective_settings.as_object_mut() { + root.insert( + "auth".to_string(), + json!({ "OPENAI_API_KEY": PROXY_TOKEN_PLACEHOLDER }), + ); + } + + let config_str = effective_settings + .get("config") + .and_then(|v| v.as_str()) + .unwrap_or(""); + let updated_config = Self::apply_codex_proxy_toml_config_for_provider( + config_str, + &proxy_codex_base_url, + Some(provider), + ); + effective_settings["config"] = json!(updated_config); + + self.write_codex_live(&effective_settings)?; + Ok(()) + } + fn get_current_provider_for_app(&self, app_type: &AppType) -> Result, String> { let Some(current_id) = crate::settings::get_effective_current_provider(&self.db, app_type) .map_err(|e| format!("获取 {app_type:?} 当前供应商失败: {e}"))? @@ -1096,8 +1138,15 @@ impl ProxyService { .get("config") .and_then(|v| v.as_str()) .unwrap_or(""); - let updated_config = - Self::apply_codex_proxy_toml_config(config_str, &proxy_codex_base_url); + let codex_provider = self + .get_current_provider_for_app(&AppType::Codex) + .ok() + .flatten(); + let updated_config = Self::apply_codex_proxy_toml_config_for_provider( + config_str, + &proxy_codex_base_url, + codex_provider.as_ref(), + ); live_config["config"] = json!(updated_config); self.write_codex_live(&live_config)?; @@ -1150,8 +1199,15 @@ impl ProxyService { .get("config") .and_then(|v| v.as_str()) .unwrap_or(""); - let updated_config = - Self::apply_codex_proxy_toml_config(config_str, &proxy_codex_base_url); + let codex_provider = self + .get_current_provider_for_app(&AppType::Codex) + .ok() + .flatten(); + let updated_config = Self::apply_codex_proxy_toml_config_for_provider( + config_str, + &proxy_codex_base_url, + codex_provider.as_ref(), + ); live_config["config"] = json!(updated_config); self.write_codex_live(&live_config)?; @@ -1217,8 +1273,15 @@ impl ProxyService { .get("config") .and_then(|v| v.as_str()) .unwrap_or(""); - let updated_config = - Self::apply_codex_proxy_toml_config(config_str, &proxy_codex_base_url); + let codex_provider = self + .get_current_provider_for_app(&AppType::Codex) + .ok() + .flatten(); + let updated_config = Self::apply_codex_proxy_toml_config_for_provider( + config_str, + &proxy_codex_base_url, + codex_provider.as_ref(), + ); live_config["config"] = json!(updated_config); let _ = self.write_codex_live(&live_config); @@ -1748,6 +1811,9 @@ impl ProxyService { if matches!(app_type_enum, AppType::Claude) { self.sync_claude_live_from_provider_while_proxy_active(&provider) .await?; + } else if live_taken_over && matches!(app_type_enum, AppType::Codex) { + self.sync_codex_live_from_provider_while_proxy_active(&provider) + .await?; } } @@ -1854,10 +1920,35 @@ impl ProxyService { /// 接管 Codex 时，本地客户端必须继续以 Responses wire API 访问代理。 /// 真实上游是否走 Chat Completions 由 provider 配置决定，并在代理内部转换。 - fn apply_codex_proxy_toml_config(toml_str: &str, proxy_url: &str) -> String { + fn apply_codex_proxy_toml_config_for_provider( + toml_str: &str, + proxy_url: &str, + provider: Option<&Provider>, + ) -> String { let updated = Self::update_toml_base_url(toml_str, proxy_url); - crate::codex_config::update_codex_toml_field(&updated, "wire_api", "responses") - .unwrap_or(updated) + let mut updated = + crate::codex_config::update_codex_toml_field(&updated, "wire_api", "responses") + .unwrap_or(updated); + + if provider + .map(crate::proxy::providers::codex_provider_uses_chat_completions) + .unwrap_or(false) + { + updated = crate::codex_config::update_codex_toml_field( + &updated, + "model", + crate::proxy::providers::CODEX_CHAT_CLIENT_MODEL, + ) + .unwrap_or(updated); + } else if let Some(upstream_model) = + provider.and_then(crate::proxy::providers::codex_provider_upstream_model) + { + updated = + crate::codex_config::update_codex_toml_field(&updated, "model", &upstream_model) + .unwrap_or(updated); + } + + updated } fn read_claude_live(&self) -> Result { @@ -2308,7 +2399,8 @@ wire_api = "chat" "#; let proxy_url = "http://127.0.0.1:5000/v1"; - let output = ProxyService::apply_codex_proxy_toml_config(input, proxy_url); + let output = + ProxyService::apply_codex_proxy_toml_config_for_provider(input, proxy_url, None); let parsed: toml::Value = toml::from_str(&output).expect("updated config should be valid TOML"); @@ -2327,6 +2419,136 @@ wire_api = "chat" ); } + #[test] + fn apply_codex_proxy_toml_config_uses_safe_client_model_for_chat_provider() { + let input = r#" +model_provider = "deepseek" +model = "deepseek-v4-flash" + +[model_providers.deepseek] +name = "DeepSeek" +base_url = "https://api.deepseek.com/v1" +wire_api = "responses" +"#; + let mut provider = Provider::with_id( + "deepseek".to_string(), + "DeepSeek".to_string(), + json!({ + "config": input + }), + None, + ); + provider.meta = Some(ProviderMeta { + api_format: Some("openai_chat".to_string()), + ..Default::default() + }); + + let proxy_url = "http://127.0.0.1:5000/v1"; + let output = ProxyService::apply_codex_proxy_toml_config_for_provider( + input, + proxy_url, + Some(&provider), + ); + let parsed: toml::Value = + toml::from_str(&output).expect("updated config should be valid TOML"); + + assert_eq!( + parsed.get("model").and_then(|v| v.as_str()), + Some(crate::proxy::providers::CODEX_CHAT_CLIENT_MODEL) + ); + assert_eq!( + parsed + .get("model_providers") + .and_then(|v| v.get("deepseek")) + .and_then(|v| v.get("base_url")) + .and_then(|v| v.as_str()), + Some(proxy_url) + ); + } + + #[test] + fn apply_codex_proxy_toml_config_preserves_model_for_responses_provider() { + let input = r#" +model_provider = "responses" +model = "upstream-responses-model" + +[model_providers.responses] +name = "Responses" +base_url = "https://responses.example/v1" +wire_api = "responses" +"#; + let mut provider = Provider::with_id( + "responses".to_string(), + "Responses".to_string(), + json!({ + "config": input + }), + None, + ); + provider.meta = Some(ProviderMeta { + api_format: Some("openai_responses".to_string()), + ..Default::default() + }); + + let output = ProxyService::apply_codex_proxy_toml_config_for_provider( + input, + "http://127.0.0.1:5000/v1", + Some(&provider), + ); + let parsed: toml::Value = + toml::from_str(&output).expect("updated config should be valid TOML"); + + assert_eq!( + parsed.get("model").and_then(|v| v.as_str()), + Some("upstream-responses-model") + ); + } + + #[test] + fn apply_codex_proxy_toml_config_restores_upstream_model_for_responses_provider() { + let input = r#" +model_provider = "responses" +model = "gpt-5.4" + +[model_providers.responses] +name = "Responses" +base_url = "http://127.0.0.1:5000/v1" +wire_api = "responses" +"#; + let mut provider = Provider::with_id( + "responses".to_string(), + "Responses".to_string(), + json!({ + "config": r#"model_provider = "responses" +model = "upstream-responses-model" + +[model_providers.responses] +name = "Responses" +base_url = "https://responses.example/v1" +wire_api = "responses" +"# + }), + None, + ); + provider.meta = Some(ProviderMeta { + api_format: Some("openai_responses".to_string()), + ..Default::default() + }); + + let output = ProxyService::apply_codex_proxy_toml_config_for_provider( + input, + "http://127.0.0.1:5000/v1", + Some(&provider), + ); + let parsed: toml::Value = + toml::from_str(&output).expect("updated config should be valid TOML"); + + assert_eq!( + parsed.get("model").and_then(|v| v.as_str()), + Some("upstream-responses-model") + ); + } + #[test] fn update_toml_base_url_falls_back_to_top_level_base_url() { let input = r#" @@ -3185,6 +3407,117 @@ requires_openai_auth = true ); } + #[tokio::test] + #[serial] + async fn hot_switch_codex_chat_provider_uses_safe_model_without_changing_live_provider() { + let _home = TempHome::new(); + crate::settings::reload_settings().expect("reload settings"); + + let db = Arc::new(Database::memory().expect("init db")); + let service = ProxyService::new(db.clone()); + + let provider_a = Provider::with_id( + "a".to_string(), + "Responses".to_string(), + json!({ + "auth": { + "OPENAI_API_KEY": "responses-key" + }, + "config": r#"model_provider = "stable" +model = "responses-model" + +[model_providers.stable] +name = "Stable" +base_url = "https://responses.example/v1" +wire_api = "responses" +requires_openai_auth = true +"# + }), + None, + ); + let mut provider_b = Provider::with_id( + "b".to_string(), + "DeepSeek".to_string(), + json!({ + "auth": { + "OPENAI_API_KEY": "deepseek-key" + }, + "config": r#"model_provider = "deepseek" +model = "deepseek-v4-flash" + +[model_providers.deepseek] +name = "DeepSeek" +base_url = "https://api.deepseek.com/v1" +wire_api = "responses" +requires_openai_auth = true +"# + }), + None, + ); + provider_b.meta = Some(ProviderMeta { + api_format: Some("openai_chat".to_string()), + ..Default::default() + }); + + db.save_provider("codex", &provider_a) + .expect("save provider a"); + db.save_provider("codex", &provider_b) + .expect("save provider b"); + db.set_current_provider("codex", "a") + .expect("set current provider"); + crate::settings::set_current_provider(&AppType::Codex, Some("a")) + .expect("set local current provider"); + db.save_live_backup( + "codex", + &serde_json::to_string(&provider_a.settings_config).expect("serialize provider a"), + ) + .await + .expect("seed live backup"); + service + .write_codex_live(&json!({ + "auth": { + "OPENAI_API_KEY": PROXY_TOKEN_PLACEHOLDER + }, + "config": r#"model_provider = "stable" +model = "responses-model" + +[model_providers.stable] +name = "Stable" +base_url = "http://127.0.0.1:15721/v1" +wire_api = "responses" +requires_openai_auth = true +"# + })) + .expect("seed taken-over Codex live config"); + + service + .hot_switch_provider("codex", "b") + .await + .expect("hot switch Codex provider"); + + let live = service.read_codex_live().expect("read Codex live config"); + let live_config = live + .get("config") + .and_then(|v| v.as_str()) + .expect("live config string"); + let parsed_live: toml::Value = toml::from_str(live_config).expect("parse live config"); + + assert_eq!( + parsed_live.get("model_provider").and_then(|v| v.as_str()), + Some("stable") + ); + assert_eq!( + parsed_live.get("model").and_then(|v| v.as_str()), + Some(crate::proxy::providers::CODEX_CHAT_CLIENT_MODEL) + ); + assert_eq!( + live.get("auth") + .and_then(|auth| auth.get("OPENAI_API_KEY")) + .and_then(|v| v.as_str()), + Some(PROXY_TOKEN_PLACEHOLDER) + ); + } + #[tokio::test] #[serial] async fn update_live_backup_from_provider_keeps_new_codex_mcp_entries_on_conflict() { diff --git a/src/components/providers/forms/CodexFormFields.tsx b/src/components/providers/forms/CodexFormFields.tsx index 6690e7cdd..2e470768d 100644 --- a/src/components/providers/forms/CodexFormFields.tsx +++ b/src/components/providers/forms/CodexFormFields.tsx @@ -90,6 +90,7 @@ export function CodexFormFields({ const [fetchedModels, setFetchedModels] = useState([]); const [isFetchingModels, setIsFetchingModels] = useState(false); + const isChatApiFormat = apiFormat === "openai_chat"; const handleFetchModels = useCallback(() => { if (!codexBaseUrl || !codexApiKey) { @@ -202,7 +203,11 @@ export function CodexFormFields({ htmlFor="codexModelName" className="block text-sm font-medium text-foreground" > - {t("codexConfig.modelName", { defaultValue: "模型名称" })} + {isChatApiFormat + ? t("codexConfig.upstreamModelName", { + defaultValue: "上游模型名称", + }) + : t("codexConfig.modelName", { defaultValue: "模型名称" })}