huskies: merge 1210 story Gateway-side "digging in" notification for long tool-only turns
This commit is contained in:
@@ -130,6 +130,10 @@ pub struct BotContext {
|
|||||||
/// Minimum seconds between repeated `compact` suggestions for the same
|
/// Minimum seconds between repeated `compact` suggestions for the same
|
||||||
/// room. From `bot.toml`'s `compact_suggest_cooldown_secs`.
|
/// room. From `bot.toml`'s `compact_suggest_cooldown_secs`.
|
||||||
pub compact_suggest_cooldown_secs: i64,
|
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 {
|
impl BotContext {
|
||||||
@@ -356,6 +360,7 @@ mod tests {
|
|||||||
compact_seed_max_bytes: 8_000,
|
compact_seed_max_bytes: 8_000,
|
||||||
cache_read_suggest_threshold: 50_000,
|
cache_read_suggest_threshold: 50_000,
|
||||||
compact_suggest_cooldown_secs: 3_600,
|
compact_suggest_cooldown_secs: 3_600,
|
||||||
|
digging_in_threshold_secs: 15,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1,12 +1,14 @@
|
|||||||
//! Matrix handle_message — runs the LLM turn for a verified incoming message and
|
//! Matrix handle_message — runs the LLM turn for a verified incoming message and
|
||||||
//! streams the assistant reply back to the room.
|
//! streams the assistant reply back to the room.
|
||||||
|
|
||||||
|
use crate::chat::ChatTransport;
|
||||||
use crate::chat::util::drain_complete_paragraphs;
|
use crate::chat::util::drain_complete_paragraphs;
|
||||||
use crate::llm::providers::claude_code::{ClaudeCodeProvider, ClaudeCodeResult};
|
use crate::llm::providers::claude_code::{ClaudeCodeProvider, ClaudeCodeResult};
|
||||||
use crate::slog;
|
use crate::slog;
|
||||||
use matrix_sdk::ruma::OwnedRoomId;
|
use matrix_sdk::ruma::OwnedRoomId;
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
use std::sync::atomic::{AtomicBool, Ordering};
|
use std::sync::atomic::{AtomicBool, Ordering};
|
||||||
|
use std::time::Duration;
|
||||||
use tokio::sync::watch;
|
use tokio::sync::watch;
|
||||||
|
|
||||||
use super::super::context::BotContext;
|
use super::super::context::BotContext;
|
||||||
@@ -15,6 +17,33 @@ use super::super::history::{ConversationEntry, ConversationRole, save_history};
|
|||||||
|
|
||||||
use super::format_user_prompt;
|
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<dyn ChatTransport>,
|
||||||
|
room_id: String,
|
||||||
|
sent_any_text: Arc<AtomicBool>,
|
||||||
|
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(
|
pub(in crate::chat::transport::matrix::bot) async fn handle_message(
|
||||||
room_id_str: String,
|
room_id_str: String,
|
||||||
room_id: OwnedRoomId,
|
room_id: OwnedRoomId,
|
||||||
@@ -135,10 +164,21 @@ pub(in crate::chat::transport::matrix::bot) async fn handle_message(
|
|||||||
);
|
);
|
||||||
tokio::pin!(chat_fut);
|
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
|
// Permission requests are handled by the persistent permission_listener
|
||||||
// task spawned at bot startup (story 884) — they no longer route through
|
// task spawned at bot startup (story 884) — they no longer route through
|
||||||
// per-message handlers. Just await chat_fut.
|
// per-message handlers. Just await chat_fut.
|
||||||
let result = (&mut chat_fut).await;
|
let result = (&mut chat_fut).await;
|
||||||
|
digging_in_task.abort();
|
||||||
|
|
||||||
// Flush any remaining text that didn't end with a paragraph boundary.
|
// Flush any remaining text that didn't end with a paragraph boundary.
|
||||||
let remaining = buffer.lock().unwrap().trim().to_string();
|
let remaining = buffer.lock().unwrap().trim().to_string();
|
||||||
@@ -275,3 +315,173 @@ pub(in crate::chat::transport::matrix::bot) async fn handle_message(
|
|||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
// Tests
|
// 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<Vec<(String, String)>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
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<String> {
|
||||||
|
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<MessageId, String> {
|
||||||
|
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<dyn ChatTransport>,
|
||||||
|
"!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<dyn ChatTransport>,
|
||||||
|
"!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<dyn ChatTransport>,
|
||||||
|
"!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<dyn ChatTransport>,
|
||||||
|
"!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<dyn ChatTransport>,
|
||||||
|
"!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<dyn ChatTransport>,
|
||||||
|
"!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);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -1861,6 +1861,7 @@ mod tests {
|
|||||||
compact_seed_max_bytes: 8_000,
|
compact_seed_max_bytes: 8_000,
|
||||||
cache_read_suggest_threshold: 50_000,
|
cache_read_suggest_threshold: 50_000,
|
||||||
compact_suggest_cooldown_secs: 3_600,
|
compact_suggest_cooldown_secs: 3_600,
|
||||||
|
digging_in_threshold_secs: 15,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -337,6 +337,7 @@ pub async fn run_bot(
|
|||||||
compact_seed_max_bytes: config.compact_seed_max_bytes,
|
compact_seed_max_bytes: config.compact_seed_max_bytes,
|
||||||
cache_read_suggest_threshold: config.cache_read_suggest_threshold,
|
cache_read_suggest_threshold: config.cache_read_suggest_threshold,
|
||||||
compact_suggest_cooldown_secs: config.compact_suggest_cooldown_secs,
|
compact_suggest_cooldown_secs: config.compact_suggest_cooldown_secs,
|
||||||
|
digging_in_threshold_secs: config.digging_in_threshold_secs,
|
||||||
};
|
};
|
||||||
|
|
||||||
slog!(
|
slog!(
|
||||||
|
|||||||
@@ -172,6 +172,49 @@ history_size = 50
|
|||||||
assert_eq!(config.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]
|
#[test]
|
||||||
fn load_reads_display_name() {
|
fn load_reads_display_name() {
|
||||||
let tmp = tempfile::tempdir().unwrap();
|
let tmp = tempfile::tempdir().unwrap();
|
||||||
|
|||||||
@@ -31,6 +31,12 @@ pub(super) fn default_compact_suggest_cooldown_secs() -> i64 {
|
|||||||
3_600
|
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 {
|
pub(super) fn default_transport() -> String {
|
||||||
"matrix".to_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).
|
/// room, so a busy room isn't spammed every turn. Defaults to 3600 (1h).
|
||||||
#[serde(default = "default_compact_suggest_cooldown_secs")]
|
#[serde(default = "default_compact_suggest_cooldown_secs")]
|
||||||
pub compact_suggest_cooldown_secs: i64,
|
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,
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -949,6 +949,7 @@ mod tests {
|
|||||||
compact_seed_max_bytes: 8_000,
|
compact_seed_max_bytes: 8_000,
|
||||||
cache_read_suggest_threshold: 50_000,
|
cache_read_suggest_threshold: 50_000,
|
||||||
compact_suggest_cooldown_secs: 3_600,
|
compact_suggest_cooldown_secs: 3_600,
|
||||||
|
digging_in_threshold_secs: 15,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -102,6 +102,7 @@ mod tests {
|
|||||||
compact_seed_max_bytes: 8_000,
|
compact_seed_max_bytes: 8_000,
|
||||||
cache_read_suggest_threshold: 50_000,
|
cache_read_suggest_threshold: 50_000,
|
||||||
compact_suggest_cooldown_secs: 3_600,
|
compact_suggest_cooldown_secs: 3_600,
|
||||||
|
digging_in_threshold_secs: 15,
|
||||||
};
|
};
|
||||||
run_projects_list(&ctx).await
|
run_projects_list(&ctx).await
|
||||||
}
|
}
|
||||||
@@ -224,6 +225,7 @@ mod tests {
|
|||||||
compact_seed_max_bytes: 8_000,
|
compact_seed_max_bytes: 8_000,
|
||||||
cache_read_suggest_threshold: 50_000,
|
cache_read_suggest_threshold: 50_000,
|
||||||
compact_suggest_cooldown_secs: 3_600,
|
compact_suggest_cooldown_secs: 3_600,
|
||||||
|
digging_in_threshold_secs: 15,
|
||||||
};
|
};
|
||||||
let response = run_projects_list(&ctx).await;
|
let response = run_projects_list(&ctx).await;
|
||||||
assert!(
|
assert!(
|
||||||
|
|||||||
Reference in New Issue
Block a user