Use captured environment state for permissions and daemon recovery (#46568)

## Why

When a selected environment is unavailable, permission fallbacks can use the turn's initial environment and workspace roots. Daemon recovery can also miss a local environment that finishes starting after the turn begins.

## What changed

- Resolve fallback permissions from thread defaults using the captured selection's workspace root URIs, including when setup is starting or has failed.
- Apply this resolution to Guardian reviews, permission instructions, and sandbox context selection.
- Capture daemon recovery from the latest step inputs and refresh environment readiness before checking recovery eligibility.

## Testing

Add regression tests for Guardian resolving project-root restrictions against an unavailable captured environment's workspace and for daemon recovery including a local environment that finishes starting during a turn.

GitOrigin-RevId: c76d83d786fdc9c444b78ecbc335b02a20805c58

codex-rs/core/src/environment_selection.rs

@@ -1018,13 +1018,21 @@ impl TurnEnvironmentSnapshot {
 
     /// Returns the first selected environment's host folders, even if setup is not ready yet.
     pub(crate) fn primary_workspace_roots(&self) -> Vec<AbsolutePathBuf> {
+        self.primary_workspace_root_uris()
+            .iter()
+            .filter_map(|root| root.to_abs_path().ok())
+            .collect()
+    }
+
+    /// Returns the first selection's executor paths, even when setup is not ready yet.
+    pub(crate) fn primary_workspace_root_uris(&self) -> &[PathUri] {
         let selection = match self.environments.first() {
             Some(TurnEnvironmentState::Ready(environment)) => &environment.selection,
             Some(TurnEnvironmentState::Starting(environment)) => &environment.selection,
             Some(TurnEnvironmentState::Failed { selection, .. }) => selection,
-            None => return Vec::new(),
+            None => return &[],
         };
-        ThreadEnvironments::primary_workspace_roots_for(std::slice::from_ref(selection))
+        &selection.workspace_roots
     }
 
     /// Returns the primary environment's resolved permissions, or the provided fallback.

codex-rs/core/src/guardian/prompt.rs

@@ -231,7 +231,7 @@ fn parent_turn_permissions(
         .and_then(|environment| environment.cwd().to_abs_path().ok());
     let permission_profile = environment
         .map(TurnEnvironment::permission_profile_with_workspace_roots)
-        .unwrap_or_else(|| turn.permission_profile());
+        .unwrap_or_else(|| turn.permission_profile_for_environments(context.environments()));
     let file_system_policy = permission_profile.file_system_sandbox_policy();
     // Remote restrictions must not be interpreted using the filesystem running Guardian.
     // Older executors may not report their temp folders. If a rule explicitly denies those

codex-rs/core/src/guardian/tests.rs

@@ -673,6 +673,72 @@ async fn background_approval_permissions_use_the_owning_environment() -> anyhow:
     Ok(())
 }
 
+#[tokio::test(flavor = "current_thread")]
+async fn guardian_uses_thread_permissions_for_an_unavailable_captured_environment()
+-> anyhow::Result<()> {
+    let (session, mut turn) = crate::session::tests::make_session_and_context().await;
+    let original_root = test_path_buf("/original-workspace").abs();
+    let captured_root = test_path_buf("/captured-workspace").abs();
+    let thread_profile = PermissionProfile::from_runtime_permissions(
+        &FileSystemSandboxPolicy::restricted(vec![
+            FileSystemSandboxEntry::new(
+                FileSystemPath::Special {
+                    value: codex_protocol::permissions::FileSystemSpecialPath::Root,
+                },
+                FileSystemAccessMode::Read,
+            ),
+            FileSystemSandboxEntry::new(
+                FileSystemPath::Special {
+                    value: codex_protocol::permissions::FileSystemSpecialPath::project_roots(Some(
+                        "private".into(),
+                    )),
+                },
+                FileSystemAccessMode::Deny,
+            ),
+        ]),
+        NetworkSandboxPolicy::Restricted,
+    );
+    let permissions = &mut Arc::make_mut(&mut turn.config).permissions;
+    permissions.set_permission_profile(thread_profile)?;
+    permissions.set_workspace_roots(vec![original_root.clone()]);
+    let TurnEnvironmentState::Ready(original) = &mut turn.initial_environments.environments[0]
+    else {
+        panic!("initial environment should be ready");
+    };
+    original.config_mut().permission_profile =
+        PermissionProfileSnapshot::legacy(PermissionProfile::Disabled);
+    let mut selection = original.selection();
+    selection.cwd = PathUri::from_abs_path(&captured_root);
+    selection.workspace_roots = vec![selection.cwd.clone()];
+    selection.config = codex_protocol::protocol::EnvironmentConfigState::Failed("offline".into());
+    let captured = crate::environment_selection::TurnEnvironmentSnapshot {
+        environments: vec![TurnEnvironmentState::Failed {
+            selection,
+            error: "offline".into(),
+        }],
+    };
+    let turn = Arc::new(turn);
+    let context = GuardianReviewContext::from_resolved_settings(
+        Arc::clone(&turn),
+        &turn.initial_settings,
+        &captured,
+    );
+    let prompt = build_guardian_prompt_items_with_parent_turn(
+        &session,
+        session.conversation_history_snapshot().await.as_ref(),
+        Some(&context),
+        ApprovalRequestReasons::default(),
+        guardian_exec_command_request("shell"),
+        GuardianPromptMode::Full,
+        /*reviewed_node_repl_evidence_sequence*/ 0,
+    )
+    .await?;
+    let text = guardian_prompt_text(&prompt.context.into_user_inputs()?);
+    assert!(text.contains(captured_root.join("private").to_string_lossy().as_ref()));
+    assert!(!text.contains(original_root.join("private").to_string_lossy().as_ref()));
+    Ok(())
+}
+
 #[tokio::test(flavor = "current_thread")]
 async fn build_guardian_prompt_delta_mode_preserves_original_numbering() -> anyhow::Result<()> {
     let (session, turn) = guardian_test_session_and_turn_with_base_url("http://localhost").await;

codex-rs/core/src/session/daemon_recovery.rs

@@ -20,14 +20,13 @@ impl Session {
         }
         let context = &task.turn_context;
         context.extension_data.get::<RecordedTurnInput>()?;
+        let inputs = context.next_step_input.load();
         // Remote identities/configuration are not persisted across daemon restarts.
-        if context.initial_environments.environments.len() != 1 {
+        if inputs.environments.environments.len() != 1 {
             return None;
         }
-        let environment = context
-            .initial_environments
-            .single_local_environment()?
-            .selection();
+        let environments = inputs.environments.refresh_readiness();
+        let environment = environments.single_local_environment()?.selection();
         if environment.config != EnvironmentConfigState::FromThread {
             return None;
         }
@@ -35,18 +34,9 @@ impl Session {
             context.sub_id.clone(),
             crate::TurnStartOptions {
                 final_output_json_schema: context.final_output_json_schema.clone(),
-                service_tier: Some(
-                    context
-                        .next_step_input
-                        .load()
-                        .settings
-                        .service_tier
-                        .clone()
-                        .unwrap_or_else(|| {
-                            codex_protocol::config_types::SERVICE_TIER_DEFAULT_REQUEST_VALUE
-                                .to_string()
-                        }),
-                ),
+                service_tier: Some(inputs.settings.service_tier.clone().unwrap_or_else(|| {
+                    codex_protocol::config_types::SERVICE_TIER_DEFAULT_REQUEST_VALUE.to_string()
+                })),
                 cyber_access_program: context.cyber_access_program,
                 ..Default::default()
             },

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

@@ -3753,8 +3753,8 @@ impl Session {
             if !discovery.sandbox_contexts().is_empty() {
                 extension_data.insert(discovery.sandbox_contexts().clone());
             }
-        } else if !environments
-            .permission_profile_or_else(|| turn_context.permission_profile())
+        } else if !turn_context
+            .permission_profile_for_environments(&environments)
             .file_system_sandbox_policy()
             .has_full_disk_read_access()
         {

codex-rs/core/src/session/turn_context.rs

@@ -541,6 +541,22 @@ impl TurnContext {
             .permission_profile_or_else(|| self.config.permissions.effective_permission_profile())
     }
 
+    /// Uses this selection's permissions, or the turn's thread defaults for these folders.
+    pub(crate) fn permission_profile_for_environments(
+        &self,
+        environments: &TurnEnvironmentSnapshot,
+    ) -> PermissionProfile {
+        environments.permission_profile_or_else(|| {
+            self.config
+                .permissions
+                .permission_profile()
+                .clone()
+                .materialize_project_roots_with_path_uris(
+                    environments.primary_workspace_root_uris(),
+                )
+        })
+    }
+
     pub(crate) fn file_system_sandbox_policy(&self) -> FileSystemSandboxPolicy {
         self.permission_profile().file_system_sandbox_policy()
     }

codex-rs/core/src/session/world_state.rs

@@ -137,9 +137,8 @@ impl Session {
             .current_for_prefix_rules(turn_context.allow_prefix_rules());
         if turn_context.config.include_permissions_instructions {
             let environment = step_context.environments.primary();
-            let permission_profile = step_context
-                .environments
-                .permission_profile_or_else(|| turn_context.permission_profile());
+            let permission_profile =
+                turn_context.permission_profile_for_environments(&step_context.environments);
             #[allow(deprecated)]
             let cwd = environment
                 .and_then(|environment| environment.cwd().to_abs_path().ok())

codex-rs/core/tests/suite/turn_input_submission.rs

@@ -13,6 +13,7 @@ use codex_core::context::InternalContextSource;
 use codex_core::context::InternalModelContextFragment;
 use codex_extension_api::ExtensionRegistryBuilder;
 use codex_extension_api::TurnStartAdmission;
+use codex_features::Feature;
 use codex_protocol::AgentPath;
 use codex_protocol::config_types::CollaborationMode;
 use codex_protocol::config_types::ModeKind;
@@ -1005,3 +1006,85 @@ async fn sampling_is_ready_for_daemon_recovery(
     .await;
     Ok(())
 }
+
+#[tokio::test(flavor = "current_thread")]
+async fn daemon_recovery_includes_local_environment_that_finished_starting() -> anyhow::Result<()> {
+    let (release, gate) = oneshot::channel();
+    let (server, _completions) = start_streaming_sse_server(vec![
+        vec![StreamingSseChunk {
+            gate: None,
+            body: responses::sse(vec![
+                ev_response_created("wait"),
+                responses::ev_function_call(
+                    "wait-local",
+                    "wait_for_environment",
+                    r#"{"environment_id":"local"}"#,
+                ),
+                ev_completed("wait"),
+            ]),
+        }],
+        vec![StreamingSseChunk {
+            gate: Some(gate),
+            body: responses::sse_completed("done"),
+        }],
+    ])
+    .await;
+    let test = test_codex()
+        .with_config(|config| {
+            config.features.enable(Feature::DeferredExecutor).unwrap();
+        })
+        .build_with_streaming_server(&server)
+        .await?;
+    let cwd = test.config.cwd.join("new-workspace");
+    std::fs::create_dir(&cwd)?;
+    let selection = local(cwd.clone());
+    // A different workspace starts a new attachment. On this single-threaded runtime,
+    // turn startup captures it before the spawned setup task can run.
+    let started = test
+        .codex
+        .start_turn_if_idle(
+            user_message_request("wait for the environment").with_thread_settings(
+                ThreadSettingsOverrides {
+                    environments: Some(TurnEnvironmentSelections::new(
+                        cwd,
+                        vec![selection.clone()],
+                    )),
+                    ..Default::default()
+                },
+            ),
+        )
+        .await?;
+    let StartIfIdleSubmission::Started { turn_id } = started else {
+        anyhow::bail!("turn should start");
+    };
+
+    timeout(
+        Duration::from_secs(5),
+        server.wait_for_request_count(/*count*/ 2),
+    )
+    .await?;
+    let request: Value = serde_json::from_slice(&server.requests().await[1])?;
+    let output = request["input"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .find(|item| item["type"] == "function_call_output" && item["call_id"] == "wait-local")
+        .expect("the second request should contain the wait result");
+    assert_eq!(
+        serde_json::from_str::<Value>(output["output"].as_str().unwrap())?,
+        serde_json::json!({"environment_id": "local", "status": "ready"}),
+    );
+    assert_eq!(
+        test.codex
+            .interrupted_turn()
+            .await
+            .map(|(id, _, environment)| (id, environment)),
+        Some((turn_id, selection)),
+    );
+    release.send(()).expect("the model response is waiting");
+    wait_for_event(&test.codex, |event| {
+        matches!(event, EventMsg::TurnComplete(_))
+    })
+    .await;
+    Ok(())
+}