[codex] Bind shell snapshots to retained thread environments (#28421)

## Why

Shell snapshots are currently session-scoped even though shell and cwd
are properties of a selected turn environment. That makes snapshot
refresh depend on separate session-cwd plumbing, prevents retained
environments from retaining their snapshot work, and can make snapshot
construction use a different shell than command execution.

This follows #27955 by making the retained thread-environment service
own environment snapshot lifecycles. Session configuration remains the
requested selection state, while `ThreadEnvironments` remains the source
of successfully resolved environments.

## What changed

- Configure the shell-snapshot builder before initial environment
resolution.
- Start each local environment snapshot task when its `TurnEnvironment`
is built and retain that shared task while environment ID and cwd still
match.
- Inherit retained environment snapshots into spawned child threads.
- Carry the selected `TurnEnvironment` through shell runtimes so
snapshot construction and command execution use the same
environment-specific shell and cwd.
- Load project instructions and warm plugins/skills after initial
environment resolution.
- Continue decoding invalid UTF-8 instruction files lossily without
emitting a startup warning.
- Keep requested selections in `SessionConfiguration`; failed or
duplicate resolutions only affect the resolved environment snapshot.

## Validation

- `cargo check -p codex-core --tests`
- `just test -p codex-home instructions` (6 passed)
- Focused environment, instruction, shell-snapshot, and user-shell tests
(84 passed)
- Focused shell-snapshot, user-shell, and unified-exec tests (126
passed; two event-timing tests passed on retry)

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<BoxFuture<'static, Option<Arc<ShellSnapshotFile>>>>;
+
+#[derive(Clone)]
 pub(crate) struct TurnEnvironment {
     pub(crate) environment_id: String,
     pub(crate) environment: Arc<Environment>,
@@ -50,6 +57,7 @@ pub(crate) struct TurnEnvironment {
     cwd: AbsolutePathBuf,
     cwd_uri: PathUri,
     pub(crate) shell: Option<shell::Shell>,
+    pub(crate) shell_snapshot: ShellSnapshotTask,
 }
 
 impl TurnEnvironment {
@@ -66,9 +74,20 @@ impl TurnEnvironment {
             cwd,
             cwd_uri,
             shell,
+            shell_snapshot: futures::future::ready(None).boxed().shared(),
         }
     }
 
+    pub(crate) fn shell_snapshot(&self, cwd: &AbsolutePathBuf) -> Option<AbsolutePathBuf> {
+        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 {
         &self.cwd
     }
@@ -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()