fix(tui): stop awaiting the op channel on the input path (#6150)

The engine drains rx_op only between turns, so an awaited send into a
saturated 32-slot mailbox froze all input for the rest of the turn. The
SendMessage dispatch already reserved capacity off the render thread; this
finishes the audit for the remaining input-path sites:

- try_send for droppable ops whose rejection is reported and retryable:
  sidebar/apply CancelSubAgent, PreviewOutboundRequest, bang shell input,
  PurgeContext, and the single-op settings updates.
- try_reserve before committing state, then send_reserved_op, where the op
  must land if the UI changed: BacktrackConfirm's SyncSession and the
  search-provider update.
- sync_mode_update uses try_send because ChangeMode's live authority is
  published even on a full channel and applied at the next drain.

Left awaited, now documented on EngineHandle::send and at the site: the
ordered must-deliver ops of committed transitions (session/provider
reload, Shutdown+SyncSession+SetCompaction) where a drop would desync
engine and UI. Removing those wedges needs an engine-side mid-turn op
drain or deferred application — a separate slice.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

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

@@ -140,6 +140,15 @@ impl EngineHandle {
     }
 
     /// Send an operation to the engine
+    ///
+    /// This awaits channel capacity, and the engine drains `rx_op` only
+    /// between turns — so on the UI event loop an awaited send into a
+    /// saturated mailbox freezes input for the rest of the turn (#6150).
+    /// Input-path callers instead either `try_send` a droppable op (report
+    /// the rejection) or `try_reserve_owned` before committing UI state and
+    /// hand off with `send_reserved_op`. An awaited `send` remains correct
+    /// only where the operation is part of a committed, ordered transition
+    /// (session/provider reload) whose drop would desync engine and UI.
     pub async fn send(&self, op: Op) -> Result<()> {
         let authority = Self::change_mode_authority(&op);
         let permit = self.tx_op.clone().reserve_owned().await?;

crates/tui/src/tui/ui/apply.rs

@@ -1663,24 +1663,22 @@ pub(crate) async fn apply_command_result(
                 let inputs =
                     build_preview_request_inputs(app, config, engine_handle, hypothetical_prompt)
                         .await;
-                if let Err(err) = engine_handle
-                    .send(Op::PreviewOutboundRequest {
-                        inputs: Box::new(inputs),
-                        json,
-                        base_prompt_only,
-                    })
-                    .await
-                {
+                // #6150: the input path never awaits a full op channel; a
+                // rejected preview is reported and retryable.
+                if let Err(err) = engine_handle.try_send(Op::PreviewOutboundRequest {
+                    inputs: Box::new(inputs),
+                    json,
+                    base_prompt_only,
+                }) {
                     app.status_message = Some(format!("Cannot preview request: {err}"));
                 }
             }
             AppAction::CancelSubAgent { agent_id } => {
                 app.status_message = Some(format!("Cancelling {agent_id}..."));
                 if engine_handle
-                    .send(Op::CancelSubAgent {
+                    .try_send(Op::CancelSubAgent {
                         agent_id: agent_id.clone(),
                     })
-                    .await
                     .is_err()
                 {
                     app.status_message = Some(format!("Could not cancel {agent_id}"));
@@ -1917,9 +1915,14 @@ pub(crate) async fn apply_command_result(
                 }
             }
             AppAction::UpdateStreamChunkTimeout(timeout_secs) => {
-                let _ = engine_handle
-                    .send(Op::SetStreamChunkTimeout { timeout_secs })
-                    .await;
+                // #6150: the input path never awaits a full op channel.
+                if engine_handle
+                    .try_send(Op::SetStreamChunkTimeout { timeout_secs })
+                    .is_err()
+                {
+                    app.status_message =
+                        Some("Engine busy — setting not applied; try again".to_string());
+                }
             }
             AppAction::UpdateSubagentRuntimeConfig {
                 enabled,
@@ -1929,8 +1932,8 @@ pub(crate) async fn apply_command_result(
                 api_timeout_secs,
                 heartbeat_timeout_secs,
             } => {
-                let _ = engine_handle
-                    .send(Op::SetSubagentRuntimeConfig {
+                if engine_handle
+                    .try_send(Op::SetSubagentRuntimeConfig {
                         enabled,
                         max_subagents,
                         launch_concurrency,
@@ -1938,15 +1941,30 @@ pub(crate) async fn apply_command_result(
                         api_timeout_secs,
                         heartbeat_timeout_secs,
                     })
-                    .await;
+                    .is_err()
+                {
+                    app.status_message =
+                        Some("Engine busy — setting not applied; try again".to_string());
+                }
             }
             AppAction::UpdateSearchProvider { provider } => {
-                let effective_provider = config.set_search_provider(provider);
-                let _ = engine_handle
-                    .send(Op::SetSearchProvider {
-                        provider: effective_provider,
-                    })
-                    .await;
+                // Reserve before committing the config change so a full
+                // channel cannot desync the engine from it.
+                match engine_handle.tx_op.clone().try_reserve_owned() {
+                    Ok(permit) => {
+                        let effective_provider = config.set_search_provider(provider);
+                        engine_handle.send_reserved_op(
+                            permit,
+                            Op::SetSearchProvider {
+                                provider: effective_provider,
+                            },
+                        );
+                    }
+                    Err(_) => {
+                        app.status_message =
+                            Some("Engine busy — provider not applied; try again".to_string());
+                    }
+                }
             }
             AppAction::UpdatePromptSuggestion { enabled } => {
                 config.prompt_suggestion = Some(enabled);
@@ -1957,7 +1975,13 @@ pub(crate) async fn apply_command_result(
                 }
             }
             AppAction::SetAdvisorEnabled { enabled } => {
-                let _ = engine_handle.send(Op::SetAdvisorEnabled { enabled }).await;
+                if engine_handle
+                    .try_send(Op::SetAdvisorEnabled { enabled })
+                    .is_err()
+                {
+                    app.status_message =
+                        Some("Engine busy — setting not applied; try again".to_string());
+                }
             }
             AppAction::OpenConfigView => {
                 if app.view_stack.top_kind() != Some(ModalKind::Config) {
@@ -2309,8 +2333,12 @@ pub(crate) async fn apply_command_result(
                 try_queue_manual_compaction(app, config, engine_handle, focus);
             }
             AppAction::PurgeContext => {
-                app.status_message = Some("Agent purging context...".to_string());
-                let _ = engine_handle.send(Op::PurgeContext).await;
+                if engine_handle.try_send(Op::PurgeContext).is_err() {
+                    app.status_message =
+                        Some("Engine busy — purge not sent; try again".to_string());
+                } else {
+                    app.status_message = Some("Agent purging context...".to_string());
+                }
             }
             AppAction::TaskAdd { prompt } => {
                 let owner_session_id = app

crates/tui/src/tui/ui/handlers.rs

@@ -449,17 +449,31 @@ pub(crate) async fn handle_bang_shell_input(
         }
     };
 
-    engine_handle
-        .send(Op::RunShellCommand {
-            command: command.to_string(),
-            mode: app.mode,
-            allow_shell: app.allow_shell,
-            trust_mode: app.trust_mode,
-            auto_approve: app_auto_approve_enabled(app),
-            approval_mode: app.approval_mode,
-        })
-        .await?;
-    app.status_message = Some(format!("Shell command submitted: {command}"));
+    // #6150: composer input never awaits a full op channel — a saturated
+    // engine reports busy instead of freezing the loop.
+    match engine_handle.tx_op.clone().try_reserve_owned() {
+        Ok(permit) => {
+            engine_handle.send_reserved_op(
+                permit,
+                Op::RunShellCommand {
+                    command: command.to_string(),
+                    mode: app.mode,
+                    allow_shell: app.allow_shell,
+                    trust_mode: app.trust_mode,
+                    auto_approve: app_auto_approve_enabled(app),
+                    approval_mode: app.approval_mode,
+                },
+            );
+            app.status_message = Some(format!("Shell command submitted: {command}"));
+        }
+        Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
+            app.status_message =
+                Some("Engine busy — shell command not sent; try again".to_string());
+        }
+        Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
+            return Err(anyhow::anyhow!("engine channel closed"));
+        }
+    }
     Ok(true)
 }
 
@@ -1436,6 +1450,11 @@ pub(crate) async fn handle_view_events(
                             }
                         };
                         sync_runtime_workspace_state(task_manager, app.workspace.clone()).await;
+                        // #6150 audit: these sends may await a full op channel,
+                        // and that await is load-bearing — the session switch
+                        // is already committed UI-side, so each op must land in
+                        // order (drop = engine/UI desync). A wedge is possible
+                        // only while a saturated engine finishes its turn.
                         if respawn {
                             let _ = engine_handle.send(Op::Shutdown).await;
                             *engine_handle =
@@ -2091,11 +2110,12 @@ pub(crate) async fn handle_view_events(
             }
             ViewEvent::SidebarAgentCancel { agent_id } => {
                 app.status_message = Some(format!("Cancelling {agent_id}..."));
+                // #6150: the input path never awaits a full op channel. The
+                // cancel is retryable; a rejected send surfaces immediately.
                 if engine_handle
-                    .send(Op::CancelSubAgent {
+                    .try_send(Op::CancelSubAgent {
                         agent_id: agent_id.clone(),
                     })
-                    .await
                     .is_err()
                 {
                     app.status_message = Some(format!("Could not cancel {agent_id}"));
@@ -2574,18 +2594,39 @@ pub(crate) async fn handle_view_events(
             }
             ViewEvent::BacktrackConfirm => {
                 if let Some(depth) = app.backtrack.confirm() {
-                    apply_backtrack(app, depth);
-                    let _ = engine_handle
-                        .send(Op::SyncSession {
-                            session_id: app.current_session_id.clone(),
-                            messages: app.api_messages.clone(),
-                            system_prompt: app.system_prompt.clone(),
-                            system_prompt_override: false,
-                            model: app.model.clone(),
-                            workspace: app.workspace.clone(),
-                            mode: app.mode,
-                        })
-                        .await;
+                    // Reserve the slot before mutating history (#6150): the
+                    // loop must not await a full op channel, and applying the
+                    // backtrack without delivering SyncSession would desync
+                    // the engine's messages from ours.
+                    match engine_handle.tx_op.clone().try_reserve_owned() {
+                        Ok(permit) => {
+                            apply_backtrack(app, depth);
+                            engine_handle.send_reserved_op(
+                                permit,
+                                Op::SyncSession {
+                                    session_id: app.current_session_id.clone(),
+                                    messages: app.api_messages.clone(),
+                                    system_prompt: app.system_prompt.clone(),
+                                    system_prompt_override: false,
+                                    model: app.model.clone(),
+                                    workspace: app.workspace.clone(),
+                                    mode: app.mode,
+                                },
+                            );
+                        }
+                        Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
+                            app.status_message = Some(
+                                "Engine busy — backtrack not applied; try again in a moment"
+                                    .to_string(),
+                            );
+                            app.needs_redraw = true;
+                        }
+                        Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
+                            app.status_message =
+                                Some("Engine stopped — backtrack not applied".to_string());
+                            app.needs_redraw = true;
+                        }
+                    }
                 }
             }
             ViewEvent::BacktrackCancel => {

crates/tui/src/tui/ui/provider_routes.rs

@@ -576,16 +576,17 @@ pub(crate) fn reasoning_effort_receipt_for_route(
 }
 
 pub(crate) async fn sync_mode_update(app: &App, engine_handle: &EngineHandle) {
-    let _ = engine_handle
-        .send(Op::ChangeMode {
-            mode: app.mode,
-            allow_shell: app.allow_shell,
-            trust_mode: app.trust_mode,
-            auto_approve: app_auto_approve_enabled(app),
-            approval_mode: app.approval_mode,
-            configured_sandbox_mode: app.configured_sandbox_mode.clone(),
-        })
-        .await;
+    // #6150: non-blocking send on the input path. ChangeMode is safe to drop
+    // on a full channel — `try_send` still publishes the live authority
+    // snapshot, which the drain applies before the next queued op.
+    let _ = engine_handle.try_send(Op::ChangeMode {
+        mode: app.mode,
+        allow_shell: app.allow_shell,
+        trust_mode: app.trust_mode,
+        auto_approve: app_auto_approve_enabled(app),
+        approval_mode: app.approval_mode,
+        configured_sandbox_mode: app.configured_sandbox_mode.clone(),
+    });
 }
 
 /// Apply a `/provider` switch by resolving a complete route candidate before