From 3f37a6cf34559c3d0c42d6c91123e3103b197ddc Mon Sep 17 00:00:00 2001 From: Huskies Agent Date: Sat, 18 Jul 2026 10:02:34 +0000 Subject: [PATCH] huskies: merge 1210 story Gateway-side "digging in" notification for long tool-only turns --- .../src/chat/transport/matrix/bot/context.rs | 5 + .../matrix/bot/messages/handle_message.rs | 210 ++++++++++++++++++ .../matrix/bot/messages/on_room_message.rs | 1 + server/src/chat/transport/matrix/bot/run.rs | 1 + .../transport/matrix/config/tests_core.rs | 43 ++++ .../src/chat/transport/matrix/config/types.rs | 12 + server/src/chat/transport/matrix/health.rs | 1 + server/src/chat/transport/matrix/projects.rs | 2 + 8 files changed, 275 insertions(+) diff --git a/server/src/chat/transport/matrix/bot/context.rs b/server/src/chat/transport/matrix/bot/context.rs index 1f2b3656..20c7398b 100644 --- a/server/src/chat/transport/matrix/bot/context.rs +++ b/server/src/chat/transport/matrix/bot/context.rs @@ -130,6 +130,10 @@ pub struct BotContext { /// Minimum seconds between repeated `compact` suggestions for the same /// room. From `bot.toml`'s `compact_suggest_cooldown_secs`. pub compact_suggest_cooldown_secs: i64, + /// Seconds a turn may spend on tool calls before emitting any + /// user-facing text before the bot posts a "digging in" notice. From + /// `bot.toml`'s `digging_in_threshold_secs`. + pub digging_in_threshold_secs: u64, } impl BotContext { @@ -356,6 +360,7 @@ mod tests { compact_seed_max_bytes: 8_000, cache_read_suggest_threshold: 50_000, compact_suggest_cooldown_secs: 3_600, + digging_in_threshold_secs: 15, } } 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 d0881662..3910f913 100644 --- a/server/src/chat/transport/matrix/bot/messages/handle_message.rs +++ b/server/src/chat/transport/matrix/bot/messages/handle_message.rs @@ -1,12 +1,14 @@ //! Matrix handle_message — runs the LLM turn for a verified incoming message and //! streams the assistant reply back to the room. +use crate::chat::ChatTransport; use crate::chat::util::drain_complete_paragraphs; use crate::llm::providers::claude_code::{ClaudeCodeProvider, ClaudeCodeResult}; use crate::slog; use matrix_sdk::ruma::OwnedRoomId; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Duration; use tokio::sync::watch; use super::super::context::BotContext; @@ -15,6 +17,33 @@ use super::super::history::{ConversationEntry, ConversationRole, save_history}; use super::format_user_prompt; +/// Text posted to the room by [`spawn_digging_in_watcher`] when a turn runs +/// long without emitting any user-facing text. +const DIGGING_IN_MESSAGE: &str = "Still digging in — this turn is taking a bit longer than usual."; + +/// Spawns a background watcher that posts a single "digging in" notice to +/// `room_id` if `threshold` elapses before `sent_any_text` becomes `true`. +/// +/// Callers must abort the returned [`tokio::task::JoinHandle`] once the turn +/// completes so a turn that finishes under the threshold — with or without +/// text — never triggers the notice after the fact. +pub(in crate::chat::transport::matrix::bot) fn spawn_digging_in_watcher( + transport: Arc, + room_id: String, + sent_any_text: Arc, + threshold: Duration, +) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + tokio::time::sleep(threshold).await; + if !sent_any_text.load(Ordering::Relaxed) { + let html = markdown_to_html(DIGGING_IN_MESSAGE); + let _ = transport + .send_message(&room_id, DIGGING_IN_MESSAGE, &html) + .await; + } + }) +} + pub(in crate::chat::transport::matrix::bot) async fn handle_message( room_id_str: String, room_id: OwnedRoomId, @@ -135,10 +164,21 @@ pub(in crate::chat::transport::matrix::bot) async fn handle_message( ); tokio::pin!(chat_fut); + // Notify the room if this turn spends longer than the configured + // threshold on tool calls before emitting any user-facing text, so a + // long silent turn doesn't look hung. Aborted below once the turn ends. + let digging_in_task = spawn_digging_in_watcher( + Arc::clone(&ctx.transport), + room_id_str.clone(), + Arc::clone(&sent_any_chunk), + Duration::from_secs(ctx.digging_in_threshold_secs), + ); + // Permission requests are handled by the persistent permission_listener // task spawned at bot startup (story 884) — they no longer route through // per-message handlers. Just await chat_fut. let result = (&mut chat_fut).await; + digging_in_task.abort(); // Flush any remaining text that didn't end with a paragraph boundary. let remaining = buffer.lock().unwrap().trim().to_string(); @@ -275,3 +315,173 @@ pub(in crate::chat::transport::matrix::bot) async fn handle_message( // --------------------------------------------------------------------------- // Tests // --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + use crate::chat::MessageId; + use async_trait::async_trait; + + /// Captures every message sent through it, for assertion in tests. + struct CapturingTransport { + sent: std::sync::Mutex>, + } + + impl CapturingTransport { + fn new() -> Self { + Self { + sent: std::sync::Mutex::new(Vec::new()), + } + } + + fn sent_count(&self) -> usize { + self.sent.lock().unwrap().len() + } + + fn last_message(&self) -> Option { + self.sent + .lock() + .unwrap() + .last() + .map(|(_, plain)| plain.clone()) + } + } + + #[async_trait] + impl ChatTransport for CapturingTransport { + async fn send_message( + &self, + room_id: &str, + plain: &str, + _html: &str, + ) -> Result { + self.sent + .lock() + .unwrap() + .push((room_id.to_string(), plain.to_string())); + Ok("msg-id".to_string()) + } + + async fn edit_message( + &self, + _room_id: &str, + _original_message_id: &str, + _plain: &str, + _html: &str, + ) -> Result<(), String> { + Ok(()) + } + + async fn send_typing(&self, _room_id: &str, _typing: bool) -> Result<(), String> { + Ok(()) + } + } + + /// AC 1: a turn that runs longer than the threshold without sending any + /// text gets exactly one "digging in" notice. + #[tokio::test] + async fn digging_in_fires_after_threshold_when_no_text_sent() { + let transport = Arc::new(CapturingTransport::new()); + let sent_any_text = Arc::new(AtomicBool::new(false)); + let handle = spawn_digging_in_watcher( + transport.clone() as Arc, + "!room:example.com".to_string(), + Arc::clone(&sent_any_text), + Duration::from_millis(30), + ); + handle.await.unwrap(); + assert_eq!(transport.sent_count(), 1); + assert!( + transport + .last_message() + .unwrap() + .to_lowercase() + .contains("digging in"), + "notice should mention 'digging in'" + ); + } + + /// AC 2: if text is sent before the threshold elapses, the watcher must + /// not post anything. + #[tokio::test] + async fn digging_in_does_not_fire_when_text_sent_before_threshold() { + let transport = Arc::new(CapturingTransport::new()); + let sent_any_text = Arc::new(AtomicBool::new(false)); + let handle = spawn_digging_in_watcher( + transport.clone() as Arc, + "!room:example.com".to_string(), + Arc::clone(&sent_any_text), + Duration::from_millis(30), + ); + sent_any_text.store(true, Ordering::Relaxed); + handle.await.unwrap(); + assert_eq!(transport.sent_count(), 0); + } + + /// AC 2: a fast turn that completes (and is aborted by its caller) + /// before the threshold elapses must not post anything, even if it + /// never sent any text either. + #[tokio::test] + async fn digging_in_does_not_fire_when_aborted_before_threshold() { + let transport = Arc::new(CapturingTransport::new()); + let sent_any_text = Arc::new(AtomicBool::new(false)); + let handle = spawn_digging_in_watcher( + transport.clone() as Arc, + "!room:example.com".to_string(), + Arc::clone(&sent_any_text), + Duration::from_millis(200), + ); + handle.abort(); + tokio::time::sleep(Duration::from_millis(250)).await; + assert_eq!(transport.sent_count(), 0); + } + + /// AC 3: the notice fires at most once per turn — even waiting well past + /// the threshold never produces a second message. + #[tokio::test] + async fn digging_in_fires_at_most_once_even_after_extra_wait() { + let transport = Arc::new(CapturingTransport::new()); + let sent_any_text = Arc::new(AtomicBool::new(false)); + let handle = spawn_digging_in_watcher( + transport.clone() as Arc, + "!room:example.com".to_string(), + Arc::clone(&sent_any_text), + Duration::from_millis(20), + ); + handle.await.unwrap(); + tokio::time::sleep(Duration::from_millis(100)).await; + assert_eq!(transport.sent_count(), 1); + } + + /// AC 5: state is fresh per call — spawning a second, independent + /// watcher (as a new turn would) does not inherit any state from a + /// prior watcher's `sent_any_text` flag. + #[tokio::test] + async fn digging_in_state_does_not_leak_across_turns() { + let transport = Arc::new(CapturingTransport::new()); + + // First "turn": text sent before threshold, so no notice. + let first_sent_any_text = Arc::new(AtomicBool::new(false)); + let first_handle = spawn_digging_in_watcher( + transport.clone() as Arc, + "!room:example.com".to_string(), + Arc::clone(&first_sent_any_text), + Duration::from_millis(30), + ); + first_sent_any_text.store(true, Ordering::Relaxed); + first_handle.await.unwrap(); + assert_eq!(transport.sent_count(), 0); + + // Second "turn": fresh flag, no text sent — must fire independently + // of the first turn's outcome. + let second_sent_any_text = Arc::new(AtomicBool::new(false)); + let second_handle = spawn_digging_in_watcher( + transport.clone() as Arc, + "!room:example.com".to_string(), + Arc::clone(&second_sent_any_text), + Duration::from_millis(30), + ); + second_handle.await.unwrap(); + assert_eq!(transport.sent_count(), 1); + } +} 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 0d3659e0..df98013c 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 @@ -1861,6 +1861,7 @@ mod tests { compact_seed_max_bytes: 8_000, cache_read_suggest_threshold: 50_000, compact_suggest_cooldown_secs: 3_600, + digging_in_threshold_secs: 15, } } diff --git a/server/src/chat/transport/matrix/bot/run.rs b/server/src/chat/transport/matrix/bot/run.rs index 7ca2295d..f598ffb4 100644 --- a/server/src/chat/transport/matrix/bot/run.rs +++ b/server/src/chat/transport/matrix/bot/run.rs @@ -337,6 +337,7 @@ pub async fn run_bot( compact_seed_max_bytes: config.compact_seed_max_bytes, cache_read_suggest_threshold: config.cache_read_suggest_threshold, compact_suggest_cooldown_secs: config.compact_suggest_cooldown_secs, + digging_in_threshold_secs: config.digging_in_threshold_secs, }; slog!( diff --git a/server/src/chat/transport/matrix/config/tests_core.rs b/server/src/chat/transport/matrix/config/tests_core.rs index 53840695..d13b48ee 100644 --- a/server/src/chat/transport/matrix/config/tests_core.rs +++ b/server/src/chat/transport/matrix/config/tests_core.rs @@ -172,6 +172,49 @@ history_size = 50 assert_eq!(config.history_size, 50); } +/// AC4: `digging_in_threshold_secs` defaults to 15 when unset in bot.toml. +#[test] +fn load_uses_default_digging_in_threshold_secs() { + let tmp = tempfile::tempdir().unwrap(); + let sk = tmp.path().join(".huskies"); + fs::create_dir_all(&sk).unwrap(); + fs::write( + sk.join("bot.toml"), + r#" +homeserver = "https://matrix.example.com" +username = "@bot:example.com" +password = "secret" +room_ids = ["!abc:example.com"] +enabled = true +"#, + ) + .unwrap(); + let config = BotConfig::load(tmp.path()).unwrap(); + assert_eq!(config.digging_in_threshold_secs, 15); +} + +/// AC4: `digging_in_threshold_secs` can be overridden in bot.toml. +#[test] +fn load_respects_custom_digging_in_threshold_secs() { + let tmp = tempfile::tempdir().unwrap(); + let sk = tmp.path().join(".huskies"); + fs::create_dir_all(&sk).unwrap(); + fs::write( + sk.join("bot.toml"), + r#" +homeserver = "https://matrix.example.com" +username = "@bot:example.com" +password = "secret" +room_ids = ["!abc:example.com"] +enabled = true +digging_in_threshold_secs = 30 +"#, + ) + .unwrap(); + let config = BotConfig::load(tmp.path()).unwrap(); + assert_eq!(config.digging_in_threshold_secs, 30); +} + #[test] fn load_reads_display_name() { let tmp = tempfile::tempdir().unwrap(); diff --git a/server/src/chat/transport/matrix/config/types.rs b/server/src/chat/transport/matrix/config/types.rs index c2cdba1d..2ea7e755 100644 --- a/server/src/chat/transport/matrix/config/types.rs +++ b/server/src/chat/transport/matrix/config/types.rs @@ -31,6 +31,12 @@ pub(super) fn default_compact_suggest_cooldown_secs() -> i64 { 3_600 } +/// Default threshold (seconds) a turn may spend on tool calls before +/// emitting any user-facing text before the bot posts a "digging in" notice. +pub(super) fn default_digging_in_threshold_secs() -> u64 { + 15 +} + pub(super) fn default_transport() -> String { "matrix".to_string() } @@ -223,4 +229,10 @@ pub struct BotConfig { /// room, so a busy room isn't spammed every turn. Defaults to 3600 (1h). #[serde(default = "default_compact_suggest_cooldown_secs")] pub compact_suggest_cooldown_secs: i64, + + /// Seconds a turn may spend on tool calls before emitting any + /// user-facing text before the bot posts a "digging in" notice to the + /// room, so a silent long-running turn doesn't look hung. Defaults to 15. + #[serde(default = "default_digging_in_threshold_secs")] + pub digging_in_threshold_secs: u64, } diff --git a/server/src/chat/transport/matrix/health.rs b/server/src/chat/transport/matrix/health.rs index 10193cc2..3a4a2cd9 100644 --- a/server/src/chat/transport/matrix/health.rs +++ b/server/src/chat/transport/matrix/health.rs @@ -949,6 +949,7 @@ mod tests { compact_seed_max_bytes: 8_000, cache_read_suggest_threshold: 50_000, compact_suggest_cooldown_secs: 3_600, + digging_in_threshold_secs: 15, } } } diff --git a/server/src/chat/transport/matrix/projects.rs b/server/src/chat/transport/matrix/projects.rs index 52e84795..d73d9224 100644 --- a/server/src/chat/transport/matrix/projects.rs +++ b/server/src/chat/transport/matrix/projects.rs @@ -102,6 +102,7 @@ mod tests { compact_seed_max_bytes: 8_000, cache_read_suggest_threshold: 50_000, compact_suggest_cooldown_secs: 3_600, + digging_in_threshold_secs: 15, }; run_projects_list(&ctx).await } @@ -224,6 +225,7 @@ mod tests { compact_seed_max_bytes: 8_000, cache_read_suggest_threshold: 50_000, compact_suggest_cooldown_secs: 3_600, + digging_in_threshold_secs: 15, }; let response = run_projects_list(&ctx).await; assert!(