feat(engine): backport run completion and GNU LLVM bindings

crates/tui/src/core/engine.rs

@@ -975,6 +975,7 @@ impl Drop for ReservedSteer {
 /// Handle to communicate with the engine
 #[derive(Clone)]
 pub struct EngineHandle {
+    run_completion: Option<tokio::sync::watch::Receiver<Option<EngineRunCompletion>>>,
     /// Send operations to the engine
     pub tx_op: mpsc::Sender<Op>,
     /// Receive events from the engine
@@ -2410,6 +2411,7 @@ impl Engine {
             advisor_emission_guard: None,
         };
         let handle = EngineHandle {
+            run_completion: None,
             tx_op,
             rx_event: Arc::new(RwLock::new(rx_event)),
             cancel_token: shared_cancel_token,
@@ -8625,15 +8627,12 @@ fn apply_patch_permission_paths(input: &Value) -> Vec<String> {
 
 /// Spawn the engine in a background task
 pub fn spawn_engine(config: EngineConfig, api_config: &Config) -> EngineHandle {
-    let (engine, handle) = Engine::new(config, api_config);
+    let (engine, mut handle) = Engine::new(config, api_config);
 
-    spawn_supervised(
-        "engine-event-loop",
-        std::panic::Location::caller(),
-        async move {
-            engine.run().await;
-        },
-    );
+    // Keep the large run future boxed before entering the supervisor.
+    let (run, completion) = run_completion::observe(Box::pin(engine.run()));
+    handle.run_completion = Some(completion);
+    spawn_supervised("engine-event-loop", std::panic::Location::caller(), run);
 
     handle
 }
@@ -8645,16 +8644,13 @@ pub(crate) fn spawn_engine_with_authoritative_route_config(
     api_config: &Config,
     authoritative_route_config: Arc<parking_lot::RwLock<Config>>,
 ) -> EngineHandle {
-    let (mut engine, handle) = Engine::new(config, api_config);
+    let (mut engine, mut handle) = Engine::new(config, api_config);
     engine.authoritative_route_config = Some(authoritative_route_config);
 
-    spawn_supervised(
-        "engine-event-loop",
-        std::panic::Location::caller(),
-        async move {
-            engine.run().await;
-        },
-    );
+    // Keep the large run future boxed before entering the supervisor.
+    let (run, completion) = run_completion::observe(Box::pin(engine.run()));
+    handle.run_completion = Some(completion);
+    spawn_supervised("engine-event-loop", std::panic::Location::caller(), run);
 
     handle
 }
@@ -8687,6 +8683,18 @@ pub(crate) enum MockApprovalEvent {
 
 #[cfg(test)]
 impl MockEngineHandle {
+    pub(crate) fn observe_run_for_test<F>(
+        &mut self,
+        future: F,
+    ) -> impl Future<Output = ()> + Send + 'static + use<F>
+    where
+        F: Future<Output = ()> + Send + 'static,
+    {
+        let (run, completion) = run_completion::observe(Box::pin(future));
+        self.handle.run_completion = Some(completion);
+        run
+    }
+
     pub(crate) async fn recv_approval_event(&mut self) -> Option<MockApprovalEvent> {
         match self.rx_approval.recv().await? {
             ApprovalDecision::Approved { id } => Some(MockApprovalEvent::Approved { id }),
@@ -8749,6 +8757,7 @@ pub(crate) fn mock_engine_handle() -> MockEngineHandle {
     let mut steer_control_state = SteerControlState::default();
     steer_control_state.begin_turn();
     let handle = EngineHandle {
+        run_completion: None,
         tx_op,
         rx_event: Arc::new(RwLock::new(rx_event)),
         cancel_token: shared_cancel_token,
@@ -9077,6 +9086,8 @@ impl SubAgentWiring {
 mod approval;
 mod context;
 mod handle;
+mod run_completion;
+pub use run_completion::EngineRunCompletion;
 pub mod preview;
 use crate::compaction::estimate_input_tokens_conservative;
 #[cfg(test)]

crates/tui/src/core/engine/README.md

@@ -0,0 +1,24 @@
+# Engine run completion
+
+`EngineHandle::wait_run_completion` observes the particular background run
+created by either engine spawn factory. `NormalReturn` means that run returned
+and its owned future was dropped. `AbnormalTermination` covers panic,
+run cancellation, or loss of the observation sender without a terminal value.
+Handles constructed directly with `Engine::new` return `NotObserved`.
+
+Each waiter receives the same terminal value. Cancelling one wait neither stops
+the engine nor consumes another waiter's result. Waiting does not initiate
+shutdown; callers must use the existing shutdown protocol. The observation
+does not keep another engine or operation-mailbox sender alive.
+
+The one-shot exec consumer uses a two-second absolute deadline shared between
+sending `Shutdown` and waiting for completion after the event loop. This replaces
+the fork's previously unbounded shutdown-mailbox send with a bounded cleanup wait.
+An unconfirmed normal return produces a warning without changing the task's
+result. Existing early error returns and broken-pipe process exits do not pass
+through this cleanup. The deadline bounds cooperative asynchronous waits; it
+cannot preempt a blocking poll or destructor.
+
+Completion does not certify durable persistence, successful work, or termination
+of detached children, blocking writers, event forwarders, or the supervisor's
+later diagnostics.

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

@@ -16,11 +16,30 @@ use tokio::sync::mpsc;
 
 use super::approval::{ApprovalDecision, UserInputDecision};
 use super::{
-    CancelMode, CancelReason, EngineHandle, LiveRuntimeAuthority, Op, ReservedSteer,
-    RuntimePermissionAuthority, SteerWithdrawal, UserInputResponse,
+    CancelMode, CancelReason, EngineHandle, EngineRunCompletion, LiveRuntimeAuthority, Op,
+    ReservedSteer, RuntimePermissionAuthority, SteerWithdrawal, UserInputResponse,
 };
 
 impl EngineHandle {
+    /// Wait for this exact background run to return or terminate abnormally.
+    /// Cancelling a wait does not stop the engine or consume another waiter's result.
+    /// Directly constructed handles return `NotObserved`; completion does not prove
+    /// persistence succeeded or detached children and blocking writers stopped.
+    pub async fn wait_run_completion(&self) -> EngineRunCompletion {
+        let Some(mut completion) = self.run_completion.clone() else {
+            return EngineRunCompletion::NotObserved;
+        };
+        loop {
+            let outcome = *completion.borrow_and_update();
+            if let Some(outcome) = outcome {
+                return outcome;
+            }
+            if completion.changed().await.is_err() {
+                return EngineRunCompletion::AbnormalTermination;
+            }
+        }
+    }
+
     /// True when the caller must preflight a concrete provider client before
     /// committing UI/runtime turn state. Test and embedding handles with an
     /// injected model client return false because that client owns model I/O.

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

@@ -0,0 +1,243 @@
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use tokio::sync::watch;
+
+/// Completion of one observed engine run, not a durable save or child-task drain.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum EngineRunCompletion {
+    NormalReturn,
+    AbnormalTermination,
+    /// The handle was constructed without an observed background run.
+    NotObserved,
+}
+
+pub(super) struct ObservedRun<F> {
+    future: Option<Pin<Box<F>>>,
+    completion: Option<watch::Sender<Option<EngineRunCompletion>>>,
+}
+
+pub(super) fn observe<F: Future<Output = ()>>(
+    future: Pin<Box<F>>,
+) -> (ObservedRun<F>, watch::Receiver<Option<EngineRunCompletion>>) {
+    let (completion, receiver) = watch::channel(None);
+    (
+        ObservedRun {
+            future: Some(future),
+            completion: Some(completion),
+        },
+        receiver,
+    )
+}
+
+impl<F> ObservedRun<F> {
+    fn finish(&mut self, outcome: EngineRunCompletion) {
+        drop(self.future.take());
+        if let Some(completion) = self.completion.take() {
+            let _ = completion.send_replace(Some(outcome));
+        }
+    }
+}
+
+impl<F: Future<Output = ()>> Future for ObservedRun<F> {
+    type Output = ();
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
+        let this = self.get_mut();
+        match this
+            .future
+            .as_mut()
+            .expect("run already completed")
+            .as_mut()
+            .poll(cx)
+        {
+            Poll::Pending => Poll::Pending,
+            Poll::Ready(()) => {
+                this.finish(EngineRunCompletion::NormalReturn);
+                Poll::Ready(())
+            }
+        }
+    }
+}
+
+impl<F> Drop for ObservedRun<F> {
+    fn drop(&mut self) {
+        self.finish(EngineRunCompletion::AbnormalTermination);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::core::engine::mock_engine_handle;
+    use std::sync::{
+        Arc,
+        atomic::{AtomicBool, Ordering},
+    };
+    use tokio::sync::oneshot;
+
+    struct Run {
+        release: oneshot::Receiver<()>,
+        dropped: Arc<AtomicBool>,
+        panic: bool,
+        completion: Option<watch::Receiver<Option<EngineRunCompletion>>>,
+    }
+
+    impl Future for Run {
+        type Output = ();
+        fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
+            assert!(!self.panic, "controlled run panic");
+            Pin::new(&mut self.release).poll(cx).map(|_| ())
+        }
+    }
+
+    impl Drop for Run {
+        fn drop(&mut self) {
+            if let Some(completion) = &self.completion {
+                assert!(
+                    completion.borrow().is_none(),
+                    "completion published before run drop"
+                );
+            }
+            self.dropped.store(true, Ordering::Release);
+        }
+    }
+
+    fn fixture(
+        panic: bool,
+    ) -> (
+        ObservedRun<Run>,
+        crate::core::engine::EngineHandle,
+        oneshot::Sender<()>,
+        Arc<AtomicBool>,
+    ) {
+        let (release, receiver) = oneshot::channel();
+        let dropped = Arc::new(AtomicBool::new(false));
+        let (mut run, completion) = observe(Box::pin(Run {
+            release: receiver,
+            dropped: Arc::clone(&dropped),
+            panic,
+            completion: None,
+        }));
+        run.future.as_mut().unwrap().as_mut().get_mut().completion = Some(completion.clone());
+        let mut handle = mock_engine_handle().handle;
+        handle.run_completion = Some(completion);
+        (run, handle, release, dropped)
+    }
+
+    #[tokio::test]
+    async fn run_completion_drops_future_before_notifying_all_waiters() {
+        let (run, handle, release, dropped) = fixture(false);
+        let task = tokio::spawn(run);
+        let mut cancelled = Box::pin(handle.wait_run_completion());
+        std::future::poll_fn(|cx| {
+            assert!(cancelled.as_mut().poll(cx).is_pending());
+            Poll::Ready(())
+        })
+        .await;
+        drop(cancelled);
+        release.send(()).unwrap();
+        let clone = handle.clone();
+        let (first, second) =
+            tokio::join!(handle.wait_run_completion(), clone.wait_run_completion());
+        assert_eq!(first, EngineRunCompletion::NormalReturn);
+        assert_eq!(second, first);
+        assert!(dropped.load(Ordering::Acquire));
+        task.await.unwrap();
+        assert_eq!(handle.wait_run_completion().await, first);
+    }
+
+    #[test]
+    fn run_completion_supervised_panic_is_abnormal_after_future_drop() {
+        let _env = crate::test_support::lock_test_env();
+        let home = tempfile::tempdir().unwrap();
+        let _home = crate::test_support::EnvVarGuard::set("CODEWHALE_HOME", home.path());
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+        runtime.block_on(async {
+            let (run, handle, _release, dropped) = fixture(true);
+            let task = crate::utils::spawn_supervised(
+                "completion-test",
+                std::panic::Location::caller(),
+                run,
+            );
+            task.await.expect("existing supervisor catches the panic");
+            assert_eq!(
+                handle.wait_run_completion().await,
+                EngineRunCompletion::AbnormalTermination
+            );
+            assert!(dropped.load(Ordering::Acquire));
+        });
+    }
+
+    #[tokio::test]
+    async fn run_completion_unpolled_abort_drops_future_before_notification() {
+        let (run, handle, _release, dropped) = fixture(false);
+        let task = tokio::spawn(run);
+        task.abort();
+        assert!(task.await.unwrap_err().is_cancelled());
+        assert_eq!(
+            handle.wait_run_completion().await,
+            EngineRunCompletion::AbnormalTermination
+        );
+        assert!(dropped.load(Ordering::Acquire));
+    }
+
+    #[tokio::test]
+    async fn run_completion_polled_abort_drops_future_before_notification() {
+        let (mut run, handle, _release, dropped) = fixture(false);
+        std::future::poll_fn(|cx| {
+            assert!(Pin::new(&mut run).poll(cx).is_pending());
+            Poll::Ready(())
+        })
+        .await;
+        let task = tokio::spawn(run);
+        task.abort();
+        assert!(task.await.unwrap_err().is_cancelled());
+        assert_eq!(
+            handle.wait_run_completion().await,
+            EngineRunCompletion::AbnormalTermination
+        );
+        assert!(dropped.load(Ordering::Acquire));
+    }
+
+    #[test]
+    fn run_completion_runtime_drop_is_abnormal() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+        let (run, handle, _release, dropped) = fixture(false);
+        runtime.spawn(run);
+        drop(runtime);
+        assert!(dropped.load(Ordering::Acquire));
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+        assert_eq!(
+            runtime.block_on(handle.wait_run_completion()),
+            EngineRunCompletion::AbnormalTermination
+        );
+    }
+
+    #[tokio::test]
+    async fn run_completion_old_run_cannot_complete_another_run() {
+        let (old, old_handle, release, _) = fixture(false);
+        let (_new, new_handle, _new_release, _) = fixture(false);
+        release.send(()).unwrap();
+        old.await;
+        assert_eq!(
+            old_handle.wait_run_completion().await,
+            EngineRunCompletion::NormalReturn
+        );
+        let mut pending = Box::pin(new_handle.wait_run_completion());
+        std::future::poll_fn(|cx| {
+            assert!(pending.as_mut().poll(cx).is_pending());
+            Poll::Ready(())
+        })
+        .await;
+    }
+}