openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-06 20:20:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

rebase

Browse Source
This commit is contained in:
Roy Han
2026-03-31 15:07:23 -07:00
parent 81b4ff7937 103acdfb06
commit ba125df5bd
1807 changed files with 41233 additions and 163241 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/codex_message_processor.rs
+230 -65
View File
@@ -251,6 +251,8 @@ use codex_git_utils::git_diff_to_remote;
use codex_login::ServerOptions as LoginServerOptions;
use codex_login::ShutdownHandle;
use codex_login::auth::login_with_chatgpt_auth_tokens;
use codex_login::complete_device_code_login;
use codex_login::request_device_code;
use codex_login::run_login_server;
use codex_protocol::ThreadId;
use codex_protocol::config_types::CollaborationMode;
@@ -341,12 +343,39 @@ struct ThreadListFilters {
search_term: Option<String>,
}
// Duration before a ChatGPT login attempt is abandoned.
// Duration before a browser ChatGPT login attempt is abandoned.
const LOGIN_CHATGPT_TIMEOUT: Duration = Duration::from_secs(10 * 60);
const LOGIN_ISSUER_OVERRIDE_ENV_VAR: &str = "CODEX_APP_SERVER_LOGIN_ISSUER";
const APP_LIST_LOAD_TIMEOUT: Duration = Duration::from_secs(90);
struct ActiveLogin {
shutdown_handle: ShutdownHandle,
login_id: Uuid,
enum ActiveLogin {
Browser {
shutdown_handle: ShutdownHandle,
login_id: Uuid,
},
DeviceCode {
cancel: CancellationToken,
login_id: Uuid,
},
}
impl ActiveLogin {
fn login_id(&self) -> Uuid {
match self {
ActiveLogin::Browser { login_id, .. } | ActiveLogin::DeviceCode { login_id, .. } => {
*login_id
}
}
}
fn cancel(&self) {
match self {
ActiveLogin::Browser {
shutdown_handle, ..
} => shutdown_handle.shutdown(),
ActiveLogin::DeviceCode { cancel, .. } => cancel.cancel(),
}
}
}
#[derive(Clone, Copy, Debug)]
@@ -367,7 +396,7 @@ enum ThreadShutdownResult {
impl Drop for ActiveLogin {
fn drop(&mut self) {
self.shutdown_handle.shutdown();
self.cancel();
}
}
@@ -408,6 +437,7 @@ struct ListenerTaskContext {
thread_state_manager: ThreadStateManager,
outgoing: Arc<OutgoingMessageSender>,
analytics_events_client: AnalyticsEventsClient,
general_analytics_enabled: bool,
thread_watch_manager: ThreadWatchManager,
fallback_model_provider: String,
codex_home: PathBuf,
@@ -961,6 +991,9 @@ impl CodexMessageProcessor {
LoginAccountParams::Chatgpt => {
self.login_chatgpt_v2(request_id).await;
}
LoginAccountParams::ChatgptDeviceCode => {
self.login_chatgpt_device_code_v2(request_id).await;
}
LoginAccountParams::ChatgptAuthTokens {
access_token,
chatgpt_account_id,
@@ -990,7 +1023,7 @@ impl CodexMessageProcessor {
&mut self,
params: &LoginApiKeyParams,
) -> std::result::Result<(), JSONRPCErrorError> {
if self.auth_manager.is_external_auth_active() {
if self.auth_manager.is_external_chatgpt_auth_active() {
return Err(self.external_auth_active_error());
}
@@ -1069,7 +1102,7 @@ impl CodexMessageProcessor {
) -> std::result::Result<LoginServerOptions, JSONRPCErrorError> {
let config = self.config.as_ref();
if self.auth_manager.is_external_auth_active() {
if self.auth_manager.is_external_chatgpt_auth_active() {
return Err(self.external_auth_active_error());
}
@@ -1081,7 +1114,7 @@ impl CodexMessageProcessor {
});
}
Ok(LoginServerOptions {
let mut opts = LoginServerOptions {
open_browser: false,
..LoginServerOptions::new(
config.codex_home.clone(),
@@ -1089,7 +1122,32 @@ impl CodexMessageProcessor {
config.forced_chatgpt_workspace_id.clone(),
config.cli_auth_credentials_store_mode,
)
})
};
#[cfg(debug_assertions)]
if let Ok(issuer) = std::env::var(LOGIN_ISSUER_OVERRIDE_ENV_VAR)
&& !issuer.trim().is_empty()
{
opts.issuer = issuer;
}
Ok(opts)
}
fn login_chatgpt_device_code_start_error(err: IoError) -> JSONRPCErrorError {
let is_not_found = err.kind() == std::io::ErrorKind::NotFound;
JSONRPCErrorError {
code: if is_not_found {
INVALID_REQUEST_ERROR_CODE
} else {
INTERNAL_ERROR_CODE
},
message: if is_not_found {
err.to_string()
} else {
format!("failed to request device code: {err}")
},
data: None,
}
}
async fn login_chatgpt_v2(&mut self, request_id: ConnectionRequestId) {
@@ -1105,7 +1163,7 @@ impl CodexMessageProcessor {
if let Some(existing) = guard.take() {
drop(existing);
}
*guard = Some(ActiveLogin {
*guard = Some(ActiveLogin::Browser {
shutdown_handle: shutdown_handle.clone(),
login_id,
});
@@ -1175,7 +1233,7 @@ impl CodexMessageProcessor {
// Clear the active login if it matches this attempt. It may have been replaced or cancelled.
let mut guard = active_login.lock().await;
if guard.as_ref().map(|l| l.login_id) == Some(login_id) {
if guard.as_ref().map(ActiveLogin::login_id) == Some(login_id) {
*guard = None;
}
});
@@ -1201,12 +1259,114 @@ impl CodexMessageProcessor {
}
}
async fn login_chatgpt_device_code_v2(&mut self, request_id: ConnectionRequestId) {
match self.login_chatgpt_common().await {
Ok(opts) => match request_device_code(&opts).await {
Ok(device_code) => {
let login_id = Uuid::new_v4();
let cancel = CancellationToken::new();
{
let mut guard = self.active_login.lock().await;
if let Some(existing) = guard.take() {
drop(existing);
}
*guard = Some(ActiveLogin::DeviceCode {
cancel: cancel.clone(),
login_id,
});
}
let verification_url = device_code.verification_url.clone();
let user_code = device_code.user_code.clone();
let response =
codex_app_server_protocol::LoginAccountResponse::ChatgptDeviceCode {
login_id: login_id.to_string(),
verification_url,
user_code,
};
self.outgoing.send_response(request_id, response).await;
let outgoing_clone = self.outgoing.clone();
let active_login = self.active_login.clone();
let auth_manager = self.auth_manager.clone();
let cloud_requirements = self.cloud_requirements.clone();
let chatgpt_base_url = self.config.chatgpt_base_url.clone();
let codex_home = self.config.codex_home.clone();
let cli_overrides = self.current_cli_overrides();
tokio::spawn(async move {
let (success, error_msg) = tokio::select! {
_ = cancel.cancelled() => {
(false, Some("Login was not completed".to_string()))
}
r = complete_device_code_login(opts, device_code) => {
match r {
Ok(()) => (true, None),
Err(err) => (false, Some(err.to_string())),
}
}
};
let payload_v2 = AccountLoginCompletedNotification {
login_id: Some(login_id.to_string()),
success,
error: error_msg,
};
outgoing_clone
.send_server_notification(ServerNotification::AccountLoginCompleted(
payload_v2,
))
.await;
if success {
auth_manager.reload();
replace_cloud_requirements_loader(
cloud_requirements.as_ref(),
auth_manager.clone(),
chatgpt_base_url,
codex_home,
);
sync_default_client_residency_requirement(
&cli_overrides,
cloud_requirements.as_ref(),
)
.await;
let auth = auth_manager.auth_cached();
let payload_v2 = AccountUpdatedNotification {
auth_mode: auth.as_ref().map(CodexAuth::api_auth_mode),
plan_type: auth.as_ref().and_then(CodexAuth::account_plan_type),
};
outgoing_clone
.send_server_notification(ServerNotification::AccountUpdated(
payload_v2,
))
.await;
}
let mut guard = active_login.lock().await;
if guard.as_ref().map(ActiveLogin::login_id) == Some(login_id) {
*guard = None;
}
});
}
Err(err) => {
let error = Self::login_chatgpt_device_code_start_error(err);
self.outgoing.send_error(request_id, error).await;
}
},
Err(err) => {
self.outgoing.send_error(request_id, err).await;
}
}
}
async fn cancel_login_chatgpt_common(
&mut self,
login_id: Uuid,
) -> std::result::Result<(), CancelLoginError> {
let mut guard = self.active_login.lock().await;
if guard.as_ref().map(|l| l.login_id) == Some(login_id) {
if guard.as_ref().map(ActiveLogin::login_id) == Some(login_id) {
if let Some(active) = guard.take() {
drop(active);
}
@@ -1379,7 +1539,7 @@ impl CodexMessageProcessor {
}
async fn refresh_token_if_requested(&self, do_refresh: bool) -> RefreshTokenRequestOutcome {
if self.auth_manager.is_external_auth_active() {
if self.auth_manager.is_external_chatgpt_auth_active() {
return RefreshTokenRequestOutcome::NotAttemptedOrSucceeded;
}
if do_refresh && let Err(err) = self.auth_manager.refresh_token().await {
@@ -1935,6 +2095,7 @@ impl CodexMessageProcessor {
thread_state_manager: self.thread_state_manager.clone(),
outgoing: Arc::clone(&self.outgoing),
analytics_events_client: self.analytics_events_client.clone(),
general_analytics_enabled: self.config.features.enabled(Feature::GeneralAnalytics),
thread_watch_manager: self.thread_watch_manager.clone(),
fallback_model_provider: self.config.model_provider_id.clone(),
codex_home: self.config.codex_home.clone(),
@@ -2167,15 +2328,17 @@ impl CodexMessageProcessor {
sandbox: config_snapshot.sandbox_policy.into(),
reasoning_effort: config_snapshot.reasoning_effort,
};
listener_task_context
.analytics_events_client
.track_response(
request_id.connection_id.0,
ClientResponse::ThreadStart {
request_id: request_id.request_id.clone(),
response: response.clone(),
},
);
if listener_task_context.general_analytics_enabled {
listener_task_context
.analytics_events_client
.track_response(
request_id.connection_id.0,
ClientResponse::ThreadStart {
request_id: request_id.request_id.clone(),
response: response.clone(),
},
);
}
listener_task_context
.outgoing
@@ -3654,13 +3817,15 @@ impl CodexMessageProcessor {
sandbox: session_configured.sandbox_policy.into(),
reasoning_effort: session_configured.reasoning_effort,
};
self.analytics_events_client.track_response(
request_id.connection_id.0,
ClientResponse::ThreadResume {
request_id: request_id.request_id.clone(),
response: response.clone(),
},
);
if self.config.features.enabled(Feature::GeneralAnalytics) {
self.analytics_events_client.track_response(
request_id.connection_id.0,
ClientResponse::ThreadResume {
request_id: request_id.request_id.clone(),
response: response.clone(),
},
);
}
self.outgoing.send_response(request_id, response).await;
}
@@ -4268,13 +4433,15 @@ impl CodexMessageProcessor {
sandbox: session_configured.sandbox_policy.into(),
reasoning_effort: session_configured.reasoning_effort,
};
self.analytics_events_client.track_response(
request_id.connection_id.0,
ClientResponse::ThreadFork {
request_id: request_id.request_id.clone(),
response: response.clone(),
},
);
if self.config.features.enabled(Feature::GeneralAnalytics) {
self.analytics_events_client.track_response(
request_id.connection_id.0,
ClientResponse::ThreadFork {
request_id: request_id.request_id.clone(),
response: response.clone(),
},
);
}
self.outgoing.send_response(request_id, response).await;
@@ -6179,16 +6346,6 @@ impl CodexMessageProcessor {
return;
}
let analytics_turn_start_params = params.clone();
self.analytics_events_client.track_request(
request_id.connection_id.0,
request_id.request_id.clone(),
ClientRequest::TurnStart {
request_id: request_id.request_id.clone(),
params: analytics_turn_start_params,
},
);
let collaboration_modes_config = CollaborationModesConfig {
default_mode_request_user_input: thread.enabled(Feature::DefaultModeRequestUserInput),
};
@@ -6264,13 +6421,6 @@ impl CodexMessageProcessor {
};
let response = TurnStartResponse { turn };
self.analytics_events_client.track_response(
request_id.connection_id.0,
ClientResponse::TurnStart {
request_id: request_id.request_id.clone(),
response: response.clone(),
},
);
self.outgoing.send_response(request_id, response).await;
}
Err(err) => {
@@ -6874,6 +7024,7 @@ impl CodexMessageProcessor {
thread_state_manager: self.thread_state_manager.clone(),
outgoing: Arc::clone(&self.outgoing),
analytics_events_client: self.analytics_events_client.clone(),
general_analytics_enabled: self.config.features.enabled(Feature::GeneralAnalytics),
thread_watch_manager: self.thread_watch_manager.clone(),
fallback_model_provider: self.config.model_provider_id.clone(),
codex_home: self.config.codex_home.clone(),
@@ -6962,6 +7113,7 @@ impl CodexMessageProcessor {
thread_state_manager: self.thread_state_manager.clone(),
outgoing: Arc::clone(&self.outgoing),
analytics_events_client: self.analytics_events_client.clone(),
general_analytics_enabled: self.config.features.enabled(Feature::GeneralAnalytics),
thread_watch_manager: self.thread_watch_manager.clone(),
fallback_model_provider: self.config.model_provider_id.clone(),
codex_home: self.config.codex_home.clone(),
@@ -6993,7 +7145,8 @@ impl CodexMessageProcessor {
outgoing,
thread_manager,
thread_state_manager,
analytics_events_client,
analytics_events_client: _,
general_analytics_enabled: _,
thread_watch_manager,
fallback_model_provider,
codex_home,
@@ -7040,7 +7193,6 @@ impl CodexMessageProcessor {
conversation_id,
conversation.clone(),
thread_manager.clone(),
analytics_events_client.clone(),
thread_outgoing,
thread_state.clone(),
thread_watch_manager.clone(),
@@ -8640,7 +8792,7 @@ mod tests {
fn config_load_error_marks_non_auth_cloud_requirements_failures_without_relogin() {
let err = std::io::Error::other(CloudRequirementsLoadError::new(
CloudRequirementsLoadErrorCode::RequestFailed,
None,
/*status_code*/ None,
"failed to load your workspace-managed config",
));
@@ -8854,7 +9006,8 @@ mod tests {
fn merge_persisted_resume_metadata_skips_missing_values() -> Result<()> {
let mut request_overrides = None;
let mut typesafe_overrides = ConfigOverrides::default();
let persisted_metadata = test_thread_metadata(None, None)?;
let persisted_metadata =
test_thread_metadata(/*model*/ None, /*reasoning_effort*/ None)?;
merge_persisted_resume_metadata(
&mut request_overrides,
@@ -8906,7 +9059,7 @@ mod tests {
path.clone(),
&head,
&session_meta,
None,
/*git*/ None,
"test-provider",
timestamp.clone(),
)
@@ -9116,9 +9269,9 @@ mod tests {
source,
Some("atlas".to_string()),
Some("explorer".to_string()),
None,
None,
None,
/*git_sha*/ None,
/*git_branch*/ None,
/*git_origin_url*/ None,
);
let thread = summary_to_thread(summary);
@@ -9137,7 +9290,9 @@ mod tests {
manager.connection_initialized(connection).await;
manager
.try_ensure_connection_subscribed(thread_id, connection, false)
.try_ensure_connection_subscribed(
thread_id, connection, /*experimental_raw_events*/ false,
)
.await
.expect("connection should be live");
{
@@ -9181,11 +9336,19 @@ mod tests {
manager.connection_initialized(connection_a).await;
manager.connection_initialized(connection_b).await;
manager
.try_ensure_connection_subscribed(thread_id, connection_a, false)
.try_ensure_connection_subscribed(
thread_id,
connection_a,
/*experimental_raw_events*/ false,
)
.await
.expect("connection_a should be live");
manager
.try_ensure_connection_subscribed(thread_id, connection_b, false)
.try_ensure_connection_subscribed(
thread_id,
connection_b,
/*experimental_raw_events*/ false,
)
.await
.expect("connection_b should be live");
{
@@ -9218,7 +9381,9 @@ mod tests {
assert!(
manager
.try_ensure_connection_subscribed(thread_id, connection, false)
.try_ensure_connection_subscribed(
thread_id, connection, /*experimental_raw_events*/ false
)
.await
.is_none()
);
codex-rs/app-server/src/codex_message_processor/plugin_app_helpers.rs
+1 -1
View File
@@ -137,7 +137,7 @@ mod tests {
&all_connectors,
&[],
&[AppConnectorId("alpha".to_string())],
false,
/*codex_apps_ready*/ false,
),
Vec::new()
);
codex-rs/app-server/src/config_api.rs
+151 -6
View File
@@ -12,7 +12,9 @@ use codex_app_server_protocol::ConfigWriteResponse;
use codex_app_server_protocol::ExperimentalFeatureEnablementSetParams;
use codex_app_server_protocol::ExperimentalFeatureEnablementSetResponse;
use codex_app_server_protocol::JSONRPCErrorError;
use codex_app_server_protocol::NetworkDomainPermission;
use codex_app_server_protocol::NetworkRequirements;
use codex_app_server_protocol::NetworkUnixSocketPermission;
use codex_app_server_protocol::SandboxMode;
use codex_core::AnalyticsEventsClient;
use codex_core::ThreadManager;
@@ -410,6 +412,20 @@ fn map_residency_requirement_to_api(
fn map_network_requirements_to_api(
network: codex_core::config_loader::NetworkRequirementsToml,
) -> NetworkRequirements {
let allowed_domains = network
.domains
.as_ref()
.and_then(codex_core::config_loader::NetworkDomainPermissionsToml::allowed_domains);
let denied_domains = network
.domains
.as_ref()
.and_then(codex_core::config_loader::NetworkDomainPermissionsToml::denied_domains);
let allow_unix_sockets = network
.unix_sockets
.as_ref()
.map(codex_core::config_loader::NetworkUnixSocketPermissionsToml::allow_unix_sockets)
.filter(|entries| !entries.is_empty());
NetworkRequirements {
enabled: network.enabled,
http_port: network.http_port,
@@ -417,13 +433,58 @@ fn map_network_requirements_to_api(
allow_upstream_proxy: network.allow_upstream_proxy,
dangerously_allow_non_loopback_proxy: network.dangerously_allow_non_loopback_proxy,
dangerously_allow_all_unix_sockets: network.dangerously_allow_all_unix_sockets,
allowed_domains: network.allowed_domains,
denied_domains: network.denied_domains,
allow_unix_sockets: network.allow_unix_sockets,
domains: network.domains.map(|domains| {
domains
.entries
.into_iter()
.map(|(pattern, permission)| {
(pattern, map_network_domain_permission_to_api(permission))
})
.collect()
}),
managed_allowed_domains_only: network.managed_allowed_domains_only,
allowed_domains,
denied_domains,
unix_sockets: network.unix_sockets.map(|unix_sockets| {
unix_sockets
.entries
.into_iter()
.map(|(path, permission)| {
(path, map_network_unix_socket_permission_to_api(permission))
})
.collect()
}),
allow_unix_sockets,
allow_local_binding: network.allow_local_binding,
}
}
fn map_network_domain_permission_to_api(
permission: codex_core::config_loader::NetworkDomainPermissionToml,
) -> NetworkDomainPermission {
match permission {
codex_core::config_loader::NetworkDomainPermissionToml::Allow => {
NetworkDomainPermission::Allow
}
codex_core::config_loader::NetworkDomainPermissionToml::Deny => {
NetworkDomainPermission::Deny
}
}
}
fn map_network_unix_socket_permission_to_api(
permission: codex_core::config_loader::NetworkUnixSocketPermissionToml,
) -> NetworkUnixSocketPermission {
match permission {
codex_core::config_loader::NetworkUnixSocketPermissionToml::Allow => {
NetworkUnixSocketPermission::Allow
}
codex_core::config_loader::NetworkUnixSocketPermissionToml::None => {
NetworkUnixSocketPermission::None
}
}
}
fn map_error(err: ConfigServiceError) -> JSONRPCErrorError {
if let Some(code) = err.write_error_code() {
return config_write_error(code, err.to_string());
@@ -452,7 +513,11 @@ mod tests {
use codex_core::AnalyticsEventsClient;
use codex_core::AuthManager;
use codex_core::CodexAuth;
use codex_core::config_loader::NetworkDomainPermissionToml as CoreNetworkDomainPermissionToml;
use codex_core::config_loader::NetworkDomainPermissionsToml as CoreNetworkDomainPermissionsToml;
use codex_core::config_loader::NetworkRequirementsToml as CoreNetworkRequirementsToml;
use codex_core::config_loader::NetworkUnixSocketPermissionToml as CoreNetworkUnixSocketPermissionToml;
use codex_core::config_loader::NetworkUnixSocketPermissionsToml as CoreNetworkUnixSocketPermissionsToml;
use codex_features::Feature;
use codex_protocol::protocol::AskForApproval as CoreAskForApproval;
use pretty_assertions::assert_eq;
@@ -505,10 +570,25 @@ mod tests {
allow_upstream_proxy: Some(false),
dangerously_allow_non_loopback_proxy: Some(false),
dangerously_allow_all_unix_sockets: Some(true),
allowed_domains: Some(vec!["api.openai.com".to_string()]),
domains: Some(CoreNetworkDomainPermissionsToml {
entries: std::collections::BTreeMap::from([
(
"api.openai.com".to_string(),
CoreNetworkDomainPermissionToml::Allow,
),
(
"example.com".to_string(),
CoreNetworkDomainPermissionToml::Deny,
),
]),
}),
managed_allowed_domains_only: Some(false),
denied_domains: Some(vec!["example.com".to_string()]),
allow_unix_sockets: Some(vec!["/tmp/proxy.sock".to_string()]),
unix_sockets: Some(CoreNetworkUnixSocketPermissionsToml {
entries: std::collections::BTreeMap::from([(
"/tmp/proxy.sock".to_string(),
CoreNetworkUnixSocketPermissionToml::Allow,
)]),
}),
allow_local_binding: Some(true),
}),
};
@@ -550,14 +630,79 @@ mod tests {
allow_upstream_proxy: Some(false),
dangerously_allow_non_loopback_proxy: Some(false),
dangerously_allow_all_unix_sockets: Some(true),
domains: Some(std::collections::BTreeMap::from([
("api.openai.com".to_string(), NetworkDomainPermission::Allow,),
("example.com".to_string(), NetworkDomainPermission::Deny),
])),
managed_allowed_domains_only: Some(false),
allowed_domains: Some(vec!["api.openai.com".to_string()]),
denied_domains: Some(vec!["example.com".to_string()]),
unix_sockets: Some(std::collections::BTreeMap::from([(
"/tmp/proxy.sock".to_string(),
NetworkUnixSocketPermission::Allow,
)])),
allow_unix_sockets: Some(vec!["/tmp/proxy.sock".to_string()]),
allow_local_binding: Some(true),
}),
);
}
#[test]
fn map_requirements_toml_to_api_omits_unix_socket_none_entries_from_legacy_network_fields() {
let requirements = ConfigRequirementsToml {
allowed_approval_policies: None,
allowed_sandbox_modes: None,
allowed_web_search_modes: None,
guardian_developer_instructions: None,
feature_requirements: None,
mcp_servers: None,
apps: None,
rules: None,
enforce_residency: None,
network: Some(CoreNetworkRequirementsToml {
enabled: None,
http_port: None,
socks_port: None,
allow_upstream_proxy: None,
dangerously_allow_non_loopback_proxy: None,
dangerously_allow_all_unix_sockets: None,
domains: None,
managed_allowed_domains_only: None,
unix_sockets: Some(CoreNetworkUnixSocketPermissionsToml {
entries: std::collections::BTreeMap::from([(
"/tmp/ignored.sock".to_string(),
CoreNetworkUnixSocketPermissionToml::None,
)]),
}),
allow_local_binding: None,
}),
};
let mapped = map_requirements_toml_to_api(requirements);
assert_eq!(
mapped.network,
Some(NetworkRequirements {
enabled: None,
http_port: None,
socks_port: None,
allow_upstream_proxy: None,
dangerously_allow_non_loopback_proxy: None,
dangerously_allow_all_unix_sockets: None,
domains: None,
managed_allowed_domains_only: None,
allowed_domains: None,
denied_domains: None,
unix_sockets: Some(std::collections::BTreeMap::from([(
"/tmp/ignored.sock".to_string(),
NetworkUnixSocketPermission::None,
)])),
allow_unix_sockets: None,
allow_local_binding: None,
}),
);
}
#[test]
fn map_requirements_toml_to_api_normalizes_allowed_web_search_modes() {
let requirements = ConfigRequirementsToml {
codex-rs/app-server/src/filters.rs
+1 -1
View File
@@ -90,7 +90,7 @@ mod tests {
#[test]
fn compute_source_filters_defaults_to_interactive_sources() {
let (allowed_sources, filter) = compute_source_filters(None);
let (allowed_sources, filter) = compute_source_filters(/*source_kinds*/ None);
assert_eq!(allowed_sources, INTERACTIVE_SESSION_SOURCES.to_vec());
assert_eq!(filter, None);
codex-rs/app-server/src/in_process.rs
+4 -1
View File
@@ -74,6 +74,7 @@ use codex_app_server_protocol::Result;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ServerRequest;
use codex_arg0::Arg0DispatchPaths;
use codex_core::AppServerRpcTransport;
use codex_core::config::Config;
use codex_core::config_loader::CloudRequirementsLoader;
use codex_core::config_loader::LoaderOverrides;
@@ -393,6 +394,7 @@ fn start_uninitialized(args: InProcessStartArgs) -> InProcessClientHandle {
config_warnings: args.config_warnings,
session_source: args.session_source,
enable_codex_api_key_env: args.enable_codex_api_key_env,
rpc_transport: AppServerRpcTransport::InProcess,
});
let mut thread_created_rx = processor.thread_created_receiver();
let mut session = ConnectionSessionState::default();
@@ -786,7 +788,8 @@ mod tests {
#[tokio::test]
async fn in_process_start_clamps_zero_channel_capacity() {
let client = start_test_client_with_capacity(SessionSource::Cli, 0).await;
let client =
start_test_client_with_capacity(SessionSource::Cli, /*channel_capacity*/ 0).await;
let response = loop {
match client
.request(ClientRequest::ConfigRequirementsRead {
codex-rs/app-server/src/lib.rs
+13 -1
View File
@@ -36,6 +36,7 @@ use codex_app_server_protocol::ConfigWarningNotification;
use codex_app_server_protocol::JSONRPCMessage;
use codex_app_server_protocol::TextPosition as AppTextPosition;
use codex_app_server_protocol::TextRange as AppTextRange;
use codex_core::AppServerRpcTransport;
use codex_core::ExecPolicyError;
use codex_core::check_execpolicy_for_warnings;
use codex_core::config_loader::ConfigLoadError;
@@ -623,6 +624,7 @@ pub async fn run_main_with_transport(
config_warnings,
session_source,
enable_codex_api_key_env: false,
rpc_transport: analytics_rpc_transport(transport),
});
let mut thread_created_rx = processor.thread_created_receiver();
let mut running_turn_count_rx = processor.subscribe_running_assistant_turn_count();
@@ -846,6 +848,13 @@ pub async fn run_main_with_transport(
Ok(())
}
fn analytics_rpc_transport(transport: AppServerTransport) -> AppServerRpcTransport {
match transport {
AppServerTransport::Stdio => AppServerRpcTransport::Stdio,
AppServerTransport::WebSocket { .. } => AppServerRpcTransport::Websocket,
}
}
#[cfg(test)]
mod tests {
use super::LogFormat;
@@ -860,7 +869,10 @@ mod tests {
#[test]
fn log_format_from_env_value_defaults_for_non_json_values() {
assert_eq!(LogFormat::from_env_value(None), LogFormat::Default);
assert_eq!(
LogFormat::from_env_value(/*value*/ None),
LogFormat::Default
);
assert_eq!(LogFormat::from_env_value(Some("")), LogFormat::Default);
assert_eq!(LogFormat::from_env_value(Some("text")), LogFormat::Default);
assert_eq!(LogFormat::from_env_value(Some("jsonl")), LogFormat::Default);
codex-rs/app-server/src/message_processor.rs
+32 -15
View File
@@ -56,6 +56,7 @@ use codex_app_server_protocol::experimental_required_message;
use codex_arg0::Arg0DispatchPaths;
use codex_chatgpt::connectors;
use codex_core::AnalyticsEventsClient;
use codex_core::AppServerRpcTransport;
use codex_core::AuthManager;
use codex_core::ThreadManager;
use codex_core::config::Config;
@@ -70,9 +71,10 @@ use codex_core::models_manager::collaboration_mode_presets::CollaborationModesCo
use codex_exec_server::EnvironmentManager;
use codex_features::Feature;
use codex_feedback::CodexFeedback;
use codex_login::AuthMode as LoginAuthMode;
use codex_login::auth::ExternalAuth;
use codex_login::auth::ExternalAuthRefreshContext;
use codex_login::auth::ExternalAuthRefreshReason;
use codex_login::auth::ExternalAuthRefresher;
use codex_login::auth::ExternalAuthTokens;
use codex_protocol::ThreadId;
use codex_protocol::protocol::SessionSource;
@@ -87,6 +89,7 @@ use toml::Value as TomlValue;
use tracing::Instrument;
const EXTERNAL_AUTH_REFRESH_TIMEOUT: Duration = Duration::from_secs(10);
#[derive(Clone)]
struct ExternalAuthRefreshBridge {
outgoing: Arc<OutgoingMessageSender>,
@@ -101,7 +104,11 @@ impl ExternalAuthRefreshBridge {
}
#[async_trait]
impl ExternalAuthRefresher for ExternalAuthRefreshBridge {
impl ExternalAuth for ExternalAuthRefreshBridge {
fn auth_mode(&self) -> LoginAuthMode {
LoginAuthMode::Chatgpt
}
async fn refresh(
&self,
context: ExternalAuthRefreshContext,
@@ -143,11 +150,11 @@ impl ExternalAuthRefresher for ExternalAuthRefreshBridge {
let response: ChatgptAuthTokensRefreshResponse =
serde_json::from_value(result).map_err(std::io::Error::other)?;
Ok(ExternalAuthTokens {
access_token: response.access_token,
chatgpt_account_id: response.chatgpt_account_id,
chatgpt_plan_type: response.chatgpt_plan_type,
})
Ok(ExternalAuthTokens::chatgpt(
response.access_token,
response.chatgpt_account_id,
response.chatgpt_plan_type,
))
}
}
@@ -162,6 +169,7 @@ pub(crate) struct MessageProcessor {
fs_watch_manager: FsWatchManager,
config: Arc<Config>,
config_warnings: Arc<Vec<ConfigWarningNotification>>,
rpc_transport: AppServerRpcTransport,
}
#[derive(Clone, Debug, Default)]
@@ -186,6 +194,7 @@ pub(crate) struct MessageProcessorArgs {
pub(crate) config_warnings: Vec<ConfigWarningNotification>,
pub(crate) session_source: SessionSource,
pub(crate) enable_codex_api_key_env: bool,
pub(crate) rpc_transport: AppServerRpcTransport,
}
impl MessageProcessor {
@@ -205,11 +214,15 @@ impl MessageProcessor {
config_warnings,
session_source,
enable_codex_api_key_env,
rpc_transport,
} = args;
let auth_manager = AuthManager::shared(
let auth_manager = AuthManager::shared_with_external_auth(
config.codex_home.clone(),
enable_codex_api_key_env,
config.cli_auth_credentials_store_mode,
Arc::new(ExternalAuthRefreshBridge {
outgoing: outgoing.clone(),
}),
);
let thread_manager = Arc::new(ThreadManager::new(
config.as_ref(),
@@ -223,9 +236,6 @@ impl MessageProcessor {
environment_manager,
));
auth_manager.set_forced_chatgpt_workspace_id(config.forced_chatgpt_workspace_id.clone());
auth_manager.set_external_auth_refresher(Arc::new(ExternalAuthRefreshBridge {
outgoing: outgoing.clone(),
}));
let analytics_events_client = AnalyticsEventsClient::new(
Arc::clone(&auth_manager),
config.chatgpt_base_url.trim_end_matches('/').to_string(),
@@ -280,11 +290,12 @@ impl MessageProcessor {
fs_watch_manager,
config,
config_warnings: Arc::new(config_warnings),
rpc_transport,
}
}
pub(crate) fn clear_runtime_references(&self) {
self.auth_manager.clear_external_auth_refresher();
self.auth_manager.clear_external_auth();
}
pub(crate) async fn process_request(
@@ -571,7 +582,7 @@ impl MessageProcessor {
session.app_server_client_name = Some(name.clone());
session.client_version = Some(version.clone());
let originator = name.clone();
if let Err(error) = set_default_originator(originator) {
if let Err(error) = set_default_originator(originator.clone()) {
match error {
SetOriginatorError::InvalidHeaderValue => {
let error = JSONRPCErrorError {
@@ -594,8 +605,14 @@ impl MessageProcessor {
}
}
}
self.analytics_events_client
.track_initialize(connection_id.0, analytics_initialize_params);
if self.config.features.enabled(Feature::GeneralAnalytics) {
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());
let user_agent_suffix = format!("{name}; {version}");
if let Ok(mut suffix) = USER_AGENT_SUFFIX.lock() {
codex-rs/app-server/src/message_processor/tracing_tests.rs
+18 -8
View File
@@ -20,6 +20,7 @@ use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartResponse;
use codex_app_server_protocol::UserInput;
use codex_arg0::Arg0DispatchPaths;
use codex_core::AppServerRpcTransport;
use codex_core::config::Config;
use codex_core::config::ConfigBuilder;
use codex_core::config_loader::CloudRequirementsLoader;
@@ -147,7 +148,7 @@ impl TracingHarness {
}),
},
},
None,
/*trace*/ None,
)
.await;
assert!(harness.session.initialized);
@@ -213,7 +214,7 @@ async fn build_test_config(codex_home: &Path, server_uri: &str) -> Result<Config
codex_home,
server_uri,
&BTreeMap::new(),
8_192,
/*auto_compact_limit*/ 8_192,
Some(false),
"mock_provider",
"compact",
@@ -246,6 +247,7 @@ fn build_test_processor(
config_warnings: Vec::new(),
session_source: SessionSource::VSCode,
enable_codex_api_key_env: false,
rpc_transport: AppServerRpcTransport::Stdio,
});
(processor, outgoing_rx)
}
@@ -509,7 +511,9 @@ async fn thread_start_jsonrpc_span_exports_server_span_and_parents_children() ->
..
} = RemoteTrace::new("00000000000000000000000000000011", "0000000000000022");
let _: ThreadStartResponse = harness.start_thread(20_002, None).await;
let _: ThreadStartResponse = harness
.start_thread(/*request_id*/ 20_002, /*trace*/ None)
.await;
let untraced_spans = wait_for_exported_spans(harness.tracing, |spans| {
spans.iter().any(|span| {
span.span_kind == SpanKind::Server
@@ -538,10 +542,16 @@ async fn thread_start_jsonrpc_span_exports_server_span_and_parents_children() ->
.span_context
.trace_id(),
);
assert_has_internal_descendant_at_min_depth(&untraced_spans, untraced_server_span, 1);
assert_has_internal_descendant_at_min_depth(
&untraced_spans,
untraced_server_span,
/*min_depth*/ 1,
);
let baseline_len = untraced_spans.len();
let _: ThreadStartResponse = harness.start_thread(20_003, Some(remote_trace)).await;
let _: ThreadStartResponse = harness
.start_thread(/*request_id*/ 20_003, Some(remote_trace))
.await;
let spans = wait_for_new_exported_spans(harness.tracing, baseline_len, |spans| {
spans.iter().any(|span| {
span.span_kind == SpanKind::Server
@@ -561,8 +571,8 @@ async fn thread_start_jsonrpc_span_exports_server_span_and_parents_children() ->
assert!(server_request_span.parent_span_is_remote);
assert_eq!(server_request_span.span_context.trace_id(), remote_trace_id);
assert_ne!(server_request_span.span_context.span_id(), SpanId::INVALID);
assert_has_internal_descendant_at_min_depth(&spans, server_request_span, 1);
assert_has_internal_descendant_at_min_depth(&spans, server_request_span, 2);
assert_has_internal_descendant_at_min_depth(&spans, server_request_span, /*min_depth*/ 1);
assert_has_internal_descendant_at_min_depth(&spans, server_request_span, /*min_depth*/ 2);
harness.shutdown().await;
Ok(())
@@ -572,7 +582,7 @@ async fn thread_start_jsonrpc_span_exports_server_span_and_parents_children() ->
async fn turn_start_jsonrpc_span_parents_core_turn_spans() -> Result<()> {
let _guard = tracing_test_guard().lock().await;
let mut harness = TracingHarness::new().await?;
let thread_start_response = harness.start_thread(2, None).await;
let thread_start_response = harness.start_thread(/*request_id*/ 2, /*trace*/ None).await;
let thread_id = thread_start_response.thread.id.clone();
harness.reset_tracing();
codex-rs/app-server/src/outgoing_message.rs
+3 -3
View File
@@ -886,7 +886,7 @@ mod tests {
.register_request_context(RequestContext::new(
request_id.clone(),
tracing::info_span!("app_server.request", rpc.method = "thread/start"),
None,
/*parent_trace*/ None,
))
.await;
assert_eq!(outgoing.request_context_count().await, 1);
@@ -997,14 +997,14 @@ mod tests {
.register_request_context(RequestContext::new(
closed_connection_request,
tracing::info_span!("app_server.request", rpc.method = "turn/interrupt"),
None,
/*parent_trace*/ None,
))
.await;
outgoing
.register_request_context(RequestContext::new(
open_connection_request,
tracing::info_span!("app_server.request", rpc.method = "turn/start"),
None,
/*parent_trace*/ None,
))
.await;
assert_eq!(outgoing.request_context_count().await, 2);
codex-rs/app-server/src/thread_status.rs
+8 -4
View File
@@ -530,7 +530,7 @@ mod tests {
#[test]
fn resolves_in_progress_turn_to_active_status() {
let status = resolve_thread_status(ThreadStatus::Idle, true);
let status = resolve_thread_status(ThreadStatus::Idle, /*has_in_progress_turn*/ true);
assert_eq!(
status,
ThreadStatus::Active {
@@ -538,7 +538,8 @@ mod tests {
}
);
let status = resolve_thread_status(ThreadStatus::NotLoaded, true);
let status =
resolve_thread_status(ThreadStatus::NotLoaded, /*has_in_progress_turn*/ true);
assert_eq!(
status,
ThreadStatus::Active {
@@ -550,11 +551,14 @@ mod tests {
#[test]
fn keeps_status_when_no_in_progress_turn() {
assert_eq!(
resolve_thread_status(ThreadStatus::Idle, false),
resolve_thread_status(ThreadStatus::Idle, /*has_in_progress_turn*/ false),
ThreadStatus::Idle
);
assert_eq!(
resolve_thread_status(ThreadStatus::SystemError, false),
resolve_thread_status(
ThreadStatus::SystemError,
/*has_in_progress_turn*/ false
),
ThreadStatus::SystemError
);
}
codex-rs/app-server/src/transport/auth.rs
+40 -11
View File
@@ -506,8 +506,14 @@ mod tests {
}),
);
let tampered = token.replace(".eyJleHAi", ".eyJleHBi");
let err = verify_signed_bearer_token(&tampered, shared_secret, None, None, 30)
.expect_err("tampered jwt should fail");
let err = verify_signed_bearer_token(
&tampered,
shared_secret,
/*issuer*/ None,
/*audience*/ None,
/*max_clock_skew_seconds*/ 30,
)
.expect_err("tampered jwt should fail");
assert_eq!(err.status_code(), StatusCode::UNAUTHORIZED);
}
@@ -522,8 +528,14 @@ mod tests {
"aud": "audience",
}),
);
verify_signed_bearer_token(&token, shared_secret, Some("issuer"), Some("audience"), 30)
.expect("valid signed token should verify");
verify_signed_bearer_token(
&token,
shared_secret,
Some("issuer"),
Some("audience"),
/*max_clock_skew_seconds*/ 30,
)
.expect("valid signed token should verify");
}
#[test]
@@ -536,8 +548,14 @@ mod tests {
"aud": ["other-audience", "audience"],
}),
);
verify_signed_bearer_token(&token, shared_secret, None, Some("audience"), 30)
.expect("jwt audience arrays should verify");
verify_signed_bearer_token(
&token,
shared_secret,
/*issuer*/ None,
Some("audience"),
/*max_clock_skew_seconds*/ 30,
)
.expect("jwt audience arrays should verify");
}
#[test]
@@ -550,9 +568,14 @@ mod tests {
);
let header_segment = URL_SAFE_NO_PAD.encode(br#"{"alg":"none","typ":"JWT"}"#);
let token = format!("{header_segment}.{claims_segment}.");
let err =
verify_signed_bearer_token(&token, b"0123456789abcdef0123456789abcdef", None, None, 30)
.expect_err("alg=none jwt should be rejected");
let err = verify_signed_bearer_token(
&token,
b"0123456789abcdef0123456789abcdef",
/*issuer*/ None,
/*audience*/ None,
/*max_clock_skew_seconds*/ 30,
)
.expect_err("alg=none jwt should be rejected");
assert_eq!(err.status_code(), StatusCode::UNAUTHORIZED);
}
@@ -565,8 +588,14 @@ mod tests {
"iss": "issuer",
}),
);
let err = verify_signed_bearer_token(&token, shared_secret, None, None, 30)
.expect_err("jwt without exp should be rejected");
let err = verify_signed_bearer_token(
&token,
shared_secret,
/*issuer*/ None,
/*audience*/ None,
/*max_clock_skew_seconds*/ 30,
)
.expect_err("jwt without exp should be rejected");
assert_eq!(err.status_code(), StatusCode::UNAUTHORIZED);
}
codex-rs/app-server/src/transport/mod.rs
+6 -6
View File
@@ -646,7 +646,7 @@ mod tests {
initialized,
Arc::new(AtomicBool::new(true)),
opted_out_notification_methods,
None,
/*disconnect_sender*/ None,
),
);
@@ -686,7 +686,7 @@ mod tests {
Arc::new(AtomicBool::new(true)),
Arc::new(AtomicBool::new(true)),
Arc::new(RwLock::new(HashSet::from(["configWarning".to_string()]))),
None,
/*disconnect_sender*/ None,
),
);
@@ -726,7 +726,7 @@ mod tests {
Arc::new(AtomicBool::new(true)),
Arc::new(AtomicBool::new(true)),
Arc::new(RwLock::new(HashSet::new())),
None,
/*disconnect_sender*/ None,
),
);
@@ -772,7 +772,7 @@ mod tests {
Arc::new(AtomicBool::new(true)),
Arc::new(AtomicBool::new(false)),
Arc::new(RwLock::new(HashSet::new())),
None,
/*disconnect_sender*/ None,
),
);
@@ -834,7 +834,7 @@ mod tests {
Arc::new(AtomicBool::new(true)),
Arc::new(AtomicBool::new(true)),
Arc::new(RwLock::new(HashSet::new())),
None,
/*disconnect_sender*/ None,
),
);
@@ -1006,7 +1006,7 @@ mod tests {
Arc::new(AtomicBool::new(true)),
Arc::new(AtomicBool::new(true)),
Arc::new(RwLock::new(HashSet::new())),
None,
/*disconnect_sender*/ None,
),
);