From b9c30ba407f31b68e2e49d18aaaec5d34268e9aa Mon Sep 17 00:00:00 2001 From: Zyy Date: Fri, 12 Jun 2026 11:24:06 +0800 Subject: [PATCH] feat: add openai gpt parity routes Co-authored-by: Codex --- src/app.rs | 7 +- src/codex/client.rs | 32 ++ src/codex/types.rs | 64 ++- src/http/v1.rs | 362 +++++++++++++++- src/translation/codex_to_openai.rs | 523 +++++++++++++++++++++++ src/translation/openai_to_codex.rs | 259 +++++++++++- tests/chat_completions_route_test.rs | 595 +++++++++++++++++++++++++++ tests/codex_client_test.rs | 18 +- tests/codex_websocket_test.rs | 23 +- tests/routes_chat_test.rs | 16 +- tests/v1_upstream_route_test.rs | 245 ++++++++++- 11 files changed, 2079 insertions(+), 65 deletions(-) create mode 100644 tests/chat_completions_route_test.rs diff --git a/src/app.rs b/src/app.rs index 4dbcc5e48..3128262f2 100644 --- a/src/app.rs +++ b/src/app.rs @@ -13,7 +13,10 @@ use crate::{ }, health::health, middleware::attach_request_id, - v1::{debug_models, model_catalog, model_detail, model_info, models, responses}, + v1::{ + chat_completions, debug_models, model_catalog, model_detail, model_info, models, + responses, + }, }, state::AppState, }; @@ -22,7 +25,7 @@ pub fn build_router(state: AppState) -> Router { Router::new() .route("/health", get(health)) .route("/v1/responses", post(responses)) - .route("/v1/chat/completions", post(responses)) + .route("/v1/chat/completions", post(chat_completions)) .route("/v1/models", get(models)) .route("/v1/models/catalog", get(model_catalog)) .route("/v1/models/{model_id}", get(model_detail)) diff --git a/src/codex/client.rs b/src/codex/client.rs index dface1bd9..d5b66b4a7 100644 --- a/src/codex/client.rs +++ b/src/codex/client.rs @@ -39,6 +39,12 @@ pub struct CodexRequestContext<'a> { pub account_id: Option<&'a str>, pub request_id: &'a str, pub turn_state: Option<&'a str>, + pub turn_metadata: Option<&'a str>, + pub beta_features: Option<&'a str>, + pub include_timing_metrics: Option<&'a str>, + pub version: Option<&'a str>, + pub codex_window_id: Option<&'a str>, + pub parent_thread_id: Option<&'a str>, pub cookie_header: Option<&'a str>, } @@ -158,10 +164,36 @@ impl CodexBackendClient { if let Some(cookie_header) = context.cookie_header { headers.insert(COOKIE, HeaderValue::from_str(cookie_header)?); } + insert_optional_header(&mut headers, "x-codex-turn-metadata", context.turn_metadata)?; + insert_optional_header(&mut headers, "x-codex-beta-features", context.beta_features)?; + insert_optional_header( + &mut headers, + "x-responsesapi-include-timing-metrics", + context.include_timing_metrics, + )?; + insert_optional_header(&mut headers, "version", context.version)?; + insert_optional_header(&mut headers, "x-codex-window-id", context.codex_window_id)?; + insert_optional_header( + &mut headers, + "x-codex-parent-thread-id", + context.parent_thread_id, + )?; Ok(headers) } } +fn insert_optional_header( + headers: &mut HeaderMap, + name: &'static str, + value: Option<&str>, +) -> CodexClientResult<()> { + let Some(value) = value else { + return Ok(()); + }; + headers.insert(HeaderName::from_static(name), HeaderValue::from_str(value)?); + Ok(()) +} + pub fn build_reqwest_client(force_http11: bool) -> Result { // Codex Desktop 指纹依赖 reqwest/rustls 组合，升级前必须重新验证 TLS 行为。 let builder = Client::builder() diff --git a/src/codex/types.rs b/src/codex/types.rs index 39cc4ba31..bf9b41a63 100644 --- a/src/codex/types.rs +++ b/src/codex/types.rs @@ -13,11 +13,69 @@ pub struct CodexResponsesRequest { #[serde(skip_serializing_if = "Option::is_none")] pub tools: Option>, #[serde(skip_serializing_if = "Option::is_none")] + pub tool_choice: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub parallel_tool_calls: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub text: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub service_tier: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub previous_response_id: Option, - #[serde(default, skip_serializing_if = "is_false")] + #[serde(skip_serializing_if = "Option::is_none")] + pub prompt_cache_key: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub include: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub client_metadata: Option, + #[serde(skip)] pub use_websocket: bool, + #[serde(skip)] + pub turn_state: Option, + #[serde(skip)] + pub turn_metadata: Option, + #[serde(skip)] + pub beta_features: Option, + #[serde(skip)] + pub version: Option, + #[serde(skip)] + pub include_timing_metrics: Option, + #[serde(skip)] + pub codex_window_id: Option, + #[serde(skip)] + pub parent_thread_id: Option, } -fn is_false(value: &bool) -> bool { - !*value +impl CodexResponsesRequest { + pub fn new_http_sse( + model: impl Into, + instructions: impl Into, + input: Vec, + ) -> Self { + Self { + model: model.into(), + instructions: instructions.into(), + input, + stream: true, + store: false, + reasoning: None, + tools: None, + tool_choice: None, + parallel_tool_calls: None, + text: None, + service_tier: None, + previous_response_id: None, + prompt_cache_key: None, + include: None, + client_metadata: None, + use_websocket: false, + turn_state: None, + turn_metadata: None, + beta_features: None, + version: None, + include_timing_metrics: None, + codex_window_id: None, + parent_thread_id: None, + } + } } diff --git a/src/http/v1.rs b/src/http/v1.rs index ddc932499..02dd3a5db 100644 --- a/src/http/v1.rs +++ b/src/http/v1.rs @@ -39,7 +39,12 @@ use crate::{ logs::event::{EventLevel, EventLog}, models::catalog::ModelCatalog, state::AppState, - translation::codex_to_openai::openai_error, + translation::{ + codex_to_openai::{ + chat_completion_from_codex_sse, chat_completion_stream_from_codex_sse, openai_error, + }, + openai_to_codex::{translate_chat_to_codex, ChatCompletionRequest}, + }, }; const MODEL_CREATED_TIMESTAMP: i64 = 1_700_000_000; @@ -51,8 +56,27 @@ struct ResponsesBody { instructions: Option, reasoning: Option, tools: Option>, + service_tier: Option, + tool_choice: Option, + parallel_tool_calls: Option, + text: Option, + prompt_cache_key: Option, + include: Option>, + client_metadata: Option, previous_response_id: Option, stream: Option, + #[serde(rename = "turnState")] + turn_state: Option, + #[serde(rename = "turnMetadata")] + turn_metadata: Option, + #[serde(rename = "betaFeatures")] + beta_features: Option, + #[serde(rename = "includeTimingMetrics")] + include_timing_metrics: Option, + #[serde(rename = "codexWindowId")] + codex_window_id: Option, + #[serde(rename = "parentThreadId")] + parent_thread_id: Option, } pub async fn responses( @@ -69,7 +93,7 @@ pub async fn responses( let default_model = state.config().model.default_model.clone(); let body = serde_json::from_slice::(&body) .unwrap_or_else(|_| default_body(default_model.clone())); - let client_stream = body.stream.unwrap_or(false); + let client_stream = body.stream.unwrap_or(true); let requested_model = body.model.clone().unwrap_or(default_model); let catalog = ModelCatalog::from_config(&state.config().model); if !catalog.is_recognized_model_name(&requested_model) { @@ -80,17 +104,50 @@ pub async fn responses( .into_response(); } let parsed_model = catalog.parse_model_name(&requested_model); - let codex_request = CodexResponsesRequest { - model: parsed_model.model_id.clone(), - instructions: body.instructions.unwrap_or_default(), - input: body.input.unwrap_or_default(), - stream: true, - store: false, - reasoning: body.reasoning, - tools: body.tools, - previous_response_id: body.previous_response_id, - use_websocket: false, - }; + let mut codex_request = CodexResponsesRequest::new_http_sse( + parsed_model.model_id.clone(), + body.instructions.unwrap_or_default(), + body.input.unwrap_or_default(), + ); + codex_request.reasoning = body.reasoning; + codex_request.tools = body.tools; + codex_request.tool_choice = body.tool_choice; + codex_request.parallel_tool_calls = body.parallel_tool_calls; + codex_request.text = body.text; + codex_request.prompt_cache_key = body.prompt_cache_key; + codex_request.include = body.include; + codex_request.client_metadata = body.client_metadata; + codex_request.previous_response_id = body.previous_response_id; + codex_request.reasoning = responses_reasoning( + codex_request.reasoning.take(), + parsed_model.reasoning_effort.as_deref(), + state.config().model.default_reasoning_effort.as_deref(), + ); + codex_request.service_tier = body + .service_tier + .or(parsed_model.service_tier) + .or_else(|| state.config().model.service_tier.clone()) + .map(normalize_service_tier_for_upstream); + ensure_reasoning_include(&mut codex_request); + codex_request.turn_state = body + .turn_state + .or_else(|| header_string(&headers, "x-codex-turn-state")); + codex_request.turn_metadata = body + .turn_metadata + .or_else(|| header_string(&headers, "x-codex-turn-metadata")); + codex_request.beta_features = body + .beta_features + .or_else(|| header_string(&headers, "x-codex-beta-features")); + codex_request.include_timing_metrics = body + .include_timing_metrics + .or_else(|| header_string(&headers, "x-responsesapi-include-timing-metrics")); + codex_request.version = header_string(&headers, "version"); + codex_request.codex_window_id = body + .codex_window_id + .or_else(|| header_string(&headers, "x-codex-window-id")); + codex_request.parent_thread_id = body + .parent_thread_id + .or_else(|| header_string(&headers, "x-codex-parent-thread-id")); let acquired = { state .account_pool() @@ -237,6 +294,200 @@ pub async fn responses( } } +pub async fn chat_completions( + State(state): State, + Extension(request_id): Extension, + headers: HeaderMap, + body: Bytes, +) -> Response { + let started_at = Instant::now(); + if !authorize_client_api_key(&state, &headers).await { + return missing_client_api_key_response().into_response(); + } + + let Ok(chat_request) = serde_json::from_slice::(&body) else { + return ( + StatusCode::BAD_REQUEST, + Json(openai_error( + "Invalid chat completion request", + "invalid_request", + )), + ) + .into_response(); + }; + let client_stream = chat_request.stream; + let requested_model = chat_request.model.clone(); + let catalog = ModelCatalog::from_config(&state.config().model); + if !catalog.is_recognized_model_name(&requested_model) { + return model_not_found_response().into_response(); + } + let parsed_model = catalog.parse_model_name(&requested_model); + let display_model = ModelCatalog::build_display_model_name(&parsed_model); + let mut codex_request = match translate_chat_to_codex(chat_request) { + Ok(request) => request, + Err(_) => { + return ( + StatusCode::BAD_REQUEST, + Json(openai_error( + "Invalid chat completion request", + "invalid_request", + )), + ) + .into_response(); + } + }; + codex_request.model = parsed_model.model_id.clone(); + if codex_request.reasoning.is_none() { + let effort = parsed_model + .reasoning_effort + .clone() + .or_else(|| state.config().model.default_reasoning_effort.clone()); + if let Some(effort) = effort { + codex_request.reasoning = Some(json!({"effort": effort, "summary": "auto"})); + } + } + if codex_request.service_tier.is_none() { + codex_request.service_tier = parsed_model + .service_tier + .clone() + .or_else(|| state.config().model.service_tier.clone()); + } + codex_request.service_tier = codex_request + .service_tier + .map(normalize_service_tier_for_upstream); + let include_reasoning = codex_request.reasoning.is_some(); + + let acquired = { + state + .account_pool() + .lock() + .await + .acquire_with(AccountAcquireRequest::new(&codex_request.model, Utc::now())) + }; + let Some(acquired) = acquired else { + return no_available_accounts_response().into_response(); + }; + let account = acquired.account; + let log_context = V1LogContext::new( + request_id.as_str(), + &account.id, + &codex_request.model, + client_stream, + started_at, + ); + + let response = send_codex_request_with_refresh_retry( + &state, + &codex_request, + &account, + request_id.as_str(), + ) + .await; + state.account_pool().lock().await.release(&account.id); + + let response = match response { + Ok(response) => response, + Err(error) => { + let error_response = codex_client_error_response(error); + log_v1_response( + &state, + &log_context, + error_response.0, + EventLevel::Error, + "v1 chat completions upstream request failed", + json!({"stream": false}), + ) + .await; + return error_response.into_response(); + } + }; + if persist_upstream_cookies(&state, &account.id, &response.set_cookie_headers) + .await + .is_err() + { + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(openai_error( + "Failed to persist upstream cookies", + "cookie_store_error", + )), + ) + .into_response(); + } + if let Some(usage) = response.usage { + if record_usage(&state, &account.id, usage).await.is_err() { + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(openai_error( + "Failed to record account usage", + "usage_store_error", + )), + ) + .into_response(); + } + } + + if client_stream { + match chat_completion_stream_from_codex_sse( + &response.body, + &display_model, + include_reasoning, + ) { + Ok(Some(body)) => Response::builder() + .status(StatusCode::OK) + .header(CONTENT_TYPE, "text/event-stream") + .header(CACHE_CONTROL, "no-cache") + .body(Body::from(body)) + .unwrap_or_else(|_| { + ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(openai_error( + "Failed to build stream response", + "stream_response_error", + )), + ) + .into_response() + }), + Ok(None) => ( + StatusCode::BAD_GATEWAY, + Json(openai_error( + "Codex response did not include response.completed", + "empty_upstream_response", + )), + ) + .into_response(), + Err(_) => ( + StatusCode::BAD_GATEWAY, + Json(openai_error( + "Invalid Codex SSE response", + "invalid_upstream_sse", + )), + ) + .into_response(), + } + } else { + match chat_completion_from_codex_sse(&response.body, &display_model, include_reasoning) { + Ok(Some(body)) => (StatusCode::OK, Json(body)).into_response(), + Ok(None) => ( + StatusCode::BAD_GATEWAY, + Json(openai_error( + "Codex response did not include response.completed", + "empty_upstream_response", + )), + ) + .into_response(), + Err(_) => ( + StatusCode::BAD_GATEWAY, + Json(openai_error( + "Invalid Codex SSE response", + "invalid_upstream_sse", + )), + ) + .into_response(), + } + } +} + fn no_available_accounts_response() -> (StatusCode, Json) { ( StatusCode::SERVICE_UNAVAILABLE, @@ -254,11 +505,80 @@ fn default_body(default_model: String) -> ResponsesBody { instructions: Some(String::new()), reasoning: None, tools: None, + service_tier: None, + tool_choice: None, + parallel_tool_calls: None, + text: None, + prompt_cache_key: None, + include: None, + client_metadata: None, previous_response_id: None, stream: None, + turn_state: None, + turn_metadata: None, + beta_features: None, + include_timing_metrics: None, + codex_window_id: None, + parent_thread_id: None, } } +fn responses_reasoning( + client_reasoning: Option, + suffix_effort: Option<&str>, + default_effort: Option<&str>, +) -> Option { + let effort = client_reasoning + .as_ref() + .and_then(|reasoning| reasoning.get("effort")) + .and_then(Value::as_str) + .or(suffix_effort) + .or(default_effort); + let summary = client_reasoning + .as_ref() + .and_then(|reasoning| reasoning.get("summary")) + .and_then(Value::as_str) + .unwrap_or("auto"); + if effort.is_none() && client_reasoning.is_none() { + return None; + } + let mut reasoning = json!({"summary": summary}); + if let Some(effort) = effort { + reasoning["effort"] = Value::String(effort.to_string()); + } + Some(reasoning) +} + +fn normalize_service_tier_for_upstream(service_tier: String) -> String { + if service_tier == "fast" { + "priority".to_string() + } else { + service_tier + } +} + +fn ensure_reasoning_include(request: &mut CodexResponsesRequest) { + if request.reasoning.is_none() { + return; + } + let include = request.include.get_or_insert_with(Vec::new); + if include + .iter() + .any(|item| item == "reasoning.encrypted_content") + { + return; + } + include.push("reasoning.encrypted_content".to_string()); +} + +fn header_string(headers: &HeaderMap, name: &str) -> Option { + headers + .get(name) + .and_then(|value| value.to_str().ok()) + .filter(|value| !value.trim().is_empty()) + .map(ToString::to_string) +} + async fn send_codex_request_with_refresh_retry( state: &AppState, request: &CodexResponsesRequest, @@ -301,7 +621,13 @@ async fn send_codex_request( access_token: &account.access_token, account_id: account.account_id.as_deref(), request_id, - turn_state: None, + turn_state: request.turn_state.as_deref(), + turn_metadata: request.turn_metadata.as_deref(), + beta_features: request.beta_features.as_deref(), + include_timing_metrics: request.include_timing_metrics.as_deref(), + version: request.version.as_deref(), + codex_window_id: request.codex_window_id.as_deref(), + parent_thread_id: request.parent_thread_id.as_deref(), cookie_header: cookie_header.as_deref(), }, ) @@ -350,7 +676,13 @@ async fn send_codex_stream_request( access_token: &account.access_token, account_id: account.account_id.as_deref(), request_id, - turn_state: None, + turn_state: request.turn_state.as_deref(), + turn_metadata: request.turn_metadata.as_deref(), + beta_features: request.beta_features.as_deref(), + include_timing_metrics: request.include_timing_metrics.as_deref(), + version: request.version.as_deref(), + codex_window_id: request.codex_window_id.as_deref(), + parent_thread_id: request.parent_thread_id.as_deref(), cookie_header: cookie_header.as_deref(), }, ) diff --git a/src/translation/codex_to_openai.rs b/src/translation/codex_to_openai.rs index 64f5a6a35..3da738f8b 100644 --- a/src/translation/codex_to_openai.rs +++ b/src/translation/codex_to_openai.rs @@ -1,4 +1,10 @@ +use std::collections::{BTreeMap, BTreeSet}; + +use chrono::Utc; use serde_json::{json, Value}; +use uuid::Uuid; + +use crate::codex::sse::{parse_sse_events, SseError}; pub fn openai_error(message: &str, code: &str) -> Value { json!({ @@ -10,3 +16,520 @@ pub fn openai_error(message: &str, code: &str) -> Value { } }) } + +pub fn chat_completion_from_codex_sse( + body: &str, + model: &str, + include_reasoning: bool, +) -> Result, SseError> { + let events = parse_sse_events(body)?; + let mut content = String::new(); + let mut reasoning = String::new(); + let mut function_call_items = BTreeMap::new(); + let mut finished_call_ids = BTreeSet::new(); + let mut tool_calls = Vec::new(); + let mut completed_response = None; + for event in events { + let Ok(value) = serde_json::from_str::(&event.data) else { + continue; + }; + match event.event.as_deref() { + Some("response.output_text.delta") => { + if let Some(delta) = value.get("delta").and_then(Value::as_str) { + content.push_str(delta); + } + } + Some("response.reasoning_summary_text.delta") => { + if include_reasoning { + if let Some(delta) = value.get("delta").and_then(Value::as_str) { + reasoning.push_str(delta); + } + } + } + Some("response.output_item.added") => { + if let Some((item_id, call_id, name)) = function_call_item(value.get("item")) { + function_call_items.insert(item_id, FunctionCallInfo { call_id, name }); + } + } + Some("response.function_call_arguments.done") => { + if let Some(tool_call) = + tool_call_from_done_event(&value, &function_call_items, &mut finished_call_ids) + { + tool_calls.push(tool_call); + } + } + Some("response.output_item.done") => { + if let Some(tool_call) = tool_call_from_output_item( + value.get("item"), + &function_call_items, + &mut finished_call_ids, + ) { + tool_calls.push(tool_call); + } + } + Some("response.completed") => { + completed_response = value.get("response").cloned(); + } + _ => {} + } + } + let Some(response) = completed_response else { + return Ok(None); + }; + if content.is_empty() { + content = output_text_from_response(&response); + } + let mut message = json!({ + "role": "assistant", + "content": if content.is_empty() && !tool_calls.is_empty() { + Value::Null + } else { + Value::String(content) + }, + }); + if include_reasoning && !reasoning.is_empty() { + message["reasoning_content"] = Value::String(reasoning); + } + let has_tool_calls = !tool_calls.is_empty(); + if has_tool_calls { + message["tool_calls"] = Value::Array(tool_calls); + } + + Ok(Some(json!({ + "id": format!("chatcmpl-{}", Uuid::new_v4().simple()), + "object": "chat.completion", + "created": Utc::now().timestamp(), + "model": model, + "choices": [{ + "index": 0, + "message": message, + "finish_reason": if has_tool_calls { "tool_calls" } else { "stop" }, + }], + "usage": openai_usage(response.get("usage")), + }))) +} + +pub fn chat_completion_stream_from_codex_sse( + body: &str, + model: &str, + include_reasoning: bool, +) -> Result, SseError> { + let events = parse_sse_events(body)?; + let chunk_id = format!("chatcmpl-{}", Uuid::new_v4().simple()); + let created = Utc::now().timestamp(); + let mut output = String::new(); + let mut completed_response = None; + let mut has_tool_calls = false; + let mut function_call_items = BTreeMap::new(); + let mut tool_call_indices = BTreeMap::new(); + let mut call_ids_with_deltas = BTreeSet::new(); + let mut next_tool_call_index = 0usize; + + push_sse_data( + &mut output, + json!({ + "id": chunk_id, + "object": "chat.completion.chunk", + "created": created, + "model": model, + "choices": [{ + "index": 0, + "delta": {"role": "assistant"}, + "finish_reason": null, + }], + }), + ); + + for event in events { + let Ok(value) = serde_json::from_str::(&event.data) else { + continue; + }; + match event.event.as_deref() { + Some("response.output_text.delta") => { + if let Some(delta) = value.get("delta").and_then(Value::as_str) { + push_sse_data( + &mut output, + json!({ + "id": chunk_id, + "object": "chat.completion.chunk", + "created": created, + "model": model, + "choices": [{ + "index": 0, + "delta": {"content": delta}, + "finish_reason": null, + }], + }), + ); + } + } + Some("response.reasoning_summary_text.delta") if include_reasoning => { + if let Some(delta) = value.get("delta").and_then(Value::as_str) { + push_sse_data( + &mut output, + json!({ + "id": chunk_id, + "object": "chat.completion.chunk", + "created": created, + "model": model, + "choices": [{ + "index": 0, + "delta": {"reasoning_content": delta}, + "finish_reason": null, + }], + }), + ); + } + } + Some("response.output_item.added") => { + if let Some((item_id, call_id, name)) = function_call_item(value.get("item")) { + has_tool_calls = true; + let index = next_tool_call_index; + next_tool_call_index = next_tool_call_index.saturating_add(1); + function_call_items.insert( + item_id, + FunctionCallInfo { + call_id: call_id.clone(), + name: name.clone(), + }, + ); + tool_call_indices.insert(call_id.clone(), index); + push_sse_data( + &mut output, + json!({ + "id": chunk_id, + "object": "chat.completion.chunk", + "created": created, + "model": model, + "choices": [{ + "index": 0, + "delta": { + "tool_calls": [{ + "index": index, + "id": call_id, + "type": "function", + "function": { + "name": name, + "arguments": "", + }, + }], + }, + "finish_reason": null, + }], + }), + ); + } + } + Some("response.function_call_arguments.delta") => { + if let Some((call_id, index, delta)) = + stream_tool_call_delta(&value, &function_call_items, &tool_call_indices) + { + has_tool_calls = true; + call_ids_with_deltas.insert(call_id); + push_sse_data( + &mut output, + json!({ + "id": chunk_id, + "object": "chat.completion.chunk", + "created": created, + "model": model, + "choices": [{ + "index": 0, + "delta": { + "tool_calls": [{ + "index": index, + "function": { + "arguments": delta, + }, + }], + }, + "finish_reason": null, + }], + }), + ); + } + } + Some("response.function_call_arguments.done") => { + if let Some((call_id, index, arguments)) = stream_tool_call_done( + &value, + &function_call_items, + &tool_call_indices, + &call_ids_with_deltas, + ) { + has_tool_calls = true; + push_sse_data( + &mut output, + json!({ + "id": chunk_id, + "object": "chat.completion.chunk", + "created": created, + "model": model, + "choices": [{ + "index": 0, + "delta": { + "tool_calls": [{ + "index": index, + "function": { + "arguments": arguments, + }, + }], + }, + "finish_reason": null, + }], + }), + ); + call_ids_with_deltas.insert(call_id); + } + } + Some("response.output_item.done") => { + if is_function_call_item(value.get("item")) { + has_tool_calls = true; + } + } + Some("response.completed") => { + completed_response = value.get("response").cloned(); + } + _ => {} + } + } + + let Some(response) = completed_response else { + return Ok(None); + }; + + push_sse_data( + &mut output, + json!({ + "id": chunk_id, + "object": "chat.completion.chunk", + "created": created, + "model": model, + "choices": [{ + "index": 0, + "delta": {}, + "finish_reason": if has_tool_calls { "tool_calls" } else { "stop" }, + }], + "usage": openai_usage(response.get("usage")), + }), + ); + output.push_str("data: [DONE]\n\n"); + + Ok(Some(output)) +} + +fn output_text_from_response(response: &Value) -> String { + response + .get("output_text") + .and_then(Value::as_str) + .map(ToString::to_string) + .unwrap_or_else(|| { + response + .get("output") + .and_then(Value::as_array) + .map(|items| { + items + .iter() + .filter_map(output_text_from_item) + .collect::>() + .join("\n\n") + }) + .unwrap_or_default() + }) +} + +fn output_text_from_item(item: &Value) -> Option { + let content = item.get("content")?.as_array()?; + let text = content + .iter() + .filter_map(|part| { + let part_type = part.get("type")?.as_str()?; + if part_type != "output_text" && part_type != "text" { + return None; + } + part.get("text")?.as_str() + }) + .collect::(); + (!text.is_empty()).then_some(text) +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct FunctionCallInfo { + call_id: String, + name: String, +} + +fn function_call_item(item: Option<&Value>) -> Option<(String, String, String)> { + let item = item?; + if item.get("type").and_then(Value::as_str) != Some("function_call") { + return None; + } + let item_id = item.get("id").and_then(Value::as_str)?; + let call_id = item.get("call_id").and_then(Value::as_str)?; + let name = item.get("name").and_then(Value::as_str)?; + Some((item_id.to_string(), call_id.to_string(), name.to_string())) +} + +fn tool_call_from_done_event( + event: &Value, + function_call_items: &BTreeMap, + finished_call_ids: &mut BTreeSet, +) -> Option { + let event_id = event + .get("call_id") + .or_else(|| event.get("item_id")) + .and_then(Value::as_str)?; + let info = function_call_items.get(event_id); + let call_id = info + .map(|info| info.call_id.as_str()) + .unwrap_or(event_id) + .to_string(); + if !finished_call_ids.insert(call_id.clone()) { + return None; + } + let name = event + .get("name") + .and_then(Value::as_str) + .filter(|name| !name.is_empty()) + .or_else(|| info.map(|info| info.name.as_str())) + .unwrap_or("unknown"); + let arguments = event + .get("arguments") + .and_then(Value::as_str) + .unwrap_or_default(); + Some(openai_tool_call(call_id, name, arguments)) +} + +fn tool_call_from_output_item( + item: Option<&Value>, + function_call_items: &BTreeMap, + finished_call_ids: &mut BTreeSet, +) -> Option { + let item = item?; + if item.get("type").and_then(Value::as_str) != Some("function_call") { + return None; + } + let item_id = item.get("id").and_then(Value::as_str); + let info = item_id.and_then(|item_id| function_call_items.get(item_id)); + let call_id = item + .get("call_id") + .and_then(Value::as_str) + .or_else(|| info.map(|info| info.call_id.as_str())) + .or(item_id)?; + if !finished_call_ids.insert(call_id.to_string()) { + return None; + } + let name = item + .get("name") + .and_then(Value::as_str) + .filter(|name| !name.is_empty()) + .or_else(|| info.map(|info| info.name.as_str())) + .unwrap_or("unknown"); + let arguments = item + .get("arguments") + .and_then(Value::as_str) + .unwrap_or_default(); + Some(openai_tool_call(call_id.to_string(), name, arguments)) +} + +fn openai_tool_call(call_id: String, name: &str, arguments: &str) -> Value { + json!({ + "id": call_id, + "type": "function", + "function": { + "name": name, + "arguments": arguments, + }, + }) +} + +fn stream_tool_call_delta( + event: &Value, + function_call_items: &BTreeMap, + tool_call_indices: &BTreeMap, +) -> Option<(String, usize, String)> { + let event_id = event + .get("call_id") + .or_else(|| event.get("item_id")) + .and_then(Value::as_str)?; + let info = function_call_items.get(event_id); + let call_id = info + .map(|info| info.call_id.as_str()) + .unwrap_or(event_id) + .to_string(); + let index = *tool_call_indices.get(&call_id).unwrap_or(&0); + let delta = event.get("delta").and_then(Value::as_str)?.to_string(); + Some((call_id, index, delta)) +} + +fn stream_tool_call_done( + event: &Value, + function_call_items: &BTreeMap, + tool_call_indices: &BTreeMap, + call_ids_with_deltas: &BTreeSet, +) -> Option<(String, usize, String)> { + let event_id = event + .get("call_id") + .or_else(|| event.get("item_id")) + .and_then(Value::as_str)?; + let info = function_call_items.get(event_id); + let call_id = info + .map(|info| info.call_id.as_str()) + .unwrap_or(event_id) + .to_string(); + if call_ids_with_deltas.contains(&call_id) { + return None; + } + let index = *tool_call_indices.get(&call_id).unwrap_or(&0); + let arguments = event + .get("arguments") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(); + Some((call_id, index, arguments)) +} + +fn push_sse_data(output: &mut String, value: Value) { + output.push_str("data: "); + output.push_str(&value.to_string()); + output.push_str("\n\n"); +} + +fn is_function_call_item(item: Option<&Value>) -> bool { + item.and_then(|item| item.get("type")) + .and_then(Value::as_str) + == Some("function_call") +} + +fn openai_usage(usage: Option<&Value>) -> Value { + let prompt_tokens = usage + .and_then(|usage| usage.get("input_tokens")) + .and_then(Value::as_u64) + .unwrap_or_default(); + let completion_tokens = usage + .and_then(|usage| usage.get("output_tokens")) + .and_then(Value::as_u64) + .unwrap_or_default(); + let cached_tokens = usage + .and_then(|usage| usage.get("input_tokens_details")) + .and_then(|details| details.get("cached_tokens")) + .and_then(Value::as_u64) + .unwrap_or_default(); + let reasoning_tokens = usage + .and_then(|usage| usage.get("output_tokens_details")) + .and_then(|details| details.get("reasoning_tokens")) + .and_then(Value::as_u64); + let mut openai_usage = json!({ + "prompt_tokens": prompt_tokens, + "completion_tokens": completion_tokens, + "total_tokens": prompt_tokens.saturating_add(completion_tokens), + "prompt_tokens_details": { + "cached_tokens": cached_tokens, + }, + }); + if let Some(reasoning_tokens) = reasoning_tokens { + openai_usage["completion_tokens_details"] = json!({ + "reasoning_tokens": reasoning_tokens, + }); + } + openai_usage +} diff --git a/src/translation/openai_to_codex.rs b/src/translation/openai_to_codex.rs index ae5dbc855..813a21cd3 100644 --- a/src/translation/openai_to_codex.rs +++ b/src/translation/openai_to_codex.rs @@ -1,5 +1,5 @@ use serde::{Deserialize, Serialize}; -use serde_json::json; +use serde_json::{json, Value}; use crate::{ codex::types::CodexResponsesRequest, @@ -12,12 +12,37 @@ pub struct ChatCompletionRequest { #[serde(default)] pub stream: bool, pub messages: Vec, + #[serde(default)] + pub reasoning_effort: Option, + #[serde(default)] + pub service_tier: Option, + #[serde(default)] + pub tools: Option>, + #[serde(default)] + pub tool_choice: Option, + #[serde(default)] + pub parallel_tool_calls: Option, + #[serde(default)] + pub functions: Option>, + #[serde(default)] + pub response_format: Option, + #[serde(default)] + pub user: Option, } #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ChatMessage { pub role: String, - pub content: String, + #[serde(default)] + pub content: Option, + #[serde(default)] + pub name: Option, + #[serde(default)] + pub tool_calls: Option>, + #[serde(default)] + pub tool_call_id: Option, + #[serde(default)] + pub function_call: Option, } pub fn translate_chat_to_codex(req: ChatCompletionRequest) -> AppResult { @@ -26,20 +51,218 @@ pub fn translate_chat_to_codex(req: ChatCompletionRequest) -> AppResult String { + let instructions = messages + .iter() + .filter(|message| message.role == "system" || message.role == "developer") + .map(|message| extract_text(&message.content)) + .filter(|text| !text.is_empty()) + .collect::>() + .join("\n\n"); + if instructions.is_empty() { + "You are a helpful assistant.".to_string() + } else { + instructions + } +} + +fn chat_input(messages: Vec) -> Vec { + let mut input = Vec::new(); + for message in messages { + match message.role.as_str() { + "system" | "developer" => {} + "assistant" => push_assistant_message(&mut input, message), + "tool" => input.push(json!({ + "type": "function_call_output", + "call_id": message.tool_call_id.unwrap_or_else(|| "unknown".to_string()), + "output": extract_text(&message.content), + })), + "function" => input.push(json!({ + "type": "function_call_output", + "call_id": format!("fc_{}", message.name.unwrap_or_else(|| "unknown".to_string())), + "output": extract_text(&message.content), + })), + _ => input.push(json!({ + "role": "user", + "content": extract_content(&message.content), + })), + } + } + input +} + +fn push_assistant_message(input: &mut Vec, message: ChatMessage) { + let text = extract_text(&message.content); + let has_tool_calls = message + .tool_calls + .as_ref() + .is_some_and(|calls| !calls.is_empty()); + if !text.is_empty() || (!has_tool_calls && message.function_call.is_none()) { + input.push(json!({"role": "assistant", "content": text})); + } + + if let Some(tool_calls) = message.tool_calls { + for tool_call in tool_calls { + let call_id = tool_call + .get("id") + .and_then(Value::as_str) + .unwrap_or("unknown"); + let function = tool_call.get("function").unwrap_or(&Value::Null); + let name = function + .get("name") + .and_then(Value::as_str) + .unwrap_or("unknown"); + let arguments = function + .get("arguments") + .and_then(Value::as_str) + .unwrap_or(""); + input.push(json!({ + "type": "function_call", + "call_id": call_id, + "name": name, + "arguments": arguments, + })); + } + } + + if let Some(function_call) = message.function_call { + let name = function_call + .get("name") + .and_then(Value::as_str) + .unwrap_or("unknown"); + let arguments = function_call + .get("arguments") + .and_then(Value::as_str) + .unwrap_or(""); + input.push(json!({ + "type": "function_call", + "call_id": format!("fc_{name}"), + "name": name, + "arguments": arguments, + })); + } +} + +fn extract_text(content: &Option) -> String { + let Some(content) = content else { + return String::new(); + }; + if let Some(text) = content.as_str() { + return text.to_string(); + } + let Some(parts) = content.as_array() else { + return String::new(); + }; + parts + .iter() + .filter_map(|part| { + (part.get("type").and_then(Value::as_str) == Some("text")) + .then(|| part.get("text").and_then(Value::as_str)) + .flatten() + }) + .collect::>() + .join("\n") +} + +fn extract_content(content: &Option) -> Value { + let Some(content) = content else { + return Value::String(String::new()); + }; + if content.as_str().is_some() { + return content.clone(); + } + let Some(parts) = content.as_array() else { + return Value::String(String::new()); + }; + let has_image = parts + .iter() + .any(|part| part.get("type").and_then(Value::as_str) == Some("image_url")); + if !has_image { + return Value::String(extract_text(&Some(content.clone()))); + } + let codex_parts = parts + .iter() + .filter_map(codex_content_part) + .collect::>(); + if codex_parts.is_empty() { + Value::String(String::new()) + } else { + Value::Array(codex_parts) + } +} + +fn codex_content_part(part: &Value) -> Option { + match part.get("type").and_then(Value::as_str)? { + "text" => part + .get("text") + .and_then(Value::as_str) + .map(|text| json!({"type": "input_text", "text": text})), + "image_url" => image_url(part).map(|url| json!({"type": "input_image", "image_url": url})), + _ => None, + } +} + +fn image_url(part: &Value) -> Option<&str> { + let image = part.get("image_url")?; + image + .as_str() + .or_else(|| image.get("url").and_then(Value::as_str)) +} + +fn codex_tools(tools: Option>, functions: Option>) -> Option> { + if let Some(tools) = tools.filter(|tools| !tools.is_empty()) { + return Some(tools); + } + functions + .filter(|functions| !functions.is_empty()) + .map(|functions| { + functions + .into_iter() + .map(|function| json!({"type": "function", "function": function})) + .collect() + }) +} + +fn response_format_text(response_format: Option) -> Option { + let format = response_format?; + let kind = format.get("type").and_then(Value::as_str)?; + match kind { + "json_object" => Some(json!({"format": {"type": "json_object"}})), + "json_schema" => { + let schema = format.get("json_schema")?; + let name = schema + .get("name") + .and_then(Value::as_str) + .unwrap_or("response"); + let mut codex_format = json!({ + "type": "json_schema", + "name": name, + "schema": schema.get("schema").cloned().unwrap_or_else(|| json!({})), + }); + if let Some(strict) = schema.get("strict").and_then(Value::as_bool) { + codex_format["strict"] = Value::Bool(strict); + } + Some(json!({"format": codex_format})) + } + _ => None, + } } diff --git a/tests/chat_completions_route_test.rs b/tests/chat_completions_route_test.rs new file mode 100644 index 000000000..c000e9a82 --- /dev/null +++ b/tests/chat_completions_route_test.rs @@ -0,0 +1,595 @@ +use std::collections::BTreeMap; + +use axum::{ + body::{to_bytes, Body}, + http::{Request, StatusCode}, +}; +use chrono::Utc; +use serde_json::{json, Value}; +use tower::ServiceExt; +use wiremock::{ + matchers::{body_json, header, method, path}, + Mock, MockServer, ResponseTemplate, +}; + +use codex_proxy_rs::{ + app::build_router, + auth::{api_key::ApiKeyHasher, api_key_repository::ClientApiKeyRepository}, + config::{ + AdminConfig, ApiConfig, AppConfig, AuthConfig, DatabaseConfig, LoggingConfig, ModelConfig, + QuotaConfig, QuotaWarningThresholds, SecurityConfig, ServerConfig, TlsConfig, + UsageStatsConfig, + }, + crypto::SecretBox, + state::AppState, + storage::db::connect_sqlite, +}; + +struct ImportedApp { + app: axum::Router, + client_api_key: String, + _tempdir: tempfile::TempDir, +} + +fn test_config(database_url: String, base_url: String) -> AppConfig { + AppConfig { + server: ServerConfig { + host: "127.0.0.1".to_string(), + port: 0, + }, + api: ApiConfig { base_url }, + model: ModelConfig { + default_model: "gpt-5.5".to_string(), + default_reasoning_effort: None, + service_tier: None, + aliases: BTreeMap::new(), + }, + auth: AuthConfig { + refresh_margin_seconds: 300, + refresh_enabled: true, + refresh_concurrency: 2, + max_concurrent_per_account: 3, + request_interval_ms: 50, + rotation_strategy: "least_used".to_string(), + tier_priority: Vec::new(), + }, + quota: QuotaConfig { + refresh_interval_minutes: 5, + warning_thresholds: QuotaWarningThresholds { + primary: vec![80, 90], + secondary: vec![80, 90], + }, + skip_exhausted: true, + }, + usage_stats: UsageStatsConfig { + history_retention_days: None, + }, + database: DatabaseConfig { url: database_url }, + security: SecurityConfig { + master_key_file: "data/master.key".to_string(), + api_key_pepper_file: "data/api-key-pepper.key".to_string(), + }, + tls: TlsConfig { + force_http11: false, + }, + admin: AdminConfig { + session_ttl_minutes: 1440, + }, + logging: LoggingConfig { + directory: "logs".to_string(), + max_file_bytes: 10_485_760, + retention_days: 14, + enabled: false, + capacity: 2_000, + capture_body: false, + }, + } +} + +async fn build_imported_app(base_url: String) -> ImportedApp { + let tempdir = tempfile::tempdir().unwrap(); + let db = tempdir.path().join("chat-completions.sqlite"); + let url = format!("sqlite://{}", db.display()); + let pool = connect_sqlite(&url).await.unwrap(); + seed_admin_session(&pool, "session_1").await; + let secret_box = SecretBox::new([61u8; 32]); + let hasher = ApiKeyHasher::new([62u8; 32]); + let generated = hasher.generate_client_api_key("test"); + ClientApiKeyRepository::new(pool.clone()) + .insert_generated("test", &generated) + .await + .unwrap(); + let app = build_router(AppState::with_pool_secret_and_api_key_hasher( + test_config(url, base_url), + pool, + secret_box, + hasher, + )); + + let import_response = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri("/admin/accounts/import") + .header("content-type", "application/json") + .header("cookie", "cpr_admin_session=session_1") + .body(Body::from( + json!({ + "accounts": [{ + "id": "acct_chat", + "accountId": "chatgpt-account", + "token": "access-secret", + "refreshToken": "refresh-secret", + "status": "active" + }] + }) + .to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(import_response.status(), StatusCode::OK); + + ImportedApp { + app, + client_api_key: generated.plaintext, + _tempdir: tempdir, + } +} + +#[tokio::test] +async fn chat_completions_translates_messages_and_returns_openai_response() { + let server = MockServer::start().await; + let sse_body = concat!( + "event: response.output_text.delta\n", + "data: {\"delta\":\"hello\"}\n", + "\n", + "event: response.completed\n", + "data: {\"response\":{\"id\":\"resp_chat\",\"object\":\"response\",\"status\":\"completed\",\"usage\":{\"input_tokens\":9,\"output_tokens\":3}}}\n", + "\n", + ); + Mock::given(method("POST")) + .and(path("/codex/responses")) + .and(header("authorization", "Bearer access-secret")) + .and(header("chatgpt-account-id", "chatgpt-account")) + .and(body_json(json!({ + "model": "gpt-5.5", + "instructions": "You are concise.", + "input": [{ + "role": "user", + "content": "Say hello" + }], + "stream": true, + "store": false + }))) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(sse_body), + ) + .mount(&server) + .await; + let imported = build_imported_app(server.uri()).await; + + let response = imported + .app + .oneshot( + Request::builder() + .method("POST") + .uri("/v1/chat/completions") + .header( + "authorization", + format!("Bearer {}", imported.client_api_key), + ) + .header("content-type", "application/json") + .body(Body::from( + json!({ + "model": "gpt-5.5", + "messages": [ + {"role": "system", "content": "You are concise."}, + {"role": "user", "content": "Say hello"} + ] + }) + .to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(response.status(), StatusCode::OK); + let body = response_json(response).await; + assert!(body["id"].as_str().unwrap().starts_with("chatcmpl-")); + assert_eq!(body["object"], "chat.completion"); + assert_eq!(body["model"], "gpt-5.5"); + assert_eq!(body["choices"][0]["message"]["role"], "assistant"); + assert_eq!(body["choices"][0]["message"]["content"], "hello"); + assert_eq!(body["choices"][0]["finish_reason"], "stop"); + assert_eq!(body["usage"]["prompt_tokens"], 9); + assert_eq!(body["usage"]["completion_tokens"], 3); + assert_eq!(body["usage"]["total_tokens"], 12); +} + +#[tokio::test] +async fn chat_completions_streams_openai_chunks_from_codex_sse() { + let server = MockServer::start().await; + let sse_body = concat!( + "event: response.output_text.delta\n", + "data: {\"delta\":\"hel\"}\n", + "\n", + "event: response.output_text.delta\n", + "data: {\"delta\":\"lo\"}\n", + "\n", + "event: response.completed\n", + "data: {\"response\":{\"id\":\"resp_chat\",\"object\":\"response\",\"status\":\"completed\",\"usage\":{\"input_tokens\":9,\"output_tokens\":3,\"input_tokens_details\":{\"cached_tokens\":4}}}}\n", + "\n", + ); + Mock::given(method("POST")) + .and(path("/codex/responses")) + .and(header("authorization", "Bearer access-secret")) + .and(header("chatgpt-account-id", "chatgpt-account")) + .and(body_json(json!({ + "model": "gpt-5.5", + "instructions": "You are a helpful assistant.", + "input": [{ + "role": "user", + "content": "Say hello" + }], + "stream": true, + "store": false + }))) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(sse_body), + ) + .mount(&server) + .await; + let imported = build_imported_app(server.uri()).await; + + let response = imported + .app + .oneshot( + Request::builder() + .method("POST") + .uri("/v1/chat/completions") + .header( + "authorization", + format!("Bearer {}", imported.client_api_key), + ) + .header("content-type", "application/json") + .body(Body::from( + json!({ + "model": "gpt-5.5", + "stream": true, + "messages": [ + {"role": "user", "content": "Say hello"} + ] + }) + .to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(response.status(), StatusCode::OK); + let frames = response_sse_data(response).await; + assert_eq!(frames.last().unwrap(), "[DONE]"); + let chunks = frames + .iter() + .filter(|frame| frame.as_str() != "[DONE]") + .map(|frame| serde_json::from_str::(frame).unwrap()) + .collect::>(); + + assert_eq!(chunks[0]["object"], "chat.completion.chunk"); + assert_eq!(chunks[0]["model"], "gpt-5.5"); + assert_eq!(chunks[0]["choices"][0]["delta"]["role"], "assistant"); + assert_eq!(chunks[1]["choices"][0]["delta"]["content"], "hel"); + assert_eq!(chunks[2]["choices"][0]["delta"]["content"], "lo"); + assert_eq!(chunks[3]["choices"][0]["finish_reason"], "stop"); + assert_eq!(chunks[3]["usage"]["prompt_tokens"], 9); + assert_eq!(chunks[3]["usage"]["completion_tokens"], 3); + assert_eq!(chunks[3]["usage"]["total_tokens"], 12); + assert_eq!( + chunks[3]["usage"]["prompt_tokens_details"]["cached_tokens"], + 4 + ); +} + +#[tokio::test] +async fn chat_completions_forwards_parallel_tools_and_model_suffix_options() { + let server = MockServer::start().await; + let sse_body = concat!( + "event: response.output_text.delta\n", + "data: {\"delta\":\"ok\"}\n", + "\n", + "event: response.completed\n", + "data: {\"response\":{\"id\":\"resp_chat\",\"object\":\"response\",\"status\":\"completed\",\"usage\":{\"input_tokens\":2,\"output_tokens\":1}}}\n", + "\n", + ); + Mock::given(method("POST")) + .and(path("/codex/responses")) + .and(header("authorization", "Bearer access-secret")) + .and(header("chatgpt-account-id", "chatgpt-account")) + .and(body_json(json!({ + "model": "gpt-5.5", + "instructions": "You are a helpful assistant.", + "input": [{ + "role": "user", + "content": "Use tools if needed" + }], + "stream": true, + "store": false, + "reasoning": { + "effort": "high", + "summary": "auto" + }, + "service_tier": "priority", + "parallel_tool_calls": true + }))) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(sse_body), + ) + .mount(&server) + .await; + let imported = build_imported_app(server.uri()).await; + + let response = imported + .app + .oneshot( + Request::builder() + .method("POST") + .uri("/v1/chat/completions") + .header( + "authorization", + format!("Bearer {}", imported.client_api_key), + ) + .header("content-type", "application/json") + .body(Body::from( + json!({ + "model": "gpt-5.5-high-fast", + "parallel_tool_calls": true, + "messages": [ + {"role": "user", "content": "Use tools if needed"} + ] + }) + .to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(response.status(), StatusCode::OK); + let body = response_json(response).await; + assert_eq!(body["model"], "gpt-5.5-high-fast"); +} + +#[tokio::test] +async fn chat_completions_collects_tool_calls_and_reasoning_content() { + let server = MockServer::start().await; + let sse_body = concat!( + "event: response.reasoning_summary_text.delta\n", + "data: {\"delta\":\"thinking \"}\n", + "\n", + "event: response.output_item.added\n", + "data: {\"output_index\":0,\"item\":{\"type\":\"function_call\",\"id\":\"item_1\",\"call_id\":\"call_abc\",\"name\":\"lookup\"}}\n", + "\n", + "event: response.function_call_arguments.delta\n", + "data: {\"item_id\":\"item_1\",\"delta\":\"{\\\"city\\\"\"}\n", + "\n", + "event: response.function_call_arguments.delta\n", + "data: {\"item_id\":\"item_1\",\"delta\":\":\\\"Paris\\\"}\"}\n", + "\n", + "event: response.function_call_arguments.done\n", + "data: {\"item_id\":\"item_1\",\"arguments\":\"{\\\"city\\\":\\\"Paris\\\"}\"}\n", + "\n", + "event: response.completed\n", + "data: {\"response\":{\"id\":\"resp_chat\",\"object\":\"response\",\"status\":\"completed\",\"usage\":{\"input_tokens\":5,\"output_tokens\":2,\"output_tokens_details\":{\"reasoning_tokens\":7}}}}\n", + "\n", + ); + Mock::given(method("POST")) + .and(path("/codex/responses")) + .and(header("authorization", "Bearer access-secret")) + .and(header("chatgpt-account-id", "chatgpt-account")) + .and(body_json(json!({ + "model": "gpt-5.5", + "instructions": "You are a helpful assistant.", + "input": [{ + "role": "user", + "content": "Find it" + }], + "stream": true, + "store": false, + "reasoning": { + "effort": "high", + "summary": "auto" + } + }))) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(sse_body), + ) + .mount(&server) + .await; + let imported = build_imported_app(server.uri()).await; + + let response = imported + .app + .oneshot( + Request::builder() + .method("POST") + .uri("/v1/chat/completions") + .header( + "authorization", + format!("Bearer {}", imported.client_api_key), + ) + .header("content-type", "application/json") + .body(Body::from( + json!({ + "model": "gpt-5.5", + "reasoning_effort": "high", + "messages": [ + {"role": "user", "content": "Find it"} + ] + }) + .to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(response.status(), StatusCode::OK); + let body = response_json(response).await; + let message = &body["choices"][0]["message"]; + assert_eq!(message["content"], Value::Null); + assert_eq!(message["reasoning_content"], "thinking "); + assert_eq!(message["tool_calls"][0]["id"], "call_abc"); + assert_eq!(message["tool_calls"][0]["type"], "function"); + assert_eq!(message["tool_calls"][0]["function"]["name"], "lookup"); + assert_eq!( + message["tool_calls"][0]["function"]["arguments"], + "{\"city\":\"Paris\"}" + ); + assert_eq!(body["choices"][0]["finish_reason"], "tool_calls"); + assert_eq!( + body["usage"]["completion_tokens_details"]["reasoning_tokens"], + 7 + ); +} + +#[tokio::test] +async fn chat_completions_streams_tool_call_chunks_from_codex_sse() { + let server = MockServer::start().await; + let sse_body = concat!( + "event: response.output_item.added\n", + "data: {\"output_index\":0,\"item\":{\"type\":\"function_call\",\"id\":\"item_1\",\"call_id\":\"call_abc\",\"name\":\"lookup\"}}\n", + "\n", + "event: response.function_call_arguments.delta\n", + "data: {\"item_id\":\"item_1\",\"delta\":\"{\\\"city\\\"\"}\n", + "\n", + "event: response.function_call_arguments.delta\n", + "data: {\"item_id\":\"item_1\",\"delta\":\":\\\"Paris\\\"}\"}\n", + "\n", + "event: response.completed\n", + "data: {\"response\":{\"id\":\"resp_chat\",\"object\":\"response\",\"status\":\"completed\",\"usage\":{\"input_tokens\":5,\"output_tokens\":2}}}\n", + "\n", + ); + Mock::given(method("POST")) + .and(path("/codex/responses")) + .and(header("authorization", "Bearer access-secret")) + .and(header("chatgpt-account-id", "chatgpt-account")) + .and(body_json(json!({ + "model": "gpt-5.5", + "instructions": "You are a helpful assistant.", + "input": [{ + "role": "user", + "content": "Find it" + }], + "stream": true, + "store": false + }))) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(sse_body), + ) + .mount(&server) + .await; + let imported = build_imported_app(server.uri()).await; + + let response = imported + .app + .oneshot( + Request::builder() + .method("POST") + .uri("/v1/chat/completions") + .header( + "authorization", + format!("Bearer {}", imported.client_api_key), + ) + .header("content-type", "application/json") + .body(Body::from( + json!({ + "model": "gpt-5.5", + "stream": true, + "messages": [ + {"role": "user", "content": "Find it"} + ] + }) + .to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(response.status(), StatusCode::OK); + let frames = response_sse_data(response).await; + let chunks = frames + .iter() + .filter(|frame| frame.as_str() != "[DONE]") + .map(|frame| serde_json::from_str::(frame).unwrap()) + .collect::>(); + + let start_call = &chunks[1]["choices"][0]["delta"]["tool_calls"][0]; + assert_eq!(start_call["index"], 0); + assert_eq!(start_call["id"], "call_abc"); + assert_eq!(start_call["type"], "function"); + assert_eq!(start_call["function"]["name"], "lookup"); + assert_eq!(start_call["function"]["arguments"], ""); + assert_eq!( + chunks[2]["choices"][0]["delta"]["tool_calls"][0]["function"]["arguments"], + "{\"city\"" + ); + assert_eq!( + chunks[3]["choices"][0]["delta"]["tool_calls"][0]["function"]["arguments"], + ":\"Paris\"}" + ); + assert_eq!(chunks[4]["choices"][0]["finish_reason"], "tool_calls"); +} + +async fn seed_admin_session(pool: &sqlx::SqlitePool, session_id: &str) { + let now = Utc::now().to_rfc3339(); + sqlx::query( + "insert into admin_users (id, password_hash, created_at, updated_at) values (?, ?, ?, ?)", + ) + .bind("admin_1") + .bind("hash") + .bind(&now) + .bind(&now) + .execute(pool) + .await + .unwrap(); + sqlx::query( + "insert into admin_sessions (id, user_id, expires_at, created_at) values (?, ?, ?, ?)", + ) + .bind(session_id) + .bind("admin_1") + .bind("2999-01-01T00:00:00Z") + .bind(now) + .execute(pool) + .await + .unwrap(); +} + +async fn response_json(response: axum::response::Response) -> Value { + let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + serde_json::from_slice(&bytes).unwrap() +} + +async fn response_sse_data(response: axum::response::Response) -> Vec { + let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + String::from_utf8(bytes.to_vec()) + .unwrap() + .lines() + .filter_map(|line| line.strip_prefix("data: ").map(ToString::to_string)) + .collect() +} diff --git a/tests/codex_client_test.rs b/tests/codex_client_test.rs index cf5f73107..93720e33d 100644 --- a/tests/codex_client_test.rs +++ b/tests/codex_client_test.rs @@ -56,22 +56,18 @@ async fn codex_backend_client_should_send_desktop_headers_and_capture_response_m let response = client .create_response( - &CodexResponsesRequest { - model: "gpt-5.5".to_string(), - instructions: String::new(), - input: Vec::new(), - stream: true, - store: false, - reasoning: None, - tools: None, - previous_response_id: None, - use_websocket: false, - }, + &CodexResponsesRequest::new_http_sse("gpt-5.5", "", Vec::new()), CodexRequestContext { access_token: "access-token", account_id: Some("chatgpt-account"), request_id: "req_1", turn_state: Some("turn_1"), + turn_metadata: None, + beta_features: None, + include_timing_metrics: None, + version: None, + codex_window_id: None, + parent_thread_id: None, cookie_header: Some("cf_clearance=old"), }, ) diff --git a/tests/codex_websocket_test.rs b/tests/codex_websocket_test.rs index 445a6d36b..9f888fbf6 100644 --- a/tests/codex_websocket_test.rs +++ b/tests/codex_websocket_test.rs @@ -26,16 +26,17 @@ fn transport_for_request_should_require_websocket_for_previous_response_id() { ); } +#[test] +fn use_websocket_should_not_serialize_to_upstream_json() { + let mut request = base_request(); + request.use_websocket = true; + + let body = serde_json::to_value(&request).unwrap(); + + assert!(body.get("use_websocket").is_none()); + assert!(body.get("useWebSocket").is_none()); +} + fn base_request() -> CodexResponsesRequest { - CodexResponsesRequest { - model: "gpt-5.5".to_string(), - instructions: String::new(), - input: Vec::new(), - stream: true, - store: false, - reasoning: None, - tools: None, - previous_response_id: None, - use_websocket: false, - } + CodexResponsesRequest::new_http_sse("gpt-5.5", "", Vec::new()) } diff --git a/tests/routes_chat_test.rs b/tests/routes_chat_test.rs index 6fca46f19..b58297ce3 100644 --- a/tests/routes_chat_test.rs +++ b/tests/routes_chat_test.rs @@ -1,3 +1,5 @@ +use serde_json::json; + use codex_proxy_rs::translation::openai_to_codex::{ translate_chat_to_codex, ChatCompletionRequest, ChatMessage, }; @@ -9,8 +11,20 @@ fn chat_completion_translates_to_codex_response_request() { stream: true, messages: vec![ChatMessage { role: "user".to_string(), - content: "hello".to_string(), + content: Some(json!("hello")), + name: None, + tool_calls: None, + tool_call_id: None, + function_call: None, }], + reasoning_effort: None, + service_tier: None, + tools: None, + tool_choice: None, + parallel_tool_calls: None, + functions: None, + response_format: None, + user: None, }; let codex = translate_chat_to_codex(req).unwrap(); diff --git a/tests/v1_upstream_route_test.rs b/tests/v1_upstream_route_test.rs index 88a3b4704..19bd10d9c 100644 --- a/tests/v1_upstream_route_test.rs +++ b/tests/v1_upstream_route_test.rs @@ -300,7 +300,9 @@ async fn v1_responses_should_use_imported_account_and_record_usage() { ) .header("content-type", "application/json") .header("x-request-id", "req_non_stream") - .body(Body::from(r#"{"model":"gpt-5.5","input":[]}"#)) + .body(Body::from( + r#"{"model":"gpt-5.5","input":[],"stream":false}"#, + )) .unwrap(), ) .await @@ -332,6 +334,176 @@ async fn v1_responses_should_use_imported_account_and_record_usage() { assert_eq!(event.4["usage"]["inputTokens"], 7); } +#[tokio::test] +async fn v1_responses_should_forward_parity_fields_and_context_headers() { + let server = MockServer::start().await; + let sse_body = concat!( + "event: response.completed\n", + "data: {\"response\":{\"id\":\"resp_fields\",\"object\":\"response\",\"usage\":{\"input_tokens\":1,\"output_tokens\":1}}}\n", + "\n", + ); + Mock::given(method("POST")) + .and(path("/codex/responses")) + .and(header("authorization", "Bearer access-secret")) + .and(header("chatgpt-account-id", "chatgpt-account")) + .and(header("x-codex-turn-state", "turn-body")) + .and(header("x-codex-turn-metadata", "meta-header")) + .and(header("x-codex-beta-features", "beta-header")) + .and(header("x-responsesapi-include-timing-metrics", "true")) + .and(header("version", "2026-06-12")) + .and(header("x-codex-window-id", "window-1")) + .and(header("x-codex-parent-thread-id", "parent-1")) + .and(body_json(json!({ + "model": "gpt-5.5", + "instructions": "", + "input": [], + "stream": true, + "store": false, + "reasoning": { + "effort": "high", + "summary": "auto" + }, + "service_tier": "priority", + "tool_choice": { + "type": "function", + "function": {"name": "lookup"} + }, + "parallel_tool_calls": true, + "text": { + "format": { + "type": "json_schema", + "name": "Answer", + "schema": {"type": "object"}, + "strict": true + } + }, + "prompt_cache_key": "pcache", + "include": ["reasoning.encrypted_content"], + "client_metadata": { + "safe": "yes" + } + }))) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(sse_body), + ) + .mount(&server) + .await; + let imported = build_imported_app(server.uri()).await; + + let response = imported + .app + .oneshot( + Request::builder() + .method("POST") + .uri("/v1/responses") + .header( + "authorization", + format!("Bearer {}", imported.client_api_key), + ) + .header("content-type", "application/json") + .header("x-codex-turn-state", "turn-header") + .header("x-codex-turn-metadata", "meta-header") + .header("x-codex-beta-features", "beta-header") + .header("x-responsesapi-include-timing-metrics", "true") + .header("version", "2026-06-12") + .header("x-codex-window-id", "window-1") + .header("x-codex-parent-thread-id", "parent-1") + .body(Body::from( + json!({ + "model": "gpt-5.5-fast", + "stream": false, + "input": [], + "reasoning": {"effort": "high"}, + "service_tier": "fast", + "tool_choice": { + "type": "function", + "function": {"name": "lookup"} + }, + "parallel_tool_calls": true, + "text": { + "format": { + "type": "json_schema", + "name": "Answer", + "schema": {"type": "object"}, + "strict": true + } + }, + "prompt_cache_key": "pcache", + "include": ["reasoning.encrypted_content"], + "client_metadata": { + "safe": "yes" + }, + "turnState": "turn-body" + }) + .to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(response.status(), StatusCode::OK); + let body = response_json(response).await; + assert_eq!(body["id"], "resp_fields"); +} + +#[tokio::test] +async fn v1_responses_should_include_encrypted_reasoning_by_default() { + let server = MockServer::start().await; + let sse_body = concat!( + "event: response.completed\n", + "data: {\"response\":{\"id\":\"resp_reasoning_include\",\"object\":\"response\",\"usage\":{\"input_tokens\":1,\"output_tokens\":1}}}\n", + "\n", + ); + Mock::given(method("POST")) + .and(path("/codex/responses")) + .and(body_json(json!({ + "model": "gpt-5.5", + "instructions": "", + "input": [], + "stream": true, + "store": false, + "reasoning": { + "effort": "high", + "summary": "auto" + }, + "include": ["reasoning.encrypted_content"] + }))) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(sse_body), + ) + .mount(&server) + .await; + let imported = build_imported_app(server.uri()).await; + + let response = imported + .app + .oneshot( + Request::builder() + .method("POST") + .uri("/v1/responses") + .header( + "authorization", + format!("Bearer {}", imported.client_api_key), + ) + .header("content-type", "application/json") + .body(Body::from( + r#"{"model":"gpt-5.5","stream":false,"input":[],"reasoning":{"effort":"high"}}"#, + )) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(response.status(), StatusCode::OK); + let body = response_json(response).await; + assert_eq!(body["id"], "resp_reasoning_include"); +} + #[tokio::test] async fn v1_responses_should_reconstruct_non_stream_output_text_from_sse_deltas() { let server = MockServer::start().await; @@ -368,7 +540,9 @@ async fn v1_responses_should_reconstruct_non_stream_output_text_from_sse_deltas( format!("Bearer {}", imported.client_api_key), ) .header("content-type", "application/json") - .body(Body::from(r#"{"model":"gpt-5.5","input":[]}"#)) + .body(Body::from( + r#"{"model":"gpt-5.5","input":[],"stream":false}"#, + )) .unwrap(), ) .await @@ -418,7 +592,9 @@ async fn v1_responses_should_use_done_output_items_when_completed_output_is_empt format!("Bearer {}", imported.client_api_key), ) .header("content-type", "application/json") - .body(Body::from(r#"{"model":"gpt-5.5","input":[]}"#)) + .body(Body::from( + r#"{"model":"gpt-5.5","input":[],"stream":false}"#, + )) .unwrap(), ) .await @@ -508,6 +684,65 @@ async fn v1_responses_should_passthrough_stream_and_record_usage_and_log() { assert_eq!(event.4["usage"]["outputTokens"], 5); } +#[tokio::test] +async fn v1_responses_should_default_to_streaming_when_stream_is_omitted() { + let server = MockServer::start().await; + let sse_body = concat!( + "event: response.output_text.delta\n", + "data: {\"delta\":\"pong\"}\n", + "\n", + "event: response.completed\n", + "data: {\"response\":{\"id\":\"resp_default_stream\",\"object\":\"response\",\"usage\":{\"input_tokens\":3,\"output_tokens\":5}}}\n", + "\n", + ); + Mock::given(method("POST")) + .and(path("/codex/responses")) + .and(body_json(json!({ + "model": "gpt-5.5", + "instructions": "", + "input": [], + "stream": true, + "store": false + }))) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(sse_body), + ) + .mount(&server) + .await; + let imported = build_imported_app(server.uri()).await; + + let response = imported + .app + .oneshot( + Request::builder() + .method("POST") + .uri("/v1/responses") + .header( + "authorization", + format!("Bearer {}", imported.client_api_key), + ) + .header("content-type", "application/json") + .body(Body::from(r#"{"model":"gpt-5.5","input":[]}"#)) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(response.status(), StatusCode::OK); + let content_type = response + .headers() + .get("content-type") + .and_then(|value| value.to_str().ok()) + .unwrap_or_default() + .to_string(); + assert!(content_type.starts_with("text/event-stream")); + let body = response_text(response).await; + assert!(body.contains("event: response.output_text.delta")); + assert!(body.contains("event: response.completed")); +} + #[tokio::test] async fn v1_responses_should_refresh_after_401_and_retry_non_stream() { let server = MockServer::start().await; @@ -549,7 +784,9 @@ async fn v1_responses_should_refresh_after_401_and_retry_non_stream() { ) .header("content-type", "application/json") .header("x-request-id", "req_refresh_non_stream") - .body(Body::from(r#"{"model":"gpt-5.5","input":[]}"#)) + .body(Body::from( + r#"{"model":"gpt-5.5","input":[],"stream":false}"#, + )) .unwrap(), ) .await