feat(engine): conversation-insertion primitives — keepInbox steer queue + steer events + deterministic cancel

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

@@ -8,15 +8,16 @@
 //! `submit_user_input` / `cancel_user_input`, and `steer` — moves here
 //! so the agent loop's mailbox API is reviewable on its own.
 
+use std::sync::atomic::Ordering;
+
 use anyhow::Result;
 use tokio::sync::mpsc;
 
 use super::approval::{ApprovalDecision, UserInputDecision};
 use super::{
-    CancelReason, EngineHandle, LiveRuntimeAuthority, Op, RuntimePermissionAuthority,
-    UserInputResponse,
+    CancelMode, CancelReason, EngineHandle, LiveRuntimeAuthority, Op, ReservedSteer,
+    RuntimePermissionAuthority, UserInputResponse,
 };
-
 impl EngineHandle {
     /// True when the caller must preflight a concrete provider client before
     /// committing UI/runtime turn state. Test and embedding handles with an
@@ -122,21 +123,59 @@ impl EngineHandle {
 
     /// 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.
-    pub(crate) async fn reserve_steer(&self) -> Result<mpsc::OwnedPermit<String>> {
-        Ok(self.tx_steer.clone().reserve_owned().await?)
+    /// without a cancellation point between those two operations. Its target
+    /// is frozen before the channel wait, so a later session switch or stop
+    /// cannot retarget the send.
+    pub(crate) async fn reserve_steer(&self) -> Result<ReservedSteer> {
+        let id = self.next_steer_id();
+        let target = self
+            .steer_control
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .active_target()
+            .map_err(anyhow::Error::msg)?;
+        let permit = self.tx_steer.clone().reserve_owned().await?;
+        self.steer_control
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .register(id.clone(), target)
+            .map_err(anyhow::Error::msg)?;
+        Ok(ReservedSteer {
+            permit: Some(permit),
+            sent: false,
+            id,
+            target,
+            control: self.steer_control.clone(),
+        })
+    }
+
+    /// Allocate the next opaque steer id. Unique within this engine session;
+    /// shared across handle clones via the counter in `EngineHandle`.
+    fn next_steer_id(&self) -> String {
+        let seq = self.next_steer_id.fetch_add(1, Ordering::Relaxed) + 1;
+        format!("steer-{seq}")
     }
 
-    /// Cancel the current request (user-initiated path — keeps the
-    /// public `cancel()` signature stable). Equivalent to
-    /// `cancel_with_reason(CancelReason::User)`.
+    /// Stop the current request and discard its uncommitted steer inputs.
+    /// Call `cancel_with_mode` explicitly for interrupt/keep-inbox semantics.
     pub fn cancel(&self) {
-        self.cancel_with_reason(CancelReason::User);
+        self.cancel_with_mode(CancelReason::User, CancelMode::StopDropInbox);
     }
 
     /// Cancel the current request and latch the reason so downstream
     /// "request cancelled" error messages can name a cause.
     pub fn cancel_with_reason(&self, reason: CancelReason) {
+        self.cancel_with_mode(reason, CancelMode::StopDropInbox);
+    }
+
+    /// Atomically publish the steer disposition and cancel the active turn.
+    /// A stop barrier is visible before the token fires, so concurrent or
+    /// already-reserved sends cannot escape into a later turn.
+    pub fn cancel_with_mode(&self, reason: CancelReason, mode: CancelMode) {
+        self.steer_control
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .cancel(mode);
         match self.cancel_reason.lock() {
             Ok(mut slot) => *slot = Some(reason),
             Err(poisoned) => *poisoned.into_inner() = Some(reason),
@@ -230,10 +269,31 @@ impl EngineHandle {
         Ok(())
     }
 
+    /// Withdraw a queued steer before the engine injects it.
+    ///
+    /// Fire-and-forget: the id is recorded in a set shared with the engine,
+    /// which checks it at every steer collection and injection point. A
+    /// withdrawn steer is never appended to the transcript; when the engine
+    /// next encounters it, the steer is skipped and reported once via
+    /// `Event::SteerDropped`. Withdrawing an id that was already committed —
+    /// or never existed — is a no-op with no event. The mark survives across
+    /// turns (a parked steer may only surface in a later turn) and is cleared
+    /// on session switch and shutdown.
+    pub fn withdraw_steer(&self, steer_id: &str) {
+        self.steer_control
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .withdraw(steer_id);
+    }
+
     /// Steer an in-flight turn with additional user input.
-    pub async fn steer(&self, content: impl Into<String>) -> Result<()> {
-        self.tx_steer.send(content.into()).await?;
-        Ok(())
+    ///
+    /// Returns the opaque steer id assigned at enqueue time. The engine
+    /// echoes it back in `Event::SteerCommitted` / `Event::SteerDropped`, so
+    /// hosts can correlate those events with the queued input without
+    /// re-hashing content.
+    pub async fn steer(&self, content: impl Into<String>) -> Result<String> {
+        Ok(self.reserve_steer().await?.send(content.into()))
     }
 
     /// Request a snapshot of the current session state.