//! I/O side of the notifications service.
//!
//! This is the **only** file inside `service/notifications/` that may perform
//! side effects: reading from the CRDT content store, loading configuration,
//! and spawning the background listener task.

use crate::chat::ChatTransport;
use crate::config::ProjectConfig;
use crate::io::story_metadata::parse_front_matter;
use crate::io::watcher::WatcherEvent;
use crate::slog;
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Instant;
use tokio::sync::broadcast;

use super::events::classify;
use super::filter::{STAGE_TRANSITION_DEBOUNCE, should_send_rate_limit};
use super::format::{
    format_blocked_notification, format_error_notification, format_rate_limit_notification,
    format_stage_notification, stage_display_name,
};
use super::route::rooms_for_notification;

/// Read the story name from the CRDT content store's YAML front matter.
///
/// Returns `None` if the item is not in the content store or has no parseable name.
pub fn read_story_name(_project_root: &Path, _stage: &str, item_id: &str) -> Option<String> {
    let contents = crate::db::read_content(item_id)?;
    let meta = parse_front_matter(&contents).ok()?;
    meta.name
}

/// Look up a story name from the CRDT content store regardless of stage.
///
/// 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<String> {
    read_story_name(project_root, "", item_id)
}

/// 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<Mutex<HashSet<String>>>`
/// for WhatsApp ambient senders.
pub fn spawn_notification_listener(
    transport: Arc<dyn ChatTransport>,
    get_room_ids: impl Fn() -> Vec<String> + Send + 'static,
    watcher_rx: broadcast::Receiver<WatcherEvent>,
    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<String, Instant> = 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<String, (String, String, Option<String>)> =
            HashMap::new();
        let mut flush_deadline: Option<tokio::time::Instant> = 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 &rooms_for_notification(&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;
            }

            let event = match recv_result.unwrap() {
                Ok(ev) => ev,
                Err(broadcast::error::RecvError::Lagged(n)) => {
                    slog!("[bot] Notification listener lagged, skipped {n} events");
                    continue;
                }
                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 &rooms_for_notification(&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;
                }
            };

            use super::events::EventAction;
            match classify(&event) {
                EventAction::StageTransition => {
                    // WorkItem with a known from_stage — extract the fields.
                    let WatcherEvent::WorkItem {
                        ref stage,
                        ref item_id,
                        ref from_stage,
                        ..
                    } = event
                    else {
                        continue;
                    };
                    let from_display = stage_display_name(from_stage.as_deref().unwrap_or(""));

                    // Look up the story name in the expected stage directory; fall
                    // back to a full search so stale events still show the name.
                    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.
                    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);
                }
                EventAction::MergeFailure => {
                    let WatcherEvent::MergeFailure {
                        ref story_id,
                        ref reason,
                    } = event
                    else {
                        continue;
                    };
                    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 &rooms_for_notification(&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}");
                        }
                    }
                }
                EventAction::RateLimitWarning => {
                    let WatcherEvent::RateLimitWarning {
                        ref story_id,
                        ref agent_name,
                    } = event
                    else {
                        continue;
                    };
                    if !config.rate_limit_notifications {
                        slog!(
                            "[bot] RateLimitWarning suppressed by config for \
                             {story_id}:{agent_name}"
                        );
                        continue;
                    }
                    let debounce_key = format!("{story_id}:{agent_name}");
                    let now = Instant::now();
                    if !should_send_rate_limit(
                        rate_limit_last_notified.get(&debounce_key).copied(),
                        now,
                    ) {
                        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 &rooms_for_notification(&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}"
                            );
                        }
                    }
                }
                EventAction::StoryBlocked => {
                    let WatcherEvent::StoryBlocked {
                        ref story_id,
                        ref reason,
                    } = event
                    else {
                        continue;
                    };
                    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 &rooms_for_notification(&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}");
                        }
                    }
                }
                EventAction::LogOnly => {
                    // Hard-block: log server-side for debugging; do NOT post to chat.
                    // Hard-block auto-resume is normal operation — the status command
                    // already surfaces rate-limit state via emoji.
                    if let WatcherEvent::RateLimitHardBlock {
                        ref story_id,
                        ref agent_name,
                        reset_at,
                    } = event
                    {
                        slog!(
                            "[bot] Rate-limit hard block for {story_id}/{agent_name}, \
                             auto-resume at {reset_at}"
                        );
                    }
                }
                EventAction::ReloadConfig => {
                    if let Ok(new_cfg) = ProjectConfig::load(&project_root) {
                        config = new_cfg;
                    }
                }
                EventAction::Skip => {}
            }
        }
    });
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::chat::MessageId;
    use async_trait::async_trait;

    // ── MockTransport ───────────────────────────────────────────────────────

    type CallLog = Arc<std::sync::Mutex<Vec<(String, String, String)>>>;

    /// Records every `send_message` call for inspection in tests.
    struct MockTransport {
        calls: CallLog,
    }

    impl MockTransport {
        fn new() -> (Arc<Self>, 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<MessageId, String> {
            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();
        // Seed story via CRDT (the only source of truth).
        crate::db::ensure_content_store();
        crate::db::write_item_with_content(
            "365_story_rate_limit",
            "2_current",
            "---\nname: Rate Limit Test Story\n---\n",
        );

        let (watcher_tx, watcher_rx) = broadcast::channel::<WatcherEvent>(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::<WatcherEvent>(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::<WatcherEvent>(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();
        // Seed story via CRDT (the only source of truth).
        crate::db::ensure_content_store();
        crate::db::write_item_with_content("10_story_foo", "3_qa", "---\nname: Foo Story\n---\n");

        let (watcher_tx, watcher_rx) = broadcast::channel::<WatcherEvent>(16);
        let (transport, calls) = MockTransport::new();

        let rooms: Arc<std::sync::Mutex<std::collections::HashSet<String>>> =
            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: Some("2_current".to_string()),
            })
            .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::<WatcherEvent>(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: Some("2_current".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");
    }

    // ── read_story_name ─────────────────────────────────────────────────────

    #[test]
    fn read_story_name_reads_from_front_matter() {
        crate::db::ensure_content_store();
        crate::db::write_item_with_content(
            "9942_story_my_feature",
            "2_current",
            "---\nname: My Cool Feature\n---\n# Story\n",
        );

        let tmp = tempfile::tempdir().unwrap();
        let name = read_story_name(tmp.path(), "2_current", "9942_story_my_feature");
        assert_eq!(name.as_deref(), Some("My Cool Feature"));
    }

    #[test]
    fn read_story_name_returns_none_for_missing_file() {
        crate::db::ensure_content_store();
        let tmp = tempfile::tempdir().unwrap();
        let name = read_story_name(tmp.path(), "2_current", "99_story_missing_notif_test");
        assert_eq!(name, None);
    }

    #[test]
    fn read_story_name_returns_none_for_missing_name_field() {
        crate::db::ensure_content_store();
        crate::db::write_item_with_content(
            "9943_story_no_name",
            "2_current",
            "---\ncoverage_baseline: 50%\n---\n# Story\n",
        );

        let tmp = tempfile::tempdir().unwrap();
        let name = read_story_name(tmp.path(), "2_current", "9943_story_no_name");
        assert_eq!(name, None);
    }

    // ── 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();
        // Seed story via CRDT (the only source of truth).
        crate::db::ensure_content_store();
        crate::db::write_item_with_content(
            "425_story_blocking_test",
            "2_current",
            "---\nname: Blocking Test Story\n---\n",
        );

        let (watcher_tx, watcher_rx) = broadcast::channel::<WatcherEvent>(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::<WatcherEvent>(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");
    }

    // ── 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::<WatcherEvent>(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"
        );
    }

    /// RateLimitHardBlock is never posted to Matrix — it is logged server-side only.
    #[tokio::test]
    async fn rate_limit_hard_block_never_sent_to_matrix() {
        let tmp = tempfile::tempdir().unwrap();

        let (watcher_tx, watcher_rx) = broadcast::channel::<WatcherEvent>(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(), 0, "RateLimitHardBlock must not post to Matrix");
    }

    /// 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::<WatcherEvent>(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::<WatcherEvent>(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"
        );
    }

    // ── Bug 549: synthetic events with from_stage=None must not notify ──────

    /// Synthetic events (reassign, creation) have from_stage=None and must
    /// not produce stage-transition notifications.  Before the fix, the
    /// inferred_from_stage fallback would emit e.g. "QA → Merge" for a
    /// reassign event within the merge stage.
    #[tokio::test]
    async fn synthetic_event_without_from_stage_does_not_notify() {
        let tmp = tempfile::tempdir().unwrap();

        let (watcher_tx, watcher_rx) = broadcast::channel::<WatcherEvent>(16);
        let (transport, calls) = MockTransport::new();

        spawn_notification_listener(
            transport,
            || vec!["!room1:example.org".to_string()],
            watcher_rx,
            tmp.path().to_path_buf(),
        );

        // Synthetic reassign event within 4_merge — no actual stage change.
        watcher_tx
            .send(WatcherEvent::WorkItem {
                stage: "4_merge".to_string(),
                item_id: "549_story_skip_qa".to_string(),
                action: "reassign".to_string(),
                commit_msg: String::new(),
                from_stage: None,
            })
            .unwrap();

        tokio::time::sleep(std::time::Duration::from_millis(350)).await;

        let calls = calls.lock().unwrap();
        assert_eq!(
            calls.len(),
            0,
            "Synthetic events with from_stage=None must not generate notifications"
        );
    }

    /// Stories that skip QA (qa: server) move directly from Current to Merge.
    /// The notification must say "Current → Merge", not "QA → Merge".
    #[tokio::test]
    async fn skip_qa_shows_current_to_merge_not_qa_to_merge() {
        let tmp = tempfile::tempdir().unwrap();
        let stage_dir = tmp.path().join(".huskies").join("work").join("4_merge");
        std::fs::create_dir_all(&stage_dir).unwrap();
        std::fs::write(
            stage_dir.join("549_story_skip_qa.md"),
            "---\nname: Skip QA Story\n---\n",
        )
        .unwrap();

        let (watcher_tx, watcher_rx) = broadcast::channel::<WatcherEvent>(16);
        let (transport, calls) = MockTransport::new();

        spawn_notification_listener(
            transport,
            || vec!["!room1:example.org".to_string()],
            watcher_rx,
            tmp.path().to_path_buf(),
        );

        // Story skips QA: from_stage is 2_current, not 3_qa.
        watcher_tx
            .send(WatcherEvent::WorkItem {
                stage: "4_merge".to_string(),
                item_id: "549_story_skip_qa".to_string(),
                action: "merge".to_string(),
                commit_msg: "huskies: merge 549_story_skip_qa".to_string(),
                from_stage: Some("2_current".to_string()),
            })
            .unwrap();

        tokio::time::sleep(std::time::Duration::from_millis(350)).await;

        let calls = calls.lock().unwrap();
        assert_eq!(calls.len(), 1, "Should send exactly one notification");
        assert!(
            calls[0].1.contains("Current \u{2192} Merge"),
            "Notification should say 'Current → Merge', got: {}",
            calls[0].1
        );
        assert!(
            !calls[0].1.contains("QA \u{2192} Merge"),
            "Must NOT say 'QA → Merge' when QA was skipped"
        );
    }
}