Use step environments for Guardian approval reviews (#37618)

## Why

Deferred environments can become ready after a turn starts. Guardian approval
reviews must use the environment selected for the current step so their working
directory and permission context are not taken from the stale turn snapshot.

## What changed

- Carry `StepContext` through tool execution and approval handling.
- Add a Guardian review context that pairs the parent turn with the current
  step's environment snapshot.
- Use that snapshot for Guardian prompts, reviewer sessions, and delegated
  reviewer threads while retaining turn-level fallbacks for callers without a
  step context.

## Testing

Added an integration test that makes a deferred remote environment ready before
an escalated command and verifies Guardian receives that environment's working
directory and denied-read policy instead of the initial local environment's.

GitOrigin-RevId: 779096bc7d0e0c7453a009c02141029714b9b407

codex-rs/core/src/unified_exec/async_watcher.rs

@@ -56,7 +56,7 @@ pub(crate) fn start_streaming_output(
     } = process.output_handles().clone();
 
     let session_ref = Arc::clone(&context.session);
-    let turn_ref = Arc::clone(&context.turn);
+    let turn_ref = Arc::clone(&context.step_context.turn);
     let call_id = context.call_id.clone();
 
     tokio::spawn(async move {

codex-rs/core/src/unified_exec/async_watcher_tests.rs

@@ -49,7 +49,11 @@ async fn streaming_output_harness() -> anyhow::Result<StreamingOutputHarness> {
             .await?,
     );
     let (session, turn, rx_event) = make_session_and_context_with_rx().await;
-    let context = UnifiedExecContext::new(session, turn, "streaming-output-test".to_string());
+    let context = UnifiedExecContext::new(
+        session,
+        crate::session::step_context::StepContext::for_test(turn),
+        "streaming-output-test".to_string(),
+    );
     let transcript = Arc::new(tokio::sync::Mutex::new(HeadTailBuffer::default()));
     start_streaming_output(&process, &context, Arc::clone(&transcript));
 
@@ -154,11 +158,11 @@ async fn exit_watcher_waits_for_late_network_denial_before_classifying_end() ->
     late_denial_armed_rx.await.expect("late denial armed");
 
     #[allow(deprecated)]
-    let cwd = context.turn.cwd.clone().into();
+    let cwd = context.step_context.turn.cwd.clone().into();
     spawn_exit_watcher(
         Arc::clone(&process),
         Arc::clone(&context.session),
-        Arc::clone(&context.turn),
+        Arc::clone(&context.step_context.turn),
         context.call_id,
         vec!["proof".to_string()],
         cwd,

codex-rs/core/src/unified_exec/mod.rs

@@ -38,6 +38,7 @@ use tokio::sync::Mutex;
 
 use crate::sandboxing::SandboxPermissions;
 use crate::session::session::Session;
+use crate::session::step_context::StepContext;
 use crate::session::turn_context::TurnContext;
 use crate::session::turn_context::TurnEnvironment;
 use crate::shell::ShellType;
@@ -74,15 +75,15 @@ pub(crate) const MAX_UNIFIED_EXEC_PROCESSES: usize = 64;
 
 pub(crate) struct UnifiedExecContext {
     pub session: Arc<Session>,
-    pub turn: Arc<TurnContext>,
+    pub step_context: Arc<StepContext>,
     pub call_id: String,
 }
 
 impl UnifiedExecContext {
-    pub fn new(session: Arc<Session>, turn: Arc<TurnContext>, call_id: String) -> Self {
+    pub fn new(session: Arc<Session>, step_context: Arc<StepContext>, call_id: String) -> Self {
         Self {
             session,
-            turn,
+            step_context,
             call_id,
         }
     }

codex-rs/core/src/unified_exec/mod_tests.rs

@@ -124,8 +124,11 @@ async fn exec_command_with_tty(
             )
             .await?,
     );
-    let context =
-        UnifiedExecContext::new(Arc::clone(session), Arc::clone(turn), "call".to_string());
+    let context = UnifiedExecContext::new(
+        Arc::clone(session),
+        crate::session::step_context::StepContext::for_test(Arc::clone(turn)),
+        "call".to_string(),
+    );
     let started_at = Instant::now();
     let process_started_alive = !process.has_exited() && process.exit_code().is_none();
     if process_started_alive {

codex-rs/core/src/unified_exec/process_manager.rs

@@ -344,7 +344,7 @@ async fn emit_failed_initial_exec_end_if_unstored(
 
     emit_failed_exec_end_for_unified_exec(
         Arc::clone(&context.session),
-        Arc::clone(&context.turn),
+        Arc::clone(&context.step_context.turn),
         context.call_id.clone(),
         request.command.clone(),
         cwd,
@@ -449,12 +449,13 @@ impl UnifiedExecProcessManager {
         let transcript = Arc::new(tokio::sync::Mutex::new(HeadTailBuffer::default()));
         let event_ctx = ToolEventCtx::new(
             context.session.as_ref(),
-            context.turn.as_ref(),
+            context.step_context.turn.as_ref(),
             &context.call_id,
             /*turn_diff_tracker*/ None,
         );
         let plugin_attribution = cwd.to_abs_path().ok().and_then(|cwd| {
             context
+                .step_context
                 .turn
                 .plugin_attribution_for_command(&request.command, &cwd)
         });
@@ -627,7 +628,7 @@ impl UnifiedExecProcessManager {
             let exit = exit_code.unwrap_or(-1);
             emit_exec_end_for_unified_exec(
                 Arc::clone(&context.session),
-                Arc::clone(&context.turn),
+                Arc::clone(&context.step_context.turn),
                 context.call_id.clone(),
                 request.command.clone(),
                 cwd.clone(),
@@ -655,7 +656,12 @@ impl UnifiedExecProcessManager {
             chunk_id,
             wall_time,
             raw_output: collected,
-            truncation_policy: context.turn.model_info.truncation_policy.into(),
+            truncation_policy: context
+                .step_context
+                .turn
+                .model_info
+                .truncation_policy
+                .into(),
             max_output_tokens: request.max_output_tokens,
             process_id: response_process_id,
             exit_code,
@@ -952,7 +958,7 @@ impl UnifiedExecProcessManager {
         spawn_exit_watcher(
             Arc::clone(&process),
             Arc::clone(&context.session),
-            Arc::clone(&context.turn),
+            Arc::clone(&context.step_context.turn),
             context.call_id.clone(),
             command.to_vec(),
             cwd,
@@ -1132,8 +1138,9 @@ impl UnifiedExecProcessManager {
         cwd: PathUri,
         context: &UnifiedExecContext,
     ) -> Result<(UnifiedExecProcess, Option<DeferredNetworkApproval>), UnifiedExecError> {
+        let turn = &context.step_context.turn;
         let local_policy_env = create_env(
-            &context.turn.config.permissions.shell_environment_policy,
+            &turn.config.permissions.shell_environment_policy,
             /*thread_id*/ None,
         );
         let mut env = local_policy_env.clone();
@@ -1146,7 +1153,7 @@ impl UnifiedExecProcessManager {
         let env = apply_unified_exec_env(env);
         let exec_server_env_config = ExecServerEnvConfig {
             policy: exec_env_policy_from_shell_policy(
-                &context.turn.config.permissions.shell_environment_policy,
+                &turn.config.permissions.shell_environment_policy,
             ),
             local_policy_env,
         };
@@ -1158,9 +1165,9 @@ impl UnifiedExecProcessManager {
             .exec_policy
             .create_exec_approval_requirement_for_command(ExecApprovalRequest {
                 command: &request.command,
-                approval_policy: context.turn.approval_policy(),
+                approval_policy: turn.approval_policy(),
                 permission_profile: request.turn_environment.permission_profile().clone(),
-                windows_sandbox_level: context.turn.windows_sandbox_level,
+                windows_sandbox_level: turn.windows_sandbox_level,
                 sandbox_permissions: if request.additional_permissions_preapproved {
                     crate::sandboxing::SandboxPermissions::UseDefault
                 } else {
@@ -1180,8 +1187,7 @@ impl UnifiedExecProcessManager {
             turn_environment: request.turn_environment.clone(),
             env,
             exec_server_env_config: Some(exec_server_env_config),
-            explicit_env_overrides: context
-                .turn
+            explicit_env_overrides: turn
                 .config
                 .permissions
                 .shell_environment_policy
@@ -1198,18 +1204,12 @@ impl UnifiedExecProcessManager {
         };
         let tool_ctx = ToolCtx {
             session: context.session.clone(),
-            turn: context.turn.clone(),
+            step_context: Arc::clone(&context.step_context),
             call_id: context.call_id.clone(),
             tool_name: ToolName::plain("exec_command"),
         };
         orchestrator
-            .run(
-                &mut runtime,
-                &req,
-                &tool_ctx,
-                &context.turn,
-                context.turn.approval_policy(),
-            )
+            .run(&mut runtime, &req, &tool_ctx, turn, turn.approval_policy())
             .await
             .map(|result| (result.output, result.deferred_network_approval))
             .map_err(|err| match err {

codex-rs/core/src/unified_exec/process_manager_tests.rs

@@ -377,7 +377,7 @@ async fn failed_initial_end_for_unstored_process_uses_fallback_output() {
     let (session, turn, rx_event) = crate::session::tests::make_session_and_context_with_rx().await;
     let context = UnifiedExecContext::new(
         Arc::clone(&session),
-        Arc::clone(&turn),
+        crate::session::step_context::StepContext::for_test(Arc::clone(&turn)),
         "call-unified-denied".to_string(),
     );
     let request = ExecCommandRequest {