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 1286b29a65 into sapling-pr-archive-bolinfest

Browse Source
This commit is contained in:
Michael Bolin
2026-01-30 09:51:35 -08:00
committed by GitHub
parent c6be423f7d 1286b29a65
commit 6a67d24b1f
87 changed files with 2930 additions and 293 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/agent/role.rs
+9 -13
View File
@@ -92,20 +92,16 @@ Rules:
},
AgentRole::Explorer => AgentProfile {
model: Some(EXPLORER_MODEL),
reasoning_effort: Some(ReasoningEffort::Low),
description: r#"Use for fast codebase understanding and information gathering.
`explorer` are extremely fast agents so use them as much as you can to speed up the resolution of the global task.
Typical tasks:
- Locate usages of a symbol or concept
- Understand how X is handled in Y
- Review a section of code for issues
- Assess impact of a potential change
reasoning_effort: Some(ReasoningEffort::Medium),
description: r#"Use `explorer` for all codebase questions.
Explorers are fast and authoritative.
Always prefer them over manual search or file reading.
Rules:
- Be explicit in what you are looking for. A good usage of `explorer` would mean that don't need to read the same code after the explorer send you the result.
- **Always** prefer asking explorers rather than exploring the codebase yourself.
- Spawn multiple explorers in parallel when useful and wait for all results.
- You can ask the `explorer` to return file name, lines, entire code snippets, ...
- Reuse the same explorer when it is relevant. If later in your process you have more questions on some code an explorer already covered, reuse this same explorer to be more efficient.
- Ask explorers first and precisely.
- Do not re-read or re-search code they cover.
- Trust explorer results without verification.
- Run explorers in parallel when useful.
- Reuse existing explorers for related questions.
"#,
..Default::default()
},
codex-rs/core/src/codex.rs
+145 -18
View File
@@ -26,6 +26,7 @@ use crate::features::maybe_push_unstable_features_warning;
use crate::models_manager::manager::ModelsManager;
use crate::parse_command::parse_command;
use crate::parse_turn_item;
use crate::rollout::session_index;
use crate::stream_events_utils::HandleOutputCtx;
use crate::stream_events_utils::handle_non_tool_response_item;
use crate::stream_events_utils::handle_output_item_done;
@@ -354,6 +355,8 @@ impl Codex {
sandbox_policy: config.sandbox_policy.clone(),
windows_sandbox_level: WindowsSandboxLevel::from_config(&config),
cwd: config.cwd.clone(),
codex_home: config.codex_home.clone(),
thread_name: None,
original_config_do_not_use: Arc::clone(&config),
session_source,
dynamic_tools,
@@ -544,6 +547,10 @@ pub(crate) struct SessionConfiguration {
/// `ConfigureSession` operation so that the business-logic layer can
/// operate deterministically.
cwd: PathBuf,
/// Directory containing all Codex state for this session.
codex_home: PathBuf,
/// Optional user-facing name for the thread, updated during the session.
thread_name: Option<String>,
// TODO(pakrym): Remove config from here
original_config_do_not_use: Arc<Config>,
@@ -553,6 +560,10 @@ pub(crate) struct SessionConfiguration {
}
impl SessionConfiguration {
pub(crate) fn codex_home(&self) -> &PathBuf {
&self.codex_home
}
fn thread_config_snapshot(&self) -> ThreadConfigSnapshot {
ThreadConfigSnapshot {
model: self.collaboration_mode.model().to_string(),
@@ -623,6 +634,11 @@ impl Session {
per_turn_config
}
pub(crate) async fn codex_home(&self) -> PathBuf {
let state = self.state.lock().await;
state.session_configuration.codex_home().clone()
}
#[allow(clippy::too_many_arguments)]
fn make_turn_context(
auth_manager: Option<Arc<AuthManager>>,
@@ -683,7 +699,7 @@ impl Session {
#[allow(clippy::too_many_arguments)]
async fn new(
session_configuration: SessionConfiguration,
mut session_configuration: SessionConfiguration,
config: Arc<Config>,
auth_manager: Arc<AuthManager>,
models_manager: Arc<ModelsManager>,
@@ -863,6 +879,16 @@ impl Session {
otel_manager.clone(),
);
}
let thread_name =
match session_index::find_thread_name_by_id(&config.codex_home, &conversation_id).await
{
Ok(name) => name,
Err(err) => {
warn!("Failed to read session index for thread name: {err}");
None
}
};
session_configuration.thread_name = thread_name.clone();
let state = SessionState::new(session_configuration.clone());
let services = SessionServices {
@@ -904,6 +930,7 @@ impl Session {
msg: EventMsg::SessionConfigured(SessionConfiguredEvent {
session_id: conversation_id,
forked_from_id,
thread_name: session_configuration.thread_name.clone(),
model: session_configuration.collaboration_mode.model().to_string(),
model_provider_id: config.model_provider_id.clone(),
approval_policy: session_configuration.approval_policy.value(),
@@ -1285,27 +1312,31 @@ impl Session {
previous: Option<&Arc<TurnContext>>,
next: &TurnContext,
) -> Option<ResponseItem> {
let personality = next.personality?;
if let Some(prev) = previous
&& prev.personality == Some(personality)
{
if !self.features.enabled(Feature::Personality) {
return None;
}
let model_info = next.client.get_model_info();
let personality_message = Self::personality_message_for(&model_info, personality);
let previous = previous?;
personality_message.map(|personality_message| {
DeveloperInstructions::personality_spec_message(personality_message).into()
})
// if a personality is specified and it's different from the previous one, build a personality update item
if let Some(personality) = next.personality
&& next.personality != previous.personality
{
let model_info = next.client.get_model_info();
let personality_message = Self::personality_message_for(&model_info, personality);
personality_message.map(|personality_message| {
DeveloperInstructions::personality_spec_message(personality_message).into()
})
} else {
None
}
}
fn personality_message_for(model_info: &ModelInfo, personality: Personality) -> Option<String> {
model_info
.model_instructions_template
.model_messages
.as_ref()
.and_then(|template| template.personality_messages.as_ref())
.and_then(|messages| messages.0.get(&personality))
.cloned()
.and_then(|spec| spec.get_personality_message(Some(personality)))
.filter(|message| !message.is_empty())
}
fn build_collaboration_mode_update_item(
@@ -1447,8 +1478,7 @@ impl Session {
.lock()
.await
.session_configuration
.original_config_do_not_use
.codex_home
.codex_home()
.clone();
if !features.enabled(Feature::ExecPolicy) {
@@ -1845,15 +1875,33 @@ impl Session {
items.push(DeveloperInstructions::new(developer_instructions.to_string()).into());
}
// Add developer instructions from collaboration_mode if they exist and are non-empty
let collaboration_mode = {
let (collaboration_mode, base_instructions) = {
let state = self.state.lock().await;
state.session_configuration.collaboration_mode.clone()
(
state.session_configuration.collaboration_mode.clone(),
state.session_configuration.base_instructions.clone(),
)
};
if let Some(collab_instructions) =
DeveloperInstructions::from_collaboration_mode(&collaboration_mode)
{
items.push(collab_instructions.into());
}
if self.features.enabled(Feature::Personality)
&& let Some(personality) = turn_context.personality
{
let model_info = turn_context.client.get_model_info();
let has_baked_personality = model_info.supports_personality()
&& base_instructions == model_info.get_model_instructions(Some(personality));
if !has_baked_personality
&& let Some(personality_message) =
Self::personality_message_for(&model_info, personality)
{
items.push(
DeveloperInstructions::personality_spec_message(personality_message).into(),
);
}
}
if let Some(user_instructions) = turn_context.user_instructions.as_deref() {
items.push(
UserInstructions {
@@ -2418,6 +2466,9 @@ async fn submission_loop(sess: Arc<Session>, config: Arc<Config>, rx_sub: Receiv
Op::ThreadRollback { num_turns } => {
handlers::thread_rollback(&sess, sub.id.clone(), num_turns).await;
}
Op::SetThreadName { name } => {
handlers::set_thread_name(&sess, sub.id.clone(), name).await;
}
Op::RunUserShellCommand { command } => {
handlers::run_user_shell_command(
&sess,
@@ -2461,6 +2512,7 @@ mod handlers {
use crate::mcp::collect_mcp_snapshot_from_manager;
use crate::mcp::effective_mcp_servers;
use crate::review_prompts::resolve_review_request;
use crate::rollout::session_index;
use crate::tasks::CompactTask;
use crate::tasks::RegularTask;
use crate::tasks::UndoTask;
@@ -2477,6 +2529,7 @@ mod handlers {
use codex_protocol::protocol::ReviewDecision;
use codex_protocol::protocol::ReviewRequest;
use codex_protocol::protocol::SkillsListEntry;
use codex_protocol::protocol::ThreadNameUpdatedEvent;
use codex_protocol::protocol::ThreadRolledBackEvent;
use codex_protocol::protocol::TurnAbortReason;
use codex_protocol::protocol::WarningEvent;
@@ -2930,6 +2983,72 @@ mod handlers {
.await;
}
/// Persists the thread name in the session index, updates in-memory state, and emits
/// a `ThreadNameUpdated` event on success.
///
/// This appends the name to `CODEX_HOME/sessions_index.jsonl` via `session_index::append_thread_name` for the
/// current `thread_id`, then updates `SessionConfiguration::thread_name`.
///
/// Returns an error event if the name is empty or session persistence is disabled.
pub async fn set_thread_name(sess: &Arc<Session>, sub_id: String, name: String) {
let Some(name) = crate::util::normalize_thread_name(&name) else {
let event = Event {
id: sub_id,
msg: EventMsg::Error(ErrorEvent {
message: "Thread name cannot be empty.".to_string(),
codex_error_info: Some(CodexErrorInfo::BadRequest),
}),
};
sess.send_event_raw(event).await;
return;
};
let persistence_enabled = {
let rollout = sess.services.rollout.lock().await;
rollout.is_some()
};
if !persistence_enabled {
let event = Event {
id: sub_id,
msg: EventMsg::Error(ErrorEvent {
message: "Session persistence is disabled; cannot rename thread.".to_string(),
codex_error_info: Some(CodexErrorInfo::Other),
}),
};
sess.send_event_raw(event).await;
return;
};
let codex_home = sess.codex_home().await;
if let Err(e) =
session_index::append_thread_name(&codex_home, sess.conversation_id, &name).await
{
let event = Event {
id: sub_id,
msg: EventMsg::Error(ErrorEvent {
message: format!("Failed to set thread name: {e}"),
codex_error_info: Some(CodexErrorInfo::Other),
}),
};
sess.send_event_raw(event).await;
return;
}
{
let mut state = sess.state.lock().await;
state.session_configuration.thread_name = Some(name.clone());
}
sess.send_event_raw(Event {
id: sub_id,
msg: EventMsg::ThreadNameUpdated(ThreadNameUpdatedEvent {
thread_id: sess.conversation_id,
thread_name: Some(name),
}),
})
.await;
}
pub async fn shutdown(sess: &Arc<Session>, sub_id: String) -> bool {
sess.abort_all_tasks(TurnAbortReason::Interrupted).await;
sess.services
@@ -4418,6 +4537,8 @@ mod tests {
sandbox_policy: config.sandbox_policy.clone(),
windows_sandbox_level: WindowsSandboxLevel::from_config(&config),
cwd: config.cwd.clone(),
codex_home: config.codex_home.clone(),
thread_name: None,
original_config_do_not_use: Arc::clone(&config),
session_source: SessionSource::Exec,
dynamic_tools: Vec::new(),
@@ -4499,6 +4620,8 @@ mod tests {
sandbox_policy: config.sandbox_policy.clone(),
windows_sandbox_level: WindowsSandboxLevel::from_config(&config),
cwd: config.cwd.clone(),
codex_home: config.codex_home.clone(),
thread_name: None,
original_config_do_not_use: Arc::clone(&config),
session_source: SessionSource::Exec,
dynamic_tools: Vec::new(),
@@ -4764,6 +4887,8 @@ mod tests {
sandbox_policy: config.sandbox_policy.clone(),
windows_sandbox_level: WindowsSandboxLevel::from_config(&config),
cwd: config.cwd.clone(),
codex_home: config.codex_home.clone(),
thread_name: None,
original_config_do_not_use: Arc::clone(&config),
session_source: SessionSource::Exec,
dynamic_tools: Vec::new(),
@@ -4878,6 +5003,8 @@ mod tests {
sandbox_policy: config.sandbox_policy.clone(),
windows_sandbox_level: WindowsSandboxLevel::from_config(&config),
cwd: config.cwd.clone(),
codex_home: config.codex_home.clone(),
thread_name: None,
original_config_do_not_use: Arc::clone(&config),
session_source: SessionSource::Exec,
dynamic_tools: Vec::new(),
codex-rs/core/src/codex_delegate.rs
+4
View File
@@ -208,6 +208,10 @@ async fn forward_events(
id: _,
msg: EventMsg::SessionConfigured(_),
} => {}
Event {
id: _,
msg: EventMsg::ThreadNameUpdated(_),
} => {}
Event {
id,
msg: EventMsg::ExecApprovalRequest(event),
codex-rs/core/src/config/mod.rs
+28 -7
View File
@@ -18,6 +18,7 @@ use crate::config::types::ShellEnvironmentPolicyToml;
use crate::config::types::SkillsConfig;
use crate::config::types::Tui;
use crate::config::types::UriBasedFileOpener;
use crate::config_loader::CloudRequirementsLoader;
use crate::config_loader::ConfigLayerStack;
use crate::config_loader::ConfigRequirements;
use crate::config_loader::LoaderOverrides;
@@ -366,6 +367,7 @@ pub struct ConfigBuilder {
cli_overrides: Option<Vec<(String, TomlValue)>>,
harness_overrides: Option<ConfigOverrides>,
loader_overrides: Option<LoaderOverrides>,
cloud_requirements: Option<CloudRequirementsLoader>,
fallback_cwd: Option<PathBuf>,
}
@@ -390,6 +392,11 @@ impl ConfigBuilder {
self
}
pub fn cloud_requirements(mut self, cloud_requirements: CloudRequirementsLoader) -> Self {
self.cloud_requirements = Some(cloud_requirements);
self
}
pub fn fallback_cwd(mut self, fallback_cwd: Option<PathBuf>) -> Self {
self.fallback_cwd = fallback_cwd;
self
@@ -401,6 +408,7 @@ impl ConfigBuilder {
cli_overrides,
harness_overrides,
loader_overrides,
cloud_requirements,
fallback_cwd,
} = self;
let codex_home = codex_home.map_or_else(find_codex_home, std::io::Result::Ok)?;
@@ -413,9 +421,14 @@ impl ConfigBuilder {
None => AbsolutePathBuf::current_dir()?,
};
harness_overrides.cwd = Some(cwd.to_path_buf());
let config_layer_stack =
load_config_layers_state(&codex_home, Some(cwd), &cli_overrides, loader_overrides)
.await?;
let config_layer_stack = load_config_layers_state(
&codex_home,
Some(cwd),
&cli_overrides,
loader_overrides,
cloud_requirements,
)
.await?;
let merged_toml = config_layer_stack.effective_config();
// Note that each layer in ConfigLayerStack should have resolved
@@ -511,6 +524,7 @@ pub async fn load_config_as_toml_with_cli_overrides(
Some(cwd.clone()),
&cli_overrides,
LoaderOverrides::default(),
None,
)
.await?;
@@ -609,9 +623,14 @@ pub async fn load_global_mcp_servers(
// There is no cwd/project context for this query, so this will not include
// MCP servers defined in in-repo .codex/ folders.
let cwd: Option<AbsolutePathBuf> = None;
let config_layer_stack =
load_config_layers_state(codex_home, cwd, &cli_overrides, LoaderOverrides::default())
.await?;
let config_layer_stack = load_config_layers_state(
codex_home,
cwd,
&cli_overrides,
LoaderOverrides::default(),
None,
)
.await?;
let merged_toml = config_layer_stack.effective_config();
let Some(servers_value) = merged_toml.get("mcp_servers") else {
return Ok(BTreeMap::new());
@@ -2612,7 +2631,8 @@ profile = "project"
let cwd = AbsolutePathBuf::try_from(codex_home.path())?;
let config_layer_stack =
load_config_layers_state(codex_home.path(), Some(cwd), &Vec::new(), overrides).await?;
load_config_layers_state(codex_home.path(), Some(cwd), &Vec::new(), overrides, None)
.await?;
let cfg = deserialize_config_toml_with_base(
config_layer_stack.effective_config(),
codex_home.path(),
@@ -2739,6 +2759,7 @@ profile = "project"
Some(cwd),
&[("model".to_string(), TomlValue::String("cli".to_string()))],
overrides,
None,
)
.await?;
codex-rs/core/src/config/service.rs
+1
View File
@@ -376,6 +376,7 @@ impl ConfigService {
cwd,
&self.cli_overrides,
self.loader_overrides.clone(),
None,
)
.await
}
codex-rs/core/src/config_loader/README.md
+2 -1
View File
@@ -10,7 +10,7 @@ This module is the canonical place to **load and describe Codex configuration la
Exported from `codex_core::config_loader`:
- `load_config_layers_state(codex_home, cwd_opt, cli_overrides, overrides) -> ConfigLayerStack`
- `load_config_layers_state(codex_home, cwd_opt, cli_overrides, overrides, cloud_requirements) -> ConfigLayerStack`
- `ConfigLayerStack`
- `effective_config() -> toml::Value`
- `origins() -> HashMap<String, ConfigLayerMetadata>`
@@ -49,6 +49,7 @@ let layers = load_config_layers_state(
Some(cwd),
&cli_overrides,
LoaderOverrides::default(),
None,
).await?;
let effective = layers.effective_config();
codex-rs/core/src/config_loader/cloud_requirements.rs
+56
View File
@@ -0,0 +1,56 @@
use crate::config_loader::ConfigRequirementsToml;
use futures::future::BoxFuture;
use futures::future::FutureExt;
use futures::future::Shared;
use std::fmt;
use std::future::Future;
#[derive(Clone)]
pub struct CloudRequirementsLoader {
// TODO(gt): This should return a Result once we can fail-closed.
fut: Shared<BoxFuture<'static, Option<ConfigRequirementsToml>>>,
}
impl CloudRequirementsLoader {
pub fn new<F>(fut: F) -> Self
where
F: Future<Output = Option<ConfigRequirementsToml>> + Send + 'static,
{
Self {
fut: fut.boxed().shared(),
}
}
pub async fn get(&self) -> Option<ConfigRequirementsToml> {
self.fut.clone().await
}
}
impl fmt::Debug for CloudRequirementsLoader {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("CloudRequirementsLoader").finish()
}
}
#[cfg(test)]
mod tests {
use super::*;
use pretty_assertions::assert_eq;
use std::sync::Arc;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;
#[tokio::test]
async fn shared_future_runs_once() {
let counter = Arc::new(AtomicUsize::new(0));
let counter_clone = Arc::clone(&counter);
let loader = CloudRequirementsLoader::new(async move {
counter_clone.fetch_add(1, Ordering::SeqCst);
Some(ConfigRequirementsToml::default())
});
let (first, second) = tokio::join!(loader.get(), loader.get());
assert_eq!(first, second);
assert_eq!(counter.load(Ordering::SeqCst), 1);
}
}
codex-rs/core/src/config_loader/config_requirements.rs
+31
View File
@@ -13,6 +13,7 @@ use crate::config::ConstraintError;
pub enum RequirementSource {
Unknown,
MdmManagedPreferences { domain: String, key: String },
CloudRequirements,
SystemRequirementsToml { file: AbsolutePathBuf },
LegacyManagedConfigTomlFromFile { file: AbsolutePathBuf },
LegacyManagedConfigTomlFromMdm,
@@ -25,6 +26,9 @@ impl fmt::Display for RequirementSource {
RequirementSource::MdmManagedPreferences { domain, key } => {
write!(f, "MDM {domain}:{key}")
}
RequirementSource::CloudRequirements => {
write!(f, "cloud requirements")
}
RequirementSource::SystemRequirementsToml { file } => {
write!(f, "{}", file.as_path().display())
}
@@ -448,6 +452,33 @@ mod tests {
Ok(())
}
#[test]
fn constraint_error_includes_cloud_requirements_source() -> Result<()> {
let source: ConfigRequirementsToml = from_str(
r#"
allowed_approval_policies = ["on-request"]
"#,
)?;
let source_location = RequirementSource::CloudRequirements;
let mut target = ConfigRequirementsWithSources::default();
target.merge_unset_fields(source_location.clone(), source);
let requirements = ConfigRequirements::try_from(target)?;
assert_eq!(
requirements.approval_policy.can_set(&AskForApproval::Never),
Err(ConstraintError::InvalidValue {
field_name: "approval_policy",
candidate: "Never".into(),
allowed: "[OnRequest]".into(),
requirement_source: source_location,
})
);
Ok(())
}
#[test]
fn deserialize_allowed_approval_policies() -> Result<()> {
let toml_str = r#"
codex-rs/core/src/config_loader/mod.rs
+11
View File
@@ -1,3 +1,4 @@
mod cloud_requirements;
mod config_requirements;
mod diagnostics;
mod fingerprint;
@@ -30,6 +31,7 @@ use std::io;
use std::path::Path;
use toml::Value as TomlValue;
pub use cloud_requirements::CloudRequirementsLoader;
pub use config_requirements::ConfigRequirements;
pub use config_requirements::ConfigRequirementsToml;
pub use config_requirements::McpServerIdentity;
@@ -69,6 +71,7 @@ const DEFAULT_PROJECT_ROOT_MARKERS: &[&str] = &[".git"];
/// earlier layer cannot be overridden by a later layer:
///
/// - admin: managed preferences (*)
/// - cloud: managed cloud requirements
/// - system `/etc/codex/requirements.toml`
///
/// For backwards compatibility, we also load from
@@ -98,6 +101,7 @@ pub async fn load_config_layers_state(
cwd: Option<AbsolutePathBuf>,
cli_overrides: &[(String, TomlValue)],
overrides: LoaderOverrides,
cloud_requirements: Option<CloudRequirementsLoader>, // TODO(gt): Once exec and app-server are wired up, we can remove the option.
) -> io::Result<ConfigLayerStack> {
let mut config_requirements_toml = ConfigRequirementsWithSources::default();
@@ -110,6 +114,13 @@ pub async fn load_config_layers_state(
)
.await?;
if let Some(loader) = cloud_requirements
&& let Some(requirements) = loader.get().await
{
config_requirements_toml
.merge_unset_fields(RequirementSource::CloudRequirements, requirements);
}
// Honor /etc/codex/requirements.toml.
if cfg!(unix) {
load_requirements_toml(
codex-rs/core/src/config_loader/tests.rs
+103
View File
@@ -4,11 +4,15 @@ use crate::config::CONFIG_TOML_FILE;
use crate::config::ConfigBuilder;
use crate::config::ConfigOverrides;
use crate::config::ConfigToml;
use crate::config::ConstraintError;
use crate::config::ProjectConfig;
use crate::config_loader::CloudRequirementsLoader;
use crate::config_loader::ConfigLayerEntry;
use crate::config_loader::ConfigLoadError;
use crate::config_loader::ConfigRequirements;
use crate::config_loader::ConfigRequirementsToml;
use crate::config_loader::config_requirements::ConfigRequirementsWithSources;
use crate::config_loader::config_requirements::RequirementSource;
use crate::config_loader::fingerprint::version_for_toml;
use crate::config_loader::load_requirements_toml;
use codex_protocol::config_types::TrustLevel;
@@ -65,6 +69,7 @@ async fn returns_config_error_for_invalid_user_config_toml() {
Some(cwd),
&[] as &[(String, TomlValue)],
LoaderOverrides::default(),
None,
)
.await
.expect_err("expected error");
@@ -94,6 +99,7 @@ async fn returns_config_error_for_invalid_managed_config_toml() {
Some(cwd),
&[] as &[(String, TomlValue)],
overrides,
None,
)
.await
.expect_err("expected error");
@@ -182,6 +188,7 @@ extra = true
Some(cwd),
&[] as &[(String, TomlValue)],
overrides,
None,
)
.await
.expect("load config");
@@ -218,6 +225,7 @@ async fn returns_empty_when_all_layers_missing() {
Some(cwd),
&[] as &[(String, TomlValue)],
overrides,
None,
)
.await
.expect("load layers");
@@ -315,6 +323,7 @@ flag = false
Some(cwd),
&[] as &[(String, TomlValue)],
overrides,
None,
)
.await
.expect("load config");
@@ -354,6 +363,7 @@ allowed_sandbox_modes = ["read-only"]
),
),
},
None,
)
.await?;
@@ -414,6 +424,7 @@ allowed_approval_policies = ["never"]
),
),
},
None,
)
.await?;
@@ -472,6 +483,91 @@ allowed_approval_policies = ["never", "on-request"]
Ok(())
}
#[tokio::test(flavor = "current_thread")]
async fn cloud_requirements_are_not_overwritten_by_system_requirements() -> anyhow::Result<()> {
let tmp = tempdir()?;
let requirements_file = tmp.path().join("requirements.toml");
tokio::fs::write(
&requirements_file,
r#"
allowed_approval_policies = ["on-request"]
"#,
)
.await?;
let mut config_requirements_toml = ConfigRequirementsWithSources::default();
config_requirements_toml.merge_unset_fields(
RequirementSource::CloudRequirements,
ConfigRequirementsToml {
allowed_approval_policies: Some(vec![AskForApproval::Never]),
allowed_sandbox_modes: None,
mcp_servers: None,
},
);
load_requirements_toml(&mut config_requirements_toml, &requirements_file).await?;
assert_eq!(
config_requirements_toml
.allowed_approval_policies
.as_ref()
.map(|sourced| sourced.value.clone()),
Some(vec![AskForApproval::Never])
);
assert_eq!(
config_requirements_toml
.allowed_approval_policies
.as_ref()
.map(|sourced| sourced.source.clone()),
Some(RequirementSource::CloudRequirements)
);
Ok(())
}
#[tokio::test]
async fn load_config_layers_includes_cloud_requirements() -> 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 requirements = ConfigRequirementsToml {
allowed_approval_policies: Some(vec![AskForApproval::Never]),
allowed_sandbox_modes: None,
mcp_servers: None,
};
let expected = requirements.clone();
let cloud_requirements = CloudRequirementsLoader::new(async move { Some(requirements) });
let layers = load_config_layers_state(
&codex_home,
Some(cwd),
&[] as &[(String, TomlValue)],
LoaderOverrides::default(),
Some(cloud_requirements),
)
.await?;
assert_eq!(
layers.requirements_toml().allowed_approval_policies,
expected.allowed_approval_policies
);
assert_eq!(
layers
.requirements()
.approval_policy
.can_set(&AskForApproval::OnRequest),
Err(ConstraintError::InvalidValue {
field_name: "approval_policy",
candidate: "OnRequest".into(),
allowed: "[Never]".into(),
requirement_source: RequirementSource::CloudRequirements,
})
);
Ok(())
}
#[tokio::test]
async fn project_layers_prefer_closest_cwd() -> std::io::Result<()> {
let tmp = tempdir()?;
@@ -501,6 +597,7 @@ async fn project_layers_prefer_closest_cwd() -> std::io::Result<()> {
Some(cwd),
&[] as &[(String, TomlValue)],
LoaderOverrides::default(),
None,
)
.await?;
@@ -632,6 +729,7 @@ async fn project_layer_is_added_when_dot_codex_exists_without_config_toml() -> s
Some(cwd),
&[] as &[(String, TomlValue)],
LoaderOverrides::default(),
None,
)
.await?;
@@ -691,6 +789,7 @@ async fn project_layers_disabled_when_untrusted_or_unknown() -> std::io::Result<
Some(cwd.clone()),
&[] as &[(String, TomlValue)],
LoaderOverrides::default(),
None,
)
.await?;
let project_layers_untrusted: Vec<_> = layers_untrusted
@@ -728,6 +827,7 @@ async fn project_layers_disabled_when_untrusted_or_unknown() -> std::io::Result<
Some(cwd),
&[] as &[(String, TomlValue)],
LoaderOverrides::default(),
None,
)
.await?;
let project_layers_unknown: Vec<_> = layers_unknown
@@ -788,6 +888,7 @@ async fn invalid_project_config_ignored_when_untrusted_or_unknown() -> std::io::
Some(cwd.clone()),
&[] as &[(String, TomlValue)],
LoaderOverrides::default(),
None,
)
.await?;
let project_layers: Vec<_> = layers
@@ -843,6 +944,7 @@ async fn cli_overrides_with_relative_paths_do_not_break_trust_check() -> std::io
Some(cwd),
&cli_overrides,
LoaderOverrides::default(),
None,
)
.await?;
@@ -884,6 +986,7 @@ async fn project_root_markers_supports_alternate_markers() -> std::io::Result<()
Some(cwd),
&[] as &[(String, TomlValue)],
LoaderOverrides::default(),
None,
)
.await?;
codex-rs/core/src/lib.rs
+2
View File
@@ -100,12 +100,14 @@ pub mod turn_diff_tracker;
pub use rollout::ARCHIVED_SESSIONS_SUBDIR;
pub use rollout::INTERACTIVE_SESSION_SOURCES;
pub use rollout::RolloutRecorder;
pub use rollout::RolloutRecorderParams;
pub use rollout::SESSIONS_SUBDIR;
pub use rollout::SessionMeta;
pub use rollout::find_archived_thread_path_by_id_str;
#[deprecated(note = "use find_thread_path_by_id_str")]
pub use rollout::find_conversation_path_by_id_str;
pub use rollout::find_thread_path_by_id_str;
pub use rollout::find_thread_path_by_name_str;
pub use rollout::list::Cursor;
pub use rollout::list::ThreadItem;
pub use rollout::list::ThreadSortKey;
codex-rs/core/src/mcp_connection_manager.rs
+43 -6
View File
@@ -769,6 +769,32 @@ fn filter_tools(tools: Vec<ToolInfo>, filter: ToolFilter) -> Vec<ToolInfo> {
.collect()
}
fn normalize_codex_apps_tool_title(
server_name: &str,
connector_name: Option<&str>,
value: &str,
) -> String {
if server_name != CODEX_APPS_MCP_SERVER_NAME {
return value.to_string();
}
let Some(connector_name) = connector_name
.map(str::trim)
.filter(|name| !name.is_empty())
else {
return value.to_string();
};
let prefix = format!("{connector_name}_");
if let Some(stripped) = value.strip_prefix(&prefix)
&& !stripped.is_empty()
{
return stripped.to_string();
}
value.to_string()
}
fn resolve_bearer_token(
server_name: &str,
bearer_token_env_var: Option<&str>,
@@ -926,12 +952,23 @@ async fn list_tools_for_client(
Ok(resp
.tools
.into_iter()
.map(|tool| ToolInfo {
server_name: server_name.to_owned(),
tool_name: tool.tool.name.clone(),
tool: tool.tool,
connector_id: tool.connector_id,
connector_name: tool.connector_name,
.map(|tool| {
let connector_name = tool.connector_name;
let mut tool_def = tool.tool;
if let Some(title) = tool_def.title.as_deref() {
let normalized_title =
normalize_codex_apps_tool_title(server_name, connector_name.as_deref(), title);
if tool_def.title.as_deref() != Some(normalized_title.as_str()) {
tool_def.title = Some(normalized_title);
}
}
ToolInfo {
server_name: server_name.to_owned(),
tool_name: tool_def.name.clone(),
tool: tool_def,
connector_id: tool.connector_id,
connector_name,
}
})
.collect())
}
codex-rs/core/src/mcp_tool_call.rs
+314 -3
View File
@@ -1,20 +1,30 @@
use std::time::Duration;
use std::time::Instant;
use tracing::error;
use crate::codex::Session;
use crate::codex::TurnContext;
use crate::mcp::CODEX_APPS_MCP_SERVER_NAME;
use crate::protocol::EventMsg;
use crate::protocol::McpInvocation;
use crate::protocol::McpToolCallBeginEvent;
use crate::protocol::McpToolCallEndEvent;
use codex_protocol::models::FunctionCallOutputPayload;
use codex_protocol::models::ResponseInputItem;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::SandboxPolicy;
use codex_protocol::request_user_input::RequestUserInputArgs;
use codex_protocol::request_user_input::RequestUserInputQuestion;
use codex_protocol::request_user_input::RequestUserInputQuestionOption;
use codex_protocol::request_user_input::RequestUserInputResponse;
use mcp_types::ToolAnnotations;
use std::sync::Arc;
/// Handles the specified tool call dispatches the appropriate
/// `McpToolCallBegin` and `McpToolCallEnd` events to the `Session`.
pub(crate) async fn handle_mcp_tool_call(
sess: &Session,
sess: Arc<Session>,
turn_context: &TurnContext,
call_id: String,
server: String,
@@ -48,11 +58,79 @@ pub(crate) async fn handle_mcp_tool_call(
arguments: arguments_value.clone(),
};
if let Some(decision) =
maybe_request_mcp_tool_approval(sess.as_ref(), turn_context, &call_id, &server, &tool_name)
.await
{
let result = match decision {
McpToolApprovalDecision::Accept => {
let tool_call_begin_event = EventMsg::McpToolCallBegin(McpToolCallBeginEvent {
call_id: call_id.clone(),
invocation: invocation.clone(),
});
notify_mcp_tool_call_event(sess.as_ref(), turn_context, tool_call_begin_event)
.await;
let start = Instant::now();
let result = sess
.call_tool(&server, &tool_name, arguments_value.clone())
.await
.map_err(|e| format!("tool call error: {e:?}"));
if let Err(e) = &result {
tracing::warn!("MCP tool call error: {e:?}");
}
let tool_call_end_event = EventMsg::McpToolCallEnd(McpToolCallEndEvent {
call_id: call_id.clone(),
invocation,
duration: start.elapsed(),
result: result.clone(),
});
notify_mcp_tool_call_event(
sess.as_ref(),
turn_context,
tool_call_end_event.clone(),
)
.await;
result
}
McpToolApprovalDecision::Decline => {
let message = "user rejected MCP tool call".to_string();
notify_mcp_tool_call_skip(
sess.as_ref(),
turn_context,
&call_id,
invocation,
message,
)
.await
}
McpToolApprovalDecision::Cancel => {
let message = "user cancelled MCP tool call".to_string();
notify_mcp_tool_call_skip(
sess.as_ref(),
turn_context,
&call_id,
invocation,
message,
)
.await
}
};
let status = if result.is_ok() { "ok" } else { "error" };
turn_context
.client
.get_otel_manager()
.counter("codex.mcp.call", 1, &[("status", status)]);
return ResponseInputItem::McpToolCallOutput { call_id, result };
}
let tool_call_begin_event = EventMsg::McpToolCallBegin(McpToolCallBeginEvent {
call_id: call_id.clone(),
invocation: invocation.clone(),
});
notify_mcp_tool_call_event(sess, turn_context, tool_call_begin_event).await;
notify_mcp_tool_call_event(sess.as_ref(), turn_context, tool_call_begin_event).await;
let start = Instant::now();
// Perform the tool call.
@@ -70,7 +148,7 @@ pub(crate) async fn handle_mcp_tool_call(
result: result.clone(),
});
notify_mcp_tool_call_event(sess, turn_context, tool_call_end_event.clone()).await;
notify_mcp_tool_call_event(sess.as_ref(), turn_context, tool_call_end_event.clone()).await;
let status = if result.is_ok() { "ok" } else { "error" };
turn_context
@@ -84,3 +162,236 @@ pub(crate) async fn handle_mcp_tool_call(
async fn notify_mcp_tool_call_event(sess: &Session, turn_context: &TurnContext, event: EventMsg) {
sess.send_event(turn_context, event).await;
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum McpToolApprovalDecision {
Accept,
Decline,
Cancel,
}
struct McpToolApprovalMetadata {
annotations: ToolAnnotations,
connector_name: Option<String>,
tool_title: Option<String>,
}
const MCP_TOOL_APPROVAL_QUESTION_ID_PREFIX: &str = "mcp_tool_call_approval";
const MCP_TOOL_APPROVAL_ACCEPT: &str = "Accept";
const MCP_TOOL_APPROVAL_DECLINE: &str = "Decline";
const MCP_TOOL_APPROVAL_CANCEL: &str = "Cancel";
async fn maybe_request_mcp_tool_approval(
sess: &Session,
turn_context: &TurnContext,
call_id: &str,
server: &str,
tool_name: &str,
) -> Option<McpToolApprovalDecision> {
if is_full_access_mode(turn_context) {
return None;
}
if server != CODEX_APPS_MCP_SERVER_NAME {
return None;
}
let metadata = lookup_mcp_tool_metadata(sess, server, tool_name).await?;
if !requires_mcp_tool_approval(&metadata.annotations) {
return None;
}
let question_id = format!("{MCP_TOOL_APPROVAL_QUESTION_ID_PREFIX}_{call_id}");
let question = build_mcp_tool_approval_question(
question_id.clone(),
tool_name,
metadata.tool_title.as_deref(),
metadata.connector_name.as_deref(),
&metadata.annotations,
);
let args = RequestUserInputArgs {
questions: vec![question],
};
let response = sess
.request_user_input(turn_context, call_id.to_string(), args)
.await;
Some(parse_mcp_tool_approval_response(response, &question_id))
}
fn is_full_access_mode(turn_context: &TurnContext) -> bool {
matches!(turn_context.approval_policy, AskForApproval::Never)
&& matches!(
turn_context.sandbox_policy,
SandboxPolicy::DangerFullAccess | SandboxPolicy::ExternalSandbox { .. }
)
}
async fn lookup_mcp_tool_metadata(
sess: &Session,
server: &str,
tool_name: &str,
) -> Option<McpToolApprovalMetadata> {
let tools = sess
.services
.mcp_connection_manager
.read()
.await
.list_all_tools()
.await;
tools.into_values().find_map(|tool_info| {
if tool_info.server_name == server && tool_info.tool_name == tool_name {
tool_info
.tool
.annotations
.map(|annotations| McpToolApprovalMetadata {
annotations,
connector_name: tool_info.connector_name,
tool_title: tool_info.tool.title,
})
} else {
None
}
})
}
fn build_mcp_tool_approval_question(
question_id: String,
tool_name: &str,
tool_title: Option<&str>,
connector_name: Option<&str>,
annotations: &ToolAnnotations,
) -> RequestUserInputQuestion {
let destructive = annotations.destructive_hint == Some(true);
let open_world = annotations.open_world_hint == Some(true);
let reason = match (destructive, open_world) {
(true, true) => "may modify data and access external systems",
(true, false) => "may modify or delete data",
(false, true) => "may access external systems",
(false, false) => "may have side effects",
};
let tool_label = tool_title.unwrap_or(tool_name);
let app_label = connector_name
.map(|name| format!("The {name} app"))
.unwrap_or_else(|| "This app".to_string());
let question = format!(
"{app_label} wants to run the tool \"{tool_label}\", which {reason}. Allow this action?"
);
RequestUserInputQuestion {
id: question_id,
header: "Approve app tool call?".to_string(),
question,
is_other: false,
is_secret: false,
options: Some(vec![
RequestUserInputQuestionOption {
label: MCP_TOOL_APPROVAL_ACCEPT.to_string(),
description: "Run the tool and continue.".to_string(),
},
RequestUserInputQuestionOption {
label: MCP_TOOL_APPROVAL_DECLINE.to_string(),
description: "Decline this tool call and continue.".to_string(),
},
RequestUserInputQuestionOption {
label: MCP_TOOL_APPROVAL_CANCEL.to_string(),
description: "Cancel this tool call".to_string(),
},
]),
}
}
fn parse_mcp_tool_approval_response(
response: Option<RequestUserInputResponse>,
question_id: &str,
) -> McpToolApprovalDecision {
let Some(response) = response else {
return McpToolApprovalDecision::Cancel;
};
let answers = response
.answers
.get(question_id)
.map(|answer| answer.answers.as_slice());
let Some(answers) = answers else {
return McpToolApprovalDecision::Cancel;
};
if answers
.iter()
.any(|answer| answer == MCP_TOOL_APPROVAL_ACCEPT)
{
McpToolApprovalDecision::Accept
} else if answers
.iter()
.any(|answer| answer == MCP_TOOL_APPROVAL_CANCEL)
{
McpToolApprovalDecision::Cancel
} else {
McpToolApprovalDecision::Decline
}
}
fn requires_mcp_tool_approval(annotations: &ToolAnnotations) -> bool {
annotations.read_only_hint == Some(false)
&& (annotations.destructive_hint == Some(true) || annotations.open_world_hint == Some(true))
}
async fn notify_mcp_tool_call_skip(
sess: &Session,
turn_context: &TurnContext,
call_id: &str,
invocation: McpInvocation,
message: String,
) -> Result<mcp_types::CallToolResult, String> {
let tool_call_begin_event = EventMsg::McpToolCallBegin(McpToolCallBeginEvent {
call_id: call_id.to_string(),
invocation: invocation.clone(),
});
notify_mcp_tool_call_event(sess, turn_context, tool_call_begin_event).await;
let tool_call_end_event = EventMsg::McpToolCallEnd(McpToolCallEndEvent {
call_id: call_id.to_string(),
invocation,
duration: Duration::ZERO,
result: Err(message.clone()),
});
notify_mcp_tool_call_event(sess, turn_context, tool_call_end_event).await;
Err(message)
}
#[cfg(test)]
mod tests {
use super::*;
use pretty_assertions::assert_eq;
fn annotations(
read_only: Option<bool>,
destructive: Option<bool>,
open_world: Option<bool>,
) -> ToolAnnotations {
ToolAnnotations {
destructive_hint: destructive,
idempotent_hint: None,
open_world_hint: open_world,
read_only_hint: read_only,
title: None,
}
}
#[test]
fn approval_required_when_read_only_false_and_destructive() {
let annotations = annotations(Some(false), Some(true), None);
assert_eq!(requires_mcp_tool_approval(&annotations), true);
}
#[test]
fn approval_required_when_read_only_false_and_open_world() {
let annotations = annotations(Some(false), None, Some(true));
assert_eq!(requires_mcp_tool_approval(&annotations), true);
}
#[test]
fn approval_not_required_when_read_only_true() {
let annotations = annotations(Some(true), Some(true), Some(true));
assert_eq!(requires_mcp_tool_approval(&annotations), false);
}
}
codex-rs/core/src/models_manager/model_info.rs
+28 -27
View File
@@ -1,19 +1,17 @@
use std::collections::BTreeMap;
use codex_protocol::config_types::Personality;
use codex_protocol::config_types::Verbosity;
use codex_protocol::openai_models::ApplyPatchToolType;
use codex_protocol::openai_models::ConfigShellToolType;
use codex_protocol::openai_models::ModelInfo;
use codex_protocol::openai_models::ModelInstructionsTemplate;
use codex_protocol::openai_models::ModelInstructionsVariables;
use codex_protocol::openai_models::ModelMessages;
use codex_protocol::openai_models::ModelVisibility;
use codex_protocol::openai_models::PersonalityMessages;
use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::openai_models::ReasoningEffortPreset;
use codex_protocol::openai_models::TruncationMode;
use codex_protocol::openai_models::TruncationPolicyConfig;
use crate::config::Config;
use crate::features::Feature;
use crate::truncate::approx_bytes_for_tokens;
use tracing::warn;
@@ -29,8 +27,11 @@ const GPT_5_1_CODEX_MAX_INSTRUCTIONS: &str = include_str!("../../gpt-5.1-codex-m
const GPT_5_2_CODEX_INSTRUCTIONS: &str = include_str!("../../gpt-5.2-codex_prompt.md");
const GPT_5_2_CODEX_INSTRUCTIONS_TEMPLATE: &str =
include_str!("../../templates/model_instructions/gpt-5.2-codex_instructions_template.md");
const PERSONALITY_FRIENDLY: &str = include_str!("../../templates/personalities/friendly.md");
const PERSONALITY_PRAGMATIC: &str = include_str!("../../templates/personalities/pragmatic.md");
const GPT_5_2_CODEX_PERSONALITY_FRIENDLY: &str =
include_str!("../../templates/personalities/gpt-5.2-codex_friendly.md");
const GPT_5_2_CODEX_PERSONALITY_PRAGMATIC: &str =
include_str!("../../templates/personalities/gpt-5.2-codex_pragmatic.md");
pub(crate) const CONTEXT_WINDOW_272K: i64 = 272_000;
@@ -54,7 +55,7 @@ macro_rules! model_info {
priority: 99,
upgrade: None,
base_instructions: BASE_INSTRUCTIONS.to_string(),
model_instructions_template: None,
model_messages: None,
supports_reasoning_summaries: false,
support_verbosity: false,
default_verbosity: None,
@@ -100,8 +101,11 @@ pub(crate) fn with_config_overrides(mut model: ModelInfo, config: &Config) -> Mo
if let Some(base_instructions) = &config.base_instructions {
model.base_instructions = base_instructions.clone();
model.model_instructions_template = None;
model.model_messages = None;
} else if !config.features.enabled(Feature::Personality) {
model.model_messages = None;
}
model
}
@@ -169,15 +173,13 @@ pub(crate) fn find_model_info_for_slug(slug: &str) -> ModelInfo {
model_info!(
slug,
base_instructions: GPT_5_2_CODEX_INSTRUCTIONS.to_string(),
model_instructions_template: Some(ModelInstructionsTemplate {
template: GPT_5_2_CODEX_INSTRUCTIONS_TEMPLATE.to_string(),
personality_messages: Some(PersonalityMessages(BTreeMap::from([(
Personality::Friendly,
PERSONALITY_FRIENDLY.to_string(),
), (
Personality::Pragmatic,
PERSONALITY_PRAGMATIC.to_string(),
)]))),
model_messages: Some(ModelMessages {
instructions_template: Some(GPT_5_2_CODEX_INSTRUCTIONS_TEMPLATE.to_string()),
instructions_variables: Some(ModelInstructionsVariables {
personality_default: Some("".to_string()),
personality_friendly: Some(GPT_5_2_CODEX_PERSONALITY_FRIENDLY.to_string()),
personality_pragmatic: Some(GPT_5_2_CODEX_PERSONALITY_PRAGMATIC.to_string()),
}),
}),
apply_patch_tool_type: Some(ApplyPatchToolType::Freeform),
shell_type: ConfigShellToolType::ShellCommand,
@@ -213,15 +215,14 @@ pub(crate) fn find_model_info_for_slug(slug: &str) -> ModelInfo {
truncation_policy: TruncationPolicyConfig::tokens(10_000),
context_window: Some(CONTEXT_WINDOW_272K),
supported_reasoning_levels: supported_reasoning_level_low_medium_high_xhigh(),
model_instructions_template: Some(ModelInstructionsTemplate {
template: GPT_5_2_CODEX_INSTRUCTIONS_TEMPLATE.to_string(),
personality_messages: Some(PersonalityMessages(BTreeMap::from([(
Personality::Friendly,
PERSONALITY_FRIENDLY.to_string(),
), (
Personality::Pragmatic,
PERSONALITY_PRAGMATIC.to_string(),
)]))),
base_instructions: GPT_5_2_CODEX_INSTRUCTIONS.to_string(),
model_messages: Some(ModelMessages {
instructions_template: Some(GPT_5_2_CODEX_INSTRUCTIONS_TEMPLATE.to_string()),
instructions_variables: Some(ModelInstructionsVariables {
personality_default: Some("".to_string()),
personality_friendly: Some(GPT_5_2_CODEX_PERSONALITY_FRIENDLY.to_string()),
personality_pragmatic: Some(GPT_5_2_CODEX_PERSONALITY_PRAGMATIC.to_string()),
}),
}),
)
} else if slug.starts_with("gpt-5.1-codex-max") {
codex-rs/core/src/rollout/metadata.rs
+12
View File
@@ -16,6 +16,7 @@ use codex_protocol::protocol::SessionSource;
use codex_state::BackfillStats;
use codex_state::DB_ERROR_METRIC;
use codex_state::DB_METRIC_BACKFILL;
use codex_state::DB_METRIC_BACKFILL_DURATION_MS;
use codex_state::ExtractionOutcome;
use codex_state::ThreadMetadataBuilder;
use codex_state::apply_rollout_item;
@@ -128,6 +129,7 @@ pub(crate) async fn backfill_sessions(
config: &Config,
otel: Option<&OtelManager>,
) {
let timer = otel.and_then(|otel| otel.start_timer(DB_METRIC_BACKFILL_DURATION_MS, &[]).ok());
let sessions_root = config.codex_home.join(rollout::SESSIONS_SUBDIR);
let archived_root = config.codex_home.join(rollout::ARCHIVED_SESSIONS_SUBDIR);
let mut rollout_paths: Vec<(PathBuf, bool)> = Vec::new();
@@ -210,6 +212,16 @@ pub(crate) async fn backfill_sessions(
&[("status", "failed")],
);
}
if let Some(timer) = timer.as_ref() {
let status = if stats.failed == 0 {
"success"
} else if stats.upserted == 0 {
"failed"
} else {
"partial_failure"
};
let _ = timer.record(&[("status", status)]);
}
}
async fn file_modified_time_utc(path: &Path) -> Option<DateTime<Utc>> {
codex-rs/core/src/rollout/mod.rs
+2
View File
@@ -12,6 +12,7 @@ pub mod list;
pub(crate) mod metadata;
pub(crate) mod policy;
pub mod recorder;
pub(crate) mod session_index;
pub(crate) mod truncation;
pub use codex_protocol::protocol::SessionMeta;
@@ -23,6 +24,7 @@ pub use list::find_thread_path_by_id_str as find_conversation_path_by_id_str;
pub use list::rollout_date_parts;
pub use recorder::RolloutRecorder;
pub use recorder::RolloutRecorderParams;
pub use session_index::find_thread_path_by_name_str;
#[cfg(test)]
pub mod tests;
codex-rs/core/src/rollout/policy.rs
+1
View File
@@ -58,6 +58,7 @@ pub(crate) fn should_persist_event_msg(ev: &EventMsg) -> bool {
| EventMsg::AgentReasoningSectionBreak(_)
| EventMsg::RawResponseItem(_)
| EventMsg::SessionConfigured(_)
| EventMsg::ThreadNameUpdated(_)
| EventMsg::McpToolCallBegin(_)
| EventMsg::McpToolCallEnd(_)
| EventMsg::WebSearchBegin(_)
codex-rs/core/src/rollout/recorder.rs
+2 -2
View File
@@ -428,13 +428,13 @@ impl RolloutRecorder {
}
},
Err(e) => {
warn!("failed to parse rollout line: {v:?}, error: {e}");
warn!("failed to parse rollout line: {e}");
parse_errors = parse_errors.saturating_add(1);
}
}
}
info!(
tracing::debug!(
"Resumed rollout with {} items, thread ID: {:?}, parse errors: {}",
items.len(),
thread_id,
codex-rs/core/src/rollout/session_index.rs
+325
View File
@@ -0,0 +1,325 @@
use std::fs::File;
use std::io::Read;
use std::io::Seek;
use std::io::SeekFrom;
use std::path::Path;
use std::path::PathBuf;
use codex_protocol::ThreadId;
use serde::Deserialize;
use serde::Serialize;
use tokio::io::AsyncWriteExt;
const SESSION_INDEX_FILE: &str = "session_index.jsonl";
const READ_CHUNK_SIZE: usize = 8192;
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct SessionIndexEntry {
pub id: ThreadId,
pub thread_name: String,
pub updated_at: String,
}
/// Append a thread name update to the session index.
/// The index is append-only; the most recent entry wins when resolving names or ids.
pub async fn append_thread_name(
codex_home: &Path,
thread_id: ThreadId,
name: &str,
) -> std::io::Result<()> {
use time::OffsetDateTime;
use time::format_description::well_known::Rfc3339;
let updated_at = OffsetDateTime::now_utc()
.format(&Rfc3339)
.unwrap_or_else(|_| "unknown".to_string());
let entry = SessionIndexEntry {
id: thread_id,
thread_name: name.to_string(),
updated_at,
};
append_session_index_entry(codex_home, &entry).await
}
/// Append a raw session index entry to `session_index.jsonl`.
/// The file is append-only; consumers scan from the end to find the newest match.
pub async fn append_session_index_entry(
codex_home: &Path,
entry: &SessionIndexEntry,
) -> std::io::Result<()> {
let path = session_index_path(codex_home);
let mut file = tokio::fs::OpenOptions::new()
.create(true)
.append(true)
.open(&path)
.await?;
let mut line = serde_json::to_string(entry).map_err(std::io::Error::other)?;
line.push('\n');
file.write_all(line.as_bytes()).await?;
file.flush().await?;
Ok(())
}
/// Find the latest thread name for a thread id, if any.
pub async fn find_thread_name_by_id(
codex_home: &Path,
thread_id: &ThreadId,
) -> std::io::Result<Option<String>> {
let path = session_index_path(codex_home);
if !path.exists() {
return Ok(None);
}
let id = *thread_id;
let entry = tokio::task::spawn_blocking(move || scan_index_from_end_by_id(&path, &id))
.await
.map_err(std::io::Error::other)??;
Ok(entry.map(|entry| entry.thread_name))
}
/// Find the most recently updated thread id for a thread name, if any.
pub async fn find_thread_id_by_name(
codex_home: &Path,
name: &str,
) -> std::io::Result<Option<ThreadId>> {
if name.trim().is_empty() {
return Ok(None);
}
let path = session_index_path(codex_home);
if !path.exists() {
return Ok(None);
}
let name = name.to_string();
let entry = tokio::task::spawn_blocking(move || scan_index_from_end_by_name(&path, &name))
.await
.map_err(std::io::Error::other)??;
Ok(entry.map(|entry| entry.id))
}
/// Locate a recorded thread rollout file by thread name using newest-first ordering.
/// Returns `Ok(Some(path))` if found, `Ok(None)` if not present.
pub async fn find_thread_path_by_name_str(
codex_home: &Path,
name: &str,
) -> std::io::Result<Option<PathBuf>> {
let Some(thread_id) = find_thread_id_by_name(codex_home, name).await? else {
return Ok(None);
};
super::list::find_thread_path_by_id_str(codex_home, &thread_id.to_string()).await
}
fn session_index_path(codex_home: &Path) -> PathBuf {
codex_home.join(SESSION_INDEX_FILE)
}
fn scan_index_from_end_by_id(
path: &Path,
thread_id: &ThreadId,
) -> std::io::Result<Option<SessionIndexEntry>> {
scan_index_from_end(path, |entry| entry.id == *thread_id)
}
fn scan_index_from_end_by_name(
path: &Path,
name: &str,
) -> std::io::Result<Option<SessionIndexEntry>> {
scan_index_from_end(path, |entry| entry.thread_name == name)
}
fn scan_index_from_end<F>(
path: &Path,
mut predicate: F,
) -> std::io::Result<Option<SessionIndexEntry>>
where
F: FnMut(&SessionIndexEntry) -> bool,
{
let mut file = File::open(path)?;
let mut remaining = file.metadata()?.len();
let mut line_rev: Vec<u8> = Vec::new();
let mut buf = vec![0u8; READ_CHUNK_SIZE];
while remaining > 0 {
let read_size = usize::try_from(remaining.min(READ_CHUNK_SIZE as u64))
.map_err(std::io::Error::other)?;
remaining -= read_size as u64;
file.seek(SeekFrom::Start(remaining))?;
file.read_exact(&mut buf[..read_size])?;
for &byte in buf[..read_size].iter().rev() {
if byte == b'\n' {
if let Some(entry) = parse_line_from_rev(&mut line_rev, &mut predicate)? {
return Ok(Some(entry));
}
continue;
}
line_rev.push(byte);
}
}
if let Some(entry) = parse_line_from_rev(&mut line_rev, &mut predicate)? {
return Ok(Some(entry));
}
Ok(None)
}
fn parse_line_from_rev<F>(
line_rev: &mut Vec<u8>,
predicate: &mut F,
) -> std::io::Result<Option<SessionIndexEntry>>
where
F: FnMut(&SessionIndexEntry) -> bool,
{
if line_rev.is_empty() {
return Ok(None);
}
line_rev.reverse();
let line = std::mem::take(line_rev);
let Ok(mut line) = String::from_utf8(line) else {
return Ok(None);
};
if line.ends_with('\r') {
line.pop();
}
let trimmed = line.trim();
if trimmed.is_empty() {
return Ok(None);
}
let Ok(entry) = serde_json::from_str::<SessionIndexEntry>(trimmed) else {
return Ok(None);
};
if predicate(&entry) {
return Ok(Some(entry));
}
Ok(None)
}
#[cfg(test)]
mod tests {
use super::*;
use pretty_assertions::assert_eq;
use tempfile::TempDir;
fn write_index(path: &Path, lines: &[SessionIndexEntry]) -> std::io::Result<()> {
let mut out = String::new();
for entry in lines {
out.push_str(&serde_json::to_string(entry).unwrap());
out.push('\n');
}
std::fs::write(path, out)
}
#[test]
fn find_thread_id_by_name_prefers_latest_entry() -> std::io::Result<()> {
let temp = TempDir::new()?;
let path = session_index_path(temp.path());
let id1 = ThreadId::new();
let id2 = ThreadId::new();
let lines = vec![
SessionIndexEntry {
id: id1,
thread_name: "same".to_string(),
updated_at: "2024-01-01T00:00:00Z".to_string(),
},
SessionIndexEntry {
id: id2,
thread_name: "same".to_string(),
updated_at: "2024-01-02T00:00:00Z".to_string(),
},
];
write_index(&path, &lines)?;
let found = scan_index_from_end_by_name(&path, "same")?;
assert_eq!(found.map(|entry| entry.id), Some(id2));
Ok(())
}
#[test]
fn find_thread_name_by_id_prefers_latest_entry() -> std::io::Result<()> {
let temp = TempDir::new()?;
let path = session_index_path(temp.path());
let id = ThreadId::new();
let lines = vec![
SessionIndexEntry {
id,
thread_name: "first".to_string(),
updated_at: "2024-01-01T00:00:00Z".to_string(),
},
SessionIndexEntry {
id,
thread_name: "second".to_string(),
updated_at: "2024-01-02T00:00:00Z".to_string(),
},
];
write_index(&path, &lines)?;
let found = scan_index_from_end_by_id(&path, &id)?;
assert_eq!(
found.map(|entry| entry.thread_name),
Some("second".to_string())
);
Ok(())
}
#[test]
fn scan_index_returns_none_when_entry_missing() -> std::io::Result<()> {
let temp = TempDir::new()?;
let path = session_index_path(temp.path());
let id = ThreadId::new();
let lines = vec![SessionIndexEntry {
id,
thread_name: "present".to_string(),
updated_at: "2024-01-01T00:00:00Z".to_string(),
}];
write_index(&path, &lines)?;
let missing_name = scan_index_from_end_by_name(&path, "missing")?;
assert_eq!(missing_name, None);
let missing_id = scan_index_from_end_by_id(&path, &ThreadId::new())?;
assert_eq!(missing_id, None);
Ok(())
}
#[test]
fn scan_index_finds_latest_match_among_mixed_entries() -> std::io::Result<()> {
let temp = TempDir::new()?;
let path = session_index_path(temp.path());
let id_target = ThreadId::new();
let id_other = ThreadId::new();
let expected = SessionIndexEntry {
id: id_target,
thread_name: "target".to_string(),
updated_at: "2024-01-03T00:00:00Z".to_string(),
};
let expected_other = SessionIndexEntry {
id: id_other,
thread_name: "target".to_string(),
updated_at: "2024-01-02T00:00:00Z".to_string(),
};
// Resolution is based on append order (scan from end), not updated_at.
let lines = vec![
SessionIndexEntry {
id: id_target,
thread_name: "target".to_string(),
updated_at: "2024-01-01T00:00:00Z".to_string(),
},
expected_other.clone(),
expected.clone(),
SessionIndexEntry {
id: ThreadId::new(),
thread_name: "another".to_string(),
updated_at: "2024-01-04T00:00:00Z".to_string(),
},
];
write_index(&path, &lines)?;
let found_by_name = scan_index_from_end_by_name(&path, "target")?;
assert_eq!(found_by_name, Some(expected.clone()));
let found_by_id = scan_index_from_end_by_id(&path, &id_target)?;
assert_eq!(found_by_id, Some(expected));
let found_other_by_id = scan_index_from_end_by_id(&path, &id_other)?;
assert_eq!(found_other_by_id, Some(expected_other));
Ok(())
}
}
codex-rs/core/src/shell.rs
+1
View File
@@ -137,6 +137,7 @@ fn get_shell_path(
let default_shell_path = get_user_shell_path();
if let Some(default_shell_path) = default_shell_path
&& detect_shell_type(&default_shell_path) == Some(shell_type)
&& file_exists(&default_shell_path).is_some()
{
return Some(default_shell_path);
}
codex-rs/core/src/skills/manager.rs
+1
View File
@@ -88,6 +88,7 @@ impl SkillsManager {
Some(cwd_abs),
&cli_overrides,
LoaderOverrides::default(),
None,
)
.await
{
codex-rs/core/src/state_db.rs
+3 -7
View File
@@ -11,6 +11,7 @@ use codex_otel::OtelManager;
use codex_protocol::ThreadId;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::SessionSource;
use codex_state::DB_METRIC_COMPARE_ERROR;
pub use codex_state::LogEntry;
use codex_state::STATE_DB_FILENAME;
use codex_state::ThreadMetadataBuilder;
@@ -32,12 +33,6 @@ pub(crate) async fn init_if_enabled(
) -> Option<StateDbHandle> {
let state_path = config.codex_home.join(STATE_DB_FILENAME);
if !config.features.enabled(Feature::Sqlite) {
// We delete the file on best effort basis to maintain retro-compatibility in the future.
let wal_path = state_path.with_extension("sqlite-wal");
let shm_path = state_path.with_extension("sqlite-shm");
for path in [state_path.as_path(), wal_path.as_path(), shm_path.as_path()] {
tokio::fs::remove_file(path).await.ok();
}
return None;
}
let existed = tokio::fs::try_exists(&state_path).await.unwrap_or(false);
@@ -282,9 +277,10 @@ pub async fn apply_rollout_items(
pub fn record_discrepancy(stage: &str, reason: &str) {
// We access the global metric because the call sites might not have access to the broader
// OtelManager.
tracing::warn!("state db record_discrepancy: {stage}{reason}");
if let Some(metric) = codex_otel::metrics::global() {
let _ = metric.counter(
"codex.db.discrepancy",
DB_METRIC_COMPARE_ERROR,
1,
&[("stage", stage), ("reason", reason)],
);
codex-rs/core/src/tools/handlers/mcp.rs
+2 -1
View File
@@ -1,4 +1,5 @@
use async_trait::async_trait;
use std::sync::Arc;
use crate::function_tool::FunctionCallError;
use crate::mcp_tool_call::handle_mcp_tool_call;
@@ -42,7 +43,7 @@ impl ToolHandler for McpHandler {
let arguments_str = raw_arguments;
let response = handle_mcp_tool_call(
session.as_ref(),
Arc::clone(&session),
turn.as_ref(),
call_id.clone(),
server,
codex-rs/core/src/tools/handlers/request_user_input.rs
+13 -1
View File
@@ -49,7 +49,19 @@ impl ToolHandler for RequestUserInputHandler {
)));
}
let args: RequestUserInputArgs = parse_arguments(&arguments)?;
let mut args: RequestUserInputArgs = parse_arguments(&arguments)?;
let missing_options = args
.questions
.iter()
.any(|question| question.options.as_ref().is_none_or(Vec::is_empty));
if missing_options {
return Err(FunctionCallError::RespondToModel(
"request_user_input requires non-empty options for every question".to_string(),
));
}
for question in &mut args.questions {
question.is_other = true;
}
let response = session
.request_user_input(turn.as_ref(), call_id, args)
.await
codex-rs/core/src/tools/spec.rs
+2 -11
View File
@@ -571,7 +571,7 @@ fn create_request_user_input_tool() -> ToolSpec {
let options_schema = JsonSchema::Array {
description: Some(
"Optional 2-3 mutually exclusive choices. Put the recommended option first and suffix its label with \"(Recommended)\". Do not include an \"Other\" option in this list; use isOther on the question to request a free form choice. If the question is free form in nature, please do not have any option."
"Provide 2-3 mutually exclusive choices. Put the recommended option first and suffix its label with \"(Recommended)\". Do not include an \"Other\" option in this list; the client will add a free-form \"Other\" option automatically."
.to_string(),
),
items: Box::new(JsonSchema::Object {
@@ -602,15 +602,6 @@ fn create_request_user_input_tool() -> ToolSpec {
description: Some("Single-sentence prompt shown to the user.".to_string()),
},
);
question_props.insert(
"isOther".to_string(),
JsonSchema::Boolean {
description: Some(
"True when this question should include a free-form \"Other\" option. Otherwise false."
.to_string(),
),
},
);
question_props.insert("options".to_string(), options_schema);
let questions_schema = JsonSchema::Array {
@@ -621,7 +612,7 @@ fn create_request_user_input_tool() -> ToolSpec {
"id".to_string(),
"header".to_string(),
"question".to_string(),
"isOther".to_string(),
"options".to_string(),
]),
additional_properties: Some(false.into()),
}),
codex-rs/core/src/util.rs
+76
View File
@@ -2,10 +2,13 @@ use std::path::Path;
use std::path::PathBuf;
use std::time::Duration;
use codex_protocol::ThreadId;
use rand::Rng;
use tracing::debug;
use tracing::error;
use crate::parse_command::shlex_join;
const INITIAL_DELAY_MS: u64 = 200;
const BACKOFF_FACTOR: f64 = 2.0;
@@ -72,6 +75,32 @@ pub fn resolve_path(base: &Path, path: &PathBuf) -> PathBuf {
}
}
/// Trim a thread name and return `None` if it is empty after trimming.
pub fn normalize_thread_name(name: &str) -> Option<String> {
let trimmed = name.trim();
if trimmed.is_empty() {
None
} else {
Some(trimmed.to_string())
}
}
pub fn resume_command(thread_name: Option<&str>, thread_id: Option<ThreadId>) -> Option<String> {
let resume_target = thread_name
.filter(|name| !name.is_empty())
.map(str::to_string)
.or_else(|| thread_id.map(|thread_id| thread_id.to_string()));
resume_target.map(|target| {
let needs_double_dash = target.starts_with('-');
let escaped = shlex_join(&[target]);
if needs_double_dash {
format!("codex resume -- {escaped}")
} else {
format!("codex resume {escaped}")
}
})
}
#[cfg(test)]
mod tests {
use super::*;
@@ -107,4 +136,51 @@ mod tests {
feedback_tags!(model = "gpt-5", cached = true, debug_only = OnlyDebug);
}
#[test]
fn normalize_thread_name_trims_and_rejects_empty() {
assert_eq!(normalize_thread_name(" "), None);
assert_eq!(
normalize_thread_name(" my thread "),
Some("my thread".to_string())
);
}
#[test]
fn resume_command_prefers_name_over_id() {
let thread_id = ThreadId::from_string("123e4567-e89b-12d3-a456-426614174000").unwrap();
let command = resume_command(Some("my-thread"), Some(thread_id));
assert_eq!(command, Some("codex resume my-thread".to_string()));
}
#[test]
fn resume_command_with_only_id() {
let thread_id = ThreadId::from_string("123e4567-e89b-12d3-a456-426614174000").unwrap();
let command = resume_command(None, Some(thread_id));
assert_eq!(
command,
Some("codex resume 123e4567-e89b-12d3-a456-426614174000".to_string())
);
}
#[test]
fn resume_command_with_no_name_or_id() {
let command = resume_command(None, None);
assert_eq!(command, None);
}
#[test]
fn resume_command_quotes_thread_name_when_needed() {
let command = resume_command(Some("-starts-with-dash"), None);
assert_eq!(
command,
Some("codex resume -- -starts-with-dash".to_string())
);
let command = resume_command(Some("two words"), None);
assert_eq!(command, Some("codex resume 'two words'".to_string()));
let command = resume_command(Some("quote'case"), None);
assert_eq!(command, Some("codex resume \"quote'case\"".to_string()));
}
}