openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-25 18:28:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

Reapply usage nudge with account refresh dedupe

Browse Source
This commit is contained in:
Richard Lee
2026-04-10 18:42:59 -07:00
parent 58933237cd
commit ef96d7e4d8
82 changed files with 3418 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/README.md
+24 -11
View File
@@ -147,6 +147,7 @@ Example with notification opt-out:
- `thread/unarchive` — move an archived rollout file back into the sessions directory; returns the restored `thread` on success and emits `thread/unarchived`.
- `thread/compact/start` — trigger conversation history compaction for a thread; returns `{}` immediately while progress streams through standard turn/item notifications.
- `thread/shellCommand` — run a user-initiated `!` shell command against a thread; this runs unsandboxed with full access rather than inheriting the thread sandbox policy. Returns `{}` immediately while progress streams through standard turn/item notifications and any active turn receives the formatted output in its message stream.
- `thread/addCreditsNudgeEmail` — ask the backend to notify the workspace owner that the thread hit a usage limit; returns `{}` immediately and emits `account/addCreditsNudgeEmail/completed` with the result.
- `thread/backgroundTerminals/clean` — terminate all running background terminals for a thread (experimental; requires `capabilities.experimentalApi`); returns `{}` when the cleanup request is accepted.
- `thread/rollback` — drop the last N turns from the agent’s in-memory context and persist a rollback marker in the rollout so future resumes see the pruned history; returns the updated `thread` (with `turns` populated) on success.
- `turn/start` — add user input to a thread and begin Codex generation; responds with the initial `turn` object and streams `turn/started`, `item/*`, and `turn/completed` notifications. For `collaborationMode`, `settings.developer_instructions: null` means "use built-in instructions for the selected mode".
@@ -457,6 +458,15 @@ If the thread does not already have an active turn, the server starts a standalo
{ "id": 26, "result": {} }
```
### Example: Notify Workspace Owner About Usage Limit
Use `thread/addCreditsNudgeEmail` when a usage-limit prompt needs to notify the thread's workspace owner. The request returns immediately with `{}`. The final delivery result is emitted separately as `account/addCreditsNudgeEmail/completed` for the same `threadId`.
```json
{ "method": "thread/addCreditsNudgeEmail", "id": 27, "params": { "threadId": "thr_b" } }
{ "id": 27, "result": {} }
```
### Example: Start a turn (send user input)
Turns attach user input (text or images) to a thread and trigger Codex generation. The `input` field is a list of discriminated unions:
@@ -1357,19 +1367,19 @@ The JSON-RPC auth/account surface exposes request/response methods plus server-i
### Authentication modes
Codex supports these authentication modes. The current mode is surfaced in `account/updated` (`authMode`), which also includes the current ChatGPT `planType` when available, and can be inferred from `account/read`.
Codex supports these authentication modes. The current mode is surfaced in `account/updated` (`authMode`), which also includes the current ChatGPT `planType` when available, the current `workspaceRole` when live account metadata can be fetched, and the derived `isWorkspaceOwner` flag. The same cached account state can be read from `account/read`.
- **API key (`apiKey`)**: Caller supplies an OpenAI API key via `account/login/start` with `type: "apiKey"`. The API key is saved and used for API requests.
- **ChatGPT managed (`chatgpt`)** (recommended): Codex owns the ChatGPT OAuth flow and refresh tokens. Start via `account/login/start` with `type: "chatgpt"` for the browser flow or `type: "chatgptDeviceCode"` for device code; Codex persists tokens to disk and refreshes them automatically.
### API Overview
- `account/read` — fetch current account info; optionally refresh tokens.
- `account/read` — fetch cached current account info; optionally refresh tokens. ChatGPT workspace role is refreshed in the background and delivered by `account/updated` so this request does not wait for live account metadata.
- `account/login/start` — begin login (`apiKey`, `chatgpt`, `chatgptDeviceCode`).
- `account/login/completed` (notify) — emitted when a login attempt finishes (success or error).
- `account/login/cancel` — cancel a pending managed ChatGPT login by `loginId`.
- `account/logout` — sign out; triggers `account/updated`.
- `account/updated` (notify) — emitted whenever auth mode changes (`authMode`: `apikey`, `chatgpt`, or `null`) and includes the current ChatGPT `planType` when available.
- `account/updated` (notify) — emitted whenever auth mode changes (`authMode`: `apikey`, `chatgpt`, or `null`) and includes the current ChatGPT `planType`, `workspaceRole`, and `isWorkspaceOwner`.
- `account/rateLimits/read` — fetch ChatGPT rate limits; updates arrive via `account/rateLimits/updated` (notify).
- `account/rateLimits/updated` (notify) — emitted whenever a user's ChatGPT rate limits change.
- `mcpServer/oauthLogin/completed` (notify) — emitted after a `mcpServer/oauth/login` flow finishes for a server; payload includes `{ name, success, error? }`.
@@ -1386,16 +1396,19 @@ Request:
Response examples:
```json
{ "id": 1, "result": { "account": null, "requiresOpenaiAuth": false } } // No OpenAI auth needed (e.g., OSS/local models)
{ "id": 1, "result": { "account": null, "requiresOpenaiAuth": true } } // OpenAI auth required (typical for OpenAI-hosted models)
{ "id": 1, "result": { "account": { "type": "apiKey" }, "requiresOpenaiAuth": true } }
{ "id": 1, "result": { "account": { "type": "chatgpt", "email": "user@example.com", "planType": "pro" }, "requiresOpenaiAuth": true } }
{ "id": 1, "result": { "account": null, "workspaceRole": null, "isWorkspaceOwner": null, "requiresOpenaiAuth": false } } // No OpenAI auth needed (e.g., OSS/local models)
{ "id": 1, "result": { "account": null, "workspaceRole": null, "isWorkspaceOwner": null, "requiresOpenaiAuth": true } } // OpenAI auth required (typical for OpenAI-hosted models)
{ "id": 1, "result": { "account": { "type": "apiKey" }, "workspaceRole": null, "isWorkspaceOwner": null, "requiresOpenaiAuth": true } }
{ "id": 1, "result": { "account": { "type": "chatgpt", "email": "user@example.com", "planType": "pro" }, "workspaceRole": null, "isWorkspaceOwner": true, "requiresOpenaiAuth": true } }
{ "method": "account/updated", "params": { "authMode": "chatgpt", "planType": "pro", "workspaceRole": "account-admin", "isWorkspaceOwner": true } } // emitted when live workspace metadata arrives
```
Field notes:
- `refreshToken` (bool): set `true` to force a token refresh.
- `requiresOpenaiAuth` reflects the active provider; when `false`, Codex can run without OpenAI credentials.
- `workspaceRole` is a nullable live account role from ChatGPT account metadata: `account-owner`, `account-admin`, or `standard-user`. It is normally delivered asynchronously in `account/updated`; clients should treat `null` from `account/read` as "not known yet".
- `isWorkspaceOwner` is a nullable convenience flag. When `workspaceRole` is available, owners and admins map to `true` and standard users map to `false`; otherwise the server may fall back to the managed-account token's workspace-owner claim.
### 2) Log in with an API key
@@ -1414,7 +1427,7 @@ Field notes:
3. Notifications:
```json
{ "method": "account/login/completed", "params": { "loginId": null, "success": true, "error": null } }
{ "method": "account/updated", "params": { "authMode": "apikey", "planType": null } }
{ "method": "account/updated", "params": { "authMode": "apikey", "planType": null, "workspaceRole": null, "isWorkspaceOwner": null } }
```
### 3) Log in with ChatGPT (browser flow)
@@ -1428,7 +1441,7 @@ Field notes:
3. Wait for notifications:
```json
{ "method": "account/login/completed", "params": { "loginId": "<uuid>", "success": true, "error": null } }
{ "method": "account/updated", "params": { "authMode": "chatgpt", "planType": "plus" } }
{ "method": "account/updated", "params": { "authMode": "chatgpt", "planType": "plus", "workspaceRole": "account-owner", "isWorkspaceOwner": true } }
```
### 4) Log in with ChatGPT (device code flow)
@@ -1442,7 +1455,7 @@ Field notes:
3. Wait for notifications:
```json
{ "method": "account/login/completed", "params": { "loginId": "<uuid>", "success": true, "error": null } }
{ "method": "account/updated", "params": { "authMode": "chatgpt", "planType": "plus" } }
{ "method": "account/updated", "params": { "authMode": "chatgpt", "planType": "plus", "workspaceRole": "account-owner", "isWorkspaceOwner": true } }
```
### 5) Cancel a ChatGPT login
@@ -1457,7 +1470,7 @@ Field notes:
```json
{ "method": "account/logout", "id": 6 }
{ "id": 6, "result": {} }
{ "method": "account/updated", "params": { "authMode": null, "planType": null } }
{ "method": "account/updated", "params": { "authMode": null, "planType": null, "workspaceRole": null, "isWorkspaceOwner": null } }
```
### 7) Rate limits (ChatGPT)
codex-rs/app-server/src/bespoke_event_handling.rs
+24
View File
@@ -13,6 +13,8 @@ use crate::thread_state::resolve_server_request_on_thread_listener;
use crate::thread_status::ThreadWatchActiveGuard;
use crate::thread_status::ThreadWatchManager;
use codex_app_server_protocol::AccountRateLimitsUpdatedNotification;
use codex_app_server_protocol::AddCreditsNudgeEmailNotification;
use codex_app_server_protocol::AddCreditsNudgeEmailResult;
use codex_app_server_protocol::AdditionalPermissionProfile as V2AdditionalPermissionProfile;
use codex_app_server_protocol::AgentMessageDeltaNotification;
use codex_app_server_protocol::ApplyPatchApprovalParams;
@@ -117,6 +119,7 @@ use codex_protocol::dynamic_tools::DynamicToolCallOutputContentItem as CoreDynam
use codex_protocol::dynamic_tools::DynamicToolResponse as CoreDynamicToolResponse;
use codex_protocol::items::parse_hook_prompt_message;
use codex_protocol::plan_tool::UpdatePlanArgs;
use codex_protocol::protocol::AddCreditsNudgeEmailStatus as CoreAddCreditsNudgeEmailStatus;
use codex_protocol::protocol::CodexErrorInfo as CoreCodexErrorInfo;
use codex_protocol::protocol::Event;
use codex_protocol::protocol::EventMsg;
@@ -223,6 +226,26 @@ pub(crate) async fn apply_bespoke_event_handling(
)
.await;
}
EventMsg::AddCreditsNudgeEmailResponse(event) => {
if let ApiVersion::V2 = api_version {
let result = match event.result {
Ok(CoreAddCreditsNudgeEmailStatus::Sent) => AddCreditsNudgeEmailResult::Sent,
Ok(CoreAddCreditsNudgeEmailStatus::CooldownActive) => {
AddCreditsNudgeEmailResult::CooldownActive
}
Err(message) => AddCreditsNudgeEmailResult::Failed { message },
};
let notification = AddCreditsNudgeEmailNotification {
thread_id: conversation_id.to_string(),
result,
};
outgoing
.send_server_notification(ServerNotification::AddCreditsNudgeEmailCompleted(
notification,
))
.await;
}
}
EventMsg::SkillsUpdateAvailable => {
if let ApiVersion::V2 = api_version {
outgoing
@@ -4039,6 +4062,7 @@ mod tests {
unlimited: false,
balance: Some("5".to_string()),
}),
spend_control: None,
plan_type: None,
};
codex-rs/app-server/src/codex_message_processor.rs
+263 -16
View File
@@ -116,6 +116,8 @@ use codex_app_server_protocol::SkillsConfigWriteResponse;
use codex_app_server_protocol::SkillsListParams;
use codex_app_server_protocol::SkillsListResponse;
use codex_app_server_protocol::Thread;
use codex_app_server_protocol::ThreadAddCreditsNudgeEmailParams;
use codex_app_server_protocol::ThreadAddCreditsNudgeEmailResponse;
use codex_app_server_protocol::ThreadArchiveParams;
use codex_app_server_protocol::ThreadArchiveResponse;
use codex_app_server_protocol::ThreadArchivedNotification;
@@ -184,6 +186,7 @@ use codex_app_server_protocol::WindowsSandboxSetupCompletedNotification;
use codex_app_server_protocol::WindowsSandboxSetupMode;
use codex_app_server_protocol::WindowsSandboxSetupStartParams;
use codex_app_server_protocol::WindowsSandboxSetupStartResponse;
use codex_app_server_protocol::WorkspaceRole;
use codex_app_server_protocol::build_turns_from_rollout_items;
use codex_arg0::Arg0DispatchPaths;
use codex_backend_client::Client as BackendClient;
@@ -200,6 +203,7 @@ use codex_core::SteerInputError;
use codex_core::ThreadConfigSnapshot;
use codex_core::ThreadManager;
use codex_core::ThreadSortKey as CoreThreadSortKey;
use codex_core::WorkspaceRole as CoreWorkspaceRole;
use codex_core::append_thread_name;
use codex_core::config::Config;
use codex_core::config::ConfigOverrides;
@@ -321,6 +325,7 @@ use std::sync::RwLock;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
use std::time::Duration;
use std::time::Instant;
use std::time::SystemTime;
use tokio::sync::Mutex;
use tokio::sync::broadcast;
@@ -363,6 +368,7 @@ struct ThreadListFilters {
const LOGIN_CHATGPT_TIMEOUT: Duration = Duration::from_secs(10 * 60);
const LOGIN_ISSUER_OVERRIDE_ENV_VAR: &str = "CODEX_APP_SERVER_LOGIN_ISSUER";
const APP_LIST_LOAD_TIMEOUT: Duration = Duration::from_secs(90);
const WORKSPACE_ROLE_REFRESH_RETRY_COOLDOWN: Duration = Duration::from_secs(60);
enum ActiveLogin {
Browser {
@@ -434,6 +440,7 @@ pub(crate) struct CodexMessageProcessor {
command_exec_manager: CommandExecManager,
pending_fuzzy_searches: Arc<Mutex<HashMap<String, Arc<AtomicBool>>>>,
fuzzy_search_sessions: Arc<Mutex<HashMap<String, FuzzyFileSearchSession>>>,
workspace_role_refresh_state: Arc<Mutex<WorkspaceRoleRefreshState>>,
background_tasks: TaskTracker,
feedback: CodexFeedback,
log_db: Option<LogDbLayer>,
@@ -486,6 +493,156 @@ pub(crate) struct CodexMessageProcessorArgs {
pub(crate) log_db: Option<LogDbLayer>,
}
async fn resolve_workspace_role_and_owner_for_auth(
chatgpt_base_url: &str,
auth: Option<&CodexAuth>,
) -> (Option<WorkspaceRole>, Option<bool>) {
let ownership =
codex_core::resolve_workspace_role_and_owner_for_auth(chatgpt_base_url, auth).await;
(
ownership.workspace_role.map(workspace_role_to_v2),
ownership.is_workspace_owner,
)
}
fn workspace_role_to_v2(role: CoreWorkspaceRole) -> WorkspaceRole {
match role {
CoreWorkspaceRole::AccountOwner => WorkspaceRole::AccountOwner,
CoreWorkspaceRole::AccountAdmin => WorkspaceRole::AccountAdmin,
CoreWorkspaceRole::StandardUser => WorkspaceRole::StandardUser,
}
}
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
struct AuthIdentity {
account_id: Option<String>,
account_email: Option<String>,
chatgpt_user_id: Option<String>,
}
fn auth_identity(auth: &CodexAuth) -> AuthIdentity {
AuthIdentity {
account_id: auth.get_account_id(),
account_email: auth.get_account_email(),
chatgpt_user_id: auth.get_chatgpt_user_id(),
}
}
#[derive(Default)]
struct WorkspaceRoleRefreshState {
in_flight: HashSet<AuthIdentity>,
last_attempt_at: HashMap<AuthIdentity, Instant>,
last_payload_by_identity: HashMap<AuthIdentity, AccountUpdatedNotification>,
}
impl WorkspaceRoleRefreshState {
fn try_start(&mut self, identity: &AuthIdentity) -> bool {
if self.last_payload_by_identity.contains_key(identity) {
return false;
}
if let Some(last_attempt) = self.last_attempt_at.get(identity)
&& last_attempt.elapsed() < WORKSPACE_ROLE_REFRESH_RETRY_COOLDOWN
{
return false;
}
if !self.in_flight.insert(identity.clone()) {
return false;
}
self.last_attempt_at
.insert(identity.clone(), Instant::now());
true
}
fn finish(&mut self, identity: &AuthIdentity, payload: AccountUpdatedNotification) -> bool {
self.in_flight.remove(identity);
if self
.last_payload_by_identity
.get(identity)
.is_some_and(|cached_payload| cached_payload == &payload)
{
return false;
}
self.last_payload_by_identity
.insert(identity.clone(), payload);
true
}
fn finish_without_payload(&mut self, identity: &AuthIdentity) {
self.in_flight.remove(identity);
}
}
fn cached_workspace_role_and_owner_for_auth(
auth: Option<&CodexAuth>,
) -> (Option<WorkspaceRole>, Option<bool>) {
(None, auth.and_then(CodexAuth::is_workspace_owner))
}
fn account_updated_notification_for_auth(
auth: Option<&CodexAuth>,
workspace_role: Option<WorkspaceRole>,
is_workspace_owner: Option<bool>,
) -> AccountUpdatedNotification {
AccountUpdatedNotification {
auth_mode: auth.map(CodexAuth::api_auth_mode),
plan_type: auth.and_then(CodexAuth::account_plan_type),
workspace_role,
is_workspace_owner,
}
}
fn spawn_live_workspace_role_update_for_auth(
outgoing: Arc<OutgoingMessageSender>,
auth_manager: Arc<AuthManager>,
chatgpt_base_url: String,
workspace_role_refresh_state: Arc<Mutex<WorkspaceRoleRefreshState>>,
auth: Option<CodexAuth>,
) {
let Some(auth) = auth.filter(CodexAuth::is_chatgpt_auth) else {
return;
};
let expected_identity = auth_identity(&auth);
tokio::spawn(async move {
{
let mut state = workspace_role_refresh_state.lock().await;
if !state.try_start(&expected_identity) {
return;
}
}
let (workspace_role, is_workspace_owner) =
resolve_workspace_role_and_owner_for_auth(&chatgpt_base_url, Some(&auth)).await;
let Some(workspace_role) = workspace_role else {
let mut state = workspace_role_refresh_state.lock().await;
state.finish_without_payload(&expected_identity);
return;
};
let current_auth = auth_manager.auth_cached();
if current_auth.as_ref().map(auth_identity) != Some(expected_identity.clone()) {
let mut state = workspace_role_refresh_state.lock().await;
state.finish_without_payload(&expected_identity);
return;
}
let payload = account_updated_notification_for_auth(
current_auth.as_ref(),
Some(workspace_role),
is_workspace_owner,
);
{
let mut state = workspace_role_refresh_state.lock().await;
if !state.finish(&expected_identity, payload.clone()) {
return;
}
}
outgoing
.send_server_notification(ServerNotification::AccountUpdated(payload))
.await;
});
}
impl CodexMessageProcessor {
pub(crate) fn handle_config_mutation(&self) {
self.clear_plugin_related_caches();
@@ -498,10 +655,19 @@ impl CodexMessageProcessor {
fn current_account_updated_notification(&self) -> AccountUpdatedNotification {
let auth = self.auth_manager.auth_cached();
AccountUpdatedNotification {
auth_mode: auth.as_ref().map(CodexAuth::api_auth_mode),
plan_type: auth.as_ref().and_then(CodexAuth::account_plan_type),
}
let (workspace_role, is_workspace_owner) =
cached_workspace_role_and_owner_for_auth(auth.as_ref());
account_updated_notification_for_auth(auth.as_ref(), workspace_role, is_workspace_owner)
}
fn spawn_live_workspace_role_update(&self, auth: Option<CodexAuth>) {
spawn_live_workspace_role_update_for_auth(
self.outgoing.clone(),
self.auth_manager.clone(),
self.config.chatgpt_base_url.clone(),
self.workspace_role_refresh_state.clone(),
auth,
);
}
async fn load_thread(
@@ -558,6 +724,9 @@ impl CodexMessageProcessor {
command_exec_manager: CommandExecManager::default(),
pending_fuzzy_searches: Arc::new(Mutex::new(HashMap::new())),
fuzzy_search_sessions: Arc::new(Mutex::new(HashMap::new())),
workspace_role_refresh_state: Arc::new(
Mutex::new(WorkspaceRoleRefreshState::default()),
),
background_tasks: TaskTracker::new(),
feedback,
log_db,
@@ -785,6 +954,10 @@ impl CodexMessageProcessor {
self.thread_shell_command(to_connection_request_id(request_id), params)
.await;
}
ClientRequest::ThreadAddCreditsNudgeEmail { request_id, params } => {
self.thread_add_credits_nudge_email(to_connection_request_id(request_id), params)
.await;
}
ClientRequest::SkillsList { request_id, params } => {
self.skills_list(to_connection_request_id(request_id), params)
.await;
@@ -1104,6 +1277,7 @@ impl CodexMessageProcessor {
self.current_account_updated_notification(),
))
.await;
self.spawn_live_workspace_role_update(self.auth_manager.auth_cached());
}
Err(error) => {
self.outgoing.send_error(request_id, error).await;
@@ -1194,6 +1368,7 @@ impl CodexMessageProcessor {
let auth_manager = self.auth_manager.clone();
let cloud_requirements = self.cloud_requirements.clone();
let chatgpt_base_url = self.config.chatgpt_base_url.clone();
let workspace_role_refresh_state = self.workspace_role_refresh_state.clone();
let codex_home = self.config.codex_home.clone();
let cli_overrides = self.current_cli_overrides();
let auth_url = server.auth_url.clone();
@@ -1228,7 +1403,7 @@ impl CodexMessageProcessor {
replace_cloud_requirements_loader(
cloud_requirements.as_ref(),
auth_manager.clone(),
chatgpt_base_url,
chatgpt_base_url.clone(),
codex_home,
);
sync_default_client_residency_requirement(
@@ -1237,17 +1412,28 @@ impl CodexMessageProcessor {
)
.await;
// Notify clients with the actual current auth mode.
// Notify clients with the actual current auth mode immediately; the
// live workspace role is fetched below without delaying this update.
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),
};
let (workspace_role, is_workspace_owner) =
cached_workspace_role_and_owner_for_auth(auth.as_ref());
let payload_v2 = account_updated_notification_for_auth(
auth.as_ref(),
workspace_role,
is_workspace_owner,
);
outgoing_clone
.send_server_notification(ServerNotification::AccountUpdated(
payload_v2,
))
.await;
spawn_live_workspace_role_update_for_auth(
outgoing_clone.clone(),
auth_manager.clone(),
chatgpt_base_url.clone(),
workspace_role_refresh_state.clone(),
auth,
);
}
// Clear the active login if it matches this attempt. It may have been replaced or cancelled.
@@ -1311,6 +1497,7 @@ impl CodexMessageProcessor {
let auth_manager = self.auth_manager.clone();
let cloud_requirements = self.cloud_requirements.clone();
let chatgpt_base_url = self.config.chatgpt_base_url.clone();
let workspace_role_refresh_state = self.workspace_role_refresh_state.clone();
let codex_home = self.config.codex_home.clone();
let cli_overrides = self.current_cli_overrides();
tokio::spawn(async move {
@@ -1342,7 +1529,7 @@ impl CodexMessageProcessor {
replace_cloud_requirements_loader(
cloud_requirements.as_ref(),
auth_manager.clone(),
chatgpt_base_url,
chatgpt_base_url.clone(),
codex_home,
);
sync_default_client_residency_requirement(
@@ -1352,15 +1539,25 @@ impl CodexMessageProcessor {
.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),
};
let (workspace_role, is_workspace_owner) =
cached_workspace_role_and_owner_for_auth(auth.as_ref());
let payload_v2 = account_updated_notification_for_auth(
auth.as_ref(),
workspace_role,
is_workspace_owner,
);
outgoing_clone
.send_server_notification(ServerNotification::AccountUpdated(
payload_v2,
))
.await;
spawn_live_workspace_role_update_for_auth(
outgoing_clone.clone(),
auth_manager.clone(),
chatgpt_base_url.clone(),
workspace_role_refresh_state.clone(),
auth,
);
}
let mut guard = active_login.lock().await;
@@ -1509,6 +1706,7 @@ impl CodexMessageProcessor {
self.current_account_updated_notification(),
))
.await;
self.spawn_live_workspace_role_update(self.auth_manager.auth_cached());
}
async fn logout_common(&self) -> std::result::Result<Option<AuthMode>, JSONRPCErrorError> {
@@ -1546,6 +1744,8 @@ impl CodexMessageProcessor {
let payload_v2 = AccountUpdatedNotification {
auth_mode: current_auth_method,
plan_type: None,
workspace_role: None,
is_workspace_owner: None,
};
self.outgoing
.send_server_notification(ServerNotification::AccountUpdated(payload_v2))
@@ -1644,13 +1844,16 @@ impl CodexMessageProcessor {
if !requires_openai_auth {
let response = GetAccountResponse {
account: None,
workspace_role: None,
is_workspace_owner: None,
requires_openai_auth,
};
self.outgoing.send_response(request_id, response).await;
return;
}
let account = match self.auth_manager.auth_cached() {
let auth = self.auth_manager.auth_cached();
let account = match auth.as_ref() {
Some(auth) => match auth.auth_mode() {
CoreAuthMode::ApiKey => Some(Account::ApiKey {}),
CoreAuthMode::Chatgpt | CoreAuthMode::ChatgptAuthTokens => {
@@ -1677,12 +1880,17 @@ impl CodexMessageProcessor {
},
None => None,
};
let (workspace_role, is_workspace_owner) =
cached_workspace_role_and_owner_for_auth(auth.as_ref());
let response = GetAccountResponse {
account,
workspace_role,
is_workspace_owner,
requires_openai_auth,
};
self.outgoing.send_response(request_id, response).await;
self.spawn_live_workspace_role_update(auth);
}
async fn get_account_rate_limits(&self, request_id: ConnectionRequestId) {
@@ -1700,6 +1908,11 @@ impl CodexMessageProcessor {
self.outgoing.send_response(request_id, response).await;
}
Err(error) => {
tracing::warn!(
?request_id,
error = %error.message,
"account/rateLimits/read request failed"
);
self.outgoing.send_error(request_id, error).await;
}
}
@@ -3409,6 +3622,40 @@ impl CodexMessageProcessor {
}
}
async fn thread_add_credits_nudge_email(
&self,
request_id: ConnectionRequestId,
params: ThreadAddCreditsNudgeEmailParams,
) {
let ThreadAddCreditsNudgeEmailParams { thread_id } = params;
let (_, thread) = match self.load_thread(&thread_id).await {
Ok(v) => v,
Err(error) => {
self.outgoing.send_error(request_id, error).await;
return;
}
};
match self
.submit_core_op(&request_id, thread.as_ref(), Op::SendAddCreditsNudgeEmail)
.await
{
Ok(_) => {
self.outgoing
.send_response(request_id, ThreadAddCreditsNudgeEmailResponse {})
.await;
}
Err(err) => {
self.send_internal_error(
request_id,
format!("failed to request add-credits nudge email: {err}"),
)
.await;
}
}
}
async fn thread_list(&self, request_id: ConnectionRequestId, params: ThreadListParams) {
let ThreadListParams {
cursor,
codex-rs/app-server/src/outgoing_message.rs
+7 -1
View File
@@ -735,6 +735,7 @@ mod tests {
}),
secondary: None,
credits: None,
spend_control: None,
plan_type: Some(PlanType::Plus),
},
});
@@ -754,6 +755,7 @@ mod tests {
},
"secondary": null,
"credits": null,
"spendControl": null,
"planType": "plus"
}
},
@@ -769,6 +771,8 @@ mod tests {
let notification = ServerNotification::AccountUpdated(AccountUpdatedNotification {
auth_mode: Some(AuthMode::ApiKey),
plan_type: None,
workspace_role: None,
is_workspace_owner: None,
});
let jsonrpc_notification = OutgoingMessage::AppServerNotification(notification);
@@ -777,7 +781,9 @@ mod tests {
"method": "account/updated",
"params": {
"authMode": "apikey",
"planType": null
"planType": null,
"workspaceRole": null,
"isWorkspaceOwner": null
},
}),
serde_json::to_value(jsonrpc_notification)
codex-rs/app-server/tests/common/auth_fixtures.rs
+14
View File
@@ -60,6 +60,11 @@ impl ChatGptAuthFixture {
self
}
pub fn is_org_owner(mut self, is_org_owner: bool) -> Self {
self.claims.is_org_owner = Some(is_org_owner);
self
}
pub fn email(mut self, email: impl Into<String>) -> Self {
self.claims.email = Some(email.into());
self
@@ -82,6 +87,7 @@ pub struct ChatGptIdTokenClaims {
pub plan_type: Option<String>,
pub chatgpt_user_id: Option<String>,
pub chatgpt_account_id: Option<String>,
pub is_org_owner: Option<bool>,
}
impl ChatGptIdTokenClaims {
@@ -108,6 +114,11 @@ impl ChatGptIdTokenClaims {
self.chatgpt_account_id = Some(chatgpt_account_id.into());
self
}
pub fn is_org_owner(mut self, is_org_owner: bool) -> Self {
self.is_org_owner = Some(is_org_owner);
self
}
}
pub fn encode_id_token(claims: &ChatGptIdTokenClaims) -> Result<String> {
@@ -126,6 +137,9 @@ pub fn encode_id_token(claims: &ChatGptIdTokenClaims) -> Result<String> {
if let Some(chatgpt_account_id) = &claims.chatgpt_account_id {
auth_payload.insert("chatgpt_account_id".to_string(), json!(chatgpt_account_id));
}
if let Some(is_org_owner) = claims.is_org_owner {
auth_payload.insert("is_org_owner".to_string(), json!(is_org_owner));
}
if !auth_payload.is_empty() {
payload.insert(
"https://api.openai.com/auth".to_string(),
codex-rs/app-server/tests/common/mcp_process.rs
+11
View File
@@ -58,6 +58,7 @@ use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ReviewStartParams;
use codex_app_server_protocol::ServerRequest;
use codex_app_server_protocol::SkillsListParams;
use codex_app_server_protocol::ThreadAddCreditsNudgeEmailParams;
use codex_app_server_protocol::ThreadArchiveParams;
use codex_app_server_protocol::ThreadCompactStartParams;
use codex_app_server_protocol::ThreadForkParams;
@@ -414,6 +415,16 @@ impl McpProcess {
self.send_request("thread/shellCommand", params).await
}
/// Send a `thread/addCreditsNudgeEmail` JSON-RPC request.
pub async fn send_thread_add_credits_nudge_email_request(
&mut self,
params: ThreadAddCreditsNudgeEmailParams,
) -> anyhow::Result<i64> {
let params = Some(serde_json::to_value(params)?);
self.send_request("thread/addCreditsNudgeEmail", params)
.await
}
/// Send a `thread/rollback` JSON-RPC request.
pub async fn send_thread_rollback_request(
&mut self,
codex-rs/app-server/tests/suite/v2/account.rs
+369 -1
View File
@@ -28,6 +28,7 @@ use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ServerRequest;
use codex_app_server_protocol::TurnCompletedNotification;
use codex_app_server_protocol::TurnStatus;
use codex_app_server_protocol::WorkspaceRole;
use codex_config::types::AuthCredentialsStoreMode;
use codex_login::login_with_api_key;
use codex_protocol::account::PlanType as AccountPlanType;
@@ -55,6 +56,7 @@ struct CreateConfigTomlParams {
forced_workspace_id: Option<String>,
requires_openai_auth: Option<bool>,
base_url: Option<String>,
chatgpt_base_url: Option<String>,
}
fn create_config_toml(codex_home: &Path, params: CreateConfigTomlParams) -> std::io::Result<()> {
@@ -62,6 +64,9 @@ fn create_config_toml(codex_home: &Path, params: CreateConfigTomlParams) -> std:
let base_url = params
.base_url
.unwrap_or_else(|| "http://127.0.0.1:0/v1".to_string());
let chatgpt_base_url = params
.chatgpt_base_url
.unwrap_or_else(|| "http://127.0.0.1:0/backend-api".to_string());
let forced_line = if let Some(method) = params.forced_method {
format!("forced_login_method = \"{method}\"\n")
} else {
@@ -82,6 +87,7 @@ fn create_config_toml(codex_home: &Path, params: CreateConfigTomlParams) -> std:
model = "mock-model"
approval_policy = "never"
sandbox_mode = "danger-full-access"
chatgpt_base_url = "{chatgpt_base_url}"
{forced_line}
{forced_workspace_line}
@@ -122,6 +128,49 @@ async fn mock_device_code_usercode_failure(server: &MockServer, status: u16) {
.await;
}
async fn mock_accounts_check_role(server: &MockServer, account_id: &str, role: &str) {
Mock::given(method("GET"))
.and(path("/backend-api/accounts/check/v4-2023-04-27"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"accounts": {
account_id: {
"account": {
"account_user_role": role,
}
}
},
"account_ordering": [account_id],
})))
.mount(server)
.await;
}
async fn mock_slow_accounts_check_role(
server: &MockServer,
account_id: &str,
role: &str,
delay: Duration,
) {
Mock::given(method("GET"))
.and(path("/backend-api/accounts/check/v4-2023-04-27"))
.respond_with(
ResponseTemplate::new(200)
.set_delay(delay)
.set_body_json(json!({
"accounts": {
account_id: {
"account": {
"account_user_role": role,
}
}
},
"account_ordering": [account_id],
})),
)
.mount(server)
.await;
}
async fn mock_device_code_token_success(server: &MockServer) {
Mock::given(method("POST"))
.and(path("/api/accounts/deviceauth/token"))
@@ -221,10 +270,12 @@ async fn set_auth_token_updates_account_and_notifies() -> Result<()> {
CreateConfigTomlParams {
requires_openai_auth: Some(true),
base_url: Some(format!("{}/v1", mock_server.uri())),
chatgpt_base_url: Some(format!("{}/backend-api", mock_server.uri())),
..Default::default()
},
)?;
write_models_cache(codex_home.path())?;
mock_accounts_check_role(&mock_server, "org-embedded", "standard-user").await;
let access_token = encode_id_token(
&ChatGptIdTokenClaims::new()
@@ -262,6 +313,20 @@ async fn set_auth_token_updates_account_and_notifies() -> Result<()> {
};
assert_eq!(payload.auth_mode, Some(AuthMode::ChatgptAuthTokens));
assert_eq!(payload.plan_type, Some(AccountPlanType::Pro));
assert_eq!(payload.workspace_role, None);
assert_eq!(payload.is_workspace_owner, None);
let note = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("account/updated"),
)
.await??;
let parsed: ServerNotification = note.try_into()?;
let ServerNotification::AccountUpdated(payload) = parsed else {
bail!("unexpected notification: {parsed:?}");
};
assert_eq!(payload.workspace_role, Some(WorkspaceRole::StandardUser));
assert_eq!(payload.is_workspace_owner, Some(false));
let get_id = mcp
.send_get_account_request(GetAccountParams {
@@ -281,6 +346,8 @@ async fn set_auth_token_updates_account_and_notifies() -> Result<()> {
email: "embedded@example.com".to_string(),
plan_type: AccountPlanType::Pro,
}),
workspace_role: None,
is_workspace_owner: None,
requires_openai_auth: true,
}
);
@@ -348,6 +415,8 @@ async fn account_read_refresh_token_is_noop_in_external_mode() -> Result<()> {
email: "embedded@example.com".to_string(),
plan_type: AccountPlanType::Pro,
}),
workspace_role: None,
is_workspace_owner: None,
requires_openai_auth: true,
}
);
@@ -1505,6 +1574,8 @@ async fn get_account_with_api_key() -> Result<()> {
let expected = GetAccountResponse {
account: Some(Account::ApiKey {}),
workspace_role: None,
is_workspace_owner: None,
requires_openai_auth: true,
};
assert_eq!(received, expected);
@@ -1539,6 +1610,8 @@ async fn get_account_when_auth_not_required() -> Result<()> {
let expected = GetAccountResponse {
account: None,
workspace_role: None,
is_workspace_owner: None,
requires_openai_auth: false,
};
assert_eq!(received, expected);
@@ -1559,7 +1632,8 @@ async fn get_account_with_chatgpt() -> Result<()> {
codex_home.path(),
ChatGptAuthFixture::new("access-chatgpt")
.email("user@example.com")
.plan_type("pro"),
.plan_type("pro")
.is_org_owner(/*is_org_owner*/ true),
AuthCredentialsStoreMode::File,
)?;
@@ -1583,12 +1657,304 @@ async fn get_account_with_chatgpt() -> Result<()> {
email: "user@example.com".to_string(),
plan_type: AccountPlanType::Pro,
}),
workspace_role: None,
is_workspace_owner: Some(true),
requires_openai_auth: true,
};
assert_eq!(received, expected);
Ok(())
}
#[tokio::test]
async fn get_account_with_chatgpt_emits_workspace_role_from_accounts_check() -> Result<()> {
let codex_home = TempDir::new()?;
let mock_server = MockServer::start().await;
create_config_toml(
codex_home.path(),
CreateConfigTomlParams {
requires_openai_auth: Some(true),
chatgpt_base_url: Some(format!("{}/backend-api", mock_server.uri())),
..Default::default()
},
)?;
mock_accounts_check_role(&mock_server, "org-embedded", "account-owner").await;
write_chatgpt_auth(
codex_home.path(),
ChatGptAuthFixture::new("access-chatgpt")
.account_id("org-embedded")
.email("user@example.com")
.plan_type("pro")
.chatgpt_account_id("org-embedded"),
AuthCredentialsStoreMode::File,
)?;
let mut mcp = McpProcess::new_with_env(codex_home.path(), &[("OPENAI_API_KEY", None)]).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let params = GetAccountParams {
refresh_token: false,
};
let request_id = mcp.send_get_account_request(params).await?;
let resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
)
.await??;
let received: GetAccountResponse = to_response(resp)?;
let expected = GetAccountResponse {
account: Some(Account::Chatgpt {
email: "user@example.com".to_string(),
plan_type: AccountPlanType::Pro,
}),
workspace_role: None,
is_workspace_owner: None,
requires_openai_auth: true,
};
assert_eq!(received, expected);
let note = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("account/updated"),
)
.await??;
let parsed: ServerNotification = note.try_into()?;
let ServerNotification::AccountUpdated(payload) = parsed else {
bail!("unexpected notification: {parsed:?}");
};
assert_eq!(payload.workspace_role, Some(WorkspaceRole::AccountOwner));
assert_eq!(payload.is_workspace_owner, Some(true));
Ok(())
}
#[tokio::test]
async fn get_account_with_chatgpt_dedupes_workspace_role_refresh_after_success() -> Result<()> {
let codex_home = TempDir::new()?;
let mock_server = MockServer::start().await;
create_config_toml(
codex_home.path(),
CreateConfigTomlParams {
requires_openai_auth: Some(true),
chatgpt_base_url: Some(format!("{}/backend-api", mock_server.uri())),
..Default::default()
},
)?;
Mock::given(method("GET"))
.and(path("/backend-api/accounts/check/v4-2023-04-27"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"accounts": {
"org-embedded": {
"account": {
"account_user_role": "account-owner",
}
}
},
"account_ordering": ["org-embedded"],
})))
.expect(1)
.mount(&mock_server)
.await;
write_chatgpt_auth(
codex_home.path(),
ChatGptAuthFixture::new("access-chatgpt")
.account_id("org-embedded")
.email("user@example.com")
.plan_type("pro")
.chatgpt_account_id("org-embedded"),
AuthCredentialsStoreMode::File,
)?;
let mut mcp = McpProcess::new_with_env(codex_home.path(), &[("OPENAI_API_KEY", None)]).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let first_request_id = mcp
.send_get_account_request(GetAccountParams {
refresh_token: false,
})
.await?;
let first_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(first_request_id)),
)
.await??;
let first_account: GetAccountResponse = to_response(first_resp)?;
assert_eq!(
first_account,
GetAccountResponse {
account: Some(Account::Chatgpt {
email: "user@example.com".to_string(),
plan_type: AccountPlanType::Pro,
}),
workspace_role: None,
is_workspace_owner: None,
requires_openai_auth: true,
}
);
let note = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("account/updated"),
)
.await??;
let parsed: ServerNotification = note.try_into()?;
let ServerNotification::AccountUpdated(payload) = parsed else {
bail!("unexpected notification: {parsed:?}");
};
assert_eq!(payload.workspace_role, Some(WorkspaceRole::AccountOwner));
assert_eq!(payload.is_workspace_owner, Some(true));
let second_request_id = mcp
.send_get_account_request(GetAccountParams {
refresh_token: false,
})
.await?;
let second_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(second_request_id)),
)
.await??;
let second_account: GetAccountResponse = to_response(second_resp)?;
assert_eq!(
second_account,
GetAccountResponse {
account: Some(Account::Chatgpt {
email: "user@example.com".to_string(),
plan_type: AccountPlanType::Pro,
}),
workspace_role: None,
is_workspace_owner: None,
requires_openai_auth: true,
}
);
let duplicate_note = timeout(
Duration::from_millis(250),
mcp.read_stream_until_notification_message("account/updated"),
)
.await;
assert!(
duplicate_note.is_err(),
"duplicate account/updated notification should not be emitted once workspace role is cached"
);
mock_server.verify().await;
Ok(())
}
#[tokio::test]
async fn get_account_with_chatgpt_does_not_guess_workspace_role_from_other_accounts() -> Result<()>
{
let codex_home = TempDir::new()?;
let mock_server = MockServer::start().await;
create_config_toml(
codex_home.path(),
CreateConfigTomlParams {
requires_openai_auth: Some(true),
chatgpt_base_url: Some(format!("{}/backend-api", mock_server.uri())),
..Default::default()
},
)?;
mock_accounts_check_role(&mock_server, "org-other", "account-owner").await;
write_chatgpt_auth(
codex_home.path(),
ChatGptAuthFixture::new("access-chatgpt")
.account_id("org-current")
.email("user@example.com")
.plan_type("pro")
.chatgpt_account_id("org-current"),
AuthCredentialsStoreMode::File,
)?;
let mut mcp = McpProcess::new_with_env(codex_home.path(), &[("OPENAI_API_KEY", None)]).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let params = GetAccountParams {
refresh_token: false,
};
let request_id = mcp.send_get_account_request(params).await?;
let resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
)
.await??;
let received: GetAccountResponse = to_response(resp)?;
let expected = GetAccountResponse {
account: Some(Account::Chatgpt {
email: "user@example.com".to_string(),
plan_type: AccountPlanType::Pro,
}),
workspace_role: None,
is_workspace_owner: None,
requires_openai_auth: true,
};
assert_eq!(received, expected);
Ok(())
}
#[tokio::test]
async fn get_account_with_chatgpt_does_not_wait_for_accounts_check() -> Result<()> {
let codex_home = TempDir::new()?;
let mock_server = MockServer::start().await;
create_config_toml(
codex_home.path(),
CreateConfigTomlParams {
requires_openai_auth: Some(true),
chatgpt_base_url: Some(format!("{}/backend-api", mock_server.uri())),
..Default::default()
},
)?;
mock_slow_accounts_check_role(
&mock_server,
"org-embedded",
"standard-user",
Duration::from_secs(2),
)
.await;
write_chatgpt_auth(
codex_home.path(),
ChatGptAuthFixture::new("access-chatgpt")
.account_id("org-embedded")
.email("user@example.com")
.plan_type("pro")
.chatgpt_account_id("org-embedded")
.is_org_owner(/*is_org_owner*/ true),
AuthCredentialsStoreMode::File,
)?;
let mut mcp = McpProcess::new_with_env(codex_home.path(), &[("OPENAI_API_KEY", None)]).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let request_id = mcp
.send_get_account_request(GetAccountParams {
refresh_token: false,
})
.await?;
let resp: JSONRPCResponse = timeout(
Duration::from_millis(500),
mcp.read_stream_until_response_message(RequestId::Integer(request_id)),
)
.await??;
let received: GetAccountResponse = to_response(resp)?;
assert_eq!(
received,
GetAccountResponse {
account: Some(Account::Chatgpt {
email: "user@example.com".to_string(),
plan_type: AccountPlanType::Pro,
}),
workspace_role: None,
is_workspace_owner: Some(true),
requires_openai_auth: true,
}
);
Ok(())
}
#[tokio::test]
async fn get_account_with_chatgpt_missing_plan_claim_returns_unknown() -> Result<()> {
let codex_home = TempDir::new()?;
@@ -1625,6 +1991,8 @@ async fn get_account_with_chatgpt_missing_plan_claim_returns_unknown() -> Result
email: "user@example.com".to_string(),
plan_type: AccountPlanType::Unknown,
}),
workspace_role: None,
is_workspace_owner: None,
requires_openai_auth: true,
};
assert_eq!(received, expected);
codex-rs/app-server/tests/suite/v2/mod.rs
+1
View File
@@ -32,6 +32,7 @@ mod request_user_input;
mod review;
mod safety_check_downgrade;
mod skills_list;
mod thread_add_credits_nudge_email;
mod thread_archive;
mod thread_fork;
mod thread_list;
codex-rs/app-server/tests/suite/v2/rate_limits.rs
+9 -1
View File
@@ -133,7 +133,10 @@ async fn get_account_rate_limits_returns_snapshot() -> Result<()> {
}
}
}
]
],
"spend_control": {
"reached": true
}
});
Mock::given(method("GET"))
@@ -172,6 +175,7 @@ async fn get_account_rate_limits_returns_snapshot() -> Result<()> {
resets_at: Some(secondary_reset_timestamp),
}),
credits: None,
spend_control: Some(codex_app_server_protocol::SpendControlSnapshot { reached: true }),
plan_type: Some(AccountPlanType::Pro),
},
rate_limits_by_limit_id: Some(
@@ -192,6 +196,9 @@ async fn get_account_rate_limits_returns_snapshot() -> Result<()> {
resets_at: Some(secondary_reset_timestamp),
}),
credits: None,
spend_control: Some(codex_app_server_protocol::SpendControlSnapshot {
reached: true,
}),
plan_type: Some(AccountPlanType::Pro),
},
),
@@ -207,6 +214,7 @@ async fn get_account_rate_limits_returns_snapshot() -> Result<()> {
}),
secondary: None,
credits: None,
spend_control: None,
plan_type: Some(AccountPlanType::Pro),
},
),
codex-rs/app-server/tests/suite/v2/thread_add_credits_nudge_email.rs
+96
View File
@@ -0,0 +1,96 @@
use anyhow::Result;
use app_test_support::McpProcess;
use app_test_support::create_mock_responses_server_sequence;
use app_test_support::to_response;
use codex_app_server_protocol::AddCreditsNudgeEmailNotification;
use codex_app_server_protocol::AddCreditsNudgeEmailResult;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ThreadAddCreditsNudgeEmailParams;
use codex_app_server_protocol::ThreadAddCreditsNudgeEmailResponse;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use pretty_assertions::assert_eq;
use std::path::Path;
use tempfile::TempDir;
use tokio::time::timeout;
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
#[tokio::test]
async fn thread_add_credits_nudge_email_submits_core_op_and_emits_completion() -> Result<()> {
let tmp = TempDir::new()?;
let codex_home = tmp.path().join("codex_home");
std::fs::create_dir(&codex_home)?;
let server = create_mock_responses_server_sequence(vec![]).await;
create_config_toml(codex_home.as_path(), &server.uri())?;
let mut mcp = McpProcess::new(codex_home.as_path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let start_id = mcp
.send_thread_start_request(ThreadStartParams::default())
.await?;
let start_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(start_id)),
)
.await??;
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(start_resp)?;
let nudge_id = mcp
.send_thread_add_credits_nudge_email_request(ThreadAddCreditsNudgeEmailParams {
thread_id: thread.id.clone(),
})
.await?;
let nudge_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(nudge_id)),
)
.await??;
let _: ThreadAddCreditsNudgeEmailResponse =
to_response::<ThreadAddCreditsNudgeEmailResponse>(nudge_resp)?;
let notification: AddCreditsNudgeEmailNotification = serde_json::from_value(
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("account/addCreditsNudgeEmail/completed"),
)
.await??
.params
.expect("account/addCreditsNudgeEmail/completed params"),
)?;
assert_eq!(notification.thread_id, thread.id);
assert_eq!(
notification.result,
AddCreditsNudgeEmailResult::Failed {
message: "codex account authentication required to notify workspace owner".to_string(),
}
);
Ok(())
}
fn create_config_toml(codex_home: &Path, server_uri: &str) -> std::io::Result<()> {
std::fs::write(
codex_home.join("config.toml"),
format!(
r#"
model = "mock-model"
approval_policy = "never"
sandbox_mode = "read-only"
model_provider = "mock_provider"
[model_providers.mock_provider]
name = "Mock provider for test"
base_url = "{server_uri}/v1"
wire_api = "responses"
request_max_retries = 0
stream_max_retries = 0
"#
),
)
}