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

Merge branch 'main' into codex/fix-agent-job-fanout-fill

Browse Source
This commit is contained in:
jif-oai
2026-04-21 19:33:39 +01:00
committed by GitHub
parent 5556d563bb 5bab04dcd7
commit d09040cd31
184 changed files with 13548 additions and 3133 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/Cargo.toml
+2
View File
@@ -72,8 +72,10 @@ serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true }
sha2 = { workspace = true }
tempfile = { workspace = true }
thiserror = { workspace = true }
time = { workspace = true }
toml = { workspace = true }
toml_edit = { workspace = true }
tokio = { workspace = true, features = [
"io-std",
"macros",
codex-rs/app-server/README.md
+2 -1
View File
@@ -1148,7 +1148,7 @@ the client can offer session-scoped and/or persistent approval choices.
### Permission requests
The built-in `request_permissions` tool sends an `item/permissions/requestApproval` JSON-RPC request to the client with the requested permission profile. This v2 payload mirrors the command-execution `additionalPermissions` shape: it can request network access and additional filesystem access.
The built-in `request_permissions` tool sends an `item/permissions/requestApproval` JSON-RPC request to the client with the requested permission profile. This v2 payload mirrors the command-execution `additionalPermissions` shape: it can request network access and additional filesystem access. The `cwd` field identifies the directory used to resolve cwd-relative permissions such as `:cwd`, `:project_roots`, and relative deny globs.
```json
{
@@ -1158,6 +1158,7 @@ The built-in `request_permissions` tool sends an `item/permissions/requestApprov
"threadId": "thr_123",
"turnId": "turn_123",
"itemId": "call_123",
"cwd": "/Users/me/project",
"reason": "Select a workspace root",
"permissions": {
"fileSystem": {
codex-rs/app-server/src/bespoke_event_handling.rs
+178 -23
View File
@@ -120,6 +120,7 @@ use codex_protocol::ThreadId;
use codex_protocol::dynamic_tools::DynamicToolCallOutputContentItem as CoreDynamicToolCallOutputContentItem;
use codex_protocol::dynamic_tools::DynamicToolResponse as CoreDynamicToolResponse;
use codex_protocol::items::parse_hook_prompt_message;
use codex_protocol::models::PermissionProfile as CorePermissionProfile;
use codex_protocol::plan_tool::UpdatePlanArgs;
use codex_protocol::protocol::CodexErrorInfo as CoreCodexErrorInfo;
use codex_protocol::protocol::Event;
@@ -908,27 +909,32 @@ pub(crate) async fn apply_bespoke_event_handling(
.note_permission_requested(&conversation_id.to_string())
.await;
let requested_permissions = request.permissions.clone();
let request_cwd = match request.cwd.clone() {
Some(cwd) => cwd,
None => conversation.config_snapshot().await.cwd,
};
let params = PermissionsRequestApprovalParams {
thread_id: conversation_id.to_string(),
turn_id: request.turn_id.clone(),
item_id: request.call_id.clone(),
cwd: request_cwd.clone(),
reason: request.reason,
permissions: request.permissions.into(),
};
let (pending_request_id, rx) = outgoing
.send_request(ServerRequestPayload::PermissionsRequestApproval(params))
.await;
let pending_response = PendingRequestPermissionsResponse {
call_id: request.call_id,
requested_permissions,
request_cwd,
pending_request_id,
receiver: rx,
request_permissions_guard: permission_guard,
};
tokio::spawn(async move {
on_request_permissions_response(
request.call_id,
requested_permissions,
pending_request_id,
rx,
conversation,
thread_state,
permission_guard,
)
.await;
on_request_permissions_response(pending_response, conversation, thread_state)
.await;
});
} else {
error!(
@@ -2706,20 +2712,26 @@ fn mcp_server_elicitation_response_from_client_result(
}
async fn on_request_permissions_response(
call_id: String,
requested_permissions: CoreRequestPermissionProfile,
pending_request_id: RequestId,
receiver: oneshot::Receiver<ClientRequestResult>,
pending_response: PendingRequestPermissionsResponse,
conversation: Arc<CodexThread>,
thread_state: Arc<Mutex<ThreadState>>,
request_permissions_guard: ThreadWatchActiveGuard,
) {
let PendingRequestPermissionsResponse {
call_id,
requested_permissions,
request_cwd,
pending_request_id,
receiver,
request_permissions_guard,
} = pending_response;
let response = receiver.await;
resolve_server_request_on_thread_listener(&thread_state, pending_request_id).await;
drop(request_permissions_guard);
let Some(response) =
request_permissions_response_from_client_result(requested_permissions, response)
else {
let Some(response) = request_permissions_response_from_client_result(
requested_permissions,
response,
request_cwd.as_path(),
) else {
return;
};
@@ -2734,9 +2746,19 @@ async fn on_request_permissions_response(
}
}
struct PendingRequestPermissionsResponse {
call_id: String,
requested_permissions: CoreRequestPermissionProfile,
request_cwd: AbsolutePathBuf,
pending_request_id: RequestId,
receiver: oneshot::Receiver<ClientRequestResult>,
request_permissions_guard: ThreadWatchActiveGuard,
}
fn request_permissions_response_from_client_result(
requested_permissions: CoreRequestPermissionProfile,
response: std::result::Result<ClientRequestResult, oneshot::error::RecvError>,
cwd: &std::path::Path,
) -> Option<CoreRequestPermissionsResponse> {
let value = match response {
Ok(Ok(value)) => value,
@@ -2765,12 +2787,14 @@ fn request_permissions_response_from_client_result(
scope: codex_app_server_protocol::PermissionGrantScope::Turn,
}
});
let granted_permissions: CorePermissionProfile = response.permissions.into();
let permissions = if granted_permissions.is_empty() {
CoreRequestPermissionProfile::default()
} else {
intersect_permission_profiles(requested_permissions.into(), granted_permissions, cwd).into()
};
Some(CoreRequestPermissionsResponse {
permissions: intersect_permission_profiles(
requested_permissions.into(),
response.permissions.into(),
)
.into(),
permissions,
scope: response.scope.to_core(),
})
}
@@ -3144,6 +3168,10 @@ mod tests {
use codex_protocol::models::FileSystemPermissions as CoreFileSystemPermissions;
use codex_protocol::models::MessagePhase;
use codex_protocol::models::NetworkPermissions as CoreNetworkPermissions;
use codex_protocol::permissions::FileSystemAccessMode;
use codex_protocol::permissions::FileSystemPath;
use codex_protocol::permissions::FileSystemSandboxEntry;
use codex_protocol::permissions::FileSystemSpecialPath;
use codex_protocol::plan_tool::PlanItemArg;
use codex_protocol::plan_tool::StepStatus;
use codex_protocol::protocol::CollabResumeBeginEvent;
@@ -3828,6 +3856,7 @@ mod tests {
let response = request_permissions_response_from_client_result(
CoreRequestPermissionProfile::default(),
Ok(Err(error)),
std::env::current_dir().expect("current dir").as_path(),
);
assert_eq!(response, None);
@@ -3914,12 +3943,14 @@ mod tests {
),
];
let cwd = std::env::current_dir().expect("current dir");
for (granted_permissions, expected_permissions) in cases {
let response = request_permissions_response_from_client_result(
requested_permissions.clone(),
Ok(Ok(serde_json::json!({
"permissions": granted_permissions,
}))),
cwd.as_path(),
)
.expect("response should be accepted");
@@ -3941,6 +3972,7 @@ mod tests {
"scope": "session",
"permissions": {},
}))),
std::env::current_dir().expect("current dir").as_path(),
)
.expect("response should be accepted");
@@ -3953,6 +3985,129 @@ mod tests {
);
}
#[test]
fn request_permissions_response_accepts_explicit_child_grant_for_requested_cwd_scope() {
let temp_dir = TempDir::new().expect("temp dir");
let cwd = AbsolutePathBuf::from_absolute_path(temp_dir.path()).expect("absolute cwd");
let child = cwd.join("child");
let requested_permissions = CoreRequestPermissionProfile {
file_system: Some(CoreFileSystemPermissions {
entries: vec![FileSystemSandboxEntry {
path: FileSystemPath::Special {
value: FileSystemSpecialPath::CurrentWorkingDirectory,
},
access: FileSystemAccessMode::Write,
}],
glob_scan_max_depth: None,
}),
..Default::default()
};
let response = request_permissions_response_from_client_result(
requested_permissions,
Ok(Ok(serde_json::json!({
"permissions": {
"fileSystem": {
"write": [child],
},
},
}))),
cwd.as_path(),
)
.expect("response should be accepted");
assert_eq!(
response.permissions,
CoreRequestPermissionProfile {
file_system: Some(CoreFileSystemPermissions::from_read_write_roots(
/*read*/ None,
Some(vec![child]),
)),
..Default::default()
}
);
}
#[test]
fn request_permissions_response_rejects_child_grant_outside_requested_cwd_scope() {
let temp_dir = TempDir::new().expect("temp dir");
let request_cwd = AbsolutePathBuf::from_absolute_path(temp_dir.path().join("request-cwd"))
.expect("absolute request cwd");
let later_cwd = AbsolutePathBuf::from_absolute_path(temp_dir.path().join("later-cwd"))
.expect("absolute later cwd");
let later_child = later_cwd.join("child");
let requested_permissions = CoreRequestPermissionProfile {
file_system: Some(CoreFileSystemPermissions {
entries: vec![FileSystemSandboxEntry {
path: FileSystemPath::Special {
value: FileSystemSpecialPath::CurrentWorkingDirectory,
},
access: FileSystemAccessMode::Write,
}],
glob_scan_max_depth: None,
}),
..Default::default()
};
let response = request_permissions_response_from_client_result(
requested_permissions,
Ok(Ok(serde_json::json!({
"permissions": {
"fileSystem": {
"write": [later_child],
},
},
}))),
request_cwd.as_path(),
)
.expect("response should be accepted");
assert_eq!(
response.permissions,
CoreRequestPermissionProfile::default()
);
}
#[test]
fn request_permissions_response_ignores_broader_cwd_grant_for_requested_child_path() {
let temp_dir = TempDir::new().expect("temp dir");
let cwd = AbsolutePathBuf::from_absolute_path(temp_dir.path()).expect("absolute cwd");
let child = cwd.join("child");
let requested_permissions = CoreRequestPermissionProfile {
file_system: Some(CoreFileSystemPermissions::from_read_write_roots(
/*read*/ None,
Some(vec![child]),
)),
..Default::default()
};
let response = request_permissions_response_from_client_result(
requested_permissions,
Ok(Ok(serde_json::json!({
"permissions": {
"fileSystem": {
"entries": [{
"path": {
"type": "special",
"value": {
"kind": "current_working_directory"
}
},
"access": "write"
}],
},
},
}))),
cwd.as_path(),
)
.expect("response should be accepted");
assert_eq!(
response.permissions,
CoreRequestPermissionProfile::default()
);
}
#[test]
fn collab_resume_begin_maps_to_item_started_resume_agent() {
let event = CollabResumeBeginEvent {
codex-rs/app-server/src/codex_message_processor.rs
+127 -307
View File
@@ -2,7 +2,7 @@ use crate::bespoke_event_handling::apply_bespoke_event_handling;
use crate::bespoke_event_handling::maybe_emit_hook_prompt_item_completed;
use crate::command_exec::CommandExecManager;
use crate::command_exec::StartCommandExecParams;
use crate::config_api::apply_runtime_feature_enablement;
use crate::config_manager::ConfigManager;
use crate::error_code::INPUT_TOO_LARGE_ERROR_CODE;
use crate::error_code::INTERNAL_ERROR_CODE;
use crate::error_code::INVALID_PARAMS_ERROR_CODE;
@@ -212,8 +212,6 @@ use codex_arg0::Arg0DispatchPaths;
use codex_backend_client::AddCreditsNudgeCreditType as BackendAddCreditsNudgeCreditType;
use codex_backend_client::Client as BackendClient;
use codex_chatgpt::connectors;
use codex_cloud_requirements::cloud_requirements_loader;
use codex_config::ThreadConfigLoader;
use codex_config::types::McpServerTransportConfig;
use codex_core::CodexThread;
use codex_core::ForkSnapshot;
@@ -231,9 +229,6 @@ use codex_core::config::edit::ConfigEdit;
use codex_core::config::edit::ConfigEditsBuilder;
use codex_core::config_loader::CloudRequirementsLoadError;
use codex_core::config_loader::CloudRequirementsLoadErrorCode;
use codex_core::config_loader::CloudRequirementsLoader;
use codex_core::config_loader::LoaderOverrides;
use codex_core::config_loader::load_config_layers_state;
use codex_core::config_loader::project_trust_key;
use codex_core::exec::ExecCapturePolicy;
use codex_core::exec::ExecExpiration;
@@ -280,7 +275,6 @@ 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::default_client::set_default_client_residency_requirement;
use codex_login::login_with_api_key;
use codex_login::request_device_code;
use codex_login::run_login_server;
@@ -348,16 +342,13 @@ use codex_thread_store::ThreadStore;
use codex_thread_store::ThreadStoreError;
use codex_thread_store::UpdateThreadMetadataParams as StoreUpdateThreadMetadataParams;
use codex_utils_absolute_path::AbsolutePathBuf;
use codex_utils_json_to_toml::json_to_toml;
use codex_utils_pty::DEFAULT_OUTPUT_BYTES_CAP;
use std::collections::BTreeMap;
use std::collections::HashMap;
use std::collections::HashSet;
use std::io::Error as IoError;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::RwLock;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
use std::time::Duration;
@@ -478,10 +469,7 @@ pub(crate) struct CodexMessageProcessor {
arg0_paths: Arg0DispatchPaths,
config: Arc<Config>,
thread_store: Arc<dyn ThreadStore>,
cli_overrides: Arc<RwLock<Vec<(String, TomlValue)>>>,
runtime_feature_enablement: Arc<RwLock<BTreeMap<String, bool>>>,
cloud_requirements: Arc<RwLock<CloudRequirementsLoader>>,
thread_config_loader: Arc<dyn ThreadConfigLoader>,
config_manager: ConfigManager,
active_login: Arc<Mutex<Option<ActiveLogin>>>,
pending_thread_unloads: Arc<Mutex<HashSet<ThreadId>>>,
thread_state_manager: ThreadStateManager,
@@ -638,11 +626,10 @@ pub(crate) struct CodexMessageProcessorArgs {
pub(crate) outgoing: Arc<OutgoingMessageSender>,
pub(crate) analytics_events_client: AnalyticsEventsClient,
pub(crate) arg0_paths: Arg0DispatchPaths,
/// Startup config used as the process baseline. Fresh effective config loads
/// go through `config_manager`.
pub(crate) config: Arc<Config>,
pub(crate) cli_overrides: Arc<RwLock<Vec<(String, TomlValue)>>>,
pub(crate) runtime_feature_enablement: Arc<RwLock<BTreeMap<String, bool>>>,
pub(crate) cloud_requirements: Arc<RwLock<CloudRequirementsLoader>>,
pub(crate) thread_config_loader: Arc<dyn ThreadConfigLoader>,
pub(crate) config_manager: ConfigManager,
pub(crate) feedback: CodexFeedback,
pub(crate) log_db: Option<LogDbLayer>,
}
@@ -727,10 +714,7 @@ impl CodexMessageProcessor {
analytics_events_client,
arg0_paths,
config,
cli_overrides,
runtime_feature_enablement,
cloud_requirements,
thread_config_loader,
config_manager,
feedback,
log_db,
} = args;
@@ -742,10 +726,7 @@ impl CodexMessageProcessor {
arg0_paths,
thread_store: configured_thread_store(&config),
config,
cli_overrides,
runtime_feature_enablement,
cloud_requirements,
thread_config_loader,
config_manager,
active_login: Arc::new(Mutex::new(None)),
pending_thread_unloads: Arc::new(Mutex::new(HashSet::new())),
thread_state_manager: ThreadStateManager::new(),
@@ -763,44 +744,14 @@ impl CodexMessageProcessor {
&self,
fallback_cwd: Option<PathBuf>,
) -> Result<Config, JSONRPCErrorError> {
let cloud_requirements = self.current_cloud_requirements();
let mut config = codex_core::config::ConfigBuilder::default()
.cli_overrides(self.current_cli_overrides())
.fallback_cwd(fallback_cwd)
.cloud_requirements(cloud_requirements)
.build()
self.config_manager
.load_latest_config(fallback_cwd)
.await
.map_err(|err| JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to reload config: {err}"),
data: None,
})?;
apply_runtime_feature_enablement(&mut config, &self.current_runtime_feature_enablement());
config.codex_self_exe = self.arg0_paths.codex_self_exe.clone();
config.codex_linux_sandbox_exe = self.arg0_paths.codex_linux_sandbox_exe.clone();
config.main_execve_wrapper_exe = self.arg0_paths.main_execve_wrapper_exe.clone();
Ok(config)
}
fn current_cloud_requirements(&self) -> CloudRequirementsLoader {
self.cloud_requirements
.read()
.map(|guard| guard.clone())
.unwrap_or_default()
}
fn current_cli_overrides(&self) -> Vec<(String, TomlValue)> {
self.cli_overrides
.read()
.map(|guard| guard.clone())
.unwrap_or_default()
}
fn current_runtime_feature_enablement(&self) -> BTreeMap<String, bool> {
self.runtime_feature_enablement
.read()
.map(|guard| guard.clone())
.unwrap_or_default()
})
}
/// If a client sends `developer_instructions: null` during a mode switch,
@@ -1186,6 +1137,11 @@ impl CodexMessageProcessor {
self.command_exec_terminate(to_connection_request_id(request_id), params)
.await;
}
ClientRequest::DeviceKeyCreate { .. }
| ClientRequest::DeviceKeyPublic { .. }
| ClientRequest::DeviceKeySign { .. } => {
warn!("Device key request reached CodexMessageProcessor unexpectedly");
}
ClientRequest::ConfigRead { .. }
| ClientRequest::ConfigValueWrite { .. }
| ClientRequest::ConfigBatchWrite { .. }
@@ -1419,10 +1375,8 @@ impl CodexMessageProcessor {
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 config_manager = self.config_manager.clone();
let chatgpt_base_url = self.config.chatgpt_base_url.clone();
let codex_home = self.config.codex_home.to_path_buf();
let cli_overrides = self.current_cli_overrides();
let auth_url = server.auth_url.clone();
tokio::spawn(async move {
let (success, error_msg) = match tokio::time::timeout(
@@ -1452,17 +1406,13 @@ impl CodexMessageProcessor {
if success {
auth_manager.reload();
replace_cloud_requirements_loader(
cloud_requirements.as_ref(),
config_manager.replace_cloud_requirements_loader(
auth_manager.clone(),
chatgpt_base_url,
codex_home,
);
sync_default_client_residency_requirement(
&cli_overrides,
cloud_requirements.as_ref(),
)
.await;
config_manager
.sync_default_client_residency_requirement()
.await;
// Notify clients with the actual current auth mode.
let auth = auth_manager.auth_cached();
@@ -1536,10 +1486,8 @@ impl CodexMessageProcessor {
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 config_manager = self.config_manager.clone();
let chatgpt_base_url = self.config.chatgpt_base_url.clone();
let codex_home = self.config.codex_home.to_path_buf();
let cli_overrides = self.current_cli_overrides();
tokio::spawn(async move {
let (success, error_msg) = tokio::select! {
_ = cancel.cancelled() => {
@@ -1566,17 +1514,13 @@ impl CodexMessageProcessor {
if success {
auth_manager.reload();
replace_cloud_requirements_loader(
cloud_requirements.as_ref(),
config_manager.replace_cloud_requirements_loader(
auth_manager.clone(),
chatgpt_base_url,
codex_home,
);
sync_default_client_residency_requirement(
&cli_overrides,
cloud_requirements.as_ref(),
)
.await;
config_manager
.sync_default_client_residency_requirement()
.await;
let auth = auth_manager.auth_cached();
let payload_v2 = AccountUpdatedNotification {
@@ -1706,14 +1650,12 @@ impl CodexMessageProcessor {
return;
}
self.auth_manager.reload();
replace_cloud_requirements_loader(
self.cloud_requirements.as_ref(),
self.config_manager.replace_cloud_requirements_loader(
self.auth_manager.clone(),
self.config.chatgpt_base_url.clone(),
self.config.codex_home.to_path_buf(),
);
let cli_overrides = self.current_cli_overrides();
sync_default_client_residency_requirement(&cli_overrides, self.cloud_requirements.as_ref())
self.config_manager
.sync_default_client_residency_requirement()
.await;
self.outgoing
@@ -2424,8 +2366,6 @@ impl CodexMessageProcessor {
personality,
);
typesafe_overrides.ephemeral = ephemeral;
let cloud_requirements = self.current_cloud_requirements();
let cli_overrides = self.current_cli_overrides();
let listener_task_context = ListenerTaskContext {
thread_manager: Arc::clone(&self.thread_manager),
thread_state_manager: self.thread_state_manager.clone(),
@@ -2438,15 +2378,11 @@ impl CodexMessageProcessor {
codex_home: self.config.codex_home.to_path_buf(),
};
let request_trace = request_context.request_trace();
let runtime_feature_enablement = self.current_runtime_feature_enablement();
let thread_config_loader = Arc::clone(&self.thread_config_loader);
let config_manager = self.config_manager.clone();
let thread_start_task = async move {
Self::thread_start_task(
listener_task_context,
cli_overrides,
runtime_feature_enablement,
cloud_requirements,
thread_config_loader,
config_manager,
request_id,
app_server_client_name,
app_server_client_version,
@@ -2520,10 +2456,7 @@ impl CodexMessageProcessor {
#[allow(clippy::too_many_arguments)]
async fn thread_start_task(
listener_task_context: ListenerTaskContext,
cli_overrides: Vec<(String, TomlValue)>,
runtime_feature_enablement: BTreeMap<String, bool>,
cloud_requirements: CloudRequirementsLoader,
thread_config_loader: Arc<dyn ThreadConfigLoader>,
config_manager: ConfigManager,
request_id: ConnectionRequestId,
app_server_client_name: Option<String>,
app_server_client_version: Option<String>,
@@ -2537,16 +2470,9 @@ impl CodexMessageProcessor {
request_trace: Option<W3cTraceContext>,
) {
let requested_cwd = typesafe_overrides.cwd.clone();
let mut config = match derive_config_from_params(
&cli_overrides,
config_overrides.clone(),
typesafe_overrides.clone(),
Arc::clone(&thread_config_loader),
&cloud_requirements,
&listener_task_context.codex_home,
&runtime_feature_enablement,
)
.await
let mut config = match config_manager
.load_with_overrides(config_overrides.clone(), typesafe_overrides.clone())
.await
{
Ok(config) => config,
Err(err) => {
@@ -2585,6 +2511,7 @@ impl CodexMessageProcessor {
let trust_target = resolve_root_git_project_for_trust(LOCAL_FS.as_ref(), &config.cwd)
.await
.unwrap_or_else(|| config.cwd.clone());
let current_cli_overrides = config_manager.current_cli_overrides();
let cli_overrides_with_trust;
let cli_overrides_for_reload = if let Err(err) =
codex_core::config::set_project_trust_level(
@@ -2606,7 +2533,7 @@ impl CodexMessageProcessor {
project_trust_key(trust_target.as_path()),
TomlValue::Table(project),
);
cli_overrides_with_trust = cli_overrides
cli_overrides_with_trust = current_cli_overrides
.iter()
.cloned()
.chain(std::iter::once((
@@ -2616,19 +2543,17 @@ impl CodexMessageProcessor {
.collect::<Vec<_>>();
cli_overrides_with_trust.as_slice()
} else {
&cli_overrides
current_cli_overrides.as_slice()
};
config = match derive_config_from_params(
cli_overrides_for_reload,
config_overrides,
typesafe_overrides,
thread_config_loader,
&cloud_requirements,
&listener_task_context.codex_home,
&runtime_feature_enablement,
)
.await
config = match config_manager
.load_with_cli_overrides(
cli_overrides_for_reload,
config_overrides,
typesafe_overrides,
/*fallback_cwd*/ None,
)
.await
{
Ok(config) => config,
Err(err) => {
@@ -4412,19 +4337,10 @@ impl CodexMessageProcessor {
.await;
// Derive a Config using the same logic as new conversation, honoring overrides if provided.
let cloud_requirements = self.current_cloud_requirements();
let cli_overrides = self.current_cli_overrides();
let runtime_feature_enablement = self.current_runtime_feature_enablement();
let config = match derive_config_for_cwd(
&cli_overrides,
request_overrides,
typesafe_overrides,
history_cwd,
&cloud_requirements,
&self.config.codex_home,
&runtime_feature_enablement,
)
.await
let config = match self
.config_manager
.load_for_cwd(request_overrides, typesafe_overrides, history_cwd)
.await
{
Ok(config) => config,
Err(err) => {
@@ -4992,19 +4908,10 @@ impl CodexMessageProcessor {
);
typesafe_overrides.ephemeral = ephemeral.then_some(true);
// Derive a Config using the same logic as new conversation, honoring overrides if provided.
let cloud_requirements = self.current_cloud_requirements();
let cli_overrides = self.current_cli_overrides();
let runtime_feature_enablement = self.current_runtime_feature_enablement();
let config = match derive_config_for_cwd(
&cli_overrides,
request_overrides,
typesafe_overrides,
history_cwd,
&cloud_requirements,
&self.config.codex_home,
&runtime_feature_enablement,
)
.await
let config = match self
.config_manager
.load_for_cwd(request_overrides, typesafe_overrides, history_cwd)
.await
{
Ok(config) => config,
Err(err) => {
@@ -6041,17 +5948,19 @@ impl CodexMessageProcessor {
params: McpServerToolCallParams,
) {
let outgoing = Arc::clone(&self.outgoing);
let (_, thread) = match self.load_thread(&params.thread_id).await {
let thread_id = params.thread_id.clone();
let (_, thread) = match self.load_thread(&thread_id).await {
Ok(thread) => thread,
Err(error) => {
self.outgoing.send_error(request_id, error).await;
return;
}
};
let meta = with_mcp_tool_call_thread_id_meta(params.meta, &thread_id);
tokio::spawn(async move {
let result = thread
.call_mcp_tool(&params.server, &params.tool, params.arguments, params.meta)
.call_mcp_tool(&params.server, &params.tool, params.arguments, meta)
.await;
match result {
Ok(result) => {
@@ -6577,8 +6486,6 @@ impl CodexMessageProcessor {
return;
}
};
let cli_overrides = self.current_cli_overrides();
let host_name = codex_config::host_name();
let mut data = Vec::new();
for cwd in cwds {
let extra_roots = extra_roots_by_cwd
@@ -6599,17 +6506,10 @@ impl CodexMessageProcessor {
continue;
}
};
let config_layer_stack = match load_config_layers_state(
LOCAL_FS.as_ref(),
&self.config.codex_home,
Some(cwd_abs.clone()),
&cli_overrides,
LoaderOverrides::default(),
CloudRequirementsLoader::default(),
self.thread_config_loader.as_ref(),
host_name.as_deref(),
)
.await
let config_layer_stack = match self
.config_manager
.load_config_layers_for_cwd(cwd_abs.clone())
.await
{
Ok(config_layer_stack) => config_layer_stack,
Err(err) => {
@@ -8765,30 +8665,25 @@ impl CodexMessageProcessor {
WindowsSandboxSetupMode::Unelevated => CoreWindowsSandboxSetupMode::Unelevated,
};
let config = Arc::clone(&self.config);
let cloud_requirements = self.current_cloud_requirements();
let config_manager = self.config_manager.clone();
let command_cwd = params
.cwd
.map(PathBuf::from)
.unwrap_or_else(|| config.cwd.to_path_buf());
let cli_overrides = self.current_cli_overrides();
let runtime_feature_enablement = self.current_runtime_feature_enablement();
let outgoing = Arc::clone(&self.outgoing);
let connection_id = request_id.connection_id;
tokio::spawn(async move {
let derived_config = derive_config_for_cwd(
&cli_overrides,
/*request_overrides*/ None,
ConfigOverrides {
cwd: Some(command_cwd.clone()),
..Default::default()
},
Some(command_cwd.clone()),
&cloud_requirements,
&config.codex_home,
&runtime_feature_enablement,
)
.await;
let derived_config = config_manager
.load_for_cwd(
/*request_overrides*/ None,
ConfigOverrides {
cwd: Some(command_cwd.clone()),
..Default::default()
},
Some(command_cwd.clone()),
)
.await;
let setup_result = match derived_config {
Ok(config) => {
let setup_request = WindowsSandboxSetupRequest {
@@ -9518,116 +9413,6 @@ fn validate_dynamic_tools(tools: &[ApiDynamicToolSpec]) -> Result<(), String> {
Ok(())
}
fn replace_cloud_requirements_loader(
cloud_requirements: &RwLock<CloudRequirementsLoader>,
auth_manager: Arc<AuthManager>,
chatgpt_base_url: String,
codex_home: PathBuf,
) {
let loader = cloud_requirements_loader(auth_manager, chatgpt_base_url, codex_home);
if let Ok(mut guard) = cloud_requirements.write() {
*guard = loader;
} else {
warn!("failed to update cloud requirements loader");
}
}
async fn sync_default_client_residency_requirement(
cli_overrides: &[(String, TomlValue)],
cloud_requirements: &RwLock<CloudRequirementsLoader>,
) {
let loader = cloud_requirements
.read()
.map(|guard| guard.clone())
.unwrap_or_default();
match codex_core::config::ConfigBuilder::default()
.cli_overrides(cli_overrides.to_vec())
.cloud_requirements(loader)
.build()
.await
{
Ok(config) => set_default_client_residency_requirement(config.enforce_residency.value()),
Err(err) => warn!(
error = %err,
"failed to sync default client residency requirement after auth refresh"
),
}
}
/// Derive the effective [`Config`] by layering three override sources.
///
/// Precedence (lowest to highest):
/// - `cli_overrides`: process-wide startup `--config` flags.
/// - `request_overrides`: per-request dotted-path overrides (`params.config`), converted JSON->TOML.
/// - `typesafe_overrides`: Request objects such as `NewThreadParams` and
/// `ThreadStartParams` support a limited set of _explicit_ config overrides, so
/// `typesafe_overrides` is a `ConfigOverrides` derived from the respective request object.
/// Because the overrides are defined explicitly in the `*Params`, this takes priority over
/// the more general "bag of config options" provided by `cli_overrides` and `request_overrides`.
async fn derive_config_from_params(
cli_overrides: &[(String, TomlValue)],
request_overrides: Option<HashMap<String, serde_json::Value>>,
typesafe_overrides: ConfigOverrides,
thread_config_loader: Arc<dyn ThreadConfigLoader>,
cloud_requirements: &CloudRequirementsLoader,
codex_home: &Path,
runtime_feature_enablement: &BTreeMap<String, bool>,
) -> std::io::Result<Config> {
let merged_cli_overrides = cli_overrides
.iter()
.cloned()
.chain(
request_overrides
.unwrap_or_default()
.into_iter()
.map(|(k, v)| (k, json_to_toml(v))),
)
.collect::<Vec<_>>();
let mut config = codex_core::config::ConfigBuilder::default()
.codex_home(codex_home.to_path_buf())
.cli_overrides(merged_cli_overrides)
.harness_overrides(typesafe_overrides)
.cloud_requirements(cloud_requirements.clone())
.thread_config_loader(thread_config_loader)
.build()
.await?;
apply_runtime_feature_enablement(&mut config, runtime_feature_enablement);
Ok(config)
}
async fn derive_config_for_cwd(
cli_overrides: &[(String, TomlValue)],
request_overrides: Option<HashMap<String, serde_json::Value>>,
typesafe_overrides: ConfigOverrides,
cwd: Option<PathBuf>,
cloud_requirements: &CloudRequirementsLoader,
codex_home: &Path,
runtime_feature_enablement: &BTreeMap<String, bool>,
) -> std::io::Result<Config> {
let merged_cli_overrides = cli_overrides
.iter()
.cloned()
.chain(
request_overrides
.unwrap_or_default()
.into_iter()
.map(|(k, v)| (k, json_to_toml(v))),
)
.collect::<Vec<_>>();
let mut config = codex_core::config::ConfigBuilder::default()
.codex_home(codex_home.to_path_buf())
.cli_overrides(merged_cli_overrides)
.harness_overrides(typesafe_overrides)
.fallback_cwd(cwd)
.cloud_requirements(cloud_requirements.clone())
.build()
.await?;
apply_runtime_feature_enablement(&mut config, runtime_feature_enablement);
Ok(config)
}
async fn read_history_cwd_from_state_db(
config: &Config,
thread_id: Option<ThreadId>,
@@ -9891,6 +9676,32 @@ fn thread_store_archive_error(operation: &str, err: ThreadStoreError) -> JSONRPC
}
}
const MCP_TOOL_THREAD_ID_META_KEY: &str = "threadId";
fn with_mcp_tool_call_thread_id_meta(
meta: Option<serde_json::Value>,
thread_id: &str,
) -> Option<serde_json::Value> {
match meta {
Some(serde_json::Value::Object(mut map)) => {
map.insert(
MCP_TOOL_THREAD_ID_META_KEY.to_string(),
serde_json::Value::String(thread_id.to_string()),
);
Some(serde_json::Value::Object(map))
}
None => {
let mut map = serde_json::Map::new();
map.insert(
MCP_TOOL_THREAD_ID_META_KEY.to_string(),
serde_json::Value::String(thread_id.to_string()),
);
Some(serde_json::Value::Object(map))
}
other => other,
}
}
fn summary_from_stored_thread(
thread: StoredThread,
fallback_provider: &str,
@@ -10523,6 +10334,8 @@ mod tests {
use codex_config::SessionThreadConfig;
use codex_config::StaticThreadConfigLoader;
use codex_config::ThreadConfigSource;
use codex_core::config_loader::CloudRequirementsLoader;
use codex_core::config_loader::LoaderOverrides;
use codex_model_provider_info::ModelProviderInfo;
use codex_model_provider_info::WireApi;
use codex_protocol::ThreadId;
@@ -10536,8 +10349,10 @@ mod tests {
use codex_utils_absolute_path::test_support::test_path_buf;
use pretty_assertions::assert_eq;
use serde_json::json;
use std::collections::BTreeMap;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::RwLock;
use tempfile::TempDir;
#[test]
@@ -10811,21 +10626,13 @@ mod tests {
requires_openai_auth: false,
supports_websockets: true,
};
let config = derive_config_from_params(
&[],
Some(HashMap::from([
("model_provider".to_string(), json!("request")),
("features.plugins".to_string(), json!(true)),
(
"model_providers.session".to_string(),
json!({
"name": "request",
"base_url": "http://127.0.0.1:9999/api/codex",
"wire_api": "responses",
}),
),
])),
ConfigOverrides::default(),
let config_manager = ConfigManager::new(
temp_dir.path().to_path_buf(),
Arc::new(RwLock::new(Vec::new())),
Arc::new(RwLock::new(BTreeMap::new())),
LoaderOverrides::default(),
Arc::new(RwLock::new(CloudRequirementsLoader::default())),
Arg0DispatchPaths::default(),
Arc::new(StaticThreadConfigLoader::new(vec![
ThreadConfigSource::Session(SessionThreadConfig {
model_provider: Some("session".to_string()),
@@ -10836,11 +10643,24 @@ mod tests {
features: BTreeMap::from([("plugins".to_string(), false)]),
}),
])),
&CloudRequirementsLoader::default(),
temp_dir.path(),
&BTreeMap::new(),
)
.await?;
);
let config = config_manager
.load_with_overrides(
Some(HashMap::from([
("model_provider".to_string(), json!("request")),
("features.plugins".to_string(), json!(true)),
(
"model_providers.session".to_string(),
json!({
"name": "request",
"base_url": "http://127.0.0.1:9999/api/codex",
"wire_api": "responses",
}),
),
])),
ConfigOverrides::default(),
)
.await?;
assert_eq!(config.model_provider_id, "session");
assert_eq!(config.model_provider, session_provider);
codex-rs/app-server/src/config/external_agent_config.rs
+1185
View File
File diff suppressed because it is too large Load Diff
codex-rs/app-server/src/config/external_agent_config_tests.rs
+1622
View File
File diff suppressed because it is too large Load Diff
codex-rs/app-server/src/config/mod.rs
+1
View File
@@ -0,0 +1 @@
pub(crate) mod external_agent_config;
codex-rs/app-server/src/config_api.rs
+39 -134
View File
@@ -1,3 +1,5 @@
use crate::config_manager::ConfigManager;
use crate::config_manager_service::ConfigManagerError;
use crate::error_code::INTERNAL_ERROR_CODE;
use crate::error_code::INVALID_REQUEST_ERROR_CODE;
use async_trait::async_trait;
@@ -19,11 +21,7 @@ use codex_app_server_protocol::NetworkUnixSocketPermission;
use codex_app_server_protocol::SandboxMode;
use codex_core::ThreadManager;
use codex_core::config::Config;
use codex_core::config::ConfigService;
use codex_core::config::ConfigServiceError;
use codex_core::config_loader::CloudRequirementsLoader;
use codex_core::config_loader::ConfigRequirementsToml;
use codex_core::config_loader::LoaderOverrides;
use codex_core::config_loader::ResidencyRequirement as CoreResidencyRequirement;
use codex_core::config_loader::SandboxModeRequirement as CoreSandboxModeRequirement;
use codex_core::plugins::PluginId;
@@ -34,12 +32,8 @@ use codex_features::feature_for_key;
use codex_protocol::config_types::WebSearchMode;
use codex_protocol::protocol::Op;
use serde_json::json;
use std::collections::BTreeMap;
use std::collections::BTreeSet;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::RwLock;
use toml::Value as TomlValue;
use tracing::warn;
const SUPPORTED_EXPERIMENTAL_FEATURE_ENABLEMENT: &[&str] = &[
@@ -72,86 +66,36 @@ impl UserConfigReloader for ThreadManager {
#[derive(Clone)]
pub(crate) struct ConfigApi {
codex_home: PathBuf,
cli_overrides: Arc<RwLock<Vec<(String, TomlValue)>>>,
runtime_feature_enablement: Arc<RwLock<BTreeMap<String, bool>>>,
loader_overrides: LoaderOverrides,
cloud_requirements: Arc<RwLock<CloudRequirementsLoader>>,
config_manager: ConfigManager,
user_config_reloader: Arc<dyn UserConfigReloader>,
analytics_events_client: AnalyticsEventsClient,
}
impl ConfigApi {
pub(crate) fn new(
codex_home: PathBuf,
cli_overrides: Arc<RwLock<Vec<(String, TomlValue)>>>,
runtime_feature_enablement: Arc<RwLock<BTreeMap<String, bool>>>,
loader_overrides: LoaderOverrides,
cloud_requirements: Arc<RwLock<CloudRequirementsLoader>>,
config_manager: ConfigManager,
user_config_reloader: Arc<dyn UserConfigReloader>,
analytics_events_client: AnalyticsEventsClient,
) -> Self {
Self {
codex_home,
cli_overrides,
runtime_feature_enablement,
loader_overrides,
cloud_requirements,
config_manager,
user_config_reloader,
analytics_events_client,
}
}
fn config_service(&self) -> ConfigService {
ConfigService::new(
self.codex_home.clone(),
self.current_cli_overrides(),
self.loader_overrides.clone(),
self.current_cloud_requirements(),
codex_config::host_name(),
)
}
fn current_cli_overrides(&self) -> Vec<(String, TomlValue)> {
self.cli_overrides
.read()
.map(|guard| guard.clone())
.unwrap_or_default()
}
fn current_runtime_feature_enablement(&self) -> BTreeMap<String, bool> {
self.runtime_feature_enablement
.read()
.map(|guard| guard.clone())
.unwrap_or_default()
}
fn current_cloud_requirements(&self) -> CloudRequirementsLoader {
self.cloud_requirements
.read()
.map(|guard| guard.clone())
.unwrap_or_default()
}
pub(crate) async fn load_latest_config(
&self,
fallback_cwd: Option<PathBuf>,
) -> Result<Config, JSONRPCErrorError> {
let mut config = codex_core::config::ConfigBuilder::default()
.codex_home(self.codex_home.clone())
.cli_overrides(self.current_cli_overrides())
.loader_overrides(self.loader_overrides.clone())
.fallback_cwd(fallback_cwd)
.cloud_requirements(self.current_cloud_requirements())
.build()
self.config_manager
.load_latest_config(fallback_cwd)
.await
.map_err(|err| JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to resolve feature override precedence: {err}"),
data: None,
})?;
apply_runtime_feature_enablement(&mut config, &self.current_runtime_feature_enablement());
Ok(config)
})
}
pub(crate) async fn read(
@@ -159,11 +103,7 @@ impl ConfigApi {
params: ConfigReadParams,
) -> Result<ConfigReadResponse, JSONRPCErrorError> {
let fallback_cwd = params.cwd.as_ref().map(PathBuf::from);
let mut response = self
.config_service()
.read(params)
.await
.map_err(map_error)?;
let mut response = self.config_manager.read(params).await.map_err(map_error)?;
let config = self.load_latest_config(fallback_cwd).await?;
for feature_key in SUPPORTED_EXPERIMENTAL_FEATURE_ENABLEMENT {
let Some(feature) = feature_for_key(feature_key) else {
@@ -191,7 +131,7 @@ impl ConfigApi {
&self,
) -> Result<ConfigRequirementsReadResponse, JSONRPCErrorError> {
let requirements = self
.config_service()
.config_manager
.read_requirements()
.await
.map_err(map_error)?
@@ -207,7 +147,7 @@ impl ConfigApi {
let pending_changes =
collect_plugin_enabled_candidates([(&params.key_path, &params.value)].into_iter());
let response = self
.config_service()
.config_manager
.write_value(params)
.await
.map_err(map_error)?;
@@ -227,7 +167,7 @@ impl ConfigApi {
.map(|edit| (&edit.key_path, &edit.value)),
);
let response = self
.config_service()
.config_manager
.batch_write(params)
.await
.map_err(map_error)?;
@@ -278,21 +218,17 @@ impl ConfigApi {
return Ok(ExperimentalFeatureEnablementSetResponse { enablement });
}
{
let mut runtime_feature_enablement =
self.runtime_feature_enablement
.write()
.map_err(|_| JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: "failed to update feature enablement".to_string(),
data: None,
})?;
runtime_feature_enablement.extend(
self.config_manager
.extend_runtime_feature_enablement(
enablement
.iter()
.map(|(name, enabled)| (name.clone(), *enabled)),
);
}
)
.map_err(|_| JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: "failed to update feature enablement".to_string(),
data: None,
})?;
self.load_latest_config(/*fallback_cwd*/ None).await?;
self.user_config_reloader.reload_user_config().await;
@@ -309,7 +245,8 @@ impl ConfigApi {
continue;
};
let metadata =
installed_plugin_telemetry_metadata(self.codex_home.as_path(), &plugin_id).await;
installed_plugin_telemetry_metadata(self.config_manager.codex_home(), &plugin_id)
.await;
if enabled {
self.analytics_events_client.track_plugin_enabled(metadata);
} else {
@@ -319,49 +256,6 @@ impl ConfigApi {
}
}
pub(crate) fn protected_feature_keys(
config_layer_stack: &codex_core::config_loader::ConfigLayerStack,
) -> BTreeSet<String> {
let mut protected_features = config_layer_stack
.effective_config()
.get("features")
.and_then(toml::Value::as_table)
.map(|features| features.keys().cloned().collect::<BTreeSet<_>>())
.unwrap_or_default();
if let Some(feature_requirements) = config_layer_stack
.requirements_toml()
.feature_requirements
.as_ref()
{
protected_features.extend(feature_requirements.entries.keys().cloned());
}
protected_features
}
pub(crate) fn apply_runtime_feature_enablement(
config: &mut Config,
runtime_feature_enablement: &BTreeMap<String, bool>,
) {
let protected_features = protected_feature_keys(&config.config_layer_stack);
for (name, enabled) in runtime_feature_enablement {
if protected_features.contains(name) {
continue;
}
let Some(feature) = feature_for_key(name) else {
continue;
};
if let Err(err) = config.features.set_enabled(feature, *enabled) {
warn!(
feature = name,
error = %err,
"failed to apply runtime feature enablement"
);
}
}
}
fn map_requirements_toml_to_api(requirements: ConfigRequirementsToml) -> ConfigRequirements {
ConfigRequirements {
allowed_approval_policies: requirements.allowed_approval_policies.map(|policies| {
@@ -495,7 +389,7 @@ fn map_network_unix_socket_permission_to_api(
}
}
fn map_error(err: ConfigServiceError) -> JSONRPCErrorError {
fn map_error(err: ConfigManagerError) -> JSONRPCErrorError {
if let Some(code) = err.write_error_code() {
return config_write_error(code, err.to_string());
}
@@ -520,7 +414,11 @@ fn config_write_error(code: ConfigWriteErrorCode, message: impl Into<String>) ->
#[cfg(test)]
mod tests {
use super::*;
use crate::config_manager::apply_runtime_feature_enablement;
use codex_analytics::AnalyticsEventsClient;
use codex_arg0::Arg0DispatchPaths;
use codex_core::config_loader::CloudRequirementsLoader;
use codex_core::config_loader::LoaderOverrides;
use codex_core::config_loader::NetworkDomainPermissionToml as CoreNetworkDomainPermissionToml;
use codex_core::config_loader::NetworkDomainPermissionsToml as CoreNetworkDomainPermissionsToml;
use codex_core::config_loader::NetworkRequirementsToml as CoreNetworkRequirementsToml;
@@ -533,9 +431,12 @@ mod tests {
use codex_protocol::protocol::AskForApproval as CoreAskForApproval;
use pretty_assertions::assert_eq;
use serde_json::json;
use std::collections::BTreeMap;
use std::sync::RwLock;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;
use tempfile::TempDir;
use toml::Value as TomlValue;
#[derive(Default)]
struct RecordingUserConfigReloader {
@@ -830,11 +731,15 @@ mod tests {
);
let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::from_api_key("test"));
let config_api = ConfigApi::new(
codex_home.path().to_path_buf(),
Arc::new(RwLock::new(Vec::new())),
Arc::new(RwLock::new(BTreeMap::new())),
LoaderOverrides::default(),
Arc::new(RwLock::new(CloudRequirementsLoader::default())),
ConfigManager::new(
codex_home.path().to_path_buf(),
Arc::new(RwLock::new(Vec::new())),
Arc::new(RwLock::new(BTreeMap::new())),
LoaderOverrides::default(),
Arc::new(RwLock::new(CloudRequirementsLoader::default())),
Arg0DispatchPaths::default(),
Arc::new(codex_config::NoopThreadConfigLoader),
),
reloader.clone(),
AnalyticsEventsClient::new(
auth_manager,
codex-rs/app-server/src/config_manager.rs
+326
View File
@@ -0,0 +1,326 @@
use codex_arg0::Arg0DispatchPaths;
use codex_cloud_requirements::cloud_requirements_loader;
use codex_config::ThreadConfigLoader;
use codex_core::config::Config;
use codex_core::config::ConfigOverrides;
use codex_core::config_loader::CloudRequirementsLoader;
use codex_core::config_loader::ConfigLayerStack;
use codex_core::config_loader::LoaderOverrides;
use codex_core::config_loader::load_config_layers_state;
use codex_exec_server::LOCAL_FS;
use codex_features::feature_for_key;
use codex_login::AuthManager;
use codex_login::default_client::set_default_client_residency_requirement;
use codex_utils_absolute_path::AbsolutePathBuf;
use codex_utils_json_to_toml::json_to_toml;
use std::collections::BTreeMap;
use std::collections::BTreeSet;
use std::collections::HashMap;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::RwLock;
use toml::Value as TomlValue;
use tracing::warn;
/// Shared app-server entry point for loading effective Codex configuration.
#[derive(Clone)]
pub(crate) struct ConfigManager {
codex_home: PathBuf,
cli_overrides: Arc<RwLock<Vec<(String, TomlValue)>>>,
runtime_feature_enablement: Arc<RwLock<BTreeMap<String, bool>>>,
loader_overrides: LoaderOverrides,
cloud_requirements: Arc<RwLock<CloudRequirementsLoader>>,
arg0_paths: Arg0DispatchPaths,
thread_config_loader: Arc<dyn ThreadConfigLoader>,
host_name: Option<String>,
}
impl ConfigManager {
pub(crate) fn new(
codex_home: PathBuf,
cli_overrides: Arc<RwLock<Vec<(String, TomlValue)>>>,
runtime_feature_enablement: Arc<RwLock<BTreeMap<String, bool>>>,
loader_overrides: LoaderOverrides,
cloud_requirements: Arc<RwLock<CloudRequirementsLoader>>,
arg0_paths: Arg0DispatchPaths,
thread_config_loader: Arc<dyn ThreadConfigLoader>,
) -> Self {
Self::new_with_host_name(
codex_home,
cli_overrides,
runtime_feature_enablement,
loader_overrides,
cloud_requirements,
arg0_paths,
thread_config_loader,
codex_config::host_name(),
)
}
#[allow(clippy::too_many_arguments)]
fn new_with_host_name(
codex_home: PathBuf,
cli_overrides: Arc<RwLock<Vec<(String, TomlValue)>>>,
runtime_feature_enablement: Arc<RwLock<BTreeMap<String, bool>>>,
loader_overrides: LoaderOverrides,
cloud_requirements: Arc<RwLock<CloudRequirementsLoader>>,
arg0_paths: Arg0DispatchPaths,
thread_config_loader: Arc<dyn ThreadConfigLoader>,
host_name: Option<String>,
) -> Self {
Self {
codex_home,
cli_overrides,
runtime_feature_enablement,
loader_overrides,
cloud_requirements,
arg0_paths,
thread_config_loader,
host_name,
}
}
pub(crate) fn codex_home(&self) -> &Path {
self.codex_home.as_path()
}
pub(crate) fn current_cli_overrides(&self) -> Vec<(String, TomlValue)> {
self.cli_overrides
.read()
.map(|guard| guard.clone())
.unwrap_or_default()
}
pub(crate) fn current_cloud_requirements(&self) -> CloudRequirementsLoader {
self.cloud_requirements
.read()
.map(|guard| guard.clone())
.unwrap_or_default()
}
pub(crate) fn extend_runtime_feature_enablement<I>(&self, enablement: I) -> Result<(), ()>
where
I: IntoIterator<Item = (String, bool)>,
{
let mut runtime_feature_enablement =
self.runtime_feature_enablement.write().map_err(|_| ())?;
runtime_feature_enablement.extend(enablement);
Ok(())
}
pub(crate) fn replace_cloud_requirements_loader(
&self,
auth_manager: Arc<AuthManager>,
chatgpt_base_url: String,
) {
let loader =
cloud_requirements_loader(auth_manager, chatgpt_base_url, self.codex_home.clone());
if let Ok(mut guard) = self.cloud_requirements.write() {
*guard = loader;
} else {
warn!("failed to update cloud requirements loader");
}
}
pub(crate) async fn sync_default_client_residency_requirement(&self) {
match self.load_latest_config(/*fallback_cwd*/ None).await {
Ok(config) => {
set_default_client_residency_requirement(config.enforce_residency.value());
}
Err(err) => warn!(
error = %err,
"failed to sync default client residency requirement after auth refresh"
),
}
}
pub(crate) async fn load_latest_config(
&self,
fallback_cwd: Option<PathBuf>,
) -> std::io::Result<Config> {
self.load_with_cli_overrides(
&self.current_cli_overrides(),
/*request_overrides*/ None,
ConfigOverrides::default(),
fallback_cwd,
)
.await
}
pub(crate) async fn load_with_overrides(
&self,
request_overrides: Option<HashMap<String, serde_json::Value>>,
typesafe_overrides: ConfigOverrides,
) -> std::io::Result<Config> {
self.load_with_cli_overrides(
&self.current_cli_overrides(),
request_overrides,
typesafe_overrides,
/*fallback_cwd*/ None,
)
.await
}
pub(crate) async fn load_for_cwd(
&self,
request_overrides: Option<HashMap<String, serde_json::Value>>,
typesafe_overrides: ConfigOverrides,
cwd: Option<PathBuf>,
) -> std::io::Result<Config> {
self.load_with_cli_overrides(
&self.current_cli_overrides(),
request_overrides,
typesafe_overrides,
cwd,
)
.await
}
pub(crate) async fn load_with_cli_overrides(
&self,
cli_overrides: &[(String, TomlValue)],
request_overrides: Option<HashMap<String, serde_json::Value>>,
typesafe_overrides: ConfigOverrides,
fallback_cwd: Option<PathBuf>,
) -> std::io::Result<Config> {
let merged_cli_overrides = cli_overrides
.iter()
.cloned()
.chain(
request_overrides
.unwrap_or_default()
.into_iter()
.map(|(key, value)| (key, json_to_toml(value))),
)
.collect::<Vec<_>>();
let mut config = codex_core::config::ConfigBuilder::default()
.codex_home(self.codex_home.clone())
.cli_overrides(merged_cli_overrides)
.loader_overrides(self.loader_overrides.clone())
.harness_overrides(typesafe_overrides)
.fallback_cwd(fallback_cwd)
.cloud_requirements(self.current_cloud_requirements())
.thread_config_loader(Arc::clone(&self.thread_config_loader))
.host_name(self.host_name.clone())
.build()
.await?;
self.apply_runtime_feature_enablement(&mut config);
self.apply_arg0_paths(&mut config);
Ok(config)
}
pub(crate) async fn load_config_layers_for_cwd(
&self,
cwd: AbsolutePathBuf,
) -> std::io::Result<ConfigLayerStack> {
self.load_config_layers(Some(cwd)).await
}
pub(crate) async fn load_config_layers(
&self,
cwd: Option<AbsolutePathBuf>,
) -> std::io::Result<ConfigLayerStack> {
load_config_layers_state(
LOCAL_FS.as_ref(),
&self.codex_home,
cwd,
&self.current_cli_overrides(),
self.loader_overrides.clone(),
self.current_cloud_requirements(),
self.thread_config_loader.as_ref(),
self.host_name.as_deref(),
)
.await
}
fn apply_runtime_feature_enablement(&self, config: &mut Config) {
apply_runtime_feature_enablement(config, &self.current_runtime_feature_enablement());
}
fn current_runtime_feature_enablement(&self) -> BTreeMap<String, bool> {
self.runtime_feature_enablement
.read()
.map(|guard| guard.clone())
.unwrap_or_default()
}
fn apply_arg0_paths(&self, config: &mut Config) {
config.codex_self_exe = self.arg0_paths.codex_self_exe.clone();
config.codex_linux_sandbox_exe = self.arg0_paths.codex_linux_sandbox_exe.clone();
config.main_execve_wrapper_exe = self.arg0_paths.main_execve_wrapper_exe.clone();
}
#[cfg(test)]
pub(crate) fn new_for_tests(
codex_home: PathBuf,
cli_overrides: Vec<(String, TomlValue)>,
loader_overrides: LoaderOverrides,
cloud_requirements: CloudRequirementsLoader,
host_name: Option<String>,
) -> Self {
Self::new_with_host_name(
codex_home,
Arc::new(RwLock::new(cli_overrides)),
Arc::new(RwLock::new(BTreeMap::new())),
loader_overrides,
Arc::new(RwLock::new(cloud_requirements)),
Arg0DispatchPaths::default(),
Arc::new(codex_config::NoopThreadConfigLoader),
host_name,
)
}
#[cfg(test)]
pub(crate) fn without_managed_config_for_tests(codex_home: PathBuf) -> Self {
Self::new_for_tests(
codex_home,
Vec::new(),
LoaderOverrides::without_managed_config_for_tests(),
CloudRequirementsLoader::default(),
/*host_name*/ None,
)
}
}
pub(crate) fn protected_feature_keys(config_layer_stack: &ConfigLayerStack) -> BTreeSet<String> {
let mut protected_features = config_layer_stack
.effective_config()
.get("features")
.and_then(toml::Value::as_table)
.map(|features| features.keys().cloned().collect::<BTreeSet<_>>())
.unwrap_or_default();
if let Some(feature_requirements) = config_layer_stack
.requirements_toml()
.feature_requirements
.as_ref()
{
protected_features.extend(feature_requirements.entries.keys().cloned());
}
protected_features
}
pub(crate) fn apply_runtime_feature_enablement(
config: &mut Config,
runtime_feature_enablement: &BTreeMap<String, bool>,
) {
let protected_features = protected_feature_keys(&config.config_layer_stack);
for (name, enabled) in runtime_feature_enablement {
if protected_features.contains(name) {
continue;
}
let Some(feature) = feature_for_key(name) else {
continue;
};
if let Err(err) = config.features.set_enabled(feature, *enabled) {
warn!(
feature = name,
error = %err,
"failed to apply runtime feature enablement"
);
}
}
}
codex-rs/app-server/src/config_manager_service.rs
+653
View File
@@ -0,0 +1,653 @@
use crate::config_manager::ConfigManager;
use codex_app_server_protocol::Config as ApiConfig;
use codex_app_server_protocol::ConfigBatchWriteParams;
use codex_app_server_protocol::ConfigLayerMetadata;
use codex_app_server_protocol::ConfigLayerSource;
use codex_app_server_protocol::ConfigReadParams;
use codex_app_server_protocol::ConfigReadResponse;
use codex_app_server_protocol::ConfigValueWriteParams;
use codex_app_server_protocol::ConfigWriteErrorCode;
use codex_app_server_protocol::ConfigWriteResponse;
use codex_app_server_protocol::MergeStrategy;
use codex_app_server_protocol::OverriddenMetadata;
use codex_app_server_protocol::WriteStatus;
use codex_config::CONFIG_TOML_FILE;
use codex_config::config_toml::ConfigToml;
use codex_core::config::deserialize_config_toml_with_base;
use codex_core::config::edit::ConfigEdit;
use codex_core::config::edit::ConfigEditsBuilder;
use codex_core::config::validate_feature_requirements_for_config_toml;
use codex_core::config_loader::ConfigLayerEntry;
use codex_core::config_loader::ConfigLayerStack;
use codex_core::config_loader::ConfigLayerStackOrdering;
use codex_core::config_loader::ConfigRequirementsToml;
use codex_core::config_loader::merge_toml_values;
use codex_core::path_utils;
use codex_core::path_utils::SymlinkWritePaths;
use codex_core::path_utils::resolve_symlink_write_paths;
use codex_core::path_utils::write_atomically;
use codex_utils_absolute_path::AbsolutePathBuf;
use serde_json::Value as JsonValue;
use std::borrow::Cow;
use std::path::Path;
use std::path::PathBuf;
use thiserror::Error;
use tokio::task;
use toml::Value as TomlValue;
use toml_edit::Item as TomlItem;
#[derive(Debug, Error)]
pub(crate) enum ConfigManagerError {
#[error("{message}")]
Write {
code: ConfigWriteErrorCode,
message: String,
},
#[error("{context}: {source}")]
Io {
context: &'static str,
#[source]
source: std::io::Error,
},
#[error("{context}: {source}")]
Json {
context: &'static str,
#[source]
source: serde_json::Error,
},
#[error("{context}: {source}")]
Toml {
context: &'static str,
#[source]
source: toml::de::Error,
},
#[error("{context}: {source}")]
Anyhow {
context: &'static str,
#[source]
source: anyhow::Error,
},
}
impl ConfigManagerError {
fn write(code: ConfigWriteErrorCode, message: impl Into<String>) -> Self {
Self::Write {
code,
message: message.into(),
}
}
fn io(context: &'static str, source: std::io::Error) -> Self {
Self::Io { context, source }
}
fn json(context: &'static str, source: serde_json::Error) -> Self {
Self::Json { context, source }
}
fn toml(context: &'static str, source: toml::de::Error) -> Self {
Self::Toml { context, source }
}
fn anyhow(context: &'static str, source: anyhow::Error) -> Self {
Self::Anyhow { context, source }
}
pub(crate) fn write_error_code(&self) -> Option<ConfigWriteErrorCode> {
match self {
Self::Write { code, .. } => Some(code.clone()),
_ => None,
}
}
}
impl ConfigManager {
pub(crate) async fn read(
&self,
params: ConfigReadParams,
) -> Result<ConfigReadResponse, ConfigManagerError> {
let layers = match params.cwd.as_deref() {
Some(cwd) => {
let cwd = AbsolutePathBuf::try_from(PathBuf::from(cwd)).map_err(|err| {
ConfigManagerError::io("failed to resolve config cwd to an absolute path", err)
})?;
self.load_config_layers(Some(cwd)).await.map_err(|err| {
ConfigManagerError::io("failed to read configuration layers", err)
})?
}
None => self.load_thread_agnostic_config().await.map_err(|err| {
ConfigManagerError::io("failed to read configuration layers", err)
})?,
};
let effective = layers.effective_config();
let effective_config_toml: ConfigToml = effective
.try_into()
.map_err(|err| ConfigManagerError::toml("invalid configuration", err))?;
let json_value = serde_json::to_value(&effective_config_toml)
.map_err(|err| ConfigManagerError::json("failed to serialize configuration", err))?;
let config: ApiConfig = serde_json::from_value(json_value)
.map_err(|err| ConfigManagerError::json("failed to deserialize configuration", err))?;
Ok(ConfigReadResponse {
config,
origins: layers.origins(),
layers: params.include_layers.then(|| {
layers
.get_layers(
ConfigLayerStackOrdering::HighestPrecedenceFirst,
/*include_disabled*/ true,
)
.iter()
.map(|layer| layer.as_layer())
.collect()
}),
})
}
pub(crate) async fn read_requirements(
&self,
) -> Result<Option<ConfigRequirementsToml>, ConfigManagerError> {
let layers = self
.load_thread_agnostic_config()
.await
.map_err(|err| ConfigManagerError::io("failed to read configuration layers", err))?;
let requirements = layers.requirements_toml().clone();
if requirements.is_empty() {
Ok(None)
} else {
Ok(Some(requirements))
}
}
pub(crate) async fn write_value(
&self,
params: ConfigValueWriteParams,
) -> Result<ConfigWriteResponse, ConfigManagerError> {
let edits = vec![(params.key_path, params.value, params.merge_strategy)];
self.apply_edits(params.file_path, params.expected_version, edits)
.await
}
pub(crate) async fn batch_write(
&self,
params: ConfigBatchWriteParams,
) -> Result<ConfigWriteResponse, ConfigManagerError> {
let edits = params
.edits
.into_iter()
.map(|edit| (edit.key_path, edit.value, edit.merge_strategy))
.collect();
self.apply_edits(params.file_path, params.expected_version, edits)
.await
}
async fn apply_edits(
&self,
file_path: Option<String>,
expected_version: Option<String>,
edits: Vec<(String, JsonValue, MergeStrategy)>,
) -> Result<ConfigWriteResponse, ConfigManagerError> {
let allowed_path =
AbsolutePathBuf::resolve_path_against_base(CONFIG_TOML_FILE, self.codex_home());
let provided_path = match file_path {
Some(path) => AbsolutePathBuf::from_absolute_path(PathBuf::from(path))
.map_err(|err| ConfigManagerError::io("failed to resolve user config path", err))?,
None => allowed_path.clone(),
};
if !paths_match(&allowed_path, &provided_path) {
return Err(ConfigManagerError::write(
ConfigWriteErrorCode::ConfigLayerReadonly,
"Only writes to the user config are allowed",
));
}
let layers = self
.load_thread_agnostic_config()
.await
.map_err(|err| ConfigManagerError::io("failed to load configuration", err))?;
let user_layer = match layers.get_user_layer() {
Some(layer) => Cow::Borrowed(layer),
None => Cow::Owned(create_empty_user_layer(&allowed_path).await?),
};
if let Some(expected) = expected_version.as_deref()
&& expected != user_layer.version
{
return Err(ConfigManagerError::write(
ConfigWriteErrorCode::ConfigVersionConflict,
"Configuration was modified since last read. Fetch latest version and retry.",
));
}
let mut user_config = user_layer.config.clone();
let mut parsed_segments = Vec::new();
let mut config_edits = Vec::new();
for (key_path, value, strategy) in edits.into_iter() {
let segments = parse_key_path(&key_path).map_err(|message| {
ConfigManagerError::write(ConfigWriteErrorCode::ConfigValidationError, message)
})?;
let original_value = value_at_path(&user_config, &segments).cloned();
let parsed_value = parse_value(value).map_err(|message| {
ConfigManagerError::write(ConfigWriteErrorCode::ConfigValidationError, message)
})?;
apply_merge(&mut user_config, &segments, parsed_value.as_ref(), strategy).map_err(
|err| match err {
MergeError::PathNotFound => ConfigManagerError::write(
ConfigWriteErrorCode::ConfigPathNotFound,
"Path not found",
),
MergeError::Validation(message) => ConfigManagerError::write(
ConfigWriteErrorCode::ConfigValidationError,
message,
),
},
)?;
let updated_value = value_at_path(&user_config, &segments).cloned();
if original_value != updated_value {
let edit = match updated_value {
Some(value) => ConfigEdit::SetPath {
segments: segments.clone(),
value: toml_value_to_item(&value).map_err(|err| {
ConfigManagerError::anyhow("failed to build config edits", err)
})?,
},
None => ConfigEdit::ClearPath {
segments: segments.clone(),
},
};
config_edits.push(edit);
}
parsed_segments.push(segments);
}
validate_config(&user_config).map_err(|err| {
ConfigManagerError::write(
ConfigWriteErrorCode::ConfigValidationError,
format!("Invalid configuration: {err}"),
)
})?;
let user_config_toml =
deserialize_config_toml_with_base(user_config.clone(), self.codex_home()).map_err(
|err| {
ConfigManagerError::write(
ConfigWriteErrorCode::ConfigValidationError,
format!("Invalid configuration: {err}"),
)
},
)?;
validate_feature_requirements_for_config_toml(
&user_config_toml,
layers.requirements().feature_requirements.as_ref(),
)
.map_err(|err| {
ConfigManagerError::write(
ConfigWriteErrorCode::ConfigValidationError,
format!("Invalid configuration: {err}"),
)
})?;
let updated_layers = layers.with_user_config(&provided_path, user_config.clone());
let effective = updated_layers.effective_config();
validate_config(&effective).map_err(|err| {
ConfigManagerError::write(
ConfigWriteErrorCode::ConfigValidationError,
format!("Invalid configuration: {err}"),
)
})?;
if !config_edits.is_empty() {
ConfigEditsBuilder::new(self.codex_home())
.with_edits(config_edits)
.apply()
.await
.map_err(|err| ConfigManagerError::anyhow("failed to persist config.toml", err))?;
}
let overridden = first_overridden_edit(&updated_layers, &effective, &parsed_segments);
let status = overridden
.as_ref()
.map(|_| WriteStatus::OkOverridden)
.unwrap_or(WriteStatus::Ok);
Ok(ConfigWriteResponse {
status,
version: updated_layers
.get_user_layer()
.ok_or_else(|| {
ConfigManagerError::write(
ConfigWriteErrorCode::UserLayerNotFound,
"user layer not found in updated layers",
)
})?
.version
.clone(),
file_path: provided_path,
overridden_metadata: overridden,
})
}
/// Loads a "thread-agnostic" config, which means the config layers do not
/// include any in-repo .codex/ folders because there is no cwd/project root
/// associated with this query.
async fn load_thread_agnostic_config(&self) -> std::io::Result<ConfigLayerStack> {
self.load_config_layers(/*cwd*/ None).await
}
}
async fn create_empty_user_layer(
config_toml: &AbsolutePathBuf,
) -> Result<ConfigLayerEntry, ConfigManagerError> {
let SymlinkWritePaths {
read_path,
write_path,
} = resolve_symlink_write_paths(config_toml.as_path())
.map_err(|err| ConfigManagerError::io("failed to resolve user config path", err))?;
let toml_value = match read_path {
Some(path) => match tokio::fs::read_to_string(&path).await {
Ok(contents) => toml::from_str(&contents).map_err(|e| {
ConfigManagerError::toml("failed to parse existing user config.toml", e)
})?,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
write_empty_user_config(write_path.clone()).await?;
TomlValue::Table(toml::map::Map::new())
}
Err(err) => {
return Err(ConfigManagerError::io(
"failed to read user config.toml",
err,
));
}
},
None => {
write_empty_user_config(write_path).await?;
TomlValue::Table(toml::map::Map::new())
}
};
Ok(ConfigLayerEntry::new(
ConfigLayerSource::User {
file: config_toml.clone(),
},
toml_value,
))
}
async fn write_empty_user_config(write_path: PathBuf) -> Result<(), ConfigManagerError> {
task::spawn_blocking(move || write_atomically(&write_path, ""))
.await
.map_err(|err| ConfigManagerError::anyhow("config persistence task panicked", err.into()))?
.map_err(|err| ConfigManagerError::io("failed to create empty user config.toml", err))
}
fn parse_value(value: JsonValue) -> Result<Option<TomlValue>, String> {
if value.is_null() {
return Ok(None);
}
serde_json::from_value::<TomlValue>(value)
.map(Some)
.map_err(|err| format!("invalid value: {err}"))
}
fn parse_key_path(path: &str) -> Result<Vec<String>, String> {
if path.trim().is_empty() {
return Err("keyPath must not be empty".to_string());
}
Ok(path
.split('.')
.map(std::string::ToString::to_string)
.collect())
}
#[derive(Debug)]
enum MergeError {
PathNotFound,
Validation(String),
}
fn apply_merge(
root: &mut TomlValue,
segments: &[String],
value: Option<&TomlValue>,
strategy: MergeStrategy,
) -> Result<bool, MergeError> {
let Some(value) = value else {
return clear_path(root, segments);
};
let Some((last, parents)) = segments.split_last() else {
return Err(MergeError::Validation(
"keyPath must not be empty".to_string(),
));
};
let mut current = root;
for segment in parents {
match current {
TomlValue::Table(table) => {
current = table
.entry(segment.clone())
.or_insert_with(|| TomlValue::Table(toml::map::Map::new()));
}
_ => {
*current = TomlValue::Table(toml::map::Map::new());
if let TomlValue::Table(table) = current {
current = table
.entry(segment.clone())
.or_insert_with(|| TomlValue::Table(toml::map::Map::new()));
}
}
}
}
let table = current.as_table_mut().ok_or_else(|| {
MergeError::Validation("cannot set value on non-table parent".to_string())
})?;
if matches!(strategy, MergeStrategy::Upsert)
&& let Some(existing) = table.get_mut(last)
&& matches!(existing, TomlValue::Table(_))
&& matches!(value, TomlValue::Table(_))
{
merge_toml_values(existing, value);
return Ok(true);
}
let changed = table
.get(last)
.map(|existing| Some(existing) != Some(value))
.unwrap_or(true);
table.insert(last.clone(), value.clone());
Ok(changed)
}
fn clear_path(root: &mut TomlValue, segments: &[String]) -> Result<bool, MergeError> {
let Some((last, parents)) = segments.split_last() else {
return Err(MergeError::Validation(
"keyPath must not be empty".to_string(),
));
};
let mut current = root;
for segment in parents {
match current {
TomlValue::Table(table) => {
current = table.get_mut(segment).ok_or(MergeError::PathNotFound)?;
}
_ => return Err(MergeError::PathNotFound),
}
}
let Some(parent) = current.as_table_mut() else {
return Err(MergeError::PathNotFound);
};
Ok(parent.remove(last).is_some())
}
fn toml_value_to_item(value: &TomlValue) -> anyhow::Result<TomlItem> {
match value {
TomlValue::Table(table) => {
let mut table_item = toml_edit::Table::new();
table_item.set_implicit(false);
for (key, val) in table {
table_item.insert(key, toml_value_to_item(val)?);
}
Ok(TomlItem::Table(table_item))
}
other => Ok(TomlItem::Value(toml_value_to_value(other)?)),
}
}
fn toml_value_to_value(value: &TomlValue) -> anyhow::Result<toml_edit::Value> {
match value {
TomlValue::String(val) => Ok(toml_edit::Value::from(val.clone())),
TomlValue::Integer(val) => Ok(toml_edit::Value::from(*val)),
TomlValue::Float(val) => Ok(toml_edit::Value::from(*val)),
TomlValue::Boolean(val) => Ok(toml_edit::Value::from(*val)),
TomlValue::Datetime(val) => Ok(toml_edit::Value::from(*val)),
TomlValue::Array(items) => {
let mut array = toml_edit::Array::new();
for item in items {
array.push(toml_value_to_value(item)?);
}
Ok(toml_edit::Value::Array(array))
}
TomlValue::Table(table) => {
let mut inline = toml_edit::InlineTable::new();
for (key, val) in table {
inline.insert(key, toml_value_to_value(val)?);
}
Ok(toml_edit::Value::InlineTable(inline))
}
}
}
fn validate_config(value: &TomlValue) -> Result<(), toml::de::Error> {
let _: ConfigToml = value.clone().try_into()?;
Ok(())
}
fn paths_match(expected: impl AsRef<Path>, provided: impl AsRef<Path>) -> bool {
path_utils::paths_match_after_normalization(expected, provided)
}
fn value_at_path<'a>(root: &'a TomlValue, segments: &[String]) -> Option<&'a TomlValue> {
let mut current = root;
for segment in segments {
match current {
TomlValue::Table(table) => {
current = table.get(segment)?;
}
TomlValue::Array(items) => {
let idx = segment.parse::<i64>().ok()?;
let idx = usize::try_from(idx).ok()?;
current = items.get(idx)?;
}
_ => return None,
}
}
Some(current)
}
fn override_message(layer: &ConfigLayerSource) -> String {
match layer {
ConfigLayerSource::Mdm { domain, key: _ } => {
format!("Overridden by managed policy (MDM): {domain}")
}
ConfigLayerSource::System { file } => {
format!("Overridden by managed config (system): {}", file.display())
}
ConfigLayerSource::Project { dot_codex_folder } => format!(
"Overridden by project config: {}/{CONFIG_TOML_FILE}",
dot_codex_folder.display(),
),
ConfigLayerSource::SessionFlags => "Overridden by session flags".to_string(),
ConfigLayerSource::User { file } => {
format!("Overridden by user config: {}", file.display())
}
ConfigLayerSource::LegacyManagedConfigTomlFromFile { file } => {
format!(
"Overridden by legacy managed_config.toml: {}",
file.display()
)
}
ConfigLayerSource::LegacyManagedConfigTomlFromMdm => {
"Overridden by legacy managed configuration from MDM".to_string()
}
}
}
fn compute_override_metadata(
layers: &ConfigLayerStack,
effective: &TomlValue,
segments: &[String],
) -> Option<OverriddenMetadata> {
let user_value = match layers.get_user_layer() {
Some(user_layer) => value_at_path(&user_layer.config, segments),
None => return None,
};
let effective_value = value_at_path(effective, segments);
if user_value.is_some() && user_value == effective_value {
return None;
}
if user_value.is_none() && effective_value.is_none() {
return None;
}
let overriding_layer = find_effective_layer(layers, segments)?;
let message = override_message(&overriding_layer.name);
Some(OverriddenMetadata {
message,
overriding_layer,
effective_value: effective_value
.and_then(|value| serde_json::to_value(value).ok())
.unwrap_or(JsonValue::Null),
})
}
fn first_overridden_edit(
layers: &ConfigLayerStack,
effective: &TomlValue,
edits: &[Vec<String>],
) -> Option<OverriddenMetadata> {
for segments in edits {
if let Some(meta) = compute_override_metadata(layers, effective, segments) {
return Some(meta);
}
}
None
}
fn find_effective_layer(
layers: &ConfigLayerStack,
segments: &[String],
) -> Option<ConfigLayerMetadata> {
for layer in layers.layers_high_to_low() {
if let Some(meta) = value_at_path(&layer.config, segments).map(|_| layer.metadata()) {
return Some(meta);
}
}
None
}
#[cfg(test)]
#[path = "config_manager_service_tests.rs"]
mod tests;
codex-rs/app-server/src/config_manager_service_tests.rs
+777
View File
@@ -0,0 +1,777 @@
use super::*;
use anyhow::Result;
use codex_app_server_protocol::AppConfig;
use codex_app_server_protocol::AppToolApproval;
use codex_app_server_protocol::AppsConfig;
use codex_app_server_protocol::AskForApproval;
use codex_core::config_loader::CloudRequirementsLoader;
use codex_core::config_loader::FeatureRequirementsToml;
use codex_core::config_loader::LoaderOverrides;
use codex_utils_absolute_path::AbsolutePathBuf;
use pretty_assertions::assert_eq;
use std::collections::BTreeMap;
use tempfile::tempdir;
#[test]
fn toml_value_to_item_handles_nested_config_tables() {
let config = r#"
[mcp_servers.docs]
command = "docs-server"
[mcp_servers.docs.http_headers]
X-Doc = "42"
"#;
let value: TomlValue = toml::from_str(config).expect("parse config example");
let item = toml_value_to_item(&value).expect("convert to toml_edit item");
let root = item.as_table().expect("root table");
assert!(!root.is_implicit(), "root table should be explicit");
let mcp_servers = root
.get("mcp_servers")
.and_then(TomlItem::as_table)
.expect("mcp_servers table");
assert!(
!mcp_servers.is_implicit(),
"mcp_servers table should be explicit"
);
let docs = mcp_servers
.get("docs")
.and_then(TomlItem::as_table)
.expect("docs table");
assert_eq!(
docs.get("command")
.and_then(TomlItem::as_value)
.and_then(toml_edit::Value::as_str),
Some("docs-server")
);
let http_headers = docs
.get("http_headers")
.and_then(TomlItem::as_table)
.expect("http_headers table");
assert_eq!(
http_headers
.get("X-Doc")
.and_then(TomlItem::as_value)
.and_then(toml_edit::Value::as_str),
Some("42")
);
}
#[tokio::test]
async fn write_value_preserves_comments_and_order() -> Result<()> {
let tmp = tempdir().expect("tempdir");
let original = r#"# Codex user configuration
model = "gpt-5.2"
approval_policy = "on-request"
[notice]
# Preserve this comment
hide_full_access_warning = true
[features]
unified_exec = true
"#;
std::fs::write(tmp.path().join(CONFIG_TOML_FILE), original)?;
let service = ConfigManager::without_managed_config_for_tests(tmp.path().to_path_buf());
service
.write_value(ConfigValueWriteParams {
file_path: Some(tmp.path().join(CONFIG_TOML_FILE).display().to_string()),
key_path: "features.personality".to_string(),
value: serde_json::json!(true),
merge_strategy: MergeStrategy::Replace,
expected_version: None,
})
.await
.expect("write succeeds");
let updated = std::fs::read_to_string(tmp.path().join(CONFIG_TOML_FILE)).expect("read config");
let expected = r#"# Codex user configuration
model = "gpt-5.2"
approval_policy = "on-request"
[notice]
# Preserve this comment
hide_full_access_warning = true
[features]
unified_exec = true
personality = true
"#;
assert_eq!(updated, expected);
Ok(())
}
#[tokio::test]
async fn write_value_supports_nested_app_paths() -> Result<()> {
let tmp = tempdir().expect("tempdir");
std::fs::write(tmp.path().join(CONFIG_TOML_FILE), "")?;
let service = ConfigManager::without_managed_config_for_tests(tmp.path().to_path_buf());
service
.write_value(ConfigValueWriteParams {
file_path: Some(tmp.path().join(CONFIG_TOML_FILE).display().to_string()),
key_path: "apps".to_string(),
value: serde_json::json!({
"app1": {
"enabled": false,
},
}),
merge_strategy: MergeStrategy::Replace,
expected_version: None,
})
.await
.expect("write apps succeeds");
service
.write_value(ConfigValueWriteParams {
file_path: Some(tmp.path().join(CONFIG_TOML_FILE).display().to_string()),
key_path: "apps.app1.default_tools_approval_mode".to_string(),
value: serde_json::json!("prompt"),
merge_strategy: MergeStrategy::Replace,
expected_version: None,
})
.await
.expect("write apps.app1.default_tools_approval_mode succeeds");
let read = service
.read(ConfigReadParams {
include_layers: false,
cwd: None,
})
.await
.expect("config read succeeds");
assert_eq!(
read.config.apps,
Some(AppsConfig {
default: None,
apps: std::collections::HashMap::from([(
"app1".to_string(),
AppConfig {
enabled: false,
destructive_enabled: None,
open_world_enabled: None,
default_tools_approval_mode: Some(AppToolApproval::Prompt),
default_tools_enabled: None,
tools: None,
},
)]),
})
);
Ok(())
}
#[tokio::test]
async fn write_value_supports_custom_mcp_server_default_tool_approval_mode() -> Result<()> {
let tmp = tempdir().expect("tempdir");
std::fs::write(
tmp.path().join(CONFIG_TOML_FILE),
"[mcp_servers.docs]\ncommand = \"docs-server\"\n",
)?;
let service = ConfigManager::without_managed_config_for_tests(tmp.path().to_path_buf());
service
.write_value(ConfigValueWriteParams {
file_path: Some(tmp.path().join(CONFIG_TOML_FILE).display().to_string()),
key_path: "mcp_servers.docs.default_tools_approval_mode".to_string(),
value: serde_json::json!("approve"),
merge_strategy: MergeStrategy::Replace,
expected_version: None,
})
.await
.expect("write mcp server default_tools_approval_mode succeeds");
let contents = std::fs::read_to_string(tmp.path().join(CONFIG_TOML_FILE))?;
assert!(contents.contains("default_tools_approval_mode = \"approve\""));
let read = service
.read(ConfigReadParams {
include_layers: false,
cwd: None,
})
.await
.expect("config read succeeds");
assert_eq!(
read.config
.additional
.get("mcp_servers")
.and_then(|servers| servers.get("docs"))
.and_then(|docs| docs.get("default_tools_approval_mode")),
Some(&serde_json::json!("approve"))
);
Ok(())
}
#[tokio::test]
async fn read_includes_origins_and_layers() {
let tmp = tempdir().expect("tempdir");
let user_path = tmp.path().join(CONFIG_TOML_FILE);
std::fs::write(&user_path, "model = \"user\"").unwrap();
let user_file = AbsolutePathBuf::try_from(user_path.clone()).expect("user file");
let managed_path = tmp.path().join("managed_config.toml");
std::fs::write(&managed_path, "approval_policy = \"never\"").unwrap();
let managed_file = AbsolutePathBuf::try_from(managed_path.clone()).expect("managed file");
let service = ConfigManager::new_for_tests(
tmp.path().to_path_buf(),
vec![],
LoaderOverrides::with_managed_config_path_for_tests(managed_path.clone()),
CloudRequirementsLoader::default(),
/*host_name*/ None,
);
let response = service
.read(ConfigReadParams {
include_layers: true,
cwd: None,
})
.await
.expect("response");
assert_eq!(response.config.approval_policy, Some(AskForApproval::Never));
assert_eq!(
response
.origins
.get("approval_policy")
.expect("origin")
.name,
ConfigLayerSource::LegacyManagedConfigTomlFromFile {
file: managed_file.clone()
},
);
let layers = response.layers.expect("layers present");
// Local macOS machines can surface an MDM-managed config layer at the
// top of the stack; ignore it so this test stays focused on file/user/system ordering.
let layers = if matches!(
layers.first().map(|layer| &layer.name),
Some(ConfigLayerSource::LegacyManagedConfigTomlFromMdm)
) {
&layers[1..]
} else {
layers.as_slice()
};
assert_eq!(layers.len(), 3, "expected three layers");
assert_eq!(
layers.first().unwrap().name,
ConfigLayerSource::LegacyManagedConfigTomlFromFile {
file: managed_file.clone()
}
);
assert_eq!(
layers.get(1).unwrap().name,
ConfigLayerSource::User {
file: user_file.clone()
}
);
assert!(matches!(
layers.get(2).unwrap().name,
ConfigLayerSource::System { .. }
));
}
#[cfg(target_os = "macos")]
#[tokio::test]
async fn write_value_succeeds_when_managed_preferences_expand_home_directory_paths() -> Result<()> {
use base64::Engine;
let tmp = tempdir().expect("tempdir");
std::fs::write(tmp.path().join(CONFIG_TOML_FILE), "model = \"user\"\n")?;
let mut loader_overrides =
LoaderOverrides::with_managed_config_path_for_tests(tmp.path().join("managed_config.toml"));
loader_overrides.managed_preferences_base64 = Some(
base64::prelude::BASE64_STANDARD.encode(
r#"
sandbox_mode = "workspace-write"
[sandbox_workspace_write]
writable_roots = ["~/code"]
"#
.as_bytes(),
),
);
let service = ConfigManager::new_for_tests(
tmp.path().to_path_buf(),
vec![],
loader_overrides,
CloudRequirementsLoader::default(),
/*host_name*/ None,
);
let response = service
.write_value(ConfigValueWriteParams {
file_path: Some(tmp.path().join(CONFIG_TOML_FILE).display().to_string()),
key_path: "model".to_string(),
value: serde_json::json!("updated"),
merge_strategy: MergeStrategy::Replace,
expected_version: None,
})
.await
.expect("write succeeds");
assert_eq!(response.status, WriteStatus::Ok);
assert_eq!(
std::fs::read_to_string(tmp.path().join(CONFIG_TOML_FILE)).expect("read config"),
"model = \"updated\"\n"
);
Ok(())
}
#[tokio::test]
async fn write_value_reports_override() {
let tmp = tempdir().expect("tempdir");
std::fs::write(
tmp.path().join(CONFIG_TOML_FILE),
"approval_policy = \"on-request\"",
)
.unwrap();
let managed_path = tmp.path().join("managed_config.toml");
std::fs::write(&managed_path, "approval_policy = \"never\"").unwrap();
let managed_file = AbsolutePathBuf::try_from(managed_path.clone()).expect("managed file");
let service = ConfigManager::new_for_tests(
tmp.path().to_path_buf(),
vec![],
LoaderOverrides::with_managed_config_path_for_tests(managed_path.clone()),
CloudRequirementsLoader::default(),
/*host_name*/ None,
);
let result = service
.write_value(ConfigValueWriteParams {
file_path: Some(tmp.path().join(CONFIG_TOML_FILE).display().to_string()),
key_path: "approval_policy".to_string(),
value: serde_json::json!("never"),
merge_strategy: MergeStrategy::Replace,
expected_version: None,
})
.await
.expect("result");
let read_after = service
.read(ConfigReadParams {
include_layers: true,
cwd: None,
})
.await
.expect("read");
assert_eq!(
read_after.config.approval_policy,
Some(AskForApproval::Never)
);
assert_eq!(
read_after
.origins
.get("approval_policy")
.expect("origin")
.name,
ConfigLayerSource::LegacyManagedConfigTomlFromFile {
file: managed_file.clone()
}
);
assert_eq!(result.status, WriteStatus::Ok);
assert!(result.overridden_metadata.is_none());
}
#[tokio::test]
async fn version_conflict_rejected() {
let tmp = tempdir().expect("tempdir");
let user_path = tmp.path().join(CONFIG_TOML_FILE);
std::fs::write(&user_path, "model = \"user\"").unwrap();
let service = ConfigManager::without_managed_config_for_tests(tmp.path().to_path_buf());
let error = service
.write_value(ConfigValueWriteParams {
file_path: Some(tmp.path().join(CONFIG_TOML_FILE).display().to_string()),
key_path: "model".to_string(),
value: serde_json::json!("gpt-5.2"),
merge_strategy: MergeStrategy::Replace,
expected_version: Some("sha256:bogus".to_string()),
})
.await
.expect_err("should fail");
assert_eq!(
error.write_error_code(),
Some(ConfigWriteErrorCode::ConfigVersionConflict)
);
}
#[tokio::test]
async fn write_value_defaults_to_user_config_path() {
let tmp = tempdir().expect("tempdir");
std::fs::write(tmp.path().join(CONFIG_TOML_FILE), "").unwrap();
let service = ConfigManager::without_managed_config_for_tests(tmp.path().to_path_buf());
service
.write_value(ConfigValueWriteParams {
file_path: None,
key_path: "model".to_string(),
value: serde_json::json!("gpt-new"),
merge_strategy: MergeStrategy::Replace,
expected_version: None,
})
.await
.expect("write succeeds");
let contents = std::fs::read_to_string(tmp.path().join(CONFIG_TOML_FILE)).expect("read config");
assert!(
contents.contains("model = \"gpt-new\""),
"config.toml should be updated even when file_path is omitted"
);
}
#[tokio::test]
async fn invalid_user_value_rejected_even_if_overridden_by_managed() {
let tmp = tempdir().expect("tempdir");
std::fs::write(tmp.path().join(CONFIG_TOML_FILE), "model = \"user\"").unwrap();
let managed_path = tmp.path().join("managed_config.toml");
std::fs::write(&managed_path, "approval_policy = \"never\"").unwrap();
let service = ConfigManager::new_for_tests(
tmp.path().to_path_buf(),
vec![],
LoaderOverrides::with_managed_config_path_for_tests(managed_path.clone()),
CloudRequirementsLoader::default(),
/*host_name*/ None,
);
let error = service
.write_value(ConfigValueWriteParams {
file_path: Some(tmp.path().join(CONFIG_TOML_FILE).display().to_string()),
key_path: "approval_policy".to_string(),
value: serde_json::json!("bogus"),
merge_strategy: MergeStrategy::Replace,
expected_version: None,
})
.await
.expect_err("should fail validation");
assert_eq!(
error.write_error_code(),
Some(ConfigWriteErrorCode::ConfigValidationError)
);
let contents = std::fs::read_to_string(tmp.path().join(CONFIG_TOML_FILE)).expect("read config");
assert_eq!(contents.trim(), "model = \"user\"");
}
#[tokio::test]
async fn reserved_builtin_provider_override_rejected() {
let tmp = tempdir().expect("tempdir");
std::fs::write(tmp.path().join(CONFIG_TOML_FILE), "model = \"user\"\n").unwrap();
let service = ConfigManager::without_managed_config_for_tests(tmp.path().to_path_buf());
let error = service
.write_value(ConfigValueWriteParams {
file_path: Some(tmp.path().join(CONFIG_TOML_FILE).display().to_string()),
key_path: "model_providers.openai.name".to_string(),
value: serde_json::json!("OpenAI Override"),
merge_strategy: MergeStrategy::Replace,
expected_version: None,
})
.await
.expect_err("should reject reserved provider override");
assert_eq!(
error.write_error_code(),
Some(ConfigWriteErrorCode::ConfigValidationError)
);
assert!(error.to_string().contains("reserved built-in provider IDs"));
assert!(error.to_string().contains("`openai`"));
let contents = std::fs::read_to_string(tmp.path().join(CONFIG_TOML_FILE)).expect("read config");
assert_eq!(contents, "model = \"user\"\n");
}
#[tokio::test]
async fn write_value_rejects_feature_requirement_conflict() {
let tmp = tempdir().expect("tempdir");
std::fs::write(tmp.path().join(CONFIG_TOML_FILE), "").unwrap();
let service = ConfigManager::new_for_tests(
tmp.path().to_path_buf(),
vec![],
LoaderOverrides::without_managed_config_for_tests(),
CloudRequirementsLoader::new(async {
Ok(Some(ConfigRequirementsToml {
feature_requirements: Some(FeatureRequirementsToml {
entries: BTreeMap::from([("personality".to_string(), true)]),
}),
..Default::default()
}))
}),
/*host_name*/ None,
);
let error = service
.write_value(ConfigValueWriteParams {
file_path: Some(tmp.path().join(CONFIG_TOML_FILE).display().to_string()),
key_path: "features.personality".to_string(),
value: serde_json::json!(false),
merge_strategy: MergeStrategy::Replace,
expected_version: None,
})
.await
.expect_err("conflicting feature write should fail");
assert_eq!(
error.write_error_code(),
Some(ConfigWriteErrorCode::ConfigValidationError)
);
assert!(
error
.to_string()
.contains("invalid value for `features`: `features.personality=false`"),
"{error}"
);
assert_eq!(
std::fs::read_to_string(tmp.path().join(CONFIG_TOML_FILE)).unwrap(),
""
);
}
#[tokio::test]
async fn write_value_rejects_profile_feature_requirement_conflict() {
let tmp = tempdir().expect("tempdir");
std::fs::write(tmp.path().join(CONFIG_TOML_FILE), "").unwrap();
let service = ConfigManager::new_for_tests(
tmp.path().to_path_buf(),
vec![],
LoaderOverrides::without_managed_config_for_tests(),
CloudRequirementsLoader::new(async {
Ok(Some(ConfigRequirementsToml {
feature_requirements: Some(FeatureRequirementsToml {
entries: BTreeMap::from([("personality".to_string(), true)]),
}),
..Default::default()
}))
}),
/*host_name*/ None,
);
let error = service
.write_value(ConfigValueWriteParams {
file_path: Some(tmp.path().join(CONFIG_TOML_FILE).display().to_string()),
key_path: "profiles.enterprise.features.personality".to_string(),
value: serde_json::json!(false),
merge_strategy: MergeStrategy::Replace,
expected_version: None,
})
.await
.expect_err("conflicting profile feature write should fail");
assert_eq!(
error.write_error_code(),
Some(ConfigWriteErrorCode::ConfigValidationError)
);
assert!(
error.to_string().contains(
"invalid value for `features`: `profiles.enterprise.features.personality=false`"
),
"{error}"
);
assert_eq!(
std::fs::read_to_string(tmp.path().join(CONFIG_TOML_FILE)).unwrap(),
""
);
}
#[tokio::test]
async fn read_reports_managed_overrides_user_and_session_flags() {
let tmp = tempdir().expect("tempdir");
let user_path = tmp.path().join(CONFIG_TOML_FILE);
std::fs::write(&user_path, "model = \"user\"").unwrap();
let user_file = AbsolutePathBuf::try_from(user_path.clone()).expect("user file");
let managed_path = tmp.path().join("managed_config.toml");
std::fs::write(&managed_path, "model = \"system\"").unwrap();
let managed_file = AbsolutePathBuf::try_from(managed_path.clone()).expect("managed file");
let cli_overrides = vec![(
"model".to_string(),
TomlValue::String("session".to_string()),
)];
let service = ConfigManager::new_for_tests(
tmp.path().to_path_buf(),
cli_overrides,
LoaderOverrides::with_managed_config_path_for_tests(managed_path.clone()),
CloudRequirementsLoader::default(),
/*host_name*/ None,
);
let response = service
.read(ConfigReadParams {
include_layers: true,
cwd: None,
})
.await
.expect("response");
assert_eq!(response.config.model.as_deref(), Some("system"));
assert_eq!(
response.origins.get("model").expect("origin").name,
ConfigLayerSource::LegacyManagedConfigTomlFromFile {
file: managed_file.clone()
},
);
let layers = response.layers.expect("layers");
// Local macOS machines can surface an MDM-managed config layer at the
// top of the stack; ignore it so this test stays focused on file/session/user ordering.
let layers = if matches!(
layers.first().map(|layer| &layer.name),
Some(ConfigLayerSource::LegacyManagedConfigTomlFromMdm)
) {
&layers[1..]
} else {
layers.as_slice()
};
assert_eq!(
layers.first().unwrap().name,
ConfigLayerSource::LegacyManagedConfigTomlFromFile { file: managed_file }
);
assert_eq!(layers.get(1).unwrap().name, ConfigLayerSource::SessionFlags);
assert_eq!(
layers.get(2).unwrap().name,
ConfigLayerSource::User { file: user_file }
);
}
#[tokio::test]
async fn write_value_reports_managed_override() {
let tmp = tempdir().expect("tempdir");
std::fs::write(tmp.path().join(CONFIG_TOML_FILE), "").unwrap();
let managed_path = tmp.path().join("managed_config.toml");
std::fs::write(&managed_path, "approval_policy = \"never\"").unwrap();
let managed_file = AbsolutePathBuf::try_from(managed_path.clone()).expect("managed file");
let service = ConfigManager::new_for_tests(
tmp.path().to_path_buf(),
vec![],
LoaderOverrides::with_managed_config_path_for_tests(managed_path.clone()),
CloudRequirementsLoader::default(),
/*host_name*/ None,
);
let result = service
.write_value(ConfigValueWriteParams {
file_path: Some(tmp.path().join(CONFIG_TOML_FILE).display().to_string()),
key_path: "approval_policy".to_string(),
value: serde_json::json!("on-request"),
merge_strategy: MergeStrategy::Replace,
expected_version: None,
})
.await
.expect("result");
assert_eq!(result.status, WriteStatus::OkOverridden);
let overridden = result.overridden_metadata.expect("overridden metadata");
assert_eq!(
overridden.overriding_layer.name,
ConfigLayerSource::LegacyManagedConfigTomlFromFile { file: managed_file }
);
assert_eq!(overridden.effective_value, serde_json::json!("never"));
}
#[tokio::test]
async fn upsert_merges_tables_replace_overwrites() -> Result<()> {
let tmp = tempdir().expect("tempdir");
let path = tmp.path().join(CONFIG_TOML_FILE);
let base = r#"[mcp_servers.linear]
bearer_token_env_var = "TOKEN"
name = "linear"
url = "https://linear.example"
[mcp_servers.linear.env_http_headers]
existing = "keep"
[mcp_servers.linear.http_headers]
alpha = "a"
"#;
let overlay = serde_json::json!({
"bearer_token_env_var": "NEW_TOKEN",
"http_headers": {
"alpha": "updated",
"beta": "b"
},
"name": "linear",
"url": "https://linear.example"
});
std::fs::write(&path, base)?;
let service = ConfigManager::without_managed_config_for_tests(tmp.path().to_path_buf());
service
.write_value(ConfigValueWriteParams {
file_path: Some(path.display().to_string()),
key_path: "mcp_servers.linear".to_string(),
value: overlay.clone(),
merge_strategy: MergeStrategy::Upsert,
expected_version: None,
})
.await
.expect("upsert succeeds");
let upserted: TomlValue = toml::from_str(&std::fs::read_to_string(&path)?)?;
let expected_upsert: TomlValue = toml::from_str(
r#"[mcp_servers.linear]
bearer_token_env_var = "NEW_TOKEN"
name = "linear"
url = "https://linear.example"
[mcp_servers.linear.env_http_headers]
existing = "keep"
[mcp_servers.linear.http_headers]
alpha = "updated"
beta = "b"
"#,
)?;
assert_eq!(upserted, expected_upsert);
std::fs::write(&path, base)?;
service
.write_value(ConfigValueWriteParams {
file_path: Some(path.display().to_string()),
key_path: "mcp_servers.linear".to_string(),
value: overlay,
merge_strategy: MergeStrategy::Replace,
expected_version: None,
})
.await
.expect("replace succeeds");
let replaced: TomlValue = toml::from_str(&std::fs::read_to_string(&path)?)?;
let expected_replace: TomlValue = toml::from_str(
r#"[mcp_servers.linear]
bearer_token_env_var = "NEW_TOKEN"
name = "linear"
url = "https://linear.example"
[mcp_servers.linear.http_headers]
alpha = "updated"
beta = "b"
"#,
)?;
assert_eq!(replaced, expected_replace);
Ok(())
}
codex-rs/app-server/src/external_agent_config_api.rs
+7 -7
View File
@@ -1,3 +1,8 @@
use crate::config::external_agent_config::ExternalAgentConfigDetectOptions;
use crate::config::external_agent_config::ExternalAgentConfigMigrationItem as CoreMigrationItem;
use crate::config::external_agent_config::ExternalAgentConfigMigrationItemType as CoreMigrationItemType;
use crate::config::external_agent_config::ExternalAgentConfigService;
use crate::config::external_agent_config::PendingPluginImport;
use crate::error_code::INTERNAL_ERROR_CODE;
use codex_app_server_protocol::ExternalAgentConfigDetectParams;
use codex_app_server_protocol::ExternalAgentConfigDetectResponse;
@@ -7,11 +12,6 @@ use codex_app_server_protocol::ExternalAgentConfigMigrationItemType;
use codex_app_server_protocol::JSONRPCErrorError;
use codex_app_server_protocol::MigrationDetails;
use codex_app_server_protocol::PluginsMigration;
use codex_core::external_agent_config::ExternalAgentConfigDetectOptions;
use codex_core::external_agent_config::ExternalAgentConfigMigrationItem as CoreMigrationItem;
use codex_core::external_agent_config::ExternalAgentConfigMigrationItemType as CoreMigrationItemType;
use codex_core::external_agent_config::ExternalAgentConfigService;
use codex_core::external_agent_config::PendingPluginImport;
use std::io;
use std::path::PathBuf;
@@ -108,12 +108,12 @@ impl ExternalAgentConfigApi {
description: migration_item.description,
cwd: migration_item.cwd,
details: migration_item.details.map(|details| {
codex_core::external_agent_config::MigrationDetails {
crate::config::external_agent_config::MigrationDetails {
plugins: details
.plugins
.into_iter()
.map(|plugin| {
codex_core::external_agent_config::PluginsMigration {
crate::config::external_agent_config::PluginsMigration {
marketplace_name: plugin.marketplace_name,
plugin_names: plugin.plugin_names,
}
codex-rs/app-server/src/lib.rs
+3
View File
@@ -70,7 +70,10 @@ mod app_server_tracing;
mod bespoke_event_handling;
mod codex_message_processor;
mod command_exec;
mod config;
mod config_api;
mod config_manager;
mod config_manager_service;
mod dynamic_tools;
mod error_code;
mod external_agent_config_api;
codex-rs/app-server/src/message_processor.rs
+12 -9
View File
@@ -10,6 +10,7 @@ use std::sync::atomic::Ordering;
use crate::codex_message_processor::CodexMessageProcessor;
use crate::codex_message_processor::CodexMessageProcessorArgs;
use crate::config_api::ConfigApi;
use crate::config_manager::ConfigManager;
use crate::error_code::INVALID_REQUEST_ERROR_CODE;
use crate::external_agent_config_api::ExternalAgentConfigApi;
use crate::fs_api::FsApi;
@@ -294,6 +295,15 @@ impl MessageProcessor {
let cli_overrides = Arc::new(RwLock::new(cli_overrides));
let runtime_feature_enablement = Arc::new(RwLock::new(BTreeMap::new()));
let cloud_requirements = Arc::new(RwLock::new(cloud_requirements));
let config_manager = ConfigManager::new(
config.codex_home.to_path_buf(),
cli_overrides,
runtime_feature_enablement,
loader_overrides,
cloud_requirements,
arg0_paths.clone(),
thread_config_loader,
);
let codex_message_processor = CodexMessageProcessor::new(CodexMessageProcessorArgs {
auth_manager: auth_manager.clone(),
thread_manager: Arc::clone(&thread_manager),
@@ -301,10 +311,7 @@ impl MessageProcessor {
analytics_events_client: analytics_events_client.clone(),
arg0_paths,
config: Arc::clone(&config),
cli_overrides: cli_overrides.clone(),
runtime_feature_enablement: runtime_feature_enablement.clone(),
cloud_requirements: cloud_requirements.clone(),
thread_config_loader,
config_manager: config_manager.clone(),
feedback,
log_db,
});
@@ -314,11 +321,7 @@ impl MessageProcessor {
.plugins_manager()
.maybe_start_plugin_startup_tasks_for_config(&config, auth_manager.clone());
let config_api = ConfigApi::new(
config.codex_home.to_path_buf(),
cli_overrides,
runtime_feature_enablement,
loader_overrides,
cloud_requirements,
config_manager,
thread_manager.clone(),
analytics_events_client.clone(),
);
codex-rs/app-server/tests/suite/v2/experimental_feature_list.rs
+4
View File
@@ -15,6 +15,7 @@ use codex_app_server_protocol::JSONRPCError;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RequestId;
use codex_core::config::ConfigBuilder;
use codex_core::config_loader::LoaderOverrides;
use codex_features::FEATURES;
use codex_features::Stage;
use pretty_assertions::assert_eq;
@@ -32,6 +33,9 @@ async fn experimental_feature_list_returns_feature_metadata_with_stage() -> Resu
let config = ConfigBuilder::default()
.codex_home(codex_home.path().to_path_buf())
.fallback_cwd(Some(codex_home.path().to_path_buf()))
.loader_overrides(LoaderOverrides::with_managed_config_path_for_tests(
codex_home.path().join("managed_config.toml"),
))
.build()
.await?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
codex-rs/app-server/tests/suite/v2/mcp_tool.rs
+11 -2
View File
@@ -83,10 +83,11 @@ url = "{mcp_server_url}/mcp"
)
.await??;
let ThreadStartResponse { thread, .. } = to_response(thread_start_resp)?;
let thread_id = thread.id.clone();
let tool_call_request_id = mcp
.send_mcp_server_tool_call_request(McpServerToolCallParams {
thread_id: thread.id,
thread_id: thread_id.clone(),
server: TEST_SERVER_NAME.to_string(),
tool: TEST_TOOL_NAME.to_string(),
arguments: Some(json!({
@@ -114,6 +115,7 @@ url = "{mcp_server_url}/mcp"
response.structured_content,
Some(json!({
"echoed": "hello from app",
"threadId": thread_id,
}))
);
assert_eq!(response.is_error, Some(false));
@@ -203,7 +205,7 @@ impl ServerHandler for ToolAppsMcpServer {
async fn call_tool(
&self,
request: CallToolRequestParams,
_context: RequestContext<RoleServer>,
context: RequestContext<RoleServer>,
) -> Result<CallToolResult, rmcp::ErrorData> {
assert_eq!(request.name.as_ref(), TEST_TOOL_NAME);
let message = request
@@ -212,12 +214,19 @@ impl ServerHandler for ToolAppsMcpServer {
.and_then(|arguments| arguments.get("message"))
.and_then(|value| value.as_str())
.unwrap_or_default();
let thread_id = context
.meta
.0
.get("threadId")
.and_then(|value| value.as_str())
.unwrap_or_default();
let mut meta = Meta::new();
meta.0.insert("calledBy".to_string(), json!("mcp-app"));
let mut result = CallToolResult::structured(json!({
"echoed": message,
"threadId": thread_id,
}));
result.content = vec![Content::text(format!("echo: {message}"))];
result.meta = Some(meta);
codex-rs/app-server/tests/suite/v2/request_permissions.rs
+1
View File
@@ -76,6 +76,7 @@ async fn request_permissions_round_trip() -> Result<()> {
assert_eq!(params.thread_id, thread.id);
assert_eq!(params.turn_id, turn.id);
assert_eq!(params.item_id, "call1");
assert!(params.cwd.as_path().is_absolute());
assert_eq!(params.reason, Some("Select a workspace root".to_string()));
let requested_writes = params
.permissions