feat(tui): kill watched harnesses with confirmation

src/sdk/src/control_socket/server/hub_ops.rs

@@ -188,7 +188,22 @@ impl FleetOps for HubFleetOps {
         );
 
         let tee = tee_status(status);
-        let outcome = handle.task_runner().run(request, Some(tee)).await;
+        let runner = handle.task_runner();
+        let (capabilities, abort) = runner
+            .capabilities_for_dispatch(&request.worker_address, &request.abort_id)
+            .await;
+        let screen_kill = match capabilities {
+            Ok(capabilities) => capabilities.screen_kill,
+            Err(RunError::Aborted) => {
+                let outcome = Err(RunError::Aborted);
+                record_outcome(&activity, &wire_task_id, &outcome);
+                return outcome;
+            }
+            Err(_) => false,
+        };
+        let outcome = runner
+            .run_negotiated(request, Some(tee), screen_kill, Some(abort), Some(task_id))
+            .await;
         record_outcome(&activity, &wire_task_id, &outcome);
         outcome
     }

src/sdk/src/daemon/capabilities/mod.rs

@@ -62,6 +62,8 @@ pub async fn probe_capabilities(options: ProbeOptions) -> AgentCapabilities {
         // Filled in by the daemon from its own workflow store: this probe
         // describes the harness, not what has been authored on top of it.
         workflows: Vec::new(),
+        // Enabled by the embedding worker only when it installs a ScreenRouter.
+        screen_kill: false,
         // Deterministic digest of CLAUDE.md/AGENTS.md/README.md — the summary
         // of last resort so a failed probe still carries project context.
         summary: dir.fallback_summary.clone(),

src/sdk/src/daemon/providers/types.rs

@@ -40,6 +40,7 @@ pub type ExistsOnPath = Box<dyn Fn(&str) -> bool + Send + Sync>;
 #[derive(Clone, Default)]
 pub struct Abort {
     flag: Arc<AtomicBool>,
+    terminate: Arc<AtomicBool>,
     notify: Arc<Notify>,
 }
 
@@ -55,6 +56,17 @@ impl Abort {
         self.notify.notify_waiters();
     }
 
+    /// Signal cancellation that must also terminate the serving harness.
+    pub fn terminate(&self) {
+        self.terminate.store(true, Ordering::SeqCst);
+        self.abort();
+    }
+
+    /// Whether cancellation requested termination of the serving harness.
+    pub fn is_terminated(&self) -> bool {
+        self.terminate.load(Ordering::SeqCst)
+    }
+
     /// Whether cancellation has been signalled.
     pub fn is_aborted(&self) -> bool {
         self.flag.load(Ordering::SeqCst)

src/sdk/src/daemon/runtime.rs

@@ -4,7 +4,7 @@
 //! machine lives in [`super::task_loop`].
 
 use std::collections::HashMap;
-use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex as StdMutex};
 
 use tokio::sync::{Mutex as TokioMutex, Notify, Semaphore};
@@ -20,6 +20,13 @@ use super::types::{
 };
 
 impl DaemonRuntime {
+    /// Advertise screen termination support for an embedding that installs the
+    /// authenticated screen-message router.
+    pub fn enable_screen_kill(&self) {
+        self.inner
+            .screen_kill
+            .store(true, std::sync::atomic::Ordering::Relaxed);
+    }
     /// Build a runtime from `config`, an executor (`run_task`), and a
     /// lock-serialized `send`.
     pub fn new(config: DaemonConfig, run_task: RunTaskFn, send: SendFn) -> Self {
@@ -40,6 +47,7 @@ impl DaemonRuntime {
                 inflight_count: AtomicUsize::new(0),
                 inflight_idle: Notify::new(),
                 capabilities: TokioMutex::new(None),
+                screen_kill: AtomicBool::new(false),
                 accessible_dirs: StdMutex::new(accessible_dirs),
                 sessions: crate::sessions::SessionRegistry::default(),
             }),
@@ -185,6 +193,23 @@ impl DaemonRuntime {
             .and_then(|task| task.session_id.clone())
     }
 
+    /// Terminate the running task identified by sender, id, and dispatch receipt.
+    ///
+    /// The signal remains bound to the task record while the map is locked. The
+    /// correlation check prevents a delayed request from terminating a later
+    /// dispatch that reused the same task id.
+    pub fn terminate_task(&self, from: &str, task_id: &str, correlation_id: &str) -> bool {
+        let running = self.inner.running.lock().unwrap();
+        let Some(task) = running.get(&Self::task_key(from, task_id)) else {
+            return false;
+        };
+        if task.correlation_id.as_deref() != Some(correlation_id) {
+            return false;
+        }
+        task.abort.terminate();
+        true
+    }
+
     /// Record the session an executor opened for a running task.
     pub(super) fn record_task_session(&self, key: &str, session_id: String) {
         if let Some(task) = self.inner.running.lock().unwrap().get_mut(key) {

src/sdk/src/daemon/task_loop/probe.rs

@@ -10,7 +10,29 @@ impl DaemonRuntime {
     /// Answer a `capabilities` probe with the cached [`AgentCapabilities`].
     pub(super) async fn handle_capabilities(&self, from: String, frame: TaskFrame) {
         #[cfg_attr(not(feature = "workflows"), allow(unused_mut))]
-        let mut capabilities = self.get_capabilities().await;
+        let mut capabilities = if self
+            .inner
+            .screen_kill
+            .load(std::sync::atomic::Ordering::Relaxed)
+            && self.inner.capabilities.lock().await.is_none()
+        {
+            let runtime = self.clone();
+            tokio::spawn(async move {
+                runtime.get_capabilities().await;
+            });
+            AgentCapabilities {
+                cwd: Some(self.inner.config.workspace.clone()),
+                providers: self.inner.config.providers.clone(),
+                screen_kill: true,
+                ..Default::default()
+            }
+        } else {
+            self.get_capabilities().await
+        };
+        capabilities.screen_kill = self
+            .inner
+            .screen_kill
+            .load(std::sync::atomic::Ordering::Relaxed);
         // Read fresh rather than cached: the harness probe is expensive and
         // worth caching, but an operator who just installed a workflow expects
         // the next probe to advertise it.
@@ -63,13 +85,9 @@ impl DaemonRuntime {
         let abort = Abort::new();
         let controller_id = self.register_controller(abort.clone());
         let accessible_dirs = self.inner.accessible_dirs.lock().unwrap().clone();
-        // Compete for the concurrency budget like a task.
-        let permit = self
-            .inner
-            .slots
-            .acquire()
-            .await
-            .expect("semaphore is never closed");
+        // Control-plane negotiation must not wait behind the harness-task slot
+        // it may be needed to terminate. The probe is separately serialized by
+        // the capability cache lock and bounded by its own timeout.
         let capabilities = probe_capabilities(ProbeOptions {
             provider,
             run_task: self.inner.run_task.clone(),
@@ -91,7 +109,6 @@ impl DaemonRuntime {
             router: self.inner.config.router.clone(),
         })
         .await;
-        drop(permit);
         self.unregister_controller(controller_id);
         *guard = Some(capabilities.clone());
         capabilities

src/sdk/src/daemon/types.rs

@@ -9,7 +9,7 @@
 use std::collections::HashMap;
 use std::future::Future;
 use std::pin::Pin;
-use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex as StdMutex};
 
 use tokio::sync::{mpsc, Mutex as TokioMutex, Notify, Semaphore};
@@ -262,6 +262,8 @@ pub(super) struct Inner {
     pub(super) inflight_idle: Notify,
     /// Cached capability probe result.
     pub(super) capabilities: TokioMutex<Option<AgentCapabilities>>,
+    /// Whether this embedding routes authenticated screen kill messages.
+    pub(super) screen_kill: AtomicBool,
     /// Workspace roots currently approved for capability advertisement.
     ///
     /// Kept outside the immutable config so the daemon TUI can change the

src/sdk/src/hub/handle/mod.rs

@@ -130,6 +130,27 @@ impl HubHandle {
         sent
     }
 
+    /// Ask `worker` to kill the harness serving `task_id`.
+    ///
+    /// The worker resolves the task against the authenticated sender before it
+    /// touches a PTY, so this cannot be used to kill another controller's work.
+    pub async fn kill(&self, worker: &str, task_id: &str) -> Result<(), String> {
+        let (correlation_id, wire_task_id) = self
+            .runner
+            .kill_correlation_for(worker, task_id)
+            .await
+            .ok_or_else(|| {
+                format!("task {task_id} is not running with termination support on {worker}")
+            })?;
+        let body =
+            crate::tinyplace::encode_screen_message(&crate::tinyplace::ScreenMessage::Kill {
+                task_id: wire_task_id,
+                correlation_id,
+            });
+        (self.log)(&format!("hub: killing task {task_id} on {worker}"));
+        self.relay.send(worker, &body).await
+    }
+
     /// Build a handle from its wiring.
     pub(super) fn new(wiring: HandleWiring) -> Self {
         HubHandle {

src/sdk/src/hub/runner/capabilities.rs

@@ -5,7 +5,7 @@ use tokio::sync::oneshot;
 use crate::tinyplace::{encode_task_frame, AgentCapabilities, EncodeFrameInput, TaskFrameKind};
 
 use super::types::CapabilityProbeGuard;
-use super::{RunError, TaskRunner, MAX_RESETS};
+use super::{RunError, TaskRunner, CONTACT_POLL, CONTACT_WAIT, MAX_RESETS};
 
 impl Drop for CapabilityProbeGuard {
     fn drop(&mut self) {
@@ -26,6 +26,9 @@ impl TaskRunner {
     /// (the hub's socket-plane `capabilities_result`) treats any error as "no
     /// budgets to advertise" and falls open to the static facts.
     pub async fn capabilities(&self, address: &str) -> Result<AgentCapabilities, RunError> {
+        // A failed refresh must not leave support advertised by an older worker
+        // that previously occupied this address.
+        self.capabilities.lock().await.remove(address);
         if !self.relay.contact_accepted(address).await {
             let _ = self.relay.request_contact(address).await;
             return Err(RunError::Worker(
@@ -75,7 +78,13 @@ impl TaskRunner {
                 return Err(RunError::Transport(error));
             }
             match tokio::time::timeout(self.ack_window, receiver).await {
-                Ok(Ok(Ok(caps))) => return Ok(caps),
+                Ok(Ok(Ok(caps))) => {
+                    self.capabilities
+                        .lock()
+                        .await
+                        .insert(address.to_string(), caps.clone());
+                    return Ok(caps);
+                }
                 Ok(Ok(Err(error))) => return Err(RunError::Worker(error)),
                 Ok(Err(_)) => {
                     return Err(RunError::Transport(
@@ -93,4 +102,58 @@ impl TaskRunner {
             }
         }
     }
+
+    /// Negotiate capabilities while making a backend abort effective before
+    /// the task itself is registered and sent.
+    pub async fn capabilities_for_dispatch(
+        &self,
+        address: &str,
+        abort_id: &str,
+    ) -> (
+        Result<AgentCapabilities, RunError>,
+        std::sync::Arc<tokio::sync::Notify>,
+    ) {
+        let abort = std::sync::Arc::new(tokio::sync::Notify::new());
+        self.aborts
+            .lock()
+            .expect("aborts lock")
+            .insert(abort_id.to_string(), abort.clone());
+        if !self.relay.contact_accepted(address).await {
+            let _ = self.relay.request_contact(address).await;
+            let deadline = std::time::Instant::now() + CONTACT_WAIT;
+            while std::time::Instant::now() < deadline
+                && !self.relay.contact_accepted(address).await
+            {
+                tokio::select! {
+                    biased;
+                    _ = abort.notified() => {
+                        let mut aborts = self.aborts.lock().expect("aborts lock");
+                        if aborts.get(abort_id).is_some_and(|current| {
+                            std::sync::Arc::ptr_eq(current, &abort)
+                        }) {
+                            aborts.remove(abort_id);
+                        }
+                        return (Err(RunError::Aborted), abort);
+                    },
+                    _ = tokio::time::sleep(CONTACT_POLL) => {}
+                }
+            }
+        }
+        tokio::select! {
+            biased;
+            _ = abort.notified() => {
+                let mut aborts = self.aborts.lock().expect("aborts lock");
+                if aborts
+                    .get(abort_id)
+                    .is_some_and(|current| std::sync::Arc::ptr_eq(current, &abort))
+                {
+                    aborts.remove(abort_id);
+                }
+                (Err(RunError::Aborted), abort)
+            }
+            result = self.capabilities(address) => {
+                (result, abort)
+            },
+        }
+    }
 }

src/sdk/src/hub/runner/mod.rs

@@ -190,6 +190,7 @@ impl TaskRunner {
             waiters,
             system_info_waiters,
             capabilities_waiters,
+            capabilities: Arc::new(Mutex::new(HashMap::new())),
             aborts: Arc::new(std::sync::Mutex::new(HashMap::new())),
             counter: AtomicU64::new(0),
             ack_window,
@@ -221,6 +222,22 @@ impl TaskRunner {
         true
     }
 
+    /// Return the active dispatch receipt for a worker/task pair.
+    pub async fn kill_correlation_for(
+        &self,
+        worker: &str,
+        task_id: &str,
+    ) -> Option<(String, String)> {
+        self.waiters
+            .lock()
+            .await
+            .iter()
+            .find(|(_, waiter)| {
+                waiter.from == worker && waiter.task_id == task_id && waiter.screen_kill
+            })
+            .map(|(correlation, waiter)| (correlation.clone(), waiter.wire_task_id.clone()))
+    }
+
     /// Cancel every dispatch this runner has in flight.
     ///
     /// For a caller that owns a runner serving one piece of work and wants to
@@ -270,6 +287,31 @@ impl TaskRunner {
         &self,
         req: TaskRequest,
         status: Option<mpsc::UnboundedSender<String>>,
+    ) -> Result<TaskOutcome, RunError> {
+        self.run_inner(req, status, false, None, None).await
+    }
+
+    /// Run a dispatch with the screen-control support negotiated specifically
+    /// for this request.
+    pub async fn run_negotiated(
+        &self,
+        req: TaskRequest,
+        status: Option<mpsc::UnboundedSender<String>>,
+        screen_kill: bool,
+        abort: Option<Arc<Notify>>,
+        visible_task_id: Option<String>,
+    ) -> Result<TaskOutcome, RunError> {
+        self.run_inner(req, status, screen_kill, abort, visible_task_id)
+            .await
+    }
+
+    async fn run_inner(
+        &self,
+        req: TaskRequest,
+        status: Option<mpsc::UnboundedSender<String>>,
+        screen_kill: bool,
+        prepared_abort: Option<Arc<Notify>>,
+        visible_task_id: Option<String>,
     ) -> Result<TaskOutcome, RunError> {
         // Register this dispatch's abort signal FIRST — before the contact wait —
         // so a `task_abort` that arrives during contact negotiation (up to
@@ -278,7 +320,7 @@ impl TaskRunner {
         // the backend aborts by, and held for the whole call (spanning any
         // reset+resend retries). The guard removes it on every return path, so a
         // settled dispatch leaves nothing for a later `task_abort` to match.
-        let abort = Arc::new(Notify::new());
+        let abort = prepared_abort.unwrap_or_else(|| Arc::new(Notify::new()));
         self.aborts
             .lock()
             .expect("aborts lock")
@@ -322,6 +364,11 @@ impl TaskRunner {
             self.waiters.lock().await.insert(
                 cid.clone(),
                 Waiter {
+                    task_id: visible_task_id
+                        .clone()
+                        .unwrap_or_else(|| req.task_id.clone()),
+                    wire_task_id: req.task_id.clone(),
+                    screen_kill,
                     from: req.worker_address.clone(),
                     reply: tx,
                     status: status.clone(),