use crate::{
agent::cancellation::{AgentCancellation, is_run_canceled},
login::OPENAI_CODEX_RELOGIN_GUIDANCE,
model_catalog::ModelCatalogEntry,
providers::{
CODEX_MODELS_URL, CODEX_RESPONSES_URL, HttpRequest, HttpTransport, ProviderEvent,
ProviderRequest,
error::codex_session_expired_provider_error,
openai_stream::{
stream_with_transport_attempt_result,
stream_with_transport_attempt_result_with_attempt_offset,
},
},
};
use serde_json::Value;
use std::{fmt, sync::Arc, time::Duration};
use super::{
CODEX_MODEL_CATALOG_CLIENT_VERSION, Provider,
bodies::codex_responses_body,
catalog::{
extract_chatgpt_account_id, fetch_model_catalog_response_text_cancellable,
parse_codex_model_catalog_response,
},
headers::{codex_model_catalog_headers, codex_sse_headers},
};
type CodexAuthRefresh =
dyn Fn(&AgentCancellation) -> anyhow::Result<CodexRefreshedAuth> + Send + Sync;
#[derive(Clone)]
pub struct OpenAiCodexProvider<T> {
model: String,
access_token: String,
account_id: Option<String>,
transport: T,
auth_refresh: Option<Arc<CodexAuthRefresh>>,
}
pub(crate) struct CodexRefreshedAuth {
pub(crate) access_token: String,
pub(crate) account_id: Option<String>,
}
impl<T: fmt::Debug> fmt::Debug for OpenAiCodexProvider<T> {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("OpenAiCodexProvider")
.field("model", &self.model)
.field("access_token", &"<redacted>")
.field(
"account_id",
&self.account_id.as_ref().map(|_| "<redacted>"),
)
.field("auth_refresh", &self.auth_refresh.is_some())
.field("transport", &self.transport)
.finish()
}
}
impl<T> OpenAiCodexProvider<T> {
pub fn new(
model: impl Into<String>,
access_token: impl Into<String>,
account_id: Option<String>,
transport: T,
) -> Self {
let model = model.into();
Self {
model: normalize_codex_model(&model).to_string(),
access_token: access_token.into(),
account_id,
transport,
auth_refresh: None,
}
}
pub(crate) fn with_auth_refresh(
mut self,
refresh: impl Fn(&AgentCancellation) -> anyhow::Result<CodexRefreshedAuth>
+ Send
+ Sync
+ 'static,
) -> Self {
self.auth_refresh = Some(Arc::new(refresh));
self
}
pub fn build_http_request(&self, request: &ProviderRequest) -> anyhow::Result<HttpRequest> {
self.build_http_request_with_auth(request, &self.access_token, &self.account_id)
}
fn build_http_request_with_auth(
&self,
request: &ProviderRequest,
access_token: &str,
account_id: &Option<String>,
) -> anyhow::Result<HttpRequest> {
let account_id = Self::codex_account_id_for(access_token, account_id)?;
Ok(HttpRequest {
method: "POST".to_string(),
url: CODEX_RESPONSES_URL.to_string(),
headers: codex_sse_headers(access_token, &account_id),
body: codex_responses_body(&self.model, request),
})
}
pub fn build_model_catalog_request(&self) -> anyhow::Result<HttpRequest> {
let account_id = self.codex_account_id()?;
Ok(HttpRequest {
method: "GET".to_string(),
url: format!("{CODEX_MODELS_URL}?client_version={CODEX_MODEL_CATALOG_CLIENT_VERSION}"),
headers: codex_model_catalog_headers(&self.access_token, &account_id),
body: Value::Null,
})
}
pub fn discover_model_catalog(&self) -> anyhow::Result<Vec<ModelCatalogEntry>> {
self.discover_model_catalog_cancellable(&AgentCancellation::default())
}
pub fn discover_model_catalog_cancellable(
&self,
cancellation: &AgentCancellation,
) -> anyhow::Result<Vec<ModelCatalogEntry>> {
let text = fetch_model_catalog_response_text_cancellable(
self.build_model_catalog_request()?,
"openai-codex",
cancellation,
)?;
parse_codex_model_catalog_response(&text)
}
fn codex_account_id(&self) -> anyhow::Result<String> {
Self::codex_account_id_for(&self.access_token, &self.account_id)
}
pub(crate) fn codex_account_id_for(
access_token: &str,
account_id: &Option<String>,
) -> anyhow::Result<String> {
account_id
.clone()
.or_else(|| extract_chatgpt_account_id(access_token).ok())
.ok_or_else(|| anyhow::anyhow!("missing ChatGPT account id for provider 'openai-codex'; import provider-keyed OAuth auth with accountId or a JWT access token containing https://api.openai.com/auth.chatgpt_account_id"))
}
}
pub(crate) fn codex_account_id_for(
access_token: &str,
account_id: &Option<String>,
) -> anyhow::Result<String> {
OpenAiCodexProvider::<()>::codex_account_id_for(access_token, account_id)
}
pub(crate) fn normalize_codex_model(model: &str) -> &str {
match model {
"luna" => "gpt-5.6-luna",
"terra" => "gpt-5.6-terra",
"sol" => "gpt-5.6-sol",
"gpt-5.6" => "gpt-5.6-sol",
_ => model,
}
}
impl<T: HttpTransport + Send + Sync> Provider for OpenAiCodexProvider<T> {
fn stream_cancellable(
&self,
request: ProviderRequest,
cancellation: &AgentCancellation,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
let semantic_progress_timeout = request.semantic_progress_timeout_or_default();
let http_request = self.build_http_request(&request)?;
let result = stream_with_transport_attempt_result(
&self.transport,
http_request,
cancellation,
semantic_progress_timeout,
on_event,
);
match result {
Err(attempt_error)
if !attempt_error.made_semantic_progress
&& codex_session_expired_provider_error(&attempt_error.error) =>
{
self.refresh_and_retry_session_expired_request(
&request,
cancellation,
semantic_progress_timeout,
attempt_error.attempts_used,
on_event,
)
}
Err(attempt_error) => {
let _ = attempt_error.unsafe_recovery_progress;
Err(attempt_error.error)
}
Ok(()) => Ok(()),
}
}
}
impl<T: HttpTransport + Send + Sync> OpenAiCodexProvider<T> {
fn refresh_and_retry_session_expired_request(
&self,
request: &ProviderRequest,
cancellation: &AgentCancellation,
semantic_progress_timeout: Duration,
attempt_offset: usize,
on_event: &mut dyn FnMut(ProviderEvent) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
let Some(refresh) = &self.auth_refresh else {
anyhow::bail!(
"openai-codex session expired and automatic refresh is unavailable; {OPENAI_CODEX_RELOGIN_GUIDANCE}"
);
};
cancellation.check()?;
let refreshed = refresh(cancellation).map_err(|error| {
if is_run_canceled(&error) {
return error;
}
anyhow::anyhow!(
"automatic openai-codex session refresh failed; {OPENAI_CODEX_RELOGIN_GUIDANCE}: {error}"
)
})?;
let retry_request = self.build_http_request_with_auth(
request,
&refreshed.access_token,
&refreshed.account_id,
)?;
stream_with_transport_attempt_result_with_attempt_offset(
&self.transport,
retry_request,
cancellation,
semantic_progress_timeout,
attempt_offset,
on_event,
)
.map_err(|attempt_error| attempt_error.error)
.map_err(|error| {
anyhow::anyhow!(
"automatic openai-codex session refresh/retry failed; {OPENAI_CODEX_RELOGIN_GUIDANCE}: {error}"
)
})
}
}