mirror of
https://github.com/openai/codex.git
synced 2026-10-30 19:18:49 +00:00
Merge f38077e765 into sapling-pr-archive-bolinfest
This commit is contained in:
@@ -4,9 +4,9 @@
|
||||
//! from the local filesystem. It only applies to Business (aka Enterprise CBP) or Enterprise ChatGPT
|
||||
//! customers.
|
||||
//!
|
||||
//! Today, fetching is best-effort: on error or timeout, Codex continues without cloud requirements.
|
||||
//! We expect to tighten this so that Enterprise ChatGPT customers must successfully fetch these
|
||||
//! requirements before Codex will run.
|
||||
//! Fetching fails closed for eligible ChatGPT Business and Enterprise accounts. When cloud
|
||||
//! requirements cannot be loaded for those accounts, Codex fails configuration loading rather than
|
||||
//! continuing without them.
|
||||
|
||||
use async_trait::async_trait;
|
||||
use base64::Engine;
|
||||
@@ -17,6 +17,7 @@ use chrono::Utc;
|
||||
use codex_backend_client::Client as BackendClient;
|
||||
use codex_core::AuthManager;
|
||||
use codex_core::auth::CodexAuth;
|
||||
use codex_core::config_loader::CloudRequirementsLoadError;
|
||||
use codex_core::config_loader::CloudRequirementsLoader;
|
||||
use codex_core::config_loader::ConfigRequirementsToml;
|
||||
use codex_core::util::backoff;
|
||||
@@ -220,16 +221,34 @@ impl CloudRequirementsService {
|
||||
}
|
||||
}
|
||||
|
||||
async fn fetch_with_timeout(&self) -> Option<ConfigRequirementsToml> {
|
||||
async fn fetch_with_timeout(
|
||||
&self,
|
||||
) -> Result<Option<ConfigRequirementsToml>, CloudRequirementsLoadError> {
|
||||
let _timer =
|
||||
codex_otel::start_global_timer("codex.cloud_requirements.fetch.duration_ms", &[]);
|
||||
let started_at = Instant::now();
|
||||
let result = timeout(self.timeout, self.fetch())
|
||||
.await
|
||||
.inspect_err(|_| {
|
||||
tracing::warn!("Timed out waiting for cloud requirements; continuing without them");
|
||||
let message = format!(
|
||||
"Timed out waiting for cloud requirements after {}s",
|
||||
self.timeout.as_secs()
|
||||
);
|
||||
tracing::error!("{message}");
|
||||
if let Some(metrics) = codex_otel::metrics::global() {
|
||||
let _ = metrics.counter(
|
||||
"codex.cloud_requirements.load_failure",
|
||||
1,
|
||||
&[("trigger", "startup")],
|
||||
);
|
||||
}
|
||||
})
|
||||
.ok()?;
|
||||
.map_err(|_| {
|
||||
CloudRequirementsLoadError::new(format!(
|
||||
"timed out waiting for cloud requirements after {}s",
|
||||
self.timeout.as_secs()
|
||||
))
|
||||
})??;
|
||||
|
||||
match result.as_ref() {
|
||||
Some(requirements) => {
|
||||
@@ -247,18 +266,20 @@ impl CloudRequirementsService {
|
||||
}
|
||||
}
|
||||
|
||||
result
|
||||
Ok(result)
|
||||
}
|
||||
|
||||
async fn fetch(&self) -> Option<ConfigRequirementsToml> {
|
||||
let auth = self.auth_manager.auth().await?;
|
||||
async fn fetch(&self) -> Result<Option<ConfigRequirementsToml>, CloudRequirementsLoadError> {
|
||||
let Some(auth) = self.auth_manager.auth().await else {
|
||||
return Ok(None);
|
||||
};
|
||||
if !auth.is_chatgpt_auth()
|
||||
|| !matches!(
|
||||
auth.account_plan_type(),
|
||||
Some(PlanType::Business | PlanType::Enterprise)
|
||||
)
|
||||
{
|
||||
return None;
|
||||
return Ok(None);
|
||||
}
|
||||
let token_data = auth.get_token_data().ok();
|
||||
let chatgpt_user_id = token_data
|
||||
@@ -273,7 +294,7 @@ impl CloudRequirementsService {
|
||||
path = %self.cache_path.display(),
|
||||
"Using cached cloud requirements"
|
||||
);
|
||||
return signed_payload.requirements();
|
||||
return Ok(signed_payload.requirements());
|
||||
}
|
||||
Err(cache_load_status) => {
|
||||
self.log_cache_load_status(&cache_load_status);
|
||||
@@ -281,7 +302,15 @@ impl CloudRequirementsService {
|
||||
}
|
||||
|
||||
self.fetch_with_retries(&auth, chatgpt_user_id, account_id)
|
||||
.await?
|
||||
.await
|
||||
.ok_or_else(|| {
|
||||
let message = "failed to load your workspace-managed config";
|
||||
tracing::error!(
|
||||
path = %self.cache_path.display(),
|
||||
"{message}"
|
||||
);
|
||||
CloudRequirementsLoadError::new(message)
|
||||
})
|
||||
}
|
||||
|
||||
async fn fetch_with_retries(
|
||||
@@ -311,7 +340,7 @@ impl CloudRequirementsService {
|
||||
Some(contents) => match parse_cloud_requirements(contents) {
|
||||
Ok(requirements) => requirements,
|
||||
Err(err) => {
|
||||
tracing::warn!(error = %err, "Failed to parse cloud requirements");
|
||||
tracing::error!(error = %err, "Failed to parse cloud requirements");
|
||||
return None;
|
||||
}
|
||||
},
|
||||
@@ -342,7 +371,7 @@ impl CloudRequirementsService {
|
||||
Ok(true) => {}
|
||||
Ok(false) => break,
|
||||
Err(_) => {
|
||||
tracing::warn!(
|
||||
tracing::error!(
|
||||
"Timed out refreshing cloud requirements cache from remote; keeping existing cache"
|
||||
);
|
||||
}
|
||||
@@ -375,10 +404,17 @@ impl CloudRequirementsService {
|
||||
.await
|
||||
.is_none()
|
||||
{
|
||||
tracing::warn!(
|
||||
tracing::error!(
|
||||
path = %self.cache_path.display(),
|
||||
"Failed to refresh cloud requirements cache from remote"
|
||||
);
|
||||
if let Some(metrics) = codex_otel::metrics::global() {
|
||||
let _ = metrics.counter(
|
||||
"codex.cloud_requirements.load_failure",
|
||||
1,
|
||||
&[("trigger", "refresh")],
|
||||
);
|
||||
}
|
||||
}
|
||||
true
|
||||
}
|
||||
@@ -522,10 +558,10 @@ pub fn cloud_requirements_loader(
|
||||
existing_task.abort();
|
||||
}
|
||||
CloudRequirementsLoader::new(async move {
|
||||
task.await
|
||||
.inspect_err(|err| tracing::warn!(error = %err, "Cloud requirements task failed"))
|
||||
.ok()
|
||||
.flatten()
|
||||
task.await.map_err(|err| {
|
||||
tracing::error!(error = %err, "Cloud requirements task failed");
|
||||
CloudRequirementsLoadError::new(format!("cloud requirements load failed: {err}"))
|
||||
})?
|
||||
})
|
||||
}
|
||||
|
||||
@@ -692,7 +728,7 @@ mod tests {
|
||||
CLOUD_REQUIREMENTS_TIMEOUT,
|
||||
);
|
||||
let result = service.fetch().await;
|
||||
assert!(result.is_none());
|
||||
assert_eq!(result, Ok(None));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -705,7 +741,7 @@ mod tests {
|
||||
CLOUD_REQUIREMENTS_TIMEOUT,
|
||||
);
|
||||
let result = service.fetch().await;
|
||||
assert!(result.is_none());
|
||||
assert_eq!(result, Ok(None));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -721,7 +757,7 @@ mod tests {
|
||||
);
|
||||
assert_eq!(
|
||||
service.fetch().await,
|
||||
Some(ConfigRequirementsToml {
|
||||
Ok(Some(ConfigRequirementsToml {
|
||||
allowed_approval_policies: Some(vec![AskForApproval::Never]),
|
||||
allowed_sandbox_modes: None,
|
||||
allowed_web_search_modes: None,
|
||||
@@ -729,7 +765,7 @@ mod tests {
|
||||
rules: None,
|
||||
enforce_residency: None,
|
||||
network: None,
|
||||
})
|
||||
}))
|
||||
);
|
||||
}
|
||||
|
||||
@@ -789,7 +825,11 @@ mod tests {
|
||||
tokio::time::advance(CLOUD_REQUIREMENTS_TIMEOUT + Duration::from_millis(1)).await;
|
||||
|
||||
let result = handle.await.expect("cloud requirements task");
|
||||
assert!(result.is_none());
|
||||
let err = result.expect_err("cloud requirements timeout should fail closed");
|
||||
assert!(
|
||||
err.to_string()
|
||||
.contains("timed out waiting for cloud requirements")
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
@@ -812,7 +852,7 @@ mod tests {
|
||||
|
||||
assert_eq!(
|
||||
handle.await.expect("cloud requirements task"),
|
||||
Some(ConfigRequirementsToml {
|
||||
Ok(Some(ConfigRequirementsToml {
|
||||
allowed_approval_policies: Some(vec![AskForApproval::Never]),
|
||||
allowed_sandbox_modes: None,
|
||||
allowed_web_search_modes: None,
|
||||
@@ -820,7 +860,7 @@ mod tests {
|
||||
rules: None,
|
||||
enforce_residency: None,
|
||||
network: None,
|
||||
})
|
||||
}))
|
||||
);
|
||||
assert_eq!(fetcher.request_count.load(Ordering::SeqCst), 2);
|
||||
}
|
||||
@@ -839,7 +879,7 @@ mod tests {
|
||||
CLOUD_REQUIREMENTS_TIMEOUT,
|
||||
);
|
||||
|
||||
assert!(service.fetch().await.is_none());
|
||||
assert!(service.fetch().await.is_err());
|
||||
assert_eq!(fetcher.request_count.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
|
||||
@@ -868,7 +908,7 @@ mod tests {
|
||||
|
||||
assert_eq!(
|
||||
service.fetch().await,
|
||||
Some(ConfigRequirementsToml {
|
||||
Ok(Some(ConfigRequirementsToml {
|
||||
allowed_approval_policies: Some(vec![AskForApproval::Never]),
|
||||
allowed_sandbox_modes: None,
|
||||
allowed_web_search_modes: None,
|
||||
@@ -876,7 +916,7 @@ mod tests {
|
||||
rules: None,
|
||||
enforce_residency: None,
|
||||
network: None,
|
||||
})
|
||||
}))
|
||||
);
|
||||
assert_eq!(fetcher.request_count.load(Ordering::SeqCst), 0);
|
||||
}
|
||||
@@ -895,7 +935,7 @@ mod tests {
|
||||
|
||||
assert_eq!(
|
||||
service.fetch().await,
|
||||
Some(ConfigRequirementsToml {
|
||||
Ok(Some(ConfigRequirementsToml {
|
||||
allowed_approval_policies: Some(vec![AskForApproval::Never]),
|
||||
allowed_sandbox_modes: None,
|
||||
allowed_web_search_modes: None,
|
||||
@@ -903,7 +943,7 @@ mod tests {
|
||||
rules: None,
|
||||
enforce_residency: None,
|
||||
network: None,
|
||||
})
|
||||
}))
|
||||
);
|
||||
|
||||
let path = codex_home.path().join(CLOUD_REQUIREMENTS_CACHE_FILENAME);
|
||||
@@ -942,7 +982,7 @@ mod tests {
|
||||
|
||||
assert_eq!(
|
||||
service.fetch().await,
|
||||
Some(ConfigRequirementsToml {
|
||||
Ok(Some(ConfigRequirementsToml {
|
||||
allowed_approval_policies: Some(vec![AskForApproval::OnRequest]),
|
||||
allowed_sandbox_modes: None,
|
||||
allowed_web_search_modes: None,
|
||||
@@ -950,7 +990,7 @@ mod tests {
|
||||
rules: None,
|
||||
enforce_residency: None,
|
||||
network: None,
|
||||
})
|
||||
}))
|
||||
);
|
||||
assert_eq!(fetcher.request_count.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
@@ -988,7 +1028,7 @@ mod tests {
|
||||
|
||||
assert_eq!(
|
||||
service.fetch().await,
|
||||
Some(ConfigRequirementsToml {
|
||||
Ok(Some(ConfigRequirementsToml {
|
||||
allowed_approval_policies: Some(vec![AskForApproval::OnRequest]),
|
||||
allowed_sandbox_modes: None,
|
||||
allowed_web_search_modes: None,
|
||||
@@ -996,7 +1036,7 @@ mod tests {
|
||||
rules: None,
|
||||
enforce_residency: None,
|
||||
network: None,
|
||||
})
|
||||
}))
|
||||
);
|
||||
assert_eq!(fetcher.request_count.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
@@ -1038,7 +1078,7 @@ mod tests {
|
||||
|
||||
assert_eq!(
|
||||
service.fetch().await,
|
||||
Some(ConfigRequirementsToml {
|
||||
Ok(Some(ConfigRequirementsToml {
|
||||
allowed_approval_policies: Some(vec![AskForApproval::Never]),
|
||||
allowed_sandbox_modes: None,
|
||||
allowed_web_search_modes: None,
|
||||
@@ -1046,7 +1086,7 @@ mod tests {
|
||||
rules: None,
|
||||
enforce_residency: None,
|
||||
network: None,
|
||||
})
|
||||
}))
|
||||
);
|
||||
assert_eq!(fetcher.request_count.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
@@ -1089,7 +1129,7 @@ mod tests {
|
||||
|
||||
assert_eq!(
|
||||
service.fetch().await,
|
||||
Some(ConfigRequirementsToml {
|
||||
Ok(Some(ConfigRequirementsToml {
|
||||
allowed_approval_policies: Some(vec![AskForApproval::Never]),
|
||||
allowed_sandbox_modes: None,
|
||||
allowed_web_search_modes: None,
|
||||
@@ -1097,7 +1137,7 @@ mod tests {
|
||||
rules: None,
|
||||
enforce_residency: None,
|
||||
network: None,
|
||||
})
|
||||
}))
|
||||
);
|
||||
assert_eq!(fetcher.request_count.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
@@ -1171,7 +1211,7 @@ mod tests {
|
||||
CLOUD_REQUIREMENTS_TIMEOUT,
|
||||
);
|
||||
|
||||
assert!(service.fetch().await.is_none());
|
||||
assert_eq!(service.fetch().await, Ok(None));
|
||||
assert_eq!(fetcher.request_count.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
|
||||
@@ -1196,7 +1236,14 @@ mod tests {
|
||||
tokio::time::advance(Duration::from_secs(5)).await;
|
||||
tokio::task::yield_now().await;
|
||||
|
||||
assert!(handle.await.expect("cloud requirements task").is_none());
|
||||
let err = handle
|
||||
.await
|
||||
.expect("cloud requirements task")
|
||||
.expect_err("cloud requirements retry exhaustion should fail closed");
|
||||
assert_eq!(
|
||||
err.to_string(),
|
||||
"failed to load your workspace-managed config"
|
||||
);
|
||||
assert_eq!(
|
||||
fetcher.request_count.load(Ordering::SeqCst),
|
||||
CLOUD_REQUIREMENTS_MAX_ATTEMPTS
|
||||
@@ -1221,7 +1268,7 @@ mod tests {
|
||||
|
||||
assert_eq!(
|
||||
service.fetch().await,
|
||||
Some(ConfigRequirementsToml {
|
||||
Ok(Some(ConfigRequirementsToml {
|
||||
allowed_approval_policies: Some(vec![AskForApproval::Never]),
|
||||
allowed_sandbox_modes: None,
|
||||
allowed_web_search_modes: None,
|
||||
@@ -1229,10 +1276,10 @@ mod tests {
|
||||
rules: None,
|
||||
enforce_residency: None,
|
||||
network: None,
|
||||
})
|
||||
}))
|
||||
);
|
||||
|
||||
service.refresh_cache().await;
|
||||
assert!(service.refresh_cache().await);
|
||||
|
||||
let path = codex_home.path().join(CLOUD_REQUIREMENTS_CACHE_FILENAME);
|
||||
let cache_file: CloudRequirementsCacheFile =
|
||||
|
||||
@@ -4,24 +4,42 @@ use futures::future::FutureExt;
|
||||
use futures::future::Shared;
|
||||
use std::fmt;
|
||||
use std::future::Future;
|
||||
use thiserror::Error;
|
||||
|
||||
#[derive(Clone, Debug, Eq, Error, PartialEq)]
|
||||
#[error("{message}")]
|
||||
pub struct CloudRequirementsLoadError {
|
||||
message: String,
|
||||
}
|
||||
|
||||
impl CloudRequirementsLoadError {
|
||||
pub fn new(message: impl Into<String>) -> Self {
|
||||
Self {
|
||||
message: message.into(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct CloudRequirementsLoader {
|
||||
// TODO(gt): This should return a Result once we can fail-closed.
|
||||
fut: Shared<BoxFuture<'static, Option<ConfigRequirementsToml>>>,
|
||||
fut: Shared<
|
||||
BoxFuture<'static, Result<Option<ConfigRequirementsToml>, CloudRequirementsLoadError>>,
|
||||
>,
|
||||
}
|
||||
|
||||
impl CloudRequirementsLoader {
|
||||
pub fn new<F>(fut: F) -> Self
|
||||
where
|
||||
F: Future<Output = Option<ConfigRequirementsToml>> + Send + 'static,
|
||||
F: Future<Output = Result<Option<ConfigRequirementsToml>, CloudRequirementsLoadError>>
|
||||
+ Send
|
||||
+ 'static,
|
||||
{
|
||||
Self {
|
||||
fut: fut.boxed().shared(),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn get(&self) -> Option<ConfigRequirementsToml> {
|
||||
pub async fn get(&self) -> Result<Option<ConfigRequirementsToml>, CloudRequirementsLoadError> {
|
||||
self.fut.clone().await
|
||||
}
|
||||
}
|
||||
@@ -34,7 +52,7 @@ impl fmt::Debug for CloudRequirementsLoader {
|
||||
|
||||
impl Default for CloudRequirementsLoader {
|
||||
fn default() -> Self {
|
||||
Self::new(async { None })
|
||||
Self::new(async { Ok(None) })
|
||||
}
|
||||
}
|
||||
|
||||
@@ -52,7 +70,7 @@ mod tests {
|
||||
let counter_clone = Arc::clone(&counter);
|
||||
let loader = CloudRequirementsLoader::new(async move {
|
||||
counter_clone.fetch_add(1, Ordering::SeqCst);
|
||||
Some(ConfigRequirementsToml::default())
|
||||
Ok(Some(ConfigRequirementsToml::default()))
|
||||
});
|
||||
|
||||
let (first, second) = tokio::join!(loader.get(), loader.get());
|
||||
|
||||
@@ -10,6 +10,7 @@ mod state;
|
||||
|
||||
pub const CONFIG_TOML_FILE: &str = "config.toml";
|
||||
|
||||
pub use cloud_requirements::CloudRequirementsLoadError;
|
||||
pub use cloud_requirements::CloudRequirementsLoader;
|
||||
pub use config_requirements::ConfigRequirements;
|
||||
pub use config_requirements::ConfigRequirementsToml;
|
||||
|
||||
@@ -1,5 +1,14 @@
|
||||
load("//:defs.bzl", "codex_rust_crate")
|
||||
|
||||
filegroup(
|
||||
name = "model_availability_nux_fixtures",
|
||||
srcs = [
|
||||
"models.json",
|
||||
"tests/cli_responses_fixture.sse",
|
||||
],
|
||||
visibility = ["//visibility:public"],
|
||||
)
|
||||
|
||||
codex_rust_crate(
|
||||
name = "core",
|
||||
crate_name = "codex_core",
|
||||
|
||||
@@ -227,8 +227,10 @@ mod tests {
|
||||
use super::*;
|
||||
use crate::config::ConfigBuilder;
|
||||
use crate::config_loader::ConfigLayerStackOrdering;
|
||||
use crate::skills::SkillsManager;
|
||||
use codex_protocol::openai_models::ReasoningEffort;
|
||||
use pretty_assertions::assert_eq;
|
||||
use std::fs;
|
||||
use std::path::PathBuf;
|
||||
use tempfile::TempDir;
|
||||
|
||||
@@ -470,6 +472,53 @@ writable_roots = ["./sandbox-root"]
|
||||
assert_eq!(session_flags_layer_count(&config), before_layers + 1);
|
||||
}
|
||||
|
||||
#[cfg_attr(windows, ignore)]
|
||||
#[tokio::test]
|
||||
async fn apply_role_skills_config_disables_skill_for_spawned_agent() {
|
||||
let (home, mut config) = test_config_with_cli_overrides(Vec::new()).await;
|
||||
let skill_dir = home.path().join("skills").join("demo");
|
||||
fs::create_dir_all(&skill_dir).expect("create skill dir");
|
||||
let skill_path = skill_dir.join("SKILL.md");
|
||||
fs::write(
|
||||
&skill_path,
|
||||
"---\nname: demo-skill\ndescription: demo description\n---\n\n# Body\n",
|
||||
)
|
||||
.expect("write skill");
|
||||
let role_path = write_role_config(
|
||||
&home,
|
||||
"skills-role.toml",
|
||||
&format!(
|
||||
r#"[[skills.config]]
|
||||
path = "{}"
|
||||
enabled = false
|
||||
"#,
|
||||
skill_path.display()
|
||||
),
|
||||
)
|
||||
.await;
|
||||
config.agent_roles.insert(
|
||||
"custom".to_string(),
|
||||
AgentRoleConfig {
|
||||
description: None,
|
||||
config_file: Some(role_path),
|
||||
},
|
||||
);
|
||||
|
||||
apply_role_to_config(&mut config, Some("custom"))
|
||||
.await
|
||||
.expect("custom role should apply");
|
||||
|
||||
let skills_manager = SkillsManager::new(home.path().to_path_buf());
|
||||
let outcome = skills_manager.skills_for_config(&config);
|
||||
let skill = outcome
|
||||
.skills
|
||||
.iter()
|
||||
.find(|skill| skill.name == "demo-skill")
|
||||
.expect("demo skill should be discovered");
|
||||
|
||||
assert_eq!(outcome.is_skill_enabled(skill), false);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn spawn_tool_spec_build_deduplicates_user_defined_built_in_roles() {
|
||||
let user_defined_roles = BTreeMap::from([
|
||||
|
||||
+29
-10
@@ -371,14 +371,20 @@ impl Codex {
|
||||
.map_err(|err| CodexErr::Fatal(format!("failed to load rules: {err}")))?;
|
||||
|
||||
let config = Arc::new(config);
|
||||
let _ = models_manager
|
||||
.list_models(crate::models_manager::manager::RefreshStrategy::OnlineIfUncached)
|
||||
.await;
|
||||
let model = models_manager
|
||||
.get_default_model(
|
||||
&config.model,
|
||||
crate::models_manager::manager::RefreshStrategy::OnlineIfUncached,
|
||||
let refresh_strategy = match session_source {
|
||||
SessionSource::SubAgent(_) => crate::models_manager::manager::RefreshStrategy::Offline,
|
||||
_ => crate::models_manager::manager::RefreshStrategy::OnlineIfUncached,
|
||||
};
|
||||
if config.model.is_none()
|
||||
|| !matches!(
|
||||
refresh_strategy,
|
||||
crate::models_manager::manager::RefreshStrategy::Offline
|
||||
)
|
||||
{
|
||||
let _ = models_manager.list_models(refresh_strategy).await;
|
||||
}
|
||||
let model = models_manager
|
||||
.get_default_model(&config.model, refresh_strategy)
|
||||
.await;
|
||||
|
||||
// Resolve base instructions for the session. Priority order:
|
||||
@@ -1727,6 +1733,13 @@ impl Session {
|
||||
async fn record_initial_history(&self, conversation_history: InitialHistory) {
|
||||
let turn_context = self.new_default_turn().await;
|
||||
self.clear_mcp_tool_selection().await;
|
||||
let is_subagent = {
|
||||
let state = self.state.lock().await;
|
||||
matches!(
|
||||
state.session_configuration.session_source,
|
||||
SessionSource::SubAgent(_)
|
||||
)
|
||||
};
|
||||
match conversation_history {
|
||||
InitialHistory::New => {
|
||||
// Build and record initial items (user instructions + environment context)
|
||||
@@ -1741,7 +1754,9 @@ impl Session {
|
||||
}
|
||||
self.set_previous_model(None).await;
|
||||
// Ensure initial items are visible to immediate readers (e.g., tests, forks).
|
||||
self.flush_rollout().await;
|
||||
if !is_subagent {
|
||||
self.flush_rollout().await;
|
||||
}
|
||||
}
|
||||
InitialHistory::Resumed(resumed_history) => {
|
||||
let rollout_items = resumed_history.history;
|
||||
@@ -1792,7 +1807,9 @@ impl Session {
|
||||
|
||||
// Defer seeding the session's initial context until the first turn starts so
|
||||
// turn/start overrides can be merged before we write to the rollout.
|
||||
self.flush_rollout().await;
|
||||
if !is_subagent {
|
||||
self.flush_rollout().await;
|
||||
}
|
||||
}
|
||||
InitialHistory::Forked(rollout_items) => {
|
||||
let restored_tool_selection =
|
||||
@@ -1838,7 +1855,9 @@ impl Session {
|
||||
self.ensure_rollout_materialized().await;
|
||||
|
||||
// Flush after seeding history and any persisted rollout copy.
|
||||
self.flush_rollout().await;
|
||||
if !is_subagent {
|
||||
self.flush_rollout().await;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -5921,12 +5921,12 @@ mcp_oauth_callback_url = "https://example.com/callback"
|
||||
let config = ConfigBuilder::default()
|
||||
.codex_home(codex_home.path().to_path_buf())
|
||||
.cloud_requirements(CloudRequirementsLoader::new(async {
|
||||
Some(crate::config_loader::ConfigRequirementsToml {
|
||||
Ok(Some(crate::config_loader::ConfigRequirementsToml {
|
||||
allowed_sandbox_modes: Some(vec![
|
||||
crate::config_loader::SandboxModeRequirement::ReadOnly,
|
||||
]),
|
||||
..Default::default()
|
||||
})
|
||||
}))
|
||||
}))
|
||||
.build()
|
||||
.await?;
|
||||
@@ -5962,9 +5962,9 @@ mcp_oauth_callback_url = "https://example.com/callback"
|
||||
let config = ConfigBuilder::default()
|
||||
.codex_home(codex_home.path().to_path_buf())
|
||||
.fallback_cwd(Some(codex_home.path().to_path_buf()))
|
||||
.cloud_requirements(CloudRequirementsLoader::new(
|
||||
async move { Some(requirements) },
|
||||
))
|
||||
.cloud_requirements(CloudRequirementsLoader::new(async move {
|
||||
Ok(Some(requirements))
|
||||
}))
|
||||
.build()
|
||||
.await?;
|
||||
assert_eq!(
|
||||
@@ -5988,12 +5988,12 @@ mcp_oauth_callback_url = "https://example.com/callback"
|
||||
.codex_home(codex_home.path().to_path_buf())
|
||||
.fallback_cwd(Some(codex_home.path().to_path_buf()))
|
||||
.cloud_requirements(CloudRequirementsLoader::new(async {
|
||||
Some(crate::config_loader::ConfigRequirementsToml {
|
||||
Ok(Some(crate::config_loader::ConfigRequirementsToml {
|
||||
allowed_web_search_modes: Some(vec![
|
||||
crate::config_loader::WebSearchModeRequirement::Cached,
|
||||
]),
|
||||
..Default::default()
|
||||
})
|
||||
}))
|
||||
}))
|
||||
.build()
|
||||
.await?;
|
||||
@@ -6029,10 +6029,10 @@ trust_level = "untrusted"
|
||||
.codex_home(codex_home.path().to_path_buf())
|
||||
.fallback_cwd(Some(workspace.path().to_path_buf()))
|
||||
.cloud_requirements(CloudRequirementsLoader::new(async {
|
||||
Some(crate::config_loader::ConfigRequirementsToml {
|
||||
Ok(Some(crate::config_loader::ConfigRequirementsToml {
|
||||
allowed_approval_policies: Some(vec![AskForApproval::OnRequest]),
|
||||
..Default::default()
|
||||
})
|
||||
}))
|
||||
}))
|
||||
.build()
|
||||
.await?;
|
||||
@@ -6058,10 +6058,10 @@ trust_level = "untrusted"
|
||||
.codex_home(codex_home.path().to_path_buf())
|
||||
.fallback_cwd(Some(codex_home.path().to_path_buf()))
|
||||
.cloud_requirements(CloudRequirementsLoader::new(async {
|
||||
Some(crate::config_loader::ConfigRequirementsToml {
|
||||
Ok(Some(crate::config_loader::ConfigRequirementsToml {
|
||||
allowed_approval_policies: Some(vec![AskForApproval::OnRequest]),
|
||||
..Default::default()
|
||||
})
|
||||
}))
|
||||
}))
|
||||
.build()
|
||||
.await?;
|
||||
|
||||
@@ -26,7 +26,7 @@ pub const DEFAULT_OTEL_ENVIRONMENT: &str = "dev";
|
||||
pub const DEFAULT_MEMORIES_MAX_ROLLOUTS_PER_STARTUP: usize = 16;
|
||||
pub const DEFAULT_MEMORIES_MAX_ROLLOUT_AGE_DAYS: i64 = 30;
|
||||
pub const DEFAULT_MEMORIES_MIN_ROLLOUT_IDLE_HOURS: i64 = 6;
|
||||
pub const DEFAULT_MEMORIES_MAX_RAW_MEMORIES_FOR_GLOBAL: usize = 1_024;
|
||||
pub const DEFAULT_MEMORIES_MAX_RAW_MEMORIES_FOR_GLOBAL: usize = 256;
|
||||
pub const DEFAULT_MEMORIES_MAX_UNUSED_DAYS: i64 = 30;
|
||||
|
||||
#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, JsonSchema)]
|
||||
|
||||
@@ -25,6 +25,7 @@ use std::path::Path;
|
||||
use std::path::PathBuf;
|
||||
use toml::Value as TomlValue;
|
||||
|
||||
pub use codex_config::CloudRequirementsLoadError;
|
||||
pub use codex_config::CloudRequirementsLoader;
|
||||
pub use codex_config::ConfigError;
|
||||
pub use codex_config::ConfigLayerEntry;
|
||||
@@ -116,7 +117,7 @@ pub async fn load_config_layers_state(
|
||||
) -> io::Result<ConfigLayerStack> {
|
||||
let mut config_requirements_toml = ConfigRequirementsWithSources::default();
|
||||
|
||||
if let Some(requirements) = cloud_requirements.get().await {
|
||||
if let Some(requirements) = cloud_requirements.get().await.map_err(io::Error::other)? {
|
||||
config_requirements_toml
|
||||
.merge_unset_fields(RequirementSource::CloudRequirements, requirements);
|
||||
}
|
||||
|
||||
@@ -5,6 +5,7 @@ use crate::config::ConfigOverrides;
|
||||
use crate::config::ConfigToml;
|
||||
use crate::config::ConstraintError;
|
||||
use crate::config::ProjectConfig;
|
||||
use crate::config_loader::CloudRequirementsLoadError;
|
||||
use crate::config_loader::CloudRequirementsLoader;
|
||||
use crate::config_loader::ConfigLayerEntry;
|
||||
use crate::config_loader::ConfigLoadError;
|
||||
@@ -576,7 +577,7 @@ allowed_approval_policies = ["on-request"]
|
||||
..LoaderOverrides::default()
|
||||
},
|
||||
CloudRequirementsLoader::new(async {
|
||||
Some(ConfigRequirementsToml {
|
||||
Ok(Some(ConfigRequirementsToml {
|
||||
allowed_approval_policies: Some(vec![AskForApproval::Never]),
|
||||
allowed_sandbox_modes: None,
|
||||
allowed_web_search_modes: None,
|
||||
@@ -584,7 +585,7 @@ allowed_approval_policies = ["on-request"]
|
||||
rules: None,
|
||||
enforce_residency: None,
|
||||
network: None,
|
||||
})
|
||||
}))
|
||||
}),
|
||||
)
|
||||
.await?;
|
||||
@@ -671,7 +672,7 @@ async fn load_config_layers_includes_cloud_requirements() -> anyhow::Result<()>
|
||||
network: None,
|
||||
};
|
||||
let expected = requirements.clone();
|
||||
let cloud_requirements = CloudRequirementsLoader::new(async move { Some(requirements) });
|
||||
let cloud_requirements = CloudRequirementsLoader::new(async move { Ok(Some(requirements)) });
|
||||
|
||||
let layers = load_config_layers_state(
|
||||
&codex_home,
|
||||
@@ -702,6 +703,31 @@ async fn load_config_layers_includes_cloud_requirements() -> anyhow::Result<()>
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn load_config_layers_fails_when_cloud_requirements_loader_fails() -> anyhow::Result<()> {
|
||||
let tmp = tempdir()?;
|
||||
let codex_home = tmp.path().join("home");
|
||||
tokio::fs::create_dir_all(&codex_home).await?;
|
||||
let cwd = AbsolutePathBuf::from_absolute_path(tmp.path())?;
|
||||
|
||||
let err = load_config_layers_state(
|
||||
&codex_home,
|
||||
Some(cwd),
|
||||
&[] as &[(String, TomlValue)],
|
||||
LoaderOverrides::default(),
|
||||
CloudRequirementsLoader::new(async {
|
||||
Err(CloudRequirementsLoadError::new("cloud requirements failed"))
|
||||
}),
|
||||
)
|
||||
.await
|
||||
.expect_err("cloud requirements failure should fail closed");
|
||||
|
||||
assert_eq!(err.kind(), std::io::ErrorKind::Other);
|
||||
assert!(err.to_string().contains("cloud requirements failed"));
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn project_layers_prefer_closest_cwd() -> std::io::Result<()> {
|
||||
let tmp = tempdir()?;
|
||||
|
||||
@@ -5,6 +5,7 @@ use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
use std::sync::RwLock;
|
||||
|
||||
use codex_app_server_protocol::ConfigLayerSource;
|
||||
use codex_protocol::protocol::SkillScope;
|
||||
use codex_utils_absolute_path::AbsolutePathBuf;
|
||||
use toml::Value as TomlValue;
|
||||
@@ -14,6 +15,7 @@ use tracing::warn;
|
||||
use crate::config::Config;
|
||||
use crate::config::types::SkillsConfig;
|
||||
use crate::config_loader::CloudRequirementsLoader;
|
||||
use crate::config_loader::ConfigLayerStackOrdering;
|
||||
use crate::config_loader::LoaderOverrides;
|
||||
use crate::config_loader::load_config_layers_state;
|
||||
use crate::skills::SkillLoadOutcome;
|
||||
@@ -172,24 +174,31 @@ fn disabled_paths_from_stack(
|
||||
) -> HashSet<PathBuf> {
|
||||
let mut disabled = HashSet::new();
|
||||
let mut configs = HashMap::new();
|
||||
// Skills config is user-layer only for now; higher-precedence layers are ignored.
|
||||
let Some(user_layer) = config_layer_stack.get_user_layer() else {
|
||||
return disabled;
|
||||
};
|
||||
let Some(skills_value) = user_layer.config.get("skills") else {
|
||||
return disabled;
|
||||
};
|
||||
let skills: SkillsConfig = match skills_value.clone().try_into() {
|
||||
Ok(skills) => skills,
|
||||
Err(err) => {
|
||||
warn!("invalid skills config: {err}");
|
||||
return disabled;
|
||||
for layer in
|
||||
config_layer_stack.get_layers(ConfigLayerStackOrdering::LowestPrecedenceFirst, true)
|
||||
{
|
||||
if !matches!(
|
||||
layer.name,
|
||||
ConfigLayerSource::User { .. } | ConfigLayerSource::SessionFlags
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
for entry in skills.config {
|
||||
let path = normalize_override_path(entry.path.as_path());
|
||||
configs.insert(path, entry.enabled);
|
||||
let Some(skills_value) = layer.config.get("skills") else {
|
||||
continue;
|
||||
};
|
||||
let skills: SkillsConfig = match skills_value.clone().try_into() {
|
||||
Ok(skills) => skills,
|
||||
Err(err) => {
|
||||
warn!("invalid skills config: {err}");
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
for entry in skills.config {
|
||||
let path = normalize_override_path(entry.path.as_path());
|
||||
configs.insert(path, entry.enabled);
|
||||
}
|
||||
}
|
||||
|
||||
for (path, enabled) in configs {
|
||||
@@ -220,6 +229,9 @@ mod tests {
|
||||
use super::*;
|
||||
use crate::config::ConfigBuilder;
|
||||
use crate::config::ConfigOverrides;
|
||||
use crate::config_loader::ConfigLayerEntry;
|
||||
use crate::config_loader::ConfigLayerStack;
|
||||
use crate::config_loader::ConfigRequirementsToml;
|
||||
use pretty_assertions::assert_eq;
|
||||
use std::fs;
|
||||
use std::path::PathBuf;
|
||||
@@ -402,4 +414,85 @@ mod tests {
|
||||
|
||||
assert_eq!(first, second);
|
||||
}
|
||||
|
||||
#[cfg_attr(windows, ignore)]
|
||||
#[test]
|
||||
fn disabled_paths_from_stack_allows_session_flags_to_override_user_layer() {
|
||||
let tempdir = tempfile::tempdir().expect("tempdir");
|
||||
let skill_path = tempdir.path().join("skills").join("demo").join("SKILL.md");
|
||||
let user_file = AbsolutePathBuf::try_from(tempdir.path().join("config.toml"))
|
||||
.expect("user config path should be absolute");
|
||||
let user_layer = ConfigLayerEntry::new(
|
||||
ConfigLayerSource::User { file: user_file },
|
||||
toml::from_str(&format!(
|
||||
r#"[[skills.config]]
|
||||
path = "{}"
|
||||
enabled = false
|
||||
"#,
|
||||
skill_path.display()
|
||||
))
|
||||
.expect("user layer toml"),
|
||||
);
|
||||
let session_layer = ConfigLayerEntry::new(
|
||||
ConfigLayerSource::SessionFlags,
|
||||
toml::from_str(&format!(
|
||||
r#"[[skills.config]]
|
||||
path = "{}"
|
||||
enabled = true
|
||||
"#,
|
||||
skill_path.display()
|
||||
))
|
||||
.expect("session layer toml"),
|
||||
);
|
||||
let stack = ConfigLayerStack::new(
|
||||
vec![user_layer, session_layer],
|
||||
Default::default(),
|
||||
ConfigRequirementsToml::default(),
|
||||
)
|
||||
.expect("valid config layer stack");
|
||||
|
||||
assert_eq!(disabled_paths_from_stack(&stack), HashSet::new());
|
||||
}
|
||||
|
||||
#[cfg_attr(windows, ignore)]
|
||||
#[test]
|
||||
fn disabled_paths_from_stack_allows_session_flags_to_disable_user_enabled_skill() {
|
||||
let tempdir = tempfile::tempdir().expect("tempdir");
|
||||
let skill_path = tempdir.path().join("skills").join("demo").join("SKILL.md");
|
||||
let user_file = AbsolutePathBuf::try_from(tempdir.path().join("config.toml"))
|
||||
.expect("user config path should be absolute");
|
||||
let user_layer = ConfigLayerEntry::new(
|
||||
ConfigLayerSource::User { file: user_file },
|
||||
toml::from_str(&format!(
|
||||
r#"[[skills.config]]
|
||||
path = "{}"
|
||||
enabled = true
|
||||
"#,
|
||||
skill_path.display()
|
||||
))
|
||||
.expect("user layer toml"),
|
||||
);
|
||||
let session_layer = ConfigLayerEntry::new(
|
||||
ConfigLayerSource::SessionFlags,
|
||||
toml::from_str(&format!(
|
||||
r#"[[skills.config]]
|
||||
path = "{}"
|
||||
enabled = false
|
||||
"#,
|
||||
skill_path.display()
|
||||
))
|
||||
.expect("session layer toml"),
|
||||
);
|
||||
let stack = ConfigLayerStack::new(
|
||||
vec![user_layer, session_layer],
|
||||
Default::default(),
|
||||
ConfigRequirementsToml::default(),
|
||||
)
|
||||
.expect("valid config layer stack");
|
||||
|
||||
assert_eq!(
|
||||
disabled_paths_from_stack(&stack),
|
||||
HashSet::from([skill_path])
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -48,13 +48,75 @@ Quick-pass budget:
|
||||
During execution: if you hit repeated errors, confusing behavior, or suspect
|
||||
relevant prior context, redo the quick memory pass.
|
||||
|
||||
When to update memory:
|
||||
How to decide whether to verify memory:
|
||||
|
||||
- Consider both risk of drift and verification effort.
|
||||
- If a fact is likely to drift and is cheap to verify, verify it before
|
||||
answering.
|
||||
- If a fact is likely to drift but verification is expensive, slow, or
|
||||
disruptive, it is acceptable to answer from memory in an interactive turn,
|
||||
but you should say that it is memory-derived, note that it may be stale, and
|
||||
consider offering to refresh it live.
|
||||
- If a fact is lower-drift and cheap to verify, use judgment: verification is
|
||||
more important when the fact is central to the answer or especially easy to
|
||||
confirm.
|
||||
- If a fact is lower-drift and expensive to verify, it is usually fine to
|
||||
answer from memory directly.
|
||||
|
||||
When answering from memory without current verification:
|
||||
|
||||
- If you rely on memory for a fact that you did not verify in the current turn,
|
||||
say so briefly in the final answer.
|
||||
- If that fact is plausibly drift-prone or comes from an older note, older
|
||||
snapshot, or prior run summary, say that it may be stale or outdated.
|
||||
- If live verification was skipped and a refresh would be useful in the
|
||||
interactive context, consider offering to verify or refresh it live.
|
||||
- Do not present unverified memory-derived facts as confirmed-current.
|
||||
- For interactive requests, prefer a short refresh offer over silently doing
|
||||
expensive verification that the user did not ask for.
|
||||
- When the unverified fact is about prior results, commands, timing, or an
|
||||
older snapshot, a concrete refresh offer can be especially helpful.
|
||||
|
||||
When to update memory (automatic, same turn; required):
|
||||
|
||||
- Treat memory as guidance, not truth: if memory conflicts with current repo
|
||||
state, tool outputs, environment, or user feedback, current evidence wins.
|
||||
- If you discover stale/misleading memory, update memory files accordingly.
|
||||
- When user explicitly asks to remember/update memory, revise memory_summary.md
|
||||
and/or MEMORY.md.
|
||||
- Memory is writable. You are authorized to edit {{ base_path }}/MEMORY.md and
|
||||
{{ base_path }}/memory_summary.md when stale guidance is detected.
|
||||
- If any memory fact conflicts with current evidence (repo state, tool output,
|
||||
or user correction), you MUST update memory in the same turn. Do not wait for
|
||||
a separate user prompt.
|
||||
- If you detect stale memory, updating MEMORY.md is part of task completion,
|
||||
not optional cleanup.
|
||||
- A final answer without the required MEMORY.md edit is incorrect.
|
||||
- A memory entry can be partially stale: if the broad guidance is still useful
|
||||
but a stored detail is outdated (for example line numbers, exact paths, exact
|
||||
commands, or exact model/version strings), you should keep using current
|
||||
evidence in your answer and update the stale detail in MEMORY.md.
|
||||
- Correcting only the answer is not enough when you have identified a stale
|
||||
stored detail in memory.
|
||||
- If memory contains a broad point that is still right but any concrete stored
|
||||
detail is wrong or outdated, the memory is stale and MEMORY.md should be
|
||||
corrected in the same turn after you verify the replacement.
|
||||
- Required behavior after detecting stale memory:
|
||||
1. Verify the correct replacement using local evidence.
|
||||
2. Continue the task using current evidence; do not rely on stale memory.
|
||||
3. Edit memory files later in the same turn, before your final response:
|
||||
- Always update {{ base_path }}/MEMORY.md.
|
||||
- Update {{ base_path }}/memory_summary.md only if the correction affects
|
||||
reusable guidance and you have complete local file context for a
|
||||
targeted edit.
|
||||
4. Read back the changed MEMORY.md lines to confirm the update.
|
||||
5. Finalize the task after the memory updates are written.
|
||||
- Do not finish the turn until the stale memory is corrected or you have
|
||||
determined the correction is ambiguous.
|
||||
- If you verified a contradiction and did not edit MEMORY.md, the task is
|
||||
incomplete.
|
||||
- Only ask a clarifying question instead of editing when the replacement is
|
||||
ambiguous (multiple plausible targets with low confidence and no single
|
||||
verified replacement from local evidence).
|
||||
- When user explicitly asks to remember something or update the memory, revise
|
||||
the files accordingly.
|
||||
|
||||
Memory citation requirements:
|
||||
|
||||
|
||||
@@ -74,6 +74,8 @@ use codex_core::default_client::set_default_originator;
|
||||
use codex_core::find_thread_path_by_id_str;
|
||||
use codex_core::find_thread_path_by_name_str;
|
||||
|
||||
const DEFAULT_ANALYTICS_ENABLED: bool = true;
|
||||
|
||||
enum InitialOperation {
|
||||
UserTurn {
|
||||
items: Vec<UserInput>,
|
||||
@@ -316,7 +318,12 @@ pub async fn run_main(cli: Cli, arg0_paths: Arg0DispatchPaths) -> anyhow::Result
|
||||
}
|
||||
|
||||
let otel = match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
|
||||
codex_core::otel_init::build_provider(&config, env!("CARGO_PKG_VERSION"), None, false)
|
||||
codex_core::otel_init::build_provider(
|
||||
&config,
|
||||
env!("CARGO_PKG_VERSION"),
|
||||
None,
|
||||
DEFAULT_ANALYTICS_ENABLED,
|
||||
)
|
||||
})) {
|
||||
Ok(Ok(otel)) => otel,
|
||||
Ok(Err(e)) => {
|
||||
@@ -927,6 +934,11 @@ mod tests {
|
||||
use super::*;
|
||||
use pretty_assertions::assert_eq;
|
||||
|
||||
#[test]
|
||||
fn exec_defaults_analytics_to_enabled() {
|
||||
assert_eq!(DEFAULT_ANALYTICS_ENABLED, true);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn builds_uncommitted_review_request() {
|
||||
let request = build_review_request(ReviewArgs {
|
||||
|
||||
@@ -21,6 +21,7 @@ use tracing::debug;
|
||||
use tracing::error;
|
||||
use tracing::info;
|
||||
use tracing_subscriber::EnvFilter;
|
||||
use tracing_subscriber::prelude::*;
|
||||
|
||||
mod codex_tool_config;
|
||||
mod codex_tool_runner;
|
||||
@@ -45,6 +46,8 @@ pub use crate::patch_approval::PatchApprovalResponse;
|
||||
/// is a balance between throughput and memory usage – 128 messages should be
|
||||
/// plenty for an interactive CLI.
|
||||
const CHANNEL_CAPACITY: usize = 128;
|
||||
const DEFAULT_ANALYTICS_ENABLED: bool = true;
|
||||
const OTEL_SERVICE_NAME: &str = "codex_mcp_server";
|
||||
|
||||
type IncomingMessage = JsonRpcMessage<ClientRequest, Value, ClientNotification>;
|
||||
|
||||
@@ -52,12 +55,44 @@ pub async fn run_main(
|
||||
arg0_paths: Arg0DispatchPaths,
|
||||
cli_config_overrides: CliConfigOverrides,
|
||||
) -> IoResult<()> {
|
||||
// Install a simple subscriber so `tracing` output is visible. Users can
|
||||
// control the log level with `RUST_LOG`.
|
||||
tracing_subscriber::fmt()
|
||||
// Parse CLI overrides once and derive the base Config eagerly so later
|
||||
// components do not need to work with raw TOML values.
|
||||
let cli_kv_overrides = cli_config_overrides.parse_overrides().map_err(|e| {
|
||||
std::io::Error::new(
|
||||
ErrorKind::InvalidInput,
|
||||
format!("error parsing -c overrides: {e}"),
|
||||
)
|
||||
})?;
|
||||
let config = Config::load_with_cli_overrides(cli_kv_overrides)
|
||||
.await
|
||||
.map_err(|e| {
|
||||
std::io::Error::new(ErrorKind::InvalidData, format!("error loading config: {e}"))
|
||||
})?;
|
||||
|
||||
let otel = codex_core::otel_init::build_provider(
|
||||
&config,
|
||||
env!("CARGO_PKG_VERSION"),
|
||||
Some(OTEL_SERVICE_NAME),
|
||||
DEFAULT_ANALYTICS_ENABLED,
|
||||
)
|
||||
.map_err(|e| {
|
||||
std::io::Error::new(
|
||||
ErrorKind::InvalidData,
|
||||
format!("error loading otel config: {e}"),
|
||||
)
|
||||
})?;
|
||||
|
||||
let fmt_layer = tracing_subscriber::fmt::layer()
|
||||
.with_writer(std::io::stderr)
|
||||
.with_env_filter(EnvFilter::from_default_env())
|
||||
.init();
|
||||
.with_filter(EnvFilter::from_default_env());
|
||||
let otel_logger_layer = otel.as_ref().and_then(|provider| provider.logger_layer());
|
||||
let otel_tracing_layer = otel.as_ref().and_then(|provider| provider.tracing_layer());
|
||||
|
||||
let _ = tracing_subscriber::registry()
|
||||
.with(fmt_layer)
|
||||
.with(otel_logger_layer)
|
||||
.with(otel_tracing_layer)
|
||||
.try_init();
|
||||
|
||||
// Set up channels.
|
||||
let (incoming_tx, mut incoming_rx) = mpsc::channel::<IncomingMessage>(CHANNEL_CAPACITY);
|
||||
@@ -86,20 +121,6 @@ pub async fn run_main(
|
||||
}
|
||||
});
|
||||
|
||||
// Parse CLI overrides once and derive the base Config eagerly so later
|
||||
// components do not need to work with raw TOML values.
|
||||
let cli_kv_overrides = cli_config_overrides.parse_overrides().map_err(|e| {
|
||||
std::io::Error::new(
|
||||
ErrorKind::InvalidInput,
|
||||
format!("error parsing -c overrides: {e}"),
|
||||
)
|
||||
})?;
|
||||
let config = Config::load_with_cli_overrides(cli_kv_overrides)
|
||||
.await
|
||||
.map_err(|e| {
|
||||
std::io::Error::new(ErrorKind::InvalidData, format!("error loading config: {e}"))
|
||||
})?;
|
||||
|
||||
// Task: process incoming messages.
|
||||
let processor_handle = tokio::spawn({
|
||||
let outgoing_message_sender = OutgoingMessageSender::new(outgoing_tx);
|
||||
@@ -152,3 +173,55 @@ pub async fn run_main(
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use codex_core::config::ConfigBuilder;
|
||||
use codex_core::config::types::OtelExporterKind;
|
||||
use pretty_assertions::assert_eq;
|
||||
use std::collections::HashMap;
|
||||
use tempfile::TempDir;
|
||||
|
||||
#[test]
|
||||
fn mcp_server_defaults_analytics_to_enabled() {
|
||||
assert_eq!(DEFAULT_ANALYTICS_ENABLED, true);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn mcp_server_builds_otel_provider_with_logs_traces_and_metrics() -> anyhow::Result<()> {
|
||||
let codex_home = TempDir::new()?;
|
||||
let mut config = ConfigBuilder::default()
|
||||
.codex_home(codex_home.path().to_path_buf())
|
||||
.build()
|
||||
.await?;
|
||||
let exporter = OtelExporterKind::OtlpGrpc {
|
||||
endpoint: "http://localhost:4317".to_string(),
|
||||
headers: HashMap::new(),
|
||||
tls: None,
|
||||
};
|
||||
config.otel.exporter = exporter.clone();
|
||||
config.otel.trace_exporter = exporter.clone();
|
||||
config.otel.metrics_exporter = exporter;
|
||||
config.analytics_enabled = None;
|
||||
|
||||
let provider = codex_core::otel_init::build_provider(
|
||||
&config,
|
||||
"0.0.0-test",
|
||||
Some(OTEL_SERVICE_NAME),
|
||||
DEFAULT_ANALYTICS_ENABLED,
|
||||
)
|
||||
.map_err(|err| anyhow::anyhow!(err.to_string()))?
|
||||
.expect("otel provider");
|
||||
|
||||
assert!(provider.logger.is_some(), "expected log exporter");
|
||||
assert!(
|
||||
provider.tracer_provider.is_some(),
|
||||
"expected trace exporter"
|
||||
);
|
||||
assert!(provider.metrics().is_some(), "expected metrics exporter");
|
||||
provider.shutdown();
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -12,7 +12,7 @@ codex_rust_crate(
|
||||
],
|
||||
allow_empty = True,
|
||||
),
|
||||
test_data_extra = glob(["src/**/snapshots/**"]),
|
||||
test_data_extra = glob(["src/**/snapshots/**"]) + ["//codex-rs/core:model_availability_nux_fixtures"],
|
||||
integration_compile_data_extra = ["src/test_backend.rs"],
|
||||
extra_binaries = [
|
||||
"//codex-rs/cli:codex",
|
||||
|
||||
+395
-51
@@ -106,6 +106,7 @@ use tokio::sync::mpsc;
|
||||
use tokio::sync::mpsc::error::TryRecvError;
|
||||
use tokio::sync::mpsc::error::TrySendError;
|
||||
use tokio::sync::mpsc::unbounded_channel;
|
||||
use tokio::task::JoinHandle;
|
||||
use toml::Value as TomlValue;
|
||||
|
||||
mod pending_interactive_replay;
|
||||
@@ -684,6 +685,7 @@ pub(crate) struct App {
|
||||
windows_sandbox: WindowsSandboxState,
|
||||
|
||||
thread_event_channels: HashMap<ThreadId, ThreadEventChannel>,
|
||||
thread_event_listener_tasks: HashMap<ThreadId, JoinHandle<()>>,
|
||||
agent_picker_threads: HashMap<ThreadId, AgentPickerThreadEntry>,
|
||||
active_thread_id: Option<ThreadId>,
|
||||
active_thread_rx: Option<mpsc::Receiver<Event>>,
|
||||
@@ -863,6 +865,23 @@ impl App {
|
||||
self.suppress_shutdown_complete = true;
|
||||
self.chat_widget.submit_op(Op::Shutdown);
|
||||
self.server.remove_thread(&thread_id).await;
|
||||
self.abort_thread_event_listener(thread_id);
|
||||
}
|
||||
}
|
||||
|
||||
fn abort_thread_event_listener(&mut self, thread_id: ThreadId) {
|
||||
if let Some(handle) = self.thread_event_listener_tasks.remove(&thread_id) {
|
||||
handle.abort();
|
||||
}
|
||||
}
|
||||
|
||||
fn abort_all_thread_event_listeners(&mut self) {
|
||||
for handle in self
|
||||
.thread_event_listener_tasks
|
||||
.drain()
|
||||
.map(|(_, handle)| handle)
|
||||
{
|
||||
handle.abort();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -928,6 +947,14 @@ impl App {
|
||||
self.refresh_pending_thread_approvals().await;
|
||||
}
|
||||
|
||||
async fn note_thread_outbound_op(&mut self, thread_id: ThreadId, op: &Op) {
|
||||
let Some(channel) = self.thread_event_channels.get(&thread_id) else {
|
||||
return;
|
||||
};
|
||||
let mut store = channel.store.lock().await;
|
||||
store.note_outbound_op(op);
|
||||
}
|
||||
|
||||
async fn note_active_thread_outbound_op(&mut self, op: &Op) {
|
||||
if !ThreadEventStore::op_can_change_pending_replay_state(op) {
|
||||
return;
|
||||
@@ -935,11 +962,113 @@ impl App {
|
||||
let Some(thread_id) = self.active_thread_id else {
|
||||
return;
|
||||
};
|
||||
let Some(channel) = self.thread_event_channels.get(&thread_id) else {
|
||||
return;
|
||||
self.note_thread_outbound_op(thread_id, op).await;
|
||||
}
|
||||
|
||||
fn thread_label(&self, thread_id: ThreadId) -> String {
|
||||
let is_primary = self.primary_thread_id == Some(thread_id);
|
||||
let fallback_label = if is_primary {
|
||||
"Main [default]".to_string()
|
||||
} else {
|
||||
let thread_id = thread_id.to_string();
|
||||
let short_id: String = thread_id.chars().take(8).collect();
|
||||
format!("Agent ({short_id})")
|
||||
};
|
||||
let mut store = channel.store.lock().await;
|
||||
store.note_outbound_op(op);
|
||||
if let Some(entry) = self.agent_picker_threads.get(&thread_id) {
|
||||
let label = format_agent_picker_item_name(
|
||||
entry.agent_nickname.as_deref(),
|
||||
entry.agent_role.as_deref(),
|
||||
is_primary,
|
||||
);
|
||||
if label == "Agent" {
|
||||
let thread_id = thread_id.to_string();
|
||||
let short_id: String = thread_id.chars().take(8).collect();
|
||||
format!("{label} ({short_id})")
|
||||
} else {
|
||||
label
|
||||
}
|
||||
} else {
|
||||
fallback_label
|
||||
}
|
||||
}
|
||||
|
||||
async fn thread_cwd(&self, thread_id: ThreadId) -> Option<PathBuf> {
|
||||
let channel = self.thread_event_channels.get(&thread_id)?;
|
||||
let store = channel.store.lock().await;
|
||||
match store.session_configured.as_ref().map(|event| &event.msg) {
|
||||
Some(EventMsg::SessionConfigured(session)) => Some(session.cwd.clone()),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
async fn approval_request_for_thread_event(
|
||||
&self,
|
||||
thread_id: ThreadId,
|
||||
event: &Event,
|
||||
) -> Option<ApprovalRequest> {
|
||||
let thread_label = Some(self.thread_label(thread_id));
|
||||
match &event.msg {
|
||||
EventMsg::ExecApprovalRequest(ev) => Some(ApprovalRequest::Exec {
|
||||
thread_id,
|
||||
thread_label,
|
||||
id: ev.effective_approval_id(),
|
||||
command: ev.command.clone(),
|
||||
reason: ev.reason.clone(),
|
||||
available_decisions: ev.effective_available_decisions(),
|
||||
network_approval_context: ev.network_approval_context.clone(),
|
||||
additional_permissions: ev.additional_permissions.clone(),
|
||||
}),
|
||||
EventMsg::ApplyPatchApprovalRequest(ev) => Some(ApprovalRequest::ApplyPatch {
|
||||
thread_id,
|
||||
thread_label,
|
||||
id: ev.call_id.clone(),
|
||||
reason: ev.reason.clone(),
|
||||
cwd: self
|
||||
.thread_cwd(thread_id)
|
||||
.await
|
||||
.unwrap_or_else(|| self.config.cwd.clone()),
|
||||
changes: ev.changes.clone(),
|
||||
}),
|
||||
EventMsg::ElicitationRequest(ev) => Some(ApprovalRequest::McpElicitation {
|
||||
thread_id,
|
||||
thread_label,
|
||||
server_name: ev.server_name.clone(),
|
||||
request_id: ev.id.clone(),
|
||||
message: ev.message.clone(),
|
||||
}),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
async fn submit_op_to_thread(&mut self, thread_id: ThreadId, op: Op) {
|
||||
let replay_state_op =
|
||||
ThreadEventStore::op_can_change_pending_replay_state(&op).then(|| op.clone());
|
||||
let submitted = if self.active_thread_id == Some(thread_id) {
|
||||
self.chat_widget.submit_op(op)
|
||||
} else {
|
||||
crate::session_log::log_outbound_op(&op);
|
||||
match self.server.get_thread(thread_id).await {
|
||||
Ok(thread) => match thread.submit(op).await {
|
||||
Ok(_) => true,
|
||||
Err(err) => {
|
||||
self.chat_widget.add_error_message(format!(
|
||||
"Failed to submit op to thread {thread_id}: {err}"
|
||||
));
|
||||
false
|
||||
}
|
||||
},
|
||||
Err(err) => {
|
||||
self.chat_widget.add_error_message(format!(
|
||||
"Failed to find thread {thread_id} for approval response: {err}"
|
||||
));
|
||||
false
|
||||
}
|
||||
}
|
||||
};
|
||||
if submitted && let Some(op) = replay_state_op.as_ref() {
|
||||
self.note_thread_outbound_op(thread_id, op).await;
|
||||
self.refresh_pending_thread_approvals().await;
|
||||
}
|
||||
}
|
||||
|
||||
async fn refresh_pending_thread_approvals(&mut self) {
|
||||
@@ -965,32 +1094,7 @@ impl App {
|
||||
|
||||
let threads = pending_thread_ids
|
||||
.into_iter()
|
||||
.map(|thread_id| {
|
||||
let is_primary = self.primary_thread_id == Some(thread_id);
|
||||
let fallback_label = if is_primary {
|
||||
"Main [default]".to_string()
|
||||
} else {
|
||||
let thread_id = thread_id.to_string();
|
||||
let short_id: String = thread_id.chars().take(8).collect();
|
||||
format!("Agent ({short_id})")
|
||||
};
|
||||
if let Some(entry) = self.agent_picker_threads.get(&thread_id) {
|
||||
let label = format_agent_picker_item_name(
|
||||
entry.agent_nickname.as_deref(),
|
||||
entry.agent_role.as_deref(),
|
||||
is_primary,
|
||||
);
|
||||
if label == "Agent" {
|
||||
let thread_id = thread_id.to_string();
|
||||
let short_id: String = thread_id.chars().take(8).collect();
|
||||
format!("{label} ({short_id})")
|
||||
} else {
|
||||
label
|
||||
}
|
||||
} else {
|
||||
fallback_label
|
||||
}
|
||||
})
|
||||
.map(|thread_id| self.thread_label(thread_id))
|
||||
.collect();
|
||||
|
||||
self.chat_widget.set_pending_thread_approvals(threads);
|
||||
@@ -999,6 +1103,12 @@ impl App {
|
||||
async fn enqueue_thread_event(&mut self, thread_id: ThreadId, event: Event) -> Result<()> {
|
||||
let refresh_pending_thread_approvals =
|
||||
ThreadEventStore::event_can_change_pending_thread_approvals(&event);
|
||||
let inactive_approval_request = if self.active_thread_id != Some(thread_id) {
|
||||
self.approval_request_for_thread_event(thread_id, &event)
|
||||
.await
|
||||
} else {
|
||||
None
|
||||
};
|
||||
let (sender, store) = {
|
||||
let channel = self.ensure_thread_channel(thread_id);
|
||||
(channel.sender.clone(), Arc::clone(&channel.store))
|
||||
@@ -1027,6 +1137,8 @@ impl App {
|
||||
tracing::warn!("thread {thread_id} event channel closed");
|
||||
}
|
||||
}
|
||||
} else if let Some(request) = inactive_approval_request {
|
||||
self.chat_widget.push_approval_request(request);
|
||||
}
|
||||
if refresh_pending_thread_approvals {
|
||||
self.refresh_pending_thread_approvals().await;
|
||||
@@ -1034,6 +1146,19 @@ impl App {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn handle_routed_thread_event(
|
||||
&mut self,
|
||||
thread_id: ThreadId,
|
||||
event: Event,
|
||||
) -> Result<()> {
|
||||
if !self.thread_event_channels.contains_key(&thread_id) {
|
||||
tracing::debug!("dropping stale event for untracked thread {thread_id}");
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
self.enqueue_thread_event(thread_id, event).await
|
||||
}
|
||||
|
||||
async fn enqueue_primary_event(&mut self, event: Event) -> Result<()> {
|
||||
if let Some(thread_id) = self.primary_thread_id {
|
||||
return self.enqueue_thread_event(thread_id, event).await;
|
||||
@@ -1045,12 +1170,12 @@ impl App {
|
||||
self.primary_session_configured = Some(session.clone());
|
||||
self.ensure_thread_channel(thread_id);
|
||||
self.activate_thread_channel(thread_id).await;
|
||||
self.enqueue_thread_event(thread_id, event).await?;
|
||||
|
||||
let pending = std::mem::take(&mut self.pending_primary_events);
|
||||
for pending_event in pending {
|
||||
self.enqueue_thread_event(thread_id, pending_event).await?;
|
||||
}
|
||||
self.enqueue_thread_event(thread_id, event).await?;
|
||||
} else {
|
||||
self.pending_primary_events.push_back(event);
|
||||
}
|
||||
@@ -1227,6 +1352,7 @@ impl App {
|
||||
}
|
||||
|
||||
fn reset_thread_event_state(&mut self) {
|
||||
self.abort_all_thread_event_listeners();
|
||||
self.thread_event_channels.clear();
|
||||
self.agent_picker_threads.clear();
|
||||
self.active_thread_id = None;
|
||||
@@ -1597,6 +1723,7 @@ impl App {
|
||||
pending_shutdown_exit_thread_id: None,
|
||||
windows_sandbox: WindowsSandboxState::default(),
|
||||
thread_event_channels: HashMap::new(),
|
||||
thread_event_listener_tasks: HashMap::new(),
|
||||
agent_picker_threads: HashMap::new(),
|
||||
active_thread_id: None,
|
||||
active_thread_rx: None,
|
||||
@@ -2011,6 +2138,9 @@ impl App {
|
||||
AppEvent::CodexEvent(event) => {
|
||||
self.enqueue_primary_event(event).await?;
|
||||
}
|
||||
AppEvent::ThreadEvent { thread_id, event } => {
|
||||
self.handle_routed_thread_event(thread_id, event).await?;
|
||||
}
|
||||
AppEvent::Exit(mode) => {
|
||||
return Ok(self.handle_exit_mode(mode));
|
||||
}
|
||||
@@ -2026,6 +2156,9 @@ impl App {
|
||||
self.refresh_pending_thread_approvals().await;
|
||||
}
|
||||
}
|
||||
AppEvent::SubmitThreadOp { thread_id, op } => {
|
||||
self.submit_op_to_thread(thread_id, op).await;
|
||||
}
|
||||
AppEvent::DiffResult(text) => {
|
||||
// Clear the in-progress state in the bottom pane
|
||||
self.chat_widget.on_diff_complete();
|
||||
@@ -2823,9 +2956,6 @@ impl App {
|
||||
AppEvent::OpenAgentPicker => {
|
||||
self.open_agent_picker().await;
|
||||
}
|
||||
AppEvent::RefreshPendingThreadApprovals => {
|
||||
self.refresh_pending_thread_approvals().await;
|
||||
}
|
||||
AppEvent::SelectAgentThread(thread_id) => {
|
||||
self.select_agent_thread(tui, thread_id).await?;
|
||||
}
|
||||
@@ -3184,11 +3314,9 @@ impl App {
|
||||
};
|
||||
let channel =
|
||||
ThreadEventChannel::new_with_session_configured(THREAD_EVENT_CHANNEL_CAPACITY, event);
|
||||
let sender = channel.sender.clone();
|
||||
let store = Arc::clone(&channel.store);
|
||||
let app_event_tx = self.app_event_tx.clone();
|
||||
self.thread_event_channels.insert(thread_id, channel);
|
||||
tokio::spawn(async move {
|
||||
let listener_handle = tokio::spawn(async move {
|
||||
loop {
|
||||
let event = match thread.next_event().await {
|
||||
Ok(event) => event,
|
||||
@@ -3197,22 +3325,11 @@ impl App {
|
||||
break;
|
||||
}
|
||||
};
|
||||
let refresh_pending_thread_approvals =
|
||||
ThreadEventStore::event_can_change_pending_thread_approvals(&event);
|
||||
let should_send = {
|
||||
let mut guard = store.lock().await;
|
||||
guard.push_event(event.clone());
|
||||
guard.active
|
||||
};
|
||||
if refresh_pending_thread_approvals {
|
||||
app_event_tx.send(AppEvent::RefreshPendingThreadApprovals);
|
||||
}
|
||||
if should_send && let Err(err) = sender.send(event).await {
|
||||
tracing::debug!("external thread {thread_id} channel closed: {err}");
|
||||
break;
|
||||
}
|
||||
app_event_tx.send(AppEvent::ThreadEvent { thread_id, event });
|
||||
}
|
||||
});
|
||||
self.thread_event_listener_tasks
|
||||
.insert(thread_id, listener_handle);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -3612,6 +3729,152 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn enqueue_primary_event_delivers_session_configured_before_buffered_approval()
|
||||
-> Result<()> {
|
||||
let (mut app, mut app_event_rx, _op_rx) = make_test_app_with_channels().await;
|
||||
let thread_id = ThreadId::new();
|
||||
let approval_event = Event {
|
||||
id: "approval-event".to_string(),
|
||||
msg: EventMsg::ExecApprovalRequest(
|
||||
codex_protocol::protocol::ExecApprovalRequestEvent {
|
||||
call_id: "call-1".to_string(),
|
||||
approval_id: None,
|
||||
turn_id: "turn-1".to_string(),
|
||||
command: vec!["echo".to_string(), "hello".to_string()],
|
||||
cwd: PathBuf::from("/tmp/project"),
|
||||
reason: Some("needs approval".to_string()),
|
||||
network_approval_context: None,
|
||||
proposed_execpolicy_amendment: None,
|
||||
proposed_network_policy_amendments: None,
|
||||
additional_permissions: None,
|
||||
available_decisions: None,
|
||||
parsed_cmd: Vec::new(),
|
||||
},
|
||||
),
|
||||
};
|
||||
let session_configured_event = Event {
|
||||
id: "session-configured".to_string(),
|
||||
msg: EventMsg::SessionConfigured(SessionConfiguredEvent {
|
||||
session_id: thread_id,
|
||||
forked_from_id: None,
|
||||
thread_name: None,
|
||||
model: "gpt-test".to_string(),
|
||||
model_provider_id: "test-provider".to_string(),
|
||||
approval_policy: AskForApproval::Never,
|
||||
sandbox_policy: SandboxPolicy::new_read_only_policy(),
|
||||
cwd: PathBuf::from("/tmp/project"),
|
||||
reasoning_effort: None,
|
||||
history_log_id: 0,
|
||||
history_entry_count: 0,
|
||||
initial_messages: None,
|
||||
network_proxy: None,
|
||||
rollout_path: Some(PathBuf::new()),
|
||||
}),
|
||||
};
|
||||
|
||||
app.enqueue_primary_event(approval_event.clone()).await?;
|
||||
app.enqueue_primary_event(session_configured_event.clone())
|
||||
.await?;
|
||||
|
||||
let rx = app
|
||||
.active_thread_rx
|
||||
.as_mut()
|
||||
.expect("primary thread receiver should be active");
|
||||
let first_event = time::timeout(Duration::from_millis(50), rx.recv())
|
||||
.await
|
||||
.expect("timed out waiting for session configured event")
|
||||
.expect("channel closed unexpectedly");
|
||||
let second_event = time::timeout(Duration::from_millis(50), rx.recv())
|
||||
.await
|
||||
.expect("timed out waiting for buffered approval event")
|
||||
.expect("channel closed unexpectedly");
|
||||
|
||||
assert!(matches!(first_event.msg, EventMsg::SessionConfigured(_)));
|
||||
assert!(matches!(second_event.msg, EventMsg::ExecApprovalRequest(_)));
|
||||
|
||||
app.handle_codex_event_now(first_event);
|
||||
app.handle_codex_event_now(second_event);
|
||||
app.chat_widget
|
||||
.handle_key_event(KeyEvent::new(KeyCode::Char('y'), KeyModifiers::NONE));
|
||||
|
||||
while let Ok(app_event) = app_event_rx.try_recv() {
|
||||
if let AppEvent::SubmitThreadOp {
|
||||
thread_id: op_thread_id,
|
||||
..
|
||||
} = app_event
|
||||
{
|
||||
assert_eq!(op_thread_id, thread_id);
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
|
||||
panic!("expected approval action to submit a thread-scoped op");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn routed_thread_event_does_not_recreate_channel_after_reset() -> Result<()> {
|
||||
let mut app = make_test_app().await;
|
||||
let thread_id = ThreadId::new();
|
||||
app.thread_event_channels.insert(
|
||||
thread_id,
|
||||
ThreadEventChannel::new(THREAD_EVENT_CHANNEL_CAPACITY),
|
||||
);
|
||||
|
||||
app.reset_thread_event_state();
|
||||
app.handle_routed_thread_event(
|
||||
thread_id,
|
||||
Event {
|
||||
id: "stale-event".to_string(),
|
||||
msg: EventMsg::ShutdownComplete,
|
||||
},
|
||||
)
|
||||
.await?;
|
||||
|
||||
assert!(
|
||||
!app.thread_event_channels.contains_key(&thread_id),
|
||||
"stale routed events should not recreate cleared thread channels"
|
||||
);
|
||||
assert_eq!(app.active_thread_id, None);
|
||||
assert_eq!(app.primary_thread_id, None);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn reset_thread_event_state_aborts_listener_tasks() {
|
||||
struct NotifyOnDrop(Option<tokio::sync::oneshot::Sender<()>>);
|
||||
|
||||
impl Drop for NotifyOnDrop {
|
||||
fn drop(&mut self) {
|
||||
if let Some(tx) = self.0.take() {
|
||||
let _ = tx.send(());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let mut app = make_test_app().await;
|
||||
let thread_id = ThreadId::new();
|
||||
let (started_tx, started_rx) = tokio::sync::oneshot::channel();
|
||||
let (dropped_tx, dropped_rx) = tokio::sync::oneshot::channel();
|
||||
let handle = tokio::spawn(async move {
|
||||
let _notify_on_drop = NotifyOnDrop(Some(dropped_tx));
|
||||
let _ = started_tx.send(());
|
||||
std::future::pending::<()>().await;
|
||||
});
|
||||
app.thread_event_listener_tasks.insert(thread_id, handle);
|
||||
started_rx
|
||||
.await
|
||||
.expect("listener task should report it started");
|
||||
|
||||
app.reset_thread_event_state();
|
||||
|
||||
assert_eq!(app.thread_event_listener_tasks.is_empty(), true);
|
||||
time::timeout(Duration::from_millis(50), dropped_rx)
|
||||
.await
|
||||
.expect("timed out waiting for listener task abort")
|
||||
.expect("listener task drop notification should succeed");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn enqueue_thread_event_does_not_block_when_channel_full() -> Result<()> {
|
||||
let mut app = make_test_app().await;
|
||||
@@ -3761,6 +4024,85 @@ mod tests {
|
||||
assert!(app.chat_widget.pending_thread_approvals().is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn inactive_thread_approval_bubbles_into_active_view() -> Result<()> {
|
||||
let mut app = make_test_app().await;
|
||||
let main_thread_id =
|
||||
ThreadId::from_string("00000000-0000-0000-0000-000000000011").expect("valid thread");
|
||||
let agent_thread_id =
|
||||
ThreadId::from_string("00000000-0000-0000-0000-000000000022").expect("valid thread");
|
||||
|
||||
app.primary_thread_id = Some(main_thread_id);
|
||||
app.active_thread_id = Some(main_thread_id);
|
||||
app.thread_event_channels
|
||||
.insert(main_thread_id, ThreadEventChannel::new(1));
|
||||
app.thread_event_channels.insert(
|
||||
agent_thread_id,
|
||||
ThreadEventChannel::new_with_session_configured(
|
||||
1,
|
||||
Event {
|
||||
id: String::new(),
|
||||
msg: EventMsg::SessionConfigured(SessionConfiguredEvent {
|
||||
session_id: agent_thread_id,
|
||||
forked_from_id: None,
|
||||
thread_name: None,
|
||||
model: "gpt-5".to_string(),
|
||||
model_provider_id: "test-provider".to_string(),
|
||||
approval_policy: AskForApproval::OnRequest,
|
||||
sandbox_policy: SandboxPolicy::new_workspace_write_policy(),
|
||||
cwd: PathBuf::from("/tmp/agent"),
|
||||
reasoning_effort: None,
|
||||
history_log_id: 0,
|
||||
history_entry_count: 0,
|
||||
initial_messages: None,
|
||||
network_proxy: None,
|
||||
rollout_path: Some(PathBuf::from("/tmp/agent-rollout.jsonl")),
|
||||
}),
|
||||
},
|
||||
),
|
||||
);
|
||||
app.agent_picker_threads.insert(
|
||||
agent_thread_id,
|
||||
AgentPickerThreadEntry {
|
||||
agent_nickname: Some("Robie".to_string()),
|
||||
agent_role: Some("explorer".to_string()),
|
||||
is_closed: false,
|
||||
},
|
||||
);
|
||||
|
||||
app.enqueue_thread_event(
|
||||
agent_thread_id,
|
||||
Event {
|
||||
id: "ev-approval".to_string(),
|
||||
msg: EventMsg::ExecApprovalRequest(
|
||||
codex_protocol::protocol::ExecApprovalRequestEvent {
|
||||
call_id: "call-approval".to_string(),
|
||||
approval_id: None,
|
||||
turn_id: "turn-approval".to_string(),
|
||||
command: vec!["echo".to_string(), "hi".to_string()],
|
||||
cwd: PathBuf::from("/tmp/agent"),
|
||||
reason: Some("need approval".to_string()),
|
||||
network_approval_context: None,
|
||||
proposed_execpolicy_amendment: None,
|
||||
proposed_network_policy_amendments: None,
|
||||
additional_permissions: None,
|
||||
available_decisions: None,
|
||||
parsed_cmd: Vec::new(),
|
||||
},
|
||||
),
|
||||
},
|
||||
)
|
||||
.await?;
|
||||
|
||||
assert_eq!(app.chat_widget.has_active_view(), true);
|
||||
assert_eq!(
|
||||
app.chat_widget.pending_thread_approvals(),
|
||||
&["Robie [explorer]".to_string()]
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn agent_picker_item_name_snapshot() {
|
||||
let thread_id =
|
||||
@@ -4041,6 +4383,7 @@ mod tests {
|
||||
pending_shutdown_exit_thread_id: None,
|
||||
windows_sandbox: WindowsSandboxState::default(),
|
||||
thread_event_channels: HashMap::new(),
|
||||
thread_event_listener_tasks: HashMap::new(),
|
||||
agent_picker_threads: HashMap::new(),
|
||||
active_thread_id: None,
|
||||
active_thread_rx: None,
|
||||
@@ -4100,6 +4443,7 @@ mod tests {
|
||||
pending_shutdown_exit_thread_id: None,
|
||||
windows_sandbox: WindowsSandboxState::default(),
|
||||
thread_event_channels: HashMap::new(),
|
||||
thread_event_listener_tasks: HashMap::new(),
|
||||
agent_picker_threads: HashMap::new(),
|
||||
active_thread_id: None,
|
||||
active_thread_rx: None,
|
||||
|
||||
@@ -73,8 +73,17 @@ pub(crate) enum AppEvent {
|
||||
/// Switch the active thread to the selected agent.
|
||||
SelectAgentThread(ThreadId),
|
||||
|
||||
/// Recompute the list of inactive threads that still need approval.
|
||||
RefreshPendingThreadApprovals,
|
||||
/// Submit an op to the specified thread, regardless of current focus.
|
||||
SubmitThreadOp {
|
||||
thread_id: ThreadId,
|
||||
op: codex_protocol::protocol::Op,
|
||||
},
|
||||
|
||||
/// Forward an event from a non-primary thread into the app-level thread router.
|
||||
ThreadEvent {
|
||||
thread_id: ThreadId,
|
||||
event: Event,
|
||||
},
|
||||
|
||||
/// Start a new session.
|
||||
NewSession,
|
||||
|
||||
@@ -17,6 +17,7 @@ use crate::render::highlight::highlight_bash_to_lines;
|
||||
use crate::render::renderable::ColumnRenderable;
|
||||
use crate::render::renderable::Renderable;
|
||||
use codex_core::features::Features;
|
||||
use codex_protocol::ThreadId;
|
||||
use codex_protocol::mcp::RequestId;
|
||||
use codex_protocol::models::PermissionProfile;
|
||||
use codex_protocol::protocol::ElicitationAction;
|
||||
@@ -41,6 +42,8 @@ use ratatui::widgets::Wrap;
|
||||
#[derive(Clone, Debug)]
|
||||
pub(crate) enum ApprovalRequest {
|
||||
Exec {
|
||||
thread_id: ThreadId,
|
||||
thread_label: Option<String>,
|
||||
id: String,
|
||||
command: Vec<String>,
|
||||
reason: Option<String>,
|
||||
@@ -49,18 +52,40 @@ pub(crate) enum ApprovalRequest {
|
||||
additional_permissions: Option<PermissionProfile>,
|
||||
},
|
||||
ApplyPatch {
|
||||
thread_id: ThreadId,
|
||||
thread_label: Option<String>,
|
||||
id: String,
|
||||
reason: Option<String>,
|
||||
cwd: PathBuf,
|
||||
changes: HashMap<PathBuf, FileChange>,
|
||||
},
|
||||
McpElicitation {
|
||||
thread_id: ThreadId,
|
||||
thread_label: Option<String>,
|
||||
server_name: String,
|
||||
request_id: RequestId,
|
||||
message: String,
|
||||
},
|
||||
}
|
||||
|
||||
impl ApprovalRequest {
|
||||
fn thread_id(&self) -> ThreadId {
|
||||
match self {
|
||||
ApprovalRequest::Exec { thread_id, .. }
|
||||
| ApprovalRequest::ApplyPatch { thread_id, .. }
|
||||
| ApprovalRequest::McpElicitation { thread_id, .. } => *thread_id,
|
||||
}
|
||||
}
|
||||
|
||||
fn thread_label(&self) -> Option<&str> {
|
||||
match self {
|
||||
ApprovalRequest::Exec { thread_label, .. }
|
||||
| ApprovalRequest::ApplyPatch { thread_label, .. }
|
||||
| ApprovalRequest::McpElicitation { thread_label, .. } => thread_label.as_deref(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Modal overlay asking the user to approve or deny one or more requests.
|
||||
pub(crate) struct ApprovalOverlay {
|
||||
current_request: Option<ApprovalRequest>,
|
||||
@@ -158,13 +183,7 @@ impl ApprovalOverlay {
|
||||
.collect();
|
||||
|
||||
let params = SelectionViewParams {
|
||||
footer_hint: Some(Line::from(vec![
|
||||
"Press ".into(),
|
||||
key_hint::plain(KeyCode::Enter).into(),
|
||||
" to confirm or ".into(),
|
||||
key_hint::plain(KeyCode::Esc).into(),
|
||||
" to cancel".into(),
|
||||
])),
|
||||
footer_hint: Some(approval_footer_hint(request)),
|
||||
items,
|
||||
header,
|
||||
..Default::default()
|
||||
@@ -207,20 +226,39 @@ impl ApprovalOverlay {
|
||||
}
|
||||
|
||||
fn handle_exec_decision(&self, id: &str, command: &[String], decision: ReviewDecision) {
|
||||
let cell = history_cell::new_approval_decision_cell(command.to_vec(), decision.clone());
|
||||
self.app_event_tx.send(AppEvent::InsertHistoryCell(cell));
|
||||
self.app_event_tx.send(AppEvent::CodexOp(Op::ExecApproval {
|
||||
id: id.to_string(),
|
||||
turn_id: None,
|
||||
decision,
|
||||
}));
|
||||
let Some(request) = self.current_request.as_ref() else {
|
||||
return;
|
||||
};
|
||||
if request.thread_label().is_none() {
|
||||
let cell = history_cell::new_approval_decision_cell(command.to_vec(), decision.clone());
|
||||
self.app_event_tx.send(AppEvent::InsertHistoryCell(cell));
|
||||
}
|
||||
let thread_id = request.thread_id();
|
||||
self.app_event_tx.send(AppEvent::SubmitThreadOp {
|
||||
thread_id,
|
||||
op: Op::ExecApproval {
|
||||
id: id.to_string(),
|
||||
turn_id: None,
|
||||
decision,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
fn handle_patch_decision(&self, id: &str, decision: ReviewDecision) {
|
||||
self.app_event_tx.send(AppEvent::CodexOp(Op::PatchApproval {
|
||||
id: id.to_string(),
|
||||
decision,
|
||||
}));
|
||||
let Some(thread_id) = self
|
||||
.current_request
|
||||
.as_ref()
|
||||
.map(ApprovalRequest::thread_id)
|
||||
else {
|
||||
return;
|
||||
};
|
||||
self.app_event_tx.send(AppEvent::SubmitThreadOp {
|
||||
thread_id,
|
||||
op: Op::PatchApproval {
|
||||
id: id.to_string(),
|
||||
decision,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
fn handle_elicitation_decision(
|
||||
@@ -229,12 +267,21 @@ impl ApprovalOverlay {
|
||||
request_id: &RequestId,
|
||||
decision: ElicitationAction,
|
||||
) {
|
||||
self.app_event_tx
|
||||
.send(AppEvent::CodexOp(Op::ResolveElicitation {
|
||||
let Some(thread_id) = self
|
||||
.current_request
|
||||
.as_ref()
|
||||
.map(ApprovalRequest::thread_id)
|
||||
else {
|
||||
return;
|
||||
};
|
||||
self.app_event_tx.send(AppEvent::SubmitThreadOp {
|
||||
thread_id,
|
||||
op: Op::ResolveElicitation {
|
||||
server_name: server_name.to_string(),
|
||||
request_id: request_id.clone(),
|
||||
decision,
|
||||
}));
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
fn advance_queue(&mut self) {
|
||||
@@ -261,6 +308,23 @@ impl ApprovalOverlay {
|
||||
false
|
||||
}
|
||||
}
|
||||
KeyEvent {
|
||||
kind: KeyEventKind::Press,
|
||||
code: KeyCode::Char('o'),
|
||||
..
|
||||
} => {
|
||||
if let Some(request) = self.current_request.as_ref() {
|
||||
if request.thread_label().is_some() {
|
||||
self.app_event_tx
|
||||
.send(AppEvent::SelectAgentThread(request.thread_id()));
|
||||
true
|
||||
} else {
|
||||
false
|
||||
}
|
||||
} else {
|
||||
false
|
||||
}
|
||||
}
|
||||
e => {
|
||||
if let Some(idx) = self
|
||||
.options
|
||||
@@ -347,9 +411,28 @@ impl Renderable for ApprovalOverlay {
|
||||
}
|
||||
}
|
||||
|
||||
fn approval_footer_hint(request: &ApprovalRequest) -> Line<'static> {
|
||||
let mut spans = vec![
|
||||
"Press ".into(),
|
||||
key_hint::plain(KeyCode::Enter).into(),
|
||||
" to confirm or ".into(),
|
||||
key_hint::plain(KeyCode::Esc).into(),
|
||||
" to cancel".into(),
|
||||
];
|
||||
if request.thread_label().is_some() {
|
||||
spans.extend([
|
||||
" or ".into(),
|
||||
key_hint::plain(KeyCode::Char('o')).into(),
|
||||
" to open thread".into(),
|
||||
]);
|
||||
}
|
||||
Line::from(spans)
|
||||
}
|
||||
|
||||
fn build_header(request: &ApprovalRequest) -> Box<dyn Renderable> {
|
||||
match request {
|
||||
ApprovalRequest::Exec {
|
||||
thread_label,
|
||||
reason,
|
||||
command,
|
||||
network_approval_context,
|
||||
@@ -357,6 +440,13 @@ fn build_header(request: &ApprovalRequest) -> Box<dyn Renderable> {
|
||||
..
|
||||
} => {
|
||||
let mut header: Vec<Line<'static>> = Vec::new();
|
||||
if let Some(thread_label) = thread_label {
|
||||
header.push(Line::from(vec![
|
||||
"Thread: ".into(),
|
||||
thread_label.clone().bold(),
|
||||
]));
|
||||
header.push(Line::from(""));
|
||||
}
|
||||
if let Some(reason) = reason {
|
||||
header.push(Line::from(vec!["Reason: ".into(), reason.clone().italic()]));
|
||||
header.push(Line::from(""));
|
||||
@@ -381,12 +471,20 @@ fn build_header(request: &ApprovalRequest) -> Box<dyn Renderable> {
|
||||
Box::new(Paragraph::new(header).wrap(Wrap { trim: false }))
|
||||
}
|
||||
ApprovalRequest::ApplyPatch {
|
||||
thread_label,
|
||||
reason,
|
||||
cwd,
|
||||
changes,
|
||||
..
|
||||
} => {
|
||||
let mut header: Vec<Box<dyn Renderable>> = Vec::new();
|
||||
if let Some(thread_label) = thread_label {
|
||||
header.push(Box::new(Line::from(vec![
|
||||
"Thread: ".into(),
|
||||
thread_label.clone().bold(),
|
||||
])));
|
||||
header.push(Box::new(Line::from("")));
|
||||
}
|
||||
if let Some(reason) = reason
|
||||
&& !reason.is_empty()
|
||||
{
|
||||
@@ -403,16 +501,25 @@ fn build_header(request: &ApprovalRequest) -> Box<dyn Renderable> {
|
||||
Box::new(ColumnRenderable::with(header))
|
||||
}
|
||||
ApprovalRequest::McpElicitation {
|
||||
thread_label,
|
||||
server_name,
|
||||
message,
|
||||
..
|
||||
} => {
|
||||
let header = Paragraph::new(vec![
|
||||
let mut lines = Vec::new();
|
||||
if let Some(thread_label) = thread_label {
|
||||
lines.push(Line::from(vec![
|
||||
"Thread: ".into(),
|
||||
thread_label.clone().bold(),
|
||||
]));
|
||||
lines.push(Line::from(""));
|
||||
}
|
||||
lines.extend([
|
||||
Line::from(vec!["Server: ".into(), server_name.clone().bold()]),
|
||||
Line::from(""),
|
||||
Line::from(message.clone()),
|
||||
])
|
||||
.wrap(Wrap { trim: false });
|
||||
]);
|
||||
let header = Paragraph::new(lines).wrap(Wrap { trim: false });
|
||||
Box::new(header)
|
||||
}
|
||||
}
|
||||
@@ -652,6 +759,8 @@ mod tests {
|
||||
|
||||
fn make_exec_request() -> ApprovalRequest {
|
||||
ApprovalRequest::Exec {
|
||||
thread_id: ThreadId::new(),
|
||||
thread_label: None,
|
||||
id: "test".to_string(),
|
||||
command: vec!["echo".to_string(), "hi".to_string()],
|
||||
reason: Some("reason".to_string()),
|
||||
@@ -679,10 +788,10 @@ mod tests {
|
||||
let mut view = ApprovalOverlay::new(make_exec_request(), tx, Features::with_defaults());
|
||||
assert!(!view.is_complete());
|
||||
view.handle_key_event(KeyEvent::new(KeyCode::Char('y'), KeyModifiers::NONE));
|
||||
// We expect at least one CodexOp message in the queue.
|
||||
// We expect at least one thread-scoped approval op message in the queue.
|
||||
let mut saw_op = false;
|
||||
while let Ok(ev) = rx.try_recv() {
|
||||
if matches!(ev, AppEvent::CodexOp(_)) {
|
||||
if matches!(ev, AppEvent::SubmitThreadOp { .. }) {
|
||||
saw_op = true;
|
||||
break;
|
||||
}
|
||||
@@ -690,12 +799,68 @@ mod tests {
|
||||
assert!(saw_op, "expected approval decision to emit an op");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn o_opens_source_thread_for_cross_thread_approval() {
|
||||
let (tx, mut rx) = unbounded_channel::<AppEvent>();
|
||||
let tx = AppEventSender::new(tx);
|
||||
let thread_id = ThreadId::new();
|
||||
let mut view = ApprovalOverlay::new(
|
||||
ApprovalRequest::Exec {
|
||||
thread_id,
|
||||
thread_label: Some("Robie [explorer]".to_string()),
|
||||
id: "test".to_string(),
|
||||
command: vec!["echo".to_string(), "hi".to_string()],
|
||||
reason: None,
|
||||
available_decisions: vec![ReviewDecision::Approved, ReviewDecision::Abort],
|
||||
network_approval_context: None,
|
||||
additional_permissions: None,
|
||||
},
|
||||
tx,
|
||||
Features::with_defaults(),
|
||||
);
|
||||
|
||||
view.handle_key_event(KeyEvent::new(KeyCode::Char('o'), KeyModifiers::NONE));
|
||||
|
||||
let event = rx.try_recv().expect("expected select-agent-thread event");
|
||||
assert_eq!(
|
||||
matches!(event, AppEvent::SelectAgentThread(id) if id == thread_id),
|
||||
true
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn cross_thread_footer_hint_mentions_o_shortcut() {
|
||||
let (tx, _rx) = unbounded_channel::<AppEvent>();
|
||||
let tx = AppEventSender::new(tx);
|
||||
let view = ApprovalOverlay::new(
|
||||
ApprovalRequest::Exec {
|
||||
thread_id: ThreadId::new(),
|
||||
thread_label: Some("Robie [explorer]".to_string()),
|
||||
id: "test".to_string(),
|
||||
command: vec!["echo".to_string(), "hi".to_string()],
|
||||
reason: None,
|
||||
available_decisions: vec![ReviewDecision::Approved, ReviewDecision::Abort],
|
||||
network_approval_context: None,
|
||||
additional_permissions: None,
|
||||
},
|
||||
tx,
|
||||
Features::with_defaults(),
|
||||
);
|
||||
|
||||
assert_snapshot!(
|
||||
"approval_overlay_cross_thread_prompt",
|
||||
render_overlay_lines(&view, 80)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn exec_prefix_option_emits_execpolicy_amendment() {
|
||||
let (tx, mut rx) = unbounded_channel::<AppEvent>();
|
||||
let tx = AppEventSender::new(tx);
|
||||
let mut view = ApprovalOverlay::new(
|
||||
ApprovalRequest::Exec {
|
||||
thread_id: ThreadId::new(),
|
||||
thread_label: None,
|
||||
id: "test".to_string(),
|
||||
command: vec!["echo".to_string()],
|
||||
reason: None,
|
||||
@@ -717,7 +882,11 @@ mod tests {
|
||||
view.handle_key_event(KeyEvent::new(KeyCode::Char('p'), KeyModifiers::NONE));
|
||||
let mut saw_op = false;
|
||||
while let Ok(ev) = rx.try_recv() {
|
||||
if let AppEvent::CodexOp(Op::ExecApproval { decision, .. }) = ev {
|
||||
if let AppEvent::SubmitThreadOp {
|
||||
op: Op::ExecApproval { decision, .. },
|
||||
..
|
||||
} = ev
|
||||
{
|
||||
assert_eq!(
|
||||
decision,
|
||||
ReviewDecision::ApprovedExecpolicyAmendment {
|
||||
@@ -742,6 +911,8 @@ mod tests {
|
||||
let tx = AppEventSender::new(tx);
|
||||
let mut view = ApprovalOverlay::new(
|
||||
ApprovalRequest::Exec {
|
||||
thread_id: ThreadId::new(),
|
||||
thread_label: None,
|
||||
id: "test".to_string(),
|
||||
command: vec!["curl".to_string(), "https://example.com".to_string()],
|
||||
reason: None,
|
||||
@@ -779,6 +950,8 @@ mod tests {
|
||||
let tx = AppEventSender::new(tx);
|
||||
let command = vec!["echo".into(), "hello".into(), "world".into()];
|
||||
let exec_request = ApprovalRequest::Exec {
|
||||
thread_id: ThreadId::new(),
|
||||
thread_label: None,
|
||||
id: "test".into(),
|
||||
command,
|
||||
reason: None,
|
||||
@@ -893,6 +1066,8 @@ mod tests {
|
||||
let (tx, _rx) = unbounded_channel::<AppEvent>();
|
||||
let tx = AppEventSender::new(tx);
|
||||
let exec_request = ApprovalRequest::Exec {
|
||||
thread_id: ThreadId::new(),
|
||||
thread_label: None,
|
||||
id: "test".into(),
|
||||
command: vec!["cat".into(), "/tmp/readme.txt".into()],
|
||||
reason: None,
|
||||
@@ -932,6 +1107,8 @@ mod tests {
|
||||
let (tx, _rx) = unbounded_channel::<AppEvent>();
|
||||
let tx = AppEventSender::new(tx);
|
||||
let exec_request = ApprovalRequest::Exec {
|
||||
thread_id: ThreadId::new(),
|
||||
thread_label: None,
|
||||
id: "test".into(),
|
||||
command: vec!["cat".into(), "/tmp/readme.txt".into()],
|
||||
reason: Some("need filesystem access".into()),
|
||||
@@ -958,6 +1135,8 @@ mod tests {
|
||||
let (tx, _rx) = unbounded_channel::<AppEvent>();
|
||||
let tx = AppEventSender::new(tx);
|
||||
let exec_request = ApprovalRequest::Exec {
|
||||
thread_id: ThreadId::new(),
|
||||
thread_label: None,
|
||||
id: "test".into(),
|
||||
command: vec!["curl".into(), "https://example.com".into()],
|
||||
reason: Some("network request blocked".into()),
|
||||
@@ -1049,7 +1228,11 @@ mod tests {
|
||||
|
||||
let mut decision = None;
|
||||
while let Ok(ev) = rx.try_recv() {
|
||||
if let AppEvent::CodexOp(Op::ExecApproval { decision: d, .. }) = ev {
|
||||
if let AppEvent::SubmitThreadOp {
|
||||
op: Op::ExecApproval { decision: d, .. },
|
||||
..
|
||||
} = ev
|
||||
{
|
||||
decision = Some(d);
|
||||
break;
|
||||
}
|
||||
|
||||
@@ -27,8 +27,8 @@ use super::popup_consts::standard_popup_hint_line;
|
||||
use super::textarea::TextArea;
|
||||
use super::textarea::TextAreaState;
|
||||
|
||||
const BASE_BUG_ISSUE_URL: &str =
|
||||
"https://github.com/openai/codex/issues/new?template=2-bug-report.yml";
|
||||
const BASE_CLI_BUG_ISSUE_URL: &str =
|
||||
"https://github.com/openai/codex/issues/new?template=3-cli.yml";
|
||||
/// Internal routing link for employee feedback follow-ups. This must not be shown to external users.
|
||||
const CODEX_FEEDBACK_INTERNAL_URL: &str = "http://go/codex-feedback-internal";
|
||||
|
||||
@@ -389,7 +389,7 @@ fn issue_url_for_category(
|
||||
| FeedbackCategory::Other => Some(match feedback_audience {
|
||||
FeedbackAudience::OpenAiEmployee => slack_feedback_url(thread_id),
|
||||
FeedbackAudience::External => {
|
||||
format!("{BASE_BUG_ISSUE_URL}&steps=Uploaded%20thread:%20{thread_id}")
|
||||
format!("{BASE_CLI_BUG_ISSUE_URL}&steps=Uploaded%20thread:%20{thread_id}")
|
||||
}
|
||||
}),
|
||||
FeedbackCategory::GoodResult => None,
|
||||
@@ -675,10 +675,7 @@ mod tests {
|
||||
);
|
||||
let bug_url_non_employee =
|
||||
issue_url_for_category(FeedbackCategory::Bug, "t", FeedbackAudience::External);
|
||||
let expected_external_url = format!("{BASE_BUG_ISSUE_URL}&steps=Uploaded%20thread:%20t");
|
||||
assert_eq!(
|
||||
bug_url_non_employee.as_deref(),
|
||||
Some(expected_external_url.as_str())
|
||||
);
|
||||
let expected_external_url = "https://github.com/openai/codex/issues/new?template=3-cli.yml&steps=Uploaded%20thread:%20t";
|
||||
assert_eq!(bug_url_non_employee.as_deref(), Some(expected_external_url));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -812,6 +812,11 @@ impl BottomPane {
|
||||
self.is_task_running
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) fn has_active_view(&self) -> bool {
|
||||
!self.view_stack.is_empty()
|
||||
}
|
||||
|
||||
/// Return true when the pane is in the regular composer state without any
|
||||
/// overlays or popups and not running a task. This is the safe context to
|
||||
/// use Esc-Esc for backtracking from the main view.
|
||||
@@ -1115,6 +1120,8 @@ mod tests {
|
||||
|
||||
fn exec_request() -> ApprovalRequest {
|
||||
ApprovalRequest::Exec {
|
||||
thread_id: codex_protocol::ThreadId::new(),
|
||||
thread_label: None,
|
||||
id: "1".to_string(),
|
||||
command: vec!["echo".into(), "ok".into()],
|
||||
reason: None,
|
||||
|
||||
+14
@@ -0,0 +1,14 @@
|
||||
---
|
||||
source: tui/src/bottom_pane/approval_overlay.rs
|
||||
expression: "render_overlay_lines(&view, 80)"
|
||||
---
|
||||
Would you like to run the following command?
|
||||
|
||||
Thread: Robie [explorer]
|
||||
|
||||
$ echo hi
|
||||
|
||||
› 1. Yes, proceed (y)
|
||||
2. No, and tell Codex what to do differently (esc)
|
||||
|
||||
Press enter to confirm or esc to cancel or o to open thread
|
||||
@@ -2582,6 +2582,8 @@ impl ChatWidget {
|
||||
|
||||
let available_decisions = ev.effective_available_decisions();
|
||||
let request = ApprovalRequest::Exec {
|
||||
thread_id: self.thread_id.unwrap_or_default(),
|
||||
thread_label: None,
|
||||
id: ev.effective_approval_id(),
|
||||
command: ev.command,
|
||||
reason: ev.reason,
|
||||
@@ -2598,6 +2600,8 @@ impl ChatWidget {
|
||||
self.flush_answer_stream_with_separator();
|
||||
|
||||
let request = ApprovalRequest::ApplyPatch {
|
||||
thread_id: self.thread_id.unwrap_or_default(),
|
||||
thread_label: None,
|
||||
id: ev.call_id,
|
||||
reason: ev.reason,
|
||||
changes: ev.changes.clone(),
|
||||
@@ -2620,6 +2624,8 @@ impl ChatWidget {
|
||||
});
|
||||
|
||||
let request = ApprovalRequest::McpElicitation {
|
||||
thread_id: self.thread_id.unwrap_or_default(),
|
||||
thread_label: None,
|
||||
server_name: ev.server_name,
|
||||
request_id: ev.id,
|
||||
message: ev.message,
|
||||
@@ -2629,6 +2635,12 @@ impl ChatWidget {
|
||||
self.request_redraw();
|
||||
}
|
||||
|
||||
pub(crate) fn push_approval_request(&mut self, request: ApprovalRequest) {
|
||||
self.bottom_pane
|
||||
.push_approval_request(request, &self.config.features);
|
||||
self.request_redraw();
|
||||
}
|
||||
|
||||
pub(crate) fn handle_request_user_input_now(&mut self, ev: RequestUserInputEvent) {
|
||||
self.flush_answer_stream_with_separator();
|
||||
self.bottom_pane.push_user_input_request(ev);
|
||||
@@ -7543,6 +7555,11 @@ impl ChatWidget {
|
||||
self.bottom_pane.pending_thread_approvals()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) fn has_active_view(&self) -> bool {
|
||||
self.bottom_pane.has_active_view()
|
||||
}
|
||||
|
||||
pub(crate) fn show_esc_backtrack_hint(&mut self) {
|
||||
self.bottom_pane.show_esc_backtrack_hint();
|
||||
}
|
||||
|
||||
@@ -2854,7 +2854,11 @@ async fn exec_approval_uses_approval_id_when_present() {
|
||||
|
||||
let mut found = false;
|
||||
while let Ok(app_ev) = rx.try_recv() {
|
||||
if let AppEvent::CodexOp(Op::ExecApproval { id, decision, .. }) = app_ev {
|
||||
if let AppEvent::SubmitThreadOp {
|
||||
op: Op::ExecApproval { id, decision, .. },
|
||||
..
|
||||
} = app_ev
|
||||
{
|
||||
assert_eq!(id, "approval-subcommand");
|
||||
assert_matches!(decision, codex_protocol::protocol::ReviewDecision::Approved);
|
||||
found = true;
|
||||
@@ -7637,10 +7641,14 @@ async fn apply_patch_approval_sends_op_with_call_id() {
|
||||
// Approve via key press 'y'
|
||||
chat.handle_key_event(KeyEvent::new(KeyCode::Char('y'), KeyModifiers::NONE));
|
||||
|
||||
// Expect a CodexOp with PatchApproval carrying the call id.
|
||||
// Expect a thread-scoped PatchApproval op carrying the call id.
|
||||
let mut found = false;
|
||||
while let Ok(app_ev) = rx.try_recv() {
|
||||
if let AppEvent::CodexOp(Op::PatchApproval { id, decision }) = app_ev {
|
||||
if let AppEvent::SubmitThreadOp {
|
||||
op: Op::PatchApproval { id, decision },
|
||||
..
|
||||
} = app_ev
|
||||
{
|
||||
assert_eq!(id, "call-999");
|
||||
assert_matches!(decision, codex_protocol::protocol::ReviewDecision::Approved);
|
||||
found = true;
|
||||
@@ -7671,16 +7679,16 @@ async fn apply_patch_full_flow_integration_like() {
|
||||
}),
|
||||
});
|
||||
|
||||
// 2) User approves via 'y' and App receives a CodexOp
|
||||
// 2) User approves via 'y' and App receives a thread-scoped op
|
||||
chat.handle_key_event(KeyEvent::new(KeyCode::Char('y'), KeyModifiers::NONE));
|
||||
let mut maybe_op: Option<Op> = None;
|
||||
while let Ok(app_ev) = rx.try_recv() {
|
||||
if let AppEvent::CodexOp(op) = app_ev {
|
||||
if let AppEvent::SubmitThreadOp { op, .. } = app_ev {
|
||||
maybe_op = Some(op);
|
||||
break;
|
||||
}
|
||||
}
|
||||
let op = maybe_op.expect("expected CodexOp after key press");
|
||||
let op = maybe_op.expect("expected thread-scoped op after key press");
|
||||
|
||||
// 3) App forwards to widget.submit_op, which pushes onto codex_op_tx
|
||||
chat.submit_op(op);
|
||||
|
||||
Reference in New Issue
Block a user