feat(tui): kill watched harnesses with confirmation

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

@@ -62,6 +62,7 @@ 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(),
+        screen_kill: true,
         // 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

@@ -185,6 +185,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

@@ -63,13 +63,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 +87,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/hub/handle/mod.rs

@@ -130,6 +130,28 @@ 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> {
+        if !self.runner.supports_screen_kill(worker).await {
+            return Err("worker does not advertise harness termination support".to_string());
+        }
+        let correlation_id = self
+            .runner
+            .correlation_for(worker, task_id)
+            .await
+            .ok_or_else(|| format!("task {task_id} is no longer running on {worker}"))?;
+        let body =
+            crate::tinyplace::encode_screen_message(&crate::tinyplace::ScreenMessage::Kill {
+                task_id: task_id.to_string(),
+                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

@@ -60,7 +60,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(

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

@@ -183,6 +183,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,
@@ -210,6 +211,26 @@ impl TaskRunner {
         }
     }
 
+    /// Return the active dispatch receipt for a worker/task pair.
+    pub async fn correlation_for(&self, worker: &str, task_id: &str) -> Option<String> {
+        self.waiters
+            .lock()
+            .await
+            .iter()
+            .find(|(_, waiter)| waiter.from == worker && waiter.task_id == task_id)
+            .map(|(correlation, _)| correlation.clone())
+    }
+
+    /// Return whether the worker advertised screen termination during a
+    /// capability negotiation completed before its current dispatch.
+    pub async fn supports_screen_kill(&self, worker: &str) -> bool {
+        self.capabilities
+            .lock()
+            .await
+            .get(worker)
+            .is_some_and(|capabilities| capabilities.screen_kill)
+    }
+
     /// Cancel every dispatch this runner has in flight.
     ///
     /// For a caller that owns a runner serving one piece of work and wants to
@@ -311,6 +332,7 @@ impl TaskRunner {
             self.waiters.lock().await.insert(
                 cid.clone(),
                 Waiter {
+                    task_id: req.task_id.clone(),
                     from: req.worker_address.clone(),
                     reply: tx,
                     status: status.clone(),

src/sdk/src/hub/runner/types.rs

@@ -3,6 +3,8 @@
 use super::*;
 /// A registered dispatch awaiting its terminal frame.
 pub(super) struct Waiter {
+    /// The task id this dispatch carries on the wire.
+    pub(super) task_id: String,
     /// The worker address this dispatch was sent to — the only sender whose
     /// frames may settle it. See [`Probe::from`].
     pub(super) from: String,
@@ -68,6 +70,8 @@ pub struct TaskRunner {
     pub(super) system_info_waiters: SystemInfoWaiters,
     /// Capability probes waiting for a worker's `capabilities_result`.
     pub(super) capabilities_waiters: CapabilitiesWaiters,
+    /// Last successfully negotiated capabilities for each worker address.
+    pub(super) capabilities: Arc<Mutex<HashMap<String, AgentCapabilities>>>,
     /// Abort signals for in-flight dispatches, keyed by orchestrator-facing task
     /// id; [`abort_task`](Self::abort_task) notifies one to cancel its dispatch.
     pub(super) aborts: Aborts,

src/sdk/src/hub/socket/task_run.rs

@@ -162,6 +162,11 @@ pub(super) async fn handle_task_run(
         return;
     };
 
+    // Negotiate once before dispatch. Besides informing automatic provider
+    // selection, this records static control-plane support so an emergency kill
+    // never waits behind a fresh probe of a wedged worker.
+    let capabilities = runner.capabilities(&worker_address).await.ok();
+
     // An explicit provider is authoritative. Only an untargeted task consults
     // the subscription strategy, and a failed/unknown budget probe falls open
     // to the daemon's own configured default.
@@ -174,12 +179,9 @@ pub(super) async fn handle_task_run(
             if strategy == crate::runtime::SubscriptionRoutingStrategy::Manual {
                 None
             } else {
-                match runner.capabilities(&worker_address).await {
-                    Ok(capabilities) => {
-                        super::super::roster::subscription_for_strategy(&capabilities, strategy)
-                    }
-                    Err(_) => None,
-                }
+                capabilities.as_ref().and_then(|capabilities| {
+                    super::super::roster::subscription_for_strategy(capabilities, strategy)
+                })
             }
         }
     };

src/sdk/src/runtime/mod.rs

@@ -205,6 +205,18 @@ pub trait Runtime: Send + Sync {
         Box::pin(async { Ok(()) })
     }
 
+    /// Kill the harness serving `task_id` on `worker`.
+    ///
+    /// A no-op success without a hub. Interactive callers are responsible for
+    /// confirming the destructive action before invoking this method.
+    fn kill_task(
+        &self,
+        _worker: String,
+        _task_id: String,
+    ) -> BoxFuture<'static, anyhow::Result<()>> {
+        Box::pin(async { Ok(()) })
+    }
+
     /// Tell the orchestrator a harness has been handed back, with the brief.
     ///
     /// An **error** by default rather than a silent success, unlike

src/sdk/src/runtime/openhuman/mod.rs

@@ -493,6 +493,18 @@ impl Runtime for OpenHumanRuntime {
         })
     }
 
+    fn kill_task(&self, worker: String, task_id: String) -> BoxFuture<'static, anyhow::Result<()>> {
+        let hub = self.hub();
+        Box::pin(async move {
+            let Some(hub) = hub else {
+                return Ok(());
+            };
+            hub.kill(&worker, &task_id)
+                .await
+                .map_err(|e| anyhow::anyhow!(e))
+        })
+    }
+
     fn worker_op(&self, op: crate::runtime::WorkerOp) -> BoxFuture<'static, anyhow::Result<()>> {
         let hub = self.hub();
         Box::pin(async move {

src/sdk/src/tinyplace/frames/tests/codec.rs

@@ -350,6 +350,20 @@ fn empty_budgets_and_readiness_are_omitted_on_the_wire() {
     let value = serde_json::to_value(&caps).unwrap();
     assert!(value.get("budgets").is_none());
     assert!(value.get("readiness").is_none());
+    assert!(value.get("screenKill").is_none());
+}
+
+#[test]
+fn screen_kill_support_is_additive_and_defaults_off_for_older_workers() {
+    let older = parse_agent_capabilities(r#"{"providers":["claude"]}"#).unwrap();
+    assert!(!older.screen_kill);
+
+    let current = crate::tinyplace::AgentCapabilities {
+        screen_kill: true,
+        ..Default::default()
+    };
+    let value = serde_json::to_value(&current).unwrap();
+    assert_eq!(value["screenKill"], true);
 }
 
 #[test]

src/sdk/src/tinyplace/frames/types.rs

@@ -479,6 +479,16 @@ pub struct AgentCapabilities {
     /// field. Same backward-compatibility contract as the two vectors above.
     #[serde(default, skip_serializing_if = "Vec::is_empty")]
     pub workflows: Vec<WorkflowAdvert>,
+    /// Whether this worker accepts task-correlated harness termination requests.
+    ///
+    /// Older workers omit the field and therefore deserialize as `false`, so an
+    /// upgraded controller never sends them an unknown screen control message.
+    #[serde(
+        rename = "screenKill",
+        default,
+        skip_serializing_if = "std::ops::Not::not"
+    )]
+    pub screen_kill: bool,
 }
 
 /// Fleet-safe description of one named custom harness.

src/sdk/src/tinyplace/screen/tests.rs

@@ -316,6 +316,10 @@ fn messages_round_trip_through_the_envelope() {
         ScreenMessage::Unsubscribe {
             task_id: "w_1".into(),
         },
+        ScreenMessage::Kill {
+            task_id: "w_1".into(),
+            correlation_id: "cyc/w_1/0".into(),
+        },
         ScreenMessage::Ack {
             task_id: "w_1".into(),
             seq: 418,

src/sdk/src/tinyplace/screen/types.rs

@@ -178,9 +178,9 @@ pub struct ScreenFrame {
 
 /// Everything that can cross this protocol, in both directions.
 ///
-/// There is no `input` and no `resize`: the viewer never reaches back into the
-/// session, and the sender's geometry is authoritative. Both are additive later
-/// if that changes.
+/// There is no `input` and no `resize`: the viewer cannot steer the session,
+/// and the sender's geometry is authoritative. The one control operation is an
+/// explicit, task-scoped kill used by an operator to recover a hung harness.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(tag = "kind", rename_all = "snake_case")]
 pub enum ScreenMessage {
@@ -199,6 +199,14 @@ pub enum ScreenMessage {
         /// The task to stop watching.
         task_id: String,
     },
+    /// Viewer → sender: kill the harness serving an owned running task.
+    Kill {
+        /// The task whose harness should be killed.
+        task_id: String,
+        /// The unique dispatch receipt, preventing a delayed kill from matching
+        /// a later dispatch that reused the task id.
+        correlation_id: String,
+    },
     /// Viewer → sender: the highest sequence the viewer holds, which the next
     /// diff may be taken from.
     Ack {

src/tui/src/event_loop/cmd_dispatch/mod.rs

@@ -180,6 +180,17 @@ pub(super) fn run_cmd(
                 }
             });
         }
+        Cmd::KillTask { worker, task_id } => {
+            let rt = runtime.clone();
+            let tx = msg_tx.clone();
+            tokio::spawn(async move {
+                let status = match rt.kill_task(worker, task_id.clone()).await {
+                    Ok(()) => format!("Kill requested for {task_id}"),
+                    Err(e) => format!("Cannot kill {task_id}: {e}"),
+                };
+                let _ = tx.send(AppMsg::Status(status));
+            });
+        }
         Cmd::StartLocalHost { host, index } => {
             let Some(spawner) = local_hosts.cloned() else {
                 let _ = msg_tx.send(AppMsg::Status(

src/tui/src/ui/app/input.rs

@@ -509,7 +509,7 @@ impl App {
     }
 
     /// The `(worker address, task id)` the current selection asks to watch.
-    fn watch_target(&self) -> Option<(String, String)> {
+    pub(super) fn watch_target(&self) -> Option<(String, String)> {
         // Only on the Agents tab: leaving it releases the subscription.
         if self.tab() != "Agents" {
             return None;

src/tui/src/ui/app/keys/agents.rs

@@ -82,6 +82,15 @@ impl App {
         let alt = k.modifiers.contains(KeyModifiers::ALT);
 
         match k.code {
+            KeyCode::Char('K') => {
+                if let Some(target) = self.watch_target() {
+                    self.kill_armed = Some(target);
+                    self.set_status("Kill this harness? y confirm · any other key cancels");
+                } else {
+                    self.set_status("Select a running harness task first");
+                }
+                AgentsKey::Handled(None)
+            }
             // The bare arrows are the point of having focus at all.
             KeyCode::Up | KeyCode::Down => {
                 self.agent_scroll = 0;

src/tui/src/ui/app/keys/mod.rs

@@ -43,6 +43,17 @@ impl App {
         let shift = k.modifiers.contains(KeyModifiers::SHIFT);
         let alt = k.modifiers.contains(KeyModifiers::ALT);
 
+        // Killing a harness can lose in-progress work. Once armed, the prompt
+        // owns exactly one keypress: only a deliberate `y` proceeds.
+        if let Some((worker, task_id)) = self.kill_armed.take() {
+            if k.code == KeyCode::Char('y') {
+                self.set_status(format!("Killing harness for {task_id}…"));
+                return Some(Cmd::KillTask { worker, task_id });
+            }
+            self.set_status("Harness kill cancelled");
+            return None;
+        }
+
         // The hand-back question outranks even the attached harness, and has to:
         // it is asked *while still attached*, because releasing the keyboard
         // before it is answered would hide the pane the question is about. So