use super::oauth::CodexTokenManager;
use crate::catalog::Provider;
use crate::provider::{LlmResponseStream, StreamingModelProvider, display_name, validate_reasoning};
use crate::provider_connection::DEFAULT_STREAM_IDLE_TIMEOUT;
use crate::providers::http::{SseData, http_client, post_sse, responses_code};
use crate::providers::openai::responses::mappers::{ResponsesRequestPolicy, build_request};
use crate::providers::openai::responses::streaming::decode_responses;
use crate::providers::openai::responses::types::ResponsesRequest;
use crate::providers::response_stream::{OpenedStream, response_stream};
use crate::{Context, LlmModel, Result};
use aether_auth::OAuthCredentialStorage;
use reqwest::header::{AUTHORIZATION, HeaderMap, HeaderValue};
use std::sync::Arc;
use std::time::Duration;
use tracing::debug;
const CODEX_API_BASE: &str = "https://chatgpt.com/backend-api/codex";
const CODEX_CLIENT_VERSION: &str = "0.159.0";
#[derive(Clone)]
pub struct CodexProvider {
base_url: String,
client: reqwest::Client,
model: String,
request_model: Option<String>,
token_manager: Arc<CodexTokenManager>,
idle_timeout: Duration,
}
impl CodexProvider {
pub fn new(store: Arc<dyn OAuthCredentialStorage>) -> Self {
Self::from_token_manager(CodexTokenManager::new(store, super::PROVIDER_ID))
}
pub(crate) fn from_token_manager(token_manager: CodexTokenManager) -> Self {
Self {
base_url: CODEX_API_BASE.to_string(),
client: http_client(),
model: "gpt-5.5".to_string(),
request_model: None,
token_manager: Arc::new(token_manager),
idle_timeout: DEFAULT_STREAM_IDLE_TIMEOUT,
}
}
pub fn with_connection(mut self, connection: crate::ProviderConnectionConfig) -> Self {
if let Some(base_url) = connection.base_url {
self.base_url = base_url.trim_end_matches('/').to_string();
}
self.request_model = connection.request_model;
self.idle_timeout = connection.idle_timeout;
self
}
pub fn with_model(mut self, model: &str) -> Self {
self.model = model.to_string();
self
}
pub async fn forget_cached_token(&self) {
self.token_manager.clear_cache().await;
}
fn build_wire_request(&self, context: &Context) -> Result<ResponsesRequest> {
let identity = self.model();
validate_reasoning(context, identity.as_ref())?;
let wire_model = self.request_model.as_deref().unwrap_or(&self.model);
build_request(identity.as_ref(), wire_model, context, &ResponsesRequestPolicy::CODEX)
}
async fn build_headers(&self) -> Result<HeaderMap> {
let (access_token, account_id) = self.token_manager.get_valid_token().await?;
let mut authorization = HeaderValue::from_str(&format!("Bearer {access_token}"))?;
authorization.set_sensitive(true);
let mut headers = HeaderMap::new();
headers.insert(AUTHORIZATION, authorization);
headers.insert("chatgpt-account-id", HeaderValue::from_str(&account_id)?);
headers.insert("originator", HeaderValue::from_static("codex_cli_rs"));
headers.insert("version", HeaderValue::from_static(CODEX_CLIENT_VERSION));
Ok(headers)
}
async fn send_request(&self, request: ResponsesRequest, headers: HeaderMap) -> Result<OpenedStream<SseData>> {
let url = format!("{}/responses", self.base_url);
debug!("Sending request to Codex API: {url}");
debug!(?request, "Codex request body");
match post_sse(&self.client, &url, headers, &request, responses_code).await {
Ok(opened) => Ok(opened),
Err(error) => {
if error.provider().map(|provider| provider.kind) == Some(crate::ProviderErrorKind::Authentication) {
self.token_manager.clear_cache().await;
}
Err(error)
}
}
}
}
impl StreamingModelProvider for CodexProvider {
fn display_name(&self) -> String {
display_name(Provider::Codex, &self.model)
}
fn model(&self) -> Option<LlmModel> {
LlmModel::parse_for(Provider::Codex, &self.model).ok()
}
fn stream_response(&self, context: &Context) -> LlmResponseStream {
let provider = self.clone();
let context = context.clone();
response_stream(
async move {
let request = provider.build_wire_request(&context)?;
let headers = provider.build_headers().await?;
provider.send_request(request, headers).await
},
decode_responses(),
self.idle_timeout,
)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::ChatMessage;
use crate::LlmError;
use crate::ToolDefinition;
use crate::providers::test_capture_server::{CaptureServer, hello_context};
use aether_auth::{MemoryCredentialStore, OAuthCredential};
use base64::Engine;
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use futures::StreamExt;
#[tokio::test]
async fn all_codex_models_reject_disabled_before_authentication() {
for model in
crate::LlmModel::all().iter().filter(|model| model.provider_enum() == crate::catalog::Provider::Codex)
{
assert!(!model.effective_reasoning_levels().contains(&crate::ReasoningEffort::Disabled));
let provider = CodexProvider::new(Arc::new(MemoryCredentialStore::new())).with_model(&model.model_id());
let mut context = Context::new(vec![], vec![]);
context.set_reasoning_effort(crate::ReasoningEffort::Disabled);
let responses = provider.stream_response(&context).collect::<Vec<_>>().await;
assert_eq!(responses.len(), 1);
assert!(matches!(&responses[0], Err(LlmError::ReasoningValidation(_))));
assert!(!responses[0].as_ref().unwrap_err().is_retryable());
}
}
#[test]
fn context_window_uses_codex_subscription_limit() {
let provider = create_test_provider();
assert_eq!(provider.context_window(), Some(272_000));
}
#[test]
fn display_name_includes_model() {
let provider = create_test_provider();
assert_eq!(provider.display_name(), "Codex (gpt-5.5)");
assert_eq!(create_test_provider().with_model("gpt-custom").display_name(), "Codex (gpt-custom)");
}
#[tokio::test]
async fn stream_response_sends_supported_protocol_version_for_gpt_6_luna() {
let mut server = CaptureServer::start_responses().await;
let provider = server_backed_provider(&server).with_model("gpt-6-luna");
let mut context = Context::new(
vec![ChatMessage::system("You are helpful"), ChatMessage::user("Think harder")],
vec![ToolDefinition::new(
"bash",
"Run a command",
serde_json::from_str(r#"{"type": "object", "properties": {"cmd": {"type": "string"}}}"#).unwrap(),
)],
);
context.set_reasoning_effort(crate::ReasoningEffort::Max);
context.set_prompt_cache_key(Some("session-abc".to_string()));
let responses = provider.stream_response(&context).collect::<Vec<_>>().await;
let captured = server.captured().await;
assert!(responses.iter().all(Result::is_ok), "{responses:?}");
assert_eq!(captured.body["reasoning"]["effort"], "max");
assert!(captured.body["reasoning"].get("context").is_none());
assert_eq!(captured.body["model"], "gpt-6-luna");
assert_eq!(captured.body["instructions"], "You are helpful");
assert_eq!(captured.body["tools"].as_array().unwrap().len(), 1);
assert!(captured.body["tools"][0].get("strict").is_none());
assert_eq!(captured.body["parallel_tool_calls"], true);
assert_eq!(captured.body["input"][0]["role"], "user");
assert_eq!(captured.body["prompt_cache_key"], "session-abc");
assert_eq!(captured.body["store"], false);
assert_eq!(captured.body["stream"], true);
assert_eq!(captured.headers["chatgpt-account-id"], "account-1");
assert_eq!(captured.headers["version"], "0.159.0");
assert_eq!(captured.headers["accept"], "text/event-stream");
assert!(captured.headers.get("x-openai-internal-codex-responses-lite").is_none());
assert!(captured.headers.get("OpenAI-Beta").is_none());
}
#[tokio::test]
async fn stream_response_defaults_to_medium_effort_on_the_wire() {
let mut server = CaptureServer::start_responses().await;
let provider = server_backed_provider(&server);
let context = Context::new(vec![ChatMessage::user("Hello")], vec![]);
let responses = provider.stream_response(&context).collect::<Vec<_>>().await;
let captured = server.captured().await;
assert!(responses.iter().all(Result::is_ok), "{responses:?}");
assert_eq!(captured.body["reasoning"]["effort"], "medium");
assert_eq!(captured.body["reasoning"]["summary"], "auto");
}
#[tokio::test]
async fn request_model_routes_the_request_without_changing_catalog_identity() {
let mut server = CaptureServer::start_responses().await;
let provider = authenticated_provider().with_connection(crate::ProviderConnectionConfig {
base_url: Some(server.base_url.clone()),
request_model: Some("codex-deployment".to_string()),
..Default::default()
});
let responses = provider.stream_response(&hello_context()).collect::<Vec<_>>().await;
let captured = server.captured().await;
assert!(responses.iter().all(Result::is_ok), "{responses:?}");
assert_eq!(captured.body["model"], "codex-deployment");
assert_eq!(provider.model().unwrap().to_string(), "codex:gpt-5.5");
}
#[tokio::test]
async fn a_logout_reaches_a_provider_that_forgets_its_cached_token() {
let mut server = CaptureServer::start_responses().await;
let store = authenticated_store();
let provider = CodexProvider::new(Arc::clone(&store)).with_connection(crate::ProviderConnectionConfig {
base_url: Some(server.base_url.clone()),
..Default::default()
});
let before = provider.stream_response(&hello_context()).collect::<Vec<_>>().await;
server.captured().await;
store.delete("codex").await.unwrap();
provider.forget_cached_token().await;
let after = provider.stream_response(&hello_context()).collect::<Vec<_>>().await;
assert!(before.iter().all(Result::is_ok), "{before:?}");
assert!(matches!(after.as_slice(), [Err(LlmError::OAuthError(_))]), "{after:?}");
}
fn server_backed_provider(server: &CaptureServer) -> CodexProvider {
authenticated_provider().with_connection(crate::ProviderConnectionConfig {
base_url: Some(server.base_url.clone()),
..Default::default()
})
}
fn authenticated_provider() -> CodexProvider {
CodexProvider::new(authenticated_store())
}
fn authenticated_store() -> Arc<dyn OAuthCredentialStorage> {
let credential = OAuthCredential {
client_id: "test".to_string(),
access_token: test_jwt("account-1"),
refresh_token: None,
expires_at: Some(u64::MAX),
};
Arc::new(MemoryCredentialStore::new().with_credential("codex", credential))
}
fn test_jwt(account_id: &str) -> String {
let header = URL_SAFE_NO_PAD.encode(r#"{"alg":"none"}"#);
let payload = URL_SAFE_NO_PAD
.encode(serde_json::json!({"https://api.openai.com/auth": {"chatgpt_account_id": account_id}}).to_string());
format!("{header}.{payload}.signature")
}
fn create_test_provider() -> CodexProvider {
let store: Arc<dyn OAuthCredentialStorage> = Arc::new(MemoryCredentialStore::new());
CodexProvider::new(store).with_model("gpt-5.5")
}
}