| 1 | //! Public `EngineHandle` methods. |
| 2 | //! |
| 3 | //! The struct itself lives next door in `engine.rs` because two |
| 4 | //! construction sites (`Engine::new` and the test-only |
| 5 | //! `mock_engine_handle`) need access to its private mpsc channels. |
| 6 | //! The method surface — `send`, `cancel*`, `is_cancelled`, |
| 7 | //! `approve_tool_call` / `deny_tool_call` / `retry_tool_with_policy`, |
| 8 | //! `submit_user_input` / `cancel_user_input`, and `steer` — moves here |
| 9 | //! so the agent loop's mailbox API is reviewable on its own. |
| 10 | |
| 11 | use anyhow::Result; |
| 12 | use tokio::sync::mpsc; |
| 13 | |
| 14 | use super::approval::{ApprovalDecision, UserInputDecision}; |
| 15 | use super::{ |
| 16 | CancelReason, EngineHandle, LiveRuntimeAuthority, Op, RuntimePermissionAuthority, |
| 17 | UserInputResponse, |
| 18 | }; |
| 19 | |
| 20 | impl EngineHandle { |
| 21 | /// True when the caller must preflight a concrete provider client before |
| 22 | /// committing UI/runtime turn state. Test and embedding handles with an |
| 23 | /// injected model client return false because that client owns model I/O. |
| 24 | #[must_use] |
| 25 | pub(crate) fn client_preflight_required(&self) -> bool { |
| 26 | self.client_preflight_required |
| 27 | } |
| 28 | |
| 29 | /// Send an operation to the engine |
| 30 | pub async fn send(&self, op: Op) -> Result<()> { |
| 31 | let authority = Self::change_mode_authority(&op); |
| 32 | let permit = self.tx_op.reserve().await?; |
| 33 | if let Some(authority) = authority { |
| 34 | self.publish_runtime_authority(authority); |
| 35 | } |
| 36 | permit.send(op); |
| 37 | Ok(()) |
| 38 | } |
| 39 | |
| 40 | /// Try to send an operation without blocking. |
| 41 | /// |
| 42 | /// Returns `Err` if the channel is full or closed. Use this for |
| 43 | /// non-critical, refresh-type ops (e.g. `Op::ListSubAgents`) that can |
| 44 | /// safely be dropped and re-requested on the next drain cycle. |
| 45 | pub fn try_send(&self, op: Op) -> Result<()> { |
| 46 | let authority = Self::change_mode_authority(&op); |
| 47 | let result = self.tx_op.try_send(op); |
| 48 | // A full channel already guarantees that the engine will wake and |
| 49 | // drain an operation. Publish the typed authority anyway: the drain |
| 50 | // applies pending authority before handling that queued operation, so |
| 51 | // a posture edit never blocks behind refresh traffic. A closed |
| 52 | // channel has no engine left to observe the update. |
| 53 | if !matches!(&result, Err(mpsc::error::TrySendError::Closed(_))) |
| 54 | && let Some(authority) = authority |
| 55 | { |
| 56 | self.publish_runtime_authority(authority); |
| 57 | } |
| 58 | result?; |
| 59 | Ok(()) |
| 60 | } |
| 61 | |
| 62 | fn change_mode_authority(op: &Op) -> Option<LiveRuntimeAuthority> { |
| 63 | let Op::ChangeMode { |
| 64 | mode, |
| 65 | allow_shell, |
| 66 | trust_mode, |
| 67 | auto_approve, |
| 68 | approval_mode, |
| 69 | configured_sandbox_mode, |
| 70 | } = op |
| 71 | else { |
| 72 | return None; |
| 73 | }; |
| 74 | Some(LiveRuntimeAuthority::from_fields( |
| 75 | *mode, |
| 76 | *allow_shell, |
| 77 | *trust_mode, |
| 78 | *auto_approve, |
| 79 | *approval_mode, |
| 80 | configured_sandbox_mode.clone(), |
| 81 | )) |
| 82 | } |
| 83 | |
| 84 | fn publish_runtime_authority(&self, authority: LiveRuntimeAuthority) { |
| 85 | let mut state = self |
| 86 | .live_runtime_authority |
| 87 | .lock() |
| 88 | .unwrap_or_else(std::sync::PoisonError::into_inner); |
| 89 | state.revision = state.revision.wrapping_add(1).max(1); |
| 90 | state.authority = authority; |
| 91 | } |
| 92 | |
| 93 | pub(crate) fn publish_turn_authority( |
| 94 | &self, |
| 95 | mode: crate::tui::app::AppMode, |
| 96 | allow_shell: bool, |
| 97 | trust_mode: bool, |
| 98 | auto_approve: bool, |
| 99 | approval_mode: crate::tui::approval::ApprovalMode, |
| 100 | configured_sandbox_mode: Option<String>, |
| 101 | ) { |
| 102 | self.publish_runtime_authority(LiveRuntimeAuthority::from_fields( |
| 103 | mode, |
| 104 | allow_shell, |
| 105 | trust_mode, |
| 106 | auto_approve, |
| 107 | approval_mode, |
| 108 | configured_sandbox_mode, |
| 109 | )); |
| 110 | } |
| 111 | |
| 112 | /// Exact live permission authority for runtime approval and elevation |
| 113 | /// gates. This is the same typed state the active engine turn drains. |
| 114 | #[must_use] |
| 115 | pub(crate) fn runtime_permission_authority(&self) -> RuntimePermissionAuthority { |
| 116 | self.live_runtime_authority |
| 117 | .lock() |
| 118 | .unwrap_or_else(std::sync::PoisonError::into_inner) |
| 119 | .authority |
| 120 | .permission_snapshot() |
| 121 | } |
| 122 | |
| 123 | /// Reserve capacity for a runtime steer before it mutates durable state. |
| 124 | /// The owned permit lets the caller persist and dispatch synchronously, |
| 125 | /// without a cancellation point between those two operations. |
| 126 | pub(crate) async fn reserve_steer(&self) -> Result<mpsc::OwnedPermit<String>> { |
| 127 | Ok(self.tx_steer.clone().reserve_owned().await?) |
| 128 | } |
| 129 | |
| 130 | /// Cancel the current request (user-initiated path — keeps the |
| 131 | /// public `cancel()` signature stable). Equivalent to |
| 132 | /// `cancel_with_reason(CancelReason::User)`. |
| 133 | pub fn cancel(&self) { |
| 134 | self.cancel_with_reason(CancelReason::User); |
| 135 | } |
| 136 | |
| 137 | /// Cancel the current request and latch the reason so downstream |
| 138 | /// "request cancelled" error messages can name a cause. |
| 139 | pub fn cancel_with_reason(&self, reason: CancelReason) { |
| 140 | match self.cancel_reason.lock() { |
| 141 | Ok(mut slot) => *slot = Some(reason), |
| 142 | Err(poisoned) => *poisoned.into_inner() = Some(reason), |
| 143 | } |
| 144 | match self.cancel_token.lock() { |
| 145 | Ok(token) => token.cancel(), |
| 146 | Err(poisoned) => poisoned.into_inner().cancel(), |
| 147 | } |
| 148 | crate::retry_status::clear(); |
| 149 | } |
| 150 | |
| 151 | /// Check if a request is currently cancelled |
| 152 | #[must_use] |
| 153 | #[allow(dead_code)] |
| 154 | pub fn is_cancelled(&self) -> bool { |
| 155 | match self.cancel_token.lock() { |
| 156 | Ok(token) => token.is_cancelled(), |
| 157 | Err(poisoned) => poisoned.into_inner().is_cancelled(), |
| 158 | } |
| 159 | } |
| 160 | |
| 161 | /// Pause or resume the current pausable command. |
| 162 | pub fn set_paused(&self, paused: bool) { |
| 163 | match self.shared_paused.lock() { |
| 164 | Ok(mut slot) => *slot = paused, |
| 165 | Err(poisoned) => *poisoned.into_inner() = paused, |
| 166 | } |
| 167 | } |
| 168 | |
| 169 | /// Check whether the engine pause gate is set. |
| 170 | #[cfg(test)] |
| 171 | #[must_use] |
| 172 | pub fn is_paused(&self) -> bool { |
| 173 | match self.shared_paused.lock() { |
| 174 | Ok(slot) => *slot, |
| 175 | Err(poisoned) => *poisoned.into_inner(), |
| 176 | } |
| 177 | } |
| 178 | |
| 179 | /// Approve a pending tool call |
| 180 | pub async fn approve_tool_call(&self, id: impl Into<String>) -> Result<()> { |
| 181 | self.tx_approval |
| 182 | .send(ApprovalDecision::Approved { id: id.into() }) |
| 183 | .await?; |
| 184 | Ok(()) |
| 185 | } |
| 186 | |
| 187 | /// Deny a pending tool call |
| 188 | pub async fn deny_tool_call(&self, id: impl Into<String>) -> Result<()> { |
| 189 | self.tx_approval |
| 190 | .send(ApprovalDecision::Denied { id: id.into() }) |
| 191 | .await?; |
| 192 | Ok(()) |
| 193 | } |
| 194 | |
| 195 | /// Retry a tool call with an elevated sandbox policy. |
| 196 | pub async fn retry_tool_with_policy( |
| 197 | &self, |
| 198 | id: impl Into<String>, |
| 199 | policy: crate::sandbox::SandboxPolicy, |
| 200 | ) -> Result<()> { |
| 201 | self.tx_approval |
| 202 | .send(ApprovalDecision::RetryWithPolicy { |
| 203 | id: id.into(), |
| 204 | policy, |
| 205 | }) |
| 206 | .await?; |
| 207 | Ok(()) |
| 208 | } |
| 209 | |
| 210 | /// Submit a response for request_user_input. |
| 211 | pub async fn submit_user_input( |
| 212 | &self, |
| 213 | id: impl Into<String>, |
| 214 | response: UserInputResponse, |
| 215 | ) -> Result<()> { |
| 216 | self.tx_user_input |
| 217 | .send(UserInputDecision::Submitted { |
| 218 | id: id.into(), |
| 219 | response, |
| 220 | }) |
| 221 | .await?; |
| 222 | Ok(()) |
| 223 | } |
| 224 | |
| 225 | /// Cancel a request_user_input prompt. |
| 226 | pub async fn cancel_user_input(&self, id: impl Into<String>) -> Result<()> { |
| 227 | self.tx_user_input |
| 228 | .send(UserInputDecision::Cancelled { id: id.into() }) |
| 229 | .await?; |
| 230 | Ok(()) |
| 231 | } |
| 232 | |
| 233 | /// Steer an in-flight turn with additional user input. |
| 234 | pub async fn steer(&self, content: impl Into<String>) -> Result<()> { |
| 235 | self.tx_steer.send(content.into()).await?; |
| 236 | Ok(()) |
| 237 | } |
| 238 | |
| 239 | /// Request a snapshot of the current session state. |
| 240 | /// Returns the snapshot directly via a oneshot channel, avoiding |
| 241 | /// competition with the SSE event stream on the mpsc receiver. |
| 242 | pub async fn get_session_snapshot(&self) -> Result<crate::core::ops::SessionSnapshot> { |
| 243 | let (tx, rx) = tokio::sync::oneshot::channel(); |
| 244 | let tx = std::sync::Arc::new(std::sync::Mutex::new(Some(tx))); |
| 245 | self.send(Op::GetSessionSnapshot { tx }).await?; |
| 246 | rx.await |
| 247 | .map_err(|_| anyhow::anyhow!("Engine dropped session snapshot oneshot")) |
| 248 | } |
| 249 | |
| 250 | /// Request active provider request concurrency state. |
| 251 | pub async fn get_provider_runtime_status( |
| 252 | &self, |
| 253 | ) -> Result<crate::core::ops::ProviderRuntimeStatus> { |
| 254 | let (tx, rx) = tokio::sync::oneshot::channel(); |
| 255 | let tx = std::sync::Arc::new(std::sync::Mutex::new(Some(tx))); |
| 256 | self.send(Op::GetProviderRuntimeStatus { tx }).await?; |
| 257 | rx.await |
| 258 | .map_err(|_| anyhow::anyhow!("Engine dropped provider runtime status oneshot")) |
| 259 | } |
| 260 | |
| 261 | /// Force the engine-owned MCP pool to reload and reconnect, returning a |
| 262 | /// snapshot from the exact live pool that supplies the next model turn. |
| 263 | pub async fn reload_mcp( |
| 264 | &self, |
| 265 | config_path: std::path::PathBuf, |
| 266 | ) -> Result<crate::mcp::McpManagerSnapshot> { |
| 267 | let (tx, rx) = tokio::sync::oneshot::channel(); |
| 268 | let tx = std::sync::Arc::new(std::sync::Mutex::new(Some(tx))); |
| 269 | self.send(Op::ReloadMcp { config_path, tx }).await?; |
| 270 | rx.await |
| 271 | .map_err(|_| anyhow::anyhow!("Engine dropped MCP reload oneshot"))? |
| 272 | .map_err(anyhow::Error::msg) |
| 273 | } |
| 274 | } |
| 275 |