diff --git a/codex-rs/codex-home/src/instructions/mod.rs b/codex-rs/codex-home/src/instructions/mod.rs index 0840263a13f9..4f4ea765255a 100644 --- a/codex-rs/codex-home/src/instructions/mod.rs +++ b/codex-rs/codex-home/src/instructions/mod.rs @@ -48,12 +48,6 @@ impl CodexHomeUserInstructionsProvider { continue; } }; - if let Err(err) = std::str::from_utf8(&data) { - warnings.push(format!( - "Global AGENTS.md instructions from `{}` contain invalid UTF-8: {err}. Invalid byte sequences were replaced.", - path.display() - )); - } let contents = String::from_utf8_lossy(&data); let trimmed = contents.trim(); if !trimmed.is_empty() { diff --git a/codex-rs/codex-home/src/instructions/tests.rs b/codex-rs/codex-home/src/instructions/tests.rs index e88421040320..ee2ba9035c4c 100644 --- a/codex-rs/codex-home/src/instructions/tests.rs +++ b/codex-rs/codex-home/src/instructions/tests.rs @@ -126,7 +126,7 @@ async fn recoverable_override_read_error_warns_and_falls_back_to_default() { } #[tokio::test] -async fn invalid_utf8_is_lossy_and_warned() { +async fn invalid_utf8_is_lossy() { let home = TempDir::new().expect("temp dir"); let path = home.path().join(DEFAULT_AGENTS_MD_FILENAME); let mut invalid_utf8 = b"global".to_vec(); @@ -135,18 +135,13 @@ async fn invalid_utf8_is_lossy_and_warned() { fs::write(&path, &invalid_utf8).expect("write invalid utf-8"); let outcome = provider(&home).load_user_instructions().await; - let utf8_error = std::str::from_utf8(&invalid_utf8).expect_err("invalid utf-8"); - let warning = format!( - "Global AGENTS.md instructions from `{}` contain invalid UTF-8: {utf8_error}. Invalid byte sequences were replaced.", - path.display(), - ); assert_eq!( outcome, expected( &home, DEFAULT_AGENTS_MD_FILENAME, "global\u{fffd} doc", - vec![warning] + Vec::new() ) ); } diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index bd982b3c6f93..5f1246bf5b28 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -6,10 +6,10 @@ use crate::agent::role::resolve_role_config; use crate::agent::status::is_final; use crate::codex_thread::ThreadConfigSnapshot; use crate::config::Config; +use crate::environment_selection::TurnEnvironmentSnapshot; use crate::session::emit_subagent_session_started; use crate::session_prefix::format_subagent_context_line; use crate::session_prefix::format_subagent_notification_message; -use crate::shell_snapshot::ShellSnapshot; use crate::thread_manager::ResumeThreadWithHistoryOptions; use crate::thread_manager::ThreadManagerState; use crate::thread_rollout_truncation::truncate_rollout_to_last_n_fork_turns; @@ -519,11 +519,11 @@ impl AgentControl { .ok_or_else(|| CodexErr::UnsupportedOperation("thread manager dropped".to_string())) } - async fn inherited_shell_snapshot_for_source( + async fn inherited_environments_for_source( &self, state: &Arc, session_source: Option<&SessionSource>, - ) -> Option> { + ) -> Option { let Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, .. })) = session_source @@ -532,13 +532,15 @@ impl AgentControl { }; let parent_thread = state.get_thread(*parent_thread_id).await.ok()?; - let snapshot = parent_thread - .codex - .session - .services - .shell_snapshot - .load_full()?; - (!snapshot.is_failed()).then_some(snapshot) + Some( + parent_thread + .codex + .session + .services + .turn_environments + .snapshot() + .await, + ) } async fn inherited_exec_policy_for_source( diff --git a/codex-rs/core/src/agent/control/residency_tests.rs b/codex-rs/core/src/agent/control/residency_tests.rs index 88d4b004d33e..cf043971e269 100644 --- a/codex-rs/core/src/agent/control/residency_tests.rs +++ b/codex-rs/core/src/agent/control/residency_tests.rs @@ -142,7 +142,7 @@ async fn spawn_v2_subagent( /*forked_from_thread_id*/ None, Some(ThreadSource::Subagent), /*metrics_service_name*/ None, - /*inherited_shell_snapshot*/ None, + /*inherited_environments*/ None, /*inherited_exec_policy*/ None, /*environments*/ None, ) diff --git a/codex-rs/core/src/agent/control/spawn.rs b/codex-rs/core/src/agent/control/spawn.rs index 9c12d60e6abf..bc6dba1a48cf 100644 --- a/codex-rs/core/src/agent/control/spawn.rs +++ b/codex-rs/core/src/agent/control/spawn.rs @@ -4,7 +4,7 @@ use super::*; const AGENT_NAMES: &str = include_str!("../agent_names.txt"); struct SpawnAgentThreadInheritance { - shell_snapshot: Option>, + environments: Option, exec_policy: Option>, } @@ -156,8 +156,8 @@ impl AgentControl { let parent_thread_id = initial_history .get_resumed_parent_thread_id() .or(stored_parent_thread_id); - let inherited_shell_snapshot = self - .inherited_shell_snapshot_for_source(&state, Some(&session_source)) + let inherited_environments = self + .inherited_environments_for_source(&state, Some(&session_source)) .await; let inherited_exec_policy = self .inherited_exec_policy_for_source(&state, Some(&session_source), &config) @@ -170,7 +170,7 @@ impl AgentControl { agent_control: self.clone(), session_source, parent_thread_id, - inherited_shell_snapshot, + inherited_environments, inherited_exec_policy, }) .await @@ -231,8 +231,8 @@ impl AgentControl { }; let mut reservation = self.state.reserve_spawn_slot(reservation_max_threads)?; let inheritance = SpawnAgentThreadInheritance { - shell_snapshot: self - .inherited_shell_snapshot_for_source(&state, session_source.as_ref()) + environments: self + .inherited_environments_for_source(&state, session_source.as_ref()) .await, exec_policy: self .inherited_exec_policy_for_source(&state, session_source.as_ref(), &config) @@ -283,7 +283,7 @@ impl AgentControl { /*forked_from_thread_id*/ None, /*thread_source*/ Some(ThreadSource::Subagent), /*metrics_service_name*/ None, - inheritance.shell_snapshot, + inheritance.environments, inheritance.exec_policy, options.environments.clone(), )) @@ -386,7 +386,7 @@ impl AgentControl { multi_agent_version: MultiAgentVersion, ) -> CodexResult { let SpawnAgentThreadInheritance { - shell_snapshot: inherited_shell_snapshot, + environments: inherited_environments, exec_policy: inherited_exec_policy, } = inheritance; if options.fork_parent_spawn_call_id.is_none() { @@ -513,7 +513,7 @@ impl AgentControl { /*thread_source*/ Some(ThreadSource::Subagent), /*parent_thread_id*/ Some(parent_thread_id), /*forked_from_thread_id*/ Some(parent_thread_id), - inherited_shell_snapshot, + inherited_environments, inherited_exec_policy, options.environments.clone(), ) @@ -665,8 +665,8 @@ impl AgentControl { other => (other, AgentMetadata::default()), }; let notification_source = session_source.clone(); - let inherited_shell_snapshot = self - .inherited_shell_snapshot_for_source(&state, Some(&session_source)) + let inherited_environments = self + .inherited_environments_for_source(&state, Some(&session_source)) .await; let inherited_exec_policy = self .inherited_exec_policy_for_source(&state, Some(&session_source), &config) @@ -679,7 +679,7 @@ impl AgentControl { agent_control: self.clone(), session_source, parent_thread_id, - inherited_shell_snapshot, + inherited_environments, inherited_exec_policy, }) .await?; diff --git a/codex-rs/core/src/agents_md.rs b/codex-rs/core/src/agents_md.rs index 52a386d694e9..5cb70dc5fc4e 100644 --- a/codex-rs/core/src/agents_md.rs +++ b/codex-rs/core/src/agents_md.rs @@ -46,7 +46,7 @@ const AGENTS_MD_SEPARATOR: &str = "\n\n--- project-doc ---\n\n"; /// Loads project AGENTS.md content and combines it with host-provided user /// instructions. pub(crate) async fn load_project_instructions( - config: &mut Config, + config: &Config, user_instructions: Option, environments: &TurnEnvironmentSnapshot, ) -> Option { @@ -89,7 +89,7 @@ pub(crate) async fn load_project_instructions( /// `Ok(None)`. Unexpected I/O failures bubble up as `Err` so callers can /// decide how to handle them. async fn read_agents_md( - config: &mut Config, + config: &Config, fs: &dyn ExecutorFileSystem, environment_id: &str, cwd: &AbsolutePathBuf, @@ -126,8 +126,6 @@ async fn read_agents_md( Err(err) if err.kind() == io::ErrorKind::NotFound => continue, Err(err) => return Err(err), }; - warn_invalid_utf8(&p, &data, "Project", &mut config.startup_warnings); - let size = data.len() as u64; if size > remaining { data.truncate(remaining as usize); @@ -503,20 +501,6 @@ impl InstructionProvenance { } } -fn warn_invalid_utf8( - path: &AbsolutePathBuf, - data: &[u8], - source: &str, - startup_warnings: &mut Vec, -) { - if let Err(err) = std::str::from_utf8(data) { - startup_warnings.push(format!( - "{source} AGENTS.md instructions from `{}` contain invalid UTF-8: {err}. Invalid byte sequences were replaced.", - path.display() - )); - } -} - #[cfg(test)] #[path = "agents_md_tests.rs"] mod tests; diff --git a/codex-rs/core/src/agents_md_tests.rs b/codex-rs/core/src/agents_md_tests.rs index 565b7f4b999a..92cb0d5bebfa 100644 --- a/codex-rs/core/src/agents_md_tests.rs +++ b/codex-rs/core/src/agents_md_tests.rs @@ -27,7 +27,6 @@ use std::fs; use std::io; use std::ops::Deref; use std::ops::DerefMut; -use std::path::Path; use std::path::PathBuf; use std::sync::Arc; use tempfile::TempDir; @@ -223,29 +222,18 @@ impl DerefMut for TestConfig { } async fn get_user_instructions(config: &TestConfig) -> Option { - let mut warnings = Vec::new(); - load_agents_md(config, &mut warnings) - .await - .map(|loaded| loaded.text()) + load_agents_md(config).await.map(|loaded| loaded.text()) } -async fn load_agents_md(config: &TestConfig, warnings: &mut Vec) -> Option { - let mut core_config = config.config.clone(); - let existing_warning_count = core_config.startup_warnings.len(); - let environments = resolved_local_environments([("local", core_config.cwd.clone())]); - let loaded = load_project_instructions( - &mut core_config, +async fn load_agents_md(config: &TestConfig) -> Option { + let environments = resolved_local_environments([("local", config.config.cwd.clone())]); + + load_project_instructions( + &config.config, config.user_instructions.clone(), &environments, ) - .await; - warnings.extend( - core_config - .startup_warnings - .into_iter() - .skip(existing_warning_count), - ); - loaded + .await } async fn agents_md_paths(config: &TestConfig) -> std::io::Result> { @@ -281,19 +269,6 @@ fn project_provenance(path: AbsolutePathBuf, cwd: AbsolutePathBuf) -> Instructio } } -fn assert_invalid_utf8_warning(warnings: &[String], source: &str, path: &Path) { - let path_display = path.display().to_string(); - assert_eq!(warnings.len(), 1, "expected one warning, got {warnings:?}"); - let warning = &warnings[0]; - assert!( - warning.contains(&format!("{source} AGENTS.md instructions")) - && warning.contains(&path_display) - && warning.contains("invalid UTF-8") - && warning.contains("Invalid byte sequences were replaced."), - "unexpected invalid UTF-8 warning: {warning:?}" - ); -} - /// Helper that returns a `Config` pointing at `root` and using `limit` as /// the maximum number of bytes to embed from AGENTS.md. The caller can /// optionally specify a custom `instructions` string – when `None` the @@ -446,20 +421,15 @@ async fn doc_smaller_than_limit_is_returned() { } #[tokio::test] -async fn project_doc_invalid_utf8_warns_and_uses_lossy_text() { +async fn project_doc_invalid_utf8_uses_lossy_text() { let tmp = tempfile::tempdir().expect("tempdir"); let path = tmp.path().join("AGENTS.md"); fs::write(&path, b"project\xFF doc").unwrap(); let config = make_config(&tmp, /*limit*/ 4096, /*instructions*/ None).await; - let mut warnings = Vec::new(); - let res = load_agents_md(&config, &mut warnings) - .await - .expect("doc expected") - .text(); + let res = load_agents_md(&config).await.expect("doc expected").text(); assert_eq!(res, "project\u{FFFD} doc"); - assert_invalid_utf8_warning(&warnings, "Project", config.cwd.join("AGENTS.md").as_path()); } /// Oversize file is truncated to `project_doc_max_bytes`. @@ -491,10 +461,7 @@ async fn total_byte_limit_truncates_later_project_docs() { let mut config = make_config(&repo, /*limit*/ 7, /*instructions*/ None).await; config.cwd = nested.abs(); - let mut warnings = Vec::new(); - let loaded = load_agents_md(&config, &mut warnings) - .await - .expect("project instructions"); + let loaded = load_agents_md(&config).await.expect("project instructions"); let expected = LoadedAgentsMd { user_instructions: None, entries: vec![ @@ -514,13 +481,12 @@ async fn total_byte_limit_truncates_later_project_docs() { assert_eq!(loaded, expected); assert_eq!(loaded.text(), "root\n\nabc"); - assert_eq!(warnings, Vec::::new()); } #[tokio::test] async fn read_agents_md_propagates_metadata_errors() { let tmp = tempfile::tempdir().expect("tempdir"); - let mut config = make_config(&tmp, /*limit*/ 4096, /*instructions*/ None).await; + let config = make_config(&tmp, /*limit*/ 4096, /*instructions*/ None).await; let marker_path = config.cwd.join(".git"); let fs = FailingFileSystem { path: marker_path, @@ -528,7 +494,7 @@ async fn read_agents_md_propagates_metadata_errors() { }; let cwd = config.cwd.clone(); - let err = read_agents_md(&mut config.config, &fs, "local", &cwd) + let err = read_agents_md(&config.config, &fs, "local", &cwd) .await .expect_err("metadata error"); @@ -539,14 +505,14 @@ async fn read_agents_md_propagates_metadata_errors() { async fn read_agents_md_propagates_read_errors() { let tmp = tempfile::tempdir().expect("tempdir"); fs::write(tmp.path().join("AGENTS.md"), "project doc").unwrap(); - let mut config = make_config(&tmp, /*limit*/ 4096, /*instructions*/ None).await; + let config = make_config(&tmp, /*limit*/ 4096, /*instructions*/ None).await; let fs = FailingFileSystem { path: config.cwd.join("AGENTS.md"), failure: InjectedFailure::Read(io::ErrorKind::PermissionDenied), }; let cwd = config.cwd.clone(); - let err = read_agents_md(&mut config.config, &fs, "local", &cwd) + let err = read_agents_md(&config.config, &fs, "local", &cwd) .await .expect_err("read error"); @@ -557,14 +523,14 @@ async fn read_agents_md_propagates_read_errors() { async fn read_agents_md_ignores_files_removed_after_discovery() { let tmp = tempfile::tempdir().expect("tempdir"); fs::write(tmp.path().join("AGENTS.md"), "project doc").unwrap(); - let mut config = make_config(&tmp, /*limit*/ 4096, /*instructions*/ None).await; + let config = make_config(&tmp, /*limit*/ 4096, /*instructions*/ None).await; let fs = FailingFileSystem { path: config.cwd.join("AGENTS.md"), failure: InjectedFailure::Read(io::ErrorKind::NotFound), }; let cwd = config.cwd.clone(); - let loaded = read_agents_md(&mut config.config, &fs, "local", &cwd) + let loaded = read_agents_md(&config.config, &fs, "local", &cwd) .await .expect("removed file is recoverable"); @@ -649,7 +615,7 @@ async fn multiple_environment_docs_use_labeled_layout_and_preserve_source_order( ]); let user_instructions = config.user_instructions.clone(); - let loaded = load_project_instructions(&mut config.config, user_instructions, &environments) + let loaded = load_project_instructions(&config.config, user_instructions, &environments) .await .expect("instructions expected"); let inner = format!( @@ -699,14 +665,14 @@ async fn secondary_only_project_doc_uses_single_contributor_layout() { let primary = tempfile::tempdir().expect("primary tempdir"); let secondary = tempfile::tempdir().expect("secondary tempdir"); fs::write(secondary.path().join("AGENTS.md"), "secondary doc").unwrap(); - let mut config = make_config(&primary, /*limit*/ 4096, Some("global instructions")).await; + let config = make_config(&primary, /*limit*/ 4096, Some("global instructions")).await; let environments = resolved_local_environments([ ("primary", config.cwd.clone()), ("secondary", secondary.abs()), ]); let user_instructions = config.user_instructions.clone(); - let loaded = load_project_instructions(&mut config.config, user_instructions, &environments) + let loaded = load_project_instructions(&config.config, user_instructions, &environments) .await .expect("instructions expected"); let inner = format!("global instructions{AGENTS_MD_SEPARATOR}secondary doc"); @@ -725,14 +691,14 @@ async fn primary_only_project_doc_preserves_legacy_layout_with_multiple_bound_en let primary = tempfile::tempdir().expect("primary tempdir"); let secondary = tempfile::tempdir().expect("secondary tempdir"); fs::write(primary.path().join("AGENTS.md"), "primary doc").unwrap(); - let mut config = make_config(&primary, /*limit*/ 4096, Some("global instructions")).await; + let config = make_config(&primary, /*limit*/ 4096, Some("global instructions")).await; let environments = resolved_local_environments([ ("primary", config.cwd.clone()), ("secondary", secondary.abs()), ]); let user_instructions = config.user_instructions.clone(); - let loaded = load_project_instructions(&mut config.config, user_instructions, &environments) + let loaded = load_project_instructions(&config.config, user_instructions, &environments) .await .expect("instructions expected"); let inner = format!("global instructions{AGENTS_MD_SEPARATOR}primary doc"); @@ -752,14 +718,14 @@ async fn project_doc_byte_limit_is_applied_independently_per_environment() { let secondary = tempfile::tempdir().expect("secondary tempdir"); fs::write(primary.path().join("AGENTS.md"), "ABCDE").unwrap(); fs::write(secondary.path().join("AGENTS.md"), "VWXYZ").unwrap(); - let mut config = make_config(&primary, /*limit*/ 3, /*instructions*/ None).await; + let config = make_config(&primary, /*limit*/ 3, /*instructions*/ None).await; let environments = resolved_local_environments([ ("primary", config.cwd.clone()), ("secondary", secondary.abs()), ]); let user_instructions = config.user_instructions.clone(); - let loaded = load_project_instructions(&mut config.config, user_instructions, &environments) + let loaded = load_project_instructions(&config.config, user_instructions, &environments) .await .expect("instructions expected"); @@ -784,14 +750,14 @@ async fn multiple_environments_can_exceed_single_environment_project_doc_limit() let secondary_doc = "S".repeat(LIMIT); fs::write(primary.path().join("AGENTS.md"), &primary_doc).unwrap(); fs::write(secondary.path().join("AGENTS.md"), &secondary_doc).unwrap(); - let mut config = make_config(&primary, LIMIT, /*instructions*/ None).await; + let config = make_config(&primary, LIMIT, /*instructions*/ None).await; let environments = resolved_local_environments([ ("primary", config.cwd.clone()), ("secondary", secondary.abs()), ]); let loaded = load_project_instructions( - &mut config.config, + &config.config, /*user_instructions*/ None, &environments, ) @@ -811,19 +777,19 @@ async fn multiple_environments_can_exceed_single_environment_project_doc_limit() } #[tokio::test] -async fn secondary_environment_invalid_utf8_warns_without_suppressing_other_docs() { +async fn secondary_environment_invalid_utf8_does_not_suppress_other_docs() { let primary = tempfile::tempdir().expect("primary tempdir"); let secondary = tempfile::tempdir().expect("secondary tempdir"); fs::write(primary.path().join("AGENTS.md"), "primary doc").unwrap(); fs::write(secondary.path().join("AGENTS.md"), b"secondary\xFFdoc").unwrap(); - let mut config = make_config(&primary, /*limit*/ 4096, /*instructions*/ None).await; + let config = make_config(&primary, /*limit*/ 4096, /*instructions*/ None).await; let environments = resolved_local_environments([ ("primary", config.cwd.clone()), ("secondary", secondary.abs()), ]); let loaded = load_project_instructions( - &mut config.config, + &config.config, /*user_instructions*/ None, &environments, ) @@ -832,11 +798,6 @@ async fn secondary_environment_invalid_utf8_warns_without_suppressing_other_docs assert!(loaded.text().contains("primary doc")); assert!(loaded.text().contains("secondary\u{FFFD}doc")); - assert_invalid_utf8_warning( - &config.startup_warnings, - "Project", - secondary.path().join("AGENTS.md").as_path(), - ); } #[tokio::test] @@ -853,7 +814,7 @@ async fn child_agents_guidance_is_appended_once_after_environment_groups() { ]); let loaded = load_project_instructions( - &mut config.config, + &config.config, /*user_instructions*/ None, &environments, ) @@ -902,10 +863,7 @@ async fn concatenates_root_and_cwd_docs() { let mut cfg = make_config(&repo, /*limit*/ 4096, /*instructions*/ None).await; cfg.cwd = nested.abs(); - let mut warnings = Vec::new(); - let loaded = load_agents_md(&cfg, &mut warnings) - .await - .expect("doc expected"); + let loaded = load_agents_md(&cfg).await.expect("doc expected"); let root_agents = repo.path().join("AGENTS.md").abs(); let crate_agents = cfg.cwd.join("AGENTS.md"); let expected = LoadedAgentsMd { @@ -1031,10 +989,7 @@ async fn child_agents_message_after_global_instructions_uses_plain_separator() { let mut cfg = make_config(&tmp, /*limit*/ 4096, Some("global doc")).await; cfg.features.enable(Feature::ChildAgentsMd).unwrap(); - let mut warnings = Vec::new(); - let loaded = load_agents_md(&cfg, &mut warnings) - .await - .expect("instructions expected"); + let loaded = load_agents_md(&cfg).await.expect("instructions expected"); let global_agents = cfg.codex_home.join(DEFAULT_AGENTS_MD_FILENAME); let expected = LoadedAgentsMd { user_instructions: Some(UserInstructions { @@ -1064,10 +1019,7 @@ async fn instruction_sources_include_global_before_agents_md_docs() { fs::create_dir_all(&cfg.codex_home).unwrap(); fs::write(&global_agents, "global doc").unwrap(); - let mut warnings = Vec::new(); - let loaded = load_agents_md(&cfg, &mut warnings) - .await - .expect("instructions expected"); + let loaded = load_agents_md(&cfg).await.expect("instructions expected"); let project_agents = cfg.cwd.join("AGENTS.md"); let expected = LoadedAgentsMd { @@ -1103,10 +1055,7 @@ async fn child_agents_message_after_project_docs_is_not_an_instruction_source() fs::create_dir_all(&cfg.codex_home).unwrap(); fs::write(&global_agents, "global doc").unwrap(); - let mut warnings = Vec::new(); - let loaded = load_agents_md(&cfg, &mut warnings) - .await - .expect("instructions expected"); + let loaded = load_agents_md(&cfg).await.expect("instructions expected"); let project_agents = cfg.cwd.join("AGENTS.md"); let expected = LoadedAgentsMd { diff --git a/codex-rs/core/src/codex_delegate.rs b/codex-rs/core/src/codex_delegate.rs index b36522459181..d31dc138b57f 100644 --- a/codex-rs/core/src/codex_delegate.rs +++ b/codex-rs/core/src/codex_delegate.rs @@ -106,8 +106,8 @@ pub(crate) async fn run_codex_thread_interactive( agent_control: parent_session.services.agent_control.clone(), dynamic_tools: Vec::new(), metrics_service_name: None, - inherited_shell_snapshot: None, user_shell_override: None, + inherited_environments: Some(parent_ctx.environments.clone()), inherited_exec_policy: Some(Arc::clone(&parent_session.services.exec_policy)), parent_rollout_thread_trace: codex_rollout_trace::ThreadTraceContext::disabled(), parent_trace: None, diff --git a/codex-rs/core/src/environment_selection.rs b/codex-rs/core/src/environment_selection.rs index e1ab2ea93f25..8b0a2d6b5b10 100644 --- a/codex-rs/core/src/environment_selection.rs +++ b/codex-rs/core/src/environment_selection.rs @@ -15,6 +15,7 @@ use futures::future::Shared; use crate::session::turn_context::TurnEnvironment; use crate::shell::Shell; +use crate::shell_snapshot::ShellSnapshot; pub(crate) fn default_thread_environment_selections( environment_manager: &EnvironmentManager, @@ -34,18 +35,23 @@ type SnapshotTask = Shared>; pub(crate) struct ThreadEnvironments { environment_manager: Arc, + local_shell: Shell, + shell_snapshot: ShellSnapshot, snapshot_task: ArcSwap, } impl ThreadEnvironments { - pub(crate) fn new(environment_manager: Arc) -> Self { + pub(crate) fn new( + environment_manager: Arc, + local_shell: Shell, + shell_snapshot: ShellSnapshot, + current: TurnEnvironmentSnapshot, + ) -> Self { Self { environment_manager, - snapshot_task: ArcSwap::from_pointee( - futures::future::ready(TurnEnvironmentSnapshot::default()) - .boxed() - .shared(), - ), + local_shell, + shell_snapshot, + snapshot_task: ArcSwap::from_pointee(futures::future::ready(current).boxed().shared()), } } @@ -57,9 +63,18 @@ impl ThreadEnvironments { .cloned() .unwrap_or_default(); let environment_manager = Arc::clone(&self.environment_manager); + let local_shell = self.local_shell.clone(); + let shell_snapshot = self.shell_snapshot.clone(); let environments = environments.to_vec(); let (snapshot_task, snapshot) = async move { - Self::resolve_snapshot(environment_manager, previous, environments).await + Self::resolve_snapshot( + environment_manager, + local_shell, + shell_snapshot, + previous, + environments, + ) + .await } .remote_handle(); self.snapshot_task @@ -69,6 +84,8 @@ impl ThreadEnvironments { async fn resolve_snapshot( environment_manager: Arc, + local_shell: Shell, + shell_snapshot: ShellSnapshot, current: TurnEnvironmentSnapshot, environments: Vec, ) -> TurnEnvironmentSnapshot { @@ -83,8 +100,13 @@ impl ThreadEnvironments { && environment.cwd_uri() == &selected_environment.cwd }) { Some(environment) => environment.clone(), - None => match Self::resolve_selection(&environment_manager, selected_environment) - .await + None => match Self::resolve_selection( + &environment_manager, + &local_shell, + &shell_snapshot, + selected_environment, + ) + .await { Ok(environment) => environment, Err(err) => { @@ -103,6 +125,8 @@ impl ThreadEnvironments { async fn resolve_selection( environment_manager: &EnvironmentManager, + local_shell: &Shell, + shell_snapshot: &ShellSnapshot, selected_environment: &TurnEnvironmentSelection, ) -> CodexResult { let environment_id = selected_environment.environment_id.clone(); @@ -111,22 +135,26 @@ impl ThreadEnvironments { .ok_or_else(|| { CodexErr::InvalidRequest(format!("unknown turn environment id `{environment_id}`")) })?; - let shell = match environment.info().await { - Ok(info) => match Shell::from_environment_shell_info(info.shell) { - Ok(shell) => Some(shell), + let shell = if environment.is_remote() { + match environment.info().await { + Ok(info) => match Shell::from_environment_shell_info(info.shell) { + Ok(shell) => Some(shell), + Err(err) => { + tracing::warn!( + "failed to resolve shell for environment `{environment_id}`: {err}" + ); + None + } + }, Err(err) => { - tracing::warn!( - "failed to resolve shell for environment `{environment_id}`: {err}" - ); + tracing::warn!("failed to get info for environment `{environment_id}`: {err}"); None } - }, - Err(err) => { - tracing::warn!("failed to get info for environment `{environment_id}`: {err}"); - None } + } else { + Some(local_shell.clone()) }; - Ok(TurnEnvironment::new( + let mut turn_environment = TurnEnvironment::new( environment_id, environment, selected_environment.cwd.to_abs_path().map_err(|err| { @@ -136,7 +164,15 @@ impl ThreadEnvironments { )) })?, shell, - )) + ); + let task = shell_snapshot + .clone() + .build(turn_environment.clone()) + .boxed() + .shared(); + drop(tokio::spawn(task.clone())); + turn_environment.shell_snapshot = task; + Ok(turn_environment) } pub(crate) async fn snapshot(&self) -> TurnEnvironmentSnapshot { @@ -176,12 +212,16 @@ impl TurnEnvironmentSnapshot { .map(|environment| environment.environment.get_filesystem()) } - pub(crate) fn single_local_environment_cwd(&self) -> Option<&AbsolutePathBuf> { + pub(crate) fn single_local_environment(&self) -> Option<&TurnEnvironment> { let [environment] = self.turn_environments.as_slice() else { return None; }; - (!environment.environment.is_remote()).then_some(environment.cwd()) + (!environment.environment.is_remote()).then_some(environment) + } + + pub(crate) fn single_local_environment_cwd(&self) -> Option<&AbsolutePathBuf> { + self.single_local_environment().map(TurnEnvironment::cwd) } } @@ -202,7 +242,12 @@ mod tests { environment_manager: Arc, selections: &[TurnEnvironmentSelection], ) -> Arc { - let turn_environments = Arc::new(ThreadEnvironments::new(environment_manager)); + let turn_environments = Arc::new(ThreadEnvironments::new( + environment_manager, + crate::shell::default_user_shell(), + ShellSnapshot::disabled(), + TurnEnvironmentSnapshot::default(), + )); turn_environments.update_selections(selections); turn_environments.snapshot().await; turn_environments @@ -280,6 +325,34 @@ url = "ws://127.0.0.1:8765" ); } + #[tokio::test] + async fn local_environment_uses_configured_shell() { + let cwd = AbsolutePathBuf::current_dir().expect("cwd"); + let local_shell = Shell { + shell_type: crate::shell::ShellType::Zsh, + shell_path: std::path::PathBuf::from("/configured/zsh"), + }; + let turn_environments = ThreadEnvironments::new( + Arc::new(EnvironmentManager::default_for_tests()), + local_shell.clone(), + ShellSnapshot::disabled(), + TurnEnvironmentSnapshot::default(), + ); + turn_environments.update_selections(&[TurnEnvironmentSelection { + environment_id: LOCAL_ENVIRONMENT_ID.to_string(), + cwd: PathUri::from_abs_path(&cwd), + }]); + + let snapshot = turn_environments.snapshot().await; + + assert_eq!( + snapshot + .primary() + .and_then(|environment| environment.shell.as_ref()), + Some(&local_shell) + ); + } + #[tokio::test] async fn resolve_environment_selections_keeps_first_duplicate_id() { let cwd = AbsolutePathBuf::current_dir().expect("cwd"); @@ -387,7 +460,12 @@ url = "ws://127.0.0.1:8765" .await, ); let cwd = AbsolutePathBuf::current_dir().expect("cwd"); - let turn_environments = Arc::new(ThreadEnvironments::new(manager)); + let turn_environments = Arc::new(ThreadEnvironments::new( + manager, + crate::shell::default_user_shell(), + ShellSnapshot::disabled(), + TurnEnvironmentSnapshot::default(), + )); turn_environments.update_selections(&[TurnEnvironmentSelection { environment_id: REMOTE_ENVIRONMENT_ID.to_string(), cwd: PathUri::from_abs_path(&cwd), diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index f0fef709ed32..927bf046911e 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -31,7 +31,7 @@ use crate::context::NetworkRuleSaved; use crate::context::PermissionsInstructions; use crate::context::PersonalitySpecInstructions; use crate::default_skill_metadata_budget; -use crate::environment_selection::ThreadEnvironments; +use crate::environment_selection::TurnEnvironmentSnapshot; use crate::exec_policy::ExecPolicyManager; use crate::image_preparation::prepare_response_items; use crate::parse_turn_item; @@ -304,7 +304,6 @@ use crate::network_policy_decision::execpolicy_network_rule_amendment; use crate::rollout::map_session_init_error; use crate::session_startup_prewarm::SessionStartupPrewarmHandle; use crate::shell; -use crate::shell_snapshot::ShellSnapshot; use crate::state::AutoCompactWindowSnapshot; use crate::state::PendingRequestPermissions; use crate::state::SessionServices; @@ -421,8 +420,8 @@ pub(crate) struct CodexSpawnArgs { pub(crate) agent_control: AgentControl, pub(crate) dynamic_tools: Vec, pub(crate) metrics_service_name: Option, - pub(crate) inherited_shell_snapshot: Option>, pub(crate) inherited_exec_policy: Option>, + pub(crate) inherited_environments: Option, /// Parent rollout trace used only to derive fresh spawned child traces. /// /// Root sessions and non-thread-spawn subagents pass a disabled context; @@ -507,9 +506,9 @@ impl Codex { agent_control, dynamic_tools, metrics_service_name, - inherited_shell_snapshot, user_shell_override, inherited_exec_policy, + inherited_environments, parent_rollout_thread_trace, parent_trace: _, environment_selections, @@ -519,9 +518,6 @@ impl Codex { attestation_provider, inherited_multi_agent_version, } = args; - let turn_environments = Arc::new(ThreadEnvironments::new(environment_manager)); - turn_environments.update_selections(&environment_selections); - let resolved_environments = turn_environments.snapshot().await; let (tx_sub, rx_sub) = async_channel::bounded(SUBMISSION_CHANNEL_CAPACITY); let (tx_event, rx_event) = async_channel::unbounded(); @@ -533,9 +529,6 @@ impl Codex { config .startup_warnings .extend(user_instruction_provider_warnings); - let loaded_agents_md = - load_project_instructions(&mut config, user_instructions, &resolved_environments).await; - let exec_policy = if crate::guardian::is_guardian_reviewer_source(&session_source) { // Guardian review should rely on the built-in shell safety checks, // not on caller-provided exec-policy rules that could shape the @@ -614,7 +607,7 @@ impl Codex { model_reasoning_summary: config.model_reasoning_summary, service_tier, developer_instructions: config.developer_instructions.clone(), - loaded_agents_md, + loaded_agents_md: None, personality: config.personality, base_instructions, compact_prompt: config.compact_prompt.clone(), @@ -624,7 +617,7 @@ impl Codex { windows_sandbox_level: WindowsSandboxLevel::from_config(&config), environments: TurnEnvironmentSelections::new( config.cwd.clone(), - resolved_environments.to_selections(), + environment_selections, ), workspace_roots: config.workspace_roots.clone(), codex_home: config.codex_home.clone(), @@ -638,7 +631,6 @@ impl Codex { parent_thread_id, thread_source, dynamic_tools, - inherited_shell_snapshot, user_shell_override, }; @@ -649,6 +641,7 @@ impl Codex { let session = Box::pin(Session::new( session_configuration, config.clone(), + user_instructions, installation_id, auth_manager.clone(), models_manager.clone(), @@ -663,7 +656,8 @@ impl Codex { extensions, thread_extension_init, agent_control, - turn_environments, + environment_manager, + inherited_environments, analytics_events_client, thread_store, parent_rollout_thread_trace, @@ -1399,50 +1393,12 @@ impl Session { state.set_previous_turn_settings(previous_turn_settings); } - fn maybe_refresh_shell_snapshot_for_cwd( - &self, - previous_cwd: &AbsolutePathBuf, - next_cwd: &AbsolutePathBuf, - codex_home: &AbsolutePathBuf, - session_source: &SessionSource, - ) { - if previous_cwd == next_cwd { - return; - } - - if matches!( - session_source, - SessionSource::SubAgent(SubAgentSource::ThreadSpawn { .. }) - ) { - return; - } - - if self.services.shell_snapshot.load().is_some() { - self.services.shell_snapshot.store(Some(ShellSnapshot::new( - codex_home.clone(), - self.thread_id, - next_cwd.clone(), - self.services.user_shell.as_ref(), - self.services.session_telemetry.clone(), - self.services.state_db.clone(), - ))); - } - } - pub(crate) async fn update_settings( &self, updates: SessionSettingsUpdate, ) -> ConstraintResult<()> { let notify_config_contributors = !self.services.extensions.config_contributors().is_empty(); - let ( - previous_config, - new_config, - previous_cwd, - permission_profile_changed, - next_cwd, - codex_home, - session_source, - ) = { + let (previous_config, new_config, permission_profile_changed) = { let mut state = self.state.lock().await; let updated = match state.session_configuration.apply(&updates) { Ok(updated) => updated, @@ -1456,37 +1412,19 @@ impl Session { .then(|| Self::build_effective_session_config(&state.session_configuration)); let new_config = notify_config_contributors.then(|| Self::build_effective_session_config(&updated)); - let previous_cwd = state.session_configuration.cwd().clone(); let previous_permission_profile = state.session_configuration.permission_profile(); let updated_permission_profile = updated.permission_profile(); let permission_profile_changed = previous_permission_profile != updated_permission_profile; - let next_cwd = updated.cwd().clone(); - let codex_home = updated.codex_home.clone(); - let session_source = updated.session_source.clone(); if updates.environments.is_some() { self.services .turn_environments .update_selections(updated.environment_selections()); } state.session_configuration = updated; - ( - previous_config, - new_config, - previous_cwd, - permission_profile_changed, - next_cwd, - codex_home, - session_source, - ) + (previous_config, new_config, permission_profile_changed) }; self.emit_config_changed_contributors(previous_config.as_ref(), new_config.as_ref()); - self.maybe_refresh_shell_snapshot_for_cwd( - &previous_cwd, - &next_cwd, - &codex_home, - &session_source, - ); if permission_profile_changed { self.refresh_managed_network_proxy_for_current_permission_profile() .await; diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 79f5500377f4..8640900335fa 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -2,7 +2,9 @@ use super::input_queue::InputQueue; use super::*; use crate::agents_md::LoadedAgentsMd; use crate::config::ConstraintError; +use crate::environment_selection::ThreadEnvironments; use crate::environment_selection::TurnEnvironmentSnapshot; +use crate::shell_snapshot::ShellSnapshot; use crate::skills::SkillError; use crate::state::ActiveTurn; use codex_extension_api::ExtensionDataInit; @@ -104,7 +106,6 @@ pub(crate) struct SessionConfiguration { /// Optional analytics source classification for this thread. pub(super) thread_source: Option, pub(super) dynamic_tools: Vec, - pub(super) inherited_shell_snapshot: Option>, pub(super) user_shell_override: Option, } @@ -467,6 +468,7 @@ impl Session { pub(crate) async fn new( mut session_configuration: SessionConfiguration, config: Arc, + user_instructions: Option, installation_id: String, auth_manager: Arc, models_manager: SharedModelsManager, @@ -481,7 +483,8 @@ impl Session { extensions: Arc>, thread_extension_init: ExtensionDataInit, agent_control: AgentControl, - turn_environments: Arc, + environment_manager: Arc, + inherited_environments: Option, analytics_events_client: Option, thread_store: Arc, parent_rollout_thread_trace: ThreadTraceContext, @@ -619,37 +622,12 @@ impl Session { otel.name = "session_init.auth_mcp", )); - let plugin_and_skill_warmup_fut = warm_plugins_and_skills_for_session_init( - Arc::clone(&config), - Arc::clone(&plugins_manager), - Arc::clone(&skills_manager), - turn_environments.snapshot().await, - ) - .instrument(info_span!( - "session_init.plugin_skill_warmup", - otel.name = "session_init.plugin_skill_warmup", - )); - // Join all independent futures. let ( thread_persistence_result, state_db_ctx, (auth, mcp_servers, auth_statuses, tool_plugin_provenance), - plugin_skill_errors, - ) = tokio::join!( - thread_persistence_fut, - state_db_fut, - auth_and_mcp_fut, - plugin_and_skill_warmup_fut - ); - - for err in &plugin_skill_errors { - error!( - "failed to load skill {}: {}", - err.path.display(), - err.message - ); - } + ) = tokio::join!(thread_persistence_fut, state_db_fut, auth_and_mcp_fut); let mut live_thread_init = LiveThreadInitGuard::new(thread_persistence_result.map_err(|e| { @@ -823,21 +801,48 @@ impl Session { } else { shell::default_user_shell() }; - let shell_snapshot = config.features.enabled(Feature::ShellSnapshot).then(|| { - session_configuration - .inherited_shell_snapshot - .clone() - .unwrap_or_else(|| { - ShellSnapshot::new( - config.codex_home.clone(), - thread_id, - session_configuration.cwd().clone(), - &default_shell, - session_telemetry.clone(), - state_db_ctx.clone(), - ) - }) - }); + let shell_snapshot = if config.features.enabled(Feature::ShellSnapshot) { + ShellSnapshot::new( + config.codex_home.clone(), + thread_id, + session_telemetry.clone(), + state_db_ctx.clone(), + ) + } else { + ShellSnapshot::disabled() + }; + let turn_environments = Arc::new(ThreadEnvironments::new( + environment_manager, + default_shell.clone(), + shell_snapshot, + inherited_environments.unwrap_or_default(), + )); + turn_environments.update_selections(session_configuration.environment_selections()); + let resolved_environments = turn_environments.snapshot().await; + session_configuration.loaded_agents_md = load_project_instructions( + config.as_ref(), + user_instructions, + &resolved_environments, + ) + .await; + let plugin_skill_errors = warm_plugins_and_skills_for_session_init( + Arc::clone(&config), + Arc::clone(&plugins_manager), + Arc::clone(&skills_manager), + resolved_environments, + ) + .instrument(info_span!( + "session_init.plugin_skill_warmup", + otel.name = "session_init.plugin_skill_warmup", + )) + .await; + for err in &plugin_skill_errors { + error!( + "failed to load skill {}: {}", + err.path.display(), + err.message + ); + } let thread_name = thread_title_from_thread_store(live_thread_init.as_ref(), &thread_store, thread_id) .instrument(info_span!( @@ -981,7 +986,6 @@ impl Session { hooks: arc_swap::ArcSwap::from_pointee(hooks), rollout_thread_trace, user_shell: Arc::new(default_shell), - shell_snapshot: arc_swap::ArcSwapOption::from(shell_snapshot), show_raw_agent_reasoning: config.show_raw_agent_reasoning, exec_policy, auth_manager: Arc::clone(&auth_manager), diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index dd887b421845..5086fd63bd57 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -6,8 +6,10 @@ use crate::config::ConfigOverrides; use crate::config::test_config; use crate::context::ContextualUserFragment; use crate::context::TurnAborted; +use crate::environment_selection::ThreadEnvironments; use crate::function_tool::FunctionCallError; use crate::shell::default_user_shell; +use crate::shell_snapshot::ShellSnapshot; use crate::skills::SkillRenderSideEffects; use crate::skills::render::SkillMetadataBudget; use crate::test_support::models_manager_with_provider; @@ -3323,7 +3325,6 @@ async fn set_rate_limits_retains_previous_credits() { parent_thread_id: None, thread_source: None, dynamic_tools: Vec::new(), - inherited_shell_snapshot: None, user_shell_override: None, }; @@ -3430,7 +3431,6 @@ async fn set_rate_limits_updates_plan_type_when_present() { parent_thread_id: None, thread_source: None, dynamic_tools: Vec::new(), - inherited_shell_snapshot: None, user_shell_override: None, }; @@ -3962,7 +3962,6 @@ pub(crate) async fn make_session_configuration_for_tests() -> SessionConfigurati parent_thread_id: None, thread_source: None, dynamic_tools: Vec::new(), - inherited_shell_snapshot: None, user_shell_override: None, } } @@ -4046,14 +4045,18 @@ async fn emit_subagent_session_started_includes_fork_lineage_from_session_config ); } -async fn turn_environments_for_configuration( +async fn resolved_environments_for_configuration( session_configuration: &SessionConfiguration, -) -> Arc { - let turn_environments = Arc::new(ThreadEnvironments::new(Arc::new( - codex_exec_server::EnvironmentManager::default_for_tests(), - ))); +) -> (Arc, TurnEnvironmentSnapshot) { + let environment_manager = Arc::new(EnvironmentManager::default_for_tests()); + let turn_environments = ThreadEnvironments::new( + Arc::clone(&environment_manager), + default_user_shell(), + ShellSnapshot::disabled(), + TurnEnvironmentSnapshot::default(), + ); turn_environments.update_selections(session_configuration.environment_selections()); - turn_environments + (environment_manager, turn_environments.snapshot().await) } #[tokio::test] @@ -4818,7 +4821,6 @@ async fn session_new_fails_when_zsh_fork_enabled_without_packaged_zsh() { parent_thread_id: None, thread_source: None, dynamic_tools: Vec::new(), - inherited_shell_snapshot: None, user_shell_override: None, }; @@ -4830,10 +4832,11 @@ async fn session_new_fails_when_zsh_fork_enabled_without_packaged_zsh() { config.codex_home.clone(), /*bundled_skills_enabled*/ true, )); - let turn_environments = turn_environments_for_configuration(&session_configuration).await; + let environment_manager = Arc::new(EnvironmentManager::default_for_tests()); let result = Session::new( session_configuration, Arc::clone(&config), + /*user_instructions*/ None, "11111111-1111-4111-8111-111111111111".to_string(), auth_manager, models_manager, @@ -4848,7 +4851,8 @@ async fn session_new_fails_when_zsh_fork_enabled_without_packaged_zsh() { Arc::new(codex_extension_api::ExtensionRegistryBuilder::new().build()), codex_extension_api::ExtensionDataInit::default(), AgentControl::default(), - turn_environments, + environment_manager, + /*inherited_environments*/ None, /*analytics_events_client*/ None, Arc::new(codex_thread_store::LocalThreadStore::new( codex_thread_store::LocalThreadStoreConfig::from_config(config.as_ref()), @@ -4927,7 +4931,6 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { parent_thread_id: None, thread_source: None, dynamic_tools: Vec::new(), - inherited_shell_snapshot: None, user_shell_override: None, }; let per_turn_config = @@ -4944,8 +4947,15 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { ); let state = SessionState::new(session_configuration.clone()); - let turn_environments = turn_environments_for_configuration(&session_configuration).await; - let resolved_turn_environments = turn_environments.snapshot().await; + let (environment_manager, resolved_environments) = + resolved_environments_for_configuration(&session_configuration).await; + let resolved_turn_environments = resolved_environments.clone(); + let turn_environments = Arc::new(ThreadEnvironments::new( + environment_manager, + default_user_shell(), + ShellSnapshot::disabled(), + resolved_environments, + )); let environment = Arc::clone( &resolved_turn_environments .primary() @@ -4984,7 +4994,6 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { })), rollout_thread_trace: codex_rollout_trace::ThreadTraceContext::disabled(), user_shell: Arc::new(default_user_shell()), - shell_snapshot: arc_swap::ArcSwapOption::empty(), show_raw_agent_reasoning: config.show_raw_agent_reasoning, exec_policy, auth_manager: auth_manager.clone(), @@ -5160,7 +5169,6 @@ async fn make_session_with_config_and_rx( parent_thread_id: None, thread_source: None, dynamic_tools: Vec::new(), - inherited_shell_snapshot: None, user_shell_override: None, }; @@ -5172,11 +5180,12 @@ async fn make_session_with_config_and_rx( config.codex_home.clone(), /*bundled_skills_enabled*/ true, )); - let turn_environments = turn_environments_for_configuration(&session_configuration).await; + let environment_manager = Arc::new(EnvironmentManager::default_for_tests()); let session = Session::new( session_configuration, Arc::clone(&config), + /*user_instructions*/ None, "11111111-1111-4111-8111-111111111111".to_string(), auth_manager, models_manager, @@ -5191,7 +5200,8 @@ async fn make_session_with_config_and_rx( Arc::new(codex_extension_api::ExtensionRegistryBuilder::new().build()), codex_extension_api::ExtensionDataInit::default(), AgentControl::default(), - turn_environments, + environment_manager, + /*inherited_environments*/ None, /*analytics_events_client*/ None, Arc::new(codex_thread_store::LocalThreadStore::new( codex_thread_store::LocalThreadStoreConfig::from_config(config.as_ref()), @@ -5263,7 +5273,6 @@ async fn make_session_with_history_source_and_agent_control_and_rx( parent_thread_id: None, thread_source: None, dynamic_tools: Vec::new(), - inherited_shell_snapshot: None, user_shell_override: None, }; @@ -5275,11 +5284,12 @@ async fn make_session_with_history_source_and_agent_control_and_rx( config.codex_home.clone(), /*bundled_skills_enabled*/ true, )); - let turn_environments = turn_environments_for_configuration(&session_configuration).await; + let environment_manager = Arc::new(EnvironmentManager::default_for_tests()); let session = Session::new( session_configuration, Arc::clone(&config), + /*user_instructions*/ None, "11111111-1111-4111-8111-111111111111".to_string(), auth_manager, models_manager, @@ -5294,7 +5304,8 @@ async fn make_session_with_history_source_and_agent_control_and_rx( Arc::new(codex_extension_api::ExtensionRegistryBuilder::new().build()), codex_extension_api::ExtensionDataInit::default(), agent_control, - turn_environments, + environment_manager, + /*inherited_environments*/ None, /*analytics_events_client*/ None, Arc::new(codex_thread_store::LocalThreadStore::new( codex_thread_store::LocalThreadStoreConfig::from_config(config.as_ref()), @@ -6965,7 +6976,6 @@ where parent_thread_id: None, thread_source: None, dynamic_tools, - inherited_shell_snapshot: None, user_shell_override: None, }; let per_turn_config = @@ -6982,8 +6992,14 @@ where ); let state = SessionState::new(session_configuration.clone()); - let turn_environments = turn_environments_for_configuration(&session_configuration).await; - let resolved_turn_environments = turn_environments.snapshot().await; + let (environment_manager, resolved_turn_environments) = + resolved_environments_for_configuration(&session_configuration).await; + let turn_environments = Arc::new(ThreadEnvironments::new( + environment_manager, + default_user_shell(), + ShellSnapshot::disabled(), + resolved_turn_environments.clone(), + )); let environment = Arc::clone( &resolved_turn_environments .primary() @@ -7022,7 +7038,6 @@ where })), rollout_thread_trace: codex_rollout_trace::ThreadTraceContext::disabled(), user_shell: Arc::new(default_user_shell()), - shell_snapshot: arc_swap::ArcSwapOption::empty(), show_raw_agent_reasoning: config.show_raw_agent_reasoning, exec_policy, auth_manager: Arc::clone(&auth_manager), diff --git a/codex-rs/core/src/session/tests/guardian_tests.rs b/codex-rs/core/src/session/tests/guardian_tests.rs index 184472084e40..e958b39e1f7d 100644 --- a/codex-rs/core/src/session/tests/guardian_tests.rs +++ b/codex-rs/core/src/session/tests/guardian_tests.rs @@ -725,7 +725,7 @@ async fn guardian_subagent_does_not_inherit_parent_exec_policy_rules() { agent_control: AgentControl::default(), dynamic_tools: Vec::new(), metrics_service_name: None, - inherited_shell_snapshot: None, + inherited_environments: None, inherited_exec_policy: Some(Arc::new(parent_exec_policy)), parent_rollout_thread_trace: codex_rollout_trace::ThreadTraceContext::disabled(), user_shell_override: None, diff --git a/codex-rs/core/src/session/turn_context.rs b/codex-rs/core/src/session/turn_context.rs index 940fb988e5e7..acb126d7e14e 100644 --- a/codex-rs/core/src/session/turn_context.rs +++ b/codex-rs/core/src/session/turn_context.rs @@ -3,6 +3,8 @@ use crate::SkillLoadOutcome; use crate::agents_md::LoadedAgentsMd; use crate::config::GhostSnapshotConfig; use crate::environment_selection::TurnEnvironmentSnapshot; +use crate::path_utils; +use crate::shell_snapshot::ShellSnapshotFile; use codex_core_skills::HostLoadedSkills; use codex_file_system::FileSystemSandboxContext; use codex_model_provider::SharedModelProvider; @@ -19,6 +21,9 @@ use codex_sandboxing::compatibility_sandbox_policy_for_permission_profile; use codex_sandboxing::policy_transforms::effective_file_system_sandbox_policy; use codex_sandboxing::policy_transforms::effective_network_sandbox_policy; use codex_utils_path_uri::PathUri; +use futures::FutureExt; +use futures::future::BoxFuture; +use futures::future::Shared; use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering; use tracing::instrument; @@ -38,7 +43,9 @@ impl TurnSkillsContext { } } -#[derive(Clone, Debug)] +pub(crate) type ShellSnapshotTask = Shared>>>; + +#[derive(Clone)] pub(crate) struct TurnEnvironment { pub(crate) environment_id: String, pub(crate) environment: Arc, @@ -50,6 +57,7 @@ pub(crate) struct TurnEnvironment { cwd: AbsolutePathBuf, cwd_uri: PathUri, pub(crate) shell: Option, + pub(crate) shell_snapshot: ShellSnapshotTask, } impl TurnEnvironment { @@ -66,7 +74,18 @@ impl TurnEnvironment { cwd, cwd_uri, shell, + shell_snapshot: futures::future::ready(None).boxed().shared(), + } + } + + pub(crate) fn shell_snapshot(&self, cwd: &AbsolutePathBuf) -> Option { + if !path_utils::paths_match_after_normalization(self.cwd.as_path(), cwd.as_path()) { + return None; } + self.shell_snapshot + .peek()? + .as_deref() + .map(ShellSnapshotFile::path) } pub(crate) fn cwd(&self) -> &AbsolutePathBuf { @@ -85,6 +104,18 @@ impl TurnEnvironment { } } +impl std::fmt::Debug for TurnEnvironment { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("TurnEnvironment") + .field("environment_id", &self.environment_id) + .field("environment", &self.environment) + .field("cwd", &self.cwd) + .field("cwd_uri", &self.cwd_uri) + .field("shell", &self.shell) + .finish_non_exhaustive() + } +} + /// The context needed for a single turn of the thread. #[derive(Debug)] pub struct TurnContext { @@ -623,14 +654,11 @@ impl Session { let mut state = self.state.lock().await; match state.session_configuration.clone().apply(&updates) { Ok(next) => { - let previous_cwd = state.session_configuration.cwd().clone(); let previous_permission_profile = state.session_configuration.permission_profile(); let next_permission_profile = next.permission_profile(); let permission_profile_changed = previous_permission_profile != next_permission_profile; - let codex_home = next.codex_home.clone(); - let session_source = next.session_source.clone(); let previous_config = notify_config_contributors.then(|| { Self::build_effective_session_config(&state.session_configuration) }); @@ -645,9 +673,6 @@ impl Session { Ok(( next, permission_profile_changed, - previous_cwd, - codex_home, - session_source, previous_config, new_config, )) @@ -656,36 +681,23 @@ impl Session { } }; - let ( - session_configuration, - permission_profile_changed, - previous_cwd, - codex_home, - session_source, - previous_config, - new_config, - ) = match update_result { - Ok(update) => update, - Err(err) => { - let message = err.to_string(); - self.send_event_raw(Event { - id: sub_id.clone(), - msg: EventMsg::Error(ErrorEvent { - message: message.clone(), - codex_error_info: Some(CodexErrorInfo::BadRequest), - }), - }) - .await; - return Err(CodexErr::InvalidRequest(message)); - } - }; + let (session_configuration, permission_profile_changed, previous_config, new_config) = + match update_result { + Ok(update) => update, + Err(err) => { + let message = err.to_string(); + self.send_event_raw(Event { + id: sub_id.clone(), + msg: EventMsg::Error(ErrorEvent { + message: message.clone(), + codex_error_info: Some(CodexErrorInfo::BadRequest), + }), + }) + .await; + return Err(CodexErr::InvalidRequest(message)); + } + }; self.emit_config_changed_contributors(previous_config.as_ref(), new_config.as_ref()); - self.maybe_refresh_shell_snapshot_for_cwd( - &previous_cwd, - session_configuration.cwd(), - &codex_home, - &session_source, - ); if permission_profile_changed { self.refresh_managed_network_proxy_for_current_permission_profile() diff --git a/codex-rs/core/src/shell_snapshot.rs b/codex-rs/core/src/shell_snapshot.rs index a2cfbef3d718..6e6c6cd51510 100644 --- a/codex-rs/core/src/shell_snapshot.rs +++ b/codex-rs/core/src/shell_snapshot.rs @@ -6,8 +6,8 @@ use std::time::Duration; use std::time::SystemTime; use crate::StateDbHandle; -use crate::path_utils; use crate::rollout::list::find_thread_path_by_id_str; +use crate::session::turn_context::TurnEnvironment; use crate::shell::Shell; use crate::shell::ShellType; use crate::shell::get_shell; @@ -20,17 +20,23 @@ use codex_protocol::ThreadId; use codex_utils_absolute_path::AbsolutePathBuf; use tokio::fs; use tokio::process::Command; -use tokio::sync::watch; use tokio::time::timeout; use tracing::Instrument; use tracing::info_span; +#[derive(Clone)] pub(crate) struct ShellSnapshot { - cwd: AbsolutePathBuf, - state_rx: watch::Receiver>>, + config: Option>, } -struct ShellSnapshotFile { +struct ShellSnapshotConfig { + codex_home: AbsolutePathBuf, + session_id: ThreadId, + session_telemetry: SessionTelemetry, + state_db: Option, +} + +pub(crate) struct ShellSnapshotFile { path: AbsolutePathBuf, } @@ -43,77 +49,68 @@ impl ShellSnapshot { pub(crate) fn new( codex_home: AbsolutePathBuf, session_id: ThreadId, - session_cwd: AbsolutePathBuf, - shell: &Shell, session_telemetry: SessionTelemetry, state_db: Option, - ) -> Arc { - let (state_tx, state_rx) = watch::channel(None); - let snapshot = Arc::new(Self { - cwd: session_cwd.clone(), - state_rx, - }); - Self::spawn_snapshot_task( - codex_home, - session_id, - session_cwd, - shell.clone(), - state_tx, - session_telemetry, - state_db, - ); - snapshot + ) -> Self { + Self { + config: Some(Arc::new(ShellSnapshotConfig { + codex_home, + session_id, + session_telemetry, + state_db, + })), + } + } + + pub(crate) fn disabled() -> Self { + Self { config: None } } - pub(crate) fn location(&self, cwd: &AbsolutePathBuf) -> Option { - if !path_utils::paths_match_after_normalization(self.cwd.as_path(), cwd) { + pub(crate) async fn build( + self, + environment: TurnEnvironment, + ) -> Option> { + let config = self.config.as_ref()?; + if environment.environment.is_remote() { return None; } - self.state_rx - .borrow() - .as_ref() - .and_then(|snapshot| snapshot.as_ref().ok()) - .map(|snapshot| snapshot.path.clone()) - } - - pub(crate) fn is_failed(&self) -> bool { - matches!(&*self.state_rx.borrow(), Some(Err(_))) + let shell = environment.shell.clone()?; + let cwd = environment.cwd().clone(); + Self::build_for_cwd(Arc::clone(config), cwd, shell).await } - fn spawn_snapshot_task( - codex_home: AbsolutePathBuf, - session_id: ThreadId, - session_cwd: AbsolutePathBuf, - snapshot_shell: Shell, - state_tx: watch::Sender>>, - session_telemetry: SessionTelemetry, - state_db: Option, - ) { - let snapshot_span = info_span!("shell_snapshot", thread_id = %session_id); - tokio::spawn( - async move { - let timer = session_telemetry.start_timer("codex.shell_snapshot.duration_ms", &[]); - let snapshot = ShellSnapshot::try_create( - &codex_home, - session_id, - &session_cwd, - &snapshot_shell, - state_db, - ) - .await; - let success = snapshot.is_ok(); - let success_tag = if success { "true" } else { "false" }; - let _ = timer.map(|timer| timer.record(&[("success", success_tag)])); - let mut counter_tags = vec![("success", success_tag)]; - if let Some(failure_reason) = snapshot.as_ref().err() { - counter_tags.push(("failure_reason", *failure_reason)); - } - session_telemetry.counter("codex.shell_snapshot", /*inc*/ 1, &counter_tags); - let _ = state_tx.send(Some(snapshot)); + async fn build_for_cwd( + config: Arc, + cwd: AbsolutePathBuf, + shell: Shell, + ) -> Option> { + let snapshot_span = info_span!("shell_snapshot", thread_id = %config.session_id); + async { + let timer = config + .session_telemetry + .start_timer("codex.shell_snapshot.duration_ms", &[]); + let snapshot = ShellSnapshot::try_create( + &config.codex_home, + config.session_id, + &cwd, + &shell, + config.state_db.clone(), + ) + .await; + let success_tag = if snapshot.is_ok() { "true" } else { "false" }; + let _ = timer.map(|timer| timer.record(&[("success", success_tag)])); + let mut counter_tags = vec![("success", success_tag)]; + if let Some(failure_reason) = snapshot.as_ref().err() { + counter_tags.push(("failure_reason", *failure_reason)); } - .instrument(snapshot_span), - ); + config + .session_telemetry + .counter("codex.shell_snapshot", /*inc*/ 1, &counter_tags); + snapshot.ok().map(Arc::new) + } + .instrument(snapshot_span) + .await } async fn try_create( @@ -179,6 +176,12 @@ impl ShellSnapshot { } } +impl ShellSnapshotFile { + pub(crate) fn path(&self) -> AbsolutePathBuf { + self.path.clone() + } +} + impl Drop for ShellSnapshotFile { fn drop(&mut self) { if let Err(err) = std::fs::remove_file(&self.path) { diff --git a/codex-rs/core/src/shell_snapshot_tests.rs b/codex-rs/core/src/shell_snapshot_tests.rs index de07d9a136a7..dcba3e24594b 100644 --- a/codex-rs/core/src/shell_snapshot_tests.rs +++ b/codex-rs/core/src/shell_snapshot_tests.rs @@ -215,42 +215,6 @@ async fn try_create_creates_and_deletes_snapshot_file() -> Result<()> { Ok(()) } -#[test] -fn location_matches_cwd_and_channel_drop_deletes_file() -> Result<()> { - let dir = tempdir()?; - let path = dir.path().join("snapshot.sh").abs(); - std::fs::write(&path, "# snapshot")?; - let cwd = dir.path().join("worktree-a").abs(); - let other_cwd = dir.path().join("worktree-b").abs(); - let (state_tx, state_rx) = watch::channel(Some(Ok(ShellSnapshotFile { path: path.clone() }))); - let snapshot = ShellSnapshot { - cwd: cwd.clone(), - state_rx, - }; - - assert_eq!(snapshot.location(&cwd), Some(path.clone())); - assert_eq!(snapshot.location(&other_cwd), None); - - drop(state_tx); - drop(snapshot); - assert!(!path.exists()); - - Ok(()) -} - -#[test] -fn snapshot_state_distinguishes_building_from_failed() -> Result<()> { - let cwd = tempdir()?.path().abs(); - let (state_tx, state_rx) = watch::channel(None); - let snapshot = ShellSnapshot { cwd, state_rx }; - - assert!(!snapshot.is_failed()); - state_tx.send(Some(Err("failed")))?; - assert!(snapshot.is_failed()); - - Ok(()) -} - #[cfg(unix)] #[tokio::test] async fn try_create_uses_distinct_generation_paths() -> Result<()> { diff --git a/codex-rs/core/src/state/service.rs b/codex-rs/core/src/state/service.rs index 5d888bd83dc3..00380c6d6470 100644 --- a/codex-rs/core/src/state/service.rs +++ b/codex-rs/core/src/state/service.rs @@ -12,7 +12,6 @@ use crate::exec_policy::ExecPolicyManager; use crate::guardian::GuardianRejection; use crate::guardian::GuardianRejectionCircuitBreaker; use crate::mcp::McpManager; -use crate::shell_snapshot::ShellSnapshot; use crate::tools::code_mode::CodeModeService; use crate::tools::handlers::ToolSearchHandlerCache; use crate::tools::network_approval::NetworkApprovalService; @@ -53,7 +52,6 @@ pub(crate) struct SessionServices { pub(crate) hooks: ArcSwap, pub(crate) rollout_thread_trace: ThreadTraceContext, pub(crate) user_shell: Arc, - pub(crate) shell_snapshot: ArcSwapOption, pub(crate) show_raw_agent_reasoning: bool, pub(crate) exec_policy: Arc, pub(crate) auth_manager: Arc, diff --git a/codex-rs/core/src/tasks/user_shell.rs b/codex-rs/core/src/tasks/user_shell.rs index a039dfd673ba..4a2a7649446a 100644 --- a/codex-rs/core/src/tasks/user_shell.rs +++ b/codex-rs/core/src/tasks/user_shell.rs @@ -129,12 +129,13 @@ pub(crate) async fn execute_user_shell_command( // We do not source rc files or otherwise reformat the script. let use_login_shell = true; let session_shell = session.user_shell(); - let shell_snapshot = session.services.shell_snapshot.load_full(); - #[allow(deprecated)] - let shell_snapshot_location = shell_snapshot - .as_ref() - .and_then(|snapshot| snapshot.location(&turn_context.cwd)); - let display_command = session_shell.derive_exec_args(&command, use_login_shell); + let environment = turn_context.environments.single_local_environment(); + let shell = environment + .and_then(|environment| environment.shell.as_ref()) + .unwrap_or(session_shell.as_ref()); + let shell_snapshot_location = + environment.and_then(|environment| environment.shell_snapshot(environment.cwd())); + let display_command = shell.derive_exec_args(&command, use_login_shell); let mut exec_env_map = create_env( &turn_context.shell_environment_policy, Some(session.thread_id), @@ -144,7 +145,7 @@ pub(crate) async fn execute_user_shell_command( } let exec_command = prepare_user_shell_exec_command( &display_command, - session_shell.as_ref(), + shell, shell_snapshot_location.as_ref(), &turn_context.shell_environment_policy.r#set, &mut exec_env_map, @@ -153,7 +154,9 @@ pub(crate) async fn execute_user_shell_command( let call_id = Uuid::new_v4().to_string(); let raw_command = command; #[allow(deprecated)] - let cwd = turn_context.cwd.clone(); + let cwd = environment + .map(|environment| environment.cwd().clone()) + .unwrap_or_else(|| turn_context.cwd.clone()); let parsed_cmd = parse_command(&display_command); session diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index dd4852c73227..2adb713a0e51 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -4,6 +4,7 @@ use crate::attestation::AttestationProvider; use crate::codex_thread::CodexThread; use crate::config::Config; use crate::config::ThreadStoreConfig; +use crate::environment_selection::TurnEnvironmentSnapshot; use crate::environment_selection::default_thread_environment_selections; use crate::mcp::McpManager; use crate::rollout::truncation; @@ -12,7 +13,6 @@ use crate::session::CodexSpawnArgs; use crate::session::CodexSpawnOk; use crate::session::INITIAL_SUBMIT_ID; use crate::session::resolve_multi_agent_version; -use crate::shell_snapshot::ShellSnapshot; use crate::tasks::InterruptedTurnHistoryMarker; use crate::tasks::interrupted_turn_history_marker; use codex_analytics::AnalyticsEventsClient; @@ -194,7 +194,7 @@ pub(crate) struct ResumeThreadWithHistoryOptions { pub(crate) agent_control: AgentControl, pub(crate) session_source: SessionSource, pub(crate) parent_thread_id: Option, - pub(crate) inherited_shell_snapshot: Option>, + pub(crate) inherited_environments: Option, pub(crate) inherited_exec_policy: Option>, } @@ -638,7 +638,7 @@ impl ThreadManager { thread_source, options.dynamic_tools, options.metrics_service_name, - /*inherited_shell_snapshot*/ None, + /*inherited_environments*/ None, /*inherited_exec_policy*/ None, options.parent_trace, options.environments, @@ -728,7 +728,7 @@ impl ThreadManager { thread_source, Vec::new(), /*metrics_service_name*/ None, - /*inherited_shell_snapshot*/ None, + /*inherited_environments*/ None, /*inherited_exec_policy*/ None, parent_trace, environments, @@ -791,7 +791,7 @@ impl ThreadManager { thread_source, Vec::new(), /*metrics_service_name*/ None, - /*inherited_shell_snapshot*/ None, + /*inherited_environments*/ None, /*inherited_exec_policy*/ None, /*parent_trace*/ None, environments, @@ -1181,7 +1181,7 @@ impl ThreadManagerState { /*forked_from_thread_id*/ None, /*thread_source*/ None, /*metrics_service_name*/ None, - /*inherited_shell_snapshot*/ None, + /*inherited_environments*/ None, /*inherited_exec_policy*/ None, /*environments*/ None, )) @@ -1198,7 +1198,7 @@ impl ThreadManagerState { forked_from_thread_id: Option, thread_source: Option, metrics_service_name: Option, - inherited_shell_snapshot: Option>, + inherited_environments: Option, inherited_exec_policy: Option>, environments: Option>, ) -> CodexResult { @@ -1216,7 +1216,7 @@ impl ThreadManagerState { thread_source, Vec::new(), metrics_service_name, - inherited_shell_snapshot, + inherited_environments, inherited_exec_policy, /*parent_trace*/ None, environments, @@ -1236,7 +1236,7 @@ impl ThreadManagerState { agent_control, session_source, parent_thread_id, - inherited_shell_snapshot, + inherited_environments, inherited_exec_policy, } = options; let environments = @@ -1253,7 +1253,7 @@ impl ThreadManagerState { thread_source, Vec::new(), /*metrics_service_name*/ None, - inherited_shell_snapshot, + inherited_environments, inherited_exec_policy, /*parent_trace*/ None, environments, @@ -1273,7 +1273,7 @@ impl ThreadManagerState { thread_source: Option, parent_thread_id: Option, forked_from_thread_id: Option, - inherited_shell_snapshot: Option>, + inherited_environments: Option, inherited_exec_policy: Option>, environments: Option>, ) -> CodexResult { @@ -1291,7 +1291,7 @@ impl ThreadManagerState { thread_source, Vec::new(), /*metrics_service_name*/ None, - inherited_shell_snapshot, + inherited_environments, inherited_exec_policy, /*parent_trace*/ None, environments, @@ -1330,7 +1330,7 @@ impl ThreadManagerState { thread_source, dynamic_tools, metrics_service_name, - /*inherited_shell_snapshot*/ None, + /*inherited_environments*/ None, /*inherited_exec_policy*/ None, parent_trace, environments, @@ -1353,7 +1353,7 @@ impl ThreadManagerState { thread_source: Option, dynamic_tools: Vec, metrics_service_name: Option, - inherited_shell_snapshot: Option>, + inherited_environments: Option, inherited_exec_policy: Option>, parent_trace: Option, environments: Vec, @@ -1418,7 +1418,7 @@ impl ThreadManagerState { agent_control, dynamic_tools, metrics_service_name, - inherited_shell_snapshot, + inherited_environments, inherited_exec_policy, parent_rollout_thread_trace, user_shell_override, diff --git a/codex-rs/core/src/tools/handlers/shell.rs b/codex-rs/core/src/tools/handlers/shell.rs index b978a2f577a8..91f2c64d3c3a 100644 --- a/codex-rs/core/src/tools/handlers/shell.rs +++ b/codex-rs/core/src/tools/handlers/shell.rs @@ -179,6 +179,7 @@ async fn run_exec_like(args: RunExecLikeArgs) -> Result, + pub turn_environment: TurnEnvironment, pub shell_type: Option, pub hook_command: String, pub cwd: AbsolutePathBuf, @@ -240,10 +242,12 @@ impl ToolRuntime for ShellRuntime { ctx: &ToolCtx, ) -> Result { let session_shell = ctx.session.user_shell(); - let shell_snapshot = ctx.session.services.shell_snapshot.load_full(); - let shell_snapshot_location = shell_snapshot + let shell = req + .turn_environment + .shell .as_ref() - .and_then(|snapshot| snapshot.location(&req.cwd)); + .unwrap_or(session_shell.as_ref()); + let shell_snapshot_location = req.turn_environment.shell_snapshot(&req.cwd); let (file_system_sandbox_policy, _) = attempt.permissions.to_runtime_permissions(); let sandbox_permissions = sandbox_permissions_preserving_denied_reads( req.sandbox_permissions, @@ -272,7 +276,7 @@ impl ToolRuntime for ShellRuntime { let runtime_path_prepends = RuntimePathPrepends::default(); let command = maybe_wrap_shell_lc_with_snapshot( &req.command, - session_shell.as_ref(), + shell, shell_snapshot_location.as_ref(), &explicit_env_overrides, &env, @@ -284,7 +288,7 @@ impl ToolRuntime for ShellRuntime { attempt.sandbox, attempt.windows_sandbox_level, ); - let command = if matches!(session_shell.shell_type, ShellType::PowerShell) { + let command = if matches!(shell.shell_type, ShellType::PowerShell) { prefix_powershell_script_with_utf8(&command) } else { command diff --git a/codex-rs/core/src/tools/runtimes/unified_exec.rs b/codex-rs/core/src/tools/runtimes/unified_exec.rs index 2ec2704ebb5f..23c65bdeb644 100644 --- a/codex-rs/core/src/tools/runtimes/unified_exec.rs +++ b/codex-rs/core/src/tools/runtimes/unified_exec.rs @@ -13,6 +13,7 @@ use crate::guardian::review_approval_request; use crate::sandboxing::ExecOptions; use crate::sandboxing::ExecServerEnvConfig; use crate::sandboxing::SandboxPermissions; +use crate::session::turn_context::TurnEnvironment; use crate::shell::ShellType; use crate::tools::flat_tool_name; use crate::tools::network_approval::NetworkApprovalMode; @@ -41,7 +42,6 @@ use crate::unified_exec::NoopSpawnLifecycle; use crate::unified_exec::UnifiedExecError; use crate::unified_exec::UnifiedExecProcess; use crate::unified_exec::UnifiedExecProcessManager; -use codex_exec_server::Environment; use codex_network_proxy::NetworkProxy; use codex_protocol::error::CodexErr; use codex_protocol::error::SandboxErr; @@ -53,7 +53,6 @@ use codex_tools::UnifiedExecShellMode; use codex_utils_absolute_path::AbsolutePathBuf; use futures::future::BoxFuture; use std::collections::HashMap; -use std::sync::Arc; use tokio_util::sync::CancellationToken; /// Request payload used by the unified-exec runtime after approvals and @@ -66,7 +65,7 @@ pub struct UnifiedExecRequest { pub process_id: i32, pub cwd: AbsolutePathBuf, pub sandbox_cwd: AbsolutePathBuf, - pub environment: Arc, + pub turn_environment: TurnEnvironment, pub env: HashMap, pub exec_server_env_config: Option, pub explicit_env_overrides: HashMap, @@ -264,10 +263,12 @@ impl<'a> ToolRuntime for UnifiedExecRunt ) -> Result { let base_command = &req.command; let session_shell = ctx.session.user_shell(); - let shell_snapshot = ctx.session.services.shell_snapshot.load_full(); - let shell_snapshot_location = shell_snapshot + let shell = req + .turn_environment + .shell .as_ref() - .and_then(|snapshot| snapshot.location(&req.cwd)); + .unwrap_or(session_shell.as_ref()); + let shell_snapshot_location = req.turn_environment.shell_snapshot(&req.cwd); let (file_system_sandbox_policy, _) = attempt.permissions.to_runtime_permissions(); let launch_sandbox_permissions = sandbox_permissions_preserving_denied_reads( req.sandbox_permissions, @@ -281,7 +282,7 @@ impl<'a> ToolRuntime for UnifiedExecRunt if let Some(network) = managed_network { network.apply_to_env(&mut env); } - let environment_is_remote = req.environment.is_remote(); + let environment_is_remote = req.turn_environment.environment.is_remote(); let explicit_env_overrides = req.explicit_env_overrides.clone(); #[cfg(unix)] let runtime_path_prepends = { @@ -308,7 +309,7 @@ impl<'a> ToolRuntime for UnifiedExecRunt } else { maybe_wrap_shell_lc_with_snapshot( base_command, - session_shell.as_ref(), + shell, shell_snapshot_location.as_ref(), &explicit_env_overrides, &env, @@ -351,7 +352,7 @@ impl<'a> ToolRuntime for UnifiedExecRunt .await? { Some(prepared) => { - if req.environment.is_remote() { + if req.turn_environment.environment.is_remote() { return Err(ToolError::Rejected( "unified_exec zsh-fork is not supported for remote environments" .to_string(), @@ -364,7 +365,7 @@ impl<'a> ToolRuntime for UnifiedExecRunt &prepared.exec_request, req.tty, prepared.spawn_lifecycle, - req.environment.as_ref(), + req.turn_environment.environment.as_ref(), ) .await .map_err(|err| match err { @@ -403,7 +404,7 @@ impl<'a> ToolRuntime for UnifiedExecRunt &exec_env, req.tty, Box::new(NoopSpawnLifecycle), - req.environment.as_ref(), + req.turn_environment.environment.as_ref(), ) .await .map_err(|err| match err { @@ -424,10 +425,21 @@ mod tests { use crate::exec::DEFAULT_EXEC_COMMAND_TIMEOUT_MS; use crate::tools::sandboxing::ToolRuntime; use codex_exec_server::Environment; + use codex_exec_server::LOCAL_ENVIRONMENT_ID; use codex_tools::ZshForkConfig; + use std::sync::Arc; use std::time::Duration; use tempfile::tempdir; + fn test_turn_environment(cwd: AbsolutePathBuf) -> TurnEnvironment { + TurnEnvironment::new( + LOCAL_ENVIRONMENT_ID.to_string(), + Arc::new(Environment::default_for_tests()), + cwd, + /*shell*/ None, + ) + } + #[test] fn unified_exec_options_combines_default_timeout_with_network_denial_cancellation() { let cancellation = CancellationToken::new(); @@ -467,7 +479,7 @@ mod tests { process_id: 1000, cwd, sandbox_cwd: sandbox_cwd.clone(), - environment: Arc::new(Environment::default_for_tests()), + turn_environment: test_turn_environment(sandbox_cwd.clone()), env: HashMap::new(), exec_server_env_config: None, explicit_env_overrides: HashMap::new(), @@ -565,8 +577,8 @@ mod tests { hook_command: "echo hi".to_string(), process_id: 1000, cwd: cwd.clone(), - sandbox_cwd: cwd, - environment: Arc::new(Environment::default_for_tests()), + sandbox_cwd: cwd.clone(), + turn_environment: test_turn_environment(cwd), env: HashMap::new(), exec_server_env_config: None, explicit_env_overrides: HashMap::new(), diff --git a/codex-rs/core/src/unified_exec/mod.rs b/codex-rs/core/src/unified_exec/mod.rs index c14d063e9672..23dd3bf1b5fd 100644 --- a/codex-rs/core/src/unified_exec/mod.rs +++ b/codex-rs/core/src/unified_exec/mod.rs @@ -27,7 +27,6 @@ use std::collections::HashSet; use std::sync::Arc; use std::sync::Weak; -use codex_exec_server::Environment; use codex_network_proxy::NetworkProxy; use codex_protocol::models::AdditionalPermissionProfile; use codex_tools::UnifiedExecShellMode; @@ -40,6 +39,7 @@ use tokio::sync::Mutex; use crate::sandboxing::SandboxPermissions; use crate::session::session::Session; use crate::session::turn_context::TurnContext; +use crate::session::turn_context::TurnEnvironment; use crate::shell::ShellType; use crate::tools::network_approval::DeferredNetworkApproval; @@ -98,7 +98,7 @@ pub(crate) struct ExecCommandRequest { pub max_output_tokens: Option, pub cwd: AbsolutePathBuf, pub sandbox_cwd: AbsolutePathBuf, - pub environment: Arc, + pub turn_environment: TurnEnvironment, pub shell_mode: UnifiedExecShellMode, pub network: Option, pub tty: bool, diff --git a/codex-rs/core/src/unified_exec/process_manager.rs b/codex-rs/core/src/unified_exec/process_manager.rs index 755590c86b0b..702a339e2edb 100644 --- a/codex-rs/core/src/unified_exec/process_manager.rs +++ b/codex-rs/core/src/unified_exec/process_manager.rs @@ -1069,7 +1069,7 @@ impl UnifiedExecProcessManager { process_id: request.process_id, cwd, sandbox_cwd: request.sandbox_cwd.clone(), - environment: Arc::clone(&request.environment), + turn_environment: request.turn_environment.clone(), env, exec_server_env_config: Some(exec_server_env_config), explicit_env_overrides: context.turn.shell_environment_policy.r#set.clone(), diff --git a/codex-rs/core/src/unified_exec/process_manager_tests.rs b/codex-rs/core/src/unified_exec/process_manager_tests.rs index a84f6345ede6..49758ec7998f 100644 --- a/codex-rs/core/src/unified_exec/process_manager_tests.rs +++ b/codex-rs/core/src/unified_exec/process_manager_tests.rs @@ -203,9 +203,10 @@ async fn failed_initial_end_for_unstored_process_uses_fallback_output() { cwd: turn.cwd.clone(), #[allow(deprecated)] sandbox_cwd: turn.cwd.clone(), - environment: turn + turn_environment: turn .environments - .primary_environment() + .primary() + .cloned() .expect("primary environment"), shell_mode: codex_tools::UnifiedExecShellMode::Direct, network: None, diff --git a/codex-rs/core/tests/suite/agents_md.rs b/codex-rs/core/tests/suite/agents_md.rs index 4950ee6387a1..a4ab404065dc 100644 --- a/codex-rs/core/tests/suite/agents_md.rs +++ b/codex-rs/core/tests/suite/agents_md.rs @@ -30,7 +30,6 @@ use core_test_support::test_codex::RecordingUserInstructionsProvider; use core_test_support::test_codex::TestCodexBuilder; use core_test_support::test_codex::test_codex; use core_test_support::wait_for_event; -use core_test_support::wait_for_event_match; use pretty_assertions::assert_eq; use serde_json::json; use std::sync::Arc; @@ -718,8 +717,7 @@ async fn multi_environment_thread_loads_every_project_and_keeps_creation_snapsho } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn global_loading_warning_surfaces_during_thread_creation() -> Result<()> { - // Set up a malformed global instruction file and one model response. +async fn invalid_utf8_global_instructions_are_lossy() -> Result<()> { let server = responses::start_mock_server().await; let response_mock = responses::mount_sse_once( &server, @@ -736,29 +734,12 @@ async fn global_loading_warning_surfaces_during_thread_creation() -> Result<()> b"global\xFFinstructions", )?; - // Create the thread, capture its load warning, and submit one turn for rendered output. let mut builder = test_codex().with_home(home); let test = builder.build(&server).await?; - let warning = wait_for_event_match(&test.codex, |event| match event { - EventMsg::Warning(warning) - if warning - .message - .contains(source.as_path().display().to_string().as_str()) => - { - Some(warning.message.clone()) - } - _ => None, - }) - .await; test.submit_turn("inspect lossy global instructions") .await?; - // Assert the source is reported, the warning is specific, and rendering is lossily decoded. assert_eq!(test.codex.instruction_sources().await, vec![source.clone()]); - assert!( - warning.contains("invalid UTF-8"), - "expected warning to contain \"invalid UTF-8\"; observed: {warning}" - ); let expected_fragment = expected_provider_only_instruction_fragment("global\u{FFFD}instructions"); assert_single_instruction_fragment(&response_mock.single_request(), &expected_fragment); diff --git a/codex-rs/exec/tests/suite/agents_md.rs b/codex-rs/exec/tests/suite/agents_md.rs index d4a765f5c6a3..2891721ae64e 100644 --- a/codex-rs/exec/tests/suite/agents_md.rs +++ b/codex-rs/exec/tests/suite/agents_md.rs @@ -2,8 +2,6 @@ use core_test_support::responses; use core_test_support::test_codex_exec::test_codex_exec; -use predicates::prelude::PredicateBooleanExt; -use predicates::str::contains; #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn exec_includes_workspace_agents_md_in_request() -> anyhow::Result<()> { @@ -74,103 +72,3 @@ async fn exec_prefers_workspace_agents_override_md() -> anyhow::Result<()> { Ok(()) } - -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn exec_surfaces_project_instruction_loading_warnings() -> anyhow::Result<()> { - let test = test_codex_exec(); - let project_agents_path = test.cwd_path().join("AGENTS.md"); - std::fs::write(&project_agents_path, b"project\xFFinstructions")?; - - let server = responses::start_mock_server().await; - let body = responses::sse(vec![ - responses::ev_response_created("resp1"), - responses::ev_assistant_message("m1", "fixture hello"), - responses::ev_completed("resp1"), - ]); - responses::mount_sse_once(&server, body).await; - - test.cmd_with_server(&server) - .arg("--skip-git-repo-check") - .arg("tell me something") - .assert() - .success() - .stderr(contains("invalid UTF-8").and(contains(project_agents_path.display().to_string()))); - - Ok(()) -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn exec_json_surfaces_project_instruction_loading_warnings() -> anyhow::Result<()> { - let test = test_codex_exec(); - let project_agents_path = test.cwd_path().join("AGENTS.md"); - std::fs::write(&project_agents_path, b"project\xFFinstructions")?; - - let server = responses::start_mock_server().await; - let body = responses::sse(vec![ - responses::ev_response_created("resp1"), - responses::ev_assistant_message("m1", "fixture hello"), - responses::ev_completed("resp1"), - ]); - responses::mount_sse_once(&server, body).await; - - let output = test - .cmd_with_server(&server) - .arg("--skip-git-repo-check") - .arg("--json") - .arg("tell me something") - .assert() - .success() - .get_output() - .stdout - .clone(); - let events = String::from_utf8(output)? - .lines() - .map(serde_json::from_str::) - .collect::, _>>()?; - - assert!( - events.iter().any(|event| { - event["type"] == "item.completed" - && event["item"]["type"] == "error" - && event["item"]["message"].as_str().is_some_and(|message| { - message.contains("invalid UTF-8") - && message.contains(project_agents_path.display().to_string().as_str()) - }) - }), - "expected a JSONL warning event; observed: {events:?}" - ); - - Ok(()) -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn exec_surfaces_global_instruction_loading_warnings() -> anyhow::Result<()> { - let test = test_codex_exec(); - let global_agents_path = test.home_path().join("AGENTS.md"); - let global_agents_source_suffix = format!( - "{}{}AGENTS.md", - test.home_path() - .file_name() - .expect("temporary Codex home should have a file name") - .to_string_lossy(), - std::path::MAIN_SEPARATOR, - ); - std::fs::write(&global_agents_path, b"global\xFFinstructions")?; - - let server = responses::start_mock_server().await; - let body = responses::sse(vec![ - responses::ev_response_created("resp1"), - responses::ev_assistant_message("m1", "fixture hello"), - responses::ev_completed("resp1"), - ]); - responses::mount_sse_once(&server, body).await; - - test.cmd_with_server(&server) - .arg("--skip-git-repo-check") - .arg("tell me something") - .assert() - .success() - .stderr(contains("invalid UTF-8").and(contains(global_agents_source_suffix))); - - Ok(()) -}