Merge remote-tracking branch 'origin/main' into jchu/codex-cli-owner-nudge-analytics-2026.04.22

# Conflicts:
#	codex-rs/app-server/src/message_processor.rs
This commit is contained in:
James Chu
2026-04-27 16:01:11 -04:00
66 changed files with 1791 additions and 1821 deletions
+1 -1
View File
@@ -33,7 +33,7 @@ runs:
run: Copy-Item (Get-Command dotslash).Source -Destination "$env:LOCALAPPDATA\Microsoft\WindowsApps\dotslash.exe"
- name: Set up Bazel
uses: bazelbuild/setup-bazelisk@b39c379c82683a5f25d34f0d062761f62693e0b2 # v3
uses: bazel-contrib/setup-bazel@c5acdfb288317d0b5c0bbd7a396a3dc868bb0f86 # 0.19.0
- name: Configure Bazel repository cache
id: configure_bazel_repository_cache
+7 -2
View File
@@ -45,11 +45,16 @@ jobs:
GH_TOKEN: ${{ github.token }}
run: |
set -euo pipefail
# Use a rust-release version that includes all native binaries.
CODEX_VERSION=0.115.0
# Use a recent successful rust-release run that published the full
# cross-platform native payload required by the npm package layout.
# Passing the workflow URL directly avoids relying on old rust-v*
# branches remaining discoverable via `gh run list --branch ...`.
CODEX_VERSION=0.125.0
WORKFLOW_URL="https://github.com/openai/codex/actions/runs/24901475298"
OUTPUT_DIR="${RUNNER_TEMP}"
python3 ./scripts/stage_npm_packages.py \
--release-version "$CODEX_VERSION" \
--workflow-url "$WORKFLOW_URL" \
--package codex \
--output-dir "$OUTPUT_DIR"
PACK_OUTPUT="${OUTPUT_DIR}/codex-npm-${CODEX_VERSION}.tgz"
+3 -1
View File
@@ -78,7 +78,9 @@ jobs:
- uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6
- name: Set up Bazel
uses: bazelbuild/setup-bazelisk@b39c379c82683a5f25d34f0d062761f62693e0b2 # v3
uses: ./.github/actions/setup-bazel-ci
with:
target: ${{ matrix.target }}
- name: Set up Python
uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6
+5 -1
View File
@@ -3,6 +3,7 @@ name: v8-canary
on:
pull_request:
paths:
- ".github/actions/setup-bazel-ci/**"
- ".github/scripts/rusty_v8_bazel.py"
- ".github/workflows/rusty-v8-release.yml"
- ".github/workflows/v8-canary.yml"
@@ -16,6 +17,7 @@ on:
branches:
- main
paths:
- ".github/actions/setup-bazel-ci/**"
- ".github/scripts/rusty_v8_bazel.py"
- ".github/workflows/rusty-v8-release.yml"
- ".github/workflows/v8-canary.yml"
@@ -75,7 +77,9 @@ jobs:
- uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6
- name: Set up Bazel
uses: bazelbuild/setup-bazelisk@b39c379c82683a5f25d34f0d062761f62693e0b2 # v3
uses: ./.github/actions/setup-bazel-ci
with:
target: ${{ matrix.target }}
- name: Set up Python
uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6
File diff suppressed because it is too large Load Diff
@@ -1,4 +1,6 @@
use super::*;
use crate::error_code::internal_error;
use crate::error_code::invalid_request;
use codex_app_server_protocol::PluginInstallPolicy;
impl CodexMessageProcessor {
@@ -7,46 +9,33 @@ impl CodexMessageProcessor {
request_id: ConnectionRequestId,
params: PluginListParams,
) {
let result = self.plugin_list_response(params).await;
self.outgoing.send_result(request_id, result).await;
}
async fn plugin_list_response(
&self,
params: PluginListParams,
) -> Result<PluginListResponse, JSONRPCErrorError> {
let plugins_manager = self.thread_manager.plugins_manager();
let PluginListParams { cwds } = params;
let roots = cwds.unwrap_or_default();
let config = match self.load_latest_config(/*fallback_cwd*/ None).await {
Ok(config) => config,
Err(err) => {
self.outgoing.send_error(request_id, err).await;
return;
}
let config = self.load_latest_config(/*fallback_cwd*/ None).await?;
let empty_response = || PluginListResponse {
marketplaces: Vec::new(),
marketplace_load_errors: Vec::new(),
featured_plugin_ids: Vec::new(),
};
if !config.features.enabled(Feature::Plugins) {
self.outgoing
.send_response(
request_id,
PluginListResponse {
marketplaces: Vec::new(),
marketplace_load_errors: Vec::new(),
featured_plugin_ids: Vec::new(),
},
)
.await;
return;
return Ok(empty_response());
}
let auth = self.auth_manager.auth().await;
if !self
.workspace_codex_plugins_enabled(&config, auth.as_ref())
.await
{
self.outgoing
.send_response(
request_id,
PluginListResponse {
marketplaces: Vec::new(),
marketplace_load_errors: Vec::new(),
featured_plugin_ids: Vec::new(),
},
)
.await;
return;
return Ok(empty_response());
}
plugins_manager.maybe_start_non_curated_plugin_cache_refresh(&roots);
@@ -100,18 +89,11 @@ impl CodexMessageProcessor {
.await
{
Ok(Ok(outcome)) => outcome,
Ok(Err(err)) => {
self.send_marketplace_error(request_id, err, "list marketplace plugins")
.await;
return;
}
Ok(Err(err)) => return Err(Self::marketplace_error(err, "list marketplace plugins")),
Err(err) => {
self.send_internal_error(
request_id,
format!("failed to list marketplace plugins: {err}"),
)
.await;
return;
return Err(internal_error(format!(
"failed to list marketplace plugins: {err}"
)));
}
};
@@ -174,16 +156,11 @@ impl CodexMessageProcessor {
Vec::new()
};
self.outgoing
.send_response(
request_id,
PluginListResponse {
marketplaces: data,
marketplace_load_errors,
featured_plugin_ids,
},
)
.await;
Ok(PluginListResponse {
marketplaces: data,
marketplace_load_errors,
featured_plugin_ids,
})
}
pub(super) async fn plugin_read(
@@ -191,6 +168,14 @@ impl CodexMessageProcessor {
request_id: ConnectionRequestId,
params: PluginReadParams,
) {
let result = self.plugin_read_response(params).await;
self.outgoing.send_result(request_id, result).await;
}
async fn plugin_read_response(
&self,
params: PluginReadParams,
) -> Result<PluginReadResponse, JSONRPCErrorError> {
let plugins_manager = self.thread_manager.plugins_manager();
let PluginReadParams {
marketplace_path,
@@ -201,30 +186,16 @@ impl CodexMessageProcessor {
(Some(marketplace_path), None) => Ok(marketplace_path),
(None, Some(remote_marketplace_name)) => Err(remote_marketplace_name),
(Some(_), Some(_)) | (None, None) => {
self.outgoing
.send_error(
request_id,
JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: "plugin/read requires exactly one of marketplacePath or remoteMarketplaceName".to_string(),
data: None,
},
)
.await;
return;
return Err(invalid_request(
"plugin/read requires exactly one of marketplacePath or remoteMarketplaceName",
));
}
};
let config_cwd = read_source.as_ref().ok().and_then(|marketplace_path| {
marketplace_path.as_path().parent().map(Path::to_path_buf)
});
let config = match self.load_latest_config(config_cwd).await {
Ok(config) => config,
Err(err) => {
self.outgoing.send_error(request_id, err).await;
return;
}
};
let config = self.load_latest_config(config_cwd).await?;
let plugin = match read_source {
Ok(marketplace_path) => {
@@ -232,17 +203,10 @@ impl CodexMessageProcessor {
plugin_name,
marketplace_path,
};
let outcome = match plugins_manager
let outcome = plugins_manager
.read_plugin_for_config(&config, &request)
.await
{
Ok(outcome) => outcome,
Err(err) => {
self.send_marketplace_error(request_id, err, "read plugin details")
.await;
return;
}
};
.map_err(|err| Self::marketplace_error(err, "read plugin details"))?;
let environment_manager = self.thread_manager.environment_manager();
let app_summaries = plugin_app_helpers::load_plugin_app_summaries(
&config,
@@ -287,19 +251,9 @@ impl CodexMessageProcessor {
if !config.features.enabled(Feature::Plugins)
|| !config.features.enabled(Feature::RemotePlugin)
{
self.outgoing
.send_error(
request_id,
JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: format!(
"remote plugin read is not enabled for marketplace {remote_marketplace_name}"
),
data: None,
},
)
.await;
return;
return Err(invalid_request(format!(
"remote plugin read is not enabled for marketplace {remote_marketplace_name}"
)));
}
let auth = self.auth_manager.auth().await;
let remote_plugin_service_config = RemotePluginServiceConfig {
@@ -310,36 +264,20 @@ impl CodexMessageProcessor {
.chars()
.all(|ch| ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' || ch == '~')
{
self.send_invalid_request_error(
request_id,
"invalid remote plugin id: only ASCII letters, digits, `_`, `-`, and `~` are allowed"
.to_string(),
)
.await;
return;
return Err(invalid_request(
"invalid remote plugin id: only ASCII letters, digits, `_`, `-`, and `~` are allowed",
));
}
let remote_detail = match codex_core_plugins::remote::fetch_remote_plugin_detail(
let remote_detail = codex_core_plugins::remote::fetch_remote_plugin_detail(
&remote_plugin_service_config,
auth.as_ref(),
&remote_marketplace_name,
&plugin_name,
)
.await
{
Ok(remote_detail) => remote_detail,
Err(err) => {
self.outgoing
.send_error(
request_id,
remote_plugin_catalog_error_to_jsonrpc(
err,
"read remote plugin details",
),
)
.await;
return;
}
};
.map_err(|err| {
remote_plugin_catalog_error_to_jsonrpc(err, "read remote plugin details")
})?;
let plugin_apps = remote_detail
.app_ids
.iter()
@@ -357,9 +295,7 @@ impl CodexMessageProcessor {
}
};
self.outgoing
.send_response(request_id, PluginReadResponse { plugin })
.await;
Ok(PluginReadResponse { plugin })
}
pub(super) async fn plugin_install(
@@ -367,6 +303,14 @@ impl CodexMessageProcessor {
request_id: ConnectionRequestId,
params: PluginInstallParams,
) {
let result = self.plugin_install_response(params).await;
self.outgoing.send_result(request_id, result).await;
}
async fn plugin_install_response(
&self,
params: PluginInstallParams,
) -> Result<PluginInstallResponse, JSONRPCErrorError> {
let PluginInstallParams {
marketplace_path,
remote_marketplace_name,
@@ -375,44 +319,27 @@ impl CodexMessageProcessor {
let marketplace_path = match (marketplace_path, remote_marketplace_name) {
(Some(marketplace_path), None) => marketplace_path,
(None, Some(remote_marketplace_name)) => {
self.remote_plugin_install(request_id, remote_marketplace_name, plugin_name)
return self
.remote_plugin_install_response(remote_marketplace_name, plugin_name)
.await;
return;
}
(Some(_), Some(_)) | (None, None) => {
self.outgoing
.send_error(
request_id,
JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: "plugin/install requires exactly one of marketplacePath or remoteMarketplaceName".to_string(),
data: None,
},
)
.await;
return;
return Err(invalid_request(
"plugin/install requires exactly one of marketplacePath or remoteMarketplaceName",
));
}
};
let config_cwd = marketplace_path.as_path().parent().map(Path::to_path_buf);
let config = match self.load_latest_config(config_cwd.clone()).await {
Ok(config) => config,
Err(err) => {
self.outgoing.send_error(request_id, err).await;
return;
}
};
let config = self.load_latest_config(config_cwd.clone()).await?;
let auth = self.auth_manager.auth().await;
if !self
.workspace_codex_plugins_enabled(&config, auth.as_ref())
.await
{
self.send_invalid_request_error(
request_id,
"Codex plugins are disabled for this workspace".to_string(),
)
.await;
return;
return Err(invalid_request(
"Codex plugins are disabled for this workspace",
));
}
let plugins_manager = self.thread_manager.plugins_manager();
@@ -421,197 +348,103 @@ impl CodexMessageProcessor {
marketplace_path,
};
let install_result = plugins_manager.install_plugin(request).await;
match install_result {
Ok(result) => {
let config = match self.load_latest_config(config_cwd).await {
Ok(config) => config,
Err(err) => {
warn!(
"failed to reload config after plugin install, using current config: {err:?}"
);
config
}
};
self.clear_plugin_related_caches();
let plugin_mcp_servers =
load_plugin_mcp_servers(result.installed_path.as_path()).await;
if !plugin_mcp_servers.is_empty() {
if let Err(err) = self.queue_mcp_server_refresh_for_config(&config).await {
warn!(
plugin = result.plugin_id.as_key(),
"failed to queue MCP refresh after plugin install: {err:?}"
);
}
self.start_plugin_mcp_oauth_logins(&config, plugin_mcp_servers)
.await;
}
let plugin_apps = load_plugin_apps(result.installed_path.as_path()).await;
let auth = self.auth_manager.auth().await;
let apps_needing_auth = self
.plugin_apps_needing_auth_for_install(
&config,
auth.as_ref().is_some_and(CodexAuth::is_chatgpt_auth),
&result.plugin_id.as_key(),
&plugin_apps,
)
.await;
self.outgoing
.send_response(
request_id,
PluginInstallResponse {
auth_policy: result.auth_policy.into(),
apps_needing_auth,
},
)
.await;
}
Err(err) => {
if err.is_invalid_request() {
self.send_invalid_request_error(request_id, err.to_string())
.await;
return;
}
match err {
CorePluginInstallError::Marketplace(err) => {
self.send_marketplace_error(request_id, err, "install plugin")
.await;
}
CorePluginInstallError::Config(err) => {
self.send_internal_error(
request_id,
format!("failed to persist installed plugin config: {err}"),
)
.await;
}
CorePluginInstallError::Remote(err) => {
self.send_internal_error(
request_id,
format!("failed to enable remote plugin: {err}"),
)
.await;
}
CorePluginInstallError::Join(err) => {
self.send_internal_error(
request_id,
format!("failed to install plugin: {err}"),
)
.await;
}
CorePluginInstallError::Store(err) => {
self.send_internal_error(
request_id,
format!("failed to install plugin: {err}"),
)
.await;
}
}
}
}
}
async fn remote_plugin_install(
&self,
request_id: ConnectionRequestId,
remote_marketplace_name: String,
plugin_name: String,
) {
let config = match self.load_latest_config(/*fallback_cwd*/ None).await {
let result = plugins_manager
.install_plugin(request)
.await
.map_err(Self::plugin_install_error)?;
let config = match self.load_latest_config(config_cwd).await {
Ok(config) => config,
Err(err) => {
self.outgoing.send_error(request_id, err).await;
return;
warn!(
"failed to reload config after plugin install, using current config: {err:?}"
);
config
}
};
self.clear_plugin_related_caches();
let plugin_mcp_servers = load_plugin_mcp_servers(result.installed_path.as_path()).await;
if !plugin_mcp_servers.is_empty() {
if let Err(err) = self.queue_mcp_server_refresh_for_config(&config).await {
warn!(
plugin = result.plugin_id.as_key(),
"failed to queue MCP refresh after plugin install: {err:?}"
);
}
self.start_plugin_mcp_oauth_logins(&config, plugin_mcp_servers)
.await;
}
let plugin_apps = load_plugin_apps(result.installed_path.as_path()).await;
let auth = self.auth_manager.auth().await;
let apps_needing_auth = self
.plugin_apps_needing_auth_for_install(
&config,
auth.as_ref().is_some_and(CodexAuth::is_chatgpt_auth),
&result.plugin_id.as_key(),
&plugin_apps,
)
.await;
Ok(PluginInstallResponse {
auth_policy: result.auth_policy.into(),
apps_needing_auth,
})
}
async fn remote_plugin_install_response(
&self,
remote_marketplace_name: String,
plugin_name: String,
) -> Result<PluginInstallResponse, JSONRPCErrorError> {
let config = self.load_latest_config(/*fallback_cwd*/ None).await?;
if !config.features.enabled(Feature::Plugins)
|| !config.features.enabled(Feature::RemotePlugin)
{
self.outgoing
.send_error(
request_id,
JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: format!(
"remote plugin install is not enabled for marketplace {remote_marketplace_name}"
),
data: None,
},
)
.await;
return;
return Err(invalid_request(format!(
"remote plugin install is not enabled for marketplace {remote_marketplace_name}"
)));
}
if plugin_name.is_empty()
|| !plugin_name
.chars()
.all(|ch| ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' || ch == '~')
{
self.send_invalid_request_error(
request_id,
"invalid remote plugin id: only ASCII letters, digits, `_`, `-`, and `~` are allowed"
.to_string(),
)
.await;
return;
return Err(invalid_request(
"invalid remote plugin id: only ASCII letters, digits, `_`, `-`, and `~` are allowed",
));
}
let auth = self.auth_manager.auth().await;
let remote_plugin_service_config = RemotePluginServiceConfig {
chatgpt_base_url: config.chatgpt_base_url.clone(),
};
let remote_detail = match codex_core_plugins::remote::fetch_remote_plugin_detail(
let remote_detail = codex_core_plugins::remote::fetch_remote_plugin_detail(
&remote_plugin_service_config,
auth.as_ref(),
&remote_marketplace_name,
&plugin_name,
)
.await
{
Ok(remote_detail) => remote_detail,
Err(err) => {
self.outgoing
.send_error(
request_id,
remote_plugin_catalog_error_to_jsonrpc(
err,
"read remote plugin details before install",
),
)
.await;
return;
}
};
.map_err(|err| {
remote_plugin_catalog_error_to_jsonrpc(err, "read remote plugin details before install")
})?;
if remote_detail.summary.install_policy == PluginInstallPolicy::NotAvailable {
self.send_invalid_request_error(
request_id,
format!("remote plugin {plugin_name} is not available for install"),
)
.await;
return;
return Err(invalid_request(format!(
"remote plugin {plugin_name} is not available for install"
)));
}
if let Err(err) = codex_core_plugins::remote::install_remote_plugin(
codex_core_plugins::remote::install_remote_plugin(
&remote_plugin_service_config,
auth.as_ref(),
&remote_marketplace_name,
&plugin_name,
)
.await
{
self.outgoing
.send_error(
request_id,
remote_plugin_catalog_error_to_jsonrpc(err, "install remote plugin"),
)
.await;
return;
}
.map_err(|err| remote_plugin_catalog_error_to_jsonrpc(err, "install remote plugin"))?;
self.clear_plugin_related_caches();
@@ -629,15 +462,10 @@ impl CodexMessageProcessor {
)
.await;
self.outgoing
.send_response(
request_id,
PluginInstallResponse {
auth_policy: remote_detail.summary.auth_policy,
apps_needing_auth,
},
)
.await;
Ok(PluginInstallResponse {
auth_policy: remote_detail.summary.auth_policy,
apps_needing_auth,
})
}
async fn plugin_apps_needing_auth_for_install(
@@ -709,59 +537,82 @@ impl CodexMessageProcessor {
request_id: ConnectionRequestId,
params: PluginUninstallParams,
) {
let result = self.plugin_uninstall_response(params).await;
self.outgoing.send_result(request_id, result).await;
}
async fn plugin_uninstall_response(
&self,
params: PluginUninstallParams,
) -> Result<PluginUninstallResponse, JSONRPCErrorError> {
let PluginUninstallParams { plugin_id } = params;
let plugins_manager = self.thread_manager.plugins_manager();
let uninstall_result = plugins_manager.uninstall_plugin(plugin_id).await;
plugins_manager
.uninstall_plugin(plugin_id)
.await
.map_err(Self::plugin_uninstall_error)?;
self.clear_plugin_related_caches();
Ok(PluginUninstallResponse {})
}
match uninstall_result {
Ok(()) => {
self.clear_plugin_related_caches();
self.outgoing
.send_response(request_id, PluginUninstallResponse {})
.await;
}
Err(err) => {
if err.is_invalid_request() {
self.send_invalid_request_error(request_id, err.to_string())
.await;
return;
}
fn plugin_install_error(err: CorePluginInstallError) -> JSONRPCErrorError {
if err.is_invalid_request() {
return invalid_request(err.to_string());
}
match err {
CorePluginUninstallError::Config(err) => {
self.send_internal_error(
request_id,
format!("failed to clear plugin config: {err}"),
)
.await;
}
CorePluginUninstallError::Remote(err) => {
self.send_internal_error(
request_id,
format!("failed to uninstall remote plugin: {err}"),
)
.await;
}
CorePluginUninstallError::Join(err) => {
self.send_internal_error(
request_id,
format!("failed to uninstall plugin: {err}"),
)
.await;
}
CorePluginUninstallError::Store(err) => {
self.send_internal_error(
request_id,
format!("failed to uninstall plugin: {err}"),
)
.await;
}
CorePluginUninstallError::InvalidPluginId(_) => {
unreachable!("invalid plugin ids are handled above");
}
}
match err {
CorePluginInstallError::Marketplace(err) => {
Self::marketplace_error(err, "install plugin")
}
CorePluginInstallError::Config(err) => {
internal_error(format!("failed to persist installed plugin config: {err}"))
}
CorePluginInstallError::Remote(err) => {
internal_error(format!("failed to enable remote plugin: {err}"))
}
CorePluginInstallError::Join(err) => {
internal_error(format!("failed to install plugin: {err}"))
}
CorePluginInstallError::Store(err) => {
internal_error(format!("failed to install plugin: {err}"))
}
}
}
fn plugin_uninstall_error(err: CorePluginUninstallError) -> JSONRPCErrorError {
if err.is_invalid_request() {
return invalid_request(err.to_string());
}
match err {
CorePluginUninstallError::Config(err) => {
internal_error(format!("failed to clear plugin config: {err}"))
}
CorePluginUninstallError::Remote(err) => {
internal_error(format!("failed to uninstall remote plugin: {err}"))
}
CorePluginUninstallError::Join(err) => {
internal_error(format!("failed to uninstall plugin: {err}"))
}
CorePluginUninstallError::Store(err) => {
internal_error(format!("failed to uninstall plugin: {err}"))
}
CorePluginUninstallError::InvalidPluginId(_) => {
unreachable!("invalid plugin ids are handled above");
}
}
}
fn marketplace_error(err: MarketplaceError, action: &str) -> JSONRPCErrorError {
match err {
MarketplaceError::MarketplaceNotFound { .. }
| MarketplaceError::InvalidMarketplaceFile { .. }
| MarketplaceError::PluginNotFound { .. }
| MarketplaceError::PluginNotAvailable { .. }
| MarketplaceError::PluginsDisabled
| MarketplaceError::InvalidPlugin(_) => invalid_request(err.to_string()),
MarketplaceError::Io { .. } => internal_error(format!("failed to {action}: {err}")),
}
}
}
+2 -1
View File
@@ -391,7 +391,8 @@ fn start_uninitialized(args: InProcessStartArgs) -> InProcessClientHandle {
let processor_outgoing = Arc::clone(&outgoing_message_sender);
let auth_manager =
AuthManager::shared_from_config(args.config.as_ref(), args.enable_codex_api_key_env);
AuthManager::shared_from_config(args.config.as_ref(), args.enable_codex_api_key_env)
.await;
let config_manager = ConfigManager::new(
args.config.codex_home.to_path_buf(),
args.cli_overrides,
+3 -3
View File
@@ -469,7 +469,7 @@ pub async fn run_main_with_transport_options(
config_manager
.replace_thread_config_loader(Arc::clone(&discovered_thread_config_loader));
let auth_manager =
AuthManager::shared_from_config(&config, /*enable_codex_api_key_env*/ false);
AuthManager::shared_from_config(&config, /*enable_codex_api_key_env*/ false).await;
config_manager.replace_cloud_requirements_loader(auth_manager, config.chatgpt_base_url);
}
Err(err) => {
@@ -631,7 +631,7 @@ pub async fn run_main_with_transport_options(
}
let auth_manager =
AuthManager::shared_from_config(&config, /*enable_codex_api_key_env*/ false);
AuthManager::shared_from_config(&config, /*enable_codex_api_key_env*/ false).await;
let remote_control_enabled = config.features.enabled(Feature::RemoteControl);
if transport_accept_handles.is_empty() && !remote_control_enabled {
@@ -712,7 +712,7 @@ pub async fn run_main_with_transport_options(
let outgoing_message_sender = Arc::new(OutgoingMessageSender::new(outgoing_tx));
let outbound_control_tx = outbound_control_tx;
let auth_manager =
AuthManager::shared_from_config(&config, /*enable_codex_api_key_env*/ false);
AuthManager::shared_from_config(&config, /*enable_codex_api_key_env*/ false).await;
let processor = Arc::new(MessageProcessor::new(MessageProcessorArgs {
outgoing: outgoing_message_sender,
arg0_paths,
+9 -12
View File
@@ -653,14 +653,12 @@ impl MessageProcessor {
}
}
}
if self.config.features.enabled(Feature::GeneralAnalytics) {
self.analytics_events_client.track_initialize(
connection_id.0,
analytics_initialize_params,
originator,
self.rpc_transport,
);
}
self.analytics_events_client.track_initialize(
connection_id.0,
analytics_initialize_params,
originator,
self.rpc_transport,
);
set_default_client_residency_requirement(self.config.enforce_residency.value());
if let Ok(mut suffix) = USER_AGENT_SUFFIX.lock() {
*suffix = Some(user_agent_suffix);
@@ -716,10 +714,9 @@ impl MessageProcessor {
return Err(invalid_request(experimental_required_message(reason)));
}
let connection_id = connection_request_id.connection_id;
if self.config.features.enabled(Feature::GeneralAnalytics)
&& let ClientRequest::TurnStart { request_id, .. }
| ClientRequest::TurnSteer { request_id, .. }
| ClientRequest::TrackUsageLimitBanner { request_id, .. } = &codex_request
if let ClientRequest::TurnStart { request_id, .. }
| ClientRequest::TurnSteer { request_id, .. }
| ClientRequest::TrackUsageLimitBanner { request_id, .. } = &codex_request
{
self.analytics_events_client.track_request(
connection_id.0,
@@ -127,7 +127,7 @@ impl TracingHarness {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
let config = Arc::new(build_test_config(codex_home.path(), &server.uri()).await?);
let (processor, outgoing_rx) = build_test_processor(config);
let (processor, outgoing_rx) = build_test_processor(config).await;
let tracing = init_test_tracing();
tracing.exporter.reset();
tracing::callsite::rebuild_interest_cache();
@@ -257,7 +257,7 @@ async fn build_test_config(codex_home: &Path, server_uri: &str) -> Result<Config
.await?)
}
fn build_test_processor(
async fn build_test_processor(
config: Arc<Config>,
) -> (
Arc<MessageProcessor>,
@@ -266,7 +266,7 @@ fn build_test_processor(
let (outgoing_tx, outgoing_rx) = mpsc::channel(16);
let outgoing = Arc::new(OutgoingMessageSender::new(outgoing_tx));
let auth_manager =
AuthManager::shared_from_config(config.as_ref(), /*enable_codex_api_key_env*/ false);
AuthManager::shared_from_config(config.as_ref(), /*enable_codex_api_key_env*/ false).await;
let config_manager = ConfigManager::new(
config.codex_home.to_path_buf(),
Vec::new(),
@@ -497,7 +497,8 @@ async fn remote_control_start_allows_missing_auth_when_enabled() {
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
/*chatgpt_base_url*/ None,
);
)
.await;
let (transport_event_tx, _transport_event_rx) =
mpsc::channel::<TransportEvent>(CHANNEL_CAPACITY);
let shutdown_token = CancellationToken::new();
@@ -1085,7 +1086,8 @@ async fn remote_control_waits_for_account_id_before_enrolling() {
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
/*chatgpt_base_url*/ None,
);
)
.await;
let expected_server_name = gethostname().to_string_lossy().trim().to_string();
let expected_enrollment = RemoteControlEnrollment {
account_id: "account_id".to_string(),
@@ -706,7 +706,7 @@ pub(crate) async fn load_remote_control_auth(
"remote control requires ChatGPT authentication",
));
}
auth_manager.reload();
auth_manager.reload().await;
reloaded = true;
continue;
};
@@ -714,7 +714,7 @@ pub(crate) async fn load_remote_control_auth(
break auth;
}
if auth.get_account_id().is_none() && !reloaded {
auth_manager.reload();
auth_manager.reload().await;
reloaded = true;
continue;
}
@@ -1090,7 +1090,8 @@ mod tests {
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
/*chatgpt_base_url*/ None,
);
)
.await;
let mut auth_recovery = auth_manager.unauthorized_recovery();
let mut enrollment = Some(RemoteControlEnrollment {
account_id: "account_id".to_string(),
@@ -1172,7 +1173,8 @@ mod tests {
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
/*chatgpt_base_url*/ None,
);
)
.await;
let mut auth_recovery = auth_manager.unauthorized_recovery();
let mut enrollment = None;
save_auth(
@@ -79,24 +79,6 @@ async fn app_server_default_analytics_enabled_with_flag() -> Result<()> {
Ok(())
}
pub(crate) async fn enable_analytics_capture(server: &MockServer, codex_home: &Path) -> Result<()> {
let config_path = codex_home.join("config.toml");
let config_toml = std::fs::read_to_string(&config_path)?;
if !config_toml.contains("[features]") {
std::fs::write(
&config_path,
format!("{config_toml}\n[features]\ngeneral_analytics = true\n"),
)?;
} else if !config_toml.contains("general_analytics") {
std::fs::write(
&config_path,
config_toml.replace("[features]\n", "[features]\ngeneral_analytics = true\n"),
)?;
}
mount_analytics_capture(server, codex_home).await
}
pub(crate) async fn mount_analytics_capture(server: &MockServer, codex_home: &Path) -> Result<()> {
Mock::given(method("POST"))
.and(path("/codex/analytics-events/events"))
@@ -42,7 +42,7 @@ use wiremock::matchers::method;
use wiremock::matchers::path;
use super::analytics::assert_basic_thread_initialized_event;
use super::analytics::enable_analytics_capture;
use super::analytics::mount_analytics_capture;
use super::analytics::thread_initialized_event;
use super::analytics::wait_for_analytics_payload;
@@ -385,13 +385,8 @@ async fn thread_fork_tracks_thread_initialized_analytics() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml_with_chatgpt_base_url(
codex_home.path(),
&server.uri(),
&server.uri(),
/*general_analytics_enabled*/ true,
)?;
enable_analytics_capture(&server, codex_home.path()).await?;
create_config_toml_with_chatgpt_base_url(codex_home.path(), &server.uri(), &server.uri())?;
mount_analytics_capture(&server, codex_home.path()).await?;
let conversation_id = create_fake_rollout(
codex_home.path(),
@@ -496,7 +491,6 @@ async fn thread_fork_surfaces_cloud_requirements_load_errors() -> Result<()> {
codex_home.path(),
&model_server.uri(),
&chatgpt_base_url,
/*general_analytics_enabled*/ false,
)?;
write_chatgpt_auth(
codex_home.path(),
@@ -793,13 +787,7 @@ fn create_config_toml_with_chatgpt_base_url(
codex_home: &Path,
server_uri: &str,
chatgpt_base_url: &str,
general_analytics_enabled: bool,
) -> std::io::Result<()> {
let general_analytics_toml = if general_analytics_enabled {
"\ngeneral_analytics = true".to_string()
} else {
"\ngeneral_analytics = false".to_string()
};
let config_toml = codex_home.join("config.toml");
std::fs::write(
config_toml,
@@ -812,9 +800,6 @@ chatgpt_base_url = "{chatgpt_base_url}"
model_provider = "mock_provider"
[features]
{general_analytics_toml}
[model_providers.mock_provider]
name = "Mock provider for test"
base_url = "{server_uri}/v1"
@@ -84,7 +84,7 @@ use wiremock::matchers::method;
use wiremock::matchers::path;
use super::analytics::assert_basic_thread_initialized_event;
use super::analytics::enable_analytics_capture;
use super::analytics::mount_analytics_capture;
use super::analytics::thread_initialized_event;
use super::analytics::wait_for_analytics_payload;
@@ -185,10 +185,7 @@ async fn thread_goal_get_rejects_unmaterialized_thread() -> Result<()> {
let config = std::fs::read_to_string(&config_path)?;
std::fs::write(
&config_path,
config.replace(
"general_analytics = true\n",
"general_analytics = true\ngoals = true\n",
),
config.replace("personality = true\n", "personality = true\ngoals = true\n"),
)?;
let mut mcp = McpProcess::new_without_managed_config(codex_home.path()).await?;
@@ -238,13 +235,8 @@ async fn thread_resume_tracks_thread_initialized_analytics() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml_with_chatgpt_base_url(
codex_home.path(),
&server.uri(),
&server.uri(),
/*general_analytics_enabled*/ true,
)?;
enable_analytics_capture(&server, codex_home.path()).await?;
create_config_toml_with_chatgpt_base_url(codex_home.path(), &server.uri(), &server.uri())?;
mount_analytics_capture(&server, codex_home.path()).await?;
let conversation_id = create_fake_rollout_with_text_elements(
codex_home.path(),
@@ -400,10 +392,7 @@ async fn thread_resume_emits_active_goal_update_before_continuation() -> Result<
let config = std::fs::read_to_string(&config_path)?;
std::fs::write(
&config_path,
config.replace(
"general_analytics = true\n",
"general_analytics = true\ngoals = true\n",
),
config.replace("personality = true\n", "personality = true\ngoals = true\n"),
)?;
let mut mcp = McpProcess::new_without_managed_config(codex_home.path()).await?;
@@ -507,10 +496,7 @@ async fn thread_goal_set_preserves_budget_limited_same_objective() -> Result<()>
let config = std::fs::read_to_string(&config_path)?;
std::fs::write(
&config_path,
config.replace(
"general_analytics = true\n",
"general_analytics = true\ngoals = true\n",
),
config.replace("personality = true\n", "personality = true\ngoals = true\n"),
)?;
let mut mcp = McpProcess::new_without_managed_config(codex_home.path()).await?;
@@ -608,10 +594,7 @@ async fn thread_goal_clear_deletes_goal_and_notifies() -> Result<()> {
let config = std::fs::read_to_string(&config_path)?;
std::fs::write(
&config_path,
config.replace(
"general_analytics = true\n",
"general_analytics = true\ngoals = true\n",
),
config.replace("personality = true\n", "personality = true\ngoals = true\n"),
)?;
let mut mcp = McpProcess::new_without_managed_config(codex_home.path()).await?;
@@ -2414,7 +2397,6 @@ async fn thread_resume_surfaces_cloud_requirements_load_errors() -> Result<()> {
codex_home.path(),
&model_server.uri(),
&chatgpt_base_url,
/*general_analytics_enabled*/ false,
)?;
write_chatgpt_auth(
codex_home.path(),
@@ -2859,7 +2841,6 @@ model_provider = "mock_provider"
[features]
personality = true
general_analytics = true
[model_providers.mock_provider]
name = "Mock provider for test"
@@ -2890,7 +2871,6 @@ model_provider = "mock_provider"
[features]
personality = true
general_analytics = true
[model_providers.mock_provider]
name = "Mock provider for test"
@@ -2907,13 +2887,7 @@ fn create_config_toml_with_chatgpt_base_url(
codex_home: &std::path::Path,
server_uri: &str,
chatgpt_base_url: &str,
general_analytics_enabled: bool,
) -> std::io::Result<()> {
let general_analytics_toml = if general_analytics_enabled {
"\ngeneral_analytics = true".to_string()
} else {
"\ngeneral_analytics = false".to_string()
};
let config_toml = codex_home.join("config.toml");
std::fs::write(
config_toml,
@@ -2928,7 +2902,6 @@ model_provider = "mock_provider"
[features]
personality = true
{general_analytics_toml}
[model_providers.mock_provider]
name = "Mock provider for test"
@@ -34,7 +34,6 @@ use serde_json::Value;
use serde_json::json;
use std::path::Path;
use std::path::PathBuf;
use std::time::Duration;
use tempfile::TempDir;
use tokio::time::timeout;
use wiremock::Mock;
@@ -265,12 +264,7 @@ async fn thread_start_tracks_thread_initialized_analytics() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml_with_chatgpt_base_url(
codex_home.path(),
&server.uri(),
&server.uri(),
/*general_analytics_enabled*/ true,
)?;
create_config_toml_with_chatgpt_base_url(codex_home.path(), &server.uri(), &server.uri())?;
mount_analytics_capture(&server, codex_home.path()).await?;
let mut mcp = McpProcess::new_without_managed_config(codex_home.path()).await?;
@@ -293,54 +287,6 @@ async fn thread_start_tracks_thread_initialized_analytics() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn thread_start_does_not_track_thread_initialized_analytics_without_feature() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml_with_chatgpt_base_url(
codex_home.path(),
&server.uri(),
&server.uri(),
/*general_analytics_enabled*/ false,
)?;
mount_analytics_capture(&server, codex_home.path()).await?;
let mut mcp = McpProcess::new_without_managed_config(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let req_id = mcp
.send_thread_start_request(ThreadStartParams::default())
.await?;
let resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(req_id)),
)
.await??;
let _ = to_response::<ThreadStartResponse>(resp)?;
assert_no_thread_initialized_analytics(&server, Duration::from_millis(250)).await?;
Ok(())
}
async fn assert_no_thread_initialized_analytics(
server: &MockServer,
wait_duration: Duration,
) -> Result<()> {
tokio::time::sleep(wait_duration).await;
let requests = server.received_requests().await.unwrap_or_default();
for request in requests.iter().filter(|request| {
request.method == "POST" && request.url.path() == "/codex/analytics-events/events"
}) {
let payload: Value = serde_json::from_slice(&request.body)?;
assert!(
thread_initialized_event(&payload).is_err(),
"thread analytics should be gated off when general_analytics is disabled; payload={payload}"
);
}
Ok(())
}
#[tokio::test]
async fn thread_start_respects_project_config_from_cwd() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
@@ -643,7 +589,6 @@ async fn thread_start_surfaces_cloud_requirements_load_errors() -> Result<()> {
codex_home.path(),
&model_server.uri(),
&chatgpt_base_url,
/*general_analytics_enabled*/ false,
)?;
write_chatgpt_auth(
codex_home.path(),
@@ -966,13 +911,7 @@ fn create_config_toml_with_chatgpt_base_url(
codex_home: &Path,
server_uri: &str,
chatgpt_base_url: &str,
general_analytics_enabled: bool,
) -> std::io::Result<()> {
let general_analytics_toml = if general_analytics_enabled {
"\ngeneral_analytics = true".to_string()
} else {
"\ngeneral_analytics = false".to_string()
};
let config_toml = codex_home.join("config.toml");
std::fs::write(
config_toml,
@@ -985,9 +924,6 @@ chatgpt_base_url = "{chatgpt_base_url}"
model_provider = "mock_provider"
[features]
{general_analytics_toml}
[model_providers.mock_provider]
name = "Mock provider for test"
base_url = "{server_uri}/v1"
@@ -78,7 +78,6 @@ use std::path::Path;
use tempfile::TempDir;
use tokio::time::timeout;
use super::analytics::enable_analytics_capture;
use super::analytics::mount_analytics_capture;
use super::analytics::wait_for_analytics_event;
@@ -463,7 +462,7 @@ async fn turn_start_tracks_turn_event_analytics() -> Result<()> {
&server.uri(),
&server.uri(),
)?;
enable_analytics_capture(&server, codex_home.path()).await?;
mount_analytics_capture(&server, codex_home.path()).await?;
let mut mcp = McpProcess::new_without_managed_config(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
@@ -538,77 +537,6 @@ async fn turn_start_tracks_turn_event_analytics() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn turn_start_does_not_track_turn_event_analytics_without_feature() -> Result<()> {
let responses = vec![create_final_assistant_message_sse_response("Done")?];
let server = create_mock_responses_server_sequence_unchecked(responses).await;
let codex_home = TempDir::new()?;
write_mock_responses_config_toml_with_chatgpt_base_url(
codex_home.path(),
&server.uri(),
&server.uri(),
)?;
let config_path = codex_home.path().join("config.toml");
let config_toml = std::fs::read_to_string(&config_path)?;
std::fs::write(
&config_path,
format!("{config_toml}\n[features]\ngeneral_analytics = false\n"),
)?;
mount_analytics_capture(&server, codex_home.path()).await?;
let mut mcp = McpProcess::new_without_managed_config(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let thread_req = mcp
.send_thread_start_request(ThreadStartParams {
model: Some("mock-model".to_string()),
..Default::default()
})
.await?;
let thread_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(thread_req)),
)
.await??;
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(thread_resp)?;
let turn_req = mcp
.send_turn_start_request(TurnStartParams {
thread_id: thread.id,
input: vec![V2UserInput::Text {
text: "hello".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
let turn_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(turn_req)),
)
.await??;
let _ = to_response::<TurnStartResponse>(turn_resp)?;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("turn/completed"),
)
.await??;
let turn_event = wait_for_analytics_event(
&server,
std::time::Duration::from_millis(250),
"codex_turn_event",
)
.await;
assert!(
turn_event.is_err(),
"turn analytics should be gated off when general_analytics is disabled"
);
Ok(())
}
#[tokio::test]
async fn turn_start_accepts_text_at_limit_with_mention_item() -> Result<()> {
let responses = vec![create_final_assistant_message_sse_response("Done")?];
@@ -24,7 +24,7 @@ use codex_protocol::user_input::MAX_USER_INPUT_TEXT_CHARS;
use tempfile::TempDir;
use tokio::time::timeout;
use super::analytics::enable_analytics_capture;
use super::analytics::mount_analytics_capture;
use super::analytics::wait_for_analytics_event;
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
@@ -41,7 +41,7 @@ async fn turn_steer_requires_active_turn() -> Result<()> {
&server.uri(),
&server.uri(),
)?;
enable_analytics_capture(&server, &codex_home).await?;
mount_analytics_capture(&server, &codex_home).await?;
let mut mcp = McpProcess::new_without_managed_config(&codex_home).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
@@ -125,7 +125,7 @@ async fn turn_steer_rejects_oversized_text_input() -> Result<()> {
&server.uri(),
&server.uri(),
)?;
enable_analytics_capture(&server, &codex_home).await?;
mount_analytics_capture(&server, &codex_home).await?;
let mut mcp = McpProcess::new_without_managed_config(&codex_home).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
@@ -234,7 +234,7 @@ async fn turn_steer_returns_active_turn_id() -> Result<()> {
&server.uri(),
&server.uri(),
)?;
enable_analytics_capture(&server, &codex_home).await?;
mount_analytics_capture(&server, &codex_home).await?;
let mut mcp = McpProcess::new_without_managed_config(&codex_home).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
+1 -1
View File
@@ -21,7 +21,7 @@ pub(crate) async fn chatgpt_get_request_with_timeout<T: DeserializeOwned>(
) -> anyhow::Result<T> {
let chatgpt_base_url = &config.chatgpt_base_url;
let auth_manager =
AuthManager::shared_from_config(config, /*enable_codex_api_key_env*/ false);
AuthManager::shared_from_config(config, /*enable_codex_api_key_env*/ false).await;
let auth = auth_manager
.auth()
.await
+2 -2
View File
@@ -26,7 +26,7 @@ const DIRECTORY_CONNECTORS_TIMEOUT: Duration = Duration::from_secs(60);
async fn apps_enabled(config: &Config) -> bool {
let auth_manager =
AuthManager::shared_from_config(config, /*enable_codex_api_key_env*/ false);
AuthManager::shared_from_config(config, /*enable_codex_api_key_env*/ false).await;
let auth = auth_manager.auth().await;
config
.features
@@ -35,7 +35,7 @@ async fn apps_enabled(config: &Config) -> bool {
async fn connector_auth(config: &Config) -> anyhow::Result<CodexAuth> {
let auth_manager =
AuthManager::shared_from_config(config, /*enable_codex_api_key_env*/ false);
AuthManager::shared_from_config(config, /*enable_codex_api_key_env*/ false).await;
let auth = auth_manager
.auth()
.await
+3 -1
View File
@@ -362,7 +362,9 @@ pub async fn run_login_with_device_code_fallback_to_browser(
pub async fn run_login_status(cli_config_overrides: CliConfigOverrides) -> ! {
let config = load_config_or_exit(cli_config_overrides).await;
match CodexAuth::from_auth_storage(&config.codex_home, config.cli_auth_credentials_store_mode) {
match CodexAuth::from_auth_storage(&config.codex_home, config.cli_auth_credentials_store_mode)
.await
{
Ok(Some(auth)) => match auth.auth_mode() {
AuthMode::ApiKey => match auth.get_token() {
Ok(api_key) => {
+1 -1
View File
@@ -1384,7 +1384,7 @@ async fn run_debug_models_command(
.map_err(anyhow::Error::msg)?;
let config = Config::load_with_cli_overrides(cli_overrides).await?;
let auth_manager =
AuthManager::shared_from_config(&config, /*enable_codex_api_key_env*/ true);
AuthManager::shared_from_config(&config, /*enable_codex_api_key_env*/ true).await;
let models_manager =
build_models_manager(&config, auth_manager, CollaborationModesConfig::default());
models_manager
+92 -67
View File
@@ -727,7 +727,7 @@ pub fn cloud_requirements_loader(
})
}
pub fn cloud_requirements_loader_for_storage(
pub async fn cloud_requirements_loader_for_storage(
codex_home: PathBuf,
enable_codex_api_key_env: bool,
credentials_store_mode: AuthCredentialsStoreMode,
@@ -738,7 +738,8 @@ pub fn cloud_requirements_loader_for_storage(
enable_codex_api_key_env,
credentials_store_mode,
Some(chatgpt_base_url.clone()),
);
)
.await;
cloud_requirements_loader(auth_manager, chatgpt_base_url, codex_home)
}
@@ -853,7 +854,7 @@ mod tests {
Ok(())
}
fn auth_manager_with_api_key() -> Arc<AuthManager> {
async fn auth_manager_with_api_key() -> Arc<AuthManager> {
let tmp = tempdir().expect("tempdir");
let auth_json = json!({
"OPENAI_API_KEY": "sk-test-key",
@@ -861,15 +862,18 @@ mod tests {
"last_refresh": null,
});
write_auth_json(tmp.path(), auth_json).expect("write auth");
Arc::new(AuthManager::new(
tmp.path().to_path_buf(),
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
/*chatgpt_base_url*/ None,
))
Arc::new(
AuthManager::new(
tmp.path().to_path_buf(),
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
/*chatgpt_base_url*/ None,
)
.await,
)
}
fn auth_manager_with_plan_and_identity(
async fn auth_manager_with_plan_and_identity(
plan_type: &str,
chatgpt_user_id: Option<&str>,
account_id: Option<&str>,
@@ -886,12 +890,15 @@ mod tests {
),
)
.expect("write auth");
Arc::new(AuthManager::new(
tmp.path().to_path_buf(),
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
/*chatgpt_base_url*/ None,
))
Arc::new(
AuthManager::new(
tmp.path().to_path_buf(),
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
/*chatgpt_base_url*/ None,
)
.await,
)
}
fn chatgpt_auth_json(
@@ -975,7 +982,7 @@ mod tests {
manager: Arc<AuthManager>,
}
fn managed_auth_context(
async fn managed_auth_context(
plan_type: &str,
chatgpt_user_id: Option<&str>,
account_id: Option<&str>,
@@ -995,18 +1002,22 @@ mod tests {
)
.expect("write auth");
ManagedAuthContext {
manager: Arc::new(AuthManager::new(
home.path().to_path_buf(),
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
/*chatgpt_base_url*/ None,
)),
manager: Arc::new(
AuthManager::new(
home.path().to_path_buf(),
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
/*chatgpt_base_url*/ None,
)
.await,
),
_home: home,
}
}
fn auth_manager_with_plan(plan_type: &str) -> Arc<AuthManager> {
async fn auth_manager_with_plan(plan_type: &str) -> Arc<AuthManager> {
auth_manager_with_plan_and_identity(plan_type, Some("user-12345"), Some("account-12345"))
.await
}
fn parse_for_fetch(contents: Option<&str>) -> Option<ConfigRequirementsToml> {
@@ -1118,7 +1129,7 @@ mod tests {
#[tokio::test]
async fn fetch_cloud_requirements_skips_non_chatgpt_auth() {
let auth_manager = auth_manager_with_api_key();
let auth_manager = auth_manager_with_api_key().await;
let codex_home = tempdir().expect("tempdir");
let service = CloudRequirementsService::new(
auth_manager,
@@ -1134,7 +1145,7 @@ mod tests {
async fn fetch_cloud_requirements_skips_non_business_or_enterprise_plan() {
let codex_home = tempdir().expect("tempdir");
let service = CloudRequirementsService::new(
auth_manager_with_plan("pro"),
auth_manager_with_plan("pro").await,
Arc::new(StaticFetcher { contents: None }),
codex_home.path().to_path_buf(),
CLOUD_REQUIREMENTS_TIMEOUT,
@@ -1147,7 +1158,7 @@ mod tests {
async fn fetch_cloud_requirements_skips_team_like_usage_based_plan() {
let codex_home = tempdir().expect("tempdir");
let service = CloudRequirementsService::new(
auth_manager_with_plan("self_serve_business_usage_based"),
auth_manager_with_plan("self_serve_business_usage_based").await,
Arc::new(StaticFetcher {
contents: Some("allowed_approval_policies = [\"never\"]".to_string()),
}),
@@ -1161,7 +1172,7 @@ mod tests {
async fn fetch_cloud_requirements_allows_business_plan() {
let codex_home = tempdir().expect("tempdir");
let service = CloudRequirementsService::new(
auth_manager_with_plan("business"),
auth_manager_with_plan("business").await,
Arc::new(StaticFetcher {
contents: Some("allowed_approval_policies = [\"never\"]".to_string()),
}),
@@ -1193,7 +1204,7 @@ mod tests {
async fn fetch_cloud_requirements_allows_business_like_usage_based_plan() {
let codex_home = tempdir().expect("tempdir");
let service = CloudRequirementsService::new(
auth_manager_with_plan("enterprise_cbp_usage_based"),
auth_manager_with_plan("enterprise_cbp_usage_based").await,
Arc::new(StaticFetcher {
contents: Some("allowed_approval_policies = [\"never\"]".to_string()),
}),
@@ -1225,7 +1236,7 @@ mod tests {
async fn fetch_cloud_requirements_allows_hc_plan_as_enterprise() {
let codex_home = tempdir().expect("tempdir");
let service = CloudRequirementsService::new(
auth_manager_with_plan("hc"),
auth_manager_with_plan("hc").await,
Arc::new(StaticFetcher {
contents: Some("allowed_approval_policies = [\"never\"]".to_string()),
}),
@@ -1329,7 +1340,7 @@ enabled = false
#[tokio::test(start_paused = true)]
async fn fetch_cloud_requirements_times_out() {
let auth_manager = auth_manager_with_plan("enterprise");
let auth_manager = auth_manager_with_plan("enterprise").await;
let codex_home = tempdir().expect("tempdir");
let service = CloudRequirementsService::new(
auth_manager,
@@ -1356,7 +1367,7 @@ enabled = false
]));
let codex_home = tempdir().expect("tempdir");
let service = CloudRequirementsService::new(
auth_manager_with_plan("business"),
auth_manager_with_plan("business").await,
fetcher.clone(),
codex_home.path().to_path_buf(),
CLOUD_REQUIREMENTS_TIMEOUT,
@@ -1405,12 +1416,15 @@ enabled = false
),
)
.expect("write initial auth");
let auth_manager = Arc::new(AuthManager::new(
auth_home.path().to_path_buf(),
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
/*chatgpt_base_url*/ None,
));
let auth_manager = Arc::new(
AuthManager::new(
auth_home.path().to_path_buf(),
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
/*chatgpt_base_url*/ None,
)
.await,
);
write_auth_json(
auth_home.path(),
@@ -1479,12 +1493,15 @@ enabled = false
),
)
.expect("write initial auth");
let auth_manager = Arc::new(AuthManager::new(
auth_home.path().to_path_buf(),
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
/*chatgpt_base_url*/ None,
));
let auth_manager = Arc::new(
AuthManager::new(
auth_home.path().to_path_buf(),
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
/*chatgpt_base_url*/ None,
)
.await,
);
write_auth_json(
auth_home.path(),
@@ -1559,7 +1576,8 @@ enabled = false
Some("account-12345"),
"stale-access-token",
"test-refresh-token",
);
)
.await;
write_auth_json(
auth._home.path(),
chatgpt_auth_json(
@@ -1611,12 +1629,15 @@ enabled = false
),
)
.expect("write auth");
let auth_manager = Arc::new(AuthManager::new(
auth_home.path().to_path_buf(),
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
/*chatgpt_base_url*/ None,
));
let auth_manager = Arc::new(
AuthManager::new(
auth_home.path().to_path_buf(),
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
/*chatgpt_base_url*/ None,
)
.await,
);
let fetcher = Arc::new(UnauthorizedFetcher {
message:
@@ -1653,7 +1674,7 @@ enabled = false
]));
let codex_home = tempdir().expect("tempdir");
let service = CloudRequirementsService::new(
auth_manager_with_plan("business"),
auth_manager_with_plan("business").await,
fetcher.clone(),
codex_home.path().to_path_buf(),
CLOUD_REQUIREMENTS_TIMEOUT,
@@ -1678,7 +1699,7 @@ enabled = false
))]));
let codex_home = tempdir().expect("tempdir");
let service = CloudRequirementsService::new(
auth_manager_with_plan("business"),
auth_manager_with_plan("business").await,
fetcher,
codex_home.path().to_path_buf(),
CLOUD_REQUIREMENTS_TIMEOUT,
@@ -1700,7 +1721,7 @@ enabled = false
async fn fetch_cloud_requirements_uses_cache_when_valid() {
let codex_home = tempdir().expect("tempdir");
let prime_service = CloudRequirementsService::new(
auth_manager_with_plan("business"),
auth_manager_with_plan("business").await,
Arc::new(StaticFetcher {
contents: Some("allowed_approval_policies = [\"never\"]".to_string()),
}),
@@ -1711,7 +1732,7 @@ enabled = false
let fetcher = Arc::new(SequenceFetcher::new(vec![Err(request_error())]));
let service = CloudRequirementsService::new(
auth_manager_with_plan("business"),
auth_manager_with_plan("business").await,
fetcher.clone(),
codex_home.path().to_path_buf(),
CLOUD_REQUIREMENTS_TIMEOUT,
@@ -1747,7 +1768,8 @@ enabled = false
"business",
/*chatgpt_user_id*/ None,
Some("account-12345"),
),
)
.await,
Arc::new(StaticFetcher {
contents: Some("allowed_approval_policies = [\"never\"]".to_string()),
}),
@@ -1790,7 +1812,7 @@ enabled = false
async fn fetch_cloud_requirements_does_not_use_cache_when_auth_identity_is_incomplete() {
let codex_home = tempdir().expect("tempdir");
let prime_service = CloudRequirementsService::new(
auth_manager_with_plan("business"),
auth_manager_with_plan("business").await,
Arc::new(StaticFetcher {
contents: Some("allowed_approval_policies = [\"never\"]".to_string()),
}),
@@ -1807,7 +1829,8 @@ enabled = false
"business",
/*chatgpt_user_id*/ None,
Some("account-12345"),
),
)
.await,
fetcher.clone(),
codex_home.path().to_path_buf(),
CLOUD_REQUIREMENTS_TIMEOUT,
@@ -1843,7 +1866,8 @@ enabled = false
"business",
Some("user-12345"),
Some("account-12345"),
),
)
.await,
Arc::new(StaticFetcher {
contents: Some("allowed_approval_policies = [\"never\"]".to_string()),
}),
@@ -1860,7 +1884,8 @@ enabled = false
"business",
Some("user-99999"),
Some("account-12345"),
),
)
.await,
fetcher.clone(),
codex_home.path().to_path_buf(),
CLOUD_REQUIREMENTS_TIMEOUT,
@@ -1892,7 +1917,7 @@ enabled = false
async fn fetch_cloud_requirements_ignores_tampered_cache() {
let codex_home = tempdir().expect("tempdir");
let prime_service = CloudRequirementsService::new(
auth_manager_with_plan("business"),
auth_manager_with_plan("business").await,
Arc::new(StaticFetcher {
contents: Some("allowed_approval_policies = [\"never\"]".to_string()),
}),
@@ -1917,7 +1942,7 @@ enabled = false
"allowed_approval_policies = [\"never\"]".to_string(),
))]));
let service = CloudRequirementsService::new(
auth_manager_with_plan("enterprise"),
auth_manager_with_plan("enterprise").await,
fetcher.clone(),
codex_home.path().to_path_buf(),
CLOUD_REQUIREMENTS_TIMEOUT,
@@ -1975,7 +2000,7 @@ enabled = false
"allowed_approval_policies = [\"never\"]".to_string(),
))]));
let service = CloudRequirementsService::new(
auth_manager_with_plan("enterprise"),
auth_manager_with_plan("enterprise").await,
fetcher.clone(),
codex_home.path().to_path_buf(),
CLOUD_REQUIREMENTS_TIMEOUT,
@@ -2007,7 +2032,7 @@ enabled = false
async fn fetch_cloud_requirements_writes_signed_cache() {
let codex_home = tempdir().expect("tempdir");
let service = CloudRequirementsService::new(
auth_manager_with_plan("business"),
auth_manager_with_plan("business").await,
Arc::new(StaticFetcher {
contents: Some("allowed_approval_policies = [\"never\"]".to_string()),
}),
@@ -2070,7 +2095,7 @@ enabled = false
let fetcher = Arc::new(SequenceFetcher::new(vec![Ok(None), Err(request_error())]));
let codex_home = tempdir().expect("tempdir");
let service = CloudRequirementsService::new(
auth_manager_with_plan("enterprise"),
auth_manager_with_plan("enterprise").await,
fetcher.clone(),
codex_home.path().to_path_buf(),
CLOUD_REQUIREMENTS_TIMEOUT,
@@ -2088,7 +2113,7 @@ enabled = false
]));
let codex_home = tempdir().expect("tempdir");
let service = CloudRequirementsService::new(
auth_manager_with_plan("enterprise"),
auth_manager_with_plan("enterprise").await,
fetcher.clone(),
codex_home.path().to_path_buf(),
CLOUD_REQUIREMENTS_TIMEOUT,
@@ -2121,7 +2146,7 @@ enabled = false
)),
]));
let service = CloudRequirementsService::new(
auth_manager_with_plan("business"),
auth_manager_with_plan("business").await,
fetcher,
codex_home.path().to_path_buf(),
CLOUD_REQUIREMENTS_TIMEOUT,
+9 -6
View File
@@ -44,12 +44,15 @@ pub fn normalize_base_url(input: &str) -> String {
pub async fn load_auth_manager(chatgpt_base_url: Option<String>) -> Option<AuthManager> {
// TODO: pass in cli overrides once cloud tasks properly support them.
let config = Config::load_with_cli_overrides(Vec::new()).await.ok()?;
Some(AuthManager::new(
config.codex_home.to_path_buf(),
/*enable_codex_api_key_env*/ false,
config.cli_auth_credentials_store_mode,
chatgpt_base_url.or(Some(config.chatgpt_base_url)),
))
Some(
AuthManager::new(
config.codex_home.to_path_buf(),
/*enable_codex_api_key_env*/ false,
config.cli_auth_credentials_store_mode,
chatgpt_base_url.or(Some(config.chatgpt_base_url)),
)
.await,
)
}
/// Build headers for ChatGPT-backed requests: `User-Agent`, optional `Authorization`,
-6
View File
@@ -421,9 +421,6 @@
"fast_mode": {
"type": "boolean"
},
"general_analytics": {
"type": "boolean"
},
"goals": {
"type": "boolean"
},
@@ -2631,9 +2628,6 @@
"fast_mode": {
"type": "boolean"
},
"general_analytics": {
"type": "boolean"
},
"goals": {
"type": "boolean"
},
-1
View File
@@ -262,7 +262,6 @@ impl AgentControl {
parent_thread_id, ..
},
)) = notification_source.as_ref()
&& new_thread.thread.enabled(Feature::GeneralAnalytics)
{
let client_metadata = match state.get_thread(*parent_thread_id).await {
Ok(parent_thread) => {
+10 -12
View File
@@ -104,18 +104,16 @@ pub(crate) async fn run_codex_thread_interactive(
}))
.or_cancel(&cancel_token)
.await??;
if parent_session.enabled(codex_features::Feature::GeneralAnalytics) {
let thread_config = codex.thread_config_snapshot().await;
let client_metadata = parent_session.app_server_client_metadata().await;
emit_subagent_session_started(
&parent_session.services.analytics_events_client,
client_metadata,
codex.session.conversation_id,
Some(parent_session.conversation_id),
thread_config,
subagent_source,
);
}
let thread_config = codex.thread_config_snapshot().await;
let client_metadata = parent_session.app_server_client_metadata().await;
emit_subagent_session_started(
&parent_session.services.analytics_events_client,
client_metadata,
codex.session.conversation_id,
Some(parent_session.conversation_id),
thread_config,
subagent_source,
);
let codex = Arc::new(codex);
// Use a child token so parent cancel cascades but we can scope it to this task
-7
View File
@@ -18,7 +18,6 @@ use codex_analytics::CompactionStatus;
use codex_analytics::CompactionStrategy;
use codex_analytics::CompactionTrigger;
use codex_analytics::now_unix_seconds;
use codex_features::Feature;
use codex_protocol::error::CodexErr;
use codex_protocol::error::Result as CodexResult;
use codex_protocol::items::ContextCompactionItem;
@@ -295,7 +294,6 @@ async fn run_compact_task_inner_impl(
}
pub(crate) struct CompactionAnalyticsAttempt {
enabled: bool,
thread_id: String,
turn_id: String,
trigger: CompactionTrigger,
@@ -316,10 +314,8 @@ impl CompactionAnalyticsAttempt {
implementation: CompactionImplementation,
phase: CompactionPhase,
) -> Self {
let enabled = sess.enabled(Feature::GeneralAnalytics);
let active_context_tokens_before = sess.get_total_token_usage().await;
Self {
enabled,
thread_id: sess.conversation_id.to_string(),
turn_id: turn_context.sub_id.clone(),
trigger,
@@ -338,9 +334,6 @@ impl CompactionAnalyticsAttempt {
status: CompactionStatus,
error: Option<String>,
) {
if !self.enabled {
return;
}
let active_context_tokens_after = sess.get_total_token_usage().await;
sess.services
.analytics_events_client
+3 -3
View File
@@ -144,7 +144,7 @@ pub async fn list_cached_accessible_connectors_from_mcp_tools(
config: &Config,
) -> Option<Vec<AppInfo>> {
let auth_manager =
AuthManager::shared_from_config(config, /*enable_codex_api_key_env*/ false);
AuthManager::shared_from_config(config, /*enable_codex_api_key_env*/ false).await;
let auth = auth_manager.auth().await;
if !config
.features
@@ -216,7 +216,7 @@ pub async fn list_accessible_connectors_from_mcp_tools_with_environment_manager(
environment_manager: &EnvironmentManager,
) -> anyhow::Result<AccessibleConnectorsStatus> {
let auth_manager =
AuthManager::shared_from_config(config, /*enable_codex_api_key_env*/ false);
AuthManager::shared_from_config(config, /*enable_codex_api_key_env*/ false).await;
let auth = auth_manager.auth().await;
if !config
.features
@@ -434,7 +434,7 @@ async fn list_directory_connectors_for_tool_suggest_with_auth(
Some(auth)
} else {
let auth_manager =
AuthManager::shared_from_config(config, /*enable_codex_api_key_env*/ false);
AuthManager::shared_from_config(config, /*enable_codex_api_key_env*/ false).await;
loaded_auth = auth_manager.auth().await;
loaded_auth.as_ref()
};
@@ -519,6 +519,10 @@ const RESIZED_IMAGE_BYTES_ESTIMATE: i64 = 7373;
// Use a direct 32px patch count only for `detail: "original"`;
// all other image inputs continue to use `RESIZED_IMAGE_BYTES_ESTIMATE`.
const ORIGINAL_IMAGE_PATCH_SIZE: u32 = 32;
// See https://platform.openai.com/docs/guides/images-vision#model-sizing-behavior.
// Keep this hard-coded for now; move it into model capabilities if the patch
// budget starts changing often across model releases.
const ORIGINAL_IMAGE_MAX_PATCHES: usize = 10_000;
const ORIGINAL_IMAGE_ESTIMATE_CACHE_SIZE: usize = 32;
static ORIGINAL_IMAGE_ESTIMATE_CACHE: LazyLock<BlockingLruCache<[u8; 20], Option<i64>>> =
@@ -621,6 +625,7 @@ fn estimate_original_image_bytes(image_url: &str) -> Option<i64> {
let patches_high = height.saturating_add(patch_size.saturating_sub(1)) / patch_size;
let patch_count = patches_wide.saturating_mul(patches_high);
let patch_count = usize::try_from(patch_count).unwrap_or(usize::MAX);
let patch_count = patch_count.min(ORIGINAL_IMAGE_MAX_PATCHES);
Some(i64::try_from(approx_bytes_for_tokens(patch_count)).unwrap_or(i64::MAX))
})
}
@@ -26,6 +26,7 @@ use codex_utils_output_truncation::TruncationPolicy;
use codex_utils_output_truncation::truncate_text;
use image::ImageBuffer;
use image::ImageFormat;
use image::Luma;
use image::Rgba;
use pretty_assertions::assert_eq;
use regex_lite::Regex;
@@ -1900,6 +1901,38 @@ fn original_detail_images_scale_with_dimensions() {
assert_eq!(estimated, expected);
}
#[test]
fn original_detail_images_are_capped_at_max_patch_count() {
// 3201x3201 at 32px patches yields 101 * 101 = 10,201 patches,
// which exceeds the original-detail patch budget.
let width = 3201;
let height = 3201;
let image = ImageBuffer::from_pixel(width, height, Luma([12u8]));
let mut bytes = std::io::Cursor::new(Vec::new());
image
.write_to(&mut bytes, ImageFormat::Png)
.expect("encode png");
let payload = BASE64_STANDARD.encode(bytes.get_ref());
let image_url = format!("data:image/png;base64,{payload}");
let item = ResponseItem::FunctionCallOutput {
call_id: "call-original-capped".to_string(),
output: FunctionCallOutputPayload::from_content_items(vec![
FunctionCallOutputContentItem::InputImage {
image_url,
detail: Some(ImageDetail::Original),
},
]),
};
let raw_len = serde_json::to_string(&item).unwrap().len() as i64;
let estimated = estimate_response_item_model_visible_bytes(&item);
let capped_original_detail_image_bytes =
i64::try_from(approx_bytes_for_tokens(ORIGINAL_IMAGE_MAX_PATCHES)).unwrap();
let expected = raw_len - payload.len() as i64 + capped_original_detail_image_bytes;
assert_eq!(estimated, expected);
}
#[test]
fn original_detail_webp_images_scale_with_dimensions() {
// Same dimensions as the PNG case above, so the patch-based replacement cost is the same.
-10
View File
@@ -6,7 +6,6 @@ use codex_analytics::GuardianReviewDecision;
use codex_analytics::GuardianReviewFailureReason;
use codex_analytics::GuardianReviewTerminalStatus;
use codex_analytics::GuardianReviewTrackContext;
use codex_features::Feature;
use codex_protocol::config_types::ApprovalsReviewer;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::EventMsg;
@@ -161,13 +160,9 @@ pub(crate) fn is_guardian_reviewer_source(
fn track_guardian_review(
session: &Session,
turn: &TurnContext,
tracking: &GuardianReviewTrackContext,
result: GuardianReviewAnalyticsResult,
) {
if !turn.config.features.enabled(Feature::GeneralAnalytics) {
return;
}
session
.services
.analytics_events_client
@@ -279,7 +274,6 @@ async fn run_guardian_review(
{
track_guardian_review(
session.as_ref(),
turn.as_ref(),
&review_tracking,
GuardianReviewAnalyticsResult {
decision: GuardianReviewDecision::Aborted,
@@ -325,7 +319,6 @@ async fn run_guardian_review(
let approved = matches!(assessment.outcome, GuardianAssessmentOutcome::Allow);
track_guardian_review(
session.as_ref(),
turn.as_ref(),
&review_tracking,
GuardianReviewAnalyticsResult {
decision: if approved {
@@ -356,7 +349,6 @@ async fn run_guardian_review(
.to_string();
track_guardian_review(
session.as_ref(),
turn.as_ref(),
&review_tracking,
GuardianReviewAnalyticsResult {
decision: GuardianReviewDecision::Denied,
@@ -395,7 +387,6 @@ async fn run_guardian_review(
GuardianReviewError::Cancelled => {
track_guardian_review(
session.as_ref(),
turn.as_ref(),
&review_tracking,
GuardianReviewAnalyticsResult {
decision: GuardianReviewDecision::Aborted,
@@ -437,7 +428,6 @@ async fn run_guardian_review(
let rationale = format!("Automatic approval review failed: {message}");
track_guardian_review(
session.as_ref(),
turn.as_ref(),
&review_tracking,
GuardianReviewAnalyticsResult {
decision: GuardianReviewDecision::Denied,
+9 -11
View File
@@ -163,17 +163,15 @@ pub(super) async fn run(session: &Arc<Session>, config: Arc<Config>) {
.get_agent_config_snapshot(thread_id)
.await
{
if session.enabled(Feature::GeneralAnalytics) {
let client_metadata = session.app_server_client_metadata().await;
emit_subagent_session_started(
&session.services.analytics_events_client,
client_metadata,
thread_id,
/*parent_thread_id*/ None,
thread_config,
SubAgentSource::MemoryConsolidation,
);
}
let client_metadata = session.app_server_client_metadata().await;
emit_subagent_session_started(
&session.services.analytics_events_client,
client_metadata,
thread_id,
/*parent_thread_id*/ None,
thread_config,
SubAgentSource::MemoryConsolidation,
);
} else {
warn!("failed to load memory consolidation thread config for analytics: {thread_id}");
}
+1 -1
View File
@@ -29,7 +29,7 @@ pub async fn build_prompt_input(
config.ephemeral = true;
let auth_manager =
AuthManager::shared_from_config(&config, /*enable_codex_api_key_env*/ false);
AuthManager::shared_from_config(&config, /*enable_codex_api_key_env*/ false).await;
let local_runtime_paths = ExecServerRuntimePaths::from_optional_paths(
config.codex_self_exe.clone(),
+1 -52
View File
@@ -95,9 +95,7 @@ use codex_protocol::protocol::ReasoningRawContentDeltaEvent;
use codex_protocol::protocol::TurnDiffEvent;
use codex_protocol::protocol::WarningEvent;
use codex_protocol::user_input::UserInput;
use codex_tools::ResponsesApiNamespaceTool;
use codex_tools::ToolName;
use codex_tools::ToolSpec;
use codex_tools::filter_tool_suggest_discoverable_tools_for_client;
use codex_utils_stream_parser::AssistantTextChunk;
use codex_utils_stream_parser::AssistantTextStreamParser;
@@ -664,10 +662,6 @@ async fn track_turn_resolved_config_analytics(
turn_context: &TurnContext,
input: &[UserInput],
) {
if !sess.enabled(Feature::GeneralAnalytics) {
return;
}
let thread_config = {
let state = sess.state.lock().await;
state.session_configuration.thread_config_snapshot()
@@ -946,25 +940,9 @@ pub(crate) fn build_prompt(
turn_context: &TurnContext,
base_instructions: BaseInstructions,
) -> Prompt {
let deferred_dynamic_tools = turn_context
.dynamic_tools
.iter()
.filter(|tool| tool.defer_loading)
.map(|tool| ToolName::new(tool.namespace.clone(), tool.name.clone()))
.collect::<HashSet<_>>();
let tools = if deferred_dynamic_tools.is_empty() {
router.model_visible_specs()
} else {
router
.model_visible_specs()
.into_iter()
.filter_map(|spec| filter_deferred_dynamic_tool_spec(spec, &deferred_dynamic_tools))
.collect()
};
Prompt {
input,
tools,
tools: router.model_visible_specs(),
parallel_tool_calls: turn_context.model_info.supports_parallel_tool_calls,
base_instructions,
personality: turn_context.personality,
@@ -975,35 +953,6 @@ pub(crate) fn build_prompt(
}
}
fn filter_deferred_dynamic_tool_spec(
spec: ToolSpec,
deferred_dynamic_tools: &HashSet<ToolName>,
) -> Option<ToolSpec> {
match spec {
ToolSpec::Function(tool) => {
if deferred_dynamic_tools.contains(&ToolName::plain(tool.name.as_str())) {
None
} else {
Some(ToolSpec::Function(tool))
}
}
ToolSpec::Namespace(mut namespace) => {
let namespace_name = namespace.name.clone();
namespace.tools.retain(|tool| match tool {
ResponsesApiNamespaceTool::Function(tool) => !deferred_dynamic_tools.contains(
&ToolName::namespaced(namespace_name.as_str(), tool.name.as_str()),
),
});
if namespace.tools.is_empty() {
None
} else {
Some(ToolSpec::Namespace(namespace))
}
}
spec => Some(spec),
}
}
#[allow(clippy::too_many_arguments)]
#[instrument(level = "trace",
skip_all,
+53 -17
View File
@@ -71,23 +71,26 @@ impl ToolRouter {
dynamic_tools,
);
let (specs, registry) = builder.build();
let model_visible_specs = if config.code_mode_only_enabled {
specs
.iter()
.filter_map(|configured_tool| {
if !codex_code_mode::is_code_mode_nested_tool(configured_tool.name()) {
Some(configured_tool.spec.clone())
} else {
None
}
})
.collect()
} else {
specs
.iter()
.map(|configured_tool| configured_tool.spec.clone())
.collect()
};
let deferred_dynamic_tools = dynamic_tools
.iter()
.filter(|tool| tool.defer_loading)
.map(|tool| ToolName::new(tool.namespace.clone(), tool.name.clone()))
.collect::<HashSet<_>>();
let model_visible_specs = specs
.iter()
.filter_map(|configured_tool| {
if config.code_mode_only_enabled
&& codex_code_mode::is_code_mode_nested_tool(configured_tool.name())
{
return None;
}
filter_deferred_dynamic_tool_spec(
configured_tool.spec.clone(),
&deferred_dynamic_tools,
)
})
.collect();
Self {
registry,
@@ -293,6 +296,39 @@ impl ToolRouter {
self.registry.dispatch_any(invocation).await
}
}
fn filter_deferred_dynamic_tool_spec(
spec: ToolSpec,
deferred_dynamic_tools: &HashSet<ToolName>,
) -> Option<ToolSpec> {
if deferred_dynamic_tools.is_empty() {
return Some(spec);
}
match spec {
ToolSpec::Function(tool) => {
if deferred_dynamic_tools.contains(&ToolName::plain(tool.name.as_str())) {
None
} else {
Some(ToolSpec::Function(tool))
}
}
ToolSpec::Namespace(mut namespace) => {
let namespace_name = namespace.name.clone();
namespace.tools.retain(|tool| match tool {
ResponsesApiNamespaceTool::Function(tool) => !deferred_dynamic_tools.contains(
&ToolName::namespaced(namespace_name.as_str(), tool.name.as_str()),
),
});
if namespace.tools.is_empty() {
None
} else {
Some(ToolSpec::Namespace(namespace))
}
}
spec => Some(spec),
}
}
#[cfg(test)]
#[path = "router_tests.rs"]
mod tests;
+88
View File
@@ -3,8 +3,13 @@ use std::sync::Arc;
use crate::session::tests::make_session_and_context;
use crate::tools::context::ToolPayload;
use codex_protocol::dynamic_tools::DynamicToolSpec;
use codex_protocol::models::ResponseItem;
use codex_tools::ResponsesApiNamespaceTool;
use codex_tools::ToolName;
use codex_tools::ToolSpec;
use pretty_assertions::assert_eq;
use serde_json::json;
use super::ToolCall;
use super::ToolRouter;
@@ -133,3 +138,86 @@ async fn mcp_parallel_support_uses_exact_payload_server() -> anyhow::Result<()>
Ok(())
}
#[tokio::test]
async fn model_visible_specs_filter_deferred_dynamic_tools() -> anyhow::Result<()> {
let (_, turn) = make_session_and_context().await;
let hidden_tool = "hidden_dynamic_tool";
let visible_tool = "visible_dynamic_tool";
let dynamic_tools = vec![
DynamicToolSpec {
namespace: Some("codex_app".to_string()),
name: hidden_tool.to_string(),
description: "Hidden until discovered.".to_string(),
input_schema: json!({
"type": "object",
"properties": {},
"additionalProperties": false,
}),
defer_loading: true,
},
DynamicToolSpec {
namespace: Some("codex_app".to_string()),
name: visible_tool.to_string(),
description: "Visible immediately.".to_string(),
input_schema: json!({
"type": "object",
"properties": {},
"additionalProperties": false,
}),
defer_loading: false,
},
];
let router = ToolRouter::from_config(
&turn.tools_config,
ToolRouterParams {
deferred_mcp_tools: None,
mcp_tools: None,
unavailable_called_tools: Vec::new(),
parallel_mcp_server_names: HashSet::new(),
discoverable_tools: None,
dynamic_tools: &dynamic_tools,
},
);
assert!(
router
.find_spec(&ToolName::namespaced("codex_app", hidden_tool))
.is_some()
);
assert_eq!(
namespace_function_names(&router.specs(), "codex_app"),
vec![hidden_tool.to_string(), visible_tool.to_string()]
);
assert_eq!(
namespace_function_names(&router.model_visible_specs(), "codex_app"),
vec![visible_tool.to_string()]
);
Ok(())
}
fn namespace_function_names(specs: &[ToolSpec], namespace_name: &str) -> Vec<String> {
specs
.iter()
.find_map(|spec| match spec {
ToolSpec::Namespace(namespace) if namespace.name == namespace_name => Some(
namespace
.tools
.iter()
.map(|tool| match tool {
ResponsesApiNamespaceTool::Function(tool) => tool.name.clone(),
})
.collect(),
),
ToolSpec::Function(_)
| ToolSpec::Freeform(_)
| ToolSpec::ToolSearch { .. }
| ToolSpec::LocalShell {}
| ToolSpec::ImageGeneration { .. }
| ToolSpec::WebSearch { .. }
| ToolSpec::Namespace(_) => None,
})
.unwrap_or_default()
}
+2 -1
View File
@@ -1091,7 +1091,8 @@ async fn prefers_apikey_when_config_prefers_apikey_even_with_chatgpt_tokens() {
config.model_provider = model_provider;
let auth_manager =
match CodexAuth::from_auth_storage(codex_home.path(), AuthCredentialsStoreMode::File) {
match CodexAuth::from_auth_storage(codex_home.path(), AuthCredentialsStoreMode::File).await
{
Ok(Some(auth)) => codex_core::test_support::auth_manager_from_auth(auth),
Ok(None) => panic!("No CodexAuth found in codex_home"),
Err(e) => panic!("Failed to load CodexAuth: {e}"),
+143
View File
@@ -6,6 +6,7 @@ use std::path::PathBuf;
use anyhow::Result;
use codex_core::compact::SUMMARY_PREFIX;
use codex_login::CodexAuth;
use codex_protocol::dynamic_tools::DynamicToolSpec;
use codex_protocol::items::TurnItem;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseItem;
@@ -36,6 +37,7 @@ use core_test_support::wait_for_event;
use core_test_support::wait_for_event_match;
use core_test_support::wait_for_event_with_timeout;
use pretty_assertions::assert_eq;
use serde_json::Value;
use serde_json::json;
use tokio::time::Duration;
use wiremock::ResponseTemplate;
@@ -55,6 +57,53 @@ fn estimate_compact_payload_tokens(request: &responses::ResponsesRequest) -> i64
.saturating_add(approx_token_count(&request.instructions_text()))
}
fn assert_tools_payload_does_not_defer(body: &Value) {
if let Some(tools) = body.get("tools") {
assert!(
!contains_defer_loading(tools),
"model-visible tools should not include deferred declarations: {tools}"
);
}
}
fn namespace_child_tool_names(body: &Value, namespace: &str) -> Vec<String> {
body.get("tools")
.and_then(Value::as_array)
.and_then(|tools| {
tools.iter().find_map(|tool| {
if tool.get("type").and_then(Value::as_str) == Some("namespace")
&& tool.get("name").and_then(Value::as_str) == Some(namespace)
{
tool.get("tools").and_then(Value::as_array).map(|children| {
children
.iter()
.filter_map(|child| {
child
.get("name")
.and_then(Value::as_str)
.map(str::to_string)
})
.collect()
})
} else {
None
}
})
})
.unwrap_or_default()
}
fn contains_defer_loading(value: &Value) -> bool {
match value {
Value::Object(map) => {
map.get("defer_loading").and_then(Value::as_bool) == Some(true)
|| map.values().any(contains_defer_loading)
}
Value::Array(values) => values.iter().any(contains_defer_loading),
Value::Null | Value::Bool(_) | Value::Number(_) | Value::String(_) => false,
}
}
const PRETURN_CONTEXT_DIFF_CWD: &str = "/tmp/PRETURN_CONTEXT_DIFF_CWD";
const DUMMY_FUNCTION_NAME: &str = "test_tool";
const REMOTE_COMPACT_TURN_COMPLETE_TIMEOUT: Duration = Duration::from_secs(30);
@@ -358,6 +407,100 @@ async fn remote_compact_replaces_history_for_followups() -> Result<()> {
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn remote_compact_filters_deferred_dynamic_tools() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = responses::start_mock_server().await;
let mut builder = test_codex().with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing());
let mut test = builder.build(&server).await?;
let hidden_tool = "hidden_dynamic_tool";
let visible_tool = "visible_dynamic_tool";
let input_schema = json!({
"type": "object",
"properties": {},
"additionalProperties": false,
});
let dynamic_tools = vec![
DynamicToolSpec {
namespace: Some("codex_app".to_string()),
name: hidden_tool.to_string(),
description: "Hidden until discovered.".to_string(),
input_schema: input_schema.clone(),
defer_loading: true,
},
DynamicToolSpec {
namespace: Some("codex_app".to_string()),
name: visible_tool.to_string(),
description: "Visible immediately.".to_string(),
input_schema,
defer_loading: false,
},
];
let new_thread = test
.thread_manager
.start_thread_with_tools(
test.config.clone(),
dynamic_tools,
/*persist_extended_history*/ false,
)
.await?;
test.codex = new_thread.thread;
test.session_configured = new_thread.session_configured;
let codex = test.codex.clone();
let responses_mock = mount_sse_once(
&server,
sse(vec![
responses::ev_assistant_message("m1", "FIRST_REMOTE_REPLY"),
responses::ev_completed("resp-1"),
]),
)
.await;
let compact_mock = responses::mount_compact_json_once(
&server,
serde_json::json!({
"output": compacted_summary_only_output("compact summary"),
}),
)
.await;
codex
.submit(Op::UserInput {
environments: None,
items: vec![UserInput::Text {
text: "hello remote compact".into(),
text_elements: Vec::new(),
}],
final_output_json_schema: None,
responsesapi_client_metadata: None,
})
.await?;
wait_for_turn_complete(&codex).await;
codex.submit(Op::Compact).await?;
wait_for_turn_complete(&codex).await;
let first_response_body = responses_mock.single_request().body_json();
let compact_body = compact_mock.single_request().body_json();
assert_eq!(
compact_body["tools"], first_response_body["tools"],
"compact requests should send the same model-visible tools payload as /v1/responses"
);
assert_tools_payload_does_not_defer(&first_response_body);
assert_tools_payload_does_not_defer(&compact_body);
assert_eq!(
namespace_child_tool_names(&first_response_body, "codex_app"),
vec![visible_tool.to_string()]
);
assert_eq!(
namespace_child_tool_names(&compact_body, "codex_app"),
vec![visible_tool.to_string()]
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn remote_compact_runs_automatically() -> Result<()> {
skip_if_no_network!(Ok(()));
+8 -4
View File
@@ -605,7 +605,8 @@ async fn tool_search_returns_deferred_tools_without_follow_up_tool_injection() -
"apps tools/call should include turn metadata turn_id: {apps_tool_call:?}"
);
let first_request_tools = tool_names(&requests[0].body_json());
let first_request_body = requests[0].body_json();
let first_request_tools = tool_names(&first_request_body);
assert!(
first_request_tools
.iter()
@@ -823,7 +824,8 @@ async fn tool_search_returns_deferred_dynamic_tool_and_routes_follow_up_call() -
let requests = mock.requests();
assert_eq!(requests.len(), 3);
let first_request_tools = tool_names(&requests[0].body_json());
let first_request_body = requests[0].body_json();
let first_request_tools = tool_names(&first_request_body);
assert!(
first_request_tools
.iter()
@@ -853,7 +855,8 @@ async fn tool_search_returns_deferred_dynamic_tool_and_routes_follow_up_call() -
})]
);
let second_request_tools = tool_names(&requests[1].body_json());
let second_request_body = requests[1].body_json();
let second_request_tools = tool_names(&second_request_body);
assert!(
!second_request_tools.iter().any(|name| name == tool_name),
"follow-up request should rely on tool_search_output history, not tool injection: {second_request_tools:?}"
@@ -870,7 +873,8 @@ async fn tool_search_returns_deferred_dynamic_tool_and_routes_follow_up_call() -
FunctionCallOutputPayload::from_text("dynamic-search-ok".to_string())
);
let third_request_tools = tool_names(&requests[2].body_json());
let third_request_body = requests[2].body_json();
let third_request_tools = tool_names(&third_request_body);
assert!(
!third_request_tools.iter().any(|name| name == tool_name),
"post-tool follow-up should rely on tool_search_output history, not tool injection: {third_request_tools:?}"
+5 -2
View File
@@ -348,7 +348,8 @@ pub async fn run_main(cli: Cli, arg0_paths: Arg0DispatchPaths) -> anyhow::Result
/*enable_codex_api_key_env*/ false,
config_toml.cli_auth_credentials_store.unwrap_or_default(),
chatgpt_base_url,
);
)
.await;
let run_cli_overrides = cli_kv_overrides.clone();
let run_loader_overrides = loader_overrides.clone();
let run_cloud_requirements = cloud_requirements.clone();
@@ -438,7 +439,9 @@ pub async fn run_main(cli: Cli, arg0_paths: Arg0DispatchPaths) -> anyhow::Result
auth_credentials_store_mode: config.cli_auth_credentials_store_mode,
forced_login_method: config.forced_login_method,
forced_chatgpt_workspace_id: config.forced_chatgpt_workspace_id.clone(),
}) {
})
.await
{
eprintln!("{err}");
std::process::exit(1);
}
-8
View File
@@ -128,8 +128,6 @@ pub enum Feature {
CodexGitCommit,
/// Enable runtime metrics snapshots via a manual reader.
RuntimeMetrics,
/// Enable thread lifecycle analytics emitted via the app-server analytics pipeline.
GeneralAnalytics,
/// Persist rollout metadata to a local SQLite database.
Sqlite,
/// Enable startup memory extraction and file-backed memory consolidation.
@@ -712,12 +710,6 @@ pub const FEATURES: &[FeatureSpec] = &[
stage: Stage::UnderDevelopment,
default_enabled: false,
},
FeatureSpec {
id: Feature::GeneralAnalytics,
key: "general_analytics",
stage: Stage::Stable,
default_enabled: true,
},
FeatureSpec {
id: Feature::Sqlite,
key: "sqlite",
-6
View File
@@ -156,12 +156,6 @@ fn browser_controls_are_stable_and_enabled_by_default() {
assert_eq!(feature_for_key("computer_use"), Some(Feature::ComputerUse));
}
#[test]
fn general_analytics_is_stable_and_enabled_by_default() {
assert_eq!(Feature::GeneralAnalytics.stage(), Stage::Stable);
assert_eq!(Feature::GeneralAnalytics.default_enabled(), true);
}
#[test]
fn use_linux_sandbox_bwrap_is_a_removed_feature_key() {
assert_eq!(
+45 -25
View File
@@ -116,10 +116,11 @@ fn login_with_agent_identity_rejects_invalid_jwt() {
);
}
#[test]
fn missing_auth_json_returns_none() {
#[tokio::test]
async fn missing_auth_json_returns_none() {
let dir = tempdir().unwrap();
let auth = CodexAuth::from_auth_storage(dir.path(), AuthCredentialsStoreMode::File)
.await
.expect("call should succeed");
assert_eq!(auth, None);
}
@@ -143,6 +144,7 @@ async fn pro_account_with_no_api_key_uses_chatgpt_auth() {
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
)
.await
.unwrap()
.unwrap();
assert_eq!(None, auth.api_key());
@@ -196,6 +198,7 @@ async fn loads_api_key_from_auth_json() {
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
)
.await
.unwrap()
.unwrap();
assert_eq!(auth.auth_mode(), AuthMode::ApiKey);
@@ -222,15 +225,16 @@ fn logout_removes_auth_file() -> Result<(), std::io::Error> {
Ok(())
}
#[test]
fn unauthorized_recovery_reports_mode_and_step_names() {
#[tokio::test]
async fn unauthorized_recovery_reports_mode_and_step_names() {
let dir = tempdir().unwrap();
let manager = AuthManager::shared(
dir.path().to_path_buf(),
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
/*chatgpt_base_url*/ None,
);
)
.await;
let managed = UnauthorizedRecovery {
manager: Arc::clone(&manager),
step: UnauthorizedRecoveryStep::Reload,
@@ -250,8 +254,8 @@ fn unauthorized_recovery_reports_mode_and_step_names() {
assert_eq!(external.step_name(), "external_refresh");
}
#[test]
fn refresh_failure_is_scoped_to_the_matching_auth_snapshot() {
#[tokio::test]
async fn refresh_failure_is_scoped_to_the_matching_auth_snapshot() {
let codex_home = tempdir().unwrap();
write_auth_file(
AuthFileParams {
@@ -268,6 +272,7 @@ fn refresh_failure_is_scoped_to_the_matching_auth_snapshot() {
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
)
.await
.expect("load auth")
.expect("auth available");
let mut updated_auth_dot_json = auth
@@ -284,6 +289,7 @@ fn refresh_failure_is_scoped_to_the_matching_auth_snapshot() {
updated_auth_dot_json,
AuthCredentialsStoreMode::File,
)
.await
.expect("updated auth should parse");
let manager = AuthManager::from_auth_for_testing(auth.clone());
@@ -619,9 +625,9 @@ impl Drop for EnvVarGuard {
}
}
#[test]
#[tokio::test]
#[serial(codex_auth_env)]
fn load_auth_reads_agent_identity_from_env() {
async fn load_auth_reads_agent_identity_from_env() {
let codex_home = tempdir().unwrap();
let expected_record = agent_identity_record("account-123");
let agent_identity = fake_agent_identity_jwt(&expected_record).expect("fake agent identity");
@@ -632,6 +638,7 @@ fn load_auth_reads_agent_identity_from_env() {
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
)
.await
.expect("env auth should load")
.expect("env auth should be present");
@@ -645,9 +652,9 @@ fn load_auth_reads_agent_identity_from_env() {
);
}
#[test]
#[tokio::test]
#[serial(codex_auth_env)]
fn load_auth_keeps_codex_api_key_env_precedence() {
async fn load_auth_keeps_codex_api_key_env_precedence() {
let codex_home = tempdir().unwrap();
let record = agent_identity_record("account-123");
let agent_identity = fake_agent_identity_jwt(&record).expect("fake agent identity");
@@ -659,6 +666,7 @@ fn load_auth_keeps_codex_api_key_env_precedence() {
/*enable_codex_api_key_env*/ true,
AuthCredentialsStoreMode::File,
)
.await
.expect("env auth should load")
.expect("env auth should be present");
@@ -679,8 +687,9 @@ async fn enforce_login_restrictions_logs_out_for_method_mismatch() {
)
.await;
let err =
super::enforce_login_restrictions(&config).expect_err("expected method mismatch to error");
let err = super::enforce_login_restrictions(&config)
.await
.expect_err("expected method mismatch to error");
assert!(err.to_string().contains("ChatGPT login is required"));
assert!(
!codex_home.path().join("auth.json").exists(),
@@ -710,6 +719,7 @@ async fn enforce_login_restrictions_logs_out_for_workspace_mismatch() {
.await;
let err = super::enforce_login_restrictions(&config)
.await
.expect_err("expected workspace mismatch to error");
assert!(err.to_string().contains("workspace org_mine"));
assert!(
@@ -739,7 +749,9 @@ async fn enforce_login_restrictions_allows_matching_workspace() {
)
.await;
super::enforce_login_restrictions(&config).expect("matching workspace should succeed");
super::enforce_login_restrictions(&config)
.await
.expect("matching workspace should succeed");
assert!(
codex_home.path().join("auth.json").exists(),
"auth.json should remain when restrictions pass"
@@ -761,7 +773,9 @@ async fn enforce_login_restrictions_allows_api_key_if_login_method_not_set_but_f
)
.await;
super::enforce_login_restrictions(&config).expect("matching workspace should succeed");
super::enforce_login_restrictions(&config)
.await
.expect("matching workspace should succeed");
assert!(
codex_home.path().join("auth.json").exists(),
"auth.json should remain when restrictions pass"
@@ -782,6 +796,7 @@ async fn enforce_login_restrictions_blocks_env_api_key_when_chatgpt_required() {
.await;
let err = super::enforce_login_restrictions(&config)
.await
.expect_err("environment API key should not satisfy forced ChatGPT login");
assert!(
err.to_string()
@@ -818,8 +833,8 @@ fn fake_agent_identity_jwt(record: &AgentIdentityAuthRecord) -> std::io::Result<
Ok(format!("{header_b64}.{payload_b64}.{signature_b64}"))
}
#[test]
fn plan_type_maps_known_plan() {
#[tokio::test]
async fn plan_type_maps_known_plan() {
let codex_home = tempdir().unwrap();
let _jwt = write_auth_file(
AuthFileParams {
@@ -836,14 +851,15 @@ fn plan_type_maps_known_plan() {
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
)
.await
.expect("load auth")
.expect("auth available");
pretty_assertions::assert_eq!(auth.account_plan_type(), Some(AccountPlanType::Pro));
}
#[test]
fn plan_type_maps_self_serve_business_usage_based_plan() {
#[tokio::test]
async fn plan_type_maps_self_serve_business_usage_based_plan() {
let codex_home = tempdir().unwrap();
let _jwt = write_auth_file(
AuthFileParams {
@@ -860,6 +876,7 @@ fn plan_type_maps_self_serve_business_usage_based_plan() {
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
)
.await
.expect("load auth")
.expect("auth available");
@@ -869,8 +886,8 @@ fn plan_type_maps_self_serve_business_usage_based_plan() {
);
}
#[test]
fn plan_type_maps_enterprise_cbp_usage_based_plan() {
#[tokio::test]
async fn plan_type_maps_enterprise_cbp_usage_based_plan() {
let codex_home = tempdir().unwrap();
let _jwt = write_auth_file(
AuthFileParams {
@@ -887,6 +904,7 @@ fn plan_type_maps_enterprise_cbp_usage_based_plan() {
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
)
.await
.expect("load auth")
.expect("auth available");
@@ -896,8 +914,8 @@ fn plan_type_maps_enterprise_cbp_usage_based_plan() {
);
}
#[test]
fn plan_type_maps_unknown_to_unknown() {
#[tokio::test]
async fn plan_type_maps_unknown_to_unknown() {
let codex_home = tempdir().unwrap();
let _jwt = write_auth_file(
AuthFileParams {
@@ -914,14 +932,15 @@ fn plan_type_maps_unknown_to_unknown() {
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
)
.await
.expect("load auth")
.expect("auth available");
pretty_assertions::assert_eq!(auth.account_plan_type(), Some(AccountPlanType::Unknown));
}
#[test]
fn missing_plan_type_maps_to_unknown() {
#[tokio::test]
async fn missing_plan_type_maps_to_unknown() {
let codex_home = tempdir().unwrap();
let _jwt = write_auth_file(
AuthFileParams {
@@ -938,6 +957,7 @@ fn missing_plan_type_maps_to_unknown() {
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
)
.await
.expect("load auth")
.expect("auth available");
+51 -32
View File
@@ -193,7 +193,7 @@ impl From<RefreshTokenError> for std::io::Error {
}
impl CodexAuth {
fn from_auth_dot_json(
async fn from_auth_dot_json(
codex_home: &Path,
auth_dot_json: AuthDotJson,
auth_credentials_store_mode: AuthCredentialsStoreMode,
@@ -234,7 +234,7 @@ impl CodexAuth {
}
}
pub fn from_auth_storage(
pub async fn from_auth_storage(
codex_home: &Path,
auth_credentials_store_mode: AuthCredentialsStoreMode,
) -> std::io::Result<Option<Self>> {
@@ -243,6 +243,7 @@ impl CodexAuth {
/*enable_codex_api_key_env*/ false,
auth_credentials_store_mode,
)
.await
}
pub fn from_agent_identity_jwt(jwt: &str) -> std::io::Result<Self> {
@@ -522,6 +523,7 @@ pub async fn logout_with_revoke(
auth_credentials_store_mode,
/*chatgpt_base_url*/ None,
)
.await
.logout_with_revoke()
.await
}
@@ -609,12 +611,13 @@ pub struct AuthConfig {
pub forced_chatgpt_workspace_id: Option<String>,
}
pub fn enforce_login_restrictions(config: &AuthConfig) -> std::io::Result<()> {
pub async fn enforce_login_restrictions(config: &AuthConfig) -> std::io::Result<()> {
let Some(auth) = load_auth(
&config.codex_home,
/*enable_codex_api_key_env*/ true,
config.auth_credentials_store_mode,
)?
)
.await?
else {
return Ok(());
};
@@ -714,15 +717,11 @@ fn logout_all_stores(
Ok(removed_ephemeral || removed_managed)
}
fn load_auth(
async fn load_auth(
codex_home: &Path,
enable_codex_api_key_env: bool,
auth_credentials_store_mode: AuthCredentialsStoreMode,
) -> std::io::Result<Option<CodexAuth>> {
let build_auth = |auth_dot_json: AuthDotJson, storage_mode| {
CodexAuth::from_auth_dot_json(codex_home, auth_dot_json, storage_mode)
};
// API key via env var takes precedence over any other auth method.
if enable_codex_api_key_env && let Some(api_key) = read_codex_api_key_from_env() {
return Ok(Some(CodexAuth::from_api_key(api_key.as_str())));
@@ -735,7 +734,12 @@ fn load_auth(
AuthCredentialsStoreMode::Ephemeral,
);
if let Some(auth_dot_json) = ephemeral_storage.load()? {
let auth = build_auth(auth_dot_json, AuthCredentialsStoreMode::Ephemeral)?;
let auth = CodexAuth::from_auth_dot_json(
codex_home,
auth_dot_json,
AuthCredentialsStoreMode::Ephemeral,
)
.await?;
return Ok(Some(auth));
}
@@ -755,7 +759,9 @@ fn load_auth(
None => return Ok(None),
};
let auth = build_auth(auth_dot_json, auth_credentials_store_mode)?;
let auth =
CodexAuth::from_auth_dot_json(codex_home, auth_dot_json, auth_credentials_store_mode)
.await?;
Ok(Some(auth))
}
@@ -1169,6 +1175,7 @@ impl UnauthorizedRecovery {
match self
.manager
.reload_if_account_id_matches(self.expected_account_id.as_deref())
.await
{
ReloadOutcome::ReloadedChanged => {
self.step = UnauthorizedRecoveryStep::RefreshToken;
@@ -1279,7 +1286,7 @@ impl AuthManager {
/// preferred auth method. Errors loading auth are swallowed; `auth()` will
/// simply return `None` in that case so callers can treat it as an
/// unauthenticated state.
pub fn new(
pub async fn new(
codex_home: PathBuf,
enable_codex_api_key_env: bool,
auth_credentials_store_mode: AuthCredentialsStoreMode,
@@ -1290,6 +1297,7 @@ impl AuthManager {
enable_codex_api_key_env,
auth_credentials_store_mode,
)
.await
.ok()
.flatten();
Self {
@@ -1402,13 +1410,16 @@ impl AuthManager {
/// Force a reload of the auth information from auth.json. Returns
/// whether the auth value changed.
pub fn reload(&self) -> bool {
pub async fn reload(&self) -> bool {
tracing::info!("Reloading auth");
let new_auth = self.load_auth_from_storage();
let new_auth = self.load_auth_from_storage().await;
self.set_cached_auth(new_auth)
}
fn reload_if_account_id_matches(&self, expected_account_id: Option<&str>) -> ReloadOutcome {
async fn reload_if_account_id_matches(
&self,
expected_account_id: Option<&str>,
) -> ReloadOutcome {
let expected_account_id = match expected_account_id {
Some(account_id) => account_id,
None => {
@@ -1417,7 +1428,7 @@ impl AuthManager {
}
};
let new_auth = self.load_auth_from_storage();
let new_auth = self.load_auth_from_storage().await;
let new_account_id = new_auth.as_ref().and_then(CodexAuth::get_account_id);
if new_account_id.as_deref() != Some(expected_account_id) {
@@ -1488,12 +1499,13 @@ impl AuthManager {
}
}
fn load_auth_from_storage(&self) -> Option<CodexAuth> {
async fn load_auth_from_storage(&self) -> Option<CodexAuth> {
load_auth(
&self.codex_home,
self.enable_codex_api_key_env,
self.auth_credentials_store_mode,
)
.await
.ok()
.flatten()
}
@@ -1557,22 +1569,25 @@ impl AuthManager {
}
/// Convenience constructor returning an `Arc` wrapper.
pub fn shared(
pub async fn shared(
codex_home: PathBuf,
enable_codex_api_key_env: bool,
auth_credentials_store_mode: AuthCredentialsStoreMode,
chatgpt_base_url: Option<String>,
) -> Arc<Self> {
Arc::new(Self::new(
codex_home,
enable_codex_api_key_env,
auth_credentials_store_mode,
chatgpt_base_url,
))
Arc::new(
Self::new(
codex_home,
enable_codex_api_key_env,
auth_credentials_store_mode,
chatgpt_base_url,
)
.await,
)
}
/// Convenience constructor returning an `Arc` wrapper from resolved config.
pub fn shared_from_config(
pub async fn shared_from_config(
config: &impl AuthManagerConfig,
enable_codex_api_key_env: bool,
) -> Arc<Self> {
@@ -1581,7 +1596,8 @@ impl AuthManager {
enable_codex_api_key_env,
config.cli_auth_credentials_store_mode(),
Some(config.chatgpt_base_url()),
);
)
.await;
auth_manager.set_forced_chatgpt_workspace_id(config.forced_chatgpt_workspace_id());
auth_manager
}
@@ -1647,7 +1663,10 @@ impl AuthManager {
.as_ref()
.and_then(CodexAuth::get_account_id);
match self.reload_if_account_id_matches(expected_account_id.as_deref()) {
match self
.reload_if_account_id_matches(expected_account_id.as_deref())
.await
{
ReloadOutcome::ReloadedChanged => {
tracing::info!("Skipping token refresh because auth changed after guarded reload.");
Ok(())
@@ -1714,10 +1733,10 @@ impl AuthManager {
/// if a file was removed, Ok(false) if no auth file existed. On success,
/// reloads the in‑memory auth cache so callers immediately observe the
/// unauthenticated state.
pub fn logout(&self) -> std::io::Result<bool> {
pub async fn logout(&self) -> std::io::Result<bool> {
let removed = logout_all_stores(&self.codex_home, self.auth_credentials_store_mode)?;
// Always reload to clear any cached auth (even if file absent).
self.reload();
self.reload().await;
Ok(removed)
}
@@ -1730,7 +1749,7 @@ impl AuthManager {
}
let result = logout_all_stores(&self.codex_home, self.auth_credentials_store_mode)?;
// Always reload to clear any cached auth (even if file absent).
self.reload();
self.reload().await;
Ok(result)
}
@@ -1826,7 +1845,7 @@ impl AuthManager {
AuthCredentialsStoreMode::Ephemeral,
)
.map_err(RefreshTokenError::Transient)?;
self.reload();
self.reload().await;
Ok(())
}
@@ -1846,7 +1865,7 @@ impl AuthManager {
refresh_response.refresh_token,
)
.map_err(RefreshTokenError::from)?;
self.reload();
self.reload().await;
Ok(())
}
+35 -34
View File
@@ -46,7 +46,7 @@ async fn refresh_token_succeeds_updates_storage() -> Result<()> {
.mount(&server)
.await;
let ctx = RefreshTokenTestContext::new(&server)?;
let ctx = RefreshTokenTestContext::new(&server).await?;
let initial_last_refresh = Utc::now() - Duration::days(1);
let initial_tokens = build_tokens(INITIAL_ACCESS_TOKEN, INITIAL_REFRESH_TOKEN);
let initial_auth = AuthDotJson {
@@ -56,7 +56,7 @@ async fn refresh_token_succeeds_updates_storage() -> Result<()> {
last_refresh: Some(initial_last_refresh),
agent_identity: None,
};
ctx.write_auth(&initial_auth)?;
ctx.write_auth(&initial_auth).await?;
ctx.auth_manager
.refresh_token_from_authority()
@@ -110,7 +110,7 @@ async fn refresh_token_refreshes_when_auth_is_unchanged() -> Result<()> {
.mount(&server)
.await;
let ctx = RefreshTokenTestContext::new(&server)?;
let ctx = RefreshTokenTestContext::new(&server).await?;
let initial_last_refresh = Utc::now() - Duration::days(1);
let initial_tokens = build_tokens(INITIAL_ACCESS_TOKEN, INITIAL_REFRESH_TOKEN);
let initial_auth = AuthDotJson {
@@ -120,7 +120,7 @@ async fn refresh_token_refreshes_when_auth_is_unchanged() -> Result<()> {
last_refresh: Some(initial_last_refresh),
agent_identity: None,
};
ctx.write_auth(&initial_auth)?;
ctx.write_auth(&initial_auth).await?;
ctx.auth_manager
.refresh_token()
@@ -164,7 +164,7 @@ async fn refresh_token_skips_refresh_when_auth_changed() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = MockServer::start().await;
let ctx = RefreshTokenTestContext::new(&server)?;
let ctx = RefreshTokenTestContext::new(&server).await?;
let initial_last_refresh = Utc::now() - Duration::days(1);
let initial_tokens = build_tokens(INITIAL_ACCESS_TOKEN, INITIAL_REFRESH_TOKEN);
@@ -175,7 +175,7 @@ async fn refresh_token_skips_refresh_when_auth_changed() -> Result<()> {
last_refresh: Some(initial_last_refresh),
agent_identity: None,
};
ctx.write_auth(&initial_auth)?;
ctx.write_auth(&initial_auth).await?;
let disk_tokens = build_tokens("disk-access-token", "disk-refresh-token");
let disk_auth = AuthDotJson {
@@ -230,7 +230,7 @@ async fn refresh_token_errors_on_account_mismatch() -> Result<()> {
.mount(&server)
.await;
let ctx = RefreshTokenTestContext::new(&server)?;
let ctx = RefreshTokenTestContext::new(&server).await?;
let initial_last_refresh = Utc::now() - Duration::days(1);
let initial_tokens = build_tokens(INITIAL_ACCESS_TOKEN, INITIAL_REFRESH_TOKEN);
let initial_auth = AuthDotJson {
@@ -240,7 +240,7 @@ async fn refresh_token_errors_on_account_mismatch() -> Result<()> {
last_refresh: Some(initial_last_refresh),
agent_identity: None,
};
ctx.write_auth(&initial_auth)?;
ctx.write_auth(&initial_auth).await?;
let mut disk_tokens = build_tokens("disk-access-token", "disk-refresh-token");
disk_tokens.account_id = Some("other-account".to_string());
@@ -299,7 +299,7 @@ async fn returns_fresh_tokens_as_is() -> Result<()> {
.mount(&server)
.await;
let ctx = RefreshTokenTestContext::new(&server)?;
let ctx = RefreshTokenTestContext::new(&server).await?;
let stale_refresh = Utc::now() - Duration::days(9);
let fresh_access_token = access_token_with_expiration(Utc::now() + Duration::hours(1));
let initial_tokens = build_tokens(&fresh_access_token, INITIAL_REFRESH_TOKEN);
@@ -310,7 +310,7 @@ async fn returns_fresh_tokens_as_is() -> Result<()> {
last_refresh: Some(stale_refresh),
agent_identity: None,
};
ctx.write_auth(&initial_auth)?;
ctx.write_auth(&initial_auth).await?;
let cached_auth = ctx
.auth_manager
@@ -347,7 +347,7 @@ async fn refreshes_token_when_access_token_is_expired() -> Result<()> {
.mount(&server)
.await;
let ctx = RefreshTokenTestContext::new(&server)?;
let ctx = RefreshTokenTestContext::new(&server).await?;
let fresh_refresh = Utc::now() - Duration::days(1);
let expired_access_token = access_token_with_expiration(Utc::now() - Duration::hours(1));
let initial_tokens = build_tokens(&expired_access_token, INITIAL_REFRESH_TOKEN);
@@ -358,7 +358,7 @@ async fn refreshes_token_when_access_token_is_expired() -> Result<()> {
last_refresh: Some(fresh_refresh),
agent_identity: None,
};
ctx.write_auth(&initial_auth)?;
ctx.write_auth(&initial_auth).await?;
let cached_auth = ctx
.auth_manager
@@ -398,7 +398,7 @@ async fn auth_reloads_disk_auth_when_cached_auth_is_stale() -> Result<()> {
let server = MockServer::start().await;
let ctx = RefreshTokenTestContext::new(&server)?;
let ctx = RefreshTokenTestContext::new(&server).await?;
let stale_refresh = Utc::now() - Duration::days(9);
let initial_tokens = build_tokens(INITIAL_ACCESS_TOKEN, INITIAL_REFRESH_TOKEN);
let initial_auth = AuthDotJson {
@@ -408,7 +408,7 @@ async fn auth_reloads_disk_auth_when_cached_auth_is_stale() -> Result<()> {
last_refresh: Some(stale_refresh),
agent_identity: None,
};
ctx.write_auth(&initial_auth)?;
ctx.write_auth(&initial_auth).await?;
let fresh_refresh = Utc::now() - Duration::days(1);
let disk_tokens = build_tokens("disk-access-token", "disk-refresh-token");
@@ -461,7 +461,7 @@ async fn auth_reloads_disk_auth_without_calling_expired_refresh_token() -> Resul
.mount(&server)
.await;
let ctx = RefreshTokenTestContext::new(&server)?;
let ctx = RefreshTokenTestContext::new(&server).await?;
let stale_refresh = Utc::now() - Duration::days(9);
let initial_tokens = build_tokens(INITIAL_ACCESS_TOKEN, INITIAL_REFRESH_TOKEN);
let initial_auth = AuthDotJson {
@@ -471,7 +471,7 @@ async fn auth_reloads_disk_auth_without_calling_expired_refresh_token() -> Resul
last_refresh: Some(stale_refresh),
agent_identity: None,
};
ctx.write_auth(&initial_auth)?;
ctx.write_auth(&initial_auth).await?;
let fresh_refresh = Utc::now() - Duration::days(1);
let disk_tokens = build_tokens("disk-access-token", "disk-refresh-token");
@@ -522,7 +522,7 @@ async fn refresh_token_returns_permanent_error_for_expired_refresh_token() -> Re
.mount(&server)
.await;
let ctx = RefreshTokenTestContext::new(&server)?;
let ctx = RefreshTokenTestContext::new(&server).await?;
let initial_last_refresh = Utc::now() - Duration::days(1);
let initial_tokens = build_tokens(INITIAL_ACCESS_TOKEN, INITIAL_REFRESH_TOKEN);
let initial_auth = AuthDotJson {
@@ -532,7 +532,7 @@ async fn refresh_token_returns_permanent_error_for_expired_refresh_token() -> Re
last_refresh: Some(initial_last_refresh),
agent_identity: None,
};
ctx.write_auth(&initial_auth)?;
ctx.write_auth(&initial_auth).await?;
let err = ctx
.auth_manager
@@ -575,7 +575,7 @@ async fn refresh_token_does_not_retry_after_permanent_failure() -> Result<()> {
.mount(&server)
.await;
let ctx = RefreshTokenTestContext::new(&server)?;
let ctx = RefreshTokenTestContext::new(&server).await?;
let initial_last_refresh = Utc::now() - Duration::days(1);
let initial_tokens = build_tokens(INITIAL_ACCESS_TOKEN, INITIAL_REFRESH_TOKEN);
let initial_auth = AuthDotJson {
@@ -585,7 +585,7 @@ async fn refresh_token_does_not_retry_after_permanent_failure() -> Result<()> {
last_refresh: Some(initial_last_refresh),
agent_identity: None,
};
ctx.write_auth(&initial_auth)?;
ctx.write_auth(&initial_auth).await?;
let first_err = ctx
.auth_manager
@@ -642,7 +642,7 @@ async fn refresh_token_reloads_changed_auth_after_permanent_failure() -> Result<
.mount(&server)
.await;
let ctx = RefreshTokenTestContext::new(&server)?;
let ctx = RefreshTokenTestContext::new(&server).await?;
let initial_last_refresh = Utc::now() - Duration::days(1);
let initial_tokens = build_tokens(INITIAL_ACCESS_TOKEN, INITIAL_REFRESH_TOKEN);
let initial_auth = AuthDotJson {
@@ -652,7 +652,7 @@ async fn refresh_token_reloads_changed_auth_after_permanent_failure() -> Result<
last_refresh: Some(initial_last_refresh),
agent_identity: None,
};
ctx.write_auth(&initial_auth)?;
ctx.write_auth(&initial_auth).await?;
let first_err = ctx
.auth_manager
@@ -723,7 +723,7 @@ async fn refresh_token_returns_transient_error_on_server_failure() -> Result<()>
.mount(&server)
.await;
let ctx = RefreshTokenTestContext::new(&server)?;
let ctx = RefreshTokenTestContext::new(&server).await?;
let initial_last_refresh = Utc::now() - Duration::days(1);
let initial_tokens = build_tokens(INITIAL_ACCESS_TOKEN, INITIAL_REFRESH_TOKEN);
let initial_auth = AuthDotJson {
@@ -733,7 +733,7 @@ async fn refresh_token_returns_transient_error_on_server_failure() -> Result<()>
last_refresh: Some(initial_last_refresh),
agent_identity: None,
};
ctx.write_auth(&initial_auth)?;
ctx.write_auth(&initial_auth).await?;
let err = ctx
.auth_manager
@@ -776,7 +776,7 @@ async fn unauthorized_recovery_reloads_then_refreshes_tokens() -> Result<()> {
.mount(&server)
.await;
let ctx = RefreshTokenTestContext::new(&server)?;
let ctx = RefreshTokenTestContext::new(&server).await?;
let initial_last_refresh = Utc::now() - Duration::days(1);
let initial_tokens = build_tokens(INITIAL_ACCESS_TOKEN, INITIAL_REFRESH_TOKEN);
let initial_auth = AuthDotJson {
@@ -786,7 +786,7 @@ async fn unauthorized_recovery_reloads_then_refreshes_tokens() -> Result<()> {
last_refresh: Some(initial_last_refresh),
agent_identity: None,
};
ctx.write_auth(&initial_auth)?;
ctx.write_auth(&initial_auth).await?;
let disk_tokens = build_tokens("disk-access-token", "disk-refresh-token");
let disk_auth = AuthDotJson {
@@ -870,7 +870,7 @@ async fn unauthorized_recovery_errors_on_account_mismatch() -> Result<()> {
.mount(&server)
.await;
let ctx = RefreshTokenTestContext::new(&server)?;
let ctx = RefreshTokenTestContext::new(&server).await?;
let initial_last_refresh = Utc::now() - Duration::days(1);
let initial_tokens = build_tokens(INITIAL_ACCESS_TOKEN, INITIAL_REFRESH_TOKEN);
let initial_auth = AuthDotJson {
@@ -880,7 +880,7 @@ async fn unauthorized_recovery_errors_on_account_mismatch() -> Result<()> {
last_refresh: Some(initial_last_refresh),
agent_identity: None,
};
ctx.write_auth(&initial_auth)?;
ctx.write_auth(&initial_auth).await?;
let mut disk_tokens = build_tokens("disk-access-token", "disk-refresh-token");
disk_tokens.account_id = Some("other-account".to_string());
@@ -941,7 +941,7 @@ async fn unauthorized_recovery_requires_chatgpt_auth() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = MockServer::start().await;
let ctx = RefreshTokenTestContext::new(&server)?;
let ctx = RefreshTokenTestContext::new(&server).await?;
let auth = AuthDotJson {
auth_mode: Some(AuthMode::ApiKey),
openai_api_key: Some("sk-test".to_string()),
@@ -949,7 +949,7 @@ async fn unauthorized_recovery_requires_chatgpt_auth() -> Result<()> {
last_refresh: None,
agent_identity: None,
};
ctx.write_auth(&auth)?;
ctx.write_auth(&auth).await?;
let mut recovery = ctx.auth_manager.unauthorized_recovery();
assert!(!recovery.has_next());
@@ -974,7 +974,7 @@ struct RefreshTokenTestContext {
}
impl RefreshTokenTestContext {
fn new(server: &MockServer) -> Result<Self> {
async fn new(server: &MockServer) -> Result<Self> {
let codex_home = TempDir::new()?;
let endpoint = format!("{}/oauth/token", server.uri());
@@ -985,7 +985,8 @@ impl RefreshTokenTestContext {
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
/*chatgpt_base_url*/ None,
);
)
.await;
Ok(Self {
codex_home,
@@ -1000,13 +1001,13 @@ impl RefreshTokenTestContext {
.context("auth.json should exist")
}
fn write_auth(&self, auth_dot_json: &AuthDotJson) -> Result<()> {
async fn write_auth(&self, auth_dot_json: &AuthDotJson) -> Result<()> {
save_auth(
self.codex_home.path(),
auth_dot_json,
AuthCredentialsStoreMode::File,
)?;
self.auth_manager.reload();
self.auth_manager.reload().await;
Ok(())
}
}
+2 -1
View File
@@ -143,7 +143,8 @@ async fn auth_manager_logout_with_revoke_uses_cached_auth() -> Result<()> {
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
/*chatgpt_base_url*/ None,
);
)
.await;
save_auth(
codex_home.path(),
&chatgpt_auth_with_refresh_token("newer-disk-refresh-token"),
+2 -1
View File
@@ -141,7 +141,8 @@ pub async fn run_main(
arg0_paths,
Arc::new(config),
environment_manager,
);
)
.await;
async move {
while let Some(msg) = incoming_rx.recv().await {
match msg {
+3 -2
View File
@@ -49,7 +49,7 @@ pub(crate) struct MessageProcessor {
impl MessageProcessor {
/// Create a new `MessageProcessor`, retaining a handle to the outgoing
/// `Sender` so handlers can enqueue messages to be written to stdout.
pub(crate) fn new(
pub(crate) async fn new(
outgoing: OutgoingMessageSender,
arg0_paths: Arg0DispatchPaths,
config: Arc<Config>,
@@ -59,7 +59,8 @@ impl MessageProcessor {
let auth_manager = AuthManager::shared_from_config(
config.as_ref(),
/*enable_codex_api_key_env*/ false,
);
)
.await;
let thread_manager = Arc::new(ThreadManager::new(
config.as_ref(),
auth_manager,
+3 -2
View File
@@ -206,7 +206,7 @@ fn static_manager_for_tests(model_catalog: ModelsResponse) -> StaticModelsManage
)
}
fn chatgpt_auth_tokens_for_tests(codex_home: &Path) -> CodexAuth {
async fn chatgpt_auth_tokens_for_tests(codex_home: &Path) -> CodexAuth {
let auth_dot_json = codex_login::AuthDotJson {
auth_mode: Some(AuthMode::ChatgptAuthTokens),
openai_api_key: None,
@@ -232,6 +232,7 @@ c2ln",
.expect("auth.json should be written");
CodexAuth::from_auth_storage(codex_home, AuthCredentialsStoreMode::File)
.await
.expect("auth should load")
.expect("auth should be present")
}
@@ -685,7 +686,7 @@ async fn refresh_available_models_fetches_with_chatgpt_auth_tokens() {
"ChatGPT Auth Tokens",
/*priority*/ 1,
)]]);
let auth = chatgpt_auth_tokens_for_tests(codex_home.path());
let auth = chatgpt_auth_tokens_for_tests(codex_home.path()).await;
let manager = openai_manager_for_tests_with_auth(
codex_home.path().to_path_buf(),
endpoint.clone(),
+74
View File
@@ -30,6 +30,9 @@ use codex_app_server_protocol::AdditionalPermissionProfile;
use codex_app_server_protocol::AgentMessageDeltaNotification;
use codex_app_server_protocol::CommandExecutionRequestApprovalParams;
use codex_app_server_protocol::ConfigWarningNotification;
use codex_app_server_protocol::FileChangeRequestApprovalParams;
use codex_app_server_protocol::FileUpdateChange;
use codex_app_server_protocol::ItemStartedNotification;
use codex_app_server_protocol::JSONRPCErrorError;
use codex_app_server_protocol::McpServerStartupState;
use codex_app_server_protocol::McpServerStatusUpdatedNotification;
@@ -38,6 +41,7 @@ use codex_app_server_protocol::NetworkApprovalProtocol as AppServerNetworkApprov
use codex_app_server_protocol::NetworkPolicyAmendment as AppServerNetworkPolicyAmendment;
use codex_app_server_protocol::NetworkPolicyRuleAction as AppServerNetworkPolicyRuleAction;
use codex_app_server_protocol::NonSteerableTurnKind as AppServerNonSteerableTurnKind;
use codex_app_server_protocol::PatchChangeKind;
use codex_app_server_protocol::PermissionsRequestApprovalParams;
use codex_app_server_protocol::RequestId as AppServerRequestId;
use codex_app_server_protocol::ServerNotification;
@@ -70,6 +74,7 @@ use codex_protocol::models::PermissionProfile;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::Event;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::FileChange;
use codex_protocol::protocol::NetworkApprovalContext;
use codex_protocol::protocol::NetworkApprovalProtocol;
use codex_protocol::protocol::RolloutItem;
@@ -2522,6 +2527,75 @@ async fn inactive_thread_exec_approval_splits_shell_wrapped_command() {
);
}
#[tokio::test]
async fn inactive_thread_file_change_approval_recovers_buffered_changes() {
let (mut app, mut app_event_rx, _op_rx) = make_test_app_with_channels().await;
let thread_id = ThreadId::new();
app.enqueue_thread_notification(
thread_id,
ServerNotification::ItemStarted(ItemStartedNotification {
thread_id: thread_id.to_string(),
turn_id: "turn-approval".to_string(),
item: ThreadItem::FileChange {
id: "patch-approval".to_string(),
changes: vec![FileUpdateChange {
path: "README.md".to_string(),
kind: PatchChangeKind::Add,
diff: "hello\n".to_string(),
}],
status: codex_app_server_protocol::PatchApplyStatus::InProgress,
},
}),
)
.await
.expect("enqueue file change item");
let request = ServerRequest::FileChangeRequestApproval {
request_id: AppServerRequestId::Integer(9),
params: FileChangeRequestApprovalParams {
thread_id: thread_id.to_string(),
turn_id: "turn-approval".to_string(),
item_id: "patch-approval".to_string(),
reason: Some("command failed; retry without sandbox?".to_string()),
grant_root: None,
},
};
let request = app
.interactive_request_for_thread_request(thread_id, &request)
.await
.expect("expected file change approval request");
let ThreadInteractiveRequest::Approval(ApprovalRequest::ApplyPatch {
changes, reason, ..
}) = &request
else {
panic!("expected apply-patch approval request");
};
assert_eq!(
changes,
&HashMap::from([(
PathBuf::from("README.md"),
FileChange::Add {
content: "hello\n".to_string(),
},
)])
);
assert_eq!(
reason,
&Some("command failed; retry without sandbox?".to_string())
);
app.push_thread_interactive_request(request);
let cell = match app_event_rx.try_recv() {
Ok(AppEvent::InsertHistoryCell(cell)) => cell,
other => panic!("expected patch preview history cell, saw {other:?}"),
};
let rendered = lines_to_single_string(&cell.display_lines(/*width*/ 80));
assert!(rendered.contains("• Added README.md (+1 -0)"));
assert!(rendered.contains("1 +hello"));
}
#[tokio::test]
async fn inactive_thread_permissions_approval_preserves_file_system_permissions() {
let app = make_test_app().await;
+48
View File
@@ -157,6 +157,40 @@ impl ThreadEventStore {
.collect()
}
pub(super) fn file_change_changes(
&self,
turn_id: &str,
item_id: &str,
) -> Option<Vec<codex_app_server_protocol::FileUpdateChange>> {
self.buffer
.iter()
.rev()
.find_map(|event| match event {
ThreadBufferedEvent::Notification(ServerNotification::ItemStarted(
notification,
)) if turn_id_matches(turn_id, &notification.turn_id) => {
file_change_item_changes(&notification.item, item_id)
}
ThreadBufferedEvent::Notification(ServerNotification::ItemCompleted(
notification,
)) if turn_id_matches(turn_id, &notification.turn_id) => {
file_change_item_changes(&notification.item, item_id)
}
ThreadBufferedEvent::Request(_)
| ThreadBufferedEvent::Notification(_)
| ThreadBufferedEvent::HistoryEntryResponse(_)
| ThreadBufferedEvent::FeedbackSubmission(_) => None,
})
.or_else(|| {
self.turns
.iter()
.rev()
.filter(|turn| turn_id_matches(turn_id, &turn.id))
.flat_map(|turn| turn.items.iter().rev())
.find_map(|item| file_change_item_changes(item, item_id))
})
}
pub(super) fn apply_thread_rollback(&mut self, response: &ThreadRollbackResponse) {
self.turns = response.thread.turns.clone();
self.buffer.clear();
@@ -231,6 +265,20 @@ impl ThreadEventStore {
}
}
fn turn_id_matches(request_turn_id: &str, candidate_turn_id: &str) -> bool {
request_turn_id.is_empty() || request_turn_id == candidate_turn_id
}
fn file_change_item_changes(
item: &ThreadItem,
item_id: &str,
) -> Option<Vec<codex_app_server_protocol::FileUpdateChange>> {
match item {
ThreadItem::FileChange { id, changes, .. } if id == item_id => Some(changes.clone()),
_ => None,
}
}
#[derive(Debug)]
pub(super) struct ThreadEventChannel {
pub(super) sender: mpsc::Sender<ThreadBufferedEvent>,
+34 -1
View File
@@ -194,6 +194,17 @@ impl App {
store.session.as_ref().map(|session| session.cwd.clone())
}
async fn thread_file_change_changes(
&self,
thread_id: ThreadId,
turn_id: &str,
item_id: &str,
) -> Option<Vec<codex_app_server_protocol::FileUpdateChange>> {
let channel = self.thread_event_channels.get(&thread_id)?;
let store = channel.store.lock().await;
store.file_change_changes(turn_id, item_id)
}
pub(super) async fn interactive_request_for_thread_request(
&self,
thread_id: ThreadId,
@@ -264,7 +275,11 @@ impl App {
.thread_cwd(thread_id)
.await
.unwrap_or_else(|| self.config.cwd.clone()),
changes: HashMap::new(),
changes: self
.thread_file_change_changes(thread_id, &params.turn_id, &params.item_id)
.await
.map(crate::app_server_approval_conversions::file_update_changes_to_core)
.unwrap_or_default(),
}),
),
ServerRequest::McpServerElicitationRequest { request_id, params } => {
@@ -311,6 +326,7 @@ impl App {
pub(super) fn push_thread_interactive_request(&mut self, request: ThreadInteractiveRequest) {
match request {
ThreadInteractiveRequest::Approval(request) => {
self.render_inactive_patch_preview(&request);
self.chat_widget.push_approval_request(request);
}
ThreadInteractiveRequest::McpServerElicitation(request) => {
@@ -320,6 +336,23 @@ impl App {
}
}
fn render_inactive_patch_preview(&mut self, request: &ApprovalRequest) {
let ApprovalRequest::ApplyPatch {
thread_label,
cwd,
changes,
..
} = request
else {
return;
};
if thread_label.is_none() || changes.is_empty() {
return;
}
self.chat_widget
.add_to_history(history_cell::new_patch_event(changes.clone(), cwd));
}
pub(super) async fn pending_inactive_thread_requests(&self) -> Vec<(ThreadId, ServerRequest)> {
let channels: Vec<(ThreadId, Arc<Mutex<ThreadEventStore>>)> = self
.thread_event_channels
+76 -2
View File
@@ -10,7 +10,8 @@
//!
//! Backtrack operates as a small state machine:
//! - The first `Esc` in the main view "primes" the feature and captures a base thread id.
//! - A subsequent `Esc` opens the transcript overlay (`Ctrl+T`) and highlights a user message.
//! - A subsequent `Esc` opens the transcript overlay (`Ctrl+T`) and highlights a user message when
//! there is a rewind target.
//! - `Enter` requests a rollback from core and records a `pending_rollback` guard.
//! - On `EventMsg::ThreadRolledBack`, we either finish an in-flight backtrack request or queue a
//! rollback trim so it runs in event order with transcript inserts.
@@ -44,6 +45,8 @@ use crossterm::event::KeyCode;
use crossterm::event::KeyEvent;
use crossterm::event::KeyEventKind;
const NO_PREVIOUS_MESSAGE_TO_EDIT: &str = "No previous message to edit.";
/// Aggregates all backtrack-related state used by the App.
#[derive(Default)]
pub(crate) struct BacktrackState {
@@ -266,11 +269,21 @@ impl App {
self.backtrack.primed = true;
self.backtrack.nth_user_message = usize::MAX;
self.backtrack.base_id = self.chat_widget.thread_id();
self.chat_widget.show_esc_backtrack_hint();
if has_backtrack_target(&self.transcript_cells) {
self.chat_widget.show_esc_backtrack_hint();
}
}
/// Open overlay and begin backtrack preview flow (first step + highlight).
fn open_backtrack_preview(&mut self, tui: &mut tui::Tui) {
if !has_backtrack_target(&self.transcript_cells) {
self.reset_backtrack_state();
self.chat_widget
.add_info_message(NO_PREVIOUS_MESSAGE_TO_EDIT.to_string(), /*hint*/ None);
tui.frame_requester().schedule_frame();
return;
}
self.open_transcript_overlay(tui);
self.backtrack.overlay_preview_active = true;
// Composer is hidden by overlay; clear its hint.
@@ -280,6 +293,14 @@ impl App {
/// When overlay is already open, begin preview mode and select latest user message.
fn begin_overlay_backtrack_preview(&mut self, tui: &mut tui::Tui) {
if !has_backtrack_target(&self.transcript_cells) {
self.close_transcript_overlay(tui);
self.chat_widget
.add_info_message(NO_PREVIOUS_MESSAGE_TO_EDIT.to_string(), /*hint*/ None);
tui.frame_requester().schedule_frame();
return;
}
self.backtrack.primed = true;
self.backtrack.base_id = self.chat_widget.thread_id();
self.backtrack.overlay_preview_active = true;
@@ -613,6 +634,10 @@ pub(crate) fn user_count(cells: &[Arc<dyn crate::history_cell::HistoryCell>]) ->
user_positions_iter(cells).count()
}
fn has_backtrack_target(cells: &[Arc<dyn crate::history_cell::HistoryCell>]) -> bool {
user_count(cells) > 0
}
fn nth_user_position(
cells: &[Arc<dyn crate::history_cell::HistoryCell>],
nth: usize,
@@ -674,9 +699,22 @@ mod tests {
use super::*;
use crate::history_cell::AgentMessageCell;
use crate::history_cell::HistoryCell;
use pretty_assertions::assert_eq;
use ratatui::prelude::Line;
use std::sync::Arc;
fn render_lines(lines: &[Line<'static>]) -> Vec<String> {
lines
.iter()
.map(|line| {
line.spans
.iter()
.map(|span| span.content.as_ref())
.collect::<String>()
})
.collect()
}
#[test]
fn trim_transcript_for_first_user_drops_user_and_newer_cells() {
let mut cells: Vec<Arc<dyn HistoryCell>> = vec![
@@ -885,4 +923,40 @@ mod tests {
assert_eq!(agent_group_count(&cells), 2);
}
#[test]
fn backtrack_target_requires_user_message() {
let mut cells: Vec<Arc<dyn HistoryCell>> = vec![
Arc::new(AgentMessageCell::new(
vec![Line::from("assistant")],
/*is_first_line*/ true,
)) as Arc<dyn HistoryCell>,
Arc::new(crate::history_cell::new_info_event(
"Context compacted".to_string(),
/*hint*/ None,
)) as Arc<dyn HistoryCell>,
];
assert!(!has_backtrack_target(&cells));
cells.push(Arc::new(UserHistoryCell {
message: "hello".to_string(),
text_elements: Vec::new(),
local_image_paths: Vec::new(),
remote_image_urls: Vec::new(),
}) as Arc<dyn HistoryCell>);
assert!(has_backtrack_target(&cells));
}
#[test]
fn backtrack_unavailable_info_message_snapshot() {
let cell = crate::history_cell::new_info_event(
NO_PREVIOUS_MESSAGE_TO_EDIT.to_string(),
/*hint*/ None,
);
let rendered = render_lines(&cell.display_lines(/*width*/ 80)).join("\n");
insta::assert_snapshot!(rendered);
}
}
@@ -1,9 +1,14 @@
use codex_app_server_protocol::AdditionalNetworkPermissions;
use codex_app_server_protocol::FileUpdateChange;
use codex_app_server_protocol::GrantedPermissionProfile;
use codex_app_server_protocol::NetworkApprovalContext as AppServerNetworkApprovalContext;
use codex_app_server_protocol::PatchChangeKind;
use codex_protocol::protocol::FileChange;
use codex_protocol::protocol::NetworkApprovalContext;
use codex_protocol::protocol::NetworkApprovalProtocol;
use codex_protocol::request_permissions::RequestPermissionProfile as CoreRequestPermissionProfile;
use std::collections::HashMap;
use std::path::PathBuf;
pub(crate) fn network_approval_context_to_core(
value: AppServerNetworkApprovalContext,
@@ -38,21 +43,50 @@ pub(crate) fn granted_permission_profile_from_request(
}
}
pub(crate) fn file_update_changes_to_core(
changes: Vec<FileUpdateChange>,
) -> HashMap<PathBuf, FileChange> {
changes
.into_iter()
.map(|change| {
let path = PathBuf::from(change.path);
let file_change = match change.kind {
PatchChangeKind::Add => FileChange::Add {
content: change.diff,
},
PatchChangeKind::Delete => FileChange::Delete {
content: change.diff,
},
PatchChangeKind::Update { move_path } => FileChange::Update {
unified_diff: change.diff,
move_path,
},
};
(path, file_change)
})
.collect()
}
#[cfg(test)]
mod tests {
use super::file_update_changes_to_core;
use super::granted_permission_profile_from_request;
use super::network_approval_context_to_core;
use codex_app_server_protocol::FileUpdateChange;
use codex_app_server_protocol::PatchChangeKind;
use codex_protocol::models::FileSystemPermissions;
use codex_protocol::models::NetworkPermissions;
use codex_protocol::permissions::FileSystemAccessMode;
use codex_protocol::permissions::FileSystemPath;
use codex_protocol::permissions::FileSystemSandboxEntry;
use codex_protocol::permissions::FileSystemSpecialPath;
use codex_protocol::protocol::FileChange;
use codex_protocol::protocol::NetworkApprovalContext;
use codex_protocol::protocol::NetworkApprovalProtocol;
use codex_protocol::request_permissions::RequestPermissionProfile as CoreRequestPermissionProfile;
use codex_utils_absolute_path::AbsolutePathBuf;
use pretty_assertions::assert_eq;
use std::collections::HashMap;
use std::path::PathBuf;
fn absolute_path(path: &str) -> AbsolutePathBuf {
@@ -73,6 +107,23 @@ mod tests {
);
}
#[test]
fn converts_file_update_changes_to_core() {
assert_eq!(
file_update_changes_to_core(vec![FileUpdateChange {
path: "foo.txt".to_string(),
kind: PatchChangeKind::Add,
diff: "hello\n".to_string(),
}]),
HashMap::from([(
PathBuf::from("foo.txt"),
FileChange::Add {
content: "hello\n".to_string(),
},
)])
);
}
#[test]
fn converts_request_permissions_into_granted_permissions() {
assert_eq!(
@@ -9,7 +9,6 @@ use crate::bottom_pane::CancellationEvent;
use crate::bottom_pane::list_selection_view::ListSelectionView;
use crate::bottom_pane::list_selection_view::SelectionItem;
use crate::bottom_pane::list_selection_view::SelectionViewParams;
use crate::diff_render::DiffSummary;
use crate::exec_command::strip_bash_lc_and_escape;
use crate::history_cell;
use crate::key_hint;
@@ -633,8 +632,6 @@ fn build_header(request: &ApprovalRequest) -> Box<dyn Renderable> {
ApprovalRequest::ApplyPatch {
thread_label,
reason,
cwd,
changes,
..
} => {
let mut header: Vec<Box<dyn Renderable>> = Vec::new();
@@ -643,11 +640,13 @@ fn build_header(request: &ApprovalRequest) -> Box<dyn Renderable> {
"Thread: ".into(),
thread_label.clone().bold(),
])));
header.push(Box::new(Line::from("")));
}
if let Some(reason) = reason
&& !reason.is_empty()
{
if !header.is_empty() {
header.push(Box::new(Line::from("")));
}
header.push(Box::new(
Paragraph::new(Line::from_iter([
"Reason: ".into(),
@@ -655,9 +654,7 @@ fn build_header(request: &ApprovalRequest) -> Box<dyn Renderable> {
]))
.wrap(Wrap { trim: false }),
));
header.push(Box::new(Line::from("")));
}
header.push(DiffSummary::new(changes.clone(), cwd.clone()).into());
Box::new(ColumnRenderable::with(header))
}
ApprovalRequest::McpElicitation {
@@ -1556,6 +1553,32 @@ mod tests {
);
}
#[test]
fn apply_patch_prompt_with_thread_label_omits_command_line() {
let (tx, _rx) = unbounded_channel::<AppEvent>();
let tx = AppEventSender::new(tx);
let mut changes = HashMap::new();
changes.insert(
PathBuf::from("bug1.txt"),
FileChange::Add {
content: "one\ntwo\nthree\n".to_string(),
},
);
let request = ApprovalRequest::ApplyPatch {
thread_id: ThreadId::new(),
thread_label: Some("Banach [worker]".to_string()),
id: "test".to_string(),
reason: None,
cwd: absolute_path("/tmp"),
changes,
};
let view = ApprovalOverlay::new(request, tx, Features::with_defaults());
let rendered = render_overlay_lines(&view, /*width*/ 120);
assert!(rendered.contains("Thread: Banach [worker]"));
assert!(rendered.contains("o to open thread"));
assert!(!rendered.contains("$ apply_patch"));
}
#[test]
fn network_exec_prompt_title_includes_host() {
let (tx, _rx) = unbounded_channel::<AppEvent>();
@@ -1160,7 +1160,7 @@ impl ChatComposer {
/// Convert canonical composer text into the textarea's internal representation.
///
/// Bash mode stores the leading `!` as prompt state instead of editable text,
/// Shell mode stores the leading `!` as prompt state instead of editable text,
/// so full-buffer imports must absorb that prefix before rebuilding the textarea.
fn imported_text_for_textarea(
&mut self,
@@ -2934,7 +2934,7 @@ impl ChatComposer {
fn shell_mode_footer_line(&self) -> Option<Line<'static>> {
self.is_bang_shell_command()
.then_some(())
.map(|_| Line::from(vec![Span::from("Bash mode").light_red()]))
.map(|_| Line::from(vec![Span::from("Shell mode").light_red()]))
}
/// Applies any due `PasteBurst` flush at time `now`.
@@ -4660,7 +4660,7 @@ mod tests {
}
#[test]
fn shell_command_uses_bash_accent_style() {
fn shell_command_uses_shell_accent_style() {
let (tx, _rx) = unbounded_channel::<AppEvent>();
let sender = AppEventSender::new(tx);
let mut composer = ChatComposer::new(
@@ -4688,11 +4688,11 @@ mod tests {
let footer_text = (0..area.width)
.map(|x| buf[(x, footer_y)].symbol().chars().next().unwrap_or(' '))
.collect::<String>();
let bash_label_x = footer_text
.find("Bash mode")
.expect("expected bash mode footer label");
let shell_label_x = footer_text
.find("Shell mode")
.expect("expected shell mode footer label");
assert_eq!(
buf[(bash_label_x as u16, footer_y)].style().fg,
buf[(shell_label_x as u16, footer_y)].style().fg,
Some(Color::LightRed)
);
}
@@ -10,4 +10,4 @@ expression: terminal.backend()
" "
" "
" "
" gpt-5.4 high fast · ~/code/codex-1 · Context 0% used Bash mode "
" gpt-5.4 high fast · ~/code/codex-1 · Context 0% used Shell mode "
+22 -34
View File
@@ -47,6 +47,7 @@ use self::realtime::PendingSteerCompareKey;
use crate::app::app_server_requests::ResolvedAppServerRequest;
use crate::app_command::AppCommand;
use crate::app_event::RealtimeAudioDeviceKind;
use crate::app_server_approval_conversions::file_update_changes_to_core;
use crate::app_server_approval_conversions::network_approval_context_to_core;
use crate::app_server_session::ThreadSessionState;
#[cfg(not(target_os = "linux"))]
@@ -1774,36 +1775,6 @@ fn patch_approval_request_from_params(
}
}
fn app_server_patch_changes_to_core(
changes: Vec<codex_app_server_protocol::FileUpdateChange>,
) -> HashMap<PathBuf, codex_protocol::protocol::FileChange> {
changes
.into_iter()
.map(|change| {
let path = PathBuf::from(change.path);
let file_change = match change.kind {
codex_app_server_protocol::PatchChangeKind::Add => {
codex_protocol::protocol::FileChange::Add {
content: change.diff,
}
}
codex_app_server_protocol::PatchChangeKind::Delete => {
codex_protocol::protocol::FileChange::Delete {
content: change.diff,
}
}
codex_app_server_protocol::PatchChangeKind::Update { move_path } => {
codex_protocol::protocol::FileChange::Update {
unified_diff: change.diff,
move_path,
}
}
};
(path, file_change)
})
.collect()
}
fn app_server_collab_thread_id_to_core(thread_id: &str) -> Option<ThreadId> {
match ThreadId::from_string(thread_id) {
Ok(thread_id) => Some(thread_id),
@@ -6151,9 +6122,26 @@ impl ChatWidget {
}
}
fn submit_shell_command_with_history(
&mut self,
command: &str,
history_text: &str,
) -> QueueDrain {
let drain = self.submit_shell_command(command);
if drain == QueueDrain::Stop {
self.submit_op(Op::AddToHistory {
text: history_text.to_string(),
});
}
drain
}
fn submit_queued_shell_prompt(&mut self, user_message: UserMessage) -> QueueDrain {
match user_message.text.strip_prefix('!') {
Some(command) => self.submit_shell_command(command),
Some(command) => {
let history_text = user_message.text.clone();
self.submit_shell_command_with_history(command, &history_text)
}
None => {
self.submit_user_message(user_message);
QueueDrain::Stop
@@ -6249,7 +6237,7 @@ impl ChatWidget {
if shell_escape_policy == ShellEscapePolicy::Allow
&& let Some(stripped) = text.strip_prefix('!')
{
let app_command = match self.submit_shell_command(stripped) {
let app_command = match self.submit_shell_command_with_history(stripped, &text) {
QueueDrain::Continue => None,
QueueDrain::Stop => Some(AppCommand::run_user_shell_command(
stripped.trim().to_string(),
@@ -6780,7 +6768,7 @@ impl ChatWidget {
status,
codex_app_server_protocol::PatchApplyStatus::Failed
),
changes: app_server_patch_changes_to_core(changes),
changes: file_update_changes_to_core(changes),
status: match status {
codex_app_server_protocol::PatchApplyStatus::Completed => {
codex_protocol::protocol::PatchApplyStatus::Completed
@@ -7339,7 +7327,7 @@ impl ChatWidget {
call_id: id,
turn_id: notification.turn_id,
auto_approved: false,
changes: app_server_patch_changes_to_core(changes),
changes: file_update_changes_to_core(changes),
});
}
ThreadItem::McpToolCall {
@@ -1,18 +1,11 @@
---
source: tui/src/chatwidget/tests.rs
expression: terminal.backend().vt100().screen().contents()
source: tui/src/chatwidget/tests/exec_flow.rs
expression: contents
---
Would you like to make the following edits?
Reason: The model wants to apply changes
README.md (+2 -0)
1 +hello
2 +world
› 1. Yes, proceed (y)
2. Yes, and don't ask again for these files (a)
3. No, and tell Codex what to do differently (esc)
@@ -349,25 +349,6 @@ async fn live_app_server_command_execution_strips_shell_wrapper() {
);
}
#[test]
fn app_server_patch_changes_to_core_preserves_diffs() {
let changes = app_server_patch_changes_to_core(vec![FileUpdateChange {
path: "foo.txt".to_string(),
kind: PatchChangeKind::Add,
diff: "hello\n".to_string(),
}]);
assert_eq!(
changes,
HashMap::from([(
PathBuf::from("foo.txt"),
FileChange::Add {
content: "hello\n".to_string(),
},
)])
);
}
#[tokio::test]
async fn live_app_server_collab_wait_items_render_history() {
let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
+16 -49
View File
@@ -1038,6 +1038,10 @@ async fn bang_shell_enter_while_task_running_submits_run_user_shell_command() {
Ok(Op::RunUserShellCommand { command }) => assert_eq!(command, "echo hi"),
other => panic!("expected RunUserShellCommand op, got {other:?}"),
}
assert_matches!(
op_rx.try_recv(),
Ok(Op::AddToHistory { text }) if text == "!echo hi"
);
assert_matches!(rx.try_recv(), Err(TryRecvError::Empty));
}
@@ -1317,10 +1321,9 @@ async fn approval_modal_patch_snapshot() -> anyhow::Result<()> {
terminal
.draw(|f| chat.render(f.area(), f.buffer_mut()))
.expect("draw patch approval modal");
assert_chatwidget_snapshot!(
"approval_modal_patch",
terminal.backend().vt100().screen().contents()
);
let contents = terminal.backend().vt100().screen().contents();
assert!(!contents.contains("$ apply_patch"));
assert_chatwidget_snapshot!("approval_modal_patch", contents);
Ok(())
}
@@ -1465,28 +1468,11 @@ async fn apply_patch_events_emit_history_cells() {
id: "s1".into(),
msg: EventMsg::ApplyPatchApprovalRequest(ev),
});
let cells = drain_insert_history(&mut rx);
assert!(
cells.is_empty(),
drain_insert_history(&mut rx).is_empty(),
"expected approval request to surface via modal without emitting history cells"
);
let area = Rect::new(0, 0, 80, chat.desired_height(/*width*/ 80));
let mut buf = ratatui::buffer::Buffer::empty(area);
chat.render(area, &mut buf);
let mut saw_summary = false;
for y in 0..area.height {
let mut row = String::new();
for x in 0..area.width {
row.push(buf[(x, y)].symbol().chars().next().unwrap_or(' '));
}
if row.contains("foo.txt (+1 -0)") {
saw_summary = true;
break;
}
}
assert!(saw_summary, "expected approval modal to show diff summary");
// 2) Begin apply -> per-file apply block cell (no global header)
let mut changes2 = HashMap::new();
changes2.insert(
@@ -1816,7 +1802,7 @@ async fn apply_patch_untrusted_shows_approval_modal() -> anyhow::Result<()> {
}
#[tokio::test]
async fn apply_patch_request_shows_diff_summary() -> anyhow::Result<()> {
async fn apply_patch_request_omits_diff_summary_from_modal() -> anyhow::Result<()> {
let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
// Ensure we are in OnRequest so an approval is surfaced
@@ -1845,43 +1831,24 @@ async fn apply_patch_request_shows_diff_summary() -> anyhow::Result<()> {
}),
});
// No history entries yet; the modal should contain the diff summary
let cells = drain_insert_history(&mut rx);
assert!(
cells.is_empty(),
drain_insert_history(&mut rx).is_empty(),
"expected approval request to render via modal instead of history"
);
let area = Rect::new(0, 0, 80, chat.desired_height(/*width*/ 80));
let mut buf = ratatui::buffer::Buffer::empty(area);
chat.render(area, &mut buf);
let mut saw_header = false;
let mut saw_line1 = false;
let mut saw_line2 = false;
let mut contents = String::new();
for y in 0..area.height {
let mut row = String::new();
for x in 0..area.width {
row.push(buf[(x, y)].symbol().chars().next().unwrap_or(' '));
}
if row.contains("README.md (+2 -0)") {
saw_header = true;
}
if row.contains("+line one") {
saw_line1 = true;
}
if row.contains("+line two") {
saw_line2 = true;
}
if saw_header && saw_line1 && saw_line2 {
break;
contents.push(buf[(x, y)].symbol().chars().next().unwrap_or(' '));
}
contents.push('\n');
}
assert!(saw_header, "expected modal to show diff header with totals");
assert!(
saw_line1 && saw_line2,
"expected modal to show per-line diff summary"
);
assert!(!contents.contains("README.md (+2 -0)"));
assert!(!contents.contains("+line one"));
assert!(!contents.contains("+line two"));
Ok(())
}
@@ -210,6 +210,10 @@ async fn queued_bang_shell_dispatches_after_active_turn() {
Ok(Op::RunUserShellCommand { command }) => assert_eq!(command, "echo hi"),
other => panic!("expected queued shell command op, got {other:?}"),
}
assert_matches!(
op_rx.try_recv(),
Ok(Op::AddToHistory { text }) if text == "!echo hi"
);
assert!(chat.queued_user_messages.is_empty());
}
@@ -287,7 +291,10 @@ async fn queued_bang_shell_waits_for_user_shell_completion_before_next_input() {
Ok(Op::RunUserShellCommand { command }) => assert_eq!(command, "echo hi"),
other => panic!("expected queued shell command op, got {other:?}"),
}
assert_matches!(op_rx.try_recv(), Err(TryRecvError::Empty));
assert_matches!(
op_rx.try_recv(),
Ok(Op::AddToHistory { text }) if text == "!echo hi"
);
assert_eq!(chat.queued_user_messages.len(), 1);
let begin = begin_exec_with_source(
+7 -3
View File
@@ -795,7 +795,8 @@ pub async fn run_main(
/*enable_codex_api_key_env*/ false,
config_toml.cli_auth_credentials_store.unwrap_or_default(),
chatgpt_base_url,
);
)
.await;
let model_provider_override = if cli.oss {
let resolved = resolve_oss_provider(
@@ -895,7 +896,9 @@ pub async fn run_main(
auth_credentials_store_mode: config.cli_auth_credentials_store_mode,
forced_login_method: config.forced_login_method,
forced_chatgpt_workspace_id: config.forced_chatgpt_workspace_id.clone(),
}) {
})
.await
{
eprintln!("{err}");
std::process::exit(1);
}
@@ -1166,7 +1169,8 @@ async fn run_ratatui_app(
/*enable_codex_api_key_env*/ false,
initial_config.cli_auth_credentials_store_mode,
initial_config.chatgpt_base_url.clone(),
);
)
.await;
}
// If the user made an explicit trust decision, or we showed the login flow, reload config
+2 -1
View File
@@ -986,7 +986,8 @@ mod tests {
/*enable_codex_api_key_env*/ false,
AuthCredentialsStoreMode::File,
"https://chatgpt.com/backend-api/".to_string(),
),
)
.await,
feedback: codex_feedback::CodexFeedback::new(),
log_db: None,
environment_manager: Arc::new(
@@ -0,0 +1,5 @@
---
source: tui/src/app_backtrack.rs
expression: rendered
---
• No previous message to edit.