From cebe9e273709614d6415bca46c2b85895ab168bc Mon Sep 17 00:00:00 2001 From: Huskies Agent Date: Sat, 18 Jul 2026 11:34:19 +0000 Subject: [PATCH] huskies: merge 1213 story Chat "stop" command that immediately aborts the in-flight LLM turn --- server/src/chat/commands/mod.rs | 15 ++ server/src/chat/dispatcher.rs | 89 ++++++++-- .../matrix/bot/messages/handle_message.rs | 16 +- .../matrix/bot/messages/on_room_message.rs | 168 +++++++++++++++--- server/src/chat/transport/matrix/mod.rs | 3 + server/src/chat/transport/matrix/stop.rs | 103 +++++++++++ server/src/llm/providers/claude_code/mod.rs | 8 +- 7 files changed, 367 insertions(+), 35 deletions(-) create mode 100644 server/src/chat/transport/matrix/stop.rs diff --git a/server/src/chat/commands/mod.rs b/server/src/chat/commands/mod.rs index 308757db..2ada8b70 100644 --- a/server/src/chat/commands/mod.rs +++ b/server/src/chat/commands/mod.rs @@ -229,6 +229,11 @@ pub fn commands() -> &'static [BotCommand] { description: "Distill the session transcript into a seed, then reset (unlike `reset`, keeps distilled context for the next message)", handler: handle_compact_fallback, }, + BotCommand { + name: "stop", + description: "Cancel the in-flight LLM turn for this room (synonyms: `halt`, `abort`)", + handler: handle_stop_fallback, + }, BotCommand { name: "timer", description: "Schedule a deferred agent start: `timer `, `timer list`, `timer cancel `", @@ -439,6 +444,16 @@ fn handle_compact_fallback(_ctx: &CommandContext) -> Option { None } +/// Fallback handler for the `stop` command when it is not intercepted by the +/// async handler in `on_room_message`. In practice this is never called — +/// stop is detected and handled before `try_handle_command` is invoked. +/// The entry exists in the registry only so `help` lists it. +/// +/// Returns `None` to prevent the LLM from receiving "stop" as a prompt. +fn handle_stop_fallback(_ctx: &CommandContext) -> Option { + None +} + /// Fallback handler for the `cleanup_worktrees` command when it is not /// intercepted by the async handler in `on_room_message`. In practice this is /// never called — cleanup_worktrees is detected and handled before diff --git a/server/src/chat/dispatcher.rs b/server/src/chat/dispatcher.rs index 27f4c8b9..cf9c708a 100644 --- a/server/src/chat/dispatcher.rs +++ b/server/src/chat/dispatcher.rs @@ -19,6 +19,7 @@ use crate::slog; use std::collections::HashMap; use std::pin::Pin; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; use tokio::sync::{mpsc, watch}; @@ -47,6 +48,20 @@ enum SessionMsg { struct SessionHandle { tx: mpsc::UnboundedSender, + /// `true` whenever the session task is coalescing, running, or draining a + /// batch — i.e. there is something for [`ChatDispatcher::stop`] to cancel. + /// `false` while the task is idle in Phase 1, waiting for a message. + active: Arc, +} + +/// Result of a [`ChatDispatcher::stop`] call. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum StopOutcome { + /// A turn was actively coalescing, running, or draining pending messages + /// for the session, and a cancellation was sent. + Cancelled, + /// The session doesn't exist, or exists but has no turn in flight. + NothingRunning, } /// Coalescing, serialising dispatcher for chat-to-LLM message routing. @@ -79,8 +94,14 @@ impl ChatDispatcher { let coalesce_ms = self.coalesce_ms; let handle = guard.entry(session_key.clone()).or_insert_with(|| { let (tx, rx) = mpsc::unbounded_channel(); - tokio::spawn(session_task(session_key.clone(), rx, coalesce_ms)); - SessionHandle { tx } + let active = Arc::new(AtomicBool::new(false)); + tokio::spawn(session_task( + session_key.clone(), + rx, + coalesce_ms, + Arc::clone(&active), + )); + SessionHandle { tx, active } }); let _ = handle.tx.send(SessionMsg::UserMessage { text: message, @@ -90,15 +111,19 @@ impl ChatDispatcher { /// Stop the active LLM run for `session_key` and clear its pending queue. /// - /// Returns `true` if the session existed (whether or not anything was - /// actually running), `false` if no session for that key has been created. - pub fn stop(&self, session_key: &str) -> bool { + /// Returns [`StopOutcome::Cancelled`] and sends the cancellation only when + /// a turn is actually coalescing, running, or draining pending messages; + /// otherwise returns [`StopOutcome::NothingRunning`] without sending + /// anything (covers both "no session was ever created" and "session + /// exists but is idle"). + pub fn stop(&self, session_key: &str) -> StopOutcome { let guard = self.sessions.lock().unwrap(); - if let Some(handle) = guard.get(session_key) { - let _ = handle.tx.send(SessionMsg::Stop); - true - } else { - false + match guard.get(session_key) { + Some(handle) if handle.active.load(Ordering::SeqCst) => { + let _ = handle.tx.send(SessionMsg::Stop); + StopOutcome::Cancelled + } + _ => StopOutcome::NothingRunning, } } } @@ -119,6 +144,7 @@ async fn session_task( session_key: String, mut rx: mpsc::UnboundedReceiver, coalesce_ms: u64, + active: Arc, ) { let coalesce_dur = Duration::from_millis(coalesce_ms); @@ -131,6 +157,9 @@ async fn session_task( Some(SessionMsg::UserMessage { text, factory }) => break (text, factory), } }; + // From here until we loop back to Phase 1, there is something in + // flight (coalescing, running, or draining) for `stop()` to cancel. + active.store(true, Ordering::SeqCst); // ── Phase 2: coalesce window (debounce) ────────────────────────────── let mut batch: Vec = vec![first_text]; @@ -160,6 +189,7 @@ async fn session_task( } if batch.is_empty() { + active.store(false, Ordering::SeqCst); continue; // Stop received during coalesce — restart } @@ -209,6 +239,7 @@ async fn session_task( } if stopped || pending_texts.is_empty() { + active.store(false, Ordering::SeqCst); break; // back to Phase 1 } @@ -351,7 +382,12 @@ mod tests { ); // Stop immediately. - dispatcher.stop(&session); + let outcome = dispatcher.stop(&session); + assert_eq!( + outcome, + StopOutcome::Cancelled, + "a run was active, so stop must report Cancelled" + ); // Wait longer than the run would have taken if not stopped. tokio::time::sleep(Duration::from_millis(700)).await; @@ -364,4 +400,35 @@ mod tests { "stop should discard pending; got {count} spawns" ); } + + /// Stopping a session key that was never submitted reports NothingRunning + /// rather than silently no-opping. + #[tokio::test] + async fn stop_on_unknown_session_reports_nothing_running() { + let dispatcher = ChatDispatcher::new(30); + assert_eq!( + dispatcher.stop("never-seen-room"), + StopOutcome::NothingRunning + ); + } + + /// Stopping a session that exists but has no active turn (its one run + /// already completed) reports NothingRunning, not Cancelled. + #[tokio::test] + async fn stop_on_idle_session_reports_nothing_running() { + let spawn_count = Arc::new(AtomicUsize::new(0)); + let dispatcher = ChatDispatcher::new(20); + let session = "room4".to_string(); + + dispatcher.submit( + session.clone(), + "hello".to_string(), + make_factory(Arc::clone(&spawn_count), 20), + ); + + // Wait long enough for the coalesce window and the run to finish. + tokio::time::sleep(Duration::from_millis(200)).await; + + assert_eq!(dispatcher.stop(&session), StopOutcome::NothingRunning); + } } diff --git a/server/src/chat/transport/matrix/bot/messages/handle_message.rs b/server/src/chat/transport/matrix/bot/messages/handle_message.rs index 3910f913..e82dfef6 100644 --- a/server/src/chat/transport/matrix/bot/messages/handle_message.rs +++ b/server/src/chat/transport/matrix/bot/messages/handle_message.rs @@ -3,7 +3,7 @@ use crate::chat::ChatTransport; use crate::chat::util::drain_complete_paragraphs; -use crate::llm::providers::claude_code::{ClaudeCodeProvider, ClaudeCodeResult}; +use crate::llm::providers::claude_code::{CANCELLED, ClaudeCodeProvider, ClaudeCodeResult}; use crate::slog; use matrix_sdk::ruma::OwnedRoomId; use std::sync::Arc; @@ -184,6 +184,7 @@ pub(in crate::chat::transport::matrix::bot) async fn handle_message( let remaining = buffer.lock().unwrap().trim().to_string(); let did_send_any = sent_any_chunk.load(Ordering::Relaxed); + let mut was_cancelled = false; let (assistant_reply, new_session_id, turn_usage) = match result { Ok(ClaudeCodeResult { messages, @@ -213,6 +214,15 @@ pub(in crate::chat::transport::matrix::bot) async fn handle_message( slog!("[matrix-bot] session_id from chat_stream: {:?}", session_id); (reply, session_id, usage) } + Err(e) if e == CANCELLED => { + // A user-initiated "stop" — not a crash. The stop handler already + // sent a confirmation, so don't post a second message here, and + // don't run the crash-recovery path below (which would otherwise + // clear the room's session_id as if the session were poisoned). + slog!("[matrix-bot] LLM turn cancelled via stop for room {room_id}"); + was_cancelled = true; + (String::new(), None, None) + } Err(e) => { slog!("[matrix-bot] LLM error: {e}"); let err_msg = if let Some(url) = crate::llm::oauth::extract_login_url_from_error(&e) { @@ -230,6 +240,10 @@ pub(in crate::chat::transport::matrix::bot) async fn handle_message( drop(msg_tx); let _ = post_task.await; + if was_cancelled { + return; + } + // Record this exchange in the per-room conversation history and persist // the session ID so the next turn resumes with structured API messages. let mut compact_suggestion: Option = None; diff --git a/server/src/chat/transport/matrix/bot/messages/on_room_message.rs b/server/src/chat/transport/matrix/bot/messages/on_room_message.rs index df98013c..bea7b867 100644 --- a/server/src/chat/transport/matrix/bot/messages/on_room_message.rs +++ b/server/src/chat/transport/matrix/bot/messages/on_room_message.rs @@ -295,6 +295,47 @@ async fn try_handle_compact_command( true } +/// Attempt to handle an addressed message as a bare `stop`/`halt`/`abort` +/// command. +/// +/// Returns `true` when `user_message` was recognised as a stop command and +/// handled: the in-flight turn (if any) was cancelled via +/// [`crate::chat::dispatcher::ChatDispatcher::stop`] and a confirmation was +/// sent via `ctx.transport`. Returns `false` so the caller falls through to +/// the dispatcher submit path — this must run before `chat_dispatcher.submit` +/// so a bare "stop" is never itself queued as a prompt. +async fn try_handle_stop_command( + ctx: &BotContext, + sender: &str, + user_message: &str, + room_id_str: &str, +) -> bool { + if super::super::super::stop::extract_stop_command( + user_message, + &ctx.services.bot_name, + ctx.matrix_user_id.as_str(), + ) + .is_none() + { + return false; + } + slog!("[matrix-bot] stop command from {sender} for session {room_id_str}"); + let response = match ctx.services.chat_dispatcher.stop(room_id_str) { + crate::chat::dispatcher::StopOutcome::Cancelled => "Stopped.", + crate::chat::dispatcher::StopOutcome::NothingRunning => "Nothing is currently running.", + }; + let html = markdown_to_html(response); + if let Ok(msg_id) = ctx + .transport + .send_message(room_id_str, response, &html) + .await + && let Ok(event_id) = msg_id.parse() + { + ctx.bot_sent_event_ids.lock().await.insert(event_id); + } + true +} + pub(in crate::chat::transport::matrix::bot) async fn on_room_message( ev: OriginalSyncRoomMessageEvent, room: Room, @@ -490,6 +531,13 @@ pub(in crate::chat::transport::matrix::bot) async fn on_room_message( // entry silently swallows the command (bug: compact no-op in // gateway mode, distinct from the 1192/1205 registry-ordering bug). "compact", + // `stop` is a gateway-local session command (sibling of `compact`): + // it cancels THIS gateway's in-flight LLM turn via the local + // `chat_dispatcher`. It must be listed here for the same reason + // as `compact` above — once registered in `commands()` (so `help` + // lists it), the gateway proxy would otherwise forward it to the + // active project's sled instead of executing it locally. + "stop", "switch", "all_status", "new", @@ -1403,27 +1451,10 @@ pub(in crate::chat::transport::matrix::bot) async fn on_room_message( return; } - // "stop" — cancel the running LLM turn for this session and clear pending queue. - { - let stripped = crate::chat::util::strip_bot_mention( - &user_message, - &ctx.services.bot_name, - ctx.matrix_user_id.as_str(), - ) - .trim() - .to_ascii_lowercase(); - if stripped == "stop" { - slog!("[matrix-bot] stop command from {sender} for session {room_id_str}"); - ctx.services.chat_dispatcher.stop(&room_id_str); - let msg = "Stopped."; - let html = markdown_to_html(msg); - if let Ok(msg_id) = ctx.transport.send_message(&room_id_str, msg, &html).await - && let Ok(event_id) = msg_id.parse() - { - ctx.bot_sent_event_ids.lock().await.insert(event_id); - } - return; - } + // "stop"/"halt"/"abort" — cancel the running LLM turn for this session + // and clear its pending queue. + if try_handle_stop_command(&ctx, &sender, &user_message, &room_id_str).await { + return; } // Hand the message to the protocol-agnostic dispatcher instead of spawning @@ -1465,7 +1496,7 @@ pub(in crate::chat::transport::matrix::bot) async fn on_room_message( mod tests { use super::{ eval_gateway_overview_command, eval_gateway_status_command, eval_switch_command, - try_handle_compact_command, + try_handle_compact_command, try_handle_stop_command, }; use crate::chat::{ChatTransport, MessageId}; use crate::service::gateway::config::ProjectEntry; @@ -1987,4 +2018,97 @@ mod tests { "no reply should be sent for a message that isn't compact" ); } + + // ── stop (story 1213) ────────────────────────────────────────────── + + /// AC1: a bare "stop" cancels an active run and replies "Stopped.". + #[tokio::test] + async fn stop_command_cancels_active_run_and_confirms() { + use std::sync::Arc; + + let room_id: matrix_sdk::ruma::OwnedRoomId = "!test:example.com".parse().unwrap(); + let project_root_dir = tempfile::tempdir().unwrap(); + let services = crate::services::Services::new_test( + project_root_dir.path().to_path_buf(), + "Huskies".to_string(), + ); + let transport = Arc::new(CapturingTransport::new()); + let ctx = make_test_ctx(services, transport.clone()); + + // Start a long-running "turn" on the dispatcher for this room. + let factory: crate::chat::dispatcher::SpawnFn = Arc::new(|_prompt, _cancel_rx| { + Box::pin(async move { + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + }) + }); + ctx.services.chat_dispatcher.submit( + room_id.to_string(), + "long-running".to_string(), + factory, + ); + // Let the coalesce window fire so the run is actually active. + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + + let handled = + try_handle_stop_command(&ctx, "@alice:example.com", "stop", room_id.as_str()).await; + + assert!(handled, "bare stop must be recognized and handled"); + let sent = transport.sent.lock().unwrap().clone(); + assert_eq!(sent.len(), 1, "exactly one reply must be sent"); + assert_eq!(sent[0].1, "Stopped."); + } + + /// AC6: stop with nothing running replies that nothing is running, + /// instead of silently no-opping. + #[tokio::test] + async fn stop_command_with_nothing_running_reports_nothing_running() { + use std::sync::Arc; + + let room_id: matrix_sdk::ruma::OwnedRoomId = "!test:example.com".parse().unwrap(); + let project_root_dir = tempfile::tempdir().unwrap(); + let services = crate::services::Services::new_test( + project_root_dir.path().to_path_buf(), + "Huskies".to_string(), + ); + let transport = Arc::new(CapturingTransport::new()); + let ctx = make_test_ctx(services, transport.clone()); + + let handled = + try_handle_stop_command(&ctx, "@alice:example.com", "stop", room_id.as_str()).await; + + assert!(handled, "bare stop must be recognized and handled"); + let sent = transport.sent.lock().unwrap().clone(); + assert_eq!(sent.len(), 1, "exactly one reply must be sent"); + assert_eq!(sent[0].1, "Nothing is currently running."); + } + + /// AC4: a non-bare message that merely contains the word "stop" must + /// fall through untouched, never treated as an abort. + #[tokio::test] + async fn stop_with_trailing_text_falls_through() { + use std::sync::Arc; + + let room_id: matrix_sdk::ruma::OwnedRoomId = "!test:example.com".parse().unwrap(); + let project_root_dir = tempfile::tempdir().unwrap(); + let services = crate::services::Services::new_test( + project_root_dir.path().to_path_buf(), + "Huskies".to_string(), + ); + let transport = Arc::new(CapturingTransport::new()); + let ctx = make_test_ctx(services, transport.clone()); + + let handled = try_handle_stop_command( + &ctx, + "@alice:example.com", + "stop the deployment and redeploy", + room_id.as_str(), + ) + .await; + + assert!(!handled, "non-bare 'stop' text must fall through"); + assert!( + transport.sent.lock().unwrap().is_empty(), + "no reply should be sent for a message that isn't a bare stop" + ); + } } diff --git a/server/src/chat/transport/matrix/mod.rs b/server/src/chat/transport/matrix/mod.rs index b25ed418..4b9e3dcb 100644 --- a/server/src/chat/transport/matrix/mod.rs +++ b/server/src/chat/transport/matrix/mod.rs @@ -52,6 +52,9 @@ pub mod rmtree; pub mod sled_upgrade; /// Start command — handles `!start` bot commands to launch agents on stories. pub mod start; +/// Stop/halt/abort command — cancels the in-flight LLM turn for this room +/// and clears any coalesced/pending messages. +pub mod stop; /// Matrix `ChatTransport` implementation wrapping the Matrix SDK client. pub mod transport_impl; diff --git a/server/src/chat/transport/matrix/stop.rs b/server/src/chat/transport/matrix/stop.rs new file mode 100644 index 00000000..673c95f2 --- /dev/null +++ b/server/src/chat/transport/matrix/stop.rs @@ -0,0 +1,103 @@ +//! Stop command: cancel the in-flight LLM turn for the current room's chat +//! session. +//! +//! `{bot_name} stop` (or `halt`/`abort`) cancels the actively-running Claude +//! Code turn for this room via [`crate::chat::dispatcher::ChatDispatcher`] +//! and clears any coalesced/pending messages. Unlike [`super::reset`] and +//! [`crate::chat::compact`], which match on the first word and ignore +//! trailing text, this requires an EXACT match (after mention-stripping) — +//! an instruction like "stop X and do Y" must not be misread as an abort. + +use crate::chat::util::strip_bot_mention; + +/// A parsed stop/halt/abort command. +#[derive(Debug, PartialEq)] +pub struct StopCommand; + +/// Parse a stop command from a raw message body. +/// +/// Strips the bot mention prefix and requires the *entire* remaining, +/// trimmed message to equal `stop`, `halt`, or `abort` (case-insensitive). +/// Returns `None` for anything else, including a recognised word followed by +/// trailing text (e.g. "stop the deployment"). +pub fn extract_stop_command( + message: &str, + bot_name: &str, + bot_user_id: &str, +) -> Option { + let stripped = strip_bot_mention(message, bot_name, bot_user_id); + let trimmed = stripped.trim(); + + if trimmed.eq_ignore_ascii_case("stop") + || trimmed.eq_ignore_ascii_case("halt") + || trimmed.eq_ignore_ascii_case("abort") + { + Some(StopCommand) + } else { + None + } +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn extract_bare_stop() { + let cmd = extract_stop_command("stop", "Timmy", "@timmy:home.local"); + assert_eq!(cmd, Some(StopCommand)); + } + + #[test] + fn extract_with_display_name() { + let cmd = extract_stop_command("Timmy stop", "Timmy", "@timmy:home.local"); + assert_eq!(cmd, Some(StopCommand)); + } + + #[test] + fn extract_with_full_user_id() { + let cmd = extract_stop_command("@timmy:home.local stop", "Timmy", "@timmy:home.local"); + assert_eq!(cmd, Some(StopCommand)); + } + + #[test] + fn extract_halt_synonym() { + let cmd = extract_stop_command("Timmy halt", "Timmy", "@timmy:home.local"); + assert_eq!(cmd, Some(StopCommand)); + } + + #[test] + fn extract_abort_synonym() { + let cmd = extract_stop_command("abort", "Timmy", "@timmy:home.local"); + assert_eq!(cmd, Some(StopCommand)); + } + + #[test] + fn extract_case_insensitive() { + let cmd = extract_stop_command("Timmy STOP", "Timmy", "@timmy:home.local"); + assert_eq!(cmd, Some(StopCommand)); + } + + #[test] + fn extract_non_stop_returns_none() { + let cmd = extract_stop_command("Timmy help", "Timmy", "@timmy:home.local"); + assert_eq!(cmd, None); + } + + #[test] + fn extract_with_trailing_text_returns_none() { + // "stop X and do Y" is an instruction, not an abort request. + let cmd = extract_stop_command("Timmy stop X and do Y", "Timmy", "@timmy:home.local"); + assert_eq!(cmd, None); + } + + #[test] + fn extract_stop_followed_by_words_returns_none() { + let cmd = extract_stop_command("stop the deployment please", "Timmy", "@timmy:home.local"); + assert_eq!(cmd, None); + } +} diff --git a/server/src/llm/providers/claude_code/mod.rs b/server/src/llm/providers/claude_code/mod.rs index ce4902ee..2a05b5cb 100644 --- a/server/src/llm/providers/claude_code/mod.rs +++ b/server/src/llm/providers/claude_code/mod.rs @@ -41,6 +41,12 @@ mod parse; use events::process_json_event; +/// Sentinel error string `chat_stream` returns when a turn was cancelled via +/// `cancel_rx` (e.g. a user-initiated chat "stop"), as opposed to a genuine +/// Claude Code crash. Callers must check for this before running crash- +/// recovery logic (e.g. clearing a stored session_id). +pub const CANCELLED: &str = "Cancelled"; + /// Orchestrates Claude Code CLI sessions via a PTY for streaming agent chat. pub struct ClaudeCodeProvider; @@ -354,7 +360,7 @@ fn run_pty_session( let _ = child.kill(); let _ = child.wait(); let _ = reader_handle.join(); - return Err("Cancelled".to_string()); + return Err(CANCELLED.to_string()); } match line_rx.recv_timeout(std::time::Duration::from_millis(500)) {