starred/codex-proxy-rs
1
0
Fork 0
mirror of https://github.com/zyycn/codex-proxy-rs.git synced 2026-11-07 00:05:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add openai gpt parity routes

Browse Source 
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-06-12 11:25:37 +08:00
co-authored by Codex
parent e640c5e9ff
commit b9c30ba407
11 changed files with 2079 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/app.rs
+5 -2
View File
@@ -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))
src/codex/client.rs
+32
View File
@@ -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<Client, reqwest::Error> {
// Codex Desktop 指纹依赖 reqwest/rustls 组合，升级前必须重新验证 TLS 行为。
let builder = Client::builder()
src/codex/types.rs
+61 -3
View File
@@ -13,11 +13,69 @@ pub struct CodexResponsesRequest {
#[serde(skip_serializing_if = "Option::is_none")]
pub tools: Option<Vec<Value>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub tool_choice: Option<Value>,
#[serde(skip_serializing_if = "Option::is_none")]
pub parallel_tool_calls: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub text: Option<Value>,
#[serde(skip_serializing_if = "Option::is_none")]
pub service_tier: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub previous_response_id: Option<String>,
#[serde(default, skip_serializing_if = "is_false")]
#[serde(skip_serializing_if = "Option::is_none")]
pub prompt_cache_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub include: Option<Vec<String>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub client_metadata: Option<Value>,
#[serde(skip)]
pub use_websocket: bool,
#[serde(skip)]
pub turn_state: Option<String>,
#[serde(skip)]
pub turn_metadata: Option<String>,
#[serde(skip)]
pub beta_features: Option<String>,
#[serde(skip)]
pub version: Option<String>,
#[serde(skip)]
pub include_timing_metrics: Option<String>,
#[serde(skip)]
pub codex_window_id: Option<String>,
#[serde(skip)]
pub parent_thread_id: Option<String>,
}
fn is_false(value: &bool) -> bool {
!*value
impl CodexResponsesRequest {
pub fn new_http_sse(
model: impl Into<String>,
instructions: impl Into<String>,
input: Vec<Value>,
) -> 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,
}
}
}
src/http/v1.rs
+347 -15
View File
@@ -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<String>,
reasoning: Option<Value>,
tools: Option<Vec<Value>>,
service_tier: Option<String>,
tool_choice: Option<Value>,
parallel_tool_calls: Option<bool>,
text: Option<Value>,
prompt_cache_key: Option<String>,
include: Option<Vec<String>>,
client_metadata: Option<Value>,
previous_response_id: Option<String>,
stream: Option<bool>,
#[serde(rename = "turnState")]
turn_state: Option<String>,
#[serde(rename = "turnMetadata")]
turn_metadata: Option<String>,
#[serde(rename = "betaFeatures")]
beta_features: Option<String>,
#[serde(rename = "includeTimingMetrics")]
include_timing_metrics: Option<String>,
#[serde(rename = "codexWindowId")]
codex_window_id: Option<String>,
#[serde(rename = "parentThreadId")]
parent_thread_id: Option<String>,
}
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::<ResponsesBody>(&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<AppState>,
Extension(request_id): Extension<RequestId>,
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::<ChatCompletionRequest>(&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<Value>) {
(
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<Value>,
suffix_effort: Option<&str>,
default_effort: Option<&str>,
) -> Option<Value> {
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<String> {
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(),
},
)
src/translation/codex_to_openai.rs
+523
View File
@@ -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<Option<Value>, 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::<Value>(&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<Option<String>, 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::<Value>(&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::<Vec<_>>()
.join("\n\n")
})
.unwrap_or_default()
})
}
fn output_text_from_item(item: &Value) -> Option<String> {
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::<String>();
(!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<String, FunctionCallInfo>,
finished_call_ids: &mut BTreeSet<String>,
) -> Option<Value> {
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<String, FunctionCallInfo>,
finished_call_ids: &mut BTreeSet<String>,
) -> Option<Value> {
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<String, FunctionCallInfo>,
tool_call_indices: &BTreeMap<String, usize>,
) -> 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<String, FunctionCallInfo>,
tool_call_indices: &BTreeMap<String, usize>,
call_ids_with_deltas: &BTreeSet<String>,
) -> 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
}
src/translation/openai_to_codex.rs
+241 -18
View File
@@ -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<ChatMessage>,
#[serde(default)]
pub reasoning_effort: Option<String>,
#[serde(default)]
pub service_tier: Option<String>,
#[serde(default)]
pub tools: Option<Vec<Value>>,
#[serde(default)]
pub tool_choice: Option<Value>,
#[serde(default)]
pub parallel_tool_calls: Option<bool>,
#[serde(default)]
pub functions: Option<Vec<Value>>,
#[serde(default)]
pub response_format: Option<Value>,
#[serde(default)]
pub user: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ChatMessage {
pub role: String,
pub content: String,
#[serde(default)]
pub content: Option<Value>,
#[serde(default)]
pub name: Option<String>,
#[serde(default)]
pub tool_calls: Option<Vec<Value>>,
#[serde(default)]
pub tool_call_id: Option<String>,
#[serde(default)]
pub function_call: Option<Value>,
}
pub fn translate_chat_to_codex(req: ChatCompletionRequest) -> AppResult<CodexResponsesRequest> {
@@ -26,20 +51,218 @@ pub fn translate_chat_to_codex(req: ChatCompletionRequest) -> AppResult<CodexRes
"messages must not be empty".to_string(),
));
}
let input = req
.messages
.into_iter()
.map(|message| json!({ "role": message.role, "content": message.content }))
.collect();
Ok(CodexResponsesRequest {
model: req.model,
instructions: String::new(),
input,
stream: req.stream,
store: false,
reasoning: None,
tools: None,
previous_response_id: None,
use_websocket: false,
})
let instructions = chat_instructions(&req.messages);
let mut input = chat_input(req.messages);
if input.is_empty() {
input.push(json!({"role": "user", "content": ""}));
}
let mut request = CodexResponsesRequest::new_http_sse(req.model, instructions, input);
request.tools = codex_tools(req.tools, req.functions);
request.tool_choice = req.tool_choice;
request.parallel_tool_calls = req.parallel_tool_calls;
request.text = response_format_text(req.response_format);
request.service_tier = req.service_tier;
if let Some(effort) = req.reasoning_effort {
request.reasoning = Some(json!({"effort": effort, "summary": "auto"}));
}
Ok(request)
}
fn chat_instructions(messages: &[ChatMessage]) -> 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::<Vec<_>>()
.join("\n\n");
if instructions.is_empty() {
"You are a helpful assistant.".to_string()
} else {
instructions
}
}
fn chat_input(messages: Vec<ChatMessage>) -> Vec<Value> {
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<Value>, 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<Value>) -> 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::<Vec<_>>()
.join("\n")
}
fn extract_content(content: &Option<Value>) -> 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::<Vec<_>>();
if codex_parts.is_empty() {
Value::String(String::new())
} else {
Value::Array(codex_parts)
}
}
fn codex_content_part(part: &Value) -> Option<Value> {
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<Vec<Value>>, functions: Option<Vec<Value>>) -> Option<Vec<Value>> {
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<Value>) -> Option<Value> {
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,
}
}
tests/chat_completions_route_test.rs
+595
View File
@@ -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::<Value>(frame).unwrap())
.collect::<Vec<_>>();
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::<Value>(frame).unwrap())
.collect::<Vec<_>>();
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<String> {
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()
}
tests/codex_client_test.rs
+7 -11
View File
@@ -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"),
},
)
tests/codex_websocket_test.rs
+12 -11
View File
@@ -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())
}
tests/routes_chat_test.rs
+15 -1
View File
@@ -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();
tests/v1_upstream_route_test.rs
+241 -4
View File
@@ -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