fix(runtime): make permission posture live (#5025)

Make interactive Auto-Review reachable, keep permission posture live across mid-turn changes, and preserve fail-closed authorization receipts.

Repair path-included test harness compilation and make the real-PTY policy matrix assert that destructive work is held under Auto-Review while Ask and Full Access retain their documented behavior.

All substantive GitHub Actions and Buildkite #735 passed on the exact head. The lone red Claude review check was unavailable infrastructure with no findings.

crates/tui/src/child_env.rs

@@ -582,12 +582,7 @@ fn normalize_key(key: &OsStr) -> String {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use std::sync::{Mutex, OnceLock};
-
-    fn env_lock() -> &'static Mutex<()> {
-        static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
-        LOCK.get_or_init(|| Mutex::new(()))
-    }
+    use crate::test_support::EnvVarGuard;
 
     #[test]
     fn mcp_env_allowlist_inherits_base_keys() {
@@ -783,27 +778,12 @@ mod tests {
     #[cfg(windows)]
     #[test]
     fn sanitized_child_env_preserves_custom_sdk_root_vars() {
-        let _guard = env_lock().lock().expect("env lock");
-        let previous_sdk = std::env::var_os("BIMRV_SDK_ROOT");
-        let previous_secret = std::env::var_os("MY_SECRET_ROOT");
-        unsafe {
-            std::env::set_var("BIMRV_SDK_ROOT", r"F:\Lib\BimRv27.5");
-            std::env::set_var("MY_SECRET_ROOT", r"F:\Secrets");
-        }
+        let _guard = crate::test_support::lock_test_env();
+        let _sdk = EnvVarGuard::set("BIMRV_SDK_ROOT", r"F:\Lib\BimRv27.5");
+        let _secret = EnvVarGuard::set("MY_SECRET_ROOT", r"F:\Secrets");
 
         let env = sanitized_child_env(std::iter::empty::<(OsString, OsString)>());
 
-        unsafe {
-            match previous_sdk {
-                Some(value) => std::env::set_var("BIMRV_SDK_ROOT", value),
-                None => std::env::remove_var("BIMRV_SDK_ROOT"),
-            }
-            match previous_secret {
-                Some(value) => std::env::set_var("MY_SECRET_ROOT", value),
-                None => std::env::remove_var("MY_SECRET_ROOT"),
-            }
-        }
-
         assert!(
             env.iter()
                 .any(|(key, value)| key == "BIMRV_SDK_ROOT" && value == r"F:\Lib\BimRv27.5"),
@@ -962,19 +942,11 @@ mod tests {
 
     #[test]
     fn sanitized_mcp_env_passes_through_node_bootstrap() {
-        let _guard = env_lock().lock().expect("env lock");
-        let prev = std::env::var_os("NVM_DIR");
-        unsafe {
-            std::env::set_var("NVM_DIR", "/tmp/test-nvm");
-        }
+        let _guard = crate::test_support::lock_test_env();
+        let _nvm_dir = EnvVarGuard::set("NVM_DIR", "/tmp/test-nvm");
 
         let env = sanitized_mcp_env(std::iter::empty::<(OsString, OsString)>());
 
-        match prev {
-            Some(value) => unsafe { std::env::set_var("NVM_DIR", value) },
-            None => unsafe { std::env::remove_var("NVM_DIR") },
-        }
-
         let nvm_dir = env
             .iter()
             .find(|(key, _)| normalize_key(key) == "NVM_DIR")
@@ -984,23 +956,11 @@ mod tests {
 
     #[test]
     fn sanitized_mcp_env_drops_unrelated_secret_like_values() {
-        let _guard = env_lock().lock().expect("env lock");
-        let prev = std::env::var_os("DEEPSEEK_MCP_TEST_SECRET");
-        unsafe {
-            std::env::set_var("DEEPSEEK_MCP_TEST_SECRET", "should-not-leak");
-        }
+        let _guard = crate::test_support::lock_test_env();
+        let _secret = EnvVarGuard::set("DEEPSEEK_MCP_TEST_SECRET", "should-not-leak");
 
         let env = sanitized_mcp_env(std::iter::empty::<(OsString, OsString)>());
 
-        match prev {
-            Some(value) => unsafe {
-                std::env::set_var("DEEPSEEK_MCP_TEST_SECRET", value);
-            },
-            None => unsafe {
-                std::env::remove_var("DEEPSEEK_MCP_TEST_SECRET");
-            },
-        }
-
         assert!(
             env.iter().all(|(key, _)| key != "DEEPSEEK_MCP_TEST_SECRET"),
             "MCP env should not pass arbitrary parent vars"
@@ -1009,24 +969,16 @@ mod tests {
 
     #[test]
     fn reviewed_plugin_mcp_env_requires_explicit_proxy_provenance() {
-        let _guard = env_lock().lock().expect("env lock");
-        let previous = std::env::var_os("HTTP_PROXY");
-        unsafe {
-            let synthetic_proxy = format!(
-                "{}://{}:{}@{}",
-                "http", "fixture-user", "fixture-password", "127.0.0.1:9"
-            );
-            std::env::set_var("HTTP_PROXY", synthetic_proxy);
-        }
+        let _guard = crate::test_support::lock_test_env();
+        let synthetic_proxy = format!(
+            "{}://{}:{}@{}",
+            "http", "fixture-user", "fixture-password", "127.0.0.1:9"
+        );
+        let _proxy = EnvVarGuard::set("HTTP_PROXY", synthetic_proxy);
 
         let ambient = sanitized_plugin_mcp_env(std::iter::empty::<(OsString, OsString)>());
         let explicit = sanitized_plugin_mcp_env([("HTTP_PROXY", "http://proxy.invalid")]);
 
-        match previous {
-            Some(value) => unsafe { std::env::set_var("HTTP_PROXY", value) },
-            None => unsafe { std::env::remove_var("HTTP_PROXY") },
-        }
-
         assert!(
             ambient
                 .iter()
@@ -1040,23 +992,11 @@ mod tests {
 
     #[test]
     fn sanitized_child_env_drops_parent_secret_like_values() {
-        let _guard = env_lock().lock().expect("env lock");
-        let previous = std::env::var_os("DEEPSEEK_CHILD_ENV_TEST_SECRET");
-        unsafe {
-            std::env::set_var("DEEPSEEK_CHILD_ENV_TEST_SECRET", "parent-secret");
-        }
+        let _guard = crate::test_support::lock_test_env();
+        let _secret = EnvVarGuard::set("DEEPSEEK_CHILD_ENV_TEST_SECRET", "parent-secret");
 
         let env = sanitized_child_env(std::iter::empty::<(OsString, OsString)>());
 
-        match previous {
-            Some(value) => unsafe {
-                std::env::set_var("DEEPSEEK_CHILD_ENV_TEST_SECRET", value);
-            },
-            None => unsafe {
-                std::env::remove_var("DEEPSEEK_CHILD_ENV_TEST_SECRET");
-            },
-        }
-
         assert!(
             env.iter()
                 .all(|(key, _)| key != "DEEPSEEK_CHILD_ENV_TEST_SECRET")
@@ -1065,23 +1005,11 @@ mod tests {
 
     #[test]
     fn explicit_child_env_values_win_over_parent_allowlist() {
-        let _guard = env_lock().lock().expect("env lock");
-        let previous = std::env::var_os("PATH");
-        unsafe {
-            std::env::set_var("PATH", "/parent/bin");
-        }
+        let _guard = crate::test_support::lock_test_env();
+        let _path = EnvVarGuard::set("PATH", "/parent/bin");
 
         let env = sanitized_child_env([(OsString::from("PATH"), OsString::from("/explicit/bin"))]);
 
-        match previous {
-            Some(value) => unsafe {
-                std::env::set_var("PATH", value);
-            },
-            None => unsafe {
-                std::env::remove_var("PATH");
-            },
-        }
-
         let path = env
             .iter()
             .find(|(key, _)| normalize_key(key) == "PATH")
@@ -1091,38 +1019,13 @@ mod tests {
 
     #[test]
     fn sanitized_child_env_preserves_windows_toolchain_vars() {
-        let _guard = env_lock().lock().expect("env lock");
-        let prev_lib = std::env::var_os("LIB");
-        let prev_include = std::env::var_os("INCLUDE");
-        let prev_sdk = std::env::var_os("WINDOWSSDKDIR");
-        // SAFETY: serialised by env_lock above. Restoring after the
-        // assertion is also under the same guard so concurrent tests
-        // never see our staged values.
-        unsafe {
-            std::env::set_var("LIB", r"C:\sdk\lib");
-            std::env::set_var("INCLUDE", r"C:\sdk\include");
-            std::env::set_var("WINDOWSSDKDIR", r"C:\sdk");
-        }
+        let _guard = crate::test_support::lock_test_env();
+        let _lib = EnvVarGuard::set("LIB", r"C:\sdk\lib");
+        let _include = EnvVarGuard::set("INCLUDE", r"C:\sdk\include");
+        let _sdk = EnvVarGuard::set("WINDOWSSDKDIR", r"C:\sdk");
 
         let env = sanitized_child_env(std::iter::empty::<(OsString, OsString)>());
 
-        // Restore prior state before asserting so a panic still leaves
-        // the process env clean for the next test.
-        unsafe {
-            match prev_lib {
-                Some(value) => std::env::set_var("LIB", value),
-                None => std::env::remove_var("LIB"),
-            }
-            match prev_include {
-                Some(value) => std::env::set_var("INCLUDE", value),
-                None => std::env::remove_var("INCLUDE"),
-            }
-            match prev_sdk {
-                Some(value) => std::env::set_var("WINDOWSSDKDIR", value),
-                None => std::env::remove_var("WINDOWSSDKDIR"),
-            }
-        }
-
         assert!(
             env.iter()
                 .any(|(key, value)| key == "LIB" && value == r"C:\sdk\lib"),

crates/tui/src/core/authority.rs

@@ -458,6 +458,9 @@ pub(crate) enum ApprovalRequestDisposition {
     /// A forced (non-bypassable) policy hold arrived under a full-access
     /// posture that opens no modal: fail closed.
     AutoDenyFullAccessPolicyHold,
+    /// Auto-Review is autonomous: unresolved holds fail closed instead of
+    /// opening a user-approval modal.
+    AutoDenyAutoReview,
     /// approval_mode=Never: deny without a modal.
     AutoDenyNeverPosture,
     /// Open the approval modal.
@@ -481,6 +484,9 @@ pub(crate) fn resolve_approval_request_disposition(
     if session_denied {
         return ApprovalRequestDisposition::AutoDenySessionDenied;
     }
+    if authority.approval_mode_for_session() == ApprovalMode::Auto {
+        return ApprovalRequestDisposition::AutoDenyAutoReview;
+    }
     // The request exists, so the engine already resolved Prompt for the tool
     // itself. What remains is the posture question: how does this authority
     // treat an ordinary promptable tool?
@@ -642,6 +648,7 @@ mod tests {
     #[test]
     fn approval_request_disposition_preserves_legacy_branch_order() {
         let ask = authority(AppMode::Agent, false, ApprovalMode::Suggest);
+        let auto = authority(AppMode::Agent, false, ApprovalMode::Auto);
         let full_access = authority(AppMode::Agent, true, ApprovalMode::Bypass);
         let never = authority(AppMode::Agent, false, ApprovalMode::Never);
 
@@ -674,6 +681,12 @@ mod tests {
             resolve_approval_request_disposition(&never, false, false, false),
             ApprovalRequestDisposition::AutoDenyNeverPosture
         );
+        for force_prompt in [false, true] {
+            assert_eq!(
+                resolve_approval_request_disposition(&auto, false, false, force_prompt),
+                ApprovalRequestDisposition::AutoDenyAutoReview
+            );
+        }
         // Ask posture with no grant opens the modal.
         assert_eq!(
             resolve_approval_request_disposition(&ask, false, false, false),

crates/tui/src/core/engine.rs

@@ -86,6 +86,7 @@ use super::turn::{TurnContext, post_turn_snapshot, pre_turn_snapshot};
 
 const ENGINE_OP_CHANNEL_CAPACITY: usize = 32;
 const GOAL_CONTINUATION_FAILURE_DETAIL_MAX_BYTES: usize = 512;
+const PLAN_SHELL_NETWORK_DENIED_HINT: &str = "Shell command blocked: Plan mode runs shell commands in a read-only sandbox — no writes, no network. Use Act mode (`/mode act`) for any command that creates or modifies files, or that needs network access.";
 
 fn context_pressure_message(usage_percent: f64) -> Option<&'static str> {
     if usage_percent >= crate::tui::context_inspector::CONTEXT_CRITICAL_THRESHOLD_PERCENT {
@@ -541,6 +542,10 @@ pub struct EngineHandle {
     /// before it mutates turn state. Real engines own concrete provider I/O;
     /// explicit injected/mock engines own that seam themselves.
     client_preflight_required: bool,
+    /// Typed live permission authority shared with the running turn. A mode
+    /// change publishes here before its mailbox op is queued, so gates never
+    /// consult a stale per-turn copy.
+    live_runtime_authority: Arc<StdMutex<LiveRuntimeAuthorityState>>,
 }
 
 // `impl EngineHandle { ... }` moved to `engine/handle.rs` so the
@@ -586,6 +591,7 @@ pub struct Engine {
     active_route_limits: Option<codewhale_config::route::RouteLimits>,
     active_route_capabilities: codewhale_config::route::RouteCapabilities,
     rx_op: mpsc::Receiver<Op>,
+    live_runtime_authority: Arc<StdMutex<LiveRuntimeAuthorityState>>,
     /// Clone of the op-channel sender, so the engine can self-dispatch ops
     /// (e.g. a goal-continuation `SendMessage` after a turn completes).
     tx_op: mpsc::Sender<Op>,
@@ -653,6 +659,85 @@ pub struct Engine {
     shared_paused: Arc<StdMutex<bool>>,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+struct LiveRuntimeAuthority {
+    mode: AppMode,
+    allow_shell: bool,
+    trust_mode: bool,
+    auto_approve: bool,
+    approval_mode: crate::tui::approval::ApprovalMode,
+    configured_sandbox_mode: Option<String>,
+}
+
+impl LiveRuntimeAuthority {
+    fn from_fields(
+        mode: AppMode,
+        allow_shell: bool,
+        trust_mode: bool,
+        auto_approve: bool,
+        approval_mode: crate::tui::approval::ApprovalMode,
+        configured_sandbox_mode: Option<String>,
+    ) -> Self {
+        let authority = TurnAuthority::from_effective_fields(
+            mode,
+            allow_shell,
+            trust_mode,
+            auto_approve,
+            approval_mode,
+        );
+        Self::from_turn_authority(&authority, configured_sandbox_mode)
+    }
+
+    fn from_turn_authority(
+        authority: &TurnAuthority,
+        configured_sandbox_mode: Option<String>,
+    ) -> Self {
+        let approval_mode = authority.approval_mode_for_session();
+        Self {
+            mode: authority.mode,
+            allow_shell: authority.allow_shell,
+            trust_mode: authority.trust_mode,
+            auto_approve: authority.auto_approve
+                || approval_mode == crate::tui::approval::ApprovalMode::Bypass,
+            approval_mode,
+            configured_sandbox_mode,
+        }
+    }
+
+    fn permission_snapshot(&self) -> RuntimePermissionAuthority {
+        RuntimePermissionAuthority {
+            auto_approve: self.auto_approve,
+            trust_mode: self.trust_mode,
+            approval_mode: self.approval_mode,
+        }
+    }
+}
+
+#[derive(Debug)]
+struct LiveRuntimeAuthorityState {
+    revision: u64,
+    applied_revision: u64,
+    authority: LiveRuntimeAuthority,
+}
+
+impl LiveRuntimeAuthorityState {
+    fn new(authority: LiveRuntimeAuthority) -> Self {
+        Self {
+            revision: 0,
+            applied_revision: 0,
+            authority,
+        }
+    }
+}
+
+/// Runtime-facing view of the engine's exact live permission authority.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) struct RuntimePermissionAuthority {
+    pub(crate) auto_approve: bool,
+    pub(crate) trust_mode: bool,
+    pub(crate) approval_mode: crate::tui::approval::ApprovalMode,
+}
+
 fn claim_subagent_completion(
     delivered_ids: &mut HashSet<String>,
     completion: SubAgentCompletion,
@@ -1018,6 +1103,16 @@ impl Engine {
         let shared_cancel_token = Arc::new(StdMutex::new(cancel_token.clone()));
         let cancel_reason: Arc<StdMutex<Option<CancelReason>>> = Arc::new(StdMutex::new(None));
         let shared_paused = Arc::new(StdMutex::new(false));
+        let live_runtime_authority = Arc::new(StdMutex::new(LiveRuntimeAuthorityState::new(
+            LiveRuntimeAuthority::from_fields(
+                AppMode::Agent,
+                config.allow_shell,
+                config.trust_mode,
+                false,
+                crate::tui::approval::ApprovalMode::Suggest,
+                api_config.sandbox_mode.clone(),
+            ),
+        )));
         let tool_exec_lock = Arc::new(RwLock::new(()));
         let plugin_registry = config
             .plugin_registry
@@ -1183,6 +1278,7 @@ impl Engine {
             active_route_limits,
             active_route_capabilities: codewhale_config::route::RouteCapabilities::default(),
             rx_op,
+            live_runtime_authority: Arc::clone(&live_runtime_authority),
             tx_op: tx_op.clone(),
             scheduled_goal_continuation: None,
             goal_continuation_schedule_seq: 0,
@@ -1217,6 +1313,7 @@ impl Engine {
             tx_steer,
             shared_paused,
             client_preflight_required: true,
+            live_runtime_authority,
         };
 
         (engine, handle)
@@ -1527,6 +1624,108 @@ impl Engine {
         }
     }
 
+    /// Apply a user/host mode-or-posture change to the live session.
+    ///
+    /// Single authority source for mode/permission state: both the run loop
+    /// and the active turn's typed live-authority drain land here.
+    async fn apply_change_mode(
+        &mut self,
+        mode: AppMode,
+        allow_shell: bool,
+        trust_mode: bool,
+        auto_approve: bool,
+        approval_mode: crate::tui::approval::ApprovalMode,
+        configured_sandbox_mode: Option<String>,
+    ) {
+        let authority = TurnAuthority::from_effective_fields(
+            mode,
+            allow_shell,
+            trust_mode,
+            auto_approve,
+            approval_mode,
+        );
+        let effective_approval = authority.approval_mode_for_session();
+        let changed = self.current_mode != authority.mode
+            || self.session.allow_shell != authority.allow_shell
+            || self.session.trust_mode != authority.trust_mode
+            || self.session.auto_approve
+                != (authority.auto_approve
+                    || effective_approval == crate::tui::approval::ApprovalMode::Bypass)
+            || self.session.approval_mode != effective_approval
+            || self.api_config.sandbox_mode != configured_sandbox_mode;
+        self.api_config.sandbox_mode = configured_sandbox_mode;
+        self.apply_runtime_mode_policy(&authority);
+        if !changed {
+            return;
+        }
+        self.emit_session_updated().await;
+        let _ = self
+            .tx_event
+            .send(Event::status(format!(
+                "Runtime policy changed to: {} / {}",
+                mode.description(),
+                effective_approval.permission_chip_label(),
+            )))
+            .await;
+    }
+
+    fn take_pending_runtime_authority(&self) -> Option<LiveRuntimeAuthority> {
+        let mut state = self
+            .live_runtime_authority
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        if state.applied_revision == state.revision {
+            return None;
+        }
+        state.applied_revision = state.revision;
+        Some(state.authority.clone())
+    }
+
+    fn runtime_authority_snapshot(&self) -> LiveRuntimeAuthority {
+        self.live_runtime_authority
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .authority
+            .clone()
+    }
+
+    async fn apply_runtime_authority(&mut self, authority: LiveRuntimeAuthority) {
+        self.apply_change_mode(
+            authority.mode,
+            authority.allow_shell,
+            authority.trust_mode,
+            authority.auto_approve,
+            authority.approval_mode,
+            authority.configured_sandbox_mode,
+        )
+        .await;
+    }
+
+    async fn apply_pending_runtime_authority(&mut self) -> bool {
+        let Some(authority) = self.take_pending_runtime_authority() else {
+            return false;
+        };
+        self.apply_runtime_authority(authority).await;
+        true
+    }
+
+    fn record_applied_runtime_authority(&self, authority: &TurnAuthority) {
+        let applied = LiveRuntimeAuthority::from_turn_authority(
+            authority,
+            self.api_config.sandbox_mode.clone(),
+        );
+        let mut state = self
+            .live_runtime_authority
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        // Never overwrite a newer, not-yet-applied user change with the turn
+        // posture that preceded it.
+        if state.revision == state.applied_revision || state.authority == applied {
+            state.authority = applied;
+            state.applied_revision = state.revision;
+        }
+    }
+
     fn apply_runtime_mode_policy(&mut self, authority: &TurnAuthority) {
         // Mode doctrine lives in the stable prefix (#4780), so a mode change
         // has to rebuild it. `refresh_system_prompt` is hash-guarded and only
@@ -1544,6 +1743,7 @@ impl Engine {
         self.session.approval_mode = authority.approval_mode_for_session();
         self.session.auto_approve = authority.auto_approve
             || self.session.approval_mode == crate::tui::approval::ApprovalMode::Bypass;
+        self.record_applied_runtime_authority(authority);
     }
 
     fn schedule_goal_continuation(&mut self, dynamic_tools: Vec<DynamicToolSpec>) {
@@ -1732,6 +1932,15 @@ impl Engine {
                 break;
             };
 
+            // Runtime posture updates publish through shared typed state
+            // before attempting their best-effort wake-up. If the mailbox was
+            // already full, its next queued operation is the wake-up: apply
+            // the latest authority before doing any work under an obsolete
+            // policy.
+            if matches!(&input, EngineRunInput::Operation(_)) {
+                self.apply_pending_runtime_authority().await;
+            }
+
             match input {
                 EngineRunInput::SubAgentCompletion(completion) => {
                     self.handle_idle_subagent_completion(completion).await;
@@ -2090,31 +2299,13 @@ impl Engine {
                             }
                         }
                     }
-                    Op::ChangeMode {
-                        mode,
-                        allow_shell,
-                        trust_mode,
-                        auto_approve,
-                        approval_mode,
-                        configured_sandbox_mode,
-                    } => {
-                        let authority = TurnAuthority::from_effective_fields(
-                            mode,
-                            allow_shell,
-                            trust_mode,
-                            auto_approve,
-                            approval_mode,
-                        );
-                        self.api_config.sandbox_mode = configured_sandbox_mode;
-                        self.apply_runtime_mode_policy(&authority);
-                        self.emit_session_updated().await;
-                        let _ = self
-                            .tx_event
-                            .send(Event::status(format!(
-                                "Mode changed to: {}",
-                                mode.description()
-                            )))
-                            .await;
+                    Op::ChangeMode { .. } => {
+                        // The mailbox payload may predate a newer posture that
+                        // was published while the channel was full. Apply the
+                        // single live snapshot so a stale queued ChangeMode
+                        // can never roll authority backward.
+                        let authority = self.runtime_authority_snapshot();
+                        self.apply_runtime_authority(authority).await;
                     }
                     Op::SetModel {
                         model,
@@ -4455,6 +4646,34 @@ impl Engine {
         self.build_tool_context_for_turn(&authority, &route)
     }
 
+    /// Project the current engine authority onto an already-built registry.
+    /// Registries own long-lived services and tool definitions; permission,
+    /// shell, and sandbox policy are live turn state and must not be read from
+    /// the registry's start-of-turn snapshot after a Runtime posture switch.
+    fn live_tool_context(
+        &self,
+        registry: Option<&crate::tools::ToolRegistry>,
+    ) -> Option<ToolContext> {
+        let mut context = registry?.context().clone();
+        let authority = TurnAuthority::from_effective_fields(
+            self.current_mode,
+            self.session.allow_shell,
+            self.session.trust_mode,
+            self.session.auto_approve,
+            self.session.approval_mode,
+        );
+        context.trust_mode = authority.trust_mode;
+        context.auto_approve = authority.auto_approve;
+        context.shell_policy = authority.shell_policy();
+        context.elevated_sandbox_policy = Some(authority.sandbox_policy(
+            &self.session.workspace,
+            self.api_config.sandbox_mode.as_deref(),
+        ));
+        context.shell_network_denied_hint = matches!(authority.mode, AppMode::Plan)
+            .then(|| PLAN_SHELL_NETWORK_DENIED_HINT.to_string());
+        Some(context)
+    }
+
     /// Build one tool context from the already-resolved turn authority and
     /// route. A preview owns values that are deliberately not installed on the
     /// session; rebuilding either from `self.session` would give it the prior
@@ -4559,9 +4778,7 @@ impl Engine {
         );
         let mut ctx = ctx.with_elevated_sandbox_policy(policy);
         if matches!(authority.mode, AppMode::Plan) {
-            ctx = ctx.with_shell_network_denied_hint(
-                "Shell command blocked: Plan mode runs shell commands in a read-only sandbox — no writes, no network. Use Act mode (`/mode act`) for any command that creates or modifies files, or that needs network access.",
-            );
+            ctx = ctx.with_shell_network_denied_hint(PLAN_SHELL_NETWORK_DENIED_HINT);
         }
         ctx
     }
@@ -5013,6 +5230,7 @@ pub(super) enum ToolAskRuleDecision {
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub(super) enum AutoReviewPlanDecision {
     NoChange,
+    Allow,
     ForcePrompt(String),
     Block(String),
 }
@@ -5052,36 +5270,60 @@ pub(super) fn auto_review_plan_decision(
     );
     let decision = policy.evaluate(&context);
     let audit_event = policy.audit_event(&context, &decision);
-    let plan_decision = match decision.action {
-        crate::tui::auto_review::AutoReviewAction::Allow
-        | crate::tui::auto_review::AutoReviewAction::AskUser => AutoReviewPlanDecision::NoChange,
-        crate::tui::auto_review::AutoReviewAction::HoldForReview => {
-            // HoldForReview only originates from the built-in safety floor
-            // (configured rules produce Allow/Block), so name the gate
-            // honestly instead of blaming an "auto-review policy" the user
-            // may never have configured (#3883).
-            let reason = format!(
-                "Built-in safety gate requires approval: {}",
-                decision.reason
-            );
-            if matches!(
-                approval_mode,
-                crate::tui::approval::ApprovalMode::Never
-                    | crate::tui::approval::ApprovalMode::Bypass
-            ) {
-                // Never and Full Access are both non-interactive postures for
-                // approval holds. Full Access auto-runs ordinary calls, but a
-                // non-bypassable safety floor fails closed instead of opening
-                // a contradictory modal or being silently auto-approved.
-                AutoReviewPlanDecision::Block(reason)
-            } else {
-                AutoReviewPlanDecision::ForcePrompt(reason)
+    let plan_decision = if approval_mode == crate::tui::approval::ApprovalMode::Auto
+        && tool_name == REQUEST_USER_INPUT_NAME
+    {
+        // This synthetic tool does not execute user work. Let the turn loop
+        // return its ordinary autonomous guidance result instead of treating
+        // a hallucinated question as an unknown external action.
+        AutoReviewPlanDecision::Allow
+    } else {
+        match decision.action {
+            crate::tui::auto_review::AutoReviewAction::Allow
+                if approval_mode == crate::tui::approval::ApprovalMode::Auto =>
+            {
+                AutoReviewPlanDecision::Allow
+            }
+            crate::tui::auto_review::AutoReviewAction::Allow => AutoReviewPlanDecision::NoChange,
+            crate::tui::auto_review::AutoReviewAction::AskUser
+                if approval_mode == crate::tui::approval::ApprovalMode::Auto =>
+            {
+                AutoReviewPlanDecision::Block(format!(
+                    "Auto-Review held tool '{tool_name}': {}",
+                    decision.reason
+                ))
+            }
+            crate::tui::auto_review::AutoReviewAction::AskUser => AutoReviewPlanDecision::NoChange,
+            crate::tui::auto_review::AutoReviewAction::HoldForReview => {
+                // HoldForReview only originates from the built-in safety floor
+                // (configured rules produce Allow/Block), so name the gate
+                // honestly instead of blaming an "auto-review policy" the user
+                // may never have configured (#3883).
+                let reason = format!(
+                    "Built-in safety gate requires approval: {}",
+                    decision.reason
+                );
+                if matches!(
+                    approval_mode,
+                    crate::tui::approval::ApprovalMode::Auto
+                        | crate::tui::approval::ApprovalMode::Never
+                        | crate::tui::approval::ApprovalMode::Bypass
+                ) {
+                    // Auto-Review, Never, and Full Access are non-interactive for
+                    // approval holds. Full Access auto-runs ordinary calls, but a
+                    // non-bypassable safety floor always fails closed.
+                    AutoReviewPlanDecision::Block(reason)
+                } else {
+                    AutoReviewPlanDecision::ForcePrompt(reason)
+                }
+            }
+            crate::tui::auto_review::AutoReviewAction::Block => {
+                AutoReviewPlanDecision::Block(format!(
+                    "Auto-review policy blocked tool '{tool_name}': {}",
+                    decision.reason
+                ))
             }
         }
-        crate::tui::auto_review::AutoReviewAction::Block => AutoReviewPlanDecision::Block(format!(
-            "Auto-review policy blocked tool '{tool_name}': {}",
-            decision.reason
-        )),
     };
     (plan_decision, audit_event)
 }
@@ -5335,6 +5577,16 @@ pub(crate) fn mock_engine_handle() -> MockEngineHandle {
     let shared_cancel_token = Arc::new(StdMutex::new(cancel_token.clone()));
     let cancel_reason: Arc<StdMutex<Option<CancelReason>>> = Arc::new(StdMutex::new(None));
     let shared_paused = Arc::new(StdMutex::new(false));
+    let live_runtime_authority = Arc::new(StdMutex::new(LiveRuntimeAuthorityState::new(
+        LiveRuntimeAuthority::from_fields(
+            AppMode::Agent,
+            false,
+            false,
+            false,
+            crate::tui::approval::ApprovalMode::Suggest,
+            None,
+        ),
+    )));
     let handle = EngineHandle {
         tx_op,
         rx_event: Arc::new(RwLock::new(rx_event)),
@@ -5345,6 +5597,7 @@ pub(crate) fn mock_engine_handle() -> MockEngineHandle {
         tx_steer,
         shared_paused,
         client_preflight_required: false,
+        live_runtime_authority,
     };
 
     MockEngineHandle {

crates/tui/src/core/engine/handle.rs

@@ -12,7 +12,10 @@ use anyhow::Result;
 use tokio::sync::mpsc;
 
 use super::approval::{ApprovalDecision, UserInputDecision};
-use super::{CancelReason, EngineHandle, Op, UserInputResponse};
+use super::{
+    CancelReason, EngineHandle, LiveRuntimeAuthority, Op, RuntimePermissionAuthority,
+    UserInputResponse,
+};
 
 impl EngineHandle {
     /// True when the caller must preflight a concrete provider client before
@@ -25,7 +28,12 @@ impl EngineHandle {
 
     /// Send an operation to the engine
     pub async fn send(&self, op: Op) -> Result<()> {
-        self.tx_op.send(op).await?;
+        let authority = Self::change_mode_authority(&op);
+        let permit = self.tx_op.reserve().await?;
+        if let Some(authority) = authority {
+            self.publish_runtime_authority(authority);
+        }
+        permit.send(op);
         Ok(())
     }
 
@@ -35,10 +43,83 @@ impl EngineHandle {
     /// non-critical, refresh-type ops (e.g. `Op::ListSubAgents`) that can
     /// safely be dropped and re-requested on the next drain cycle.
     pub fn try_send(&self, op: Op) -> Result<()> {
-        self.tx_op.try_send(op)?;
+        let authority = Self::change_mode_authority(&op);
+        let result = self.tx_op.try_send(op);
+        // A full channel already guarantees that the engine will wake and
+        // drain an operation. Publish the typed authority anyway: the drain
+        // applies pending authority before handling that queued operation, so
+        // a posture edit never blocks behind refresh traffic. A closed
+        // channel has no engine left to observe the update.
+        if !matches!(&result, Err(mpsc::error::TrySendError::Closed(_)))
+            && let Some(authority) = authority
+        {
+            self.publish_runtime_authority(authority);
+        }
+        result?;
         Ok(())
     }
 
+    fn change_mode_authority(op: &Op) -> Option<LiveRuntimeAuthority> {
+        let Op::ChangeMode {
+            mode,
+            allow_shell,
+            trust_mode,
+            auto_approve,
+            approval_mode,
+            configured_sandbox_mode,
+        } = op
+        else {
+            return None;
+        };
+        Some(LiveRuntimeAuthority::from_fields(
+            *mode,
+            *allow_shell,
+            *trust_mode,
+            *auto_approve,
+            *approval_mode,
+            configured_sandbox_mode.clone(),
+        ))
+    }
+
+    fn publish_runtime_authority(&self, authority: LiveRuntimeAuthority) {
+        let mut state = self
+            .live_runtime_authority
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        state.revision = state.revision.wrapping_add(1).max(1);
+        state.authority = authority;
+    }
+
+    pub(crate) fn publish_turn_authority(
+        &self,
+        mode: crate::tui::app::AppMode,
+        allow_shell: bool,
+        trust_mode: bool,
+        auto_approve: bool,
+        approval_mode: crate::tui::approval::ApprovalMode,
+        configured_sandbox_mode: Option<String>,
+    ) {
+        self.publish_runtime_authority(LiveRuntimeAuthority::from_fields(
+            mode,
+            allow_shell,
+            trust_mode,
+            auto_approve,
+            approval_mode,
+            configured_sandbox_mode,
+        ));
+    }
+
+    /// Exact live permission authority for runtime approval and elevation
+    /// gates. This is the same typed state the active engine turn drains.
+    #[must_use]
+    pub(crate) fn runtime_permission_authority(&self) -> RuntimePermissionAuthority {
+        self.live_runtime_authority
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .authority
+            .permission_snapshot()
+    }
+
     /// Reserve capacity for a runtime steer before it mutates durable state.
     /// The owned permit lets the caller persist and dispatch synchronously,
     /// without a cancellation point between those two operations.

crates/tui/src/core/engine/tests.rs

@@ -4568,7 +4568,7 @@ async fn operate_conversation_reaches_provider_when_workers_are_disabled() {
 }
 
 #[test]
-fn auto_review_classifies_publish_and_force_prompts_it() {
+fn auto_review_classifies_publish_and_holds_without_prompting() {
     let (decision, audit) = auto_review_plan_decision(
         &crate::tui::auto_review::AutoReviewPolicy::default(),
         "exec_shell",
@@ -4582,7 +4582,7 @@ fn auto_review_classifies_publish_and_force_prompts_it() {
 
     assert_eq!(
         decision,
-        AutoReviewPlanDecision::ForcePrompt(
+        AutoReviewPlanDecision::Block(
             "Built-in safety gate requires approval: publish-like action requires durable review"
                 .to_string()
         )
@@ -4592,7 +4592,24 @@ fn auto_review_classifies_publish_and_force_prompts_it() {
 }
 
 #[test]
-fn auto_review_policy_does_not_force_prompt_for_shell_git_tag_list_probe() {
+fn auto_review_classifier_allow_executes_without_prompting() {
+    let (decision, audit) = auto_review_plan_decision(
+        &crate::tui::auto_review::AutoReviewPolicy::default(),
+        "read_file",
+        &json!({"path": "Cargo.toml"}),
+        crate::tui::auto_review::RunOrigin::Interactive,
+        crate::tui::approval::ApprovalMode::Auto,
+        Some("inspect the manifest"),
+        true,
+        false,
+    );
+
+    assert_eq!(decision, AutoReviewPlanDecision::Allow);
+    assert_eq!(audit["decision"], "allow");
+}
+
+#[test]
+fn auto_review_holds_unclassified_shell_probe_without_prompting() {
     let (decision, audit) = auto_review_plan_decision(
         &crate::tui::auto_review::AutoReviewPolicy::default(),
         "exec_shell",
@@ -4604,7 +4621,13 @@ fn auto_review_policy_does_not_force_prompt_for_shell_git_tag_list_probe() {
         false,
     );
 
-    assert_eq!(decision, AutoReviewPlanDecision::NoChange);
+    assert_eq!(
+        decision,
+        AutoReviewPlanDecision::Block(
+            "Auto-Review held tool 'exec_shell': destructive action requires explicit review"
+                .to_string()
+        )
+    );
     assert_eq!(audit["decision"], "ask_user");
     assert_eq!(audit["action_kind"], "shell");
 }
@@ -4704,7 +4727,7 @@ fn generic_required_tools_keep_auto_approve_behavior() {
 }
 
 #[test]
-fn auto_review_policy_does_not_change_generic_destructive_auto_approval_yet() {
+fn auto_review_holds_generic_destructive_call_without_prompting() {
     let (decision, audit) = auto_review_plan_decision(
         &crate::tui::auto_review::AutoReviewPolicy::default(),
         "exec_shell",
@@ -4716,7 +4739,13 @@ fn auto_review_policy_does_not_change_generic_destructive_auto_approval_yet() {
         false,
     );
 
-    assert_eq!(decision, AutoReviewPlanDecision::NoChange);
+    assert_eq!(
+        decision,
+        AutoReviewPlanDecision::Block(
+            "Auto-Review held tool 'exec_shell': destructive action requires explicit review"
+                .to_string()
+        )
+    );
     assert_eq!(audit["decision"], "ask_user");
     assert_eq!(audit["risk"], "destructive");
 }
@@ -6432,9 +6461,9 @@ fn print_mode_runtime_contract_metrics() {
     let _userprofile = EnvVarGuard::set("USERPROFILE", &home);
     let codewhale_home = home.join(".codewhale");
     let _codewhale_home = EnvVarGuard::set("CODEWHALE_HOME", &codewhale_home);
-    // Keep the model-visible shell fact stable across developer and CI hosts.
-    // ShellDispatcher recognizes this token without executing a shell.
-    let _shell = EnvVarGuard::set("SHELL", "bash");
+    // Keep the model-visible shell fact stable across developer and CI hosts
+    // while exercising the exact-path contract used at runtime.
+    let _shell = EnvVarGuard::set("SHELL", "/bin/bash");
     let mut mode_metrics = serde_json::Map::new();
     for (mode_name, mode, mode_instructions) in [
         ("plan", AppMode::Plan, crate::prompts::PLAN_MODE),
@@ -6572,8 +6601,9 @@ fn measure_representative_runtime_context()
     let _userprofile = EnvVarGuard::set("USERPROFILE", &home);
     let codewhale_home = home.join(".codewhale");
     let _codewhale_home = EnvVarGuard::set("CODEWHALE_HOME", &codewhale_home);
-    // Keep the model-visible shell fact stable across developer and CI hosts.
-    let _shell = EnvVarGuard::set("SHELL", "bash");
+    // Keep the model-visible shell fact stable across developer and CI hosts
+    // while exercising the exact-path contract used at runtime.
+    let _shell = EnvVarGuard::set("SHELL", "/bin/bash");
 
     let mut stages = vec![representative_stage(
         "base",
@@ -10016,6 +10046,76 @@ async fn change_mode_refreshes_session_prompt_and_updates_session() {
     );
 }
 
+#[tokio::test]
+async fn live_runtime_authority_applies_latest_posture_and_sandbox_before_tools() {
+    use crate::sandbox::SandboxPolicy;
+    use crate::tui::approval::ApprovalMode;
+
+    let tmp = tempdir().expect("tempdir");
+    let config = EngineConfig {
+        workspace: tmp.path().to_path_buf(),
+        ..Default::default()
+    };
+    let (mut engine, handle) = Engine::new(config, &Config::default());
+    let registry = ToolRegistryBuilder::new()
+        .build(engine.build_tool_context(engine.current_mode, engine.session.auto_approve));
+
+    for (mode, posture, auto_approve, sandbox_mode, expected_sandbox) in [
+        (
+            AppMode::Operate,
+            ApprovalMode::Auto,
+            false,
+            Some("read-only".to_string()),
+            SandboxPolicy::ReadOnly,
+        ),
+        (
+            AppMode::Agent,
+            ApprovalMode::Bypass,
+            true,
+            None,
+            SandboxPolicy::DangerFullAccess,
+        ),
+        (
+            AppMode::Agent,
+            ApprovalMode::Suggest,
+            false,
+            None,
+            SandboxPolicy::WorkspaceWrite {
+                writable_roots: vec![tmp.path().to_path_buf()],
+                network_access: true,
+                exclude_tmpdir: false,
+                exclude_slash_tmp: false,
+            },
+        ),
+    ] {
+        handle
+            .try_send(Op::ChangeMode {
+                mode,
+                allow_shell: true,
+                trust_mode: false,
+                auto_approve,
+                approval_mode: posture,
+                configured_sandbox_mode: sandbox_mode,
+            })
+            .expect("publish live runtime authority");
+
+        let published = handle.runtime_permission_authority();
+        assert_eq!(published.approval_mode, posture);
+        assert_eq!(published.auto_approve, auto_approve);
+        assert!(engine.apply_pending_runtime_authority().await);
+        assert_eq!(engine.current_mode, mode);
+        assert_eq!(engine.session.approval_mode, posture);
+        assert_eq!(engine.session.auto_approve, auto_approve);
+        assert_eq!(
+            engine
+                .live_tool_context(Some(&registry))
+                .expect("live registry context")
+                .elevated_sandbox_policy,
+            Some(expected_sandbox),
+        );
+    }
+}
+
 #[test]
 fn turn_approval_mode_prefers_auto_approve_flag() {
     use crate::tui::approval::ApprovalMode;
@@ -13944,6 +14044,16 @@ fn engine_handle_try_send_does_not_block_when_op_channel_is_full() {
         tx_steer: mpsc::channel(1).0,
         shared_paused: Arc::new(StdMutex::new(false)),
         client_preflight_required: true,
+        live_runtime_authority: Arc::new(StdMutex::new(LiveRuntimeAuthorityState::new(
+            LiveRuntimeAuthority::from_fields(
+                AppMode::Agent,
+                false,
+                false,
+                false,
+                crate::tui::approval::ApprovalMode::Suggest,
+                None,
+            ),
+        ))),
     };
 
     // Fill the op channel with one message (capacity = 1).
@@ -13952,9 +14062,82 @@ fn engine_handle_try_send_does_not_block_when_op_channel_is_full() {
         .try_send(Op::ListSubAgents)
         .expect("first send should succeed");
 
-    // try_send must return Err immediately — never block.
-    let result = handle.try_send(Op::ListSubAgents);
+    // A live posture update must publish immediately even though its wake-up
+    // cannot fit. The already-queued operation will wake the engine, which
+    // applies this pending authority before handling it.
+    let result = handle.try_send(Op::ChangeMode {
+        mode: AppMode::Operate,
+        allow_shell: true,
+        trust_mode: false,
+        auto_approve: false,
+        approval_mode: crate::tui::approval::ApprovalMode::Auto,
+        configured_sandbox_mode: None,
+    });
     assert!(result.is_err(), "try_send should fail when channel is full");
+    let authority = handle.runtime_permission_authority();
+    assert_eq!(
+        authority.approval_mode,
+        crate::tui::approval::ApprovalMode::Auto
+    );
+    assert!(!authority.auto_approve);
+}
+
+#[tokio::test]
+async fn full_mailbox_posture_update_supersedes_queued_change_mode() {
+    use crate::tui::approval::ApprovalMode;
+
+    let tmp = tempdir().expect("tempdir");
+    let config = EngineConfig {
+        workspace: tmp.path().to_path_buf(),
+        ..Default::default()
+    };
+    let (engine, handle) = Engine::new(config, &Config::default());
+
+    handle
+        .try_send(Op::ChangeMode {
+            mode: AppMode::Plan,
+            allow_shell: false,
+            trust_mode: false,
+            auto_approve: false,
+            approval_mode: ApprovalMode::Suggest,
+            configured_sandbox_mode: None,
+        })
+        .expect("queue older posture");
+    for _ in 1..ENGINE_OP_CHANNEL_CAPACITY {
+        handle
+            .try_send(Op::ListSubAgents)
+            .expect("fill operation mailbox");
+    }
+
+    let result = handle.try_send(Op::ChangeMode {
+        mode: AppMode::Operate,
+        allow_shell: true,
+        trust_mode: false,
+        auto_approve: false,
+        approval_mode: ApprovalMode::Auto,
+        configured_sandbox_mode: Some("read-only".to_string()),
+    });
+    assert!(
+        result.is_err(),
+        "latest posture wake-up must see a full mailbox"
+    );
+
+    let run = tokio::spawn(engine.run());
+    let snapshot = tokio::time::timeout(
+        std::time::Duration::from_secs(2),
+        handle.get_session_snapshot(),
+    )
+    .await
+    .expect("snapshot after mailbox drain")
+    .expect("session snapshot");
+
+    assert_eq!(snapshot.mode, "operate");
+    let authority = handle.runtime_permission_authority();
+    assert_eq!(authority.approval_mode, ApprovalMode::Auto);
+    assert!(!authority.auto_approve);
+
+    handle.send(Op::Shutdown).await.expect("shutdown engine");
+    run.await.expect("engine task");
 }
 
 #[tokio::test]

crates/tui/src/core/engine/tool_execution.rs

@@ -232,6 +232,7 @@ impl Engine {
         input: serde_json::Value,
         tool_registry: Option<&crate::tools::ToolRegistry>,
         tool_exec_lock: Arc<RwLock<()>>,
+        context_override: Option<crate::tools::ToolContext>,
     ) -> Result<ToolResult, ToolError> {
         let calls = parse_parallel_tool_calls(&input)?;
         let mcp_pool = if calls.iter().any(|(tool, _)| McpPool::is_mcp_tool(tool)) {
@@ -291,6 +292,7 @@ impl Engine {
             let mcp_pool = mcp_pool.clone();
             let shell_permits = shell_permits.clone();
             let workspace = self.session.workspace.clone();
+            let context_override = context_override.clone();
             tasks.push(async move {
                 let _shell_permit = if tool_name == "exec_shell" {
                     shell_permits.acquire_owned().await.ok()
@@ -307,7 +309,7 @@ impl Engine {
                     workspace,
                     Some(registry_ref),
                     mcp_pool,
-                    None,
+                    context_override,
                 )
                 .await;
                 (index, tool_name, result)

crates/tui/src/core/engine/turn_loop.rs

@@ -390,7 +390,8 @@ impl Engine {
         let mut turn_error: Option<String> = None;
         let mut context_recovery_attempts = 0u8;
         let mut tool_policy = tool_policy;
-        let mode = tool_policy.mode;
+        let mut mode = tool_policy.mode;
+        let mut questions_allowed = tool_policy.allows_questions();
         let strict_tool_mode = tool_policy.strict_tool_mode;
         let tool_catalog = std::mem::take(&mut tool_policy.catalog);
         let mut active_tool_names = std::mem::take(&mut tool_policy.active_names);
@@ -414,6 +415,13 @@ impl Engine {
                 return (TurnOutcomeStatus::Interrupted, None);
             }
 
+            if self.apply_pending_runtime_authority().await {
+                mode = self.current_mode;
+                questions_allowed = crate::core::authority::permission_posture_allows_questions(
+                    self.session.approval_mode,
+                );
+            }
+
             while let Ok(steer) = self.rx_steer.try_recv() {
                 let steer = steer.trim().to_string();
                 if steer.is_empty() {
@@ -1722,6 +1730,17 @@ impl Engine {
                 break;
             }
 
+            // A user can change Ask / Auto-Review / Full Access while the
+            // provider is streaming. Apply the newest typed authority before
+            // planning this tool batch; already-running tools are never
+            // retroactively reclassified.
+            if self.apply_pending_runtime_authority().await {
+                mode = self.current_mode;
+                questions_allowed = crate::core::authority::permission_posture_allows_questions(
+                    self.session.approval_mode,
+                );
+            }
+
             // Execute tools
             if self.shared_paused.lock().is_ok_and(|paused| *paused) {
                 let _ = self
@@ -2137,6 +2156,11 @@ impl Engine {
                     }));
                     match decision {
                         AutoReviewPlanDecision::NoChange => {}
+                        AutoReviewPlanDecision::Allow => {
+                            if !hook_requires_approval && !approval_force_prompt {
+                                approval_required = false;
+                            }
+                        }
                         AutoReviewPlanDecision::ForcePrompt(reason) => {
                             // The built-in safety floor is deliberately
                             // non-bypassable. Ask/Auto-Review surface the hold;
@@ -2337,6 +2361,40 @@ impl Engine {
                     ToolExecutionBatch::Serial(plan) => (false, vec![*plan]),
                 };
 
+                // Planning can run hooks and other async gates. If policy
+                // changed after this batch was planned, never execute it with
+                // stale approval or sandbox facts. Return one typed retry to
+                // the model; the next call is planned under the new posture.
+                if self.apply_pending_runtime_authority().await {
+                    mode = self.current_mode;
+                    questions_allowed = crate::core::authority::permission_posture_allows_questions(
+                        self.session.approval_mode,
+                    );
+                    for plan in plans {
+                        let result = Err(ToolError::permission_denied(
+                            "Runtime permission posture changed while this tool call was being planned; retry it under the current posture."
+                                .to_string(),
+                        ));
+                        let _ = self
+                            .tx_event
+                            .send(Event::ToolCallComplete {
+                                id: plan.id.clone(),
+                                name: plan.name.clone(),
+                                result: result.clone(),
+                            })
+                            .await;
+                        outcomes[plan.index] = Some(ToolExecOutcome {
+                            index: plan.index,
+                            id: plan.id,
+                            name: plan.name,
+                            input: plan.input,
+                            started_at: Instant::now(),
+                            terminal: ToolExecutionOutcome::from_legacy(result),
+                        });
+                    }
+                    continue;
+                }
+
                 // #3216 / #2211: once the turn is cancelled, do not start any
                 // further tool batches. Cancellation arrives out-of-band (the
                 // TUI cancels the shared token directly), so we can observe it
@@ -2373,6 +2431,8 @@ impl Engine {
                     continue;
                 }
 
+                let batch_tool_context = self.live_tool_context(tool_registry);
+
                 if parallel_allowed {
                     let parallel_plan_receipts: Vec<_> = plans
                         .iter()
@@ -2428,6 +2488,7 @@ impl Engine {
                         let started_at = Instant::now();
                         let shell_permits = shell_permits.clone();
                         let workspace = self.session.workspace.clone();
+                        let context_override = batch_tool_context.clone();
 
                         tool_tasks.push(async move {
                             let _shell_permit = if plan.name == "exec_shell" {
@@ -2445,7 +2506,7 @@ impl Engine {
                                 workspace,
                                 registry,
                                 mcp_pool,
-                                None,
+                                context_override,
                             )
                             .await;
 
@@ -2595,6 +2656,7 @@ impl Engine {
                                     tool_input.clone(),
                                     tool_registry,
                                     tool_exec_lock.clone(),
+                                    batch_tool_context.clone(),
                                 ) => ToolExecutionOutcome::from_legacy(result),
                             };
                             let result = terminal.legacy_result();
@@ -2650,7 +2712,7 @@ impl Engine {
 
                         if tool_name == REQUEST_USER_INPUT_NAME {
                             let started_at = Instant::now();
-                            let result = if tool_policy.allows_questions() {
+                            let result = if questions_allowed {
                                 match UserInputRequest::from_value(&tool_input) {
                                     Ok(request) => self
                                         .await_user_input(&tool_id, request)
@@ -2768,9 +2830,10 @@ impl Engine {
                                         "policy": format!("{policy:?}"),
                                         "caller": caller_type_for_tool_use(tool_caller.as_ref()),
                                     }));
-                                    let elevated_context = tool_registry.map(|r| {
-                                        r.context().clone().with_elevated_sandbox_policy(policy)
-                                    });
+                                    let elevated_context =
+                                        batch_tool_context.clone().map(|context| {
+                                            context.with_elevated_sandbox_policy(policy)
+                                        });
                                     (
                                         None,
                                         elevated_context,
@@ -2783,6 +2846,26 @@ impl Engine {
                             (None, None, None)
                         };
 
+                        // An approval wait can outlive a posture switch. Do
+                        // not start a tool from the stale plan; the
+                        // model can retry immediately under the newly applied
+                        // authority.
+                        let mut result_override = if self.apply_pending_runtime_authority().await {
+                            mode = self.current_mode;
+                            questions_allowed =
+                                crate::core::authority::permission_posture_allows_questions(
+                                    self.session.approval_mode,
+                                );
+                            result_override.or_else(|| {
+                                Some(Err(ToolError::permission_denied(
+                                    "Runtime permission posture changed before this tool call executed; retry it under the current posture."
+                                        .to_string(),
+                                )))
+                            })
+                        } else {
+                            result_override
+                        };
+
                         // Per-tool snapshot for surgical undo (#384): capture workspace
                         // state before file-modifying tools execute so `/undo` can
                         // revert the most recent write_file/edit_file/apply_patch.
@@ -2801,6 +2884,20 @@ impl Engine {
                             .await;
                         }
 
+                        if self.apply_pending_runtime_authority().await {
+                            mode = self.current_mode;
+                            questions_allowed =
+                                crate::core::authority::permission_posture_allows_questions(
+                                    self.session.approval_mode,
+                                );
+                            result_override.get_or_insert_with(|| {
+                                Err(ToolError::permission_denied(
+                                    "Runtime permission posture changed before this tool call executed; retry it under the current posture."
+                                        .to_string(),
+                                ))
+                            });
+                        }
+
                         let started_at = Instant::now();
                         let (mut result, cancelled_before_completion) =
                             if let Some(result_override) = result_override {
@@ -2821,7 +2918,7 @@ impl Engine {
                                         self.session.workspace.clone(),
                                         tool_registry,
                                         mcp_pool.clone(),
-                                        context_override,
+                                        context_override.or_else(|| batch_tool_context.clone()),
                                     ) => (result, false),
                                 }
                             };

crates/tui/src/main.rs

@@ -105,6 +105,7 @@ mod route_runtime;
 mod runtime_api;
 mod runtime_handoff;
 mod runtime_log;
+mod runtime_policy;
 mod runtime_threads;
 mod safe_label;
 mod sandbox;

crates/tui/src/runtime_api.rs

@@ -202,6 +202,7 @@ struct StreamTurnRequest {
     prompt: String,
     model: Option<String>,
     mode: Option<String>,
+    permission_posture: Option<String>,
     workspace: Option<PathBuf>,
     allow_shell: Option<bool>,
     trust_mode: Option<bool>,
@@ -2044,6 +2045,7 @@ async fn retry_thread_turn(
                 input_summary: None,
                 model: None,
                 mode: None,
+                permission_posture: None,
                 allow_shell: None,
                 trust_mode: None,
                 auto_approve: None,
@@ -2357,6 +2359,7 @@ async fn stream_turn(
         .clone()
         .unwrap_or_else(|| state.workspace.clone());
     let mode = req.mode.clone().unwrap_or_else(|| "agent".to_string());
+    let permission_posture = req.permission_posture.clone();
     let allow_shell = req.allow_shell.unwrap_or(state.config.read().allow_shell());
     let trust_mode = req.trust_mode.unwrap_or(false);
     let auto_approve = req.auto_approve.unwrap_or(false);
@@ -2368,6 +2371,7 @@ async fn stream_turn(
             model: Some(model.clone()),
             workspace: Some(workspace.clone()),
             mode: Some(mode.clone()),
+            permission_posture: permission_posture.clone(),
             allow_shell: Some(allow_shell),
             trust_mode: Some(trust_mode),
             auto_approve: Some(auto_approve),
@@ -2401,6 +2405,7 @@ async fn stream_turn(
                 input_summary: None,
                 model: Some(model.clone()),
                 mode: Some(mode.clone()),
+                permission_posture,
                 allow_shell: Some(allow_shell),
                 trust_mode: Some(trust_mode),
                 auto_approve: Some(auto_approve),

crates/tui/src/runtime_api/tests.rs

@@ -270,6 +270,7 @@ fn messages_from_thread_detail_batches_tool_results() {
         model_provider_id: None,
         workspace: PathBuf::from("."),
         mode: "agent".to_string(),
+        permission_posture: Some("ask".to_string()),
         allow_shell: false,
         trust_mode: false,
         auto_approve: false,
@@ -292,6 +293,7 @@ fn messages_from_thread_detail_batches_tool_results() {
         ended_at: Some(now),
         duration_ms: Some(0),
         usage: None,
+        permission_posture: Some("ask".to_string()),
         effective_provider: None,
         effective_provider_id: None,
         effective_billing_surface: None,
@@ -455,6 +457,7 @@ fn legacy_exact_thread_export_normalizes_provider_kind_and_id() {
             model_provider_id: None,
             workspace: PathBuf::from("."),
             mode: "agent".to_string(),
+            permission_posture: None,
             allow_shell: false,
             trust_mode: false,
             auto_approve: false,
@@ -4606,8 +4609,10 @@ fn cors_layer_skips_invalid_origins() {
 
 /// #562 / whalescale#256 — `PATCH /v1/threads/{id}` accepts the new
 /// fields (allow_shell, trust_mode, auto_approve, model, mode, title,
-/// system_prompt). Each is independently optional; an empty string clears
-/// `title` / `system_prompt` back to None.
+/// system_prompt). Legacy mode aliases remain accepted as one-way inputs and
+/// the response returns the canonical product mode. Each field is
+/// independently optional; an empty string clears `title` / `system_prompt`
+/// back to None.
 #[tokio::test]
 async fn patch_thread_accepts_extended_field_set() -> Result<()> {
     let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
@@ -4653,7 +4658,8 @@ async fn patch_thread_accepts_extended_field_set() -> Result<()> {
     assert_eq!(patched["trust_mode"], true);
     assert_eq!(patched["auto_approve"], true);
     assert_eq!(patched["model"], "deepseek-v4-pro");
-    assert_eq!(patched["mode"], "yolo");
+    assert_eq!(patched["mode"], "agent");
+    assert_eq!(patched["permission_posture"], "full_access");
     assert_eq!(patched["title"], "Whalescale UI test thread");
     assert_eq!(patched["system_prompt"], "You are a useful assistant.");
 
@@ -4927,6 +4933,61 @@ async fn create_thread_accepts_dynamic_tools_and_environments() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn create_thread_normalizes_and_persists_named_permission_posture() -> Result<()> {
+    let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
+        return Ok(());
+    };
+    let client = crate::tls::reqwest_client();
+
+    let created: serde_json::Value = client
+        .post(format!("http://{addr}/v1/threads"))
+        .json(&json!({
+            "model": "test-model",
+            "mode": "operate",
+            "permission_posture": "auto-review"
+        }))
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    assert_eq!(created["mode"], "operate");
+    assert_eq!(created["permission_posture"], "auto_review");
+    assert_eq!(created["auto_approve"], false);
+    assert_eq!(created["trust_mode"], false);
+
+    let authoritative_ask: serde_json::Value = client
+        .post(format!("http://{addr}/v1/threads"))
+        .json(&json!({
+            "model": "test-model",
+            "mode": "yolo",
+            "permission_posture": "ask",
+            "auto_approve": true
+        }))
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    assert_eq!(authoritative_ask["mode"], "agent");
+    assert_eq!(authoritative_ask["permission_posture"], "ask");
+    assert_eq!(authoritative_ask["auto_approve"], false);
+
+    let invalid = client
+        .post(format!("http://{addr}/v1/threads"))
+        .json(&json!({
+            "model": "test-model",
+            "permission_posture": "owner"
+        }))
+        .send()
+        .await?;
+    assert_eq!(invalid.status(), StatusCode::BAD_REQUEST);
+
+    handle.abort();
+    Ok(())
+}
+
 #[tokio::test]
 async fn start_turn_accepts_dynamic_tools_and_environment_id() -> Result<()> {
     let Some((addr, _runtime_threads, handle)) = spawn_test_server().await? else {
@@ -4955,7 +5016,8 @@ async fn start_turn_accepts_dynamic_tools_and_environment_id() -> Result<()> {
                     "input_schema": { "type": "object" }
                 }
             ],
-            "environment_id": "local"
+            "environment_id": "local",
+            "permission_posture": "auto-review"
         }))
         .send()
         .await?
@@ -4963,6 +5025,17 @@ async fn start_turn_accepts_dynamic_tools_and_environment_id() -> Result<()> {
         .json()
         .await?;
     assert_eq!(started["turn"]["thread_id"], thread_id);
+    assert_eq!(started["thread"]["permission_posture"], "ask");
+    assert_eq!(started["turn"]["permission_posture"], "auto_review");
+
+    let stored: serde_json::Value = client
+        .get(format!("http://{addr}/v1/threads/{thread_id}"))
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+    assert_eq!(stored["turns"][0]["permission_posture"], "auto_review");
 
     handle.abort();
     Ok(())

crates/tui/src/runtime_policy.rs

@@ -0,0 +1,158 @@
+//! Normalization for Runtime mode and permission-posture wires.
+//!
+//! Older clients can still send mode aliases and the legacy `auto_approve`
+//! bit. New callers send a named posture. Keep those inputs readable, but
+//! persist and execute one contract: Plan / Act / Operate plus Ask /
+//! Auto-Review / Full Access.
+
+use anyhow::{Result, bail};
+
+use crate::tui::app::AppMode;
+use crate::tui::approval::ApprovalMode;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) struct RuntimePolicyProjection {
+    pub(crate) mode: AppMode,
+    pub(crate) permission: ApprovalMode,
+}
+
+impl RuntimePolicyProjection {
+    /// Read a persisted compatibility shape. Unknown historical values fail
+    /// closed to Act + Ask unless the old bypass bit is explicitly present.
+    #[must_use]
+    pub(crate) fn from_persisted(
+        mode: &str,
+        permission_posture: Option<&str>,
+        auto_approve: bool,
+    ) -> Self {
+        let parsed_mode = parse_runtime_mode(mode).unwrap_or(AppMode::Agent);
+        let permission = permission_posture
+            .and_then(ApprovalMode::from_config_value)
+            .filter(|permission| *permission != ApprovalMode::Never)
+            .unwrap_or_else(|| {
+                if parsed_mode == AppMode::Yolo || auto_approve {
+                    ApprovalMode::Bypass
+                } else {
+                    ApprovalMode::Suggest
+                }
+            });
+        Self {
+            mode: visible_mode(parsed_mode),
+            permission,
+        }
+    }
+
+    /// Validate and normalize a new Runtime request. Legacy aliases are
+    /// accepted as one-way inputs, but only current values are persisted.
+    pub(crate) fn from_request(
+        mode: &str,
+        permission_posture: Option<&str>,
+        auto_approve: Option<bool>,
+    ) -> Result<Self> {
+        let parsed_mode = parse_runtime_mode(mode).ok_or_else(|| {
+            anyhow::anyhow!("unsupported Runtime mode {mode:?}; expected plan, act, or operate")
+        })?;
+        let permission = match permission_posture {
+            Some(value) => ApprovalMode::from_config_value(value).ok_or_else(|| {
+                anyhow::anyhow!(
+                    "unsupported permission posture {value:?}; expected ask, auto-review, or full-access"
+                )
+            })?,
+            None if parsed_mode == AppMode::Yolo || auto_approve.unwrap_or(false) => {
+                ApprovalMode::Bypass
+            }
+            None => ApprovalMode::Suggest,
+        };
+        if permission == ApprovalMode::Never {
+            bail!("permission posture 'never' is not part of the Runtime product contract");
+        }
+        Ok(Self {
+            mode: visible_mode(parsed_mode),
+            permission,
+        })
+    }
+
+    #[must_use]
+    pub(crate) fn mode_setting(self) -> &'static str {
+        self.mode.as_setting()
+    }
+
+    #[must_use]
+    pub(crate) fn permission_wire(self) -> &'static str {
+        match self.permission {
+            ApprovalMode::Suggest => "ask",
+            ApprovalMode::Auto => "auto_review",
+            ApprovalMode::Bypass => "full_access",
+            ApprovalMode::Never => "ask",
+        }
+    }
+
+    #[must_use]
+    pub(crate) fn auto_approve(self) -> bool {
+        self.permission == ApprovalMode::Bypass
+    }
+}
+
+#[must_use]
+pub(crate) fn parse_runtime_mode(value: &str) -> Option<AppMode> {
+    match value.trim().to_ascii_lowercase().as_str() {
+        "normal" => Some(AppMode::Agent),
+        other => AppMode::parse(other),
+    }
+}
+
+#[must_use]
+fn visible_mode(mode: AppMode) -> AppMode {
+    match mode {
+        AppMode::Auto | AppMode::Yolo => AppMode::Agent,
+        other => other,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn legacy_inputs_project_to_current_mode_and_permission_wires() {
+        for alias in ["normal", "agent", "auto"] {
+            let policy = RuntimePolicyProjection::from_persisted(alias, None, false);
+            assert_eq!(policy.mode_setting(), "agent", "{alias}");
+            assert_eq!(policy.permission_wire(), "ask", "{alias}");
+        }
+        for alias in ["yolo", "bypass"] {
+            let policy = RuntimePolicyProjection::from_persisted(alias, None, false);
+            assert_eq!(policy.mode_setting(), "agent", "{alias}");
+            assert_eq!(policy.permission_wire(), "full_access", "{alias}");
+        }
+    }
+
+    #[test]
+    fn named_posture_is_normalized_and_authoritative() {
+        let auto = RuntimePolicyProjection::from_request("operate", Some("auto-review"), None)
+            .expect("compat permission");
+        assert_eq!(auto.mode_setting(), "operate");
+        assert_eq!(auto.permission_wire(), "auto_review");
+        assert_eq!(auto.permission, ApprovalMode::Auto);
+        assert!(!auto.auto_approve());
+
+        let full = RuntimePolicyProjection::from_request("act", Some("full_access"), None)
+            .expect("current permission wire");
+        assert_eq!(full.permission_wire(), "full_access");
+        assert_eq!(full.permission, ApprovalMode::Bypass);
+        assert!(full.auto_approve());
+
+        let ask = RuntimePolicyProjection::from_request("yolo", Some("ask"), Some(true))
+            .expect("named posture overrides legacy flags");
+        assert_eq!(ask.mode_setting(), "agent");
+        assert_eq!(ask.permission_wire(), "ask");
+        assert_eq!(ask.permission, ApprovalMode::Suggest);
+        assert!(!ask.auto_approve());
+    }
+
+    #[test]
+    fn invalid_or_never_postures_are_rejected() {
+        assert!(RuntimePolicyProjection::from_request("act", Some("owner"), None).is_err());
+        assert!(RuntimePolicyProjection::from_request("act", Some("never"), None).is_err());
+    }
+}

crates/tui/src/runtime_threads.rs

@@ -46,9 +46,11 @@ use crate::route_budget::{
 use crate::route_runtime::{
     ResolvedRuntimeRoute, resolve_runtime_route, resolve_runtime_route_for_identity,
 };
+use crate::runtime_policy::RuntimePolicyProjection;
 use crate::tools::plan::new_shared_plan_state;
 use crate::tools::subagent::SubAgentStatus;
 use crate::tools::todo::new_shared_todo_list;
+#[cfg(test)]
 use crate::tui::app::AppMode;
 use codewhale_protocol::runtime::{
     DynamicToolCallContent, DynamicToolCallParams, DynamicToolCallResult, DynamicToolSpec,
@@ -432,6 +434,10 @@ pub struct ThreadRecord {
     pub model_provider_id: Option<String>,
     pub workspace: PathBuf,
     pub mode: String,
+    /// Named default permission posture for new turns. Absent on legacy
+    /// records, whose effective posture is derived from the old fields.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub permission_posture: Option<String>,
     pub allow_shell: bool,
     pub trust_mode: bool,
     pub auto_approve: bool,
@@ -467,6 +473,7 @@ fn thread_execution_state_matches(left: &ThreadRecord, right: &ThreadRecord) ->
         && left.model_provider_id == right.model_provider_id
         && left.workspace == right.workspace
         && left.mode == right.mode
+        && left.permission_posture == right.permission_posture
         && left.allow_shell == right.allow_shell
         && left.trust_mode == right.trust_mode
         && left.auto_approve == right.auto_approve
@@ -495,6 +502,10 @@ pub struct TurnRecord {
     pub duration_ms: Option<u64>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub usage: Option<Usage>,
+    /// Canonical posture that governed this turn. New records always carry
+    /// this receipt; old records deserialize with no fabricated value.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub permission_posture: Option<String>,
     /// Concrete generic provider kind selected for this turn.
     #[serde(
         default,
@@ -1477,6 +1488,8 @@ pub struct CreateThreadRequest {
     pub model_provider_id: Option<String>,
     pub workspace: Option<PathBuf>,
     pub mode: Option<String>,
+    #[serde(default)]
+    pub permission_posture: Option<String>,
     pub allow_shell: Option<bool>,
     pub trust_mode: Option<bool>,
     pub auto_approve: Option<bool>,
@@ -1505,6 +1518,7 @@ pub struct UpdateThreadRequest {
     pub auto_approve: Option<bool>,
     pub model: Option<String>,
     pub mode: Option<String>,
+    pub permission_posture: Option<String>,
     pub title: Option<String>,
     pub system_prompt: Option<String>,
     pub workspace: Option<PathBuf>,
@@ -1517,6 +1531,8 @@ pub struct StartTurnRequest {
     pub input_summary: Option<String>,
     pub model: Option<String>,
     pub mode: Option<String>,
+    #[serde(default)]
+    pub permission_posture: Option<String>,
     pub allow_shell: Option<bool>,
     pub trust_mode: Option<bool>,
     pub auto_approve: Option<bool>,
@@ -1925,8 +1941,6 @@ fn runtime_compaction_config(
 struct ActiveTurnState {
     turn_id: String,
     interrupt_requested: bool,
-    auto_approve: bool,
-    trust_mode: bool,
 }
 
 #[derive(Debug, Clone, Copy)]
@@ -2986,32 +3000,49 @@ impl RuntimeThreadManager {
     }
 
     async fn remember_thread_auto_approve(&self, thread_id: &str) {
-        {
+        let thread = {
             let _thread_mutation = self.store.thread_mutation.lock();
             let Ok(mut thread) = self.store.load_thread(thread_id) else {
                 return;
             };
-            if thread.auto_approve {
-                return;
+            if !thread.auto_approve || thread.permission_posture.as_deref() != Some("full_access") {
+                thread.auto_approve = true;
+                thread.permission_posture = Some("full_access".to_string());
+                thread.updated_at = Utc::now();
+                if let Err(err) = self.store.save_thread(&thread) {
+                    tracing::warn!(
+                        "Failed to persist full-access posture for thread {}: {}",
+                        thread_id,
+                        err
+                    );
+                    return;
+                }
             }
-            thread.auto_approve = true;
-            thread.updated_at = Utc::now();
-            if let Err(err) = self.store.save_thread(&thread) {
-                tracing::warn!(
-                    "Failed to persist auto_approve flip for thread {}: {}",
-                    thread_id,
-                    err
-                );
-            }
-        }
+            thread
+        };
 
-        {
-            let mut active = self.active.lock().await;
-            if let Some(state) = active.engines.get_mut(thread_id)
-                && let Some(turn) = state.active_turn.as_mut()
-            {
-                turn.auto_approve = true;
-            }
+        let engine = {
+            let active = self.active.lock().await;
+            active
+                .engines
+                .get(thread_id)
+                .map(|state| state.engine.clone())
+        };
+        if let Some(engine) = engine {
+            let configured_sandbox_mode = self.read_config().sandbox_mode.clone();
+            let policy = RuntimePolicyProjection::from_persisted(
+                &thread.mode,
+                thread.permission_posture.as_deref(),
+                thread.auto_approve,
+            );
+            let _ = engine.try_send(Op::ChangeMode {
+                mode: policy.mode,
+                allow_shell: thread.allow_shell,
+                trust_mode: thread.trust_mode,
+                auto_approve: policy.auto_approve(),
+                approval_mode: policy.permission,
+                configured_sandbox_mode,
+            });
         }
     }
 
@@ -3504,15 +3535,22 @@ impl RuntimeThreadManager {
             .filter(|m| !m.trim().is_empty())
             .unwrap_or(default_model);
         let workspace = req.workspace.unwrap_or_else(|| self.workspace.clone());
-        let mode = req
+        let requested_mode = req
             .mode
             .filter(|m| !m.trim().is_empty())
             .unwrap_or_else(|| "agent".to_string());
+        let policy = RuntimePolicyProjection::from_request(
+            &requested_mode,
+            req.permission_posture.as_deref(),
+            req.auto_approve,
+        )?;
+        let mode = policy.mode_setting().to_string();
+        let permission_posture = Some(policy.permission_wire().to_string());
         let allow_shell = req
             .allow_shell
             .unwrap_or_else(|| self.read_config().allow_shell());
         let trust_mode = req.trust_mode.unwrap_or(false);
-        let auto_approve = req.auto_approve.unwrap_or(false);
+        let auto_approve = policy.auto_approve();
 
         let thread = ThreadRecord {
             schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
@@ -3524,6 +3562,7 @@ impl RuntimeThreadManager {
             model_provider_id,
             workspace,
             mode,
+            permission_posture,
             allow_shell,
             trust_mode,
             auto_approve,
@@ -3668,6 +3707,7 @@ impl RuntimeThreadManager {
             && req.auto_approve.is_none()
             && req.model.is_none()
             && req.mode.is_none()
+            && req.permission_posture.is_none()
             && req.title.is_none()
             && req.system_prompt.is_none()
             && req.workspace.is_none()
@@ -3685,13 +3725,19 @@ impl RuntimeThreadManager {
         {
             bail!("mode must not be empty");
         }
+        if let Some(permission_posture) = req.permission_posture.as_ref()
+            && permission_posture.trim().is_empty()
+        {
+            bail!("permission_posture must not be empty");
+        }
         if let Some(workspace) = req.workspace.as_ref()
             && workspace.as_os_str().is_empty()
         {
             bail!("workspace must not be empty");
         }
 
-        let (thread, changes, evicted_engine) = {
+        let configured_sandbox_mode = self.read_config().sandbox_mode.clone();
+        let (thread, changes, evicted_engine, posture_engine) = {
             // Take the active guard first so a workspace mutation can check
             // and evict the cached engine atomically with the durable update.
             // Using the same order as start/compact avoids lock inversion.
@@ -3702,6 +3748,19 @@ impl RuntimeThreadManager {
                 .load_thread(id)
                 .with_context(|| format!("Thread not found: {id}"))?;
             let mut changes = serde_json::Map::new();
+            let policy_patch = if req.mode.is_some()
+                || req.permission_posture.is_some()
+                || req.auto_approve.is_some()
+            {
+                Some(runtime_policy_with_overrides(
+                    &thread,
+                    req.mode.as_deref(),
+                    req.permission_posture.as_deref(),
+                    req.auto_approve,
+                )?)
+            } else {
+                None
+            };
 
             if let Some(archived) = req.archived
                 && thread.archived != archived
@@ -3721,23 +3780,28 @@ impl RuntimeThreadManager {
                 thread.trust_mode = trust_mode;
                 changes.insert("trust_mode".to_string(), json!(trust_mode));
             }
-            if let Some(auto_approve) = req.auto_approve
-                && thread.auto_approve != auto_approve
-            {
-                thread.auto_approve = auto_approve;
-                changes.insert("auto_approve".to_string(), json!(auto_approve));
-            }
             if let Some(model) = req.model
                 && thread.model != model
             {
                 thread.model = model.clone();
                 changes.insert("model".to_string(), json!(model));
             }
-            if let Some(mode) = req.mode
-                && thread.mode != mode
-            {
-                thread.mode = mode.clone();
-                changes.insert("mode".to_string(), json!(mode));
+            if let Some(policy) = policy_patch {
+                let mode = policy.mode_setting().to_string();
+                let permission_posture = Some(policy.permission_wire().to_string());
+                let auto_approve = policy.auto_approve();
+                if thread.mode != mode {
+                    thread.mode = mode.clone();
+                    changes.insert("mode".to_string(), json!(mode));
+                }
+                if thread.permission_posture != permission_posture {
+                    thread.permission_posture = permission_posture.clone();
+                    changes.insert("permission_posture".to_string(), json!(permission_posture));
+                }
+                if thread.auto_approve != auto_approve {
+                    thread.auto_approve = auto_approve;
+                    changes.insert("auto_approve".to_string(), json!(auto_approve));
+                }
             }
             if let Some(title) = req.title {
                 // Empty string clears a previously-set title and reverts to derived.
@@ -3780,6 +3844,16 @@ impl RuntimeThreadManager {
                 bail!("workspace cannot be changed while the thread has an active turn");
             }
 
+            // A posture/mode edit must reach the live engine even while a
+            // turn is running. EngineHandle publishes the authority snapshot
+            // before queueing ChangeMode; the turn loop applies that pending
+            // update before the next tool batch.
+            let posture_changed = changes.contains_key("auto_approve")
+                || changes.contains_key("permission_posture")
+                || changes.contains_key("trust_mode")
+                || changes.contains_key("allow_shell")
+                || changes.contains_key("mode");
+
             let evicted_engine = if changes.is_empty() {
                 None
             } else {
@@ -3792,13 +3866,37 @@ impl RuntimeThreadManager {
                     None
                 }
             };
-            (thread, changes, evicted_engine)
+            let posture_engine = if posture_changed && !workspace_changed {
+                active.engines.get(id).map(|state| state.engine.clone())
+            } else {
+                None
+            };
+            (thread, changes, evicted_engine, posture_engine)
         };
 
         if let Some(engine) = evicted_engine {
             let _ = engine.send(Op::Shutdown).await;
         }
 
+        // Keep the live engine session converged with the thread record.
+        // Idle engines apply it immediately; a running turn applies it at
+        // the next mid-turn drain (before the next tool batch).
+        if let Some(engine) = posture_engine {
+            let policy = RuntimePolicyProjection::from_persisted(
+                &thread.mode,
+                thread.permission_posture.as_deref(),
+                thread.auto_approve,
+            );
+            let _ = engine.try_send(Op::ChangeMode {
+                mode: policy.mode,
+                allow_shell: thread.allow_shell,
+                trust_mode: thread.trust_mode,
+                auto_approve: policy.auto_approve(),
+                approval_mode: policy.permission,
+                configured_sandbox_mode: configured_sandbox_mode.clone(),
+            });
+        }
+
         if !changes.is_empty() {
             self.emit_event(
                 &thread.id,
@@ -4438,6 +4536,7 @@ impl RuntimeThreadManager {
                     ended_at: Some(now),
                     duration_ms: Some(0),
                     usage: None,
+                    permission_posture: None,
                     effective_provider: None,
                     effective_provider_id: None,
                     effective_billing_surface: None,
@@ -4862,6 +4961,23 @@ impl RuntimeThreadManager {
         }
 
         let thread = self.get_thread(thread_id).await?;
+        let policy =
+            if req.mode.is_some() || req.permission_posture.is_some() || req.auto_approve.is_some()
+            {
+                runtime_policy_with_overrides(
+                    &thread,
+                    req.mode.as_deref(),
+                    req.permission_posture.as_deref(),
+                    req.auto_approve,
+                )?
+            } else {
+                RuntimePolicyProjection::from_persisted(
+                    &thread.mode,
+                    thread.permission_posture.as_deref(),
+                    thread.auto_approve,
+                )
+            };
+        let mode = policy.mode;
         let engine = self.ensure_engine_loaded(&thread).await?;
 
         let client_preflight_required = {
@@ -4880,11 +4996,6 @@ impl RuntimeThreadManager {
         // Resolve the concrete provider/model before persisting a turn. Auto
         // routing can fail, and such a failure must not leave a zombie
         // in-progress record behind.
-        let mode = req
-            .mode
-            .as_deref()
-            .and_then(parse_mode_opt)
-            .unwrap_or_else(|| parse_mode(&thread.mode));
         let requested_model = req.model.as_deref().unwrap_or(&thread.model).to_string();
         let auto_model = requested_model.trim().eq_ignore_ascii_case("auto");
         let cfg_snapshot = self.config.read().clone();
@@ -4944,6 +5055,7 @@ impl RuntimeThreadManager {
         } else {
             route
         };
+        let configured_sandbox_mode = route.config.sandbox_mode.clone();
         let provider = route.identity.provider;
         let provider_identity = route.identity.clone();
         let model = route.model.clone();
@@ -4973,6 +5085,7 @@ impl RuntimeThreadManager {
             ended_at: None,
             duration_ms: None,
             usage: None,
+            permission_posture: Some(policy.permission_wire().to_string()),
             effective_provider: Some(provider.as_str().to_string()),
             effective_provider_id: provider_identity
                 .exact_id
@@ -5009,7 +5122,7 @@ impl RuntimeThreadManager {
 
         let allow_shell = req.allow_shell.unwrap_or(thread.allow_shell);
         let trust_mode = req.trust_mode.unwrap_or(thread.trust_mode);
-        let auto_approve = req.auto_approve.unwrap_or(thread.auto_approve);
+        let auto_approve = policy.auto_approve();
         let op = Op::SendMessage {
             content: prompt,
             mode,
@@ -5028,11 +5141,7 @@ impl RuntimeThreadManager {
             allowed_tools: None,
             dynamic_tools: req.dynamic_tools,
             hook_executor: None,
-            approval_mode: if auto_approve {
-                crate::tui::approval::ApprovalMode::Bypass
-            } else {
-                crate::tui::approval::ApprovalMode::Suggest
-            },
+            approval_mode: policy.permission,
             verbosity,
             provenance: crate::core::ops::UserInputProvenance::ExternalUser,
         };
@@ -5066,8 +5175,6 @@ impl RuntimeThreadManager {
             state.active_turn = Some(ActiveTurnState {
                 turn_id: turn_id.clone(),
                 interrupt_requested: false,
-                auto_approve,
-                trust_mode,
             });
             state.route_identity = provider_identity;
             state.route_model.clone_from(&model);
@@ -5099,6 +5206,14 @@ impl RuntimeThreadManager {
             // point the engine owns the operation and the spawned task owns
             // lifecycle events, monitoring, and terminal cleanup even if the
             // HTTP/client future is dropped.
+            engine.publish_turn_authority(
+                mode,
+                allow_shell,
+                trust_mode,
+                auto_approve,
+                policy.permission,
+                configured_sandbox_mode,
+            );
             let _sender = permit.send(op);
             touch_lru(&mut active.lru, thread_id);
             self.spawn_claimed_turn_monitor(
@@ -5275,6 +5390,7 @@ impl RuntimeThreadManager {
         } else {
             route
         };
+        let configured_sandbox_mode = route.config.sandbox_mode.clone();
         let route_provider = route.identity.provider;
         let route_identity = route.identity.clone();
         let route_model = route.model.clone();
@@ -5307,6 +5423,15 @@ impl RuntimeThreadManager {
             ended_at: None,
             duration_ms: None,
             usage: None,
+            permission_posture: Some(
+                RuntimePolicyProjection::from_persisted(
+                    &thread.mode,
+                    thread.permission_posture.as_deref(),
+                    thread.auto_approve,
+                )
+                .permission_wire()
+                .to_string(),
+            ),
             effective_provider: Some(route_provider.as_str().to_string()),
             effective_provider_id: route_identity
                 .exact_id
@@ -5351,8 +5476,6 @@ impl RuntimeThreadManager {
             state.active_turn = Some(ActiveTurnState {
                 turn_id: turn_id.clone(),
                 interrupt_requested: false,
-                auto_approve: current_thread.auto_approve,
-                trust_mode: current_thread.trust_mode,
             });
             state.route_identity = route_identity;
             state.route_model = route_model;
@@ -5377,6 +5500,19 @@ impl RuntimeThreadManager {
             }
 
             self.register_runtime_usage_sink(&turn_id);
+            let policy = RuntimePolicyProjection::from_persisted(
+                &current_thread.mode,
+                current_thread.permission_posture.as_deref(),
+                current_thread.auto_approve,
+            );
+            engine.publish_turn_authority(
+                policy.mode,
+                current_thread.allow_shell,
+                current_thread.trust_mode,
+                policy.auto_approve(),
+                policy.permission,
+                configured_sandbox_mode,
+            );
             let _sender = permit.send(op);
             touch_lru(&mut active.lru, thread_id);
             self.spawn_claimed_turn_monitor(
@@ -5663,7 +5799,12 @@ impl RuntimeThreadManager {
                         system_prompt_override: thread.system_prompt.is_some(),
                         model: route_model.clone(),
                         workspace: thread.workspace.clone(),
-                        mode: parse_mode(&thread.mode),
+                        mode: RuntimePolicyProjection::from_persisted(
+                            &thread.mode,
+                            thread.permission_posture.as_deref(),
+                            thread.auto_approve,
+                        )
+                        .mode,
                     })
                     .await
                     .map_err(|e| anyhow!("Failed to sync thread session: {e}"))?;
@@ -6524,12 +6665,16 @@ impl RuntimeThreadManager {
                     intent_summary,
                     ..
                 } => {
-                    let Some((auto_approve, trust_mode)) =
-                        self.active_turn_flags(&thread_id, &turn_id).await
+                    let Some(authority) = self
+                        .active_turn_authority(&thread_id, &turn_id, &engine)
+                        .await
                     else {
                         let _ = engine.deny_tool_call(&id).await;
                         continue;
                     };
+                    let auto_approve = authority.auto_approve;
+                    let trust_mode = authority.trust_mode;
+                    let approval_mode = authority.approval_mode;
 
                     let pending_request = PendingApprovalRequest {
                         id: id.clone(),
@@ -6588,6 +6733,31 @@ impl RuntimeThreadManager {
                         continue;
                     }
 
+                    // Auto-Review never opens an approval modal. The engine
+                    // resolves gated tools under Auto itself, so reaching
+                    // this branch means a host injected the event directly:
+                    // fail closed (the audit trail stays authoritative)
+                    // instead of pausing the turn.
+                    if approval_mode == crate::tui::approval::ApprovalMode::Auto {
+                        self.emit_event(
+                            &thread_id,
+                            Some(&turn_id),
+                            None,
+                            "approval.decided",
+                            json!({
+                                "approval_id": id,
+                                "decision": "deny",
+                                "remember": false,
+                                "auto": true,
+                                "posture": "auto_review",
+                            }),
+                        )
+                        .await
+                        .ok();
+                        let _ = engine.deny_tool_call(id).await;
+                        continue;
+                    }
+
                     // Register before sequencing the event. A snapshot racing
                     // this branch therefore either contains the request or
                     // subscribes from an older cursor that will replay it.
@@ -6707,10 +6877,16 @@ impl RuntimeThreadManager {
                         }),
                     )
                     .await?;
-                    let (auto_approve, trust_mode) = self
-                        .active_turn_flags(&thread_id, &turn_id)
+                    let authority = self
+                        .active_turn_authority(&thread_id, &turn_id, &engine)
                         .await
-                        .unwrap_or((false, false));
+                        .unwrap_or(crate::core::engine::RuntimePermissionAuthority {
+                            auto_approve: false,
+                            trust_mode: false,
+                            approval_mode: crate::tui::approval::ApprovalMode::Suggest,
+                        });
+                    let auto_approve = authority.auto_approve;
+                    let trust_mode = authority.trust_mode;
                     match Self::approval_decision(auto_approve, trust_mode, true) {
                         RuntimeApprovalDecision::RetryWithFullAccess => {
                             let _ = engine
@@ -7016,6 +7192,22 @@ impl RuntimeThreadManager {
         Ok(turn.turn_id == turn_id && turn.interrupt_requested)
     }
 
+    async fn active_turn_authority(
+        &self,
+        thread_id: &str,
+        turn_id: &str,
+        engine: &EngineHandle,
+    ) -> Option<crate::core::engine::RuntimePermissionAuthority> {
+        let active = self.active.lock().await;
+        let state = active.engines.get(thread_id)?;
+        let turn = state.active_turn.as_ref()?;
+        if turn.turn_id != turn_id {
+            return None;
+        }
+        Some(engine.runtime_permission_authority())
+    }
+
+    #[cfg(test)]
     async fn active_turn_flags(&self, thread_id: &str, turn_id: &str) -> Option<(bool, bool)> {
         let active = self.active.lock().await;
         let state = active.engines.get(thread_id)?;
@@ -7023,7 +7215,8 @@ impl RuntimeThreadManager {
         if turn.turn_id != turn_id {
             return None;
         }
-        Some((turn.auto_approve, turn.trust_mode))
+        let authority = state.engine.runtime_permission_authority();
+        Some((authority.auto_approve, authority.trust_mode))
     }
 
     async fn active_turn_id(&self, thread_id: &str) -> Option<String> {
@@ -7464,22 +7657,42 @@ fn enforce_lru_capacity(
     evicted
 }
 
-/// Resolves only explicit mode tokens to an app mode. Free-form prompt text is
-/// never a valid mode token: `parse_mode_opt` returns `None` unless the input is
-/// exactly `agent`/`plan`/`yolo` or numeric aliases `1`/`2`/`4`. Mode
-/// changes originate from the Tab cycle, `/mode`, the mode picker, or
-/// config/startup defaults, not from submitted natural-language prompt text.
-///
-/// Textual `auto` is a legacy alias for Agent while Auto is deferred (#3733).
-fn parse_mode_opt(mode: &str) -> Option<AppMode> {
-    match mode.trim().to_ascii_lowercase().as_str() {
-        "agent" | "auto" | "1" => Some(AppMode::Agent),
-        "plan" | "2" => Some(AppMode::Plan),
-        "yolo" | "4" | "bypass" | "bypass-permissions" | "bypasspermissions" => Some(AppMode::Yolo),
-        _ => None,
-    }
+/// Merge per-request compatibility inputs with a thread's canonical policy.
+/// A mode-only edit must preserve the effective posture of a legacy record
+/// even when that record predates `permission_posture`.
+fn runtime_policy_with_overrides(
+    thread: &ThreadRecord,
+    mode: Option<&str>,
+    permission_posture: Option<&str>,
+    auto_approve: Option<bool>,
+) -> Result<RuntimePolicyProjection> {
+    let requested_mode = mode.unwrap_or(&thread.mode);
+    let legacy_bypass_mode = mode.is_some_and(|mode| {
+        matches!(
+            mode.trim().to_ascii_lowercase().as_str(),
+            "yolo" | "4" | "bypass" | "bypass-permissions" | "bypasspermissions"
+        )
+    });
+    let inherited = RuntimePolicyProjection::from_persisted(
+        &thread.mode,
+        thread.permission_posture.as_deref(),
+        thread.auto_approve,
+    );
+    let requested_permission = match permission_posture {
+        Some(explicit) => Some(explicit),
+        None if auto_approve.is_some() || legacy_bypass_mode => None,
+        None => Some(inherited.permission_wire()),
+    };
+    RuntimePolicyProjection::from_request(requested_mode, requested_permission, auto_approve)
 }
 
+/// Compatibility parser retained for focused Runtime tests.
+#[cfg(test)]
+fn parse_mode_opt(mode: &str) -> Option<AppMode> {
+    crate::runtime_policy::parse_runtime_mode(mode)
+}
+
+#[cfg(test)]
 fn parse_mode(mode: &str) -> AppMode {
     parse_mode_opt(mode).unwrap_or(AppMode::Agent)
 }

crates/tui/src/runtime_threads/tests.rs

@@ -318,6 +318,7 @@ fn sample_thread(thread_id: &str) -> ThreadRecord {
         model_provider_id: None,
         workspace: PathBuf::from("."),
         mode: AppMode::Agent.as_setting().to_string(),
+        permission_posture: Some("ask".to_string()),
         allow_shell: false,
         trust_mode: false,
         auto_approve: false,
@@ -344,6 +345,7 @@ fn sample_turn(thread_id: &str, turn_id: &str, status: RuntimeTurnStatus) -> Tur
         ended_at: None,
         duration_ms: None,
         usage: None,
+        permission_posture: None,
         effective_provider: None,
         effective_provider_id: None,
         effective_billing_surface: None,
@@ -3422,8 +3424,6 @@ fn enforce_lru_capacity_does_not_loop_when_all_threads_are_active() {
             active_turn: Some(ActiveTurnState {
                 turn_id: "turn_a".to_string(),
                 interrupt_requested: false,
-                auto_approve: true,
-                trust_mode: false,
             }),
             route_identity: crate::config::ProviderIdentity {
                 provider: ApiProvider::Deepseek,
@@ -3441,8 +3441,6 @@ fn enforce_lru_capacity_does_not_loop_when_all_threads_are_active() {
             active_turn: Some(ActiveTurnState {
                 turn_id: "turn_b".to_string(),
                 interrupt_requested: false,
-                auto_approve: true,
-                trust_mode: false,
             }),
             route_identity: crate::config::ProviderIdentity {
                 provider: ApiProvider::Deepseek,
@@ -4210,8 +4208,6 @@ async fn update_thread_workspace_rejects_active_turn() -> Result<()> {
         state.active_turn = Some(ActiveTurnState {
             turn_id: "turn_live".to_string(),
             interrupt_requested: false,
-            auto_approve: false,
-            trust_mode: false,
         });
     }
 
@@ -4268,7 +4264,7 @@ async fn start_turn_passes_effective_auto_approve_to_engine() -> Result<()> {
     let harness = install_mock_engine(&manager, &thread.id).await;
     let mut rx_op = harness.rx_op;
 
-    let _turn = manager
+    let turn = manager
         .start_turn(
             &thread.id,
             StartTurnRequest {
@@ -4283,9 +4279,17 @@ async fn start_turn_passes_effective_auto_approve_to_engine() -> Result<()> {
             },
         )
         .await?;
+    assert_eq!(turn.permission_posture.as_deref(), Some("full_access"));
 
     match rx_op.recv().await {
-        Some(Op::SendMessage { auto_approve, .. }) => assert!(auto_approve),
+        Some(Op::SendMessage {
+            auto_approve,
+            approval_mode,
+            ..
+        }) => {
+            assert!(auto_approve);
+            assert_eq!(approval_mode, crate::tui::approval::ApprovalMode::Bypass);
+        }
         other => panic!("expected SendMessage op, got {other:?}"),
     }
 
@@ -4313,7 +4317,7 @@ async fn start_turn_can_override_thread_auto_approve_to_false() -> Result<()> {
     let harness = install_mock_engine(&manager, &thread.id).await;
     let mut rx_op = harness.rx_op;
 
-    let _turn = manager
+    let turn = manager
         .start_turn(
             &thread.id,
             StartTurnRequest {
@@ -4328,15 +4332,161 @@ async fn start_turn_can_override_thread_auto_approve_to_false() -> Result<()> {
             },
         )
         .await?;
+    assert_eq!(turn.permission_posture.as_deref(), Some("ask"));
 
     match rx_op.recv().await {
-        Some(Op::SendMessage { auto_approve, .. }) => assert!(!auto_approve),
+        Some(Op::SendMessage {
+            auto_approve,
+            approval_mode,
+            ..
+        }) => {
+            assert!(!auto_approve);
+            assert_eq!(approval_mode, crate::tui::approval::ApprovalMode::Suggest);
+        }
         other => panic!("expected SendMessage op, got {other:?}"),
     }
 
     Ok(())
 }
 
+#[tokio::test]
+async fn start_turn_enforces_and_records_auto_review_without_legacy_bypass() -> Result<()> {
+    let manager = test_manager(test_runtime_dir())?;
+    let thread = manager
+        .create_thread(CreateThreadRequest {
+            permission_posture: Some("ask".to_string()),
+            ..Default::default()
+        })
+        .await?;
+    let harness = install_mock_engine(&manager, &thread.id).await;
+    let mut rx_op = harness.rx_op;
+
+    let turn = manager
+        .start_turn(
+            &thread.id,
+            StartTurnRequest {
+                prompt: "review autonomously".to_string(),
+                permission_posture: Some("auto-review".to_string()),
+                ..Default::default()
+            },
+        )
+        .await?;
+
+    assert_eq!(turn.permission_posture.as_deref(), Some("auto_review"));
+    assert_eq!(
+        manager
+            .store
+            .load_turn(&turn.id)?
+            .permission_posture
+            .as_deref(),
+        Some("auto_review")
+    );
+    match rx_op.recv().await {
+        Some(Op::SendMessage {
+            auto_approve,
+            approval_mode,
+            ..
+        }) => {
+            assert!(!auto_approve);
+            assert_eq!(approval_mode, crate::tui::approval::ApprovalMode::Auto);
+        }
+        other => panic!("expected SendMessage op, got {other:?}"),
+    }
+    Ok(())
+}
+
+#[tokio::test]
+async fn active_turn_permission_posture_switches_use_the_engine_live_authority() -> Result<()> {
+    let manager = test_manager(test_runtime_dir())?;
+    let thread = manager
+        .create_thread(CreateThreadRequest {
+            permission_posture: Some("ask".to_string()),
+            ..Default::default()
+        })
+        .await?;
+    let mut harness = install_mock_engine(&manager, &thread.id).await;
+    let turn = manager
+        .start_turn(
+            &thread.id,
+            StartTurnRequest {
+                prompt: "keep working while permissions change".to_string(),
+                ..Default::default()
+            },
+        )
+        .await?;
+    assert!(matches!(
+        harness.rx_op.recv().await,
+        Some(Op::SendMessage {
+            approval_mode: crate::tui::approval::ApprovalMode::Suggest,
+            ..
+        })
+    ));
+
+    for (requested, canonical, expected_auto, expected_approval) in [
+        (
+            "auto-review",
+            "auto_review",
+            false,
+            crate::tui::approval::ApprovalMode::Auto,
+        ),
+        (
+            "full-access",
+            "full_access",
+            true,
+            crate::tui::approval::ApprovalMode::Bypass,
+        ),
+        (
+            "ask",
+            "ask",
+            false,
+            crate::tui::approval::ApprovalMode::Suggest,
+        ),
+    ] {
+        let updated = manager
+            .update_thread(
+                &thread.id,
+                UpdateThreadRequest {
+                    permission_posture: Some(requested.to_string()),
+                    ..Default::default()
+                },
+            )
+            .await?;
+        assert_eq!(updated.permission_posture.as_deref(), Some(canonical));
+        assert_eq!(updated.auto_approve, expected_auto);
+
+        match harness.rx_op.recv().await {
+            Some(Op::ChangeMode {
+                auto_approve,
+                approval_mode,
+                ..
+            }) => {
+                assert_eq!(auto_approve, expected_auto, "{requested}");
+                assert_eq!(approval_mode, expected_approval, "{requested}");
+            }
+            other => panic!("expected ChangeMode for {requested}, got {other:?}"),
+        }
+
+        let authority = {
+            let active = manager.active.lock().await;
+            active
+                .engines
+                .get(&thread.id)
+                .expect("active engine")
+                .engine
+                .runtime_permission_authority()
+        };
+        assert_eq!(authority.auto_approve, expected_auto, "{requested}");
+        assert_eq!(authority.approval_mode, expected_approval, "{requested}");
+        assert_eq!(
+            manager.active_turn_flags(&thread.id, &turn.id).await,
+            Some((expected_auto, false)),
+            "{requested} must replace the running turn's authority immediately"
+        );
+    }
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn compact_thread_preserves_thread_auto_approve_policy() -> Result<()> {
     let manager = test_manager(test_runtime_dir())?;
@@ -7339,6 +7489,78 @@ async fn approval_required_external_deny_is_denied() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn auto_review_force_prompt_is_denied_without_opening_a_modal() -> Result<()> {
+    let manager = test_manager(test_runtime_dir())?;
+    let thread = manager
+        .create_thread(CreateThreadRequest {
+            permission_posture: Some("auto-review".to_string()),
+            ..Default::default()
+        })
+        .await?;
+    let mut harness = install_mock_engine(&manager, &thread.id).await;
+    let turn = manager
+        .start_turn(
+            &thread.id,
+            StartTurnRequest {
+                prompt: "review the gated action".to_string(),
+                ..Default::default()
+            },
+        )
+        .await?;
+    assert!(matches!(
+        harness.rx_op.recv().await,
+        Some(Op::SendMessage {
+            approval_mode: crate::tui::approval::ApprovalMode::Auto,
+            ..
+        })
+    ));
+
+    harness
+        .tx_event
+        .send(EngineEvent::ApprovalRequired {
+            approval_key: "auto_hold".to_string(),
+            approval_grouping_key: "auto_hold".to_string(),
+            id: "tool_auto_hold".to_string(),
+            tool_name: "exec_command".to_string(),
+            description: "policy hold under auto review".to_string(),
+            input: serde_json::json!({}),
+            intent_summary: None,
+            approval_force_prompt: true,
+        })
+        .await?;
+
+    let decision = tokio::time::timeout(Duration::from_secs(2), harness.recv_approval_event())
+        .await
+        .context("Auto-Review hold should resolve without a modal")?;
+    assert_eq!(
+        decision,
+        Some(MockApprovalEvent::Denied {
+            id: "tool_auto_hold".to_string(),
+        })
+    );
+    assert_eq!(manager.pending_approvals_count(), 0);
+    assert!(manager.events_since(&thread.id, None)?.iter().any(|event| {
+        event.event == "approval.decided"
+            && event.payload.get("approval_id").and_then(Value::as_str) == Some("tool_auto_hold")
+            && event.payload.get("posture").and_then(Value::as_str) == Some("auto_review")
+    }));
+
+    harness
+        .tx_event
+        .send(EngineEvent::TurnComplete {
+            usage: Usage::default(),
+            status: TurnOutcomeStatus::Completed,
+            error: None,
+            tool_catalog: None,
+            base_url: None,
+        })
+        .await?;
+    let terminal = wait_for_terminal_turn(&manager, &turn.id, Duration::from_secs(2)).await?;
+    assert_eq!(terminal.status, RuntimeTurnStatus::Completed);
+    Ok(())
+}
+
 #[tokio::test]
 async fn approval_timeout_denies_clears_ui_and_next_turn_can_start() -> Result<()> {
     let _timeout_guard = test_approval_timeout_ms(25);
@@ -8492,6 +8714,7 @@ fn opening_manager_recovers_stale_queued_and_in_progress_work() -> Result<()> {
         model_provider_id: None,
         workspace: PathBuf::from("."),
         mode: "agent".to_string(),
+        permission_posture: None,
         allow_shell: false,
         trust_mode: false,
         auto_approve: false,
@@ -8559,6 +8782,7 @@ fn opening_manager_recovers_stale_queued_and_in_progress_work() -> Result<()> {
         ended_at: None,
         duration_ms: None,
         usage: None,
+        permission_posture: None,
         effective_provider: None,
         effective_provider_id: None,
         effective_billing_surface: None,
@@ -8584,6 +8808,7 @@ fn opening_manager_recovers_stale_queued_and_in_progress_work() -> Result<()> {
         ended_at: None,
         duration_ms: None,
         usage: None,
+        permission_posture: None,
         effective_provider: None,
         effective_provider_id: None,
         effective_billing_surface: None,
@@ -8657,6 +8882,21 @@ fn parse_mode_defaults_to_agent() {
     assert_eq!(parse_mode("plan"), AppMode::Plan);
 }
 
+#[test]
+fn mode_only_override_preserves_legacy_full_access_posture() -> Result<()> {
+    let mut thread = sample_thread("thr_legacy_full_access");
+    thread.permission_posture = None;
+    thread.auto_approve = true;
+
+    let policy = runtime_policy_with_overrides(&thread, Some("plan"), None, None)?;
+    assert_eq!(policy.mode, AppMode::Plan);
+    assert_eq!(policy.permission_wire(), "full_access");
+
+    let ask = runtime_policy_with_overrides(&thread, Some("act"), None, Some(false))?;
+    assert_eq!(ask.permission_wire(), "ask");
+    Ok(())
+}
+
 #[test]
 fn parse_mode_opt_resolves_explicit_tokens_and_aliases() {
     assert_eq!(parse_mode_opt("agent"), Some(AppMode::Agent));
@@ -8664,7 +8904,8 @@ fn parse_mode_opt_resolves_explicit_tokens_and_aliases() {
     assert_eq!(parse_mode_opt("plan"), Some(AppMode::Plan));
     assert_eq!(parse_mode_opt("2"), Some(AppMode::Plan));
     assert_eq!(parse_mode_opt("auto"), Some(AppMode::Agent));
-    assert_eq!(parse_mode_opt("3"), None);
+    assert_eq!(parse_mode_opt("operate"), Some(AppMode::Operate));
+    assert_eq!(parse_mode_opt("3"), Some(AppMode::Operate));
     assert_eq!(parse_mode_opt("yolo"), Some(AppMode::Yolo));
     assert_eq!(parse_mode_opt("4"), Some(AppMode::Yolo));
     assert_eq!(parse_mode_opt(" PLAN "), Some(AppMode::Plan));
@@ -8689,7 +8930,7 @@ fn parse_mode_wrapper_defaults_and_resolves_numeric_aliases() {
     assert_eq!(parse_mode("auto"), AppMode::Agent);
     assert_eq!(parse_mode("1"), AppMode::Agent);
     assert_eq!(parse_mode("2"), AppMode::Plan);
-    assert_eq!(parse_mode("3"), AppMode::Agent);
+    assert_eq!(parse_mode("3"), AppMode::Operate);
     assert_eq!(parse_mode("4"), AppMode::Yolo);
 }
 
@@ -8839,6 +9080,7 @@ fn seed_turns_with_user_messages(
             ended_at: Some(created_at),
             duration_ms: Some(0),
             usage: None,
+            permission_posture: None,
             effective_provider: None,
             effective_provider_id: None,
             effective_billing_surface: None,

crates/tui/src/shell_dispatcher.rs

@@ -28,6 +28,11 @@ use std::sync::Mutex;
 
 static LOG_MUTEX: Mutex<()> = Mutex::new(());
 
+#[cfg(test)]
+#[allow(dead_code)] // Direct integration-harness inclusion only needs the read barrier.
+#[path = "test_env_lock.rs"]
+pub(crate) mod test_env_lock;
+
 // ---------------------------------------------------------------------------
 // Shell kind
 // ---------------------------------------------------------------------------
@@ -42,11 +47,11 @@ pub enum ShellKind {
     WindowsPowerShell,
     /// Command Prompt (`cmd.exe`).
     Cmd,
-    /// Unix `/bin/sh` (or `$SHELL`-detected bash/zsh).
+    /// Unix `/bin/sh` fallback.
     Sh,
-    /// Bash — detected via `$SHELL` on either Unix or WSL/Git Bash on Windows.
+    /// Bash — detected via `$SHELL` on WSL/Git Bash, or constructed explicitly.
     Bash,
-    /// Any other POSIX shell from $SHELL (zsh, fish, dash, ...).
+    /// The exact shell executable selected by Unix `$SHELL`.
     Custom { binary: String, flag: String },
 }
 
@@ -69,7 +74,10 @@ impl ShellKind {
             #[cfg(not(windows))]
             ShellKind::Cmd => "cmd",
 
+            #[cfg(windows)]
             ShellKind::Sh => "sh",
+            #[cfg(not(windows))]
+            ShellKind::Sh => "/bin/sh",
             ShellKind::Bash => "bash",
             ShellKind::Custom { binary, .. } => binary,
         }
@@ -95,7 +103,17 @@ impl ShellKind {
 
     /// Returns true when this is a PowerShell-family shell.
     pub fn is_powershell(&self) -> bool {
-        matches!(self, ShellKind::Pwsh | ShellKind::WindowsPowerShell)
+        match self {
+            ShellKind::Pwsh | ShellKind::WindowsPowerShell => true,
+            ShellKind::Custom { binary, .. } => Path::new(binary)
+                .file_name()
+                .and_then(|name| name.to_str())
+                .is_some_and(|name| {
+                    let name = name.to_ascii_lowercase();
+                    name.contains("pwsh") || name.contains("powershell")
+                }),
+            ShellKind::Cmd | ShellKind::Sh | ShellKind::Bash => false,
+        }
     }
 }
 
@@ -217,8 +235,8 @@ impl ShellDispatcher {
     ///
     /// ## Detection order (Unix)
     ///
-    /// 1. `$SHELL` — if it contains `bash`, use `Bash`; otherwise use the
-    ///    actual binary path via `Custom`.
+    /// 1. `$SHELL` — preserve its actual executable via `Custom`; bare names
+    ///    are resolved against the current `PATH` once at detection time.
     /// 2. `/bin/sh` fallback.
     pub fn detect() -> Self {
         let kind = Self::detect_shell();
@@ -403,6 +421,17 @@ impl ShellDispatcher {
     // -- Detection --------------------------------------------------------
 
     fn detect_shell() -> ShellKind {
+        #[cfg(test)]
+        {
+            test_env_lock::with_test_env_lock(Self::detect_shell_unlocked)
+        }
+        #[cfg(not(test))]
+        {
+            Self::detect_shell_unlocked()
+        }
+    }
+
+    fn detect_shell_unlocked() -> ShellKind {
         #[cfg(windows)]
         {
             // 1. $env:SHELL — WSL interop or Git Bash often set this.
@@ -430,28 +459,43 @@ impl ShellDispatcher {
 
         #[cfg(not(windows))]
         {
-            // 1. $SHELL environment variable (Unix)
-            if let Ok(shell) = std::env::var("SHELL") {
-                let lower = shell.to_lowercase();
-                if lower.contains("bash") {
-                    return ShellKind::Bash;
-                }
-                if lower.contains("pwsh") {
-                    return ShellKind::Pwsh;
-                }
-                if lower.contains("powershell") {
-                    return ShellKind::WindowsPowerShell;
-                }
-                return ShellKind::Custom {
-                    binary: shell,
-                    flag: "-c".to_string(),
-                };
+            if let Ok(shell) = std::env::var("SHELL")
+                && let Some(kind) = Self::unix_shell_kind(&shell)
+            {
+                return kind;
             }
 
             ShellKind::Sh
         }
     }
 
+    #[cfg(not(windows))]
+    fn unix_shell_kind(shell: &str) -> Option<ShellKind> {
+        let shell = shell.trim();
+        if shell.is_empty() {
+            return None;
+        }
+        let path = Path::new(shell);
+        let binary = if path.is_absolute() || path.components().count() > 1 {
+            shell.to_string()
+        } else {
+            std::env::var_os("PATH")
+                .and_then(|path| {
+                    std::env::split_paths(&path)
+                        .map(|dir| dir.join(shell))
+                        .find(|candidate| candidate.is_file())
+                })
+                .map_or_else(
+                    || shell.to_string(),
+                    |path| path.to_string_lossy().into_owned(),
+                )
+        };
+        Some(ShellKind::Custom {
+            binary,
+            flag: "-c".to_string(),
+        })
+    }
+
     /// Check PATH first, then fall back to well-known install directories.
     #[cfg(windows)]
     fn find_exe(name: &str) -> bool {
@@ -522,10 +566,40 @@ mod tests {
             assert_eq!(ShellKind::WindowsPowerShell.binary(), "powershell");
             assert_eq!(ShellKind::Cmd.binary(), "cmd");
         }
+        #[cfg(windows)]
         assert_eq!(ShellKind::Sh.binary(), "sh");
+        #[cfg(not(windows))]
+        assert_eq!(ShellKind::Sh.binary(), "/bin/sh");
         assert_eq!(ShellKind::Bash.binary(), "bash");
     }
 
+    #[cfg(not(windows))]
+    #[test]
+    fn unix_shell_detection_preserves_absolute_executable_paths() {
+        let bash = ShellDispatcher::unix_shell_kind("/bin/bash").expect("bash shell");
+        assert_eq!(
+            bash,
+            ShellKind::Custom {
+                binary: "/bin/bash".to_string(),
+                flag: "-c".to_string(),
+            }
+        );
+
+        let pwsh =
+            ShellDispatcher::unix_shell_kind("/opt/homebrew/bin/pwsh").expect("PowerShell path");
+        assert!(pwsh.is_powershell());
+        assert_eq!(pwsh.binary(), "/opt/homebrew/bin/pwsh");
+
+        let dispatcher = ShellDispatcher {
+            kind: ShellDispatcher::unix_shell_kind("/bin/sh").expect("POSIX shell"),
+        };
+        let mut command = dispatcher.build_command("printf path-independent");
+        command.env_clear();
+        let output = command.output().expect("absolute shell must not need PATH");
+        assert!(output.status.success(), "{output:?}");
+        assert_eq!(output.stdout, b"path-independent");
+    }
+
     #[test]
     fn detect_returns_some_shell() {
         let dispatcher = global_dispatcher();

crates/tui/src/test_env_lock.rs

@@ -0,0 +1,193 @@
+//! Process-wide test-environment barrier owned by shell dispatch.
+//!
+//! `shell_dispatcher.rs` is also compiled directly by integration harnesses,
+//! outside the main binary crate. Keeping the barrier under that module makes
+//! shell detection self-contained while `crate::test_support` re-exports the
+//! same instance to its existing environment-mutating callers in the main crate.
+
+use std::sync::{Mutex, MutexGuard, OnceLock, TryLockError};
+use std::thread::ThreadId;
+
+fn env_lock() -> &'static Mutex<()> {
+    static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
+    LOCK.get_or_init(|| Mutex::new(()))
+}
+
+/// Who currently counts as "inside" the process-wide env lock.
+///
+/// The owner is the thread holding [`TestEnvLock`]. `adopted` holds helper
+/// threads that owner explicitly enrolled with [`join_env_scope`] — see that
+/// function for why a worker thread of the current test must not be treated as
+/// a foreign reader.
+#[derive(Default)]
+struct EnvScope {
+    /// Bumped on every acquisition, so a ticket minted by an earlier test can
+    /// never enroll a thread into a later test's environment.
+    generation: u64,
+    owner: Option<ThreadId>,
+    adopted: Vec<ThreadId>,
+}
+
+fn env_scope() -> &'static Mutex<EnvScope> {
+    static SCOPE: OnceLock<Mutex<EnvScope>> = OnceLock::new();
+    SCOPE.get_or_init(|| Mutex::new(EnvScope::default()))
+}
+
+fn lock_env_scope() -> MutexGuard<'static, EnvScope> {
+    match env_scope().lock() {
+        Ok(scope) => scope,
+        Err(poisoned) => poisoned.into_inner(),
+    }
+}
+
+fn open_env_scope() {
+    let mut scope = lock_env_scope();
+    scope.generation = scope.generation.wrapping_add(1);
+    scope.owner = Some(std::thread::current().id());
+    scope.adopted.clear();
+}
+
+fn current_thread_owns_contended_env_lock() -> bool {
+    let scope = lock_env_scope();
+    let current = std::thread::current().id();
+    scope.owner == Some(current) || scope.adopted.contains(&current)
+}
+
+/// Proof that the calling thread owns a live [`lock_test_env`] scope, handed to
+/// a worker thread so it can join that scope with [`join_env_scope`].
+///
+/// Returns `None` when the caller is not the owner, so a ticket can never be
+/// minted on behalf of a test that did not seal the environment.
+#[derive(Clone, Copy, Debug)]
+pub(crate) struct EnvScopeTicket {
+    generation: u64,
+}
+
+impl EnvScopeTicket {
+    /// Which sealed environment this ticket authorizes. Callers that gate real
+    /// disk writes on a live scope key their bookkeeping by this value, so a
+    /// straggler from generation N can never be mistaken for work belonging to
+    /// generation N+1.
+    pub(crate) fn generation(&self) -> u64 {
+        self.generation
+    }
+}
+
+/// The generation of the env scope the calling thread is currently inside, as
+/// owner or as a [`join_env_scope`]-adopted worker; `None` when the thread is a
+/// foreign reader with no sealed environment of its own.
+///
+/// This is the authorization primitive for anything that must only touch disk
+/// on behalf of a test that actually sealed `HOME`. A process-global "writes
+/// are enabled" flag cannot distinguish unrelated parallel tests.
+pub(crate) fn current_env_scope_generation() -> Option<u64> {
+    let scope = lock_env_scope();
+    let current = std::thread::current().id();
+    if scope.owner == Some(current) || scope.adopted.contains(&current) {
+        Some(scope.generation)
+    } else {
+        None
+    }
+}
+
+pub(crate) fn env_scope_ticket() -> Option<EnvScopeTicket> {
+    let scope = lock_env_scope();
+    (scope.owner == Some(std::thread::current().id())).then_some(EnvScopeTicket {
+        generation: scope.generation,
+    })
+}
+
+/// Enroll the calling thread in the ticket's env scope for as long as the
+/// returned guard lives.
+///
+/// [`with_test_env_lock`] stops a foreign test from resolving another test's
+/// temporary `HOME`. A helper thread doing work for the sealing test must see
+/// that same environment without blocking on the mutex its owner holds.
+pub(crate) fn join_env_scope(ticket: Option<EnvScopeTicket>) -> Option<EnvScopeMembership> {
+    let ticket = ticket?;
+    let mut scope = lock_env_scope();
+    if scope.owner.is_none() || scope.generation != ticket.generation {
+        return None;
+    }
+    let thread = std::thread::current().id();
+    if !scope.adopted.contains(&thread) {
+        scope.adopted.push(thread);
+    }
+    Some(EnvScopeMembership {
+        generation: ticket.generation,
+        thread,
+    })
+}
+
+pub(crate) struct EnvScopeMembership {
+    generation: u64,
+    thread: ThreadId,
+}
+
+impl Drop for EnvScopeMembership {
+    fn drop(&mut self) {
+        let mut scope = lock_env_scope();
+        if scope.generation == self.generation {
+            scope.adopted.retain(|thread| *thread != self.thread);
+        }
+    }
+}
+
+/// Owned process-wide test-environment lock.
+///
+/// Clearing the owner before the underlying mutex unlocks keeps re-entrant
+/// reader detection exact. Closing the scope also evicts adopted workers, so
+/// enrollment cannot outlive the test that granted it.
+pub(crate) struct TestEnvLock {
+    _guard: MutexGuard<'static, ()>,
+}
+
+impl Drop for TestEnvLock {
+    fn drop(&mut self) {
+        let mut scope = lock_env_scope();
+        if scope.owner == Some(std::thread::current().id()) {
+            scope.owner = None;
+            scope.adopted.clear();
+        }
+    }
+}
+
+/// Acquire the process-wide env-var mutex.
+///
+/// If a prior test panicked while holding the lock, recover the guard instead
+/// of cascading failures across unrelated tests.
+pub(crate) fn lock_test_env() -> TestEnvLock {
+    let guard = match env_lock().lock() {
+        Ok(guard) => guard,
+        Err(poisoned) => poisoned.into_inner(),
+    };
+    open_env_scope();
+    TestEnvLock { _guard: guard }
+}
+
+/// Read process-global test environment while respecting [`lock_test_env`].
+///
+/// The owner check makes the barrier re-entrant for a test that reads its own
+/// guarded override.
+pub(crate) fn with_test_env_lock<T>(read: impl FnOnce() -> T) -> T {
+    if current_thread_owns_contended_env_lock() {
+        return read();
+    }
+
+    let _guard = lock_test_env();
+    read()
+}
+
+pub(crate) fn current_thread_holds_test_env_lock() -> bool {
+    match env_lock().try_lock() {
+        Ok(guard) => {
+            drop(guard);
+            false
+        }
+        Err(TryLockError::Poisoned(poisoned)) => {
+            drop(poisoned.into_inner());
+            false
+        }
+        Err(TryLockError::WouldBlock) => current_thread_owns_contended_env_lock(),
+    }
+}

crates/tui/src/test_support.rs

@@ -2,10 +2,15 @@
 
 use std::ffi::{OsStr, OsString};
 use std::path::{Path, PathBuf};
-use std::sync::{Mutex, MutexGuard, OnceLock, TryLockError};
-use std::thread::ThreadId;
+use std::sync::{Mutex, OnceLock};
 use std::time::{SystemTime, UNIX_EPOCH};
 
+pub(crate) use crate::shell_dispatcher::test_env_lock::{
+    EnvScopeMembership, EnvScopeTicket, TestEnvLock, current_env_scope_generation,
+    current_thread_holds_test_env_lock, env_scope_ticket, join_env_scope, lock_test_env,
+    with_test_env_lock,
+};
+
 /// Process-wide state root for unit tests that do not intentionally provide an
 /// explicit config/settings path.
 ///
@@ -44,215 +49,11 @@ pub(crate) fn future_test_jwt(label: &str) -> String {
     format!("test.{payload}.{label}")
 }
 
-fn env_lock() -> &'static Mutex<()> {
-    static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
-    LOCK.get_or_init(|| Mutex::new(()))
-}
-
 fn state_io_lock() -> &'static Mutex<()> {
     static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
     LOCK.get_or_init(|| Mutex::new(()))
 }
 
-/// Who currently counts as "inside" the process-wide env lock.
-///
-/// The owner is the thread holding [`TestEnvLock`]. `adopted` holds helper
-/// threads that owner explicitly enrolled with [`join_env_scope`] — see that
-/// function for why a *worker thread of the current test* must not be treated
-/// as a foreign reader.
-#[derive(Default)]
-struct EnvScope {
-    /// Bumped on every acquisition, so a ticket minted by an earlier test can
-    /// never enroll a thread into a later test's environment.
-    generation: u64,
-    owner: Option<ThreadId>,
-    adopted: Vec<ThreadId>,
-}
-
-fn env_scope() -> &'static Mutex<EnvScope> {
-    static SCOPE: OnceLock<Mutex<EnvScope>> = OnceLock::new();
-    SCOPE.get_or_init(|| Mutex::new(EnvScope::default()))
-}
-
-fn lock_env_scope() -> MutexGuard<'static, EnvScope> {
-    match env_scope().lock() {
-        Ok(scope) => scope,
-        Err(poisoned) => poisoned.into_inner(),
-    }
-}
-
-fn open_env_scope() {
-    let mut scope = lock_env_scope();
-    scope.generation = scope.generation.wrapping_add(1);
-    scope.owner = Some(std::thread::current().id());
-    scope.adopted.clear();
-}
-
-fn current_thread_owns_contended_env_lock() -> bool {
-    let scope = lock_env_scope();
-    let current = std::thread::current().id();
-    scope.owner == Some(current) || scope.adopted.contains(&current)
-}
-
-/// Proof that the calling thread owns a live [`lock_test_env`] scope, handed to
-/// a worker thread so it can join that scope with [`join_env_scope`].
-///
-/// Returns `None` when the caller is not the owner, so a ticket can never be
-/// minted on behalf of a test that did not seal the environment.
-#[derive(Clone, Copy, Debug)]
-pub(crate) struct EnvScopeTicket {
-    generation: u64,
-}
-
-impl EnvScopeTicket {
-    /// Which sealed environment this ticket authorizes. Callers that gate real
-    /// disk writes on a live scope key their bookkeeping by this value, so a
-    /// straggler from generation N can never be mistaken for work belonging to
-    /// generation N+1.
-    pub(crate) fn generation(&self) -> u64 {
-        self.generation
-    }
-}
-
-/// The generation of the env scope the *calling thread* is currently inside, as
-/// owner or as an [`join_env_scope`]-adopted worker; `None` when the thread is a
-/// foreign reader with no sealed environment of its own.
-///
-/// This is the authorization primitive for anything that must only touch disk on
-/// behalf of a test that actually sealed `HOME`. A process-global "writes are
-/// enabled" flag cannot answer that question: it is true for the whole time
-/// *some* test has sealed the environment, including for unrelated tests running
-/// in parallel that would then resolve — and write — that test's paths, or block
-/// on its env lock.
-pub(crate) fn current_env_scope_generation() -> Option<u64> {
-    let scope = lock_env_scope();
-    let current = std::thread::current().id();
-    if scope.owner == Some(current) || scope.adopted.contains(&current) {
-        Some(scope.generation)
-    } else {
-        None
-    }
-}
-
-pub(crate) fn env_scope_ticket() -> Option<EnvScopeTicket> {
-    let scope = lock_env_scope();
-    (scope.owner == Some(std::thread::current().id())).then_some(EnvScopeTicket {
-        generation: scope.generation,
-    })
-}
-
-/// Enroll the calling thread in the ticket's env scope for as long as the
-/// returned guard lives.
-///
-/// [`with_test_env_lock`] exists to stop a *foreign* test from resolving
-/// another test's temporary `HOME`. A helper thread doing work on behalf of the
-/// sealing test is not foreign: it must see that same temporary environment,
-/// and — decisively — it must not block on a mutex its own test holds for the
-/// whole test body. Blocking there is a lock-order inversion: the helper parks
-/// holding whatever lock it took first, and the test thread then parks waiting
-/// for that lock. Enrolling makes the barrier re-entrant for the helper, which
-/// is what makes the inversion impossible rather than merely unlikely.
-///
-/// Returns `None` (declining to enroll) when the scope has already closed or
-/// moved on, so a straggler thread from a finished test still gets the
-/// foreign-reader treatment.
-pub(crate) fn join_env_scope(ticket: Option<EnvScopeTicket>) -> Option<EnvScopeMembership> {
-    let ticket = ticket?;
-    let mut scope = lock_env_scope();
-    if scope.owner.is_none() || scope.generation != ticket.generation {
-        return None;
-    }
-    let thread = std::thread::current().id();
-    if !scope.adopted.contains(&thread) {
-        scope.adopted.push(thread);
-    }
-    Some(EnvScopeMembership {
-        generation: ticket.generation,
-        thread,
-    })
-}
-
-pub(crate) struct EnvScopeMembership {
-    generation: u64,
-    thread: ThreadId,
-}
-
-impl Drop for EnvScopeMembership {
-    fn drop(&mut self) {
-        let mut scope = lock_env_scope();
-        if scope.generation == self.generation {
-            scope.adopted.retain(|thread| *thread != self.thread);
-        }
-    }
-}
-
-/// Owned process-wide test-environment lock.
-///
-/// Clearing the owner before the underlying mutex unlocks keeps re-entrant
-/// reader detection exact; a stale thread id could otherwise let the previous
-/// owner bypass a newly acquired lock during its tiny owner-registration
-/// window. Closing the scope also evicts every adopted worker thread, so an
-/// enrollment cannot outlive the test that granted it.
-pub(crate) struct TestEnvLock {
-    _guard: MutexGuard<'static, ()>,
-}
-
-impl Drop for TestEnvLock {
-    fn drop(&mut self) {
-        let mut scope = lock_env_scope();
-        if scope.owner == Some(std::thread::current().id()) {
-            scope.owner = None;
-            scope.adopted.clear();
-        }
-    }
-}
-
-/// Acquire the process-wide env-var mutex.
-///
-/// If a prior test panicked while holding the lock, recover the guard instead
-/// of cascading failures across unrelated tests.
-pub(crate) fn lock_test_env() -> TestEnvLock {
-    let guard = match env_lock().lock() {
-        Ok(guard) => guard,
-        Err(poisoned) => poisoned.into_inner(),
-    };
-    open_env_scope();
-    TestEnvLock { _guard: guard }
-}
-
-/// Read process-global test environment while respecting [`lock_test_env`].
-///
-/// Config-path writers hold the mutex for their whole test. Production path
-/// resolution normally only reads the environment, but those reads still have
-/// to wait or they can resolve another test's temporary config and later write
-/// into it. The owner check makes the barrier re-entrant for a test that reads
-/// its own guarded override.
-pub(crate) fn with_test_env_lock<T>(read: impl FnOnce() -> T) -> T {
-    if current_thread_owns_contended_env_lock() {
-        return read();
-    }
-
-    // Acquire through the owner-tracking guard so nested environment readers
-    // remain re-entrant. This matters for config loading: the outer override
-    // pass holds the barrier while helper functions read individual variables.
-    let _guard = lock_test_env();
-    read()
-}
-
-pub(crate) fn current_thread_holds_test_env_lock() -> bool {
-    match env_lock().try_lock() {
-        Ok(guard) => {
-            drop(guard);
-            false
-        }
-        Err(TryLockError::Poisoned(poisoned)) => {
-            drop(poisoned.into_inner());
-            false
-        }
-        Err(TryLockError::WouldBlock) => current_thread_owns_contended_env_lock(),
-    }
-}
-
 /// Serialize read/merge/write operations against the process-wide isolated
 /// test state root.
 ///

crates/tui/src/tui/approval.rs

@@ -81,7 +81,9 @@ impl ApprovalMode {
         match value.trim().to_ascii_lowercase().as_str() {
             "auto" | "auto-review" | "auto_review" => Some(ApprovalMode::Auto),
             "bypass" | "yolo" | "dontask" | "dont_ask" | "bypass-permissions"
-            | "bypasspermissions" | "full-access" | "full" => Some(ApprovalMode::Bypass),
+            | "bypasspermissions" | "full-access" | "full_access" | "full" => {
+                Some(ApprovalMode::Bypass)
+            }
             "suggest" | "suggested" | "on-request" | "untrusted" | "ask" => {
                 Some(ApprovalMode::Suggest)
             }
@@ -4377,6 +4379,10 @@ diff --git a/src/b.rs b/src/b.rs
             ApprovalMode::from_config_value("on-request"),
             Some(ApprovalMode::Suggest)
         );
+        assert_eq!(
+            ApprovalMode::from_config_value("full_access"),
+            Some(ApprovalMode::Bypass)
+        );
         assert_eq!(
             ApprovalMode::from_config_value("deny"),
             Some(ApprovalMode::Never)

crates/tui/src/tui/ui.rs

@@ -4770,6 +4770,20 @@ async fn run_event_loop(
                                 );
                                 let _ = engine_handle.approve_tool_call(id.clone()).await;
                             }
+                            ApprovalRequestDisposition::AutoDenyAutoReview => {
+                                log_sensitive_event(
+                                    "tool.approval.auto_deny_auto_review",
+                                    serde_json::json!({
+                                        "tool_name": tool_name,
+                                        "session_id": app.current_session_id,
+                                        "mode": app.mode.label(),
+                                    }),
+                                );
+                                let _ = engine_handle.deny_tool_call(id.clone()).await;
+                                app.status_message = Some(format!(
+                                    "Auto-Review held tool '{tool_name}' without pausing"
+                                ));
+                            }
                             ApprovalRequestDisposition::AutoDenyNeverPosture => {
                                 log_sensitive_event(
                                     "tool.approval.auto_deny",

crates/tui/src/tui/ui/tests.rs

@@ -3743,13 +3743,12 @@ fn session_approved_cache_keeps_tool_name_session_grants() {
 }
 
 #[test]
-fn forced_approval_prompt_bypasses_auto_mode_shortcut() {
+fn auto_review_force_prompt_fails_closed_without_a_modal() {
     use crate::core::authority::ApprovalRequestDisposition;
     let mut app = create_test_app();
     app.approval_mode = ApprovalMode::Auto;
 
-    // Auto-Review does not full-access auto-approve; a forced hold still
-    // reaches a modal Prompt disposition (not Full Access policy deny).
+    // Auto-Review is autonomous: a forced hold is denied without pausing.
     assert_eq!(
         resolve_ui_approval_disposition(
             &app,
@@ -3758,7 +3757,7 @@ fn forced_approval_prompt_bypasses_auto_mode_shortcut() {
             "key",
             true,
         ),
-        ApprovalRequestDisposition::Prompt
+        ApprovalRequestDisposition::AutoDenyAutoReview
     );
 }
 
@@ -3786,7 +3785,7 @@ fn forced_approval_prompt_bypasses_session_approval_shortcut() {
 }
 
 #[test]
-fn full_access_auto_approves_requests_while_auto_review_does_not() {
+fn full_access_auto_approves_requests_while_auto_review_holds_without_a_modal() {
     use crate::core::authority::ApprovalRequestDisposition;
     let mut app = create_test_app();
     app.approval_mode = ApprovalMode::Auto;
@@ -3798,7 +3797,7 @@ fn full_access_auto_approves_requests_while_auto_review_does_not() {
             "key",
             false,
         ),
-        ApprovalRequestDisposition::Prompt
+        ApprovalRequestDisposition::AutoDenyAutoReview
     );
 
     app.approval_mode = ApprovalMode::Bypass;

crates/tui/tests/qa_pty.rs

@@ -2372,8 +2372,9 @@ fn pty_text_sse(content: &str) -> String {
 /// `File.patch` call performs an update, create, delete, and byte-identical
 /// delete/create rename in a single transaction; the second response settles
 /// the turn. No provider or external network is involved.
-fn spawn_file_mutation_screen_fixture()
--> anyhow::Result<(String, std::thread::JoinHandle<anyhow::Result<()>>)> {
+fn spawn_file_mutation_screen_fixture(
+    tool_allowed: bool,
+) -> anyhow::Result<(String, std::thread::JoinHandle<anyhow::Result<()>>)> {
     let listener = TcpListener::bind("127.0.0.1:0")?;
     listener.set_nonblocking(true)?;
     let address = listener.local_addr()?;
@@ -2412,6 +2413,11 @@ diff --git a/delete.txt b/delete.txt
         ),
         pty_text_sse("FILE-MUTATION-FIXTURE-DONE"),
     ];
+    let expected_result_marker = if tool_allowed {
+        "files_applied"
+    } else {
+        "destructive action requires explicit review"
+    };
 
     let handle = std::thread::spawn(move || -> anyhow::Result<()> {
         let deadline = Instant::now() + Duration::from_secs(45);
@@ -2451,12 +2457,12 @@ diff --git a/delete.txt b/delete.txt
                         contract_errors.push("initial request omitted the fixture prompt".into());
                     }
                     1 if !(request_contract.contains("call_file_mutation_pty")
-                        && request_contract.contains("files_applied")
+                        && request_contract.contains(expected_result_marker)
                         && request_contract.contains("\"role\":\"tool\"")) =>
                     {
                         let sample = request_contract.chars().take(1_200).collect::<String>();
                         contract_errors.push(format!(
-                            "settling request omitted the successful File result: {sample}"
+                            "settling request omitted the expected File result: {sample}"
                         ));
                     }
                     0 | 1 => {}
@@ -2545,7 +2551,7 @@ fn work_surface_file_mutation_modes_are_truthful_in_real_pty_frames() -> anyhow:
             "full", 140_u16, 40_u16, "dark", false, false, "ask", "ask", true,
         ),
         (
-            "summary", 100, 32, "light", false, false, "auto", "auto", true,
+            "summary", 100, 32, "light", false, false, "auto", "auto", false,
         ),
         (
             "off",
@@ -2618,7 +2624,10 @@ fn work_surface_file_mutation_modes_are_truthful_in_real_pty_frames() -> anyhow:
             );
         }
 
-        let (base_url, server) = spawn_file_mutation_screen_fixture()?;
+        // Auto-Review deliberately has no approval escape hatch for destructive
+        // create/delete work; Ask and Full Access can complete the transaction.
+        let tool_allowed = permission_posture != "auto";
+        let (base_url, server) = spawn_file_mutation_screen_fixture(tool_allowed)?;
         let mut h = spawn_file_mutation_harness(&ws, &base_url, rows, cols, ascii_safe)?;
         enter_launch_session(&mut h)?;
         assert_real_pty_frame_geometry(h.frame(), cols, rows);
@@ -2633,31 +2642,59 @@ fn work_surface_file_mutation_modes_are_truthful_in_real_pty_frames() -> anyhow:
             h.send(b"y")?;
         }
         h.wait_for_text("FILE-MUTATION-FIXTURE-DONE", Duration::from_secs(20))?;
-        h.wait_for(
-            |frame| frame.contains("4 files") && frame.contains("done"),
-            Duration::from_secs(10),
-        )?;
+        if tool_allowed {
+            h.wait_for(
+                |frame| frame.contains("4 files") && frame.contains("done"),
+                Duration::from_secs(10),
+            )?;
+        } else {
+            h.wait_for(
+                |frame| {
+                    frame.contains("tool issue")
+                        && frame.contains("destructive action")
+                        && frame.contains("done")
+                },
+                Duration::from_secs(10),
+            )?;
+        }
         h.wait_for_idle(Duration::from_millis(250), Duration::from_secs(3))?;
-        assert!(
-            !h.frame().contains("Wrote 4 files"),
-            "completed file-operation summary leaked into ambient chrome:\n{}",
-            h.frame().debug_dump()
-        );
 
-        assert_eq!(
-            std::fs::read_to_string(ws.workspace().join("new-name.txt"))?,
-            "RENAME-SENTINEL\n"
-        );
-        assert!(!ws.workspace().join("old-name.txt").exists());
-        assert_eq!(
-            std::fs::read_to_string(ws.workspace().join("update.txt"))?,
-            "DIFF-NEW-SENTINEL\n"
-        );
-        assert_eq!(
-            std::fs::read_to_string(ws.workspace().join("create.txt"))?,
-            "CREATE-SENTINEL\n"
-        );
-        assert!(!ws.workspace().join("delete.txt").exists());
+        if tool_allowed {
+            assert!(
+                !h.frame().contains("Wrote 4 files"),
+                "completed file-operation summary leaked into ambient chrome:\n{}",
+                h.frame().debug_dump()
+            );
+            assert_eq!(
+                std::fs::read_to_string(ws.workspace().join("new-name.txt"))?,
+                "RENAME-SENTINEL\n"
+            );
+            assert!(!ws.workspace().join("old-name.txt").exists());
+            assert_eq!(
+                std::fs::read_to_string(ws.workspace().join("update.txt"))?,
+                "DIFF-NEW-SENTINEL\n"
+            );
+            assert_eq!(
+                std::fs::read_to_string(ws.workspace().join("create.txt"))?,
+                "CREATE-SENTINEL\n"
+            );
+            assert!(!ws.workspace().join("delete.txt").exists());
+        } else {
+            assert_eq!(
+                std::fs::read_to_string(ws.workspace().join("old-name.txt"))?,
+                "RENAME-SENTINEL\n"
+            );
+            assert_eq!(
+                std::fs::read_to_string(ws.workspace().join("update.txt"))?,
+                "DIFF-OLD-SENTINEL\n"
+            );
+            assert!(!ws.workspace().join("new-name.txt").exists());
+            assert!(!ws.workspace().join("create.txt").exists());
+            assert_eq!(
+                std::fs::read_to_string(ws.workspace().join("delete.txt"))?,
+                "DELETE-SENTINEL\n"
+            );
+        }
 
         let settled_frame = h.frame().text();
         std::thread::sleep(Duration::from_millis(300));
@@ -2696,10 +2733,11 @@ fn work_surface_file_mutation_modes_are_truthful_in_real_pty_frames() -> anyhow:
             }
             "summary" => {
                 assert!(
-                    scroll_until(&mut h, ScrollDir::Up, "+2 -2"),
-                    "summary mode omitted semantic stats:\n{}",
+                    scroll_until(&mut h, ScrollDir::Up, "+3 / -3"),
+                    "held Auto-Review mutation omitted semantic stats:\n{}",
                     h.frame().debug_dump()
                 );
+                assert!(h.frame().contains("explicit review"));
                 assert!(!scroll_until(&mut h, ScrollDir::Up, "DIFF-NEW-SENTINEL"));
                 assert!(!scroll_until(&mut h, ScrollDir::Down, "DIFF-NEW-SENTINEL"));
             }

scripts/runtime-contract-budget.json

@@ -5,43 +5,43 @@
     "fixture_id": "representative-v1",
     "stages": {
       "base": {
-        "bytes": 11232,
-        "identity_sha256": "cfc75fc3e8faceddaec4e0611534245d83e36cac0c121b22be050866a1a0b9a7"
+        "bytes": 11237,
+        "identity_sha256": "9e82514f76885a6d4b0441334f48f831b2feaa3a42ea93532994ab7d00cd85da"
       },
       "goal": {
-        "bytes": 14563,
+        "bytes": 14568,
         "delta_bytes": 80,
-        "identity_sha256": "26606dd13fe77413eaf8680d9663bf925658a57bbcbd8d5a882226f7e419089b"
+        "identity_sha256": "e5c14a5cb749ef6295aa47898bb8710576e96a5b240214ef6be6b60c14854b13"
       },
       "handoff": {
-        "bytes": 14951,
+        "bytes": 14956,
         "delta_bytes": 388,
-        "identity_sha256": "bd2e996753c4611ad99431b2fc7774a091519ed4f616b2be386e757bc6cf1526"
+        "identity_sha256": "179e43b0fdf4c59f19c797fdac1752097cf57688380f594a8164e69ce6ec9d43"
       },
       "instructions": {
-        "bytes": 11684,
+        "bytes": 11689,
         "delta_bytes": 131,
-        "identity_sha256": "dc5fb6700cb5b96aad1bb2c2cb39299241fb62acd785fabd57ce4177ba6b9da6"
+        "identity_sha256": "59623b7b35c7b82c60560ebadaff353c041a24e14537cdc3b35324240708e05b"
       },
       "memory": {
-        "bytes": 14483,
+        "bytes": 14488,
         "delta_bytes": 1649,
-        "identity_sha256": "95e89028c7d09988878376d4d836c90bc2c6c5f87147ce46738a59a74abc8e1f"
+        "identity_sha256": "442a18f5f8579646565804e7fa3610bb9138188438ed04f6854a9613c5f3f36a"
       },
       "project": {
-        "bytes": 11553,
+        "bytes": 11558,
         "delta_bytes": 321,
-        "identity_sha256": "435a780b13f567aa948a57eba201ad8cd154a7851cc72b44785d9d875c63e1bb"
+        "identity_sha256": "65972b42822f4d25c18f0e54fbe8f996699327de0693ec514b72f0f201d91b84"
       },
       "skill": {
-        "bytes": 12834,
+        "bytes": 12839,
         "delta_bytes": 1150,
-        "identity_sha256": "7913d1a133f8fbdc4edad68f561f1f81199b2eac318625f05548e9f521a4dff7"
+        "identity_sha256": "fbe261ae71e28e8a8111fbc2cc757bbb35eb631bb2305a5276f1bc917d76882f"
       }
     },
     "system_prompt_blocks": 6,
-    "total_bytes": 14951,
-    "total_tokens_est": 3738
+    "total_bytes": 14956,
+    "total_tokens_est": 3739
   },
   "schema_version": 1,
   "skill_discovery": {
@@ -62,22 +62,22 @@
         "mode_instructions_bytes": 802,
         "mode_instructions_tokens_est": 201,
         "system_prompt_blocks": 4,
-        "system_prompt_bytes": 11232,
-        "system_prompt_tokens_est": 2808
+        "system_prompt_bytes": 11237,
+        "system_prompt_tokens_est": 2810
       },
       "operate": {
         "mode_instructions_bytes": 1671,
         "mode_instructions_tokens_est": 418,
         "system_prompt_blocks": 4,
-        "system_prompt_bytes": 12101,
-        "system_prompt_tokens_est": 3026
+        "system_prompt_bytes": 12106,
+        "system_prompt_tokens_est": 3027
       },
       "plan": {
         "mode_instructions_bytes": 517,
         "mode_instructions_tokens_est": 130,
         "system_prompt_blocks": 4,
-        "system_prompt_bytes": 10947,
-        "system_prompt_tokens_est": 2737
+        "system_prompt_bytes": 10952,
+        "system_prompt_tokens_est": 2738
       }
     }
   },