From e2d04d0a9245634180385f5099b7dfb1e2f1ca53 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 13:11:12 +0300 Subject: [PATCH 01/35] feat(tui): kill watched harnesses with confirmation Co-authored-by: Medulla --- src/sdk/src/hub/handle/mod.rs | 13 +++++++++ src/sdk/src/runtime/backend/runtime.rs | 16 +++++++++++ src/sdk/src/runtime/mod.rs | 12 ++++++++ src/sdk/src/tinyplace/screen/tests.rs | 3 ++ src/sdk/src/tinyplace/screen/types.rs | 11 ++++++-- src/tui/src/event_loop/cmd_dispatch/mod.rs | 11 ++++++++ src/tui/src/ui/app/keys/agents.rs | 9 ++++++ src/tui/src/ui/app/keys/mod.rs | 11 ++++++++ src/tui/src/ui/app/render/agents/composer.rs | 2 +- src/tui/src/ui/app/state.rs | 1 + src/tui/src/ui/app/tests.rs | 29 ++++++++++++++++++++ src/tui/src/ui/app/types.rs | 9 ++++++ src/tui/src/worker/stream/router.rs | 21 ++++++++++++-- src/tui/src/worker/stream/tests.rs | 12 ++++++++ src/tui/tests/e2e_screen_stream.rs | 29 ++++++++++++++++++++ 15 files changed, 182 insertions(+), 7 deletions(-) diff --git a/src/sdk/src/hub/handle/mod.rs b/src/sdk/src/hub/handle/mod.rs index 66b374e1..048c6e6b 100644 --- a/src/sdk/src/hub/handle/mod.rs +++ b/src/sdk/src/hub/handle/mod.rs @@ -111,6 +111,19 @@ 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 body = + crate::tinyplace::encode_screen_message(&crate::tinyplace::ScreenMessage::Kill { + task_id: task_id.to_string(), + }); + (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 { diff --git a/src/sdk/src/runtime/backend/runtime.rs b/src/sdk/src/runtime/backend/runtime.rs index 9ff25224..89b41209 100644 --- a/src/sdk/src/runtime/backend/runtime.rs +++ b/src/sdk/src/runtime/backend/runtime.rs @@ -171,6 +171,22 @@ impl Runtime for BackendRuntime { }) } + fn kill_task( + &self, + worker: String, + task_id: String, + ) -> crate::runtime::BoxFuture<'static, anyhow::Result<()>> { + let handle = self.hub.lock().unwrap().clone(); + Box::pin(async move { + let Some(hub) = handle else { + return Ok(()); + }; + hub.kill(&worker, &task_id) + .await + .map_err(|e| anyhow::anyhow!(e)) + }) + } + fn workers(&self) -> Vec { let handle = self.hub.lock().unwrap().clone(); match handle { diff --git a/src/sdk/src/runtime/mod.rs b/src/sdk/src/runtime/mod.rs index a16e597b..7bd3c835 100644 --- a/src/sdk/src/runtime/mod.rs +++ b/src/sdk/src/runtime/mod.rs @@ -156,6 +156,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(()) }) + } + /// The managed worker-peer registry snapshot (`worker.list`). Empty when the /// runtime has no worker surface. fn workers(&self) -> Vec { diff --git a/src/sdk/src/tinyplace/screen/tests.rs b/src/sdk/src/tinyplace/screen/tests.rs index 36ef5cf5..1f7b9fd8 100644 --- a/src/sdk/src/tinyplace/screen/tests.rs +++ b/src/sdk/src/tinyplace/screen/tests.rs @@ -316,6 +316,9 @@ fn messages_round_trip_through_the_envelope() { ScreenMessage::Unsubscribe { task_id: "w_1".into(), }, + ScreenMessage::Kill { + task_id: "w_1".into(), + }, ScreenMessage::Ack { task_id: "w_1".into(), seq: 418, diff --git a/src/sdk/src/tinyplace/screen/types.rs b/src/sdk/src/tinyplace/screen/types.rs index 204a1187..2a836981 100644 --- a/src/sdk/src/tinyplace/screen/types.rs +++ b/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,11 @@ 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, + }, /// Viewer → sender: the highest sequence the viewer holds, which the next /// diff may be taken from. Ack { diff --git a/src/tui/src/event_loop/cmd_dispatch/mod.rs b/src/tui/src/event_loop/cmd_dispatch/mod.rs index 0b102427..c9b09703 100644 --- a/src/tui/src/event_loop/cmd_dispatch/mod.rs +++ b/src/tui/src/event_loop/cmd_dispatch/mod.rs @@ -196,6 +196,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::WorkerOp(op) => { let rt = runtime.clone(); let tx = msg_tx.clone(); diff --git a/src/tui/src/ui/app/keys/agents.rs b/src/tui/src/ui/app/keys/agents.rs index 2712cb75..f69afe5b 100644 --- a/src/tui/src/ui/app/keys/agents.rs +++ b/src/tui/src/ui/app/keys/agents.rs @@ -58,6 +58,15 @@ impl App { let alt = k.modifiers.contains(KeyModifiers::ALT); match k.code { + KeyCode::Char('K') => { + if let Some(target) = self.watching.clone() { + 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; diff --git a/src/tui/src/ui/app/keys/mod.rs b/src/tui/src/ui/app/keys/mod.rs index e9fa1c80..dfe5c0a6 100644 --- a/src/tui/src/ui/app/keys/mod.rs +++ b/src/tui/src/ui/app/keys/mod.rs @@ -46,6 +46,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; + } + // Resume picker owns navigation while open. if self.resume_picker.is_some() { match k.code { diff --git a/src/tui/src/ui/app/render/agents/composer.rs b/src/tui/src/ui/app/render/agents/composer.rs index bef8cd2c..6a98ec36 100644 --- a/src/tui/src/ui/app/render/agents/composer.rs +++ b/src/tui/src/ui/app/render/agents/composer.rs @@ -69,7 +69,7 @@ impl App { // that cannot send it leaves the rail unreachable, so the key that // always works is named right where the cursor is. let caption = if self.agents_rail_focused() { - "↑↓ walk agents · Enter or type to write".to_string() + "↑↓ walk agents · K kill harness · Enter or type to write".to_string() } else { format!("› {target} · Esc to pick an agent") }; diff --git a/src/tui/src/ui/app/state.rs b/src/tui/src/ui/app/state.rs index 9e182a57..5a2696f5 100644 --- a/src/tui/src/ui/app/state.rs +++ b/src/tui/src/ui/app/state.rs @@ -62,6 +62,7 @@ impl App { context_index: 0, agent_index: 0, watching: None, + kill_armed: None, agents_focus: super::types::AgentsFocus::default(), agent_scroll: 0, chat_scroll: 0, diff --git a/src/tui/src/ui/app/tests.rs b/src/tui/src/ui/app/tests.rs index 9d51cf38..de72cdbb 100644 --- a/src/tui/src/ui/app/tests.rs +++ b/src/tui/src/ui/app/tests.rs @@ -361,6 +361,35 @@ fn selecting_a_task_asks_to_watch_it() { assert!(app.watching.is_some(), "the target is remembered"); } +#[test] +fn killing_a_watched_harness_requires_confirmation() { + let mut app = app(); + select_first_task(&mut app).expect("the fixture has a selectable task"); + app.focus_agents_rail(); + + let armed = app.on_key(KeyEvent::new(KeyCode::Char('K'), KeyModifiers::SHIFT)); + assert!(armed.is_none(), "arming must not kill the harness"); + assert!(app.kill_armed.is_some()); + assert!(app.status().contains("y confirm")); + + let cmd = app.on_key(KeyEvent::new(KeyCode::Char('y'), KeyModifiers::NONE)); + assert!(matches!(cmd, Some(Cmd::KillTask { .. }))); + assert!(app.kill_armed.is_none()); +} + +#[test] +fn any_other_key_cancels_a_harness_kill() { + let mut app = app(); + select_first_task(&mut app).expect("the fixture has a selectable task"); + app.focus_agents_rail(); + app.on_key(KeyEvent::new(KeyCode::Char('K'), KeyModifiers::SHIFT)); + + let cmd = app.on_key(KeyEvent::new(KeyCode::Char('n'), KeyModifiers::NONE)); + assert!(cmd.is_none()); + assert!(app.kill_armed.is_none()); + assert!(app.status().contains("cancelled")); +} + #[test] fn reselecting_the_same_task_does_not_resubscribe() { // Every subscribe carries `resync: true`, so re-issuing one on each diff --git a/src/tui/src/ui/app/types.rs b/src/tui/src/ui/app/types.rs index c02c23ec..d2f1a481 100644 --- a/src/tui/src/ui/app/types.rs +++ b/src/tui/src/ui/app/types.rs @@ -360,6 +360,13 @@ pub enum Cmd { /// The `(worker address, task id)` to start streaming, if any. start: Option<(String, String)>, }, + /// Kill the harness serving a watched task after UI confirmation. + KillTask { + /// The worker address that owns the harness. + worker: String, + /// The dispatched task whose harness should be killed. + task_id: String, + }, /// Load the persona-memory status + directives for the Memory tab. LoadMemory, /// Fetch account-level usage from the backend for the Usage tab. @@ -614,6 +621,8 @@ pub struct App { /// new one: a subscription nobody is looking at costs the worker a sample, /// a ratchet advance and a send on every tick. pub(super) watching: Option<(String, String)>, + /// The watched `(worker, task)` awaiting destructive-action confirmation. + pub(super) kill_armed: Option<(String, String)>, /// Which half of the Agents tab the keyboard is driving. pub(super) agents_focus: AgentsFocus, pub(super) agent_scroll: usize, diff --git a/src/tui/src/worker/stream/router.rs b/src/tui/src/worker/stream/router.rs index 138b01d4..84e6ad03 100644 --- a/src/tui/src/worker/stream/router.rs +++ b/src/tui/src/worker/stream/router.rs @@ -2,9 +2,10 @@ //! subscriber may watch, and starting or stopping the stream that answers. //! //! This is the half of the protocol the worker *receives*. The vocabulary is -//! deliberately tiny — subscribe, unsubscribe, acknowledge — because the viewer -//! is a passive observer: there is no message here that can type into a session -//! or resize one, so the worst a subscriber can do is read a screen. +//! deliberately tiny — subscribe, unsubscribe, acknowledge, and an explicit kill. +//! There is no message that can type into or resize a session. Kill is resolved +//! through the same task ownership check as subscribe, so a controller can stop +//! only a harness running work that controller dispatched. //! //! Which leaves one question, and the answer is structural rather than a check. //! A subscription names a **task**, and the daemon's running-task record is @@ -93,6 +94,20 @@ impl ScreenRouter { self.log(&format!("screen: {from} unsubscribed from {task_id}")); } } + ScreenMessage::Kill { task_id } => { + let Some(session_id) = self.runtime.session_for_task(from, &task_id) else { + self.log(&format!( + "screen: refused kill from {from} on {task_id} — no such running task for this sender" + )); + return; + }; + if self.sessions.close(&session_id) { + self.registry.unsubscribe(&task_id); + self.log(&format!( + "screen: {from} killed {task_id} (session {session_id})" + )); + } + } // The sampler chains from the last frame it actually sent, so an // acknowledgement is not needed to decide what to send next. Kept in // the protocol as the viewer's liveness signal and accepted here so diff --git a/src/tui/src/worker/stream/tests.rs b/src/tui/src/worker/stream/tests.rs index 1d776181..e6b6801a 100644 --- a/src/tui/src/worker/stream/tests.rs +++ b/src/tui/src/worker/stream/tests.rs @@ -396,6 +396,18 @@ async fn an_unsubscribe_for_an_unresolvable_task_does_nothing() { assert_eq!(router.active(), 0); } +#[tokio::test] +async fn a_kill_for_a_task_this_sender_never_dispatched_is_refused() { + let mut router = empty_router(); + router.handle( + "peerA", + medulla::tinyplace::ScreenMessage::Kill { + task_id: "t1".into(), + }, + ); + assert_eq!(router.active(), 0); +} + #[tokio::test] async fn the_router_ignores_messages_it_is_not_the_receiver_for() { // An ack is accepted and does nothing; a frame arriving at the sender is a diff --git a/src/tui/tests/e2e_screen_stream.rs b/src/tui/tests/e2e_screen_stream.rs index c8f13939..5113e1f1 100644 --- a/src/tui/tests/e2e_screen_stream.rs +++ b/src/tui/tests/e2e_screen_stream.rs @@ -303,3 +303,32 @@ async fn an_unchanged_screen_costs_nothing_after_the_first_frame() { sessions.shutdown(); } + +#[tokio::test(flavor = "multi_thread")] +async fn an_owned_task_kill_stops_its_real_harness() { + let peer = "peerA"; + let task_id = "t3#0"; + let sessions = PtyManager::new(); + let session_id = sessions.open(sh("sleep 30", peer)).expect("a pty session"); + let runtime = runtime_serving(session_id.clone()); + start_task(&runtime, peer, task_id).await; + let mut router = ScreenRouter::new(sessions.clone(), runtime, send_fn(|_, _| async {})); + + router.handle( + peer, + ScreenMessage::Kill { + task_id: task_id.to_string(), + }, + ); + + assert!( + !sessions + .row(&session_id) + .expect("the killed session remains inspectable") + .state + .is_running(), + "the task-scoped kill must stop the harness process" + ); + assert_eq!(router.active(), 0, "its screen stream is also stopped"); + sessions.shutdown(); +} From d193928e8c59a09b814a4bc116522c80c442bb1f Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 14:23:23 +0300 Subject: [PATCH 02/35] fix(tui): resolve kill target from current selection The kill command now resolves the target from the current rail selection instead of relying on a cached watch target, preventing stale or mismatched kills. The daemon runtime gains a dedicated abort method that safely cancels a task while holding the lock, and the router uses it directly rather than closing sessions indirectly. --- src/sdk/src/daemon/runtime.rs | 14 ++++++++++++++ src/tui/src/ui/app/input.rs | 2 +- src/tui/src/ui/app/keys/agents.rs | 2 +- src/tui/src/ui/app/tests.rs | 18 ++++++++++++++++++ src/tui/src/worker/stream/router.rs | 10 +++------- 5 files changed, 37 insertions(+), 9 deletions(-) diff --git a/src/sdk/src/daemon/runtime.rs b/src/sdk/src/daemon/runtime.rs index 0b4f09d9..2fabadd4 100644 --- a/src/sdk/src/daemon/runtime.rs +++ b/src/sdk/src/daemon/runtime.rs @@ -185,6 +185,20 @@ impl DaemonRuntime { .and_then(|task| task.session_id.clone()) } + /// Abort the running task identified by its authenticated sender and id. + /// + /// The abort remains bound to the task record while the map is locked. This + /// avoids resolving a reusable session id and acting on it after a later + /// task has claimed the same harness. + pub fn abort_task(&self, from: &str, task_id: &str) -> bool { + let running = self.inner.running.lock().unwrap(); + let Some(task) = running.get(&Self::task_key(from, task_id)) else { + return false; + }; + task.abort.abort(); + 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) { diff --git a/src/tui/src/ui/app/input.rs b/src/tui/src/ui/app/input.rs index 432e8fb2..3e5b12fe 100644 --- a/src/tui/src/ui/app/input.rs +++ b/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; diff --git a/src/tui/src/ui/app/keys/agents.rs b/src/tui/src/ui/app/keys/agents.rs index 16db44f9..952eb267 100644 --- a/src/tui/src/ui/app/keys/agents.rs +++ b/src/tui/src/ui/app/keys/agents.rs @@ -83,7 +83,7 @@ impl App { match k.code { KeyCode::Char('K') => { - if let Some(target) = self.watching.clone() { + 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 { diff --git a/src/tui/src/ui/app/tests.rs b/src/tui/src/ui/app/tests.rs index c84fccc0..78ea951a 100644 --- a/src/tui/src/ui/app/tests.rs +++ b/src/tui/src/ui/app/tests.rs @@ -308,6 +308,24 @@ fn killing_a_watched_harness_requires_confirmation() { assert!(app.kill_armed.is_none()); } +#[test] +fn killing_resolves_the_current_rail_selection_instead_of_the_cached_watch() { + let mut app = app(); + select_first_task(&mut app).expect("the fixture has a selectable task"); + let selected = app.watch_target().expect("the selected task is watchable"); + app.watching = Some(("stale-worker".into(), "stale-task".into())); + app.focus_agents_rail(); + + app.on_key(KeyEvent::new(KeyCode::Char('K'), KeyModifiers::SHIFT)); + assert_eq!(app.kill_armed.as_ref(), Some(&selected)); + + let cmd = app.on_key(KeyEvent::new(KeyCode::Char('y'), KeyModifiers::NONE)); + let Some(Cmd::KillTask { worker, task_id }) = cmd else { + panic!("confirming should kill the selected task"); + }; + assert_eq!((worker, task_id), selected); +} + #[test] fn any_other_key_cancels_a_harness_kill() { let mut app = app(); diff --git a/src/tui/src/worker/stream/router.rs b/src/tui/src/worker/stream/router.rs index 30f3a6b5..dfbbb79e 100644 --- a/src/tui/src/worker/stream/router.rs +++ b/src/tui/src/worker/stream/router.rs @@ -97,18 +97,14 @@ impl ScreenRouter { } } ScreenMessage::Kill { task_id } => { - let Some(session_id) = self.runtime.session_for_task(from, &task_id) else { + if !self.runtime.abort_task(from, &task_id) { self.log(&format!( "screen: refused kill from {from} on {task_id} — no such running task for this sender" )); return; - }; - if self.sessions.close(&session_id) { - self.registry.unsubscribe(&task_id); - self.log(&format!( - "screen: {from} killed {task_id} (session {session_id})" - )); } + self.registry.unsubscribe(&task_id); + self.log(&format!("screen: {from} killed {task_id}")); } // The sampler chains from the last frame it actually sent, so an // acknowledgement is not needed to decide what to send next. Kept in From 52fc942d208042abfe5d4ee71f6e6d21be47cded Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 14:39:39 +0300 Subject: [PATCH 03/35] feat(daemon): distinguish task termination from interrupt The abort signal now carries a termination flag that, when set, stops the serving harness and closes its session instead of merely sending an interrupt. This lets a kill request fully tear down a task while preserving the existing interrupt behaviour for ordinary aborts. --- src/sdk/src/daemon/providers/types.rs | 12 ++++++++++++ src/sdk/src/daemon/runtime.rs | 4 ++-- src/tui/src/worker/executor/run.rs | 11 ++++++++--- src/tui/src/worker/stream/router.rs | 2 +- src/tui/tests/e2e_screen_stream.rs | 10 +++++++--- 5 files changed, 30 insertions(+), 9 deletions(-) diff --git a/src/sdk/src/daemon/providers/types.rs b/src/sdk/src/daemon/providers/types.rs index a2899c77..bb1f2bb7 100644 --- a/src/sdk/src/daemon/providers/types.rs +++ b/src/sdk/src/daemon/providers/types.rs @@ -40,6 +40,7 @@ pub type ExistsOnPath = Box bool + Send + Sync>; #[derive(Clone, Default)] pub struct Abort { flag: Arc, + terminate: Arc, notify: Arc, } @@ -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) diff --git a/src/sdk/src/daemon/runtime.rs b/src/sdk/src/daemon/runtime.rs index 2fabadd4..cfe7185a 100644 --- a/src/sdk/src/daemon/runtime.rs +++ b/src/sdk/src/daemon/runtime.rs @@ -190,12 +190,12 @@ impl DaemonRuntime { /// The abort remains bound to the task record while the map is locked. This /// avoids resolving a reusable session id and acting on it after a later /// task has claimed the same harness. - pub fn abort_task(&self, from: &str, task_id: &str) -> bool { + pub fn terminate_task(&self, from: &str, task_id: &str) -> bool { let running = self.inner.running.lock().unwrap(); let Some(task) = running.get(&Self::task_key(from, task_id)) else { return false; }; - task.abort.abort(); + task.abort.terminate(); true } diff --git a/src/tui/src/worker/executor/run.rs b/src/tui/src/worker/executor/run.rs index 8f836207..18ab62ea 100644 --- a/src/tui/src/worker/executor/run.rs +++ b/src/tui/src/worker/executor/run.rs @@ -481,9 +481,14 @@ impl PtySessionExecutor { )); } if abort.is_aborted() { - // A real interrupt, not a kill: Ctrl-C reaches the harness the - // same way the operator's would, and the session survives it. - let _ = self.sessions.write(id, &[0x03]); + if abort.is_terminated() { + self.stop_turn(id); + } else { + // A requester abort is an interrupt: Ctrl-C reaches the + // harness the same way the operator's would, and the + // reusable session survives it. + let _ = self.sessions.write(id, &[0x03]); + } return Err(format!("{} task aborted", provider.as_str())); } if !self diff --git a/src/tui/src/worker/stream/router.rs b/src/tui/src/worker/stream/router.rs index dfbbb79e..77c7c889 100644 --- a/src/tui/src/worker/stream/router.rs +++ b/src/tui/src/worker/stream/router.rs @@ -97,7 +97,7 @@ impl ScreenRouter { } } ScreenMessage::Kill { task_id } => { - if !self.runtime.abort_task(from, &task_id) { + if !self.runtime.terminate_task(from, &task_id) { self.log(&format!( "screen: refused kill from {from} on {task_id} — no such running task for this sender" )); diff --git a/src/tui/tests/e2e_screen_stream.rs b/src/tui/tests/e2e_screen_stream.rs index b941fbf2..f03d4a3e 100644 --- a/src/tui/tests/e2e_screen_stream.rs +++ b/src/tui/tests/e2e_screen_stream.rs @@ -69,7 +69,7 @@ fn sh(script: &str, label: &str) -> LaunchSpec { /// The runtime is the piece under test here as much as the sampler: it owns the /// running-task record keyed by `(sender, task id)`, and that record is the only /// thing that lets a subscription resolve. -fn runtime_serving(session_id: String) -> DaemonRuntime { +fn runtime_serving(sessions: PtyManager, session_id: String) -> DaemonRuntime { let config = DaemonConfig { providers: vec![HarnessProvider::Codex], default_provider: HarnessProvider::Codex, @@ -92,13 +92,17 @@ fn runtime_serving(session_id: String) -> DaemonRuntime { }; let run_task = Arc::new(move |options: medulla::daemon::providers::RunTaskOptions| { let session_id = session_id.clone(); + let sessions = sessions.clone(); Box::pin(async move { if let Some(report) = options.on_session { report(session_id); } // Hold the task open so its record — and therefore the subscription // resolving through it — stays live for the duration of the test. - tokio::time::sleep(Duration::from_secs(60)).await; + options.abort.cancelled().await; + if options.abort.is_terminated() { + sessions.close(&session_id); + } Err("never settles".to_string()) }) as std::pin::Pin + Send>> }) as medulla::daemon::providers::RunTaskFn; @@ -155,7 +159,7 @@ async fn a_watched_task_streams_its_real_terminal_to_the_hubs_store() { )) .expect("a pty session"); - let runtime = runtime_serving(session_id.clone()); + let runtime = runtime_serving(sessions.clone(), session_id.clone()); start_task(&runtime, peer, task_id).await; let outbox: Outbox = Arc::new(Mutex::new(Vec::new())); From af9fe0cb2ac4ae6daabf3472a2b248d96f3de5b6 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 14:39:55 +0300 Subject: [PATCH 04/35] test(e2e_screen_stream): wait for killed session to stop The test for task-scoped kill now polls the session state until the harness process actually stops, instead of asserting immediately after the kill. This makes the test robust against timing differences in process termination. --- src/tui/tests/e2e_screen_stream.rs | 25 +++++++++++++++---------- 1 file changed, 15 insertions(+), 10 deletions(-) diff --git a/src/tui/tests/e2e_screen_stream.rs b/src/tui/tests/e2e_screen_stream.rs index f03d4a3e..05693400 100644 --- a/src/tui/tests/e2e_screen_stream.rs +++ b/src/tui/tests/e2e_screen_stream.rs @@ -274,7 +274,7 @@ async fn an_unchanged_screen_costs_nothing_after_the_first_frame() { let session_id = sessions .open(sh("printf 'STILL\\n'; sleep 30", peer)) .expect("a pty session"); - let runtime = runtime_serving(session_id); + let runtime = runtime_serving(sessions.clone(), session_id); start_task(&runtime, peer, task_id).await; let outbox: Outbox = Arc::new(Mutex::new(Vec::new())); @@ -322,7 +322,7 @@ async fn an_owned_task_kill_stops_its_real_harness() { let task_id = "t3#0"; let sessions = PtyManager::new(); let session_id = sessions.open(sh("sleep 30", peer)).expect("a pty session"); - let runtime = runtime_serving(session_id.clone()); + let runtime = runtime_serving(sessions.clone(), session_id.clone()); start_task(&runtime, peer, task_id).await; let mut router = ScreenRouter::new(sessions.clone(), runtime, send_fn(|_, _| async {})); @@ -333,14 +333,19 @@ async fn an_owned_task_kill_stops_its_real_harness() { }, ); - assert!( - !sessions - .row(&session_id) - .expect("the killed session remains inspectable") - .state - .is_running(), - "the task-scoped kill must stop the harness process" - ); + let deadline = Instant::now() + PATIENCE; + while sessions + .row(&session_id) + .expect("the killed session remains inspectable") + .state + .is_running() + { + assert!( + Instant::now() < deadline, + "the task-scoped kill must stop the harness process" + ); + tokio::time::sleep(Duration::from_millis(20)).await; + } assert_eq!(router.active(), 0, "its screen stream is also stopped"); sessions.shutdown(); } From d242b6fae91c8c607690acf7149cb0f525984073 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 14:40:46 +0300 Subject: [PATCH 05/35] fix(tests): clone session id before reporting The session id was moved into the report callback, preventing its use later in the task. Cloning it first ensures the id remains available for the rest of the test flow. --- src/tui/tests/e2e_screen_stream.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/tui/tests/e2e_screen_stream.rs b/src/tui/tests/e2e_screen_stream.rs index 05693400..8b7e18d5 100644 --- a/src/tui/tests/e2e_screen_stream.rs +++ b/src/tui/tests/e2e_screen_stream.rs @@ -95,7 +95,7 @@ fn runtime_serving(sessions: PtyManager, session_id: String) -> DaemonRuntime { let sessions = sessions.clone(); Box::pin(async move { if let Some(report) = options.on_session { - report(session_id); + report(session_id.clone()); } // Hold the task open so its record — and therefore the subscription // resolving through it — stays live for the duration of the test. From f3dd305d9e5368835fb1e845399a4d5cd308bdbf Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 15:09:44 +0300 Subject: [PATCH 06/35] feat(sdk): advertise screen kill capability and correlate kills The runner now records the task id on each dispatch waiter, and the screen kill message carries a correlation id so a delayed kill cannot match a later dispatch that reused the same task id. The daemon probe advertises the new screen_kill capability, which older workers omit and therefore deserialize as false, preventing an upgraded controller from sending them an unknown screen control message. --- src/sdk/src/daemon/capabilities/mod.rs | 1 + src/sdk/src/hub/runner/mod.rs | 1 + src/sdk/src/hub/runner/types.rs | 2 ++ src/sdk/src/tinyplace/frames/types.rs | 6 ++++++ src/sdk/src/tinyplace/screen/types.rs | 3 +++ 5 files changed, 13 insertions(+) diff --git a/src/sdk/src/daemon/capabilities/mod.rs b/src/sdk/src/daemon/capabilities/mod.rs index 6ed00138..b46cd9d5 100644 --- a/src/sdk/src/daemon/capabilities/mod.rs +++ b/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(), diff --git a/src/sdk/src/hub/runner/mod.rs b/src/sdk/src/hub/runner/mod.rs index 3bfbed81..0b192ba2 100644 --- a/src/sdk/src/hub/runner/mod.rs +++ b/src/sdk/src/hub/runner/mod.rs @@ -311,6 +311,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(), diff --git a/src/sdk/src/hub/runner/types.rs b/src/sdk/src/hub/runner/types.rs index 584e0782..9918eea8 100644 --- a/src/sdk/src/hub/runner/types.rs +++ b/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, diff --git a/src/sdk/src/tinyplace/frames/types.rs b/src/sdk/src/tinyplace/frames/types.rs index 4ba0c38a..8b7c8465 100644 --- a/src/sdk/src/tinyplace/frames/types.rs +++ b/src/sdk/src/tinyplace/frames/types.rs @@ -479,6 +479,12 @@ pub struct AgentCapabilities { /// field. Same backward-compatibility contract as the two vectors above. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub workflows: Vec, + /// 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. diff --git a/src/sdk/src/tinyplace/screen/types.rs b/src/sdk/src/tinyplace/screen/types.rs index 2a836981..1aa9f1ad 100644 --- a/src/sdk/src/tinyplace/screen/types.rs +++ b/src/sdk/src/tinyplace/screen/types.rs @@ -203,6 +203,9 @@ pub enum ScreenMessage { 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. From e1bc68722170970433b652c5f5913113c0e26a0f Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 15:10:01 +0300 Subject: [PATCH 07/35] fix(hub): require correlation id for task kills The kill path now checks that the worker advertises harness termination support and resolves the active dispatch correlation id before sending the kill message. The daemon verifies this correlation id when terminating, preventing stale or mismatched kill requests from aborting a task that has since been reused by a different dispatch. --- src/sdk/src/daemon/runtime.rs | 5 ++++- src/sdk/src/hub/handle/mod.rs | 14 ++++++++++++++ src/sdk/src/hub/runner/mod.rs | 10 ++++++++++ src/tui/src/worker/stream/router.rs | 10 ++++++++-- 4 files changed, 36 insertions(+), 3 deletions(-) diff --git a/src/sdk/src/daemon/runtime.rs b/src/sdk/src/daemon/runtime.rs index cfe7185a..c8419376 100644 --- a/src/sdk/src/daemon/runtime.rs +++ b/src/sdk/src/daemon/runtime.rs @@ -190,11 +190,14 @@ impl DaemonRuntime { /// The abort remains bound to the task record while the map is locked. This /// avoids resolving a reusable session id and acting on it after a later /// task has claimed the same harness. - pub fn terminate_task(&self, from: &str, task_id: &str) -> bool { + 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 } diff --git a/src/sdk/src/hub/handle/mod.rs b/src/sdk/src/hub/handle/mod.rs index 87c00b96..c95991c9 100644 --- a/src/sdk/src/hub/handle/mod.rs +++ b/src/sdk/src/hub/handle/mod.rs @@ -135,9 +135,23 @@ impl HubHandle { /// 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 capabilities = self + .runner + .capabilities(worker) + .await + .map_err(|error| error.to_string())?; + if !capabilities.screen_kill { + 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 diff --git a/src/sdk/src/hub/runner/mod.rs b/src/sdk/src/hub/runner/mod.rs index 0b192ba2..aff0ede4 100644 --- a/src/sdk/src/hub/runner/mod.rs +++ b/src/sdk/src/hub/runner/mod.rs @@ -210,6 +210,16 @@ impl TaskRunner { } } + /// Return the active dispatch receipt for a worker/task pair. + pub async fn correlation_for(&self, worker: &str, task_id: &str) -> Option { + self.waiters + .lock() + .await + .iter() + .find(|(_, waiter)| waiter.from == worker && waiter.task_id == task_id) + .map(|(correlation, _)| correlation.clone()) + } + /// Cancel every dispatch this runner has in flight. /// /// For a caller that owns a runner serving one piece of work and wants to diff --git a/src/tui/src/worker/stream/router.rs b/src/tui/src/worker/stream/router.rs index 77c7c889..b40f7286 100644 --- a/src/tui/src/worker/stream/router.rs +++ b/src/tui/src/worker/stream/router.rs @@ -96,8 +96,14 @@ impl ScreenRouter { self.log(&format!("screen: {from} unsubscribed from {task_id}")); } } - ScreenMessage::Kill { task_id } => { - if !self.runtime.terminate_task(from, &task_id) { + ScreenMessage::Kill { + task_id, + correlation_id, + } => { + if !self + .runtime + .terminate_task(from, &task_id, &correlation_id) + { self.log(&format!( "screen: refused kill from {from} on {task_id} — no such running task for this sender" )); From 4fda36f27bbb07138e6b50be180f775372540800 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 15:10:57 +0300 Subject: [PATCH 08/35] test: add correlation_id to Kill messages in tests The test fixtures for ScreenMessage::Kill were missing the correlation_id field, which is now required by the message structure. Added the field to all three test locations to keep the test data consistent with the updated message definition. --- src/sdk/src/tinyplace/screen/tests.rs | 1 + src/tui/src/worker/stream/tests.rs | 1 + src/tui/tests/e2e_screen_stream.rs | 1 + 3 files changed, 3 insertions(+) diff --git a/src/sdk/src/tinyplace/screen/tests.rs b/src/sdk/src/tinyplace/screen/tests.rs index 1f7b9fd8..2905b22a 100644 --- a/src/sdk/src/tinyplace/screen/tests.rs +++ b/src/sdk/src/tinyplace/screen/tests.rs @@ -318,6 +318,7 @@ fn messages_round_trip_through_the_envelope() { }, ScreenMessage::Kill { task_id: "w_1".into(), + correlation_id: "cyc/w_1/0".into(), }, ScreenMessage::Ack { task_id: "w_1".into(), diff --git a/src/tui/src/worker/stream/tests.rs b/src/tui/src/worker/stream/tests.rs index c473329b..ba248e36 100644 --- a/src/tui/src/worker/stream/tests.rs +++ b/src/tui/src/worker/stream/tests.rs @@ -430,6 +430,7 @@ async fn a_kill_for_a_task_this_sender_never_dispatched_is_refused() { "peerA", medulla::tinyplace::ScreenMessage::Kill { task_id: "t1".into(), + correlation_id: "cyc/t1/0".into(), }, ); assert_eq!(router.active(), 0); diff --git a/src/tui/tests/e2e_screen_stream.rs b/src/tui/tests/e2e_screen_stream.rs index 8b7e18d5..f7fc61a7 100644 --- a/src/tui/tests/e2e_screen_stream.rs +++ b/src/tui/tests/e2e_screen_stream.rs @@ -330,6 +330,7 @@ async fn an_owned_task_kill_stops_its_real_harness() { peer, ScreenMessage::Kill { task_id: task_id.to_string(), + correlation_id: format!("cyc/{task_id}/0"), }, ); From 264ce05fffb3eef5d8fe253db19777ee30769822 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 15:12:05 +0300 Subject: [PATCH 09/35] test: cover screen kill capability and stale dispatch handling Add tests for the new screenKill agent capability, asserting it is omitted for older workers and defaults to false when absent. Also extend the TUI end-to-end screen stream test to verify that a stale kill dispatch with a reused task id does not stop the live session. --- src/sdk/src/tinyplace/frames/tests/codec.rs | 14 ++++++++++++++ src/tui/tests/e2e_screen_stream.rs | 16 ++++++++++++++++ 2 files changed, 30 insertions(+) diff --git a/src/sdk/src/tinyplace/frames/tests/codec.rs b/src/sdk/src/tinyplace/frames/tests/codec.rs index 15ef1665..fe1ddd34 100644 --- a/src/sdk/src/tinyplace/frames/tests/codec.rs +++ b/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(¤t).unwrap(); + assert_eq!(value["screenKill"], true); } #[test] diff --git a/src/tui/tests/e2e_screen_stream.rs b/src/tui/tests/e2e_screen_stream.rs index f7fc61a7..898f8c32 100644 --- a/src/tui/tests/e2e_screen_stream.rs +++ b/src/tui/tests/e2e_screen_stream.rs @@ -326,6 +326,22 @@ async fn an_owned_task_kill_stops_its_real_harness() { start_task(&runtime, peer, task_id).await; let mut router = ScreenRouter::new(sessions.clone(), runtime, send_fn(|_, _| async {})); + router.handle( + peer, + ScreenMessage::Kill { + task_id: task_id.to_string(), + correlation_id: "stale-dispatch".to_string(), + }, + ); + assert!( + sessions + .row(&session_id) + .expect("the live session remains inspectable") + .state + .is_running(), + "a stale dispatch receipt must not kill a reused task id" + ); + router.handle( peer, ScreenMessage::Kill { From 41ce3800866c586b1e53f75a9fe7481e1aaed0f1 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 15:12:51 +0300 Subject: [PATCH 10/35] style: format attribute and condition for readability Reformat the serde attribute on `screen_kill` and the `terminate_task` guard condition to fit on single lines, improving code readability without changing behavior. --- src/sdk/src/tinyplace/frames/types.rs | 6 +++++- src/tui/src/worker/stream/router.rs | 5 +---- 2 files changed, 6 insertions(+), 5 deletions(-) diff --git a/src/sdk/src/tinyplace/frames/types.rs b/src/sdk/src/tinyplace/frames/types.rs index 8b7c8465..718988d3 100644 --- a/src/sdk/src/tinyplace/frames/types.rs +++ b/src/sdk/src/tinyplace/frames/types.rs @@ -483,7 +483,11 @@ pub struct AgentCapabilities { /// /// 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")] + #[serde( + rename = "screenKill", + default, + skip_serializing_if = "std::ops::Not::not" + )] pub screen_kill: bool, } diff --git a/src/tui/src/worker/stream/router.rs b/src/tui/src/worker/stream/router.rs index b40f7286..3530b5bd 100644 --- a/src/tui/src/worker/stream/router.rs +++ b/src/tui/src/worker/stream/router.rs @@ -100,10 +100,7 @@ impl ScreenRouter { task_id, correlation_id, } => { - if !self - .runtime - .terminate_task(from, &task_id, &correlation_id) - { + if !self.runtime.terminate_task(from, &task_id, &correlation_id) { self.log(&format!( "screen: refused kill from {from} on {task_id} — no such running task for this sender" )); From e4ae395c6271e686b78063621dd4b7042413addf Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 15:12:59 +0300 Subject: [PATCH 11/35] fix(daemon): clarify terminate_task correlation semantics The terminate_task documentation now reflects that the operation is a termination signal rather than an abort, and explicitly describes how the correlation check prevents a delayed request from terminating a later dispatch that reused the same task id. --- src/sdk/src/daemon/runtime.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/sdk/src/daemon/runtime.rs b/src/sdk/src/daemon/runtime.rs index c8419376..7693381b 100644 --- a/src/sdk/src/daemon/runtime.rs +++ b/src/sdk/src/daemon/runtime.rs @@ -185,11 +185,11 @@ impl DaemonRuntime { .and_then(|task| task.session_id.clone()) } - /// Abort the running task identified by its authenticated sender and id. + /// Terminate the running task identified by sender, id, and dispatch receipt. /// - /// The abort remains bound to the task record while the map is locked. This - /// avoids resolving a reusable session id and acting on it after a later - /// task has claimed the same harness. + /// 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 { From 026dd2192f9e549a197a62376e4e6f35a473b1db Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 15:36:34 +0300 Subject: [PATCH 12/35] fix(daemon): stop probe from waiting on task slot permit The capability probe no longer acquires a concurrency slot before running, since control-plane negotiation must not block behind harness-task slots it may need to terminate. The probe remains serialized by the capability cache lock and bounded by its own timeout, and the TUI router now unsubscribes using the correct source channel when killing a task. --- src/sdk/src/daemon/task_loop/probe.rs | 11 +++-------- src/tui/src/worker/stream/router.rs | 2 +- 2 files changed, 4 insertions(+), 9 deletions(-) diff --git a/src/sdk/src/daemon/task_loop/probe.rs b/src/sdk/src/daemon/task_loop/probe.rs index 882bea69..09466d85 100644 --- a/src/sdk/src/daemon/task_loop/probe.rs +++ b/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 diff --git a/src/tui/src/worker/stream/router.rs b/src/tui/src/worker/stream/router.rs index 3530b5bd..c042eab0 100644 --- a/src/tui/src/worker/stream/router.rs +++ b/src/tui/src/worker/stream/router.rs @@ -106,7 +106,7 @@ impl ScreenRouter { )); return; } - self.registry.unsubscribe(&task_id); + self.registry.unsubscribe_for(from, &task_id); self.log(&format!("screen: {from} killed {task_id}")); } // The sampler chains from the last frame it actually sent, so an From 697445cab6390aba85e405ee30cf1f119f60f7a0 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 16:01:48 +0300 Subject: [PATCH 13/35] feat(hub): cache worker capabilities for kill checks Cache the last successfully negotiated capabilities per worker address during task dispatch, and use this cached data in the kill path instead of issuing a fresh capability probe. This ensures an emergency kill can proceed even when a worker is wedged and unable to respond to a new negotiation request. --- src/sdk/src/hub/handle/mod.rs | 7 +------ src/sdk/src/hub/runner/capabilities.rs | 8 +++++++- src/sdk/src/hub/runner/mod.rs | 11 +++++++++++ src/sdk/src/hub/runner/types.rs | 2 ++ src/sdk/src/hub/socket/task_run.rs | 14 ++++++++------ 5 files changed, 29 insertions(+), 13 deletions(-) diff --git a/src/sdk/src/hub/handle/mod.rs b/src/sdk/src/hub/handle/mod.rs index c95991c9..f64f7670 100644 --- a/src/sdk/src/hub/handle/mod.rs +++ b/src/sdk/src/hub/handle/mod.rs @@ -135,12 +135,7 @@ impl HubHandle { /// 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 capabilities = self - .runner - .capabilities(worker) - .await - .map_err(|error| error.to_string())?; - if !capabilities.screen_kill { + if !self.runner.supports_screen_kill(worker).await { return Err("worker does not advertise harness termination support".to_string()); } let correlation_id = self diff --git a/src/sdk/src/hub/runner/capabilities.rs b/src/sdk/src/hub/runner/capabilities.rs index 498f6128..de37e553 100644 --- a/src/sdk/src/hub/runner/capabilities.rs +++ b/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( diff --git a/src/sdk/src/hub/runner/mod.rs b/src/sdk/src/hub/runner/mod.rs index aff0ede4..b0a10f11 100644 --- a/src/sdk/src/hub/runner/mod.rs +++ b/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, @@ -220,6 +221,16 @@ impl TaskRunner { .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 diff --git a/src/sdk/src/hub/runner/types.rs b/src/sdk/src/hub/runner/types.rs index 9918eea8..353b4976 100644 --- a/src/sdk/src/hub/runner/types.rs +++ b/src/sdk/src/hub/runner/types.rs @@ -70,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>>, /// 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, diff --git a/src/sdk/src/hub/socket/task_run.rs b/src/sdk/src/hub/socket/task_run.rs index 43a58e7c..1387ba10 100644 --- a/src/sdk/src/hub/socket/task_run.rs +++ b/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) + }) } } }; From 742878b4f6d4f7bcf3eae77ebb58ff49d6e95f35 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 16:25:56 +0300 Subject: [PATCH 14/35] feat(daemon): gate screen-kill capability on router installation Screen termination support is now advertised only when the embedding worker installs an authenticated screen-message router, preventing false capability claims in headless or unauthenticated contexts. The daemon runtime exposes an explicit enablement method, and the hub runner now negotiates capabilities with abort awareness so a backend cancellation during negotiation is honored before task dispatch, with stale capability entries cleared on refresh failure. --- src/sdk/src/daemon/capabilities/mod.rs | 3 ++- src/sdk/src/daemon/runtime.rs | 8 ++++++++ src/sdk/src/daemon/task_loop/probe.rs | 4 ++++ src/sdk/src/daemon/types.rs | 4 +++- src/sdk/src/hub/runner/capabilities.rs | 25 +++++++++++++++++++++++++ src/sdk/src/hub/runner/mod.rs | 8 +++++--- src/sdk/src/hub/socket/task_run.rs | 15 ++++++++++++++- src/tui/src/worker_loop/commands.rs | 1 + 8 files changed, 62 insertions(+), 6 deletions(-) diff --git a/src/sdk/src/daemon/capabilities/mod.rs b/src/sdk/src/daemon/capabilities/mod.rs index b46cd9d5..bce57a1e 100644 --- a/src/sdk/src/daemon/capabilities/mod.rs +++ b/src/sdk/src/daemon/capabilities/mod.rs @@ -62,7 +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(), - screen_kill: true, + // 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(), diff --git a/src/sdk/src/daemon/runtime.rs b/src/sdk/src/daemon/runtime.rs index 7693381b..391355c0 100644 --- a/src/sdk/src/daemon/runtime.rs +++ b/src/sdk/src/daemon/runtime.rs @@ -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(), }), diff --git a/src/sdk/src/daemon/task_loop/probe.rs b/src/sdk/src/daemon/task_loop/probe.rs index 09466d85..7e04131e 100644 --- a/src/sdk/src/daemon/task_loop/probe.rs +++ b/src/sdk/src/daemon/task_loop/probe.rs @@ -11,6 +11,10 @@ impl DaemonRuntime { 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; + 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. diff --git a/src/sdk/src/daemon/types.rs b/src/sdk/src/daemon/types.rs index b5fcd359..7bcd43c0 100644 --- a/src/sdk/src/daemon/types.rs +++ b/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}; @@ -260,6 +260,8 @@ pub(super) struct Inner { pub(super) inflight_idle: Notify, /// Cached capability probe result. pub(super) capabilities: TokioMutex>, + /// 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 diff --git a/src/sdk/src/hub/runner/capabilities.rs b/src/sdk/src/hub/runner/capabilities.rs index de37e553..1cd6814a 100644 --- a/src/sdk/src/hub/runner/capabilities.rs +++ b/src/sdk/src/hub/runner/capabilities.rs @@ -17,6 +17,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 { + // 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( @@ -87,4 +90,26 @@ 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 { + let abort = std::sync::Arc::new(tokio::sync::Notify::new()); + self.aborts + .lock() + .expect("aborts lock") + .insert(abort_id.to_string(), abort.clone()); + tokio::select! { + biased; + _ = abort.notified() => { + self.aborts.lock().expect("aborts lock").remove(abort_id); + Err(RunError::Aborted) + } + result = self.capabilities(address) => result, + } + } } diff --git a/src/sdk/src/hub/runner/mod.rs b/src/sdk/src/hub/runner/mod.rs index b0a10f11..902a32e2 100644 --- a/src/sdk/src/hub/runner/mod.rs +++ b/src/sdk/src/hub/runner/mod.rs @@ -288,11 +288,13 @@ 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()); - self.aborts + let abort = self + .aborts .lock() .expect("aborts lock") - .insert(req.abort_id.clone(), abort.clone()); + .entry(req.abort_id.clone()) + .or_insert_with(|| Arc::new(Notify::new())) + .clone(); let _abort_guard = AbortGuard { aborts: self.aborts.clone(), key: req.abort_id.clone(), diff --git a/src/sdk/src/hub/socket/task_run.rs b/src/sdk/src/hub/socket/task_run.rs index 1387ba10..bbb0d53f 100644 --- a/src/sdk/src/hub/socket/task_run.rs +++ b/src/sdk/src/hub/socket/task_run.rs @@ -165,7 +165,20 @@ pub(super) async fn handle_task_run( // 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(); + let capabilities = match runner + .capabilities_for_dispatch(&worker_address, &task_id) + .await + { + Ok(capabilities) => Some(capabilities), + Err(crate::hub::RunError::Aborted) => { + let outcome = Err(crate::hub::RunError::Aborted); + let _ = socket + .emit("medulla:task_result", result_frame(&task_id, &outcome)) + .await; + return; + } + Err(_) => None, + }; // An explicit provider is authoritative. Only an untargeted task consults // the subscription strategy, and a failed/unknown budget probe falls open diff --git a/src/tui/src/worker_loop/commands.rs b/src/tui/src/worker_loop/commands.rs index ce7f3592..5f430858 100644 --- a/src/tui/src/worker_loop/commands.rs +++ b/src/tui/src/worker_loop/commands.rs @@ -90,6 +90,7 @@ fn start_worker( ) }); let daemon = worker_runtime(start, mode, provider, &transport); + daemon.enable_screen_kill(); // Screens are only worth streaming when there are screens: the headless // executor runs harnesses without a pty, so a subscriber finds nothing. let screens = From 8aa29ebb045b764036c34ad811181f416c6a8ba8 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 16:26:09 +0300 Subject: [PATCH 15/35] fix(sdk): add AtomicBool to daemon runtime imports The daemon runtime now imports AtomicBool alongside the existing atomic types, preparing for upcoming state tracking that requires a boolean atomic flag. --- src/sdk/src/daemon/runtime.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/sdk/src/daemon/runtime.rs b/src/sdk/src/daemon/runtime.rs index 391355c0..29f12543 100644 --- a/src/sdk/src/daemon/runtime.rs +++ b/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}; From 58a3d0f71a883a9ff70a6b3a2feab483e6a178de Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 16:49:28 +0300 Subject: [PATCH 16/35] fix(runner): scope kill support to the dispatch that negotiated it Kill correlation now checks the screen-kill capability captured for the specific dispatch rather than consulting the worker's latest capability advertisement, preventing kills against tasks whose worker no longer advertises termination support. The abort cleanup also verifies it is removing the same abort registration it created, and the TUI clears the kill confirmation when any status message replaces the prompt. --- src/sdk/src/hub/handle/mod.rs | 9 ++++----- src/sdk/src/hub/runner/capabilities.rs | 8 +++++++- src/sdk/src/hub/runner/mod.rs | 23 +++++++++++------------ src/sdk/src/hub/runner/types.rs | 2 ++ src/tui/src/ui/app/keys/agents.rs | 2 +- src/tui/src/ui/app/state.rs | 3 +++ 6 files changed, 28 insertions(+), 19 deletions(-) diff --git a/src/sdk/src/hub/handle/mod.rs b/src/sdk/src/hub/handle/mod.rs index f64f7670..2eab3388 100644 --- a/src/sdk/src/hub/handle/mod.rs +++ b/src/sdk/src/hub/handle/mod.rs @@ -135,14 +135,13 @@ impl HubHandle { /// 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) + .kill_correlation_for(worker, task_id) .await - .ok_or_else(|| format!("task {task_id} is no longer running on {worker}"))?; + .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: task_id.to_string(), diff --git a/src/sdk/src/hub/runner/capabilities.rs b/src/sdk/src/hub/runner/capabilities.rs index 1cd6814a..c70c840d 100644 --- a/src/sdk/src/hub/runner/capabilities.rs +++ b/src/sdk/src/hub/runner/capabilities.rs @@ -106,7 +106,13 @@ impl TaskRunner { tokio::select! { biased; _ = abort.notified() => { - self.aborts.lock().expect("aborts lock").remove(abort_id); + 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) } result = self.capabilities(address) => result, diff --git a/src/sdk/src/hub/runner/mod.rs b/src/sdk/src/hub/runner/mod.rs index 902a32e2..80c278be 100644 --- a/src/sdk/src/hub/runner/mod.rs +++ b/src/sdk/src/hub/runner/mod.rs @@ -212,25 +212,17 @@ impl TaskRunner { } /// Return the active dispatch receipt for a worker/task pair. - pub async fn correlation_for(&self, worker: &str, task_id: &str) -> Option { + pub async fn kill_correlation_for(&self, worker: &str, task_id: &str) -> Option { self.waiters .lock() .await .iter() - .find(|(_, waiter)| waiter.from == worker && waiter.task_id == task_id) + .find(|(_, waiter)| { + waiter.from == worker && waiter.task_id == task_id && waiter.screen_kill + }) .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 @@ -322,6 +314,12 @@ impl TaskRunner { } let mut attempt = 0u32; + let screen_kill = self + .capabilities + .lock() + .await + .get(&req.worker_address) + .is_some_and(|capabilities| capabilities.screen_kill); loop { let cid = format!( "{}/{}/{}", @@ -335,6 +333,7 @@ impl TaskRunner { cid.clone(), Waiter { task_id: req.task_id.clone(), + screen_kill, from: req.worker_address.clone(), reply: tx, status: status.clone(), diff --git a/src/sdk/src/hub/runner/types.rs b/src/sdk/src/hub/runner/types.rs index 353b4976..032aa092 100644 --- a/src/sdk/src/hub/runner/types.rs +++ b/src/sdk/src/hub/runner/types.rs @@ -5,6 +5,8 @@ use super::*; pub(super) struct Waiter { /// The task id this dispatch carries on the wire. pub(super) task_id: String, + /// Screen termination support negotiated for this exact dispatch. + pub(super) screen_kill: bool, /// The worker address this dispatch was sent to — the only sender whose /// frames may settle it. See [`Probe::from`]. pub(super) from: String, diff --git a/src/tui/src/ui/app/keys/agents.rs b/src/tui/src/ui/app/keys/agents.rs index 952eb267..60141ffc 100644 --- a/src/tui/src/ui/app/keys/agents.rs +++ b/src/tui/src/ui/app/keys/agents.rs @@ -84,8 +84,8 @@ impl App { 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"); + self.kill_armed = Some(target); } else { self.set_status("Select a running harness task first"); } diff --git a/src/tui/src/ui/app/state.rs b/src/tui/src/ui/app/state.rs index e310f22e..6e6d07c8 100644 --- a/src/tui/src/ui/app/state.rs +++ b/src/tui/src/ui/app/state.rs @@ -371,6 +371,9 @@ impl App { } pub fn set_status(&mut self, s: impl Into) { + // A destructive confirmation is valid only while its question remains + // visible. Any asynchronous status replacement cancels it. + self.kill_armed = None; self.status = s.into(); } From 345c76273bca30ed10e4c46a5ed755ffad2956b4 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 18:23:02 +0300 Subject: [PATCH 17/35] feat(hub): negotiate screen-kill capability per task run The screen-kill capability is now determined during task negotiation rather than looked up from the runner's cached capabilities, allowing each request to specify whether screen control is supported. A new `run_negotiated` method accepts this flag explicitly, while the existing `run` method defaults to disabling it for backward compatibility. --- src/sdk/src/hub/runner/mod.rs | 26 ++++++++++++++++++++------ src/sdk/src/hub/socket/task_run.rs | 3 ++- 2 files changed, 22 insertions(+), 7 deletions(-) diff --git a/src/sdk/src/hub/runner/mod.rs b/src/sdk/src/hub/runner/mod.rs index 149bdb79..3ca92a28 100644 --- a/src/sdk/src/hub/runner/mod.rs +++ b/src/sdk/src/hub/runner/mod.rs @@ -283,6 +283,26 @@ impl TaskRunner { &self, req: TaskRequest, status: Option>, + ) -> Result { + self.run_with_screen_kill(req, status, false).await + } + + /// Run a dispatch with the screen-control support negotiated specifically + /// for this request. + pub async fn run_negotiated( + &self, + req: TaskRequest, + status: Option>, + screen_kill: bool, + ) -> Result { + self.run_with_screen_kill(req, status, screen_kill).await + } + + async fn run_with_screen_kill( + &self, + req: TaskRequest, + status: Option>, + screen_kill: bool, ) -> Result { // Register this dispatch's abort signal FIRST — before the contact wait — // so a `task_abort` that arrives during contact negotiation (up to @@ -325,12 +345,6 @@ impl TaskRunner { } let mut attempt = 0u32; - let screen_kill = self - .capabilities - .lock() - .await - .get(&req.worker_address) - .is_some_and(|capabilities| capabilities.screen_kill); loop { let cid = format!( "{}/{}/{}", diff --git a/src/sdk/src/hub/socket/task_run.rs b/src/sdk/src/hub/socket/task_run.rs index b27e3b5e..777349e8 100644 --- a/src/sdk/src/hub/socket/task_run.rs +++ b/src/sdk/src/hub/socket/task_run.rs @@ -280,7 +280,8 @@ pub(super) async fn handle_task_run( fleet_depth: 0, }; - let outcome = runner.run(req, Some(tx)).await; + let screen_kill = capabilities.is_some_and(|capabilities| capabilities.screen_kill); + let outcome = runner.run_negotiated(req, Some(tx), screen_kill).await; match &outcome { Ok(o) => log(&format!( "hub: task {} ok ({} chars)", From b346590c39063754ece191d5b51ba4fea2134afc Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 18:23:26 +0300 Subject: [PATCH 18/35] fix(runner): wait for contact acceptance before starting task The runner now polls the relay until the contact is accepted before proceeding with task execution, rather than starting immediately. This prevents races where the task begins before the peer is ready, and the wait is abortable via the existing abort notification. --- src/sdk/src/hub/runner/capabilities.rs | 15 ++++++++++++++- .../src/tinyplace/frames/tests/capabilities.rs | 16 ++++++++++++++++ src/sdk/src/tinyplace/frames/tests/codec.rs | 13 ------------- src/sdk/src/tinyplace/frames/tests/mod.rs | 1 + 4 files changed, 31 insertions(+), 14 deletions(-) create mode 100644 src/sdk/src/tinyplace/frames/tests/capabilities.rs diff --git a/src/sdk/src/hub/runner/capabilities.rs b/src/sdk/src/hub/runner/capabilities.rs index 8c4d23de..865da629 100644 --- a/src/sdk/src/hub/runner/capabilities.rs +++ b/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) { @@ -115,6 +115,19 @@ impl TaskRunner { .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() => return Err(RunError::Aborted), + _ = tokio::time::sleep(CONTACT_POLL) => {} + } + } + } tokio::select! { biased; _ = abort.notified() => { diff --git a/src/sdk/src/tinyplace/frames/tests/capabilities.rs b/src/sdk/src/tinyplace/frames/tests/capabilities.rs new file mode 100644 index 00000000..43de798b --- /dev/null +++ b/src/sdk/src/tinyplace/frames/tests/capabilities.rs @@ -0,0 +1,16 @@ +//! Compatibility tests for additive worker capability fields. + +use crate::tinyplace::{parse_agent_capabilities, AgentCapabilities}; + +#[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 = AgentCapabilities { + screen_kill: true, + ..Default::default() + }; + let value = serde_json::to_value(¤t).unwrap(); + assert_eq!(value["screenKill"], true); +} diff --git a/src/sdk/src/tinyplace/frames/tests/codec.rs b/src/sdk/src/tinyplace/frames/tests/codec.rs index 4ac196e2..494df4cd 100644 --- a/src/sdk/src/tinyplace/frames/tests/codec.rs +++ b/src/sdk/src/tinyplace/frames/tests/codec.rs @@ -391,19 +391,6 @@ fn empty_budgets_and_readiness_are_omitted_on_the_wire() { 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(¤t).unwrap(); - assert_eq!(value["screenKill"], true); -} - #[test] fn new_capabilities_round_trip_budgets_and_readiness() { let caps = crate::tinyplace::AgentCapabilities { diff --git a/src/sdk/src/tinyplace/frames/tests/mod.rs b/src/sdk/src/tinyplace/frames/tests/mod.rs index dcce9b97..38f8a45b 100644 --- a/src/sdk/src/tinyplace/frames/tests/mod.rs +++ b/src/sdk/src/tinyplace/frames/tests/mod.rs @@ -1,4 +1,5 @@ //! Focused tests for task-frame encoding and restricted tool-mode decoding. mod codec; +mod capabilities; mod tool_mode; From 8f6e9d0b38a60464f42b7255f5f9ad0cb717fc3a Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 18:23:42 +0300 Subject: [PATCH 19/35] fix(runner): clean up stale abort registrations on abort When a task is aborted, the runner now removes its abort registration from the shared map if it still matches the current abort handle. This prevents stale entries from accumulating when aborts are superseded by newer ones, keeping the abort tracking state consistent. --- src/sdk/src/hub/runner/capabilities.rs | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/src/sdk/src/hub/runner/capabilities.rs b/src/sdk/src/hub/runner/capabilities.rs index 865da629..d2393abd 100644 --- a/src/sdk/src/hub/runner/capabilities.rs +++ b/src/sdk/src/hub/runner/capabilities.rs @@ -123,7 +123,15 @@ impl TaskRunner { { tokio::select! { biased; - _ = abort.notified() => return Err(RunError::Aborted), + _ = 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); + }, _ = tokio::time::sleep(CONTACT_POLL) => {} } } From c31d78583a4fcb3065215c55ca9eef9a77b0e72a Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 18:24:51 +0300 Subject: [PATCH 20/35] test: reorder test module declarations alphabetically Reordered the module declarations in the test file so that `capabilities` comes before `codec`, matching alphabetical order for consistency and readability. --- src/sdk/src/tinyplace/frames/tests/mod.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/sdk/src/tinyplace/frames/tests/mod.rs b/src/sdk/src/tinyplace/frames/tests/mod.rs index 38f8a45b..39b028aa 100644 --- a/src/sdk/src/tinyplace/frames/tests/mod.rs +++ b/src/sdk/src/tinyplace/frames/tests/mod.rs @@ -1,5 +1,5 @@ //! Focused tests for task-frame encoding and restricted tool-mode decoding. -mod codec; mod capabilities; +mod codec; mod tool_mode; From 2adb8c005049a51b4dc56ec588e66b1ecbc7b1e3 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 18:25:06 +0300 Subject: [PATCH 21/35] test(tinyplace): move budget window default test to capabilities The budget window default test was relocated from the codec test module to the capabilities test module, where it more naturally belongs alongside other capability-related tests. The test itself is unchanged and still verifies that the default budget window is unknown. --- src/sdk/src/tinyplace/frames/tests/capabilities.rs | 7 ++++++- src/sdk/src/tinyplace/frames/tests/codec.rs | 5 ----- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/src/sdk/src/tinyplace/frames/tests/capabilities.rs b/src/sdk/src/tinyplace/frames/tests/capabilities.rs index 43de798b..a270e90b 100644 --- a/src/sdk/src/tinyplace/frames/tests/capabilities.rs +++ b/src/sdk/src/tinyplace/frames/tests/capabilities.rs @@ -1,6 +1,11 @@ //! Compatibility tests for additive worker capability fields. -use crate::tinyplace::{parse_agent_capabilities, AgentCapabilities}; +use crate::tinyplace::{parse_agent_capabilities, AgentCapabilities, BudgetWindow}; + +#[test] +fn budget_window_defaults_to_unknown() { + assert_eq!(BudgetWindow::default(), BudgetWindow::Unknown); +} #[test] fn screen_kill_support_is_additive_and_defaults_off_for_older_workers() { diff --git a/src/sdk/src/tinyplace/frames/tests/codec.rs b/src/sdk/src/tinyplace/frames/tests/codec.rs index 494df4cd..e1c76254 100644 --- a/src/sdk/src/tinyplace/frames/tests/codec.rs +++ b/src/sdk/src/tinyplace/frames/tests/codec.rs @@ -347,11 +347,6 @@ fn harness_readiness_omits_reason_when_ready() { assert_eq!(value["reason"], "not authenticated"); } -#[test] -fn budget_window_defaults_to_unknown() { - assert_eq!(BudgetWindow::default(), BudgetWindow::Unknown); -} - #[test] fn parse_agent_capabilities_defaults_missing_arrays() { let caps = parse_agent_capabilities(r#"{"cwd":"/x"}"#).unwrap(); From 5e6983098fa15787c20135e675665a41606a868d Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 18:48:16 +0300 Subject: [PATCH 22/35] fix(hub): reuse abort signal from capability negotiation The abort signal created during capability negotiation is now passed through to the task run, ensuring that an abort requested during the negotiation phase is honored when the task actually starts running. This prevents a race where an abort could be missed between capability discovery and task execution. --- src/sdk/src/hub/runner/capabilities.rs | 6 ++++-- src/sdk/src/hub/runner/mod.rs | 14 +++++++------- src/sdk/src/hub/socket/task_run.rs | 12 ++++++++---- 3 files changed, 19 insertions(+), 13 deletions(-) diff --git a/src/sdk/src/hub/runner/capabilities.rs b/src/sdk/src/hub/runner/capabilities.rs index d2393abd..778bb30c 100644 --- a/src/sdk/src/hub/runner/capabilities.rs +++ b/src/sdk/src/hub/runner/capabilities.rs @@ -109,7 +109,7 @@ impl TaskRunner { &self, address: &str, abort_id: &str, - ) -> Result { + ) -> Result<(AgentCapabilities, std::sync::Arc), RunError> { let abort = std::sync::Arc::new(tokio::sync::Notify::new()); self.aborts .lock() @@ -148,7 +148,9 @@ impl TaskRunner { } Err(RunError::Aborted) } - result = self.capabilities(address) => result, + result = self.capabilities(address) => { + result.map(|capabilities| (capabilities, abort)) + }, } } } diff --git a/src/sdk/src/hub/runner/mod.rs b/src/sdk/src/hub/runner/mod.rs index 3ca92a28..5b1405c7 100644 --- a/src/sdk/src/hub/runner/mod.rs +++ b/src/sdk/src/hub/runner/mod.rs @@ -294,15 +294,17 @@ impl TaskRunner { req: TaskRequest, status: Option>, screen_kill: bool, + abort: Option>, ) -> Result { - self.run_with_screen_kill(req, status, screen_kill).await + self.run_inner(req, status, screen_kill, abort).await } - async fn run_with_screen_kill( + async fn run_inner( &self, req: TaskRequest, status: Option>, screen_kill: bool, + prepared_abort: Option>, ) -> Result { // Register this dispatch's abort signal FIRST — before the contact wait — // so a `task_abort` that arrives during contact negotiation (up to @@ -311,13 +313,11 @@ 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 = self - .aborts + let abort = prepared_abort.unwrap_or_else(|| Arc::new(Notify::new())); + self.aborts .lock() .expect("aborts lock") - .entry(req.abort_id.clone()) - .or_insert_with(|| Arc::new(Notify::new())) - .clone(); + .insert(req.abort_id.clone(), abort.clone()); let _abort_guard = AbortGuard { aborts: self.aborts.clone(), key: req.abort_id.clone(), diff --git a/src/sdk/src/hub/socket/task_run.rs b/src/sdk/src/hub/socket/task_run.rs index 777349e8..e7213490 100644 --- a/src/sdk/src/hub/socket/task_run.rs +++ b/src/sdk/src/hub/socket/task_run.rs @@ -169,7 +169,7 @@ pub(super) async fn handle_task_run( .capabilities_for_dispatch(&worker_address, &task_id) .await { - Ok(capabilities) => Some(capabilities), + Ok((capabilities, abort)) => Some((capabilities, abort)), Err(crate::hub::RunError::Aborted) => { let outcome = Err(crate::hub::RunError::Aborted); let _ = socket @@ -192,7 +192,7 @@ pub(super) async fn handle_task_run( if strategy == crate::runtime::SubscriptionRoutingStrategy::Manual { None } else { - capabilities.as_ref().and_then(|capabilities| { + capabilities.as_ref().and_then(|(capabilities, _)| { super::super::roster::subscription_for_strategy(capabilities, strategy) }) } @@ -280,8 +280,12 @@ pub(super) async fn handle_task_run( fleet_depth: 0, }; - let screen_kill = capabilities.is_some_and(|capabilities| capabilities.screen_kill); - let outcome = runner.run_negotiated(req, Some(tx), screen_kill).await; + let (screen_kill, abort) = capabilities + .map(|(capabilities, abort)| (capabilities.screen_kill, Some(abort))) + .unwrap_or((false, None)); + let outcome = runner + .run_negotiated(req, Some(tx), screen_kill, abort) + .await; match &outcome { Ok(o) => log(&format!( "hub: task {} ok ({} chars)", From da5af4f3b8085afb11987269ea21c875f6f77e98 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 18:48:23 +0300 Subject: [PATCH 23/35] fix(runner): delegate run to run_inner The `run` method now calls `run_inner` directly instead of `run_with_screen_kill`, passing `None` for the screen-control parameter. This aligns the public API with the internal execution path and removes the now-redundant wrapper. --- src/sdk/src/hub/runner/mod.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/sdk/src/hub/runner/mod.rs b/src/sdk/src/hub/runner/mod.rs index 5b1405c7..7d2e3ee0 100644 --- a/src/sdk/src/hub/runner/mod.rs +++ b/src/sdk/src/hub/runner/mod.rs @@ -284,7 +284,7 @@ impl TaskRunner { req: TaskRequest, status: Option>, ) -> Result { - self.run_with_screen_kill(req, status, false).await + self.run_inner(req, status, false, None).await } /// Run a dispatch with the screen-control support negotiated specifically From 218c4ede6515fdfbd2fd40ee449319a93809b381 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 18:48:37 +0300 Subject: [PATCH 24/35] fix(pty): stop turn only when orchestrator still owns session The stop_turn path now checks whether the operator has taken over the session before interrupting and closing it. This prevents the orchestrator from terminating a session that has been handed to the user, avoiding a race between control handoff and termination. --- src/tui/src/worker/executor/run.rs | 3 +-- src/tui/src/worker/pty/handle/control.rs | 14 ++++++++++++++ src/tui/src/worker/pty/manager/session.rs | 6 ++++++ 3 files changed, 21 insertions(+), 2 deletions(-) diff --git a/src/tui/src/worker/executor/run.rs b/src/tui/src/worker/executor/run.rs index 18ab62ea..c69d7109 100644 --- a/src/tui/src/worker/executor/run.rs +++ b/src/tui/src/worker/executor/run.rs @@ -277,8 +277,7 @@ impl PtySessionExecutor { /// composer, which is the failure that produces confidently wrong answers /// rather than an error. fn stop_turn(&self, id: &str) { - let _ = self.sessions.write(id, &[0x03]); - self.sessions.close(id); + self.sessions.stop_if_orchestrator(id); } /// Decide which session serves this task: reuse an idle one, or launch. diff --git a/src/tui/src/worker/pty/handle/control.rs b/src/tui/src/worker/pty/handle/control.rs index 8929c0f5..ef2b8f56 100644 --- a/src/tui/src/worker/pty/handle/control.rs +++ b/src/tui/src/worker/pty/handle/control.rs @@ -29,10 +29,24 @@ impl SessionHandle { /// running that turn. Clearing `busy` on handback would advertise a harness /// as free while it was still finishing someone else's work. pub(in super::super) fn set_control(&self, control: HarnessControl) { + let _cold = lock(&self.cold); self.operator_held .store(control == HarnessControl::User, Ordering::Release); } + /// Interrupt and close only while orchestration still owns the session. + /// Serialized with control handoff so takeover and termination cannot race. + pub(in super::super) fn stop_if_orchestrator(&self) -> bool { + let _cold = lock(&self.cold); + if self.operator_held.load(Ordering::Acquire) { + return false; + } + let _ = self.write(&[0x03]); + self.kill(); + self.mark_closed(); + true + } + /// Whether this session may serve `label`'s next turn. /// /// Its own label, or the synthetic `you:` one of a handed-back diff --git a/src/tui/src/worker/pty/manager/session.rs b/src/tui/src/worker/pty/manager/session.rs index 2c9eb670..bed9c65c 100644 --- a/src/tui/src/worker/pty/manager/session.rs +++ b/src/tui/src/worker/pty/manager/session.rs @@ -186,6 +186,12 @@ impl PtyManager { true } + /// Interrupt and close a session only if the operator has not taken it. + pub fn stop_if_orchestrator(&self, id: &str) -> bool { + self.handle(id) + .is_some_and(|session| session.stop_if_orchestrator()) + } + /// Drop an exited session's record and screen. /// /// Refuses while the child is alive, so a forgotten session can never leave From 701c292ad5d72c9294fd3a4a62a3de8e150ba3b0 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 18:55:04 +0300 Subject: [PATCH 25/35] fix(hub): return abort handle even when capability negotiation fails The capabilities negotiation now always returns the abort notification handle alongside the result, so callers can still use it to cancel a task even when capability discovery fails. This simplifies the dispatch flow by removing the need to reconstruct the abort handle from an optional tuple. --- src/sdk/src/hub/runner/capabilities.rs | 11 +++++++---- src/sdk/src/hub/socket/task_run.rs | 16 +++++++--------- 2 files changed, 14 insertions(+), 13 deletions(-) diff --git a/src/sdk/src/hub/runner/capabilities.rs b/src/sdk/src/hub/runner/capabilities.rs index 778bb30c..ad525eb8 100644 --- a/src/sdk/src/hub/runner/capabilities.rs +++ b/src/sdk/src/hub/runner/capabilities.rs @@ -109,7 +109,10 @@ impl TaskRunner { &self, address: &str, abort_id: &str, - ) -> Result<(AgentCapabilities, std::sync::Arc), RunError> { + ) -> ( + Result, + std::sync::Arc, + ) { let abort = std::sync::Arc::new(tokio::sync::Notify::new()); self.aborts .lock() @@ -130,7 +133,7 @@ impl TaskRunner { }) { aborts.remove(abort_id); } - return Err(RunError::Aborted); + return (Err(RunError::Aborted), abort); }, _ = tokio::time::sleep(CONTACT_POLL) => {} } @@ -146,10 +149,10 @@ impl TaskRunner { { aborts.remove(abort_id); } - Err(RunError::Aborted) + (Err(RunError::Aborted), abort) } result = self.capabilities(address) => { - result.map(|capabilities| (capabilities, abort)) + (result, abort) }, } } diff --git a/src/sdk/src/hub/socket/task_run.rs b/src/sdk/src/hub/socket/task_run.rs index e7213490..89378d3e 100644 --- a/src/sdk/src/hub/socket/task_run.rs +++ b/src/sdk/src/hub/socket/task_run.rs @@ -165,19 +165,19 @@ pub(super) async fn handle_task_run( // 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 = match runner + let (capabilities, abort) = match runner .capabilities_for_dispatch(&worker_address, &task_id) .await { - Ok((capabilities, abort)) => Some((capabilities, abort)), - Err(crate::hub::RunError::Aborted) => { + (Ok(capabilities), abort) => (Some(capabilities), abort), + (Err(crate::hub::RunError::Aborted), _) => { let outcome = Err(crate::hub::RunError::Aborted); let _ = socket .emit("medulla:task_result", result_frame(&task_id, &outcome)) .await; return; } - Err(_) => None, + (Err(_), abort) => (None, abort), }; // An explicit provider is authoritative. Only an untargeted task consults @@ -192,7 +192,7 @@ pub(super) async fn handle_task_run( if strategy == crate::runtime::SubscriptionRoutingStrategy::Manual { None } else { - capabilities.as_ref().and_then(|(capabilities, _)| { + capabilities.as_ref().and_then(|capabilities| { super::super::roster::subscription_for_strategy(capabilities, strategy) }) } @@ -280,11 +280,9 @@ pub(super) async fn handle_task_run( fleet_depth: 0, }; - let (screen_kill, abort) = capabilities - .map(|(capabilities, abort)| (capabilities.screen_kill, Some(abort))) - .unwrap_or((false, None)); + let screen_kill = capabilities.is_some_and(|capabilities| capabilities.screen_kill); let outcome = runner - .run_negotiated(req, Some(tx), screen_kill, abort) + .run_negotiated(req, Some(tx), screen_kill, Some(abort)) .await; match &outcome { Ok(o) => log(&format!( From c62e32a1ab07d0b0bc587a3c5ef776e74893a452 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 19:02:13 +0300 Subject: [PATCH 26/35] fix(control_socket): negotiate screen kill before dispatch The task runner now checks worker capabilities and abort state before dispatching a task, allowing the control socket to return early with an aborted outcome when the task has already been cancelled. This prevents unnecessary work and ensures the screen kill capability is properly negotiated with the worker before execution begins. --- src/sdk/src/control_socket/server/hub_ops.rs | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/src/sdk/src/control_socket/server/hub_ops.rs b/src/sdk/src/control_socket/server/hub_ops.rs index 8fe37ac6..e9344951 100644 --- a/src/sdk/src/control_socket/server/hub_ops.rs +++ b/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)) + .await; record_outcome(&activity, &wire_task_id, &outcome); outcome } From 04d256ef17b8cd4731010a22f905c21e501d64da Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 19:10:01 +0300 Subject: [PATCH 27/35] fix(hub): pass visible task id through negotiated runs The hub now forwards the externally visible task id when running negotiated tasks, so abort signals and waiter registrations reference the id callers expect rather than the internal request id. This also clears the armed kill state on any mouse activity in the TUI, cancelling a pending harness kill and showing a status message when the user interacts. --- src/sdk/src/control_socket/server/hub_ops.rs | 8 +++++++- src/sdk/src/hub/runner/mod.rs | 11 ++++++++--- src/sdk/src/hub/socket/task_run.rs | 2 +- src/tui/src/ui/app/input.rs | 3 +++ 4 files changed, 19 insertions(+), 5 deletions(-) diff --git a/src/sdk/src/control_socket/server/hub_ops.rs b/src/sdk/src/control_socket/server/hub_ops.rs index e9344951..a56bf187 100644 --- a/src/sdk/src/control_socket/server/hub_ops.rs +++ b/src/sdk/src/control_socket/server/hub_ops.rs @@ -202,7 +202,13 @@ impl FleetOps for HubFleetOps { Err(_) => false, }; let outcome = runner - .run_negotiated(request, Some(tee), screen_kill, Some(abort)) + .run_negotiated( + request, + Some(tee), + screen_kill, + Some(abort), + Some(task_id), + ) .await; record_outcome(&activity, &wire_task_id, &outcome); outcome diff --git a/src/sdk/src/hub/runner/mod.rs b/src/sdk/src/hub/runner/mod.rs index 7d2e3ee0..b44ed9e4 100644 --- a/src/sdk/src/hub/runner/mod.rs +++ b/src/sdk/src/hub/runner/mod.rs @@ -284,7 +284,7 @@ impl TaskRunner { req: TaskRequest, status: Option>, ) -> Result { - self.run_inner(req, status, false, None).await + self.run_inner(req, status, false, None, None).await } /// Run a dispatch with the screen-control support negotiated specifically @@ -295,8 +295,10 @@ impl TaskRunner { status: Option>, screen_kill: bool, abort: Option>, + visible_task_id: Option, ) -> Result { - self.run_inner(req, status, screen_kill, abort).await + self.run_inner(req, status, screen_kill, abort, visible_task_id) + .await } async fn run_inner( @@ -305,6 +307,7 @@ impl TaskRunner { status: Option>, screen_kill: bool, prepared_abort: Option>, + visible_task_id: Option, ) -> Result { // Register this dispatch's abort signal FIRST — before the contact wait — // so a `task_abort` that arrives during contact negotiation (up to @@ -357,7 +360,9 @@ impl TaskRunner { self.waiters.lock().await.insert( cid.clone(), Waiter { - task_id: req.task_id.clone(), + task_id: visible_task_id + .clone() + .unwrap_or_else(|| req.task_id.clone()), screen_kill, from: req.worker_address.clone(), reply: tx, diff --git a/src/sdk/src/hub/socket/task_run.rs b/src/sdk/src/hub/socket/task_run.rs index 89378d3e..f312ee0a 100644 --- a/src/sdk/src/hub/socket/task_run.rs +++ b/src/sdk/src/hub/socket/task_run.rs @@ -282,7 +282,7 @@ pub(super) async fn handle_task_run( let screen_kill = capabilities.is_some_and(|capabilities| capabilities.screen_kill); let outcome = runner - .run_negotiated(req, Some(tx), screen_kill, Some(abort)) + .run_negotiated(req, Some(tx), screen_kill, Some(abort), None) .await; match &outcome { Ok(o) => log(&format!( diff --git a/src/tui/src/ui/app/input.rs b/src/tui/src/ui/app/input.rs index 3e5b12fe..b647526c 100644 --- a/src/tui/src/ui/app/input.rs +++ b/src/tui/src/ui/app/input.rs @@ -57,6 +57,9 @@ impl App { /// Handle scroll and left-click mouse events for the active tab. pub(super) fn on_mouse(&mut self, m: crossterm::event::MouseEvent) -> Option { + if self.kill_armed.take().is_some() { + self.set_status("Harness kill cancelled"); + } // A modal swallows the mouse, the same way it swallows the keyboard. // The harness picker is one: a click that navigated the rail behind it // left an overlay on screen describing a row nobody was pointing at. From c2b1cfc30889ac27290e6b776549720450f5ffd2 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 19:11:23 +0300 Subject: [PATCH 28/35] refactor(control_socket): simplify run_negotiated call The call to run_negotiated in hub_ops.rs was reformatted to fit on a single line, reducing unnecessary line breaks without changing any behavior. --- src/sdk/src/control_socket/server/hub_ops.rs | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/src/sdk/src/control_socket/server/hub_ops.rs b/src/sdk/src/control_socket/server/hub_ops.rs index a56bf187..569f54d4 100644 --- a/src/sdk/src/control_socket/server/hub_ops.rs +++ b/src/sdk/src/control_socket/server/hub_ops.rs @@ -202,13 +202,7 @@ impl FleetOps for HubFleetOps { Err(_) => false, }; let outcome = runner - .run_negotiated( - request, - Some(tee), - screen_kill, - Some(abort), - Some(task_id), - ) + .run_negotiated(request, Some(tee), screen_kill, Some(abort), Some(task_id)) .await; record_outcome(&activity, &wire_task_id, &outcome); outcome From 35072e96ea6e12b370d1f02ea4203a03d7252bd8 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 19:17:44 +0300 Subject: [PATCH 29/35] refactor(keys): extract kill arming and require unmodified y The kill confirmation is now armed through a dedicated method that sets the status and stores the target as one state transition, and the confirming `y` keypress only proceeds when no modifiers are held, preventing accidental kills with modified key combinations. --- src/tui/src/ui/app/keys/agents.rs | 3 +-- src/tui/src/ui/app/keys/mod.rs | 2 +- src/tui/src/ui/app/state.rs | 7 +++++++ 3 files changed, 9 insertions(+), 3 deletions(-) diff --git a/src/tui/src/ui/app/keys/agents.rs b/src/tui/src/ui/app/keys/agents.rs index 60141ffc..84112c8a 100644 --- a/src/tui/src/ui/app/keys/agents.rs +++ b/src/tui/src/ui/app/keys/agents.rs @@ -84,8 +84,7 @@ impl App { match k.code { KeyCode::Char('K') => { if let Some(target) = self.watch_target() { - self.set_status("Kill this harness? y confirm · any other key cancels"); - self.kill_armed = Some(target); + self.arm_kill(target); } else { self.set_status("Select a running harness task first"); } diff --git a/src/tui/src/ui/app/keys/mod.rs b/src/tui/src/ui/app/keys/mod.rs index 8f5428cd..0cbdc2be 100644 --- a/src/tui/src/ui/app/keys/mod.rs +++ b/src/tui/src/ui/app/keys/mod.rs @@ -47,7 +47,7 @@ impl App { // 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') { + if k.code == KeyCode::Char('y') && k.modifiers.is_empty() { self.set_status(format!("Killing harness for {task_id}…")); return Some(Cmd::KillTask { worker, task_id }); } diff --git a/src/tui/src/ui/app/state.rs b/src/tui/src/ui/app/state.rs index 196e8d47..f87ef0b9 100644 --- a/src/tui/src/ui/app/state.rs +++ b/src/tui/src/ui/app/state.rs @@ -379,6 +379,13 @@ impl App { self.status = s.into(); } + /// Show and arm the harness-kill confirmation as one invariant-preserving + /// state transition. + pub(super) fn arm_kill(&mut self, target: (String, String)) { + self.set_status("Kill this harness? y confirm · any other key cancels"); + self.kill_armed = Some(target); + } + /// Replace the Context-tab chunks. pub fn set_contexts(&mut self, c: Vec) { self.contexts = c; From dd64e1301137760d5e23ba252ea2099320a1ea99 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 19:24:43 +0300 Subject: [PATCH 30/35] fix(hub): use wire task id when killing screen tasks The kill path now sends the original wire task id to the worker instead of the visible task id, ensuring the worker can correctly identify and terminate the intended task. Additionally, the capabilities probe now returns cached values immediately while refreshing them in the background when screen kill is active, avoiding a blocking wait during task termination. --- src/sdk/src/daemon/task_loop/probe.rs | 20 +++++++++++++++++++- src/sdk/src/hub/handle/mod.rs | 4 ++-- src/sdk/src/hub/runner/mod.rs | 9 +++++++-- src/sdk/src/hub/runner/types.rs | 2 ++ 4 files changed, 30 insertions(+), 5 deletions(-) diff --git a/src/sdk/src/daemon/task_loop/probe.rs b/src/sdk/src/daemon/task_loop/probe.rs index 7e04131e..77e124c9 100644 --- a/src/sdk/src/daemon/task_loop/probe.rs +++ b/src/sdk/src/daemon/task_loop/probe.rs @@ -10,7 +10,25 @@ 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 diff --git a/src/sdk/src/hub/handle/mod.rs b/src/sdk/src/hub/handle/mod.rs index 670d71dc..53bda115 100644 --- a/src/sdk/src/hub/handle/mod.rs +++ b/src/sdk/src/hub/handle/mod.rs @@ -135,7 +135,7 @@ impl HubHandle { /// 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 = self + let (correlation_id, wire_task_id) = self .runner .kill_correlation_for(worker, task_id) .await @@ -144,7 +144,7 @@ impl HubHandle { })?; let body = crate::tinyplace::encode_screen_message(&crate::tinyplace::ScreenMessage::Kill { - task_id: task_id.to_string(), + task_id: wire_task_id, correlation_id, }); (self.log)(&format!("hub: killing task {task_id} on {worker}")); diff --git a/src/sdk/src/hub/runner/mod.rs b/src/sdk/src/hub/runner/mod.rs index b44ed9e4..85a7e6d4 100644 --- a/src/sdk/src/hub/runner/mod.rs +++ b/src/sdk/src/hub/runner/mod.rs @@ -223,7 +223,11 @@ impl TaskRunner { } /// Return the active dispatch receipt for a worker/task pair. - pub async fn kill_correlation_for(&self, worker: &str, task_id: &str) -> Option { + pub async fn kill_correlation_for( + &self, + worker: &str, + task_id: &str, + ) -> Option<(String, String)> { self.waiters .lock() .await @@ -231,7 +235,7 @@ impl TaskRunner { .find(|(_, waiter)| { waiter.from == worker && waiter.task_id == task_id && waiter.screen_kill }) - .map(|(correlation, _)| correlation.clone()) + .map(|(correlation, waiter)| (correlation.clone(), waiter.wire_task_id.clone())) } /// Cancel every dispatch this runner has in flight. @@ -363,6 +367,7 @@ impl TaskRunner { 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, diff --git a/src/sdk/src/hub/runner/types.rs b/src/sdk/src/hub/runner/types.rs index f0bc244d..7cd422cd 100644 --- a/src/sdk/src/hub/runner/types.rs +++ b/src/sdk/src/hub/runner/types.rs @@ -5,6 +5,8 @@ use super::*; pub(super) struct Waiter { /// The task id this dispatch carries on the wire. pub(super) task_id: String, + /// Worker-facing task id used by the daemon's running-task registry. + pub(super) wire_task_id: String, /// Screen termination support negotiated for this exact dispatch. pub(super) screen_kill: bool, /// The worker address this dispatch was sent to — the only sender whose From c3182ba671784056f69164d07c48d69fbf2abb39 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 19:25:16 +0300 Subject: [PATCH 31/35] fix(tui): ignore non-running tasks in input routing The input handler now checks that a task is still running before routing key events to it, preventing stale or completed tasks from receiving input. --- src/tui/src/ui/app/input.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/tui/src/ui/app/input.rs b/src/tui/src/ui/app/input.rs index b647526c..189eabc6 100644 --- a/src/tui/src/ui/app/input.rs +++ b/src/tui/src/ui/app/input.rs @@ -6,7 +6,7 @@ use crossterm::event::{Event, KeyEventKind, MouseButton, MouseEventKind}; use super::rail::RailRow; -use crate::ui::agents::{agent_row_model, AgentRole, AgentRow}; +use crate::ui::agents::{agent_row_model, AgentRole, AgentRow, TaskStatus}; use crate::ui::composer::Draft; use super::types::{App, Cmd, ROUTING_SUBPAGES, SETTINGS_SUBPAGES, TOKENMAXXING_SUBPAGES}; @@ -525,6 +525,9 @@ impl App { else { return None; }; + if task.status != TaskStatus::Running { + return None; + } let lanes = self.lanes(); let lane = lanes.get(*lane_index)?; // `Agent` is main's name for a roster agent / delegated task / peer From 3f92d04468639fa375629e77a61ee4f4b5d40d0f Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 19:25:29 +0300 Subject: [PATCH 32/35] fix(keys): restrict kill command to running tasks The kill command now uses a dedicated kill_target method that only returns a target when the selected task is running, preventing accidental termination of non-running tasks. The watch_target method no longer filters by task status, allowing it to serve other purposes without the kill-specific constraint. --- src/tui/src/ui/app/input.rs | 15 ++++++++++++--- src/tui/src/ui/app/keys/agents.rs | 2 +- 2 files changed, 13 insertions(+), 4 deletions(-) diff --git a/src/tui/src/ui/app/input.rs b/src/tui/src/ui/app/input.rs index 189eabc6..58af84de 100644 --- a/src/tui/src/ui/app/input.rs +++ b/src/tui/src/ui/app/input.rs @@ -525,9 +525,6 @@ impl App { else { return None; }; - if task.status != TaskStatus::Running { - return None; - } let lanes = self.lanes(); let lane = lanes.get(*lane_index)?; // `Agent` is main's name for a roster agent / delegated task / peer @@ -549,4 +546,16 @@ impl App { .unwrap_or_else(|| agent_id.to_string()); Some((address, task.task_id.clone())) } + + /// The selected running task eligible for destructive termination. + pub(super) fn kill_target(&self) -> Option<(String, String)> { + let rows = self.rail_rows(); + let row = rows.get(self.agent_index.min(rows.len().saturating_sub(1)))?; + let RailRow::Agent(AgentRow::Sub { task, .. }) = row else { + return None; + }; + (task.status == TaskStatus::Running) + .then(|| self.watch_target()) + .flatten() + } } diff --git a/src/tui/src/ui/app/keys/agents.rs b/src/tui/src/ui/app/keys/agents.rs index 84112c8a..05078a44 100644 --- a/src/tui/src/ui/app/keys/agents.rs +++ b/src/tui/src/ui/app/keys/agents.rs @@ -83,7 +83,7 @@ impl App { match k.code { KeyCode::Char('K') => { - if let Some(target) = self.watch_target() { + if let Some(target) = self.kill_target() { self.arm_kill(target); } else { self.set_status("Select a running harness task first"); From 48b27e1eac9bdf5cd4db2b025d3aeedc1ee5237b Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 19:28:06 +0300 Subject: [PATCH 33/35] test(tui): use running-task fixture in kill confirmation tests The kill confirmation tests previously used a bare app fixture without a running task, which did not reflect the real scenario where a task is active. A new helper now seeds a running task before these tests, ensuring the kill flow is exercised against a realistic state. --- src/tui/src/ui/app/tests.rs | 23 ++++++++++++++++++++--- 1 file changed, 20 insertions(+), 3 deletions(-) diff --git a/src/tui/src/ui/app/tests.rs b/src/tui/src/ui/app/tests.rs index 78ea951a..d0ed3924 100644 --- a/src/tui/src/ui/app/tests.rs +++ b/src/tui/src/ui/app/tests.rs @@ -22,6 +22,23 @@ fn app() -> App { App::new(rt, loaded) } +fn app_with_running_task() -> App { + let rt = MockRuntime::demo(); + rt.script_event(crate::ui::TuiEvent::TaskStart { + task_id: "task-1".into(), + instruction: "Continue the auth refactor.".into(), + depth: 2, + agent_id: Some("dev-1".into()), + contract: None, + }); + let loaded = { + let mut loaded = LoadedConfig::defaults("medulla.tui.json".into()); + loaded.config.tinyplace = Some(medulla::config::TinyplaceConfig::default()); + loaded + }; + App::new(Arc::new(rt), loaded) +} + /// The index of the tab named `name`. Looked up rather than written down: the /// tab bar's order is a product decision that has changed before. fn tab(name: &str) -> usize { @@ -294,7 +311,7 @@ fn selecting_a_task_asks_to_watch_it() { #[test] fn killing_a_watched_harness_requires_confirmation() { - let mut app = app(); + let mut app = app_with_running_task(); select_first_task(&mut app).expect("the fixture has a selectable task"); app.focus_agents_rail(); @@ -310,7 +327,7 @@ fn killing_a_watched_harness_requires_confirmation() { #[test] fn killing_resolves_the_current_rail_selection_instead_of_the_cached_watch() { - let mut app = app(); + let mut app = app_with_running_task(); select_first_task(&mut app).expect("the fixture has a selectable task"); let selected = app.watch_target().expect("the selected task is watchable"); app.watching = Some(("stale-worker".into(), "stale-task".into())); @@ -328,7 +345,7 @@ fn killing_resolves_the_current_rail_selection_instead_of_the_cached_watch() { #[test] fn any_other_key_cancels_a_harness_kill() { - let mut app = app(); + let mut app = app_with_running_task(); select_first_task(&mut app).expect("the fixture has a selectable task"); app.focus_agents_rail(); app.on_key(KeyEvent::new(KeyCode::Char('K'), KeyModifiers::SHIFT)); From 594493707dbf11b010ec5f9f6fb07b185c066bdf Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 19:31:13 +0300 Subject: [PATCH 34/35] fix(tests): update TuiEvent import path in test helper The test helper `app_with_running_task` referenced `TuiEvent` via the old module path, which has been moved to `crate::ui::events`. The import is updated to reflect the new location, ensuring the test compiles and runs correctly. --- src/tui/src/ui/app/tests.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/tui/src/ui/app/tests.rs b/src/tui/src/ui/app/tests.rs index d0ed3924..e3f741a0 100644 --- a/src/tui/src/ui/app/tests.rs +++ b/src/tui/src/ui/app/tests.rs @@ -24,7 +24,7 @@ fn app() -> App { fn app_with_running_task() -> App { let rt = MockRuntime::demo(); - rt.script_event(crate::ui::TuiEvent::TaskStart { + rt.script_event(crate::ui::events::TuiEvent::TaskStart { task_id: "task-1".into(), instruction: "Continue the auth refactor.".into(), depth: 2, From c755bfed3e46b5c481d0a28abdb84bda5a9f9b4d Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 2 Aug 2026 19:32:46 +0300 Subject: [PATCH 35/35] test(tui): update running task fixture setup The test helper now constructs the app first and then removes any TaskComplete events for the demo task from the snapshot, rather than scripting a TaskStart event before app creation. This aligns the fixture with the current event flow where task completion is the relevant state to filter. --- src/tui/src/ui/app/tests.rs | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/src/tui/src/ui/app/tests.rs b/src/tui/src/ui/app/tests.rs index e3f741a0..27b03c19 100644 --- a/src/tui/src/ui/app/tests.rs +++ b/src/tui/src/ui/app/tests.rs @@ -24,19 +24,20 @@ fn app() -> App { fn app_with_running_task() -> App { let rt = MockRuntime::demo(); - rt.script_event(crate::ui::events::TuiEvent::TaskStart { - task_id: "task-1".into(), - instruction: "Continue the auth refactor.".into(), - depth: 2, - agent_id: Some("dev-1".into()), - contract: None, - }); let loaded = { let mut loaded = LoadedConfig::defaults("medulla.tui.json".into()); loaded.config.tinyplace = Some(medulla::config::TinyplaceConfig::default()); loaded }; - App::new(Arc::new(rt), loaded) + let mut app = App::new(Arc::new(rt), loaded); + app.snapshot.events.retain(|envelope| { + !matches!( + &envelope.event, + crate::ui::events::TuiEvent::TaskComplete { digest } + if digest.task_id == "task-1" + ) + }); + app } /// The index of the tab named `name`. Looked up rather than written down: the