//! Stage transition notifications for Matrix rooms. //! //! Subscribes to [`WatcherEvent`] broadcasts and posts a notification to all //! configured Matrix rooms whenever a work item moves between pipeline stages. use crate::config::ProjectConfig; use crate::io::story_metadata::parse_front_matter; use crate::io::watcher::WatcherEvent; use crate::slog; use crate::chat::ChatTransport; use std::collections::HashMap; use std::path::{Path, PathBuf}; use std::sync::Arc; use std::time::{Duration, Instant}; use tokio::sync::broadcast; /// Human-readable display name for a pipeline stage directory. pub fn stage_display_name(stage: &str) -> &'static str { match stage { "1_backlog" => "Backlog", "2_current" => "Current", "3_qa" => "QA", "4_merge" => "Merge", "5_done" => "Done", "6_archived" => "Archived", _ => "Unknown", } } /// Infer the previous pipeline stage for a given destination stage. /// /// Returns `None` for `1_backlog` since items are created there (not /// transitioned from another stage). pub fn inferred_from_stage(to_stage: &str) -> Option<&'static str> { match to_stage { "2_current" => Some("Backlog"), "3_qa" => Some("Current"), "4_merge" => Some("QA"), "5_done" => Some("Merge"), "6_archived" => Some("Done"), _ => None, } } /// Extract the numeric story number from an item ID like `"261_story_slug"`. pub fn extract_story_number(item_id: &str) -> Option<&str> { item_id .split('_') .next() .filter(|s| !s.is_empty() && s.chars().all(|c| c.is_ascii_digit())) } /// Read the story name from the work item file's YAML front matter. /// /// Returns `None` if the file doesn't exist or has no parseable name. pub fn read_story_name(project_root: &Path, stage: &str, item_id: &str) -> Option { let path = project_root .join(".huskies") .join("work") .join(stage) .join(format!("{item_id}.md")); let contents = std::fs::read_to_string(&path).ok()?; let meta = parse_front_matter(&contents).ok()?; meta.name } /// Format a stage transition notification message. /// /// Returns `(plain_text, html)` suitable for `RoomMessageEventContent::text_html`. pub fn format_stage_notification( item_id: &str, story_name: Option<&str>, from_stage: &str, to_stage: &str, ) -> (String, String) { let number = extract_story_number(item_id).unwrap_or(item_id); let name = story_name.unwrap_or(item_id); let prefix = if to_stage == "Done" { "\u{1f389} " } else { "" }; let plain = format!("{prefix}#{number} {name} \u{2014} {from_stage} \u{2192} {to_stage}"); let html = format!( "{prefix}#{number} {name} \u{2014} {from_stage} \u{2192} {to_stage}" ); (plain, html) } /// Format an error notification message for a story failure. /// /// Returns `(plain_text, html)` suitable for `RoomMessageEventContent::text_html`. pub fn format_error_notification( item_id: &str, story_name: Option<&str>, reason: &str, ) -> (String, String) { let number = extract_story_number(item_id).unwrap_or(item_id); let name = story_name.unwrap_or(item_id); let plain = format!("\u{274c} #{number} {name} \u{2014} {reason}"); let html = format!( "\u{274c} #{number} {name} \u{2014} {reason}" ); (plain, html) } /// Search all pipeline stages for a story name. /// /// Tries each known pipeline stage directory in order and returns the first /// name found. Used for events (like rate-limit warnings) that arrive without /// a known stage. fn find_story_name_any_stage(project_root: &Path, item_id: &str) -> Option { for stage in &["2_current", "3_qa", "4_merge", "1_backlog", "5_done"] { if let Some(name) = read_story_name(project_root, stage, item_id) { return Some(name); } } None } /// Format a blocked-story notification message. /// /// Returns `(plain_text, html)` suitable for `ChatTransport::send_message`. pub fn format_blocked_notification( item_id: &str, story_name: Option<&str>, reason: &str, ) -> (String, String) { let number = extract_story_number(item_id).unwrap_or(item_id); let name = story_name.unwrap_or(item_id); let plain = format!("\u{1f6ab} #{number} {name} \u{2014} BLOCKED: {reason}"); let html = format!( "\u{1f6ab} #{number} {name} \u{2014} BLOCKED: {reason}" ); (plain, html) } /// Minimum time between rate-limit notifications for the same agent. const RATE_LIMIT_DEBOUNCE: Duration = Duration::from_secs(60); /// Window during which rapid stage transitions for the same item are coalesced /// into a single notification (only the final stage is announced). const STAGE_TRANSITION_DEBOUNCE: Duration = Duration::from_millis(200); /// Format a rate limit hard block notification message with scheduled resume time. /// /// Returns `(plain_text, html)` suitable for `ChatTransport::send_message`. pub fn format_rate_limit_hard_block_notification( item_id: &str, story_name: Option<&str>, agent_name: &str, resume_at: chrono::DateTime, ) -> (String, String) { let number = extract_story_number(item_id).unwrap_or(item_id); let name = story_name.unwrap_or(item_id); let local_time = resume_at.with_timezone(&chrono::Local); let resume_str = local_time.format("%Y-%m-%d %H:%M").to_string(); let plain = format!( "\u{1f6d1} #{number} {name} \u{2014} {agent_name} hit a hard rate limit; \ will auto-resume at {resume_str}" ); let html = format!( "\u{1f6d1} #{number} {name} \u{2014} \ {agent_name} hit a hard rate limit; will auto-resume at {resume_str}" ); (plain, html) } /// Format a rate limit warning notification message. /// /// Returns `(plain_text, html)` suitable for `ChatTransport::send_message`. pub fn format_rate_limit_notification( item_id: &str, story_name: Option<&str>, agent_name: &str, ) -> (String, String) { let number = extract_story_number(item_id).unwrap_or(item_id); let name = story_name.unwrap_or(item_id); let plain = format!( "\u{26a0}\u{fe0f} #{number} {name} \u{2014} {agent_name} hit an API rate limit" ); let html = format!( "\u{26a0}\u{fe0f} #{number} {name} \u{2014} \ {agent_name} hit an API rate limit" ); (plain, html) } /// Spawn a background task that listens for watcher events and posts /// stage-transition notifications to all configured rooms via the /// [`ChatTransport`] abstraction. /// /// `get_room_ids` is called on each notification to obtain the current list of /// destination room IDs. Pass a closure that returns a static list for Matrix /// and Slack, or one that reads from a runtime `Arc>>` /// for WhatsApp ambient senders. pub fn spawn_notification_listener( transport: Arc, get_room_ids: impl Fn() -> Vec + Send + 'static, watcher_rx: broadcast::Receiver, project_root: PathBuf, ) { tokio::spawn(async move { let mut rx = watcher_rx; // Load initial config; re-loaded on ConfigChanged events. let mut config = ProjectConfig::load(&project_root).unwrap_or_default(); // Tracks when a rate-limit notification was last sent for each // "story_id:agent_name" key, to debounce repeated warnings. let mut rate_limit_last_notified: HashMap = HashMap::new(); // Pending stage-transition notifications, keyed by item_id. // Value: (from_display, to_stage_key, story_name). // Rapid successive transitions for the same item are coalesced: the // original from_display is kept while to_stage_key is updated to the // latest destination, so only one notification fires for the final stage. let mut pending_transitions: HashMap)> = HashMap::new(); let mut flush_deadline: Option = None; loop { // Wait for the next event, or flush pending transitions when the // debounce window expires. let recv_result = if let Some(deadline) = flush_deadline { tokio::time::timeout_at(deadline, rx.recv()).await.ok() } else { Some(rx.recv().await) }; if recv_result.is_none() { // Flush all coalesced stage-transition notifications. for (item_id, (from_display, to_stage_key, story_name)) in pending_transitions.drain() { let to_display = stage_display_name(&to_stage_key); let (plain, html) = format_stage_notification( &item_id, story_name.as_deref(), &from_display, to_display, ); slog!("[bot] Sending stage notification: {plain}"); for room_id in &get_room_ids() { if let Err(e) = transport.send_message(room_id, &plain, &html).await { slog!("[bot] Failed to send notification to {room_id}: {e}"); } } } flush_deadline = None; continue; } match recv_result.unwrap() { Ok(WatcherEvent::WorkItem { ref stage, ref item_id, ref from_stage, .. }) => { // Determine from_display: prefer the actual from_stage recorded // in the event (AC3); fall back to inference for synthetic events. let from_display = from_stage .as_deref() .map(stage_display_name) .or_else(|| inferred_from_stage(stage)); let Some(from_display) = from_display else { continue; // creation or unknown transition — skip }; // Look up the story name in the expected stage directory; fall // back to a full search so stale events still show the name (AC1). let story_name = read_story_name(&project_root, stage, item_id) .or_else(|| find_story_name_any_stage(&project_root, item_id)); // Buffer the transition. If this item_id is already pending (rapid // succession), update to_stage_key to the latest destination while // preserving the original from_display (AC2). pending_transitions .entry(item_id.clone()) .and_modify(|e| { e.1 = stage.clone(); if story_name.is_some() { e.2 = story_name.clone(); } }) .or_insert_with(|| { (from_display.to_string(), stage.clone(), story_name) }); // Start or extend the debounce window. flush_deadline = Some(tokio::time::Instant::now() + STAGE_TRANSITION_DEBOUNCE); } Ok(WatcherEvent::MergeFailure { ref story_id, ref reason, }) => { let story_name = read_story_name(&project_root, "4_merge", story_id); let (plain, html) = format_error_notification( story_id, story_name.as_deref(), reason, ); slog!("[bot] Sending error notification: {plain}"); for room_id in &get_room_ids() { if let Err(e) = transport.send_message(room_id, &plain, &html).await { slog!( "[bot] Failed to send error notification to {room_id}: {e}" ); } } } Ok(WatcherEvent::RateLimitWarning { ref story_id, ref agent_name, }) => { if !config.rate_limit_notifications { slog!( "[bot] RateLimitWarning suppressed by config for \ {story_id}:{agent_name}" ); continue; } // Debounce: skip if we sent a notification for this agent // within the last RATE_LIMIT_DEBOUNCE seconds. let debounce_key = format!("{story_id}:{agent_name}"); let now = Instant::now(); if let Some(&last) = rate_limit_last_notified.get(&debounce_key) && now.duration_since(last) < RATE_LIMIT_DEBOUNCE { slog!( "[bot] Rate-limit notification debounced for \ {story_id}:{agent_name}" ); continue; } rate_limit_last_notified.insert(debounce_key, now); let story_name = find_story_name_any_stage(&project_root, story_id); let (plain, html) = format_rate_limit_notification( story_id, story_name.as_deref(), agent_name, ); slog!("[bot] Sending rate-limit notification: {plain}"); for room_id in &get_room_ids() { if let Err(e) = transport.send_message(room_id, &plain, &html).await { slog!( "[bot] Failed to send rate-limit notification \ to {room_id}: {e}" ); } } } Ok(WatcherEvent::StoryBlocked { ref story_id, ref reason, }) => { let story_name = find_story_name_any_stage(&project_root, story_id); let (plain, html) = format_blocked_notification( story_id, story_name.as_deref(), reason, ); slog!("[bot] Sending blocked notification: {plain}"); for room_id in &get_room_ids() { if let Err(e) = transport.send_message(room_id, &plain, &html).await { slog!( "[bot] Failed to send blocked notification to {room_id}: {e}" ); } } } Ok(WatcherEvent::RateLimitHardBlock { ref story_id, ref agent_name, reset_at, }) => { // Debounce: reuse the same key as RateLimitWarning so both // types are rate-limited together for the same agent. let debounce_key = format!("{story_id}:{agent_name}"); let now = Instant::now(); if let Some(&last) = rate_limit_last_notified.get(&debounce_key) && now.duration_since(last) < RATE_LIMIT_DEBOUNCE { slog!( "[bot] Rate-limit hard-block notification debounced for \ {story_id}:{agent_name}" ); continue; } rate_limit_last_notified.insert(debounce_key, now); let story_name = find_story_name_any_stage(&project_root, story_id); let (plain, html) = format_rate_limit_hard_block_notification( story_id, story_name.as_deref(), agent_name, reset_at, ); slog!("[bot] Sending rate-limit hard-block notification: {plain}"); for room_id in &get_room_ids() { if let Err(e) = transport.send_message(room_id, &plain, &html).await { slog!( "[bot] Failed to send rate-limit hard-block notification \ to {room_id}: {e}" ); } } } Ok(WatcherEvent::ConfigChanged) => { // Hot-reload: pick up any changes to rate_limit_notifications. if let Ok(new_cfg) = ProjectConfig::load(&project_root) { config = new_cfg; } } Ok(_) => {} // Ignore other events Err(broadcast::error::RecvError::Lagged(n)) => { slog!( "[bot] Notification listener lagged, skipped {n} events" ); } Err(broadcast::error::RecvError::Closed) => { slog!( "[bot] Watcher channel closed, stopping notification listener" ); // Flush any coalesced transitions that haven't fired yet. for (item_id, (from_display, to_stage_key, story_name)) in pending_transitions.drain() { let to_display = stage_display_name(&to_stage_key); let (plain, html) = format_stage_notification( &item_id, story_name.as_deref(), &from_display, to_display, ); slog!("[bot] Sending stage notification: {plain}"); for room_id in &get_room_ids() { if let Err(e) = transport.send_message(room_id, &plain, &html).await { slog!( "[bot] Failed to send notification to {room_id}: {e}" ); } } } break; } } } }); } #[cfg(test)] mod tests { use super::*; use async_trait::async_trait; use crate::chat::MessageId; // ── MockTransport ─────────────────────────────────────────────────────── type CallLog = Arc>>; /// Records every `send_message` call for inspection in tests. struct MockTransport { calls: CallLog, } impl MockTransport { fn new() -> (Arc, CallLog) { let calls: CallLog = Arc::new(std::sync::Mutex::new(Vec::new())); (Arc::new(Self { calls: Arc::clone(&calls) }), calls) } } #[async_trait] impl crate::chat::ChatTransport for MockTransport { async fn send_message(&self, room_id: &str, plain: &str, html: &str) -> Result { self.calls.lock().unwrap().push((room_id.to_string(), plain.to_string(), html.to_string())); Ok("mock-msg-id".to_string()) } async fn edit_message(&self, _room_id: &str, _id: &str, _plain: &str, _html: &str) -> Result<(), String> { Ok(()) } async fn send_typing(&self, _room_id: &str, _typing: bool) -> Result<(), String> { Ok(()) } } // ── spawn_notification_listener: RateLimitWarning ─────────────────────── /// AC2 + AC3: when a RateLimitWarning event arrives, send_message is called /// with a notification that names the agent and story. #[tokio::test] async fn rate_limit_warning_sends_notification_with_agent_and_story() { let tmp = tempfile::tempdir().unwrap(); let stage_dir = tmp.path().join(".huskies").join("work").join("2_current"); std::fs::create_dir_all(&stage_dir).unwrap(); std::fs::write( stage_dir.join("365_story_rate_limit.md"), "---\nname: Rate Limit Test Story\n---\n", ) .unwrap(); let (watcher_tx, watcher_rx) = broadcast::channel::(16); let (transport, calls) = MockTransport::new(); spawn_notification_listener( transport, || vec!["!room123:example.org".to_string()], watcher_rx, tmp.path().to_path_buf(), ); watcher_tx.send(WatcherEvent::RateLimitWarning { story_id: "365_story_rate_limit".to_string(), agent_name: "coder-1".to_string(), }).unwrap(); // Give the spawned task time to process the event. tokio::time::sleep(std::time::Duration::from_millis(100)).await; let calls = calls.lock().unwrap(); assert_eq!(calls.len(), 1, "Expected exactly one notification"); let (room_id, plain, _html) = &calls[0]; assert_eq!(room_id, "!room123:example.org"); assert!(plain.contains("365"), "plain should contain story number"); assert!(plain.contains("Rate Limit Test Story"), "plain should contain story name"); assert!(plain.contains("coder-1"), "plain should contain agent name"); assert!(plain.contains("rate limit"), "plain should mention rate limit"); } /// AC4: a second RateLimitWarning for the same agent within the debounce /// window must NOT trigger a second notification. #[tokio::test] async fn rate_limit_warning_is_debounced() { let tmp = tempfile::tempdir().unwrap(); let (watcher_tx, watcher_rx) = broadcast::channel::(16); let (transport, calls) = MockTransport::new(); spawn_notification_listener( transport, || vec!["!room1:example.org".to_string()], watcher_rx, tmp.path().to_path_buf(), ); // Send the same warning twice in rapid succession. for _ in 0..2 { watcher_tx.send(WatcherEvent::RateLimitWarning { story_id: "42_story_debounce".to_string(), agent_name: "coder-2".to_string(), }).unwrap(); } tokio::time::sleep(std::time::Duration::from_millis(100)).await; let calls = calls.lock().unwrap(); assert_eq!(calls.len(), 1, "Debounce should suppress the second notification"); } /// AC4 (corollary): warnings for different agents are NOT debounced against /// each other — both should produce notifications. #[tokio::test] async fn rate_limit_warnings_for_different_agents_both_notify() { let tmp = tempfile::tempdir().unwrap(); let (watcher_tx, watcher_rx) = broadcast::channel::(16); let (transport, calls) = MockTransport::new(); spawn_notification_listener( transport, || vec!["!room1:example.org".to_string()], watcher_rx, tmp.path().to_path_buf(), ); watcher_tx.send(WatcherEvent::RateLimitWarning { story_id: "42_story_foo".to_string(), agent_name: "coder-1".to_string(), }).unwrap(); watcher_tx.send(WatcherEvent::RateLimitWarning { story_id: "42_story_foo".to_string(), agent_name: "coder-2".to_string(), }).unwrap(); tokio::time::sleep(std::time::Duration::from_millis(100)).await; let calls = calls.lock().unwrap(); assert_eq!(calls.len(), 2, "Different agents should each trigger a notification"); } // ── dynamic room IDs (WhatsApp ambient_rooms pattern) ─────────────────── /// Notifications are sent to the rooms returned by the closure at /// notification time, not at listener-spawn time. This verifies that a /// closure backed by a runtime set (e.g. WhatsApp ambient_rooms) delivers /// messages to the rooms present when the event fires. #[tokio::test] async fn stage_notification_uses_dynamic_room_ids() { let tmp = tempfile::tempdir().unwrap(); let stage_dir = tmp.path().join(".huskies").join("work").join("3_qa"); std::fs::create_dir_all(&stage_dir).unwrap(); std::fs::write( stage_dir.join("10_story_foo.md"), "---\nname: Foo Story\n---\n", ) .unwrap(); let (watcher_tx, watcher_rx) = broadcast::channel::(16); let (transport, calls) = MockTransport::new(); let rooms: Arc>> = Arc::new(std::sync::Mutex::new(std::collections::HashSet::new())); let rooms_for_closure = Arc::clone(&rooms); spawn_notification_listener( transport, move || rooms_for_closure.lock().unwrap().iter().cloned().collect(), watcher_rx, tmp.path().to_path_buf(), ); // Add a room after the listener is spawned (simulates a user messaging first). rooms.lock().unwrap().insert("phone:+15551234567".to_string()); watcher_tx.send(WatcherEvent::WorkItem { stage: "3_qa".to_string(), item_id: "10_story_foo".to_string(), action: "qa".to_string(), commit_msg: "huskies: qa 10_story_foo".to_string(), from_stage: None, }).unwrap(); // Wait longer than STAGE_TRANSITION_DEBOUNCE (200ms) so the coalesced // notification flushes. tokio::time::sleep(std::time::Duration::from_millis(350)).await; let calls = calls.lock().unwrap(); assert_eq!(calls.len(), 1, "Should deliver to the dynamically added room"); assert_eq!(calls[0].0, "phone:+15551234567"); assert!(calls[0].1.contains("10"), "plain should contain story number"); assert!(calls[0].1.contains("Foo Story"), "plain should contain story name"); } /// When no rooms are registered (e.g. no WhatsApp users have messaged yet), /// no notifications are sent and the listener does not panic. #[tokio::test] async fn stage_notification_with_no_rooms_is_silent() { let tmp = tempfile::tempdir().unwrap(); let (watcher_tx, watcher_rx) = broadcast::channel::(16); let (transport, calls) = MockTransport::new(); spawn_notification_listener( transport, Vec::new, watcher_rx, tmp.path().to_path_buf(), ); watcher_tx.send(WatcherEvent::WorkItem { stage: "3_qa".to_string(), item_id: "10_story_foo".to_string(), action: "qa".to_string(), commit_msg: "huskies: qa 10_story_foo".to_string(), from_stage: None, }).unwrap(); tokio::time::sleep(std::time::Duration::from_millis(100)).await; let calls = calls.lock().unwrap(); assert_eq!(calls.len(), 0, "No rooms means no notifications"); } // ── stage_display_name ────────────────────────────────────────────────── #[test] fn stage_display_name_maps_all_known_stages() { assert_eq!(stage_display_name("1_backlog"), "Backlog"); assert_eq!(stage_display_name("2_current"), "Current"); assert_eq!(stage_display_name("3_qa"), "QA"); assert_eq!(stage_display_name("4_merge"), "Merge"); assert_eq!(stage_display_name("5_done"), "Done"); assert_eq!(stage_display_name("6_archived"), "Archived"); assert_eq!(stage_display_name("unknown"), "Unknown"); } // ── inferred_from_stage ───────────────────────────────────────────────── #[test] fn inferred_from_stage_returns_previous_stage() { assert_eq!(inferred_from_stage("2_current"), Some("Backlog")); assert_eq!(inferred_from_stage("3_qa"), Some("Current")); assert_eq!(inferred_from_stage("4_merge"), Some("QA")); assert_eq!(inferred_from_stage("5_done"), Some("Merge")); assert_eq!(inferred_from_stage("6_archived"), Some("Done")); } #[test] fn inferred_from_stage_returns_none_for_backlog() { assert_eq!(inferred_from_stage("1_backlog"), None); } #[test] fn inferred_from_stage_returns_none_for_unknown() { assert_eq!(inferred_from_stage("9_unknown"), None); } // ── extract_story_number ──────────────────────────────────────────────── #[test] fn extract_story_number_parses_numeric_prefix() { assert_eq!( extract_story_number("261_story_bot_notifications"), Some("261") ); assert_eq!(extract_story_number("42_bug_fix_thing"), Some("42")); assert_eq!(extract_story_number("1_spike_research"), Some("1")); } #[test] fn extract_story_number_returns_none_for_non_numeric() { assert_eq!(extract_story_number("abc_story_thing"), None); assert_eq!(extract_story_number(""), None); } // ── read_story_name ───────────────────────────────────────────────────── #[test] fn read_story_name_reads_from_front_matter() { let tmp = tempfile::tempdir().unwrap(); let stage_dir = tmp .path() .join(".huskies") .join("work") .join("2_current"); std::fs::create_dir_all(&stage_dir).unwrap(); std::fs::write( stage_dir.join("42_story_my_feature.md"), "---\nname: My Cool Feature\n---\n# Story\n", ) .unwrap(); let name = read_story_name(tmp.path(), "2_current", "42_story_my_feature"); assert_eq!(name.as_deref(), Some("My Cool Feature")); } #[test] fn read_story_name_returns_none_for_missing_file() { let tmp = tempfile::tempdir().unwrap(); let name = read_story_name(tmp.path(), "2_current", "99_story_missing"); assert_eq!(name, None); } #[test] fn read_story_name_returns_none_for_missing_name_field() { let tmp = tempfile::tempdir().unwrap(); let stage_dir = tmp .path() .join(".huskies") .join("work") .join("2_current"); std::fs::create_dir_all(&stage_dir).unwrap(); std::fs::write( stage_dir.join("42_story_no_name.md"), "---\ncoverage_baseline: 50%\n---\n# Story\n", ) .unwrap(); let name = read_story_name(tmp.path(), "2_current", "42_story_no_name"); assert_eq!(name, None); } // ── format_error_notification ──────────────────────────────────────────── #[test] fn format_error_notification_with_story_name() { let (plain, html) = format_error_notification("262_story_bot_errors", Some("Bot error notifications"), "merge conflict in src/main.rs"); assert_eq!( plain, "\u{274c} #262 Bot error notifications \u{2014} merge conflict in src/main.rs" ); assert_eq!( html, "\u{274c} #262 Bot error notifications \u{2014} merge conflict in src/main.rs" ); } #[test] fn format_error_notification_without_story_name_falls_back_to_item_id() { let (plain, _html) = format_error_notification("42_bug_fix_thing", None, "tests failed"); assert_eq!( plain, "\u{274c} #42 42_bug_fix_thing \u{2014} tests failed" ); } #[test] fn format_error_notification_non_numeric_id_uses_full_id() { let (plain, _html) = format_error_notification("abc_story_thing", Some("Some Story"), "clippy errors"); assert_eq!( plain, "\u{274c} #abc_story_thing Some Story \u{2014} clippy errors" ); } // ── format_blocked_notification ───────────────────────────────────────── #[test] fn format_blocked_notification_with_story_name() { let (plain, html) = format_blocked_notification( "425_story_blocking_reason", Some("Blocking Reason Story"), "Retry limit exceeded (3/3) at coder stage", ); assert_eq!( plain, "\u{1f6ab} #425 Blocking Reason Story \u{2014} BLOCKED: Retry limit exceeded (3/3) at coder stage" ); assert_eq!( html, "\u{1f6ab} #425 Blocking Reason Story \u{2014} BLOCKED: Retry limit exceeded (3/3) at coder stage" ); } #[test] fn format_blocked_notification_falls_back_to_item_id() { let (plain, _html) = format_blocked_notification("42_story_thing", None, "empty diff"); assert_eq!( plain, "\u{1f6ab} #42 42_story_thing \u{2014} BLOCKED: empty diff" ); } // ── spawn_notification_listener: StoryBlocked ─────────────────────────── /// AC1: when a StoryBlocked event arrives, send_message is called with a /// notification that includes the story number, name, and reason. #[tokio::test] async fn story_blocked_sends_notification_with_reason() { let tmp = tempfile::tempdir().unwrap(); let stage_dir = tmp.path().join(".huskies").join("work").join("2_current"); std::fs::create_dir_all(&stage_dir).unwrap(); std::fs::write( stage_dir.join("425_story_blocking_test.md"), "---\nname: Blocking Test Story\n---\n", ) .unwrap(); let (watcher_tx, watcher_rx) = broadcast::channel::(16); let (transport, calls) = MockTransport::new(); spawn_notification_listener( transport, || vec!["!room123:example.org".to_string()], watcher_rx, tmp.path().to_path_buf(), ); watcher_tx.send(WatcherEvent::StoryBlocked { story_id: "425_story_blocking_test".to_string(), reason: "Retry limit exceeded (3/3) at coder stage".to_string(), }).unwrap(); tokio::time::sleep(std::time::Duration::from_millis(100)).await; let calls = calls.lock().unwrap(); assert_eq!(calls.len(), 1, "Expected exactly one notification"); let (room_id, plain, html) = &calls[0]; assert_eq!(room_id, "!room123:example.org"); assert!(plain.contains("425"), "plain should contain story number"); assert!(plain.contains("Blocking Test Story"), "plain should contain story name"); assert!(plain.contains("BLOCKED"), "plain should contain BLOCKED label"); assert!(plain.contains("Retry limit exceeded"), "plain should contain the reason"); assert!(html.contains("BLOCKED"), "html should contain BLOCKED label"); } /// StoryBlocked with no room registered should not panic. #[tokio::test] async fn story_blocked_with_no_rooms_is_silent() { let tmp = tempfile::tempdir().unwrap(); let (watcher_tx, watcher_rx) = broadcast::channel::(16); let (transport, calls) = MockTransport::new(); spawn_notification_listener( transport, Vec::new, watcher_rx, tmp.path().to_path_buf(), ); watcher_tx.send(WatcherEvent::StoryBlocked { story_id: "42_story_no_rooms".to_string(), reason: "empty diff".to_string(), }).unwrap(); tokio::time::sleep(std::time::Duration::from_millis(100)).await; let calls = calls.lock().unwrap(); assert_eq!(calls.len(), 0, "No rooms means no notifications"); } // ── format_rate_limit_notification ───────────────────────────────────── #[test] fn format_rate_limit_notification_includes_agent_and_story() { let (plain, html) = format_rate_limit_notification( "365_story_my_feature", Some("My Feature"), "coder-2", ); assert_eq!( plain, "\u{26a0}\u{fe0f} #365 My Feature \u{2014} coder-2 hit an API rate limit" ); assert_eq!( html, "\u{26a0}\u{fe0f} #365 My Feature \u{2014} coder-2 hit an API rate limit" ); } #[test] fn format_rate_limit_notification_falls_back_to_item_id() { let (plain, _html) = format_rate_limit_notification("42_story_thing", None, "coder-1"); assert_eq!( plain, "\u{26a0}\u{fe0f} #42 42_story_thing \u{2014} coder-1 hit an API rate limit" ); } // ── format_stage_notification ─────────────────────────────────────────── #[test] fn format_notification_done_stage_includes_party_emoji() { let (plain, html) = format_stage_notification( "353_story_done", Some("Done Story"), "Merge", "Done", ); assert_eq!( plain, "\u{1f389} #353 Done Story \u{2014} Merge \u{2192} Done" ); assert_eq!( html, "\u{1f389} #353 Done Story \u{2014} Merge \u{2192} Done" ); } #[test] fn format_notification_non_done_stage_has_no_emoji() { let (plain, _html) = format_stage_notification( "42_story_thing", Some("Some Story"), "Backlog", "Current", ); assert!(!plain.contains("\u{1f389}")); } #[test] fn format_notification_with_story_name() { let (plain, html) = format_stage_notification( "261_story_bot_notifications", Some("Bot notifications"), "Upcoming", "Current", ); assert_eq!( plain, "#261 Bot notifications \u{2014} Upcoming \u{2192} Current" ); assert_eq!( html, "#261 Bot notifications \u{2014} Upcoming \u{2192} Current" ); } #[test] fn format_notification_without_story_name_falls_back_to_item_id() { let (plain, _html) = format_stage_notification( "42_bug_fix_thing", None, "Current", "QA", ); assert_eq!( plain, "#42 42_bug_fix_thing \u{2014} Current \u{2192} QA" ); } #[test] fn format_notification_non_numeric_id_uses_full_id() { let (plain, _html) = format_stage_notification( "abc_story_thing", Some("Some Story"), "QA", "Merge", ); assert_eq!( plain, "#abc_story_thing Some Story \u{2014} QA \u{2192} Merge" ); } // ── rate_limit_notifications config flag ───────────────────────────────── /// AC1+AC2: when rate_limit_notifications = false in project.toml, /// RateLimitWarning events are suppressed (no send_message call). #[tokio::test] async fn rate_limit_warning_suppressed_when_config_false() { let tmp = tempfile::tempdir().unwrap(); let sk_dir = tmp.path().join(".huskies"); std::fs::create_dir_all(&sk_dir).unwrap(); std::fs::write( sk_dir.join("project.toml"), "rate_limit_notifications = false\n", ) .unwrap(); let (watcher_tx, watcher_rx) = broadcast::channel::(16); let (transport, calls) = MockTransport::new(); spawn_notification_listener( transport, || vec!["!room1:example.org".to_string()], watcher_rx, tmp.path().to_path_buf(), ); watcher_tx.send(WatcherEvent::RateLimitWarning { story_id: "42_story_suppress".to_string(), agent_name: "coder-1".to_string(), }).unwrap(); tokio::time::sleep(std::time::Duration::from_millis(100)).await; let calls = calls.lock().unwrap(); assert_eq!(calls.len(), 0, "RateLimitWarning should be suppressed when rate_limit_notifications = false"); } /// AC3: RateLimitHardBlock is always sent regardless of rate_limit_notifications. #[tokio::test] async fn rate_limit_hard_block_always_sent_when_config_false() { let tmp = tempfile::tempdir().unwrap(); let sk_dir = tmp.path().join(".huskies"); std::fs::create_dir_all(&sk_dir).unwrap(); std::fs::write( sk_dir.join("project.toml"), "rate_limit_notifications = false\n", ) .unwrap(); let (watcher_tx, watcher_rx) = broadcast::channel::(16); let (transport, calls) = MockTransport::new(); spawn_notification_listener( transport, || vec!["!room1:example.org".to_string()], watcher_rx, tmp.path().to_path_buf(), ); let reset_at = chrono::Utc::now() + chrono::Duration::hours(1); watcher_tx.send(WatcherEvent::RateLimitHardBlock { story_id: "42_story_hard_block".to_string(), agent_name: "coder-1".to_string(), reset_at, }).unwrap(); tokio::time::sleep(std::time::Duration::from_millis(100)).await; let calls = calls.lock().unwrap(); assert_eq!(calls.len(), 1, "RateLimitHardBlock should always be sent"); } /// AC3: StoryBlocked is always sent regardless of rate_limit_notifications. #[tokio::test] async fn story_blocked_always_sent_when_config_false() { let tmp = tempfile::tempdir().unwrap(); let sk_dir = tmp.path().join(".huskies"); std::fs::create_dir_all(&sk_dir).unwrap(); std::fs::write( sk_dir.join("project.toml"), "rate_limit_notifications = false\n", ) .unwrap(); let (watcher_tx, watcher_rx) = broadcast::channel::(16); let (transport, calls) = MockTransport::new(); spawn_notification_listener( transport, || vec!["!room1:example.org".to_string()], watcher_rx, tmp.path().to_path_buf(), ); watcher_tx.send(WatcherEvent::StoryBlocked { story_id: "42_story_blocked".to_string(), reason: "retry limit exceeded".to_string(), }).unwrap(); tokio::time::sleep(std::time::Duration::from_millis(100)).await; let calls = calls.lock().unwrap(); assert_eq!(calls.len(), 1, "StoryBlocked should always be sent"); } /// AC5: Config is hot-reloaded — disabling rate_limit_notifications after /// startup suppresses subsequent RateLimitWarning events. #[tokio::test] async fn rate_limit_warning_suppressed_after_hot_reload() { let tmp = tempfile::tempdir().unwrap(); let sk_dir = tmp.path().join(".huskies"); std::fs::create_dir_all(&sk_dir).unwrap(); // Start with notifications enabled. std::fs::write( sk_dir.join("project.toml"), "rate_limit_notifications = true\n", ) .unwrap(); let (watcher_tx, watcher_rx) = broadcast::channel::(16); let (transport, calls) = MockTransport::new(); spawn_notification_listener( transport, || vec!["!room1:example.org".to_string()], watcher_rx, tmp.path().to_path_buf(), ); // First warning is sent. watcher_tx.send(WatcherEvent::RateLimitWarning { story_id: "42_story_reload".to_string(), agent_name: "coder-1".to_string(), }).unwrap(); tokio::time::sleep(std::time::Duration::from_millis(100)).await; // Disable notifications and trigger hot-reload. std::fs::write( sk_dir.join("project.toml"), "rate_limit_notifications = false\n", ) .unwrap(); watcher_tx.send(WatcherEvent::ConfigChanged).unwrap(); tokio::time::sleep(std::time::Duration::from_millis(100)).await; // Second warning (different agent to bypass debounce) should be suppressed. watcher_tx.send(WatcherEvent::RateLimitWarning { story_id: "42_story_reload".to_string(), agent_name: "coder-2".to_string(), }).unwrap(); tokio::time::sleep(std::time::Duration::from_millis(100)).await; let calls = calls.lock().unwrap(); assert_eq!(calls.len(), 1, "Only the first warning should be sent; second should be suppressed after hot-reload"); } }