GameCult/Epiphany
Archived
Watch
9
0
Fork
You've already forked Epiphany
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Extract Codex auth routes from processor

Browse source
This commit is contained in:
Meta Vens 2026-05-17 21:09:08 +01:00
commit e4a9ab22e1
5 changed files with 910 additions and 890 deletions
Download patch file Download diff file Expand all files Collapse all files

1
notes/fresh-workspace-handoff.md
View file

@ -42,6 +42,7 @@ remembers doctrine; the branch remembers the blade.
- Durable Epiphany state types now live in root crate `epiphany-state-model`. `codex-protocol` re-exports them only so old rollout/API callers keep compiling; do not add new durable `Epiphany*` state structs under `vendor/codex`. `epiphany-core` imports the native model directly and no longer depends on `codex-protocol`. Pressure derives from native `EpiphanyTokenUsageSnapshot`; Codex token telemetry is mapped at the bridge. Codex rollout/event reconstruction now lives in `codex-core::epiphany_rollout`, where the Codex `RolloutItem` / `EventMsg` types belong. Exact retrieval path matching is native to `epiphany-core`, so `epiphany-core` has no direct vendored Codex crate dependency.
- State-update contract proof and job launch/interrupt DTO ownership now live in native `epiphany-core`. The old giant `codex_thread.rs` Epiphany state-update test fixture was cut, and `epiphany-codex-bridge` no longer imports Epiphany job launch request types from `codex_core`; vendored `CodexThread` may call the native mutation/launch law for compatibility, but it must not regain ownership of state-update semantics, heartbeat launch-plan policy, or Epiphany DTOs.
- MCP remains JSON because MCP is a JSON protocol. The latest processor cut only moved MCP refresh/OAuth/status/resource/tool-call route handling into `codex_message_processor/mcp_routes.rs`, reducing the root processor to about 8,977 lines; it did not claim MCP payloads are CultNet or make MCP an Epiphany-owned runtime contract.
- Codex account auth route handling remains in vendored Codex but now lives in `codex_message_processor/auth_routes.rs`: API-key login, ChatGPT browser/device-code/external-token login, logout, auth status, account, rate limits, and add-credits nudge handling. This preserves Codex ownership of subscription auth while cutting the root processor to about 8,094 lines.
- The May 2026 foundation migration in `notes/epiphany-ideal-architecture-rebuild-plan.md` is complete enough to defend. Read `notes/epiphany-architectural-teardown.md` before touching Epiphany control-plane code, but do not treat it as an open-ended excuse to keep shaving the altar. The rebuild moved scene projection, freshness derivation, pressure policy, targeted context shards, bounded graph traversal, job/progress view derivation, planning view derivation, reorientation resume/regather verdict policy, pure CRRC recommendation policy, fixed-lane coordinator decision policy, role-board projection policy, role/reorient result interpretation, role self-persistence review policy, and role/reorient acceptance bundle policy into `epiphany-core`; typed acceptance receipts replaced summary-string identity for live accept paths, runtime links are dual-written on launch, and result read-back now prefers runtime links.
- Read `notes/epiphany-ideal-architecture-rebuild-plan.md` immediately after the teardown. It defines the smallest coherent replacement architecture: durable `EpiphanyState`, runtime-spine-owned `RuntimeState`, separate role memory/heartbeat state, `epiphany-core` as policy owner, app-server as adapter, Aquarium as reflector, typed acceptance receipts, and derived view lenses instead of a protocol verb zoo.
- Do not continue Aquarium UI, bridge, Face, or dogfood expansion until the teardown has a source-grounded cleanup slice plan. Epiphany is the foundation; patches on patches are not a purification rite, they are how the altar becomes load-bearing garbage.

BIN
state/ledgers.msgpack
View file

Binary file not shown.

11
state/map.yaml
View file

File diff suppressed because one or more lines are too long

887
vendor/codex/codex-rs/app-server/src/codex_message_processor.rs vendored
View file

@ -1,8 +1,10 @@
mod auth_routes;
mod epiphany_automation;
mod epiphany_mutation_routes;
mod epiphany_read_routes;
mod mcp_routes;
use self::auth_routes::ActiveLogin;
pub(crate) use self::epiphany_automation::maybe_run_epiphany_coordinator_automation_for_turn_boundary;
pub(crate) use self::epiphany_automation::maybe_run_epiphany_pre_compaction_checkpoint_intervention_for_token_count;
use crate::bespoke_event_handling::apply_bespoke_event_handling;
@ -365,46 +367,8 @@ struct ThreadListFilters {
use_state_db_only: bool,
}
// Duration before a browser ChatGPT login attempt is abandoned.
const LOGIN_CHATGPT_TIMEOUT: Duration = Duration::from_secs(10 * 60);
const LOGIN_ISSUER_OVERRIDE_ENV_VAR: &str = "CODEX_APP_SERVER_LOGIN_ISSUER";
const THREAD_UNLOADING_DELAY: Duration = Duration::from_secs(30 * 60);
enum ActiveLogin {
Browser {
shutdown_handle: ShutdownHandle,
login_id: Uuid,
},
DeviceCode {
cancel: CancellationToken,
login_id: Uuid,
},
}
impl ActiveLogin {
fn login_id(&self) -> Uuid {
match self {
ActiveLogin::Browser { login_id, .. } | ActiveLogin::DeviceCode { login_id, .. } => {
*login_id
}
}
}
fn cancel(&self) {
match self {
ActiveLogin::Browser {
shutdown_handle, ..
} => shutdown_handle.shutdown(),
ActiveLogin::DeviceCode { cancel, .. } => cancel.cancel(),
}
}
}
#[derive(Clone, Copy, Debug)]
enum CancelLoginError {
NotFound,
}
enum ThreadShutdownResult {
Complete,
SubmitFailed,
@ -416,12 +380,6 @@ enum ThreadReadViewError {
Internal(String),
}
impl Drop for ActiveLogin {
fn drop(&mut self) {
self.cancel();
}
}
/// Handles JSON-RPC messages for Codex threads (and legacy conversation APIs).
pub(crate) struct CodexMessageProcessor {
auth_manager: Arc<AuthManager>,
@ -473,13 +431,6 @@ enum EnsureConversationListenerResult {
ConnectionClosed,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum RefreshTokenRequestOutcome {
NotAttemptedOrSucceeded,
FailedTransiently,
FailedPermanently,
}
struct UnloadingState {
delay: Duration,
has_subscribers_rx: watch::Receiver<bool>,
@ -1179,840 +1130,6 @@ impl CodexMessageProcessor {
}
}
async fn login_v2(&self, request_id: ConnectionRequestId, params: LoginAccountParams) {
match params {
LoginAccountParams::ApiKey { api_key } => {
self.login_api_key_v2(request_id, LoginApiKeyParams { api_key })
.await;
}
LoginAccountParams::Chatgpt => {
self.login_chatgpt_v2(request_id).await;
}
LoginAccountParams::ChatgptDeviceCode => {
self.login_chatgpt_device_code_v2(request_id).await;
}
LoginAccountParams::ChatgptAuthTokens {
access_token,
chatgpt_account_id,
chatgpt_plan_type,
} => {
self.login_chatgpt_auth_tokens(
request_id,
access_token,
chatgpt_account_id,
chatgpt_plan_type,
)
.await;
}
}
}
fn external_auth_active_error(&self) -> JSONRPCErrorError {
JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: "External auth is active. Use account/login/start (chatgptAuthTokens) to update it or account/logout to clear it."
.to_string(),
data: None,
}
}
async fn login_api_key_common(
&self,
params: &LoginApiKeyParams,
) -> std::result::Result<(), JSONRPCErrorError> {
if self.auth_manager.is_external_chatgpt_auth_active() {
return Err(self.external_auth_active_error());
}
if matches!(
self.config.forced_login_method,
Some(ForcedLoginMethod::Chatgpt)
) {
return Err(JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: "API key login is disabled. Use ChatGPT login instead.".to_string(),
data: None,
});
}
// Cancel any active login attempt.
{
let mut guard = self.active_login.lock().await;
if let Some(active) = guard.take() {
drop(active);
}
}
match login_with_api_key(
&self.config.codex_home,
&params.api_key,
self.config.cli_auth_credentials_store_mode,
) {
Ok(()) => {
self.auth_manager.reload();
Ok(())
}
Err(err) => Err(JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to save api key: {err}"),
data: None,
}),
}
}
async fn login_api_key_v2(&self, request_id: ConnectionRequestId, params: LoginApiKeyParams) {
match self.login_api_key_common(&params).await {
Ok(()) => {
let response = codex_app_server_protocol::LoginAccountResponse::ApiKey {};
self.outgoing.send_response(request_id, response).await;
let payload_login_completed = AccountLoginCompletedNotification {
login_id: None,
success: true,
error: None,
};
self.outgoing
.send_server_notification(ServerNotification::AccountLoginCompleted(
payload_login_completed,
))
.await;
self.outgoing
.send_server_notification(ServerNotification::AccountUpdated(
self.current_account_updated_notification(),
))
.await;
}
Err(error) => {
self.outgoing.send_error(request_id, error).await;
}
}
}
// Build options for a ChatGPT login attempt; performs validation.
async fn login_chatgpt_common(
&self,
) -> std::result::Result<LoginServerOptions, JSONRPCErrorError> {
let config = self.config.as_ref();
if self.auth_manager.is_external_chatgpt_auth_active() {
return Err(self.external_auth_active_error());
}
if matches!(config.forced_login_method, Some(ForcedLoginMethod::Api)) {
return Err(JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: "ChatGPT login is disabled. Use API key login instead.".to_string(),
data: None,
});
}
let opts = LoginServerOptions {
open_browser: false,
..LoginServerOptions::new(
config.codex_home.to_path_buf(),
CLIENT_ID.to_string(),
config.forced_chatgpt_workspace_id.clone(),
config.cli_auth_credentials_store_mode,
)
};
#[cfg(debug_assertions)]
let opts = {
let mut opts = opts;
if let Ok(issuer) = std::env::var(LOGIN_ISSUER_OVERRIDE_ENV_VAR)
&& !issuer.trim().is_empty()
{
opts.issuer = issuer;
}
opts
};
Ok(opts)
}
fn login_chatgpt_device_code_start_error(err: IoError) -> JSONRPCErrorError {
let is_not_found = err.kind() == std::io::ErrorKind::NotFound;
JSONRPCErrorError {
code: if is_not_found {
INVALID_REQUEST_ERROR_CODE
} else {
INTERNAL_ERROR_CODE
},
message: if is_not_found {
err.to_string()
} else {
format!("failed to request device code: {err}")
},
data: None,
}
}
async fn login_chatgpt_v2(&self, request_id: ConnectionRequestId) {
match self.login_chatgpt_common().await {
Ok(opts) => match run_login_server(opts) {
Ok(server) => {
let login_id = Uuid::new_v4();
let shutdown_handle = server.cancel_handle();
// Replace active login if present.
{
let mut guard = self.active_login.lock().await;
if let Some(existing) = guard.take() {
drop(existing);
}
*guard = Some(ActiveLogin::Browser {
shutdown_handle: shutdown_handle.clone(),
login_id,
});
}
// Spawn background task to monitor completion.
let outgoing_clone = self.outgoing.clone();
let active_login = self.active_login.clone();
let auth_manager = self.auth_manager.clone();
let config_manager = self.config_manager.clone();
let chatgpt_base_url = self.config.chatgpt_base_url.clone();
let auth_url = server.auth_url.clone();
tokio::spawn(async move {
let (success, error_msg) = match tokio::time::timeout(
LOGIN_CHATGPT_TIMEOUT,
server.block_until_done(),
)
.await
{
Ok(Ok(())) => (true, None),
Ok(Err(err)) => (false, Some(format!("Login server error: {err}"))),
Err(_elapsed) => {
shutdown_handle.shutdown();
(false, Some("Login timed out".to_string()))
}
};
let payload_v2 = AccountLoginCompletedNotification {
login_id: Some(login_id.to_string()),
success,
error: error_msg,
};
outgoing_clone
.send_server_notification(ServerNotification::AccountLoginCompleted(
payload_v2,
))
.await;
if success {
auth_manager.reload();
config_manager.replace_cloud_requirements_loader(
auth_manager.clone(),
chatgpt_base_url,
);
config_manager
.sync_default_client_residency_requirement()
.await;
// Notify clients with the actual current auth mode.
let auth = auth_manager.auth_cached();
let payload_v2 = AccountUpdatedNotification {
auth_mode: auth.as_ref().map(CodexAuth::api_auth_mode),
plan_type: auth.as_ref().and_then(CodexAuth::account_plan_type),
};
outgoing_clone
.send_server_notification(ServerNotification::AccountUpdated(
payload_v2,
))
.await;
}
// Clear the active login if it matches this attempt. It may have been replaced or cancelled.
let mut guard = active_login.lock().await;
if guard.as_ref().map(ActiveLogin::login_id) == Some(login_id) {
*guard = None;
}
});
let response = codex_app_server_protocol::LoginAccountResponse::Chatgpt {
login_id: login_id.to_string(),
auth_url,
};
self.outgoing.send_response(request_id, response).await;
}
Err(err) => {
let error = JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to start login server: {err}"),
data: None,
};
self.outgoing.send_error(request_id, error).await;
}
},
Err(err) => {
self.outgoing.send_error(request_id, err).await;
}
}
}
async fn login_chatgpt_device_code_v2(&self, request_id: ConnectionRequestId) {
match self.login_chatgpt_common().await {
Ok(opts) => match request_device_code(&opts).await {
Ok(device_code) => {
let login_id = Uuid::new_v4();
let cancel = CancellationToken::new();
{
let mut guard = self.active_login.lock().await;
if let Some(existing) = guard.take() {
drop(existing);
}
*guard = Some(ActiveLogin::DeviceCode {
cancel: cancel.clone(),
login_id,
});
}
let verification_url = device_code.verification_url.clone();
let user_code = device_code.user_code.clone();
let response =
codex_app_server_protocol::LoginAccountResponse::ChatgptDeviceCode {
login_id: login_id.to_string(),
verification_url,
user_code,
};
self.outgoing.send_response(request_id, response).await;
let outgoing_clone = self.outgoing.clone();
let active_login = self.active_login.clone();
let auth_manager = self.auth_manager.clone();
let config_manager = self.config_manager.clone();
let chatgpt_base_url = self.config.chatgpt_base_url.clone();
tokio::spawn(async move {
let (success, error_msg) = tokio::select! {
_ = cancel.cancelled() => {
(false, Some("Login was not completed".to_string()))
}
r = complete_device_code_login(opts, device_code) => {
match r {
Ok(()) => (true, None),
Err(err) => (false, Some(err.to_string())),
}
}
};
let payload_v2 = AccountLoginCompletedNotification {
login_id: Some(login_id.to_string()),
success,
error: error_msg,
};
outgoing_clone
.send_server_notification(ServerNotification::AccountLoginCompleted(
payload_v2,
))
.await;
if success {
auth_manager.reload();
config_manager.replace_cloud_requirements_loader(
auth_manager.clone(),
chatgpt_base_url,
);
config_manager
.sync_default_client_residency_requirement()
.await;
let auth = auth_manager.auth_cached();
let payload_v2 = AccountUpdatedNotification {
auth_mode: auth.as_ref().map(CodexAuth::api_auth_mode),
plan_type: auth.as_ref().and_then(CodexAuth::account_plan_type),
};
outgoing_clone
.send_server_notification(ServerNotification::AccountUpdated(
payload_v2,
))
.await;
}
let mut guard = active_login.lock().await;
if guard.as_ref().map(ActiveLogin::login_id) == Some(login_id) {
*guard = None;
}
});
}
Err(err) => {
let error = Self::login_chatgpt_device_code_start_error(err);
self.outgoing.send_error(request_id, error).await;
}
},
Err(err) => {
self.outgoing.send_error(request_id, err).await;
}
}
}
async fn cancel_login_chatgpt_common(
&self,
login_id: Uuid,
) -> std::result::Result<(), CancelLoginError> {
let mut guard = self.active_login.lock().await;
if guard.as_ref().map(ActiveLogin::login_id) == Some(login_id) {
if let Some(active) = guard.take() {
drop(active);
}
Ok(())
} else {
Err(CancelLoginError::NotFound)
}
}
async fn cancel_login_v2(
&self,
request_id: ConnectionRequestId,
params: CancelLoginAccountParams,
) {
let login_id = params.login_id;
match Uuid::parse_str(&login_id) {
Ok(uuid) => {
let status = match self.cancel_login_chatgpt_common(uuid).await {
Ok(()) => CancelLoginAccountStatus::Canceled,
Err(CancelLoginError::NotFound) => CancelLoginAccountStatus::NotFound,
};
let response = CancelLoginAccountResponse { status };
self.outgoing.send_response(request_id, response).await;
}
Err(_) => {
let error = JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: format!("invalid login id: {login_id}"),
data: None,
};
self.outgoing.send_error(request_id, error).await;
}
}
}
async fn login_chatgpt_auth_tokens(
&self,
request_id: ConnectionRequestId,
access_token: String,
chatgpt_account_id: String,
chatgpt_plan_type: Option<String>,
) {
if matches!(
self.config.forced_login_method,
Some(ForcedLoginMethod::Api)
) {
let error = JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: "External ChatGPT auth is disabled. Use API key login instead."
.to_string(),
data: None,
};
self.outgoing.send_error(request_id, error).await;
return;
}
// Cancel any active login attempt to avoid persisting managed auth state.
{
let mut guard = self.active_login.lock().await;
if let Some(active) = guard.take() {
drop(active);
}
}
if let Some(expected_workspace) = self.config.forced_chatgpt_workspace_id.as_deref()
&& chatgpt_account_id != expected_workspace
{
let error = JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: format!(
"External auth must use workspace {expected_workspace}, but received {chatgpt_account_id:?}."
),
data: None,
};
self.outgoing.send_error(request_id, error).await;
return;
}
if let Err(err) = login_with_chatgpt_auth_tokens(
&self.config.codex_home,
&access_token,
&chatgpt_account_id,
chatgpt_plan_type.as_deref(),
) {
let error = JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to set external auth: {err}"),
data: None,
};
self.outgoing.send_error(request_id, error).await;
return;
}
self.auth_manager.reload();
self.config_manager.replace_cloud_requirements_loader(
self.auth_manager.clone(),
self.config.chatgpt_base_url.clone(),
);
self.config_manager
.sync_default_client_residency_requirement()
.await;
self.outgoing
.send_response(request_id, LoginAccountResponse::ChatgptAuthTokens {})
.await;
let payload_login_completed = AccountLoginCompletedNotification {
login_id: None,
success: true,
error: None,
};
self.outgoing
.send_server_notification(ServerNotification::AccountLoginCompleted(
payload_login_completed,
))
.await;
self.outgoing
.send_server_notification(ServerNotification::AccountUpdated(
self.current_account_updated_notification(),
))
.await;
}
async fn logout_common(&self) -> std::result::Result<Option<AuthMode>, JSONRPCErrorError> {
// Cancel any active login attempt.
{
let mut guard = self.active_login.lock().await;
if let Some(active) = guard.take() {
drop(active);
}
}
match self.auth_manager.logout_with_revoke().await {
Ok(_) => {}
Err(err) => {
return Err(JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("logout failed: {err}"),
data: None,
});
}
}
// Reflect the current auth method after logout (likely None).
Ok(self
.auth_manager
.auth_cached()
.as_ref()
.map(CodexAuth::api_auth_mode))
}
async fn logout_v2(&self, request_id: ConnectionRequestId) {
match self.logout_common().await {
Ok(current_auth_method) => {
self.outgoing
.send_response(request_id, LogoutAccountResponse {})
.await;
let payload_v2 = AccountUpdatedNotification {
auth_mode: current_auth_method,
plan_type: None,
};
self.outgoing
.send_server_notification(ServerNotification::AccountUpdated(payload_v2))
.await;
}
Err(error) => {
self.outgoing.send_error(request_id, error).await;
}
}
}
async fn refresh_token_if_requested(&self, do_refresh: bool) -> RefreshTokenRequestOutcome {
if self.auth_manager.is_external_chatgpt_auth_active() {
return RefreshTokenRequestOutcome::NotAttemptedOrSucceeded;
}
if do_refresh && let Err(err) = self.auth_manager.refresh_token().await {
let failed_reason = err.failed_reason();
if failed_reason.is_none() {
tracing::warn!("failed to refresh token while getting account: {err}");
return RefreshTokenRequestOutcome::FailedTransiently;
}
return RefreshTokenRequestOutcome::FailedPermanently;
}
RefreshTokenRequestOutcome::NotAttemptedOrSucceeded
}
async fn get_auth_status(&self, request_id: ConnectionRequestId, params: GetAuthStatusParams) {
let include_token = params.include_token.unwrap_or(false);
let do_refresh = params.refresh_token.unwrap_or(false);
self.refresh_token_if_requested(do_refresh).await;
// Determine whether auth is required based on the active model provider.
// If a custom provider is configured with `requires_openai_auth == false`,
// then no auth step is required; otherwise, default to requiring auth.
let requires_openai_auth = self.config.model_provider.requires_openai_auth;
let response = if !requires_openai_auth {
GetAuthStatusResponse {
auth_method: None,
auth_token: None,
requires_openai_auth: Some(false),
}
} else {
let auth = if do_refresh {
self.auth_manager.auth_cached()
} else {
self.auth_manager.auth().await
};
match auth {
Some(auth) => {
let permanent_refresh_failure =
self.auth_manager.refresh_failure_for_auth(&auth).is_some();
let auth_mode = auth.api_auth_mode();
let (reported_auth_method, token_opt) =
if matches!(auth, CodexAuth::AgentIdentity(_))
|| include_token && permanent_refresh_failure
{
(Some(auth_mode), None)
} else {
match auth.get_token() {
Ok(token) if !token.is_empty() => {
let tok = if include_token { Some(token) } else { None };
(Some(auth_mode), tok)
}
Ok(_) => (None, None),
Err(err) => {
tracing::warn!("failed to get token for auth status: {err}");
(None, None)
}
}
};
GetAuthStatusResponse {
auth_method: reported_auth_method,
auth_token: token_opt,
requires_openai_auth: Some(true),
}
}
None => GetAuthStatusResponse {
auth_method: None,
auth_token: None,
requires_openai_auth: Some(true),
},
}
};
self.outgoing.send_response(request_id, response).await;
}
async fn get_account(&self, request_id: ConnectionRequestId, params: GetAccountParams) {
let do_refresh = params.refresh_token;
self.refresh_token_if_requested(do_refresh).await;
// Whether auth is required for the active model provider.
let requires_openai_auth = self.config.model_provider.requires_openai_auth;
if !requires_openai_auth {
let response = GetAccountResponse {
account: None,
requires_openai_auth,
};
self.outgoing.send_response(request_id, response).await;
return;
}
let account = match self.auth_manager.auth_cached() {
Some(auth) => match auth.auth_mode() {
CoreAuthMode::ApiKey => Some(Account::ApiKey {}),
CoreAuthMode::Chatgpt
| CoreAuthMode::ChatgptAuthTokens
| CoreAuthMode::AgentIdentity => {
let email = auth.get_account_email();
let plan_type = auth.account_plan_type();
match (email, plan_type) {
(Some(email), Some(plan_type)) => {
Some(Account::Chatgpt { email, plan_type })
}
_ => {
let error = JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message:
"email and plan type are required for chatgpt authentication"
.to_string(),
data: None,
};
self.outgoing.send_error(request_id, error).await;
return;
}
}
}
},
None => None,
};
let response = GetAccountResponse {
account,
requires_openai_auth,
};
self.outgoing.send_response(request_id, response).await;
}
async fn get_account_rate_limits(&self, request_id: ConnectionRequestId) {
match self.fetch_account_rate_limits().await {
Ok((rate_limits, rate_limits_by_limit_id)) => {
let response = GetAccountRateLimitsResponse {
rate_limits: rate_limits.into(),
rate_limits_by_limit_id: Some(
rate_limits_by_limit_id
.into_iter()
.map(|(limit_id, snapshot)| (limit_id, snapshot.into()))
.collect(),
),
};
self.outgoing.send_response(request_id, response).await;
}
Err(error) => {
self.outgoing.send_error(request_id, error).await;
}
}
}
async fn send_add_credits_nudge_email(
&self,
request_id: ConnectionRequestId,
params: SendAddCreditsNudgeEmailParams,
) {
match self.send_add_credits_nudge_email_inner(params).await {
Ok(status) => {
self.outgoing
.send_response(request_id, SendAddCreditsNudgeEmailResponse { status })
.await;
}
Err(error) => {
self.outgoing.send_error(request_id, error).await;
}
}
}
async fn send_add_credits_nudge_email_inner(
&self,
params: SendAddCreditsNudgeEmailParams,
) -> Result<AddCreditsNudgeEmailStatus, JSONRPCErrorError> {
let Some(auth) = self.auth_manager.auth().await else {
return Err(JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: "codex account authentication required to notify workspace owner"
.to_string(),
data: None,
});
};
if !auth.is_chatgpt_auth() {
return Err(JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: "chatgpt authentication required to notify workspace owner".to_string(),
data: None,
});
}
let client = BackendClient::from_auth(self.config.chatgpt_base_url.clone(), &auth)
.map_err(|err| JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to construct backend client: {err}"),
data: None,
})?;
match client
.send_add_credits_nudge_email(Self::backend_credit_type(params.credit_type))
.await
{
Ok(()) => Ok(AddCreditsNudgeEmailStatus::Sent),
Err(err) if err.status().is_some_and(|status| status.as_u16() == 429) => {
Ok(AddCreditsNudgeEmailStatus::CooldownActive)
}
Err(err) => Err(JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to notify workspace owner: {err}"),
data: None,
}),
}
}
fn backend_credit_type(value: AddCreditsNudgeCreditType) -> BackendAddCreditsNudgeCreditType {
match value {
AddCreditsNudgeCreditType::Credits => BackendAddCreditsNudgeCreditType::Credits,
AddCreditsNudgeCreditType::UsageLimit => BackendAddCreditsNudgeCreditType::UsageLimit,
}
}
async fn fetch_account_rate_limits(
&self,
) -> Result<
(
CoreRateLimitSnapshot,
HashMap<String, CoreRateLimitSnapshot>,
),
JSONRPCErrorError,
> {
let Some(auth) = self.auth_manager.auth().await else {
return Err(JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: "codex account authentication required to read rate limits".to_string(),
data: None,
});
};
if !auth.is_chatgpt_auth() {
return Err(JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: "chatgpt authentication required to read rate limits".to_string(),
data: None,
});
}
let client = BackendClient::from_auth(self.config.chatgpt_base_url.clone(), &auth)
.map_err(|err| JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to construct backend client: {err}"),
data: None,
})?;
let snapshots = client
.get_rate_limits_many()
.await
.map_err(|err| JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to fetch codex rate limits: {err}"),
data: None,
})?;
if snapshots.is_empty() {
return Err(JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: "failed to fetch codex rate limits: no snapshots returned".to_string(),
data: None,
});
}
let rate_limits_by_limit_id: HashMap<String, CoreRateLimitSnapshot> = snapshots
.iter()
.cloned()
.map(|snapshot| {
let limit_id = snapshot
.limit_id
.clone()
.unwrap_or_else(|| "codex".to_string());
(limit_id, snapshot)
})
.collect();
let primary = snapshots
.iter()
.find(|snapshot| snapshot.limit_id.as_deref() == Some("codex"))
.cloned()
.unwrap_or_else(|| snapshots[0].clone());
Ok((primary, rate_limits_by_limit_id))
}
async fn exec_one_off_command(
&self,
request_id: ConnectionRequestId,

901
vendor/codex/codex-rs/app-server/src/codex_message_processor/auth_routes.rs vendored Normal file
View file

@ -0,0 +1,901 @@
use super::*;
// Duration before a browser ChatGPT login attempt is abandoned.
const LOGIN_CHATGPT_TIMEOUT: Duration = Duration::from_secs(10 * 60);
const LOGIN_ISSUER_OVERRIDE_ENV_VAR: &str = "CODEX_APP_SERVER_LOGIN_ISSUER";
pub(super) enum ActiveLogin {
Browser {
shutdown_handle: ShutdownHandle,
login_id: Uuid,
},
DeviceCode {
cancel: CancellationToken,
login_id: Uuid,
},
}
impl ActiveLogin {
fn login_id(&self) -> Uuid {
match self {
ActiveLogin::Browser { login_id, .. } | ActiveLogin::DeviceCode { login_id, .. } => {
*login_id
}
}
}
fn cancel(&self) {
match self {
ActiveLogin::Browser {
shutdown_handle, ..
} => shutdown_handle.shutdown(),
ActiveLogin::DeviceCode { cancel, .. } => cancel.cancel(),
}
}
}
#[derive(Clone, Copy, Debug)]
enum CancelLoginError {
NotFound,
}
impl Drop for ActiveLogin {
fn drop(&mut self) {
self.cancel();
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum RefreshTokenRequestOutcome {
NotAttemptedOrSucceeded,
FailedTransiently,
FailedPermanently,
}
impl CodexMessageProcessor {
pub(super) async fn login_v2(
&self,
request_id: ConnectionRequestId,
params: LoginAccountParams,
) {
match params {
LoginAccountParams::ApiKey { api_key } => {
self.login_api_key_v2(request_id, LoginApiKeyParams { api_key })
.await;
}
LoginAccountParams::Chatgpt => {
self.login_chatgpt_v2(request_id).await;
}
LoginAccountParams::ChatgptDeviceCode => {
self.login_chatgpt_device_code_v2(request_id).await;
}
LoginAccountParams::ChatgptAuthTokens {
access_token,
chatgpt_account_id,
chatgpt_plan_type,
} => {
self.login_chatgpt_auth_tokens(
request_id,
access_token,
chatgpt_account_id,
chatgpt_plan_type,
)
.await;
}
}
}
fn external_auth_active_error(&self) -> JSONRPCErrorError {
JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: "External auth is active. Use account/login/start (chatgptAuthTokens) to update it or account/logout to clear it."
.to_string(),
data: None,
}
}
async fn login_api_key_common(
&self,
params: &LoginApiKeyParams,
) -> std::result::Result<(), JSONRPCErrorError> {
if self.auth_manager.is_external_chatgpt_auth_active() {
return Err(self.external_auth_active_error());
}
if matches!(
self.config.forced_login_method,
Some(ForcedLoginMethod::Chatgpt)
) {
return Err(JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: "API key login is disabled. Use ChatGPT login instead.".to_string(),
data: None,
});
}
// Cancel any active login attempt.
{
let mut guard = self.active_login.lock().await;
if let Some(active) = guard.take() {
drop(active);
}
}
match login_with_api_key(
&self.config.codex_home,
&params.api_key,
self.config.cli_auth_credentials_store_mode,
) {
Ok(()) => {
self.auth_manager.reload();
Ok(())
}
Err(err) => Err(JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to save api key: {err}"),
data: None,
}),
}
}
async fn login_api_key_v2(&self, request_id: ConnectionRequestId, params: LoginApiKeyParams) {
match self.login_api_key_common(&params).await {
Ok(()) => {
let response = codex_app_server_protocol::LoginAccountResponse::ApiKey {};
self.outgoing.send_response(request_id, response).await;
let payload_login_completed = AccountLoginCompletedNotification {
login_id: None,
success: true,
error: None,
};
self.outgoing
.send_server_notification(ServerNotification::AccountLoginCompleted(
payload_login_completed,
))
.await;
self.outgoing
.send_server_notification(ServerNotification::AccountUpdated(
self.current_account_updated_notification(),
))
.await;
}
Err(error) => {
self.outgoing.send_error(request_id, error).await;
}
}
}
// Build options for a ChatGPT login attempt; performs validation.
async fn login_chatgpt_common(
&self,
) -> std::result::Result<LoginServerOptions, JSONRPCErrorError> {
let config = self.config.as_ref();
if self.auth_manager.is_external_chatgpt_auth_active() {
return Err(self.external_auth_active_error());
}
if matches!(config.forced_login_method, Some(ForcedLoginMethod::Api)) {
return Err(JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: "ChatGPT login is disabled. Use API key login instead.".to_string(),
data: None,
});
}
let opts = LoginServerOptions {
open_browser: false,
..LoginServerOptions::new(
config.codex_home.to_path_buf(),
CLIENT_ID.to_string(),
config.forced_chatgpt_workspace_id.clone(),
config.cli_auth_credentials_store_mode,
)
};
#[cfg(debug_assertions)]
let opts = {
let mut opts = opts;
if let Ok(issuer) = std::env::var(LOGIN_ISSUER_OVERRIDE_ENV_VAR)
&& !issuer.trim().is_empty()
{
opts.issuer = issuer;
}
opts
};
Ok(opts)
}
fn login_chatgpt_device_code_start_error(err: IoError) -> JSONRPCErrorError {
let is_not_found = err.kind() == std::io::ErrorKind::NotFound;
JSONRPCErrorError {
code: if is_not_found {
INVALID_REQUEST_ERROR_CODE
} else {
INTERNAL_ERROR_CODE
},
message: if is_not_found {
err.to_string()
} else {
format!("failed to request device code: {err}")
},
data: None,
}
}
async fn login_chatgpt_v2(&self, request_id: ConnectionRequestId) {
match self.login_chatgpt_common().await {
Ok(opts) => match run_login_server(opts) {
Ok(server) => {
let login_id = Uuid::new_v4();
let shutdown_handle = server.cancel_handle();
// Replace active login if present.
{
let mut guard = self.active_login.lock().await;
if let Some(existing) = guard.take() {
drop(existing);
}
*guard = Some(ActiveLogin::Browser {
shutdown_handle: shutdown_handle.clone(),
login_id,
});
}
// Spawn background task to monitor completion.
let outgoing_clone = self.outgoing.clone();
let active_login = self.active_login.clone();
let auth_manager = self.auth_manager.clone();
let config_manager = self.config_manager.clone();
let chatgpt_base_url = self.config.chatgpt_base_url.clone();
let auth_url = server.auth_url.clone();
tokio::spawn(async move {
let (success, error_msg) = match tokio::time::timeout(
LOGIN_CHATGPT_TIMEOUT,
server.block_until_done(),
)
.await
{
Ok(Ok(())) => (true, None),
Ok(Err(err)) => (false, Some(format!("Login server error: {err}"))),
Err(_elapsed) => {
shutdown_handle.shutdown();
(false, Some("Login timed out".to_string()))
}
};
let payload_v2 = AccountLoginCompletedNotification {
login_id: Some(login_id.to_string()),
success,
error: error_msg,
};
outgoing_clone
.send_server_notification(ServerNotification::AccountLoginCompleted(
payload_v2,
))
.await;
if success {
auth_manager.reload();
config_manager.replace_cloud_requirements_loader(
auth_manager.clone(),
chatgpt_base_url,
);
config_manager
.sync_default_client_residency_requirement()
.await;
// Notify clients with the actual current auth mode.
let auth = auth_manager.auth_cached();
let payload_v2 = AccountUpdatedNotification {
auth_mode: auth.as_ref().map(CodexAuth::api_auth_mode),
plan_type: auth.as_ref().and_then(CodexAuth::account_plan_type),
};
outgoing_clone
.send_server_notification(ServerNotification::AccountUpdated(
payload_v2,
))
.await;
}
// Clear the active login if it matches this attempt. It may have been replaced or cancelled.
let mut guard = active_login.lock().await;
if guard.as_ref().map(ActiveLogin::login_id) == Some(login_id) {
*guard = None;
}
});
let response = codex_app_server_protocol::LoginAccountResponse::Chatgpt {
login_id: login_id.to_string(),
auth_url,
};
self.outgoing.send_response(request_id, response).await;
}
Err(err) => {
let error = JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to start login server: {err}"),
data: None,
};
self.outgoing.send_error(request_id, error).await;
}
},
Err(err) => {
self.outgoing.send_error(request_id, err).await;
}
}
}
async fn login_chatgpt_device_code_v2(&self, request_id: ConnectionRequestId) {
match self.login_chatgpt_common().await {
Ok(opts) => match request_device_code(&opts).await {
Ok(device_code) => {
let login_id = Uuid::new_v4();
let cancel = CancellationToken::new();
{
let mut guard = self.active_login.lock().await;
if let Some(existing) = guard.take() {
drop(existing);
}
*guard = Some(ActiveLogin::DeviceCode {
cancel: cancel.clone(),
login_id,
});
}
let verification_url = device_code.verification_url.clone();
let user_code = device_code.user_code.clone();
let response =
codex_app_server_protocol::LoginAccountResponse::ChatgptDeviceCode {
login_id: login_id.to_string(),
verification_url,
user_code,
};
self.outgoing.send_response(request_id, response).await;
let outgoing_clone = self.outgoing.clone();
let active_login = self.active_login.clone();
let auth_manager = self.auth_manager.clone();
let config_manager = self.config_manager.clone();
let chatgpt_base_url = self.config.chatgpt_base_url.clone();
tokio::spawn(async move {
let (success, error_msg) = tokio::select! {
_ = cancel.cancelled() => {
(false, Some("Login was not completed".to_string()))
}
r = complete_device_code_login(opts, device_code) => {
match r {
Ok(()) => (true, None),
Err(err) => (false, Some(err.to_string())),
}
}
};
let payload_v2 = AccountLoginCompletedNotification {
login_id: Some(login_id.to_string()),
success,
error: error_msg,
};
outgoing_clone
.send_server_notification(ServerNotification::AccountLoginCompleted(
payload_v2,
))
.await;
if success {
auth_manager.reload();
config_manager.replace_cloud_requirements_loader(
auth_manager.clone(),
chatgpt_base_url,
);
config_manager
.sync_default_client_residency_requirement()
.await;
let auth = auth_manager.auth_cached();
let payload_v2 = AccountUpdatedNotification {
auth_mode: auth.as_ref().map(CodexAuth::api_auth_mode),
plan_type: auth.as_ref().and_then(CodexAuth::account_plan_type),
};
outgoing_clone
.send_server_notification(ServerNotification::AccountUpdated(
payload_v2,
))
.await;
}
let mut guard = active_login.lock().await;
if guard.as_ref().map(ActiveLogin::login_id) == Some(login_id) {
*guard = None;
}
});
}
Err(err) => {
let error = Self::login_chatgpt_device_code_start_error(err);
self.outgoing.send_error(request_id, error).await;
}
},
Err(err) => {
self.outgoing.send_error(request_id, err).await;
}
}
}
async fn cancel_login_chatgpt_common(
&self,
login_id: Uuid,
) -> std::result::Result<(), CancelLoginError> {
let mut guard = self.active_login.lock().await;
if guard.as_ref().map(ActiveLogin::login_id) == Some(login_id) {
if let Some(active) = guard.take() {
drop(active);
}
Ok(())
} else {
Err(CancelLoginError::NotFound)
}
}
pub(super) async fn cancel_login_v2(
&self,
request_id: ConnectionRequestId,
params: CancelLoginAccountParams,
) {
let login_id = params.login_id;
match Uuid::parse_str(&login_id) {
Ok(uuid) => {
let status = match self.cancel_login_chatgpt_common(uuid).await {
Ok(()) => CancelLoginAccountStatus::Canceled,
Err(CancelLoginError::NotFound) => CancelLoginAccountStatus::NotFound,
};
let response = CancelLoginAccountResponse { status };
self.outgoing.send_response(request_id, response).await;
}
Err(_) => {
let error = JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: format!("invalid login id: {login_id}"),
data: None,
};
self.outgoing.send_error(request_id, error).await;
}
}
}
async fn login_chatgpt_auth_tokens(
&self,
request_id: ConnectionRequestId,
access_token: String,
chatgpt_account_id: String,
chatgpt_plan_type: Option<String>,
) {
if matches!(
self.config.forced_login_method,
Some(ForcedLoginMethod::Api)
) {
let error = JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: "External ChatGPT auth is disabled. Use API key login instead."
.to_string(),
data: None,
};
self.outgoing.send_error(request_id, error).await;
return;
}
// Cancel any active login attempt to avoid persisting managed auth state.
{
let mut guard = self.active_login.lock().await;
if let Some(active) = guard.take() {
drop(active);
}
}
if let Some(expected_workspace) = self.config.forced_chatgpt_workspace_id.as_deref()
&& chatgpt_account_id != expected_workspace
{
let error = JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: format!(
"External auth must use workspace {expected_workspace}, but received {chatgpt_account_id:?}."
),
data: None,
};
self.outgoing.send_error(request_id, error).await;
return;
}
if let Err(err) = login_with_chatgpt_auth_tokens(
&self.config.codex_home,
&access_token,
&chatgpt_account_id,
chatgpt_plan_type.as_deref(),
) {
let error = JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to set external auth: {err}"),
data: None,
};
self.outgoing.send_error(request_id, error).await;
return;
}
self.auth_manager.reload();
self.config_manager.replace_cloud_requirements_loader(
self.auth_manager.clone(),
self.config.chatgpt_base_url.clone(),
);
self.config_manager
.sync_default_client_residency_requirement()
.await;
self.outgoing
.send_response(request_id, LoginAccountResponse::ChatgptAuthTokens {})
.await;
let payload_login_completed = AccountLoginCompletedNotification {
login_id: None,
success: true,
error: None,
};
self.outgoing
.send_server_notification(ServerNotification::AccountLoginCompleted(
payload_login_completed,
))
.await;
self.outgoing
.send_server_notification(ServerNotification::AccountUpdated(
self.current_account_updated_notification(),
))
.await;
}
async fn logout_common(&self) -> std::result::Result<Option<AuthMode>, JSONRPCErrorError> {
// Cancel any active login attempt.
{
let mut guard = self.active_login.lock().await;
if let Some(active) = guard.take() {
drop(active);
}
}
match self.auth_manager.logout_with_revoke().await {
Ok(_) => {}
Err(err) => {
return Err(JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("logout failed: {err}"),
data: None,
});
}
}
// Reflect the current auth method after logout (likely None).
Ok(self
.auth_manager
.auth_cached()
.as_ref()
.map(CodexAuth::api_auth_mode))
}
pub(super) async fn logout_v2(&self, request_id: ConnectionRequestId) {
match self.logout_common().await {
Ok(current_auth_method) => {
self.outgoing
.send_response(request_id, LogoutAccountResponse {})
.await;
let payload_v2 = AccountUpdatedNotification {
auth_mode: current_auth_method,
plan_type: None,
};
self.outgoing
.send_server_notification(ServerNotification::AccountUpdated(payload_v2))
.await;
}
Err(error) => {
self.outgoing.send_error(request_id, error).await;
}
}
}
async fn refresh_token_if_requested(&self, do_refresh: bool) -> RefreshTokenRequestOutcome {
if self.auth_manager.is_external_chatgpt_auth_active() {
return RefreshTokenRequestOutcome::NotAttemptedOrSucceeded;
}
if do_refresh && let Err(err) = self.auth_manager.refresh_token().await {
let failed_reason = err.failed_reason();
if failed_reason.is_none() {
tracing::warn!("failed to refresh token while getting account: {err}");
return RefreshTokenRequestOutcome::FailedTransiently;
}
return RefreshTokenRequestOutcome::FailedPermanently;
}
RefreshTokenRequestOutcome::NotAttemptedOrSucceeded
}
pub(super) async fn get_auth_status(
&self,
request_id: ConnectionRequestId,
params: GetAuthStatusParams,
) {
let include_token = params.include_token.unwrap_or(false);
let do_refresh = params.refresh_token.unwrap_or(false);
self.refresh_token_if_requested(do_refresh).await;
// Determine whether auth is required based on the active model provider.
// If a custom provider is configured with `requires_openai_auth == false`,
// then no auth step is required; otherwise, default to requiring auth.
let requires_openai_auth = self.config.model_provider.requires_openai_auth;
let response = if !requires_openai_auth {
GetAuthStatusResponse {
auth_method: None,
auth_token: None,
requires_openai_auth: Some(false),
}
} else {
let auth = if do_refresh {
self.auth_manager.auth_cached()
} else {
self.auth_manager.auth().await
};
match auth {
Some(auth) => {
let permanent_refresh_failure =
self.auth_manager.refresh_failure_for_auth(&auth).is_some();
let auth_mode = auth.api_auth_mode();
let (reported_auth_method, token_opt) =
if matches!(auth, CodexAuth::AgentIdentity(_))
|| include_token && permanent_refresh_failure
{
(Some(auth_mode), None)
} else {
match auth.get_token() {
Ok(token) if !token.is_empty() => {
let tok = if include_token { Some(token) } else { None };
(Some(auth_mode), tok)
}
Ok(_) => (None, None),
Err(err) => {
tracing::warn!("failed to get token for auth status: {err}");
(None, None)
}
}
};
GetAuthStatusResponse {
auth_method: reported_auth_method,
auth_token: token_opt,
requires_openai_auth: Some(true),
}
}
None => GetAuthStatusResponse {
auth_method: None,
auth_token: None,
requires_openai_auth: Some(true),
},
}
};
self.outgoing.send_response(request_id, response).await;
}
pub(super) async fn get_account(
&self,
request_id: ConnectionRequestId,
params: GetAccountParams,
) {
let do_refresh = params.refresh_token;
self.refresh_token_if_requested(do_refresh).await;
// Whether auth is required for the active model provider.
let requires_openai_auth = self.config.model_provider.requires_openai_auth;
if !requires_openai_auth {
let response = GetAccountResponse {
account: None,
requires_openai_auth,
};
self.outgoing.send_response(request_id, response).await;
return;
}
let account = match self.auth_manager.auth_cached() {
Some(auth) => match auth.auth_mode() {
CoreAuthMode::ApiKey => Some(Account::ApiKey {}),
CoreAuthMode::Chatgpt
| CoreAuthMode::ChatgptAuthTokens
| CoreAuthMode::AgentIdentity => {
let email = auth.get_account_email();
let plan_type = auth.account_plan_type();
match (email, plan_type) {
(Some(email), Some(plan_type)) => {
Some(Account::Chatgpt { email, plan_type })
}
_ => {
let error = JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message:
"email and plan type are required for chatgpt authentication"
.to_string(),
data: None,
};
self.outgoing.send_error(request_id, error).await;
return;
}
}
}
},
None => None,
};
let response = GetAccountResponse {
account,
requires_openai_auth,
};
self.outgoing.send_response(request_id, response).await;
}
pub(super) async fn get_account_rate_limits(&self, request_id: ConnectionRequestId) {
match self.fetch_account_rate_limits().await {
Ok((rate_limits, rate_limits_by_limit_id)) => {
let response = GetAccountRateLimitsResponse {
rate_limits: rate_limits.into(),
rate_limits_by_limit_id: Some(
rate_limits_by_limit_id
.into_iter()
.map(|(limit_id, snapshot)| (limit_id, snapshot.into()))
.collect(),
),
};
self.outgoing.send_response(request_id, response).await;
}
Err(error) => {
self.outgoing.send_error(request_id, error).await;
}
}
}
pub(super) async fn send_add_credits_nudge_email(
&self,
request_id: ConnectionRequestId,
params: SendAddCreditsNudgeEmailParams,
) {
match self.send_add_credits_nudge_email_inner(params).await {
Ok(status) => {
self.outgoing
.send_response(request_id, SendAddCreditsNudgeEmailResponse { status })
.await;
}
Err(error) => {
self.outgoing.send_error(request_id, error).await;
}
}
}
async fn send_add_credits_nudge_email_inner(
&self,
params: SendAddCreditsNudgeEmailParams,
) -> Result<AddCreditsNudgeEmailStatus, JSONRPCErrorError> {
let Some(auth) = self.auth_manager.auth().await else {
return Err(JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: "codex account authentication required to notify workspace owner"
.to_string(),
data: None,
});
};
if !auth.is_chatgpt_auth() {
return Err(JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: "chatgpt authentication required to notify workspace owner".to_string(),
data: None,
});
}
let client = BackendClient::from_auth(self.config.chatgpt_base_url.clone(), &auth)
.map_err(|err| JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to construct backend client: {err}"),
data: None,
})?;
match client
.send_add_credits_nudge_email(Self::backend_credit_type(params.credit_type))
.await
{
Ok(()) => Ok(AddCreditsNudgeEmailStatus::Sent),
Err(err) if err.status().is_some_and(|status| status.as_u16() == 429) => {
Ok(AddCreditsNudgeEmailStatus::CooldownActive)
}
Err(err) => Err(JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to notify workspace owner: {err}"),
data: None,
}),
}
}
fn backend_credit_type(value: AddCreditsNudgeCreditType) -> BackendAddCreditsNudgeCreditType {
match value {
AddCreditsNudgeCreditType::Credits => BackendAddCreditsNudgeCreditType::Credits,
AddCreditsNudgeCreditType::UsageLimit => BackendAddCreditsNudgeCreditType::UsageLimit,
}
}
async fn fetch_account_rate_limits(
&self,
) -> Result<
(
CoreRateLimitSnapshot,
HashMap<String, CoreRateLimitSnapshot>,
),
JSONRPCErrorError,
> {
let Some(auth) = self.auth_manager.auth().await else {
return Err(JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: "codex account authentication required to read rate limits".to_string(),
data: None,
});
};
if !auth.is_chatgpt_auth() {
return Err(JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: "chatgpt authentication required to read rate limits".to_string(),
data: None,
});
}
let client = BackendClient::from_auth(self.config.chatgpt_base_url.clone(), &auth)
.map_err(|err| JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to construct backend client: {err}"),
data: None,
})?;
let snapshots = client
.get_rate_limits_many()
.await
.map_err(|err| JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to fetch codex rate limits: {err}"),
data: None,
})?;
if snapshots.is_empty() {
return Err(JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: "failed to fetch codex rate limits: no snapshots returned".to_string(),
data: None,
});
}
let rate_limits_by_limit_id: HashMap<String, CoreRateLimitSnapshot> = snapshots
.iter()
.cloned()
.map(|snapshot| {
let limit_id = snapshot
.limit_id
.clone()
.unwrap_or_else(|| "codex".to_string());
(limit_id, snapshot)
})
.collect();
let primary = snapshots
.iter()
.find(|snapshot| snapshot.limit_id.as_deref() == Some("codex"))
.cloned()
.unwrap_or_else(|| snapshots[0].clone());
Ok((primary, rate_limits_by_limit_id))
}
}