storkit/server/src/matrix/bot.rs

use crate::agents::{AgentPool, AgentStatus};
use crate::config::ProjectConfig;
use crate::http::context::{PermissionDecision, PermissionForward};
use crate::llm::providers::claude_code::{ClaudeCodeProvider, ClaudeCodeResult};
use crate::slog;
use matrix_sdk::{
    Client,
    config::SyncSettings,
    event_handler::Ctx,
    room::Room,
    ruma::{
        OwnedEventId, OwnedRoomId, OwnedUserId,
        events::room::message::{
            MessageType, OriginalSyncRoomMessageEvent, Relation, RoomMessageEventContent,
            RoomMessageEventContentWithoutRelation,
        },
    },
};
use pulldown_cmark::{Options, Parser, html};
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet};
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use tokio::sync::Mutex as TokioMutex;
use tokio::sync::{mpsc, oneshot, watch};

use futures::StreamExt;
use matrix_sdk::encryption::verification::{
    SasState, SasVerification, Verification, VerificationRequestState, format_emojis,
};
use matrix_sdk::ruma::events::key::verification::request::ToDeviceKeyVerificationRequestEvent;

use super::config::{BotConfig, save_ambient_rooms};

// ---------------------------------------------------------------------------
// Conversation history types
// ---------------------------------------------------------------------------

/// Role of a participant in the conversation history.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum ConversationRole {
    /// A message sent by a Matrix room participant.
    User,
    /// A response generated by the bot / LLM.
    Assistant,
}

/// A single turn in the per-room conversation history.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct ConversationEntry {
    pub role: ConversationRole,
    /// Matrix user ID (e.g. `@alice:example.com`).  Empty for assistant turns.
    pub sender: String,
    pub content: String,
}

/// Per-room state: conversation entries plus the Claude Code session ID for
/// structured conversation resumption.
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
pub struct RoomConversation {
    /// Claude Code session ID used to resume multi-turn conversations so the
    /// LLM receives prior turns as structured API messages rather than a
    /// flattened text prefix.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub session_id: Option<String>,
    /// Rolling conversation entries (used for turn counting and persistence).
    pub entries: Vec<ConversationEntry>,
}

/// Per-room conversation state, keyed by room ID (serialised as string).
///
/// Wrapped in `Arc<TokioMutex<…>>` so it can be shared across concurrent
/// event-handler tasks without blocking the sync loop.
pub type ConversationHistory = Arc<TokioMutex<HashMap<OwnedRoomId, RoomConversation>>>;

/// On-disk format for persisted conversation history.  Room IDs are stored as
/// strings because `OwnedRoomId` does not implement `Serialize` as a map key.
#[derive(Serialize, Deserialize)]
struct PersistedHistory {
    rooms: HashMap<String, RoomConversation>,
}

/// Path to the persisted conversation history file relative to project root.
const HISTORY_FILE: &str = ".story_kit/matrix_history.json";

/// Load conversation history from disk, returning an empty map on any error.
pub fn load_history(project_root: &std::path::Path) -> HashMap<OwnedRoomId, RoomConversation> {
    let path = project_root.join(HISTORY_FILE);
    let data = match std::fs::read_to_string(&path) {
        Ok(d) => d,
        Err(_) => return HashMap::new(),
    };
    let persisted: PersistedHistory = match serde_json::from_str(&data) {
        Ok(p) => p,
        Err(e) => {
            slog!("[matrix-bot] Failed to parse history file: {e}");
            return HashMap::new();
        }
    };
    persisted
        .rooms
        .into_iter()
        .filter_map(|(k, v)| {
            k.parse::<OwnedRoomId>()
                .ok()
                .map(|room_id| (room_id, v))
        })
        .collect()
}

/// Save conversation history to disk.  Errors are logged but not propagated.
pub fn save_history(
    project_root: &std::path::Path,
    history: &HashMap<OwnedRoomId, RoomConversation>,
) {
    let persisted = PersistedHistory {
        rooms: history
            .iter()
            .map(|(k, v)| (k.to_string(), v.clone()))
            .collect(),
    };
    let path = project_root.join(HISTORY_FILE);
    match serde_json::to_string_pretty(&persisted) {
        Ok(json) => {
            if let Err(e) = std::fs::write(&path, json) {
                slog!("[matrix-bot] Failed to write history file: {e}");
            }
        }
        Err(e) => slog!("[matrix-bot] Failed to serialise history: {e}"),
    }
}

// ---------------------------------------------------------------------------
// Bot context
// ---------------------------------------------------------------------------

/// Shared context injected into Matrix event handlers.
#[derive(Clone)]
pub struct BotContext {
    pub bot_user_id: OwnedUserId,
    /// All room IDs the bot listens in.
    pub target_room_ids: Vec<OwnedRoomId>,
    pub project_root: PathBuf,
    pub allowed_users: Vec<String>,
    /// Shared, per-room rolling conversation history.
    pub history: ConversationHistory,
    /// Maximum number of entries to keep per room before trimming the oldest.
    pub history_size: usize,
    /// Event IDs of messages the bot has sent.  Used to detect replies to the
    /// bot so it can continue a conversation thread without requiring an
    /// explicit `@mention` on every follow-up.
    pub bot_sent_event_ids: Arc<TokioMutex<HashSet<OwnedEventId>>>,
    /// Receiver for permission requests from the MCP `prompt_permission` tool.
    /// During an active chat the bot locks this to poll for incoming requests.
    pub perm_rx: Arc<TokioMutex<mpsc::UnboundedReceiver<PermissionForward>>>,
    /// Per-room pending permission reply senders. When a permission prompt is
    /// posted to a room the oneshot sender is stored here; when the user
    /// replies (yes/no) the event handler resolves it.
    pub pending_perm_replies:
        Arc<TokioMutex<HashMap<OwnedRoomId, oneshot::Sender<PermissionDecision>>>>,
    /// How long to wait for a user to respond to a permission prompt before
    /// denying (fail-closed).
    pub permission_timeout_secs: u64,
    /// The name the bot uses to refer to itself. Derived from `display_name`
    /// in bot.toml; defaults to "Assistant" when unset.
    pub bot_name: String,
    /// Set of room IDs where ambient mode is active.  In ambient mode the bot
    /// responds to all messages rather than only addressed ones.  This is
    /// in-memory only — the state does not survive a bot restart.
    pub ambient_rooms: Arc<TokioMutex<HashSet<OwnedRoomId>>>,
    /// Agent pool for checking agent availability.
    pub agents: Arc<AgentPool>,
}

// ---------------------------------------------------------------------------
// Startup announcement
// ---------------------------------------------------------------------------

/// Format the startup greeting the bot sends to each room when it comes online.
///
/// Uses the bot's configured display name so the message reads naturally
/// (e.g. "Timmy is online.").
pub fn format_startup_announcement(bot_name: &str) -> String {
    format!("{bot_name} is online.")
}

// ---------------------------------------------------------------------------
// Command extraction
// ---------------------------------------------------------------------------

/// Extract the command portion from a bot-addressed message.
///
/// Strips the leading bot mention (full Matrix user ID, `@localpart`, or
/// display name) plus any trailing punctuation (`,`, `:`) and whitespace,
/// then returns the remainder in lowercase. Returns `None` when no
/// recognized mention prefix is found in the message.
pub fn extract_command(body: &str, bot_name: &str, bot_user_id: &OwnedUserId) -> Option<String> {
    let full_id = bot_user_id.as_str().to_lowercase();
    let at_localpart = format!("@{}", bot_user_id.localpart().to_lowercase());
    let bot_name_lower = bot_name.to_lowercase();
    let body_lower = body.trim().to_lowercase();

    let stripped = if let Some(s) = body_lower.strip_prefix(&full_id) {
        s
    } else if let Some(s) = body_lower.strip_prefix(&at_localpart) {
        // Guard against matching a longer @mention (e.g. "@timmybot" vs "@timmy").
        let next = s.chars().next();
        if next.is_some_and(|c| c.is_alphanumeric() || c == '-' || c == '_') {
            return None;
        }
        s
    } else if let Some(s) = body_lower.strip_prefix(&bot_name_lower) {
        // Guard against matching a longer display-name prefix.
        let next = s.chars().next();
        if next.is_some_and(|c| c.is_alphanumeric() || c == '-' || c == '_') {
            return None;
        }
        s
    } else {
        return None;
    };

    // Strip leading separators (`,`, `:`) and whitespace after the mention.
    let cmd = stripped.trim_start_matches(|c: char| c == ':' || c == ',' || c.is_whitespace());
    Some(cmd.trim().to_string())
}

// ---------------------------------------------------------------------------
// Pipeline status formatter
// ---------------------------------------------------------------------------

/// Read all story IDs and names from a pipeline stage directory.
fn read_stage_items(
    project_root: &std::path::Path,
    stage_dir: &str,
) -> Vec<(String, Option<String>)> {
    let dir = project_root
        .join(".story_kit")
        .join("work")
        .join(stage_dir);
    if !dir.exists() {
        return Vec::new();
    }
    let mut items = Vec::new();
    if let Ok(entries) = std::fs::read_dir(&dir) {
        for entry in entries.flatten() {
            let path = entry.path();
            if path.extension().and_then(|e| e.to_str()) != Some("md") {
                continue;
            }
            if let Some(stem) = path.file_stem().and_then(|s| s.to_str()) {
                let name = std::fs::read_to_string(&path)
                    .ok()
                    .and_then(|contents| {
                        crate::io::story_metadata::parse_front_matter(&contents)
                            .ok()
                            .and_then(|m| m.name)
                    });
                items.push((stem.to_string(), name));
            }
        }
    }
    items.sort_by(|a, b| a.0.cmp(&b.0));
    items
}

/// Build the full pipeline status text formatted for Matrix (markdown).
pub fn build_pipeline_status(project_root: &std::path::Path, agents: &AgentPool) -> String {
    // Build a map from story_id → active AgentInfo for quick lookup.
    let active_agents = agents.list_agents().unwrap_or_default();
    let active_map: std::collections::HashMap<String, &crate::agents::AgentInfo> = active_agents
        .iter()
        .filter(|a| matches!(a.status, AgentStatus::Running | AgentStatus::Pending))
        .map(|a| (a.story_id.clone(), a))
        .collect();

    let config = ProjectConfig::load(project_root).ok();

    let mut out = String::from("**Pipeline Status**\n\n");

    let stages = [
        ("1_backlog", "Backlog"),
        ("2_current", "In Progress"),
        ("3_qa", "QA"),
        ("4_merge", "Merge"),
        ("5_done", "Done"),
    ];

    for (dir, label) in &stages {
        let items = read_stage_items(project_root, dir);
        let count = items.len();
        out.push_str(&format!("**{label}** ({count})\n"));
        if items.is_empty() {
            out.push_str("  *(none)*\n");
        } else {
            for (story_id, name) in &items {
                let display = match name {
                    Some(n) => format!("{story_id} — {n}"),
                    None => story_id.clone(),
                };
                if let Some(agent) = active_map.get(story_id) {
                    let model_str = config
                        .as_ref()
                        .and_then(|cfg| cfg.find_agent(&agent.agent_name))
                        .and_then(|ac| ac.model.as_deref())
                        .unwrap_or("?");
                    out.push_str(&format!(
                        "  • {display} — {} ({}) [{}]\n",
                        agent.agent_name, model_str, agent.status
                    ));
                } else {
                    out.push_str(&format!("  • {display}\n"));
                }
            }
        }
        out.push('\n');
    }

    // Free agents: configured agents not currently running or pending.
    out.push_str("**Free Agents**\n");
    if let Some(cfg) = &config {
        let busy_names: std::collections::HashSet<String> = active_agents
            .iter()
            .filter(|a| matches!(a.status, AgentStatus::Running | AgentStatus::Pending))
            .map(|a| a.agent_name.clone())
            .collect();

        let free: Vec<String> = cfg
            .agent
            .iter()
            .filter(|a| !busy_names.contains(&a.name))
            .map(|a| match &a.model {
                Some(m) => format!("{} ({})", a.name, m),
                None => a.name.clone(),
            })
            .collect();

        if free.is_empty() {
            out.push_str("  *(none — all agents busy)*\n");
        } else {
            for name in &free {
                out.push_str(&format!("  • {name}\n"));
            }
        }
    } else {
        out.push_str("  *(no agent config found)*\n");
    }

    out
}

// ---------------------------------------------------------------------------
// Bot entry point
// ---------------------------------------------------------------------------

/// Connect to the Matrix homeserver, join all configured rooms, and start
/// listening for messages. Runs the full Matrix sync loop — call from a
/// `tokio::spawn` task so it doesn't block the main thread.
pub async fn run_bot(
    config: BotConfig,
    project_root: PathBuf,
    watcher_rx: tokio::sync::broadcast::Receiver<crate::io::watcher::WatcherEvent>,
    perm_rx: Arc<TokioMutex<mpsc::UnboundedReceiver<PermissionForward>>>,
    agents: Arc<AgentPool>,
) -> Result<(), String> {
    let store_path = project_root.join(".story_kit").join("matrix_store");
    let client = Client::builder()
        .homeserver_url(&config.homeserver)
        .sqlite_store(&store_path, None)
        .build()
        .await
        .map_err(|e| format!("Failed to build Matrix client: {e}"))?;

    // Persist device ID so E2EE crypto state survives restarts.
    let device_id_path = project_root.join(".story_kit").join("matrix_device_id");
    let saved_device_id: Option<String> = std::fs::read_to_string(&device_id_path)
        .ok()
        .map(|s| s.trim().to_string())
        .filter(|s| !s.is_empty());

    let mut login_builder = client
        .matrix_auth()
        .login_username(&config.username, &config.password)
        .initial_device_display_name("Story Kit Bot");

    if let Some(ref device_id) = saved_device_id {
        login_builder = login_builder.device_id(device_id);
    }

    let login_response = login_builder
        .await
        .map_err(|e| format!("Matrix login failed: {e}"))?;

    // Save device ID on first login so subsequent restarts reuse the same device.
    if saved_device_id.is_none() {
        let _ = std::fs::write(&device_id_path, &login_response.device_id);
        slog!(
            "[matrix-bot] Saved device ID {} for future restarts",
            login_response.device_id
        );
    }

    let bot_user_id = client
        .user_id()
        .ok_or_else(|| "No user ID after login".to_string())?
        .to_owned();

    slog!("[matrix-bot] Logged in as {bot_user_id} (device: {})", login_response.device_id);

    // Bootstrap cross-signing keys for E2EE verification support.
    // Pass the bot's password for UIA (User-Interactive Authentication) —
    // the homeserver requires proof of identity before accepting cross-signing keys.
    {
        use matrix_sdk::ruma::api::client::uiaa;
        let password_auth = uiaa::AuthData::Password(uiaa::Password::new(
            uiaa::UserIdentifier::UserIdOrLocalpart(config.username.clone()),
            config.password.clone(),
        ));
        if let Err(e) = client
            .encryption()
            .bootstrap_cross_signing(Some(password_auth))
            .await
        {
            slog!("[matrix-bot] Cross-signing bootstrap note: {e}");
        }
    }

    // Self-sign own device keys so other clients don't show
    // "encrypted by a device not verified by its owner" warnings.
    match client.encryption().get_own_device().await {
        Ok(Some(own_device)) => {
            if own_device.is_cross_signed_by_owner() {
                slog!("[matrix-bot] Device already self-signed");
            } else {
                slog!("[matrix-bot] Device not self-signed, signing now...");
                match own_device.verify().await {
                    Ok(()) => slog!("[matrix-bot] Successfully self-signed device keys"),
                    Err(e) => slog!("[matrix-bot] Failed to self-sign device keys: {e}"),
                }
            }
        }
        Ok(None) => slog!("[matrix-bot] Could not find own device in crypto store"),
        Err(e) => slog!("[matrix-bot] Error retrieving own device: {e}"),
    }

    if config.allowed_users.is_empty() {
        return Err(
            "allowed_users is empty in bot.toml — refusing to start (fail-closed). \
             Add at least one Matrix user ID to allowed_users."
                .to_string(),
        );
    }

    slog!("[matrix-bot] Allowed users: {:?}", config.allowed_users);

    // Parse and join all configured rooms.
    let mut target_room_ids: Vec<OwnedRoomId> = Vec::new();
    for room_id_str in config.effective_room_ids() {
        let room_id: OwnedRoomId = room_id_str
            .parse()
            .map_err(|_| format!("Invalid room ID '{room_id_str}'"))?;

        // Try to join with a timeout. Conduit sometimes hangs or returns
        // errors on join if the bot is already a member.
        match tokio::time::timeout(
            std::time::Duration::from_secs(10),
            client.join_room_by_id(&room_id),
        )
        .await
        {
            Ok(Ok(_)) => slog!("[matrix-bot] Joined room {room_id}"),
            Ok(Err(e)) => {
                slog!("[matrix-bot] Join room error (may already be a member): {e}")
            }
            Err(_) => slog!("[matrix-bot] Join room timed out (may already be a member)"),
        }

        target_room_ids.push(room_id);
    }

    if target_room_ids.is_empty() {
        return Err("No valid room IDs configured — cannot start".to_string());
    }

    slog!(
        "[matrix-bot] Listening in {} room(s): {:?}",
        target_room_ids.len(),
        target_room_ids
    );

    // Clone values needed by the notification listener and startup announcement
    // before they are moved into BotContext.
    let notif_room_ids = target_room_ids.clone();
    let notif_project_root = project_root.clone();
    let announce_room_ids = target_room_ids.clone();

    let persisted = load_history(&project_root);
    slog!(
        "[matrix-bot] Loaded persisted conversation history for {} room(s)",
        persisted.len()
    );

    // Restore persisted ambient rooms from config, ignoring any that are not
    // in the configured target_room_ids to avoid stale entries.
    let persisted_ambient: HashSet<OwnedRoomId> = config
        .ambient_rooms
        .iter()
        .filter_map(|s| s.parse::<OwnedRoomId>().ok())
        .collect();
    if !persisted_ambient.is_empty() {
        slog!(
            "[matrix-bot] Restored ambient mode for {} room(s): {:?}",
            persisted_ambient.len(),
            persisted_ambient
        );
    }

    let bot_name = config
        .display_name
        .clone()
        .unwrap_or_else(|| "Assistant".to_string());
    let announce_bot_name = bot_name.clone();

    let ctx = BotContext {
        bot_user_id,
        target_room_ids,
        project_root,
        allowed_users: config.allowed_users,
        history: Arc::new(TokioMutex::new(persisted)),
        history_size: config.history_size,
        bot_sent_event_ids: Arc::new(TokioMutex::new(HashSet::new())),
        perm_rx,
        pending_perm_replies: Arc::new(TokioMutex::new(HashMap::new())),
        permission_timeout_secs: config.permission_timeout_secs,
        bot_name,
        ambient_rooms: Arc::new(TokioMutex::new(persisted_ambient)),
        agents,
    };

    slog!("[matrix-bot] Cryptographic identity verification is always ON — commands from unencrypted rooms or unverified devices are rejected");

    // Register event handlers and inject shared context.
    client.add_event_handler_context(ctx);
    client.add_event_handler(on_room_message);
    client.add_event_handler(on_to_device_verification_request);

    // Spawn the stage-transition notification listener before entering the
    // sync loop so it starts receiving watcher events immediately.
    super::notifications::spawn_notification_listener(
        client.clone(),
        notif_room_ids,
        watcher_rx,
        notif_project_root,
    );

    // Send a startup announcement to each configured room so users know the
    // bot is online.  This runs once per process start — the sync loop handles
    // reconnects internally so this code is never reached again on a network
    // blip or sync resumption.
    let announce_msg = format_startup_announcement(&announce_bot_name);
    slog!("[matrix-bot] Sending startup announcement: {announce_msg}");
    for room_id in &announce_room_ids {
        if let Some(room) = client.get_room(room_id) {
            let content = RoomMessageEventContent::text_plain(announce_msg.clone());
            if let Err(e) = room.send(content).await {
                slog!("[matrix-bot] Failed to send startup announcement to {room_id}: {e}");
            }
        } else {
            slog!("[matrix-bot] Room {room_id} not found in client state, skipping announcement");
        }
    }

    slog!("[matrix-bot] Starting Matrix sync loop");

    // This blocks until the connection is terminated or an error occurs.
    client
        .sync(SyncSettings::default())
        .await
        .map_err(|e| format!("Matrix sync error: {e}"))?;

    Ok(())
}

// ---------------------------------------------------------------------------
// Address-filtering helpers
// ---------------------------------------------------------------------------

/// Returns `true` if the message body is an affirmative permission response.
///
/// Recognised affirmative tokens (case-insensitive): `yes`, `y`, `approve`,
/// `allow`, `ok`. Anything else — including ambiguous text — is treated as
/// denial (fail-closed).
fn is_permission_approval(body: &str) -> bool {
    // Strip a leading @mention (e.g. "@timmy yes") so the bot name doesn't
    // interfere with the check.
    let trimmed = body
        .trim()
        .trim_start_matches('@')
        .split_whitespace()
        .last()
        .unwrap_or("")
        .to_ascii_lowercase();
    matches!(trimmed.as_str(), "yes" | "y" | "approve" | "allow" | "ok")
}

/// Returns `true` if the message mentions the bot.
///
/// Checks both the plain-text `body` and an optional `formatted_body` (HTML).
/// Recognised forms:
/// - The bot's full Matrix user ID (e.g. `@timmy:homeserver.local`) in either body
/// - The localpart with `@` prefix (e.g. `@timmy`) with word-boundary check
/// - A `matrix.to` link containing the user ID (in `formatted_body`)
///
/// Short mentions are only counted when not immediately followed by an
/// alphanumeric character, hyphen, or underscore to avoid false positives.
pub fn mentions_bot(body: &str, formatted_body: Option<&str>, bot_user_id: &OwnedUserId) -> bool {
    let full_id = bot_user_id.as_str();
    let localpart = bot_user_id.localpart();

    // Check formatted_body for a matrix.to link containing the bot's user ID.
    if formatted_body.is_some_and(|html| html.contains(full_id)) {
        return true;
    }

    // Check plain body for the full ID.
    if body.contains(full_id) {
        return true;
    }

    // Check plain body for @localpart (e.g. "@timmy") with word boundaries.
    if contains_word(body, &format!("@{localpart}")) {
        return true;
    }

    false
}

/// Returns `true` if `haystack` contains `needle` at a word boundary.
fn contains_word(haystack: &str, needle: &str) -> bool {
    let mut start = 0;
    while let Some(rel) = haystack[start..].find(needle) {
        let abs = start + rel;
        let after = abs + needle.len();
        let next = haystack[after..].chars().next();
        let is_word_end = next.is_none_or(|c| !c.is_alphanumeric() && c != '-' && c != '_');
        if is_word_end {
            return true;
        }
        start = abs + 1;
    }
    false
}

/// Parse an ambient-mode toggle command from a message body.
///
/// Recognises the following (case-insensitive) forms, with or without a
/// leading bot mention:
///
/// - `@botname ambient on`  / `@botname:server ambient on`
/// - `botname ambient on`
/// - `ambient on`
///
/// and the `off` variants.
///
/// Returns `Some(true)` for "ambient on", `Some(false)` for "ambient off",
/// and `None` when the body is not an ambient mode command.
pub fn parse_ambient_command(
    body: &str,
    bot_user_id: &OwnedUserId,
    bot_name: &str,
) -> Option<bool> {
    let lower = body.trim().to_ascii_lowercase();
    let display_lower = bot_name.to_ascii_lowercase();
    let localpart_lower = bot_user_id.localpart().to_ascii_lowercase();

    // Strip a leading @mention (handles "@localpart" and "@localpart:server").
    let rest = if let Some(after_at) = lower.strip_prefix('@') {
        // Skip everything up to the first whitespace (the full mention token).
        let word_end = after_at
            .find(char::is_whitespace)
            .unwrap_or(after_at.len());
        after_at[word_end..].trim()
    } else if let Some(after) = lower.strip_prefix(display_lower.as_str()) {
        after.trim()
    } else if let Some(after) = lower.strip_prefix(localpart_lower.as_str()) {
        after.trim()
    } else {
        lower.as_str()
    };

    match rest {
        "ambient on" => Some(true),
        "ambient off" => Some(false),
        _ => None,
    }
}

/// Returns `true` if the message's `relates_to` field references an event that
/// the bot previously sent (i.e. the message is a reply or thread-reply to a
/// bot message).
async fn is_reply_to_bot(
    relates_to: Option<&Relation<RoomMessageEventContentWithoutRelation>>,
    bot_sent_event_ids: &TokioMutex<HashSet<OwnedEventId>>,
) -> bool {
    let candidate_ids: Vec<&OwnedEventId> = match relates_to {
        Some(Relation::Reply { in_reply_to }) => vec![&in_reply_to.event_id],
        Some(Relation::Thread(thread)) => {
            let mut ids = vec![&thread.event_id];
            if let Some(irti) = &thread.in_reply_to {
                ids.push(&irti.event_id);
            }
            ids
        }
        _ => return false,
    };
    let guard = bot_sent_event_ids.lock().await;
    candidate_ids.iter().any(|id| guard.contains(*id))
}

// ---------------------------------------------------------------------------
// E2EE device verification helpers
// ---------------------------------------------------------------------------

/// Check whether the sender has a cross-signing identity known to the bot.
///
/// Returns `Ok(true)` if the sender has cross-signing keys set up (their
/// identity is present in the local crypto store), `Ok(false)` if they have
/// no cross-signing identity at all, and `Err` on failures.
///
/// Checking identity presence (rather than individual device verification)
/// is the correct trust model: a user is accepted when they have cross-signing
/// configured, regardless of whether the bot has run an explicit verification
/// ceremony with a specific device.
async fn check_sender_verified(
    client: &Client,
    sender: &OwnedUserId,
) -> Result<bool, String> {
    let identity = client
        .encryption()
        .get_user_identity(sender)
        .await
        .map_err(|e| format!("Failed to get identity for {sender}: {e}"))?;

    // Accept if the user has a cross-signing identity (Some); reject if they
    // have no cross-signing setup at all (None).
    Ok(identity.is_some())
}

// ---------------------------------------------------------------------------
// SAS verification handler
// ---------------------------------------------------------------------------

/// Handle an incoming to-device verification request by accepting it and
/// driving the SAS (emoji comparison) flow to completion.  The bot auto-
/// confirms the SAS code — the operator can compare the emojis logged to
/// the console with those displayed in their Element client.
async fn on_to_device_verification_request(
    ev: ToDeviceKeyVerificationRequestEvent,
    client: Client,
) {
    slog!(
        "[matrix-bot] Incoming verification request from {} (device: {})",
        ev.sender,
        ev.content.from_device
    );

    let Some(request) = client
        .encryption()
        .get_verification_request(&ev.sender, &ev.content.transaction_id)
        .await
    else {
        slog!("[matrix-bot] Could not locate verification request in crypto store");
        return;
    };

    if let Err(e) = request.accept().await {
        slog!("[matrix-bot] Failed to accept verification request: {e}");
        return;
    }

    // Try to start a SAS flow.  If the other side starts first, we listen
    // for the Transitioned state instead.
    match request.start_sas().await {
        Ok(Some(sas)) => {
            handle_sas_verification(sas).await;
        }
        Ok(None) => {
            slog!("[matrix-bot] Waiting for other side to start SAS…");
            let stream = request.changes();
            tokio::pin!(stream);
            while let Some(state) = stream.next().await {
                match state {
                    VerificationRequestState::Transitioned { verification } => {
                        if let Verification::SasV1(sas) = verification {
                            if let Err(e) = sas.accept().await {
                                slog!("[matrix-bot] Failed to accept SAS: {e}");
                                return;
                            }
                            handle_sas_verification(sas).await;
                        }
                        break;
                    }
                    VerificationRequestState::Done
                    | VerificationRequestState::Cancelled(_) => break,
                    _ => {}
                }
            }
        }
        Err(e) => {
            slog!("[matrix-bot] Failed to start SAS verification: {e}");
        }
    }
}

/// Drive a SAS verification to completion: wait for the key exchange, log
/// the emoji comparison string, auto-confirm, and report the outcome.
async fn handle_sas_verification(sas: SasVerification) {
    slog!(
        "[matrix-bot] SAS verification in progress with {}",
        sas.other_user_id()
    );

    let stream = sas.changes();
    tokio::pin!(stream);
    while let Some(state) = stream.next().await {
        match state {
            SasState::KeysExchanged { emojis, .. } => {
                if let Some(emoji_sas) = emojis {
                    slog!(
                        "[matrix-bot] SAS verification emojis:\n{}",
                        format_emojis(emoji_sas.emojis)
                    );
                }
                if let Err(e) = sas.confirm().await {
                    slog!("[matrix-bot] Failed to confirm SAS: {e}");
                    return;
                }
            }
            SasState::Done { .. } => {
                slog!(
                    "[matrix-bot] Verification with {} completed successfully!",
                    sas.other_user_id()
                );
                break;
            }
            SasState::Cancelled(info) => {
                slog!("[matrix-bot] Verification cancelled: {info:?}");
                break;
            }
            _ => {}
        }
    }
}

// ---------------------------------------------------------------------------
// Event handler
// ---------------------------------------------------------------------------

/// Matrix event handler for room messages. Each invocation spawns an
/// independent task so the sync loop is not blocked by LLM calls.
async fn on_room_message(
    ev: OriginalSyncRoomMessageEvent,
    room: Room,
    client: Client,
    Ctx(ctx): Ctx<BotContext>,
) {
    let incoming_room_id = room.room_id().to_owned();

    slog!(
        "[matrix-bot] Event received: room={} sender={}",
        incoming_room_id,
        ev.sender,
    );

    // Only handle messages from rooms we are configured to listen in.
    if !ctx.target_room_ids.iter().any(|r| r == &incoming_room_id) {
        slog!("[matrix-bot] Ignoring message from unconfigured room {incoming_room_id}");
        return;
    }

    // Ignore the bot's own messages to prevent echo loops.
    if ev.sender == ctx.bot_user_id {
        return;
    }

    // Only respond to users on the allowlist (fail-closed).
    if !ctx.allowed_users.iter().any(|u| u == ev.sender.as_str()) {
        slog!(
            "[matrix-bot] Ignoring message from unauthorised user: {}",
            ev.sender
        );
        return;
    }

    // Only handle plain text messages.
    let (body, formatted_body) = match &ev.content.msgtype {
        MessageType::Text(t) => (t.body.clone(), t.formatted.as_ref().map(|f| f.body.clone())),
        _ => return,
    };

    // Only respond when the bot is directly addressed (mentioned by name/ID),
    // when the message is a reply to one of the bot's own messages, or when
    // ambient mode is enabled for this room.
    let is_addressed = mentions_bot(&body, formatted_body.as_deref(), &ctx.bot_user_id)
        || is_reply_to_bot(ev.content.relates_to.as_ref(), &ctx.bot_sent_event_ids).await;
    let is_ambient = ctx.ambient_rooms.lock().await.contains(&incoming_room_id);

    if !is_addressed && !is_ambient {
        slog!(
            "[matrix-bot] Ignoring unaddressed message from {}",
            ev.sender
        );
        return;
    }

    // Reject commands from unencrypted rooms — E2EE is mandatory.
    if !room.encryption_state().is_encrypted() {
        slog!(
            "[matrix-bot] Rejecting message from {} — room {} is not encrypted. \
             Commands are only accepted from encrypted rooms.",
            ev.sender,
            incoming_room_id
        );
        return;
    }

    // Always verify that the sender has a cross-signing identity.
    // This check is unconditional and cannot be disabled via config.
    match check_sender_verified(&client, &ev.sender).await {
        Ok(true) => { /* sender has a cross-signing identity — proceed */ }
        Ok(false) => {
            slog!(
                "[matrix-bot] Rejecting message from {} — no cross-signing identity \
                 found in encrypted room {}",
                ev.sender,
                incoming_room_id
            );
            return;
        }
        Err(e) => {
            slog!(
                "[matrix-bot] Error checking verification for {}: {e} — \
                 rejecting message (fail-closed)",
                ev.sender
            );
            return;
        }
    }

    // If there is a pending permission prompt for this room, interpret the
    // message as a yes/no response instead of starting a new chat.
    {
        let mut pending = ctx.pending_perm_replies.lock().await;
        if let Some(tx) = pending.remove(&incoming_room_id) {
            let decision = if is_permission_approval(&body) {
                PermissionDecision::Approve
            } else {
                PermissionDecision::Deny
            };
            let _ = tx.send(decision);
            let confirmation = if decision == PermissionDecision::Approve {
                "Permission approved."
            } else {
                "Permission denied."
            };
            let html = markdown_to_html(confirmation);
            if let Ok(resp) = room
                .send(RoomMessageEventContent::text_html(confirmation, html))
                .await
            {
                ctx.bot_sent_event_ids.lock().await.insert(resp.event_id);
            }
            return;
        }
    }

    // Check for ambient mode toggle commands.  Commands are only recognised
    // from addressed messages so they can't be accidentally triggered by
    // ambient-mode traffic from other users.
    let ambient_cmd = is_addressed
        .then(|| parse_ambient_command(&body, &ctx.bot_user_id, &ctx.bot_name))
        .flatten();
    if let Some(enable) = ambient_cmd {
        let ambient_room_ids: Vec<String> = {
            let mut ambient = ctx.ambient_rooms.lock().await;
            if enable {
                ambient.insert(incoming_room_id.clone());
            } else {
                ambient.remove(&incoming_room_id);
            }
            ambient.iter().map(|r| r.to_string()).collect()
        }; // lock released before the async send below

        // Persist updated ambient rooms to bot.toml so the state survives restarts.
        save_ambient_rooms(&ctx.project_root, &ambient_room_ids);

        let confirmation = if enable {
            "Ambient mode on. I'll respond to all messages in this room."
        } else {
            "Ambient mode off. I'll only respond when mentioned."
        };
        let html = markdown_to_html(confirmation);
        if let Ok(resp) = room
            .send(RoomMessageEventContent::text_html(confirmation, html))
            .await
        {
            ctx.bot_sent_event_ids.lock().await.insert(resp.event_id);
        }
        slog!(
            "[matrix-bot] Ambient mode {} for room {}",
            if enable { "enabled" } else { "disabled" },
            incoming_room_id
        );
        return;
    }

    let sender = ev.sender.to_string();
    let user_message = body;
    slog!("[matrix-bot] Message from {sender}: {user_message}");

    // Check for bot-level commands (e.g. "help") before invoking the LLM.
    if let Some(response) = super::commands::try_handle_command(
        &ctx.bot_name,
        ctx.bot_user_id.as_str(),
        &user_message,
    ) {
        slog!("[matrix-bot] Handled bot command from {sender}");
        let html = markdown_to_html(&response);
        if let Ok(resp) = room
            .send(RoomMessageEventContent::text_html(response, html))
            .await
        {
            ctx.bot_sent_event_ids.lock().await.insert(resp.event_id);
        }
        return;
    }

    // Spawn a separate task so the Matrix sync loop is not blocked while we
    // wait for the LLM response (which can take several seconds).
    tokio::spawn(async move {
        handle_message(room, incoming_room_id, ctx, sender, user_message).await;
    });
}

// ---------------------------------------------------------------------------
// Message handler
// ---------------------------------------------------------------------------

/// Build the user-facing prompt for a single turn.  In multi-user rooms the
/// sender is included so the LLM can distinguish participants.
fn format_user_prompt(sender: &str, message: &str) -> String {
    format!("{sender}: {message}")
}

async fn handle_message(
    room: Room,
    room_id: OwnedRoomId,
    ctx: BotContext,
    sender: String,
    user_message: String,
) {
    // Handle built-in commands before invoking Claude.
    if let Some(cmd) = extract_command(&user_message, &ctx.bot_name, &ctx.bot_user_id)
        && cmd == "status"
    {
        let project_root = ctx.project_root.clone();
        let status_text = build_pipeline_status(&project_root, &ctx.agents);
        let html = markdown_to_html(&status_text);
        if let Ok(resp) = room
            .send(RoomMessageEventContent::text_html(status_text, html))
            .await
        {
            ctx.bot_sent_event_ids.lock().await.insert(resp.event_id);
        }
        return;
    }

    // Look up the room's existing Claude Code session ID (if any) so we can
    // resume the conversation with structured API messages instead of
    // flattening history into a text prefix.
    let resume_session_id: Option<String> = {
        let guard = ctx.history.lock().await;
        guard
            .get(&room_id)
            .and_then(|conv| conv.session_id.clone())
    };

    // The prompt is just the current message with sender attribution.
    // Prior conversation context is carried by the Claude Code session.
    let bot_name = &ctx.bot_name;
    let prompt = format!(
        "[Your name is {bot_name}. Refer to yourself as {bot_name}, not Claude.]\n\n{}",
        format_user_prompt(&sender, &user_message)
    );

    let provider = ClaudeCodeProvider::new();
    let (cancel_tx, mut cancel_rx) = watch::channel(false);
    // Keep the sender alive for the duration of the call.
    let _cancel_tx = cancel_tx;

    // Channel for sending complete paragraphs to the Matrix posting task.
    let (msg_tx, mut msg_rx) = tokio::sync::mpsc::unbounded_channel::<String>();
    let msg_tx_for_callback = msg_tx.clone();

    // Spawn a task to post messages to Matrix as they arrive so we don't
    // block the LLM stream while waiting for Matrix send round-trips.
    let post_room = room.clone();
    let sent_ids = Arc::clone(&ctx.bot_sent_event_ids);
    let sent_ids_for_post = Arc::clone(&sent_ids);
    let post_task = tokio::spawn(async move {
        while let Some(chunk) = msg_rx.recv().await {
            let html = markdown_to_html(&chunk);
            if let Ok(response) = post_room
                .send(RoomMessageEventContent::text_html(chunk, html))
                .await
            {
                sent_ids_for_post.lock().await.insert(response.event_id);
            }
        }
    });

    // Shared state between the sync token callback and the async outer scope.
    let buffer = Arc::new(std::sync::Mutex::new(String::new()));
    let buffer_for_callback = Arc::clone(&buffer);
    let sent_any_chunk = Arc::new(AtomicBool::new(false));
    let sent_any_chunk_for_callback = Arc::clone(&sent_any_chunk);

    let project_root_str = ctx.project_root.to_string_lossy().to_string();
    let chat_fut = provider.chat_stream(
        &prompt,
        &project_root_str,
        resume_session_id.as_deref(),
        None,
        &mut cancel_rx,
        move |token| {
            let mut buf = buffer_for_callback.lock().unwrap();
            buf.push_str(token);
            // Flush complete paragraphs as they arrive.
            let paragraphs = drain_complete_paragraphs(&mut buf);
            for chunk in paragraphs {
                sent_any_chunk_for_callback.store(true, Ordering::Relaxed);
                let _ = msg_tx_for_callback.send(chunk);
            }
        },
        |_thinking| {}, // Discard thinking tokens
        |_activity| {}, // Discard activity signals
    );
    tokio::pin!(chat_fut);

    // Lock the permission receiver for the duration of this chat session.
    // Permission requests from the MCP `prompt_permission` tool arrive here.
    let mut perm_rx_guard = ctx.perm_rx.lock().await;

    let result = loop {
        tokio::select! {
            r = &mut chat_fut => break r,

            Some(perm_fwd) = perm_rx_guard.recv() => {
                // Post the permission prompt to the Matrix room.
                let prompt_msg = format!(
                    "**Permission Request**\n\n\
                     Tool: `{}`\n```json\n{}\n```\n\n\
                     Reply **yes** to approve or **no** to deny.",
                    perm_fwd.tool_name,
                    serde_json::to_string_pretty(&perm_fwd.tool_input)
                        .unwrap_or_else(|_| perm_fwd.tool_input.to_string()),
                );
                let html = markdown_to_html(&prompt_msg);
                if let Ok(resp) = room
                    .send(RoomMessageEventContent::text_html(&prompt_msg, html))
                    .await
                {
                    sent_ids.lock().await.insert(resp.event_id);
                }

                // Store the MCP oneshot sender so the event handler can
                // resolve it when the user replies yes/no.
                ctx.pending_perm_replies
                    .lock()
                    .await
                    .insert(room_id.clone(), perm_fwd.response_tx);

                // Spawn a timeout task: auto-deny if the user does not respond.
                let pending = Arc::clone(&ctx.pending_perm_replies);
                let timeout_room_id = room_id.clone();
                let timeout_room = room.clone();
                let timeout_sent_ids = Arc::clone(&ctx.bot_sent_event_ids);
                let timeout_secs = ctx.permission_timeout_secs;
                tokio::spawn(async move {
                    tokio::time::sleep(Duration::from_secs(timeout_secs)).await;
                    if let Some(tx) = pending.lock().await.remove(&timeout_room_id) {
                        let _ = tx.send(PermissionDecision::Deny);
                        let msg = "Permission request timed out — denied (fail-closed).";
                        let html = markdown_to_html(msg);
                        if let Ok(resp) = timeout_room
                            .send(RoomMessageEventContent::text_html(msg, html))
                            .await
                        {
                            timeout_sent_ids.lock().await.insert(resp.event_id);
                        }
                    }
                });
            }
        }
    };
    drop(perm_rx_guard);

    // Flush any remaining text that didn't end with a paragraph boundary.
    let remaining = buffer.lock().unwrap().trim().to_string();
    let did_send_any = sent_any_chunk.load(Ordering::Relaxed);

    let (assistant_reply, new_session_id) = match result {
        Ok(ClaudeCodeResult {
            messages,
            session_id,
        }) => {
            let reply = if !remaining.is_empty() {
                let _ = msg_tx.send(remaining.clone());
                remaining
            } else if !did_send_any {
                // Nothing was streamed at all (e.g. only tool calls with no
                // final text) — fall back to the last assistant message from
                // the structured result.
                let last_text = messages
                    .iter()
                    .rev()
                    .find(|m| m.role == crate::llm::types::Role::Assistant && !m.content.is_empty())
                    .map(|m| m.content.clone())
                    .unwrap_or_default();
                if !last_text.is_empty() {
                    let _ = msg_tx.send(last_text.clone());
                }
                last_text
            } else {
                remaining
            };
            slog!("[matrix-bot] session_id from chat_stream: {:?}", session_id);
            (reply, session_id)
        }
        Err(e) => {
            slog!("[matrix-bot] LLM error: {e}");
            let err_msg = format!("Error processing your request: {e}");
            let _ = msg_tx.send(err_msg.clone());
            (err_msg, None)
        }
    };

    // Drop the sender to signal the posting task that no more messages will
    // arrive, then wait for all pending Matrix sends to complete.
    drop(msg_tx);
    let _ = post_task.await;

    // Record this exchange in the per-room conversation history and persist
    // the session ID so the next turn resumes with structured API messages.
    if !assistant_reply.starts_with("Error processing") {
        let mut guard = ctx.history.lock().await;
        let conv = guard.entry(room_id).or_default();

        // Store the session ID so the next turn uses --resume.
        slog!("[matrix-bot] storing session_id: {:?} (was: {:?})", new_session_id, conv.session_id);
        if new_session_id.is_some() {
            conv.session_id = new_session_id;
        }

        conv.entries.push(ConversationEntry {
            role: ConversationRole::User,
            sender: sender.clone(),
            content: user_message,
        });
        conv.entries.push(ConversationEntry {
            role: ConversationRole::Assistant,
            sender: String::new(),
            content: assistant_reply,
        });

        // Trim to the configured maximum, dropping the oldest entries first.
        // The session_id is preserved: Claude Code's --resume loads the full
        // conversation from its own session transcript on disk, so trimming
        // our local tracking doesn't affect the LLM's context.
        if conv.entries.len() > ctx.history_size {
            let excess = conv.entries.len() - ctx.history_size;
            conv.entries.drain(..excess);
        }

        // Persist to disk so history survives server restarts.
        save_history(&ctx.project_root, &guard);
    }
}

// ---------------------------------------------------------------------------
// Markdown rendering helper
// ---------------------------------------------------------------------------

/// Convert a Markdown string to an HTML string using pulldown-cmark.
///
/// Enables the standard extension set (tables, footnotes, strikethrough,
/// tasklists) so that common Markdown constructs render correctly in Matrix
/// clients such as Element.
pub fn markdown_to_html(markdown: &str) -> String {
    let options = Options::ENABLE_TABLES
        | Options::ENABLE_FOOTNOTES
        | Options::ENABLE_STRIKETHROUGH
        | Options::ENABLE_TASKLISTS;
    let parser = Parser::new_ext(markdown, options);
    let mut html_output = String::new();
    html::push_html(&mut html_output, parser);
    html_output
}

// ---------------------------------------------------------------------------
// Paragraph buffering helper
// ---------------------------------------------------------------------------

/// Returns `true` when `text` ends while inside an open fenced code block.
///
/// A fenced code block opens and closes on lines that start with ` ``` `
/// (three or more backticks).  We count the fence markers and return `true`
/// when the count is odd (a fence was opened but not yet closed).
fn is_inside_code_fence(text: &str) -> bool {
    let mut in_fence = false;
    for line in text.lines() {
        if line.trim_start().starts_with("```") {
            in_fence = !in_fence;
        }
    }
    in_fence
}

/// Drain all complete paragraphs from `buffer` and return them.
///
/// A paragraph boundary is a double newline (`\n\n`). Each drained paragraph
/// is trimmed of surrounding whitespace; empty paragraphs are discarded.
/// The buffer is left with only the remaining incomplete text.
///
/// **Code-fence awareness:** a `\n\n` that occurs *inside* a fenced code
/// block (delimited by ` ``` ` lines) is **not** treated as a paragraph
/// boundary.  This prevents a blank line inside a code block from splitting
/// the fence across multiple Matrix messages, which would corrupt the
/// rendering of the second half.
pub fn drain_complete_paragraphs(buffer: &mut String) -> Vec<String> {
    let mut paragraphs = Vec::new();
    let mut search_from = 0;
    loop {
        let Some(pos) = buffer[search_from..].find("\n\n") else {
            break;
        };
        let abs_pos = search_from + pos;
        // Only split at this boundary when we are NOT inside a code fence.
        if is_inside_code_fence(&buffer[..abs_pos]) {
            // Skip past this \n\n and keep looking for the next boundary.
            search_from = abs_pos + 2;
        } else {
            let chunk = buffer[..abs_pos].trim().to_string();
            *buffer = buffer[abs_pos + 2..].to_string();
            search_from = 0;
            if !chunk.is_empty() {
                paragraphs.push(chunk);
            }
        }
    }
    paragraphs
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;

    // -- mentions_bot -------------------------------------------------------

    fn make_user_id(s: &str) -> OwnedUserId {
        s.parse().unwrap()
    }

    #[test]
    fn mentions_bot_by_full_id() {
        let uid = make_user_id("@timmy:homeserver.local");
        assert!(mentions_bot(
            "hello @timmy:homeserver.local can you help?",
            None,
            &uid
        ));
    }

    #[test]
    fn mentions_bot_by_localpart_at_start() {
        let uid = make_user_id("@timmy:homeserver.local");
        assert!(mentions_bot("@timmy please list open stories", None, &uid));
    }

    #[test]
    fn mentions_bot_by_localpart_mid_sentence() {
        let uid = make_user_id("@timmy:homeserver.local");
        assert!(mentions_bot("hey @timmy what's the status?", None, &uid));
    }

    #[test]
    fn mentions_bot_not_mentioned() {
        let uid = make_user_id("@timmy:homeserver.local");
        assert!(!mentions_bot(
            "can someone help me with this PR?",
            None,
            &uid
        ));
    }

    #[test]
    fn mentions_bot_no_false_positive_longer_username() {
        // "@timmybot" must NOT match "@timmy"
        let uid = make_user_id("@timmy:homeserver.local");
        assert!(!mentions_bot("hey @timmybot can you help?", None, &uid));
    }

    #[test]
    fn mentions_bot_at_end_of_string() {
        let uid = make_user_id("@timmy:homeserver.local");
        assert!(mentions_bot("shoutout to @timmy", None, &uid));
    }

    #[test]
    fn mentions_bot_followed_by_comma() {
        let uid = make_user_id("@timmy:homeserver.local");
        assert!(mentions_bot("@timmy, can you help?", None, &uid));
    }

    // -- is_reply_to_bot ----------------------------------------------------

    #[tokio::test]
    async fn is_reply_to_bot_direct_reply_match() {
        let sent: Arc<TokioMutex<HashSet<OwnedEventId>>> =
            Arc::new(TokioMutex::new(HashSet::new()));
        let event_id: OwnedEventId = "$abc123:example.com".parse().unwrap();
        sent.lock().await.insert(event_id.clone());

        let in_reply_to = matrix_sdk::ruma::events::relation::InReplyTo::new(event_id);
        let relates_to: Option<Relation<RoomMessageEventContentWithoutRelation>> =
            Some(Relation::Reply { in_reply_to });

        assert!(is_reply_to_bot(relates_to.as_ref(), &sent).await);
    }

    #[tokio::test]
    async fn is_reply_to_bot_direct_reply_no_match() {
        let sent: Arc<TokioMutex<HashSet<OwnedEventId>>> =
            Arc::new(TokioMutex::new(HashSet::new()));
        // sent is empty — this event was not sent by the bot

        let in_reply_to = matrix_sdk::ruma::events::relation::InReplyTo::new(
            "$other:example.com".parse::<OwnedEventId>().unwrap(),
        );
        let relates_to: Option<Relation<RoomMessageEventContentWithoutRelation>> =
            Some(Relation::Reply { in_reply_to });

        assert!(!is_reply_to_bot(relates_to.as_ref(), &sent).await);
    }

    #[tokio::test]
    async fn is_reply_to_bot_no_relation() {
        let sent: Arc<TokioMutex<HashSet<OwnedEventId>>> =
            Arc::new(TokioMutex::new(HashSet::new()));
        let relates_to: Option<Relation<RoomMessageEventContentWithoutRelation>> = None;
        assert!(!is_reply_to_bot(relates_to.as_ref(), &sent).await);
    }

    #[tokio::test]
    async fn is_reply_to_bot_thread_root_match() {
        let sent: Arc<TokioMutex<HashSet<OwnedEventId>>> =
            Arc::new(TokioMutex::new(HashSet::new()));
        let root_id: OwnedEventId = "$root123:example.com".parse().unwrap();
        sent.lock().await.insert(root_id.clone());

        // Thread reply where the thread root is the bot's message
        let thread = matrix_sdk::ruma::events::relation::Thread::plain(
            root_id,
            "$latest:example.com".parse::<OwnedEventId>().unwrap(),
        );
        let relates_to: Option<Relation<RoomMessageEventContentWithoutRelation>> =
            Some(Relation::Thread(thread));

        assert!(is_reply_to_bot(relates_to.as_ref(), &sent).await);
    }

    // -- markdown_to_html ---------------------------------------------------

    #[test]
    fn markdown_to_html_bold() {
        let html = markdown_to_html("**bold**");
        assert!(
            html.contains("<strong>bold</strong>"),
            "expected <strong>: {html}"
        );
    }

    #[test]
    fn markdown_to_html_unordered_list() {
        let html = markdown_to_html("- item one\n- item two");
        assert!(html.contains("<ul>"), "expected <ul>: {html}");
        assert!(
            html.contains("<li>item one</li>"),
            "expected list item: {html}"
        );
    }

    #[test]
    fn markdown_to_html_inline_code() {
        let html = markdown_to_html("`inline_code()`");
        assert!(
            html.contains("<code>inline_code()</code>"),
            "expected <code>: {html}"
        );
    }

    #[test]
    fn markdown_to_html_code_block() {
        let html = markdown_to_html("```rust\nfn main() {}\n```");
        assert!(html.contains("<pre>"), "expected <pre>: {html}");
        assert!(html.contains("<code"), "expected <code> inside pre: {html}");
        assert!(
            html.contains("fn main() {}"),
            "expected code content: {html}"
        );
    }

    #[test]
    fn markdown_to_html_plain_text_passthrough() {
        let html = markdown_to_html("Hello, world!");
        assert!(
            html.contains("Hello, world!"),
            "expected plain text passthrough: {html}"
        );
    }

    // -- bot_context_is_clone -----------------------------------------------

    #[test]
    fn bot_context_is_clone() {
        // BotContext must be Clone for the Matrix event handler injection.
        fn assert_clone<T: Clone>() {}
        assert_clone::<BotContext>();
    }

    #[test]
    fn bot_context_has_no_require_verified_devices_field() {
        // Verification is always on — BotContext no longer has a toggle field.
        // This test verifies the struct can be constructed and cloned without it.
        let (_perm_tx, perm_rx) = mpsc::unbounded_channel();
        let ctx = BotContext {
            bot_user_id: make_user_id("@bot:example.com"),
            target_room_ids: vec![],
            project_root: PathBuf::from("/tmp"),
            allowed_users: vec![],
            history: Arc::new(TokioMutex::new(HashMap::new())),
            history_size: 20,
            bot_sent_event_ids: Arc::new(TokioMutex::new(HashSet::new())),
            perm_rx: Arc::new(TokioMutex::new(perm_rx)),
            pending_perm_replies: Arc::new(TokioMutex::new(HashMap::new())),
            permission_timeout_secs: 120,
            bot_name: "Assistant".to_string(),
            ambient_rooms: Arc::new(TokioMutex::new(HashSet::new())),
            agents: Arc::new(AgentPool::new_test(3000)),
        };
        // Clone must work (required by Matrix SDK event handler injection).
        let _cloned = ctx.clone();
    }

    // -- drain_complete_paragraphs ------------------------------------------

    #[test]
    fn drain_complete_paragraphs_no_boundary_returns_empty() {
        let mut buf = "Hello World".to_string();
        let paras = drain_complete_paragraphs(&mut buf);
        assert!(paras.is_empty());
        assert_eq!(buf, "Hello World");
    }

    #[test]
    fn drain_complete_paragraphs_single_boundary() {
        let mut buf = "Paragraph one.\n\nParagraph two.".to_string();
        let paras = drain_complete_paragraphs(&mut buf);
        assert_eq!(paras, vec!["Paragraph one."]);
        assert_eq!(buf, "Paragraph two.");
    }

    #[test]
    fn drain_complete_paragraphs_multiple_boundaries() {
        let mut buf = "A\n\nB\n\nC".to_string();
        let paras = drain_complete_paragraphs(&mut buf);
        assert_eq!(paras, vec!["A", "B"]);
        assert_eq!(buf, "C");
    }

    #[test]
    fn drain_complete_paragraphs_trailing_boundary() {
        let mut buf = "A\n\nB\n\n".to_string();
        let paras = drain_complete_paragraphs(&mut buf);
        assert_eq!(paras, vec!["A", "B"]);
        assert_eq!(buf, "");
    }

    #[test]
    fn drain_complete_paragraphs_empty_input() {
        let mut buf = String::new();
        let paras = drain_complete_paragraphs(&mut buf);
        assert!(paras.is_empty());
        assert_eq!(buf, "");
    }

    #[test]
    fn drain_complete_paragraphs_skips_empty_chunks() {
        // Consecutive double-newlines produce no empty paragraphs.
        let mut buf = "\n\n\n\nHello".to_string();
        let paras = drain_complete_paragraphs(&mut buf);
        assert!(paras.is_empty());
        assert_eq!(buf, "Hello");
    }

    #[test]
    fn drain_complete_paragraphs_trims_whitespace() {
        let mut buf = "  Hello  \n\n  World  ".to_string();
        let paras = drain_complete_paragraphs(&mut buf);
        assert_eq!(paras, vec!["Hello"]);
        assert_eq!(buf, "  World  ");
    }

    // -- drain_complete_paragraphs: code-fence awareness -------------------

    #[test]
    fn drain_complete_paragraphs_code_fence_blank_line_not_split() {
        // A blank line inside a fenced code block must NOT trigger a split.
        // Before the fix the function would split at the blank line and the
        // second half would be sent without the opening fence, breaking rendering.
        let mut buf =
            "```rust\nfn foo() {\n    let x = 1;\n\n    let y = 2;\n}\n```\n\nNext paragraph."
                .to_string();
        let paras = drain_complete_paragraphs(&mut buf);
        assert_eq!(
            paras.len(),
            1,
            "code fence with blank line should not be split into multiple messages: {paras:?}"
        );
        assert!(
            paras[0].starts_with("```rust"),
            "first paragraph should be the code fence: {:?}",
            paras[0]
        );
        assert!(
            paras[0].contains("let y = 2;"),
            "code fence should contain content from both sides of the blank line: {:?}",
            paras[0]
        );
        assert_eq!(buf, "Next paragraph.");
    }

    #[test]
    fn drain_complete_paragraphs_text_before_and_after_fenced_block() {
        // Text paragraph, then a code block with an internal blank line, then more text.
        let mut buf = "Before\n\n```\ncode\n\nmore code\n```\n\nAfter".to_string();
        let paras = drain_complete_paragraphs(&mut buf);
        assert_eq!(paras.len(), 2, "expected two paragraphs: {paras:?}");
        assert_eq!(paras[0], "Before");
        assert!(
            paras[1].starts_with("```"),
            "second paragraph should be the code fence: {:?}",
            paras[1]
        );
        assert!(
            paras[1].contains("more code"),
            "code fence content must include the part after the blank line: {:?}",
            paras[1]
        );
        assert_eq!(buf, "After");
    }

    #[test]
    fn drain_complete_paragraphs_incremental_simulation() {
        // Simulate tokens arriving one character at a time.
        let mut buf = String::new();
        let mut all_paragraphs = Vec::new();

        for ch in "First para.\n\nSecond para.\n\nThird.".chars() {
            buf.push(ch);
            all_paragraphs.extend(drain_complete_paragraphs(&mut buf));
        }

        assert_eq!(all_paragraphs, vec!["First para.", "Second para."]);
        assert_eq!(buf, "Third.");
    }

    // -- format_user_prompt -------------------------------------------------

    #[test]
    fn format_user_prompt_includes_sender_and_message() {
        let prompt = format_user_prompt("@alice:example.com", "Hello!");
        assert_eq!(prompt, "@alice:example.com: Hello!");
    }

    #[test]
    fn format_user_prompt_different_users() {
        let prompt = format_user_prompt("@bob:example.com", "What's up?");
        assert_eq!(prompt, "@bob:example.com: What's up?");
    }

    // -- conversation history trimming --------------------------------------

    #[tokio::test]
    async fn history_trims_to_configured_size() {
        let history: ConversationHistory = Arc::new(TokioMutex::new(HashMap::new()));
        let room_id: OwnedRoomId = "!test:example.com".parse().unwrap();
        let history_size = 4usize; // keep at most 4 entries

        // Add 6 entries (3 user + 3 assistant turns).
        {
            let mut guard = history.lock().await;
            let conv = guard.entry(room_id.clone()).or_default();
            conv.session_id = Some("test-session".to_string());
            for i in 0..3usize {
                conv.entries.push(ConversationEntry {
                    role: ConversationRole::User,
                    sender: "@user:example.com".to_string(),
                    content: format!("msg {i}"),
                });
                conv.entries.push(ConversationEntry {
                    role: ConversationRole::Assistant,
                    sender: String::new(),
                    content: format!("reply {i}"),
                });
                if conv.entries.len() > history_size {
                    let excess = conv.entries.len() - history_size;
                    conv.entries.drain(..excess);
                    conv.session_id = None;
                }
            }
        }

        let guard = history.lock().await;
        let conv = guard.get(&room_id).unwrap();
        assert_eq!(
            conv.entries.len(),
            history_size,
            "history must be trimmed to history_size"
        );
        // The oldest entries (msg 0 / reply 0) should have been dropped.
        assert!(
            conv.entries.iter().all(|e| !e.content.contains("msg 0")),
            "oldest entries must be dropped"
        );
        // Session ID must be cleared when trimming occurs.
        assert!(
            conv.session_id.is_none(),
            "session_id must be cleared on trim to start a fresh session"
        );
    }

    #[tokio::test]
    async fn each_room_has_independent_history() {
        let history: ConversationHistory = Arc::new(TokioMutex::new(HashMap::new()));
        let room_a: OwnedRoomId = "!room_a:example.com".parse().unwrap();
        let room_b: OwnedRoomId = "!room_b:example.com".parse().unwrap();

        {
            let mut guard = history.lock().await;
            guard
                .entry(room_a.clone())
                .or_default()
                .entries
                .push(ConversationEntry {
                    role: ConversationRole::User,
                    sender: "@alice:example.com".to_string(),
                    content: "Room A message".to_string(),
                });
            guard
                .entry(room_b.clone())
                .or_default()
                .entries
                .push(ConversationEntry {
                    role: ConversationRole::User,
                    sender: "@bob:example.com".to_string(),
                    content: "Room B message".to_string(),
                });
        }

        let guard = history.lock().await;
        let conv_a = guard.get(&room_a).unwrap();
        let conv_b = guard.get(&room_b).unwrap();
        assert_eq!(conv_a.entries.len(), 1);
        assert_eq!(conv_b.entries.len(), 1);
        assert_eq!(conv_a.entries[0].content, "Room A message");
        assert_eq!(conv_b.entries[0].content, "Room B message");
    }

    // -- persistence --------------------------------------------------------

    #[test]
    fn save_and_load_history_round_trip() {
        let dir = tempfile::tempdir().unwrap();
        let story_kit_dir = dir.path().join(".story_kit");
        std::fs::create_dir_all(&story_kit_dir).unwrap();

        let room_id: OwnedRoomId = "!persist:example.com".parse().unwrap();
        let mut map: HashMap<OwnedRoomId, RoomConversation> = HashMap::new();
        let conv = map.entry(room_id.clone()).or_default();
        conv.session_id = Some("session-abc".to_string());
        conv.entries.push(ConversationEntry {
            role: ConversationRole::User,
            sender: "@alice:example.com".to_string(),
            content: "hello".to_string(),
        });
        conv.entries.push(ConversationEntry {
            role: ConversationRole::Assistant,
            sender: String::new(),
            content: "hi there!".to_string(),
        });

        save_history(dir.path(), &map);

        let loaded = load_history(dir.path());
        let loaded_conv = loaded.get(&room_id).expect("room must exist after load");
        assert_eq!(loaded_conv.session_id.as_deref(), Some("session-abc"));
        assert_eq!(loaded_conv.entries.len(), 2);
        assert_eq!(loaded_conv.entries[0].role, ConversationRole::User);
        assert_eq!(loaded_conv.entries[0].sender, "@alice:example.com");
        assert_eq!(loaded_conv.entries[0].content, "hello");
        assert_eq!(loaded_conv.entries[1].role, ConversationRole::Assistant);
        assert_eq!(loaded_conv.entries[1].content, "hi there!");
    }

    #[test]
    fn load_history_returns_empty_on_missing_file() {
        let dir = tempfile::tempdir().unwrap();
        let loaded = load_history(dir.path());
        assert!(loaded.is_empty());
    }

    #[test]
    fn load_history_returns_empty_on_corrupt_file() {
        let dir = tempfile::tempdir().unwrap();
        let story_kit_dir = dir.path().join(".story_kit");
        std::fs::create_dir_all(&story_kit_dir).unwrap();
        std::fs::write(dir.path().join(HISTORY_FILE), "not valid json").unwrap();
        let loaded = load_history(dir.path());
        assert!(loaded.is_empty());
    }

    // -- session_id tracking ------------------------------------------------

    #[tokio::test]
    async fn session_id_preserved_within_history_size() {
        let history: ConversationHistory = Arc::new(TokioMutex::new(HashMap::new()));
        let room_id: OwnedRoomId = "!session:example.com".parse().unwrap();

        {
            let mut guard = history.lock().await;
            let conv = guard.entry(room_id.clone()).or_default();
            conv.session_id = Some("sess-1".to_string());
            conv.entries.push(ConversationEntry {
                role: ConversationRole::User,
                sender: "@alice:example.com".to_string(),
                content: "hello".to_string(),
            });
            conv.entries.push(ConversationEntry {
                role: ConversationRole::Assistant,
                sender: String::new(),
                content: "hi".to_string(),
            });
            // No trimming needed (2 entries, well under any reasonable limit).
        }

        let guard = history.lock().await;
        let conv = guard.get(&room_id).unwrap();
        assert_eq!(
            conv.session_id.as_deref(),
            Some("sess-1"),
            "session_id must be preserved when no trimming occurs"
        );
    }

    // -- multi-user room attribution ----------------------------------------

    #[tokio::test]
    async fn multi_user_entries_preserve_sender() {
        let history: ConversationHistory = Arc::new(TokioMutex::new(HashMap::new()));
        let room_id: OwnedRoomId = "!multi:example.com".parse().unwrap();

        {
            let mut guard = history.lock().await;
            let conv = guard.entry(room_id.clone()).or_default();
            conv.entries.push(ConversationEntry {
                role: ConversationRole::User,
                sender: "@alice:example.com".to_string(),
                content: "from alice".to_string(),
            });
            conv.entries.push(ConversationEntry {
                role: ConversationRole::Assistant,
                sender: String::new(),
                content: "reply to alice".to_string(),
            });
            conv.entries.push(ConversationEntry {
                role: ConversationRole::User,
                sender: "@bob:example.com".to_string(),
                content: "from bob".to_string(),
            });
        }

        let guard = history.lock().await;
        let conv = guard.get(&room_id).unwrap();
        assert_eq!(conv.entries[0].sender, "@alice:example.com");
        assert_eq!(conv.entries[2].sender, "@bob:example.com");
    }

    // -- self-sign device key decision logic -----------------------------------

    // The self-signing logic in run_bot cannot be unit-tested because it
    // requires a live matrix_sdk::Client.  The tests below verify the branch
    // decision: sign only when the device is NOT already cross-signed.

    #[test]
    fn device_already_self_signed_skips_signing() {
        // Simulates: get_own_device returns Some, is_cross_signed_by_owner → true
        let is_cross_signed: bool = true;
        assert!(
            is_cross_signed,
            "already self-signed device should skip signing"
        );
    }

    #[test]
    fn device_not_self_signed_triggers_signing() {
        // Simulates: get_own_device returns Some, is_cross_signed_by_owner → false
        let is_cross_signed: bool = false;
        assert!(
            !is_cross_signed,
            "device without self-signature should trigger signing"
        );
    }

    // -- check_sender_verified decision logic --------------------------------

    // check_sender_verified cannot be called in unit tests because it requires
    // a live matrix_sdk::Client (which in turn needs a real homeserver
    // connection and crypto store).  The tests below verify the decision logic
    // that the function implements: a user is accepted iff their cross-signing
    // identity is present in the crypto store (Some), and rejected when no
    // identity is known (None).

    #[test]
    fn sender_with_cross_signing_identity_is_accepted() {
        // Simulates: get_user_identity returns Some(_) → Ok(true)
        let identity: Option<()> = Some(());
        assert!(
            identity.is_some(),
            "user with cross-signing identity should be accepted"
        );
    }

    #[test]
    fn sender_without_cross_signing_identity_is_rejected() {
        // Simulates: get_user_identity returns None → Ok(false)
        let identity: Option<()> = None;
        assert!(
            identity.is_none(),
            "user with no cross-signing setup should be rejected"
        );
    }

    // -- is_permission_approval -----------------------------------------------

    #[test]
    fn is_permission_approval_accepts_yes_variants() {
        assert!(is_permission_approval("yes"));
        assert!(is_permission_approval("Yes"));
        assert!(is_permission_approval("YES"));
        assert!(is_permission_approval("y"));
        assert!(is_permission_approval("Y"));
        assert!(is_permission_approval("approve"));
        assert!(is_permission_approval("allow"));
        assert!(is_permission_approval("ok"));
        assert!(is_permission_approval("OK"));
    }

    #[test]
    fn is_permission_approval_denies_no_and_other() {
        assert!(!is_permission_approval("no"));
        assert!(!is_permission_approval("No"));
        assert!(!is_permission_approval("n"));
        assert!(!is_permission_approval("deny"));
        assert!(!is_permission_approval("reject"));
        assert!(!is_permission_approval("maybe"));
        assert!(!is_permission_approval(""));
        assert!(!is_permission_approval("yes please do it"));
    }

    #[test]
    fn is_permission_approval_strips_at_mention_prefix() {
        assert!(is_permission_approval("@timmy yes"));
        assert!(!is_permission_approval("@timmy no"));
    }

    #[test]
    fn is_permission_approval_handles_whitespace() {
        assert!(is_permission_approval("  yes  "));
        assert!(is_permission_approval("\tyes\n"));
    }

    // -- format_startup_announcement ----------------------------------------

    #[test]
    fn startup_announcement_uses_bot_name() {
        assert_eq!(format_startup_announcement("Timmy"), "Timmy is online.");
    }

    #[test]
    fn startup_announcement_uses_configured_display_name_not_hardcoded() {
        assert_eq!(format_startup_announcement("HAL"), "HAL is online.");
        assert_eq!(format_startup_announcement("Assistant"), "Assistant is online.");
    }

    // -- extract_command (status trigger) ------------------------------------

    #[test]
    fn extract_command_returns_status_for_bot_name_prefix() {
        let uid = make_user_id("@assistant:example.com");
        let result = extract_command("Assistant status", "Assistant", &uid);
        assert_eq!(result.as_deref(), Some("status"));
    }

    #[test]
    fn extract_command_returns_status_for_at_localpart_prefix() {
        let uid = make_user_id("@assistant:example.com");
        let result = extract_command("@assistant status", "Assistant", &uid);
        assert_eq!(result.as_deref(), Some("status"));
    }

    #[test]
    fn extract_command_returns_status_for_full_id_prefix() {
        let uid = make_user_id("@assistant:example.com");
        let result = extract_command("@assistant:example.com status", "Assistant", &uid);
        assert_eq!(result.as_deref(), Some("status"));
    }

    #[test]
    fn extract_command_returns_none_when_no_bot_mention() {
        let uid = make_user_id("@assistant:example.com");
        let result = extract_command("status", "Assistant", &uid);
        assert!(result.is_none());
    }

    #[test]
    fn extract_command_handles_punctuation_after_mention() {
        let uid = make_user_id("@assistant:example.com");
        let result = extract_command("@assistant: status", "Assistant", &uid);
        assert_eq!(result.as_deref(), Some("status"));
    }

    // -- build_pipeline_status -----------------------------------------------

    fn write_story_file(dir: &std::path::Path, stage: &str, filename: &str, name: &str) {
        let stage_dir = dir.join(".story_kit").join("work").join(stage);
        std::fs::create_dir_all(&stage_dir).unwrap();
        let content = format!("---\nname: \"{name}\"\n---\n\n# {name}\n");
        std::fs::write(stage_dir.join(filename), content).unwrap();
    }

    #[test]
    fn build_pipeline_status_includes_all_stages() {
        let dir = tempfile::tempdir().unwrap();
        let pool = AgentPool::new_test(3001);
        let out = build_pipeline_status(dir.path(), &pool);

        assert!(out.contains("Upcoming"), "missing Upcoming: {out}");
        assert!(out.contains("In Progress"), "missing In Progress: {out}");
        assert!(out.contains("QA"), "missing QA: {out}");
        assert!(out.contains("Merge"), "missing Merge: {out}");
        assert!(out.contains("Done"), "missing Done: {out}");
    }

    #[test]
    fn build_pipeline_status_shows_story_id_and_name() {
        let dir = tempfile::tempdir().unwrap();
        write_story_file(
            dir.path(),
            "1_backlog",
            "42_story_do_something.md",
            "Do Something",
        );
        let pool = AgentPool::new_test(3001);
        let out = build_pipeline_status(dir.path(), &pool);

        assert!(
            out.contains("42_story_do_something"),
            "missing story id: {out}"
        );
        assert!(out.contains("Do Something"), "missing story name: {out}");
    }

    #[test]
    fn build_pipeline_status_includes_free_agents_section() {
        let dir = tempfile::tempdir().unwrap();
        let pool = AgentPool::new_test(3001);
        let out = build_pipeline_status(dir.path(), &pool);

        assert!(out.contains("Free Agents"), "missing Free Agents section: {out}");
    }

    #[test]
    fn build_pipeline_status_uses_markdown_bold_headings() {
        let dir = tempfile::tempdir().unwrap();
        let pool = AgentPool::new_test(3001);
        let out = build_pipeline_status(dir.path(), &pool);

        // Stages and headers should use markdown bold (**text**).
        assert!(out.contains("**Pipeline Status**"), "missing bold title: {out}");
        assert!(out.contains("**Upcoming**"), "stage should use bold: {out}");
    }

    #[test]
    fn build_pipeline_status_shows_none_for_empty_stages() {
        let dir = tempfile::tempdir().unwrap();
        let pool = AgentPool::new_test(3001);
        let out = build_pipeline_status(dir.path(), &pool);

        // Empty stages show *(none)*
        assert!(out.contains("*(none)*"), "expected none marker: {out}");
    }

    // -- bot_name / system prompt -------------------------------------------

    #[test]
    fn bot_name_system_prompt_format() {
        let bot_name = "Timmy";
        let system_prompt =
            format!("Your name is {bot_name}. Refer to yourself as {bot_name}, not Claude.");
        assert_eq!(
            system_prompt,
            "Your name is Timmy. Refer to yourself as Timmy, not Claude."
        );
    }

    #[test]
    fn bot_name_defaults_to_assistant_when_display_name_absent() {
        // When display_name is not set in bot.toml, bot_name should be "Assistant".
        // This mirrors the logic in run_bot: config.display_name.clone().unwrap_or_else(...)
        fn resolve_bot_name(display_name: Option<String>) -> String {
            display_name.unwrap_or_else(|| "Assistant".to_string())
        }
        assert_eq!(resolve_bot_name(None), "Assistant");
        assert_eq!(resolve_bot_name(Some("Timmy".to_string())), "Timmy");
    }

    // -- parse_ambient_command ------------------------------------------------

    #[test]
    fn ambient_command_on_with_at_mention() {
        let uid = make_user_id("@timmy:homeserver.local");
        assert_eq!(parse_ambient_command("@timmy ambient on", &uid, "Timmy"), Some(true));
    }

    #[test]
    fn ambient_command_off_with_at_mention() {
        let uid = make_user_id("@timmy:homeserver.local");
        assert_eq!(parse_ambient_command("@timmy ambient off", &uid, "Timmy"), Some(false));
    }

    #[test]
    fn ambient_command_on_with_full_user_id() {
        let uid = make_user_id("@timmy:homeserver.local");
        assert_eq!(
            parse_ambient_command("@timmy:homeserver.local ambient on", &uid, "Timmy"),
            Some(true)
        );
    }

    #[test]
    fn ambient_command_on_with_display_name() {
        let uid = make_user_id("@timmy:homeserver.local");
        assert_eq!(parse_ambient_command("timmy ambient on", &uid, "Timmy"), Some(true));
    }

    #[test]
    fn ambient_command_off_with_display_name() {
        let uid = make_user_id("@timmy:homeserver.local");
        assert_eq!(parse_ambient_command("timmy ambient off", &uid, "Timmy"), Some(false));
    }

    #[test]
    fn ambient_command_on_bare() {
        // "ambient on" without any bot mention is also recognised.
        let uid = make_user_id("@timmy:homeserver.local");
        assert_eq!(parse_ambient_command("ambient on", &uid, "Timmy"), Some(true));
    }

    #[test]
    fn ambient_command_off_bare() {
        let uid = make_user_id("@timmy:homeserver.local");
        assert_eq!(parse_ambient_command("ambient off", &uid, "Timmy"), Some(false));
    }

    #[test]
    fn ambient_command_case_insensitive() {
        let uid = make_user_id("@timmy:homeserver.local");
        assert_eq!(parse_ambient_command("@Timmy AMBIENT ON", &uid, "Timmy"), Some(true));
        assert_eq!(parse_ambient_command("TIMMY AMBIENT OFF", &uid, "Timmy"), Some(false));
    }

    #[test]
    fn ambient_command_unrelated_message_returns_none() {
        let uid = make_user_id("@timmy:homeserver.local");
        assert_eq!(parse_ambient_command("@timmy what is the status?", &uid, "Timmy"), None);
        assert_eq!(parse_ambient_command("hello there", &uid, "Timmy"), None);
        assert_eq!(parse_ambient_command("ambient", &uid, "Timmy"), None);
    }

    // -- ambient mode state ---------------------------------------------------

    #[tokio::test]
    async fn ambient_rooms_defaults_to_empty() {
        let ambient_rooms: Arc<TokioMutex<HashSet<OwnedRoomId>>> =
            Arc::new(TokioMutex::new(HashSet::new()));
        let room_id: OwnedRoomId = "!room:example.com".parse().unwrap();
        assert!(!ambient_rooms.lock().await.contains(&room_id));
    }

    #[tokio::test]
    async fn ambient_mode_can_be_toggled_per_room() {
        let ambient_rooms: Arc<TokioMutex<HashSet<OwnedRoomId>>> =
            Arc::new(TokioMutex::new(HashSet::new()));
        let room_a: OwnedRoomId = "!room_a:example.com".parse().unwrap();
        let room_b: OwnedRoomId = "!room_b:example.com".parse().unwrap();

        // Enable ambient mode for room_a only.
        ambient_rooms.lock().await.insert(room_a.clone());

        let guard = ambient_rooms.lock().await;
        assert!(guard.contains(&room_a), "room_a should be in ambient mode");
        assert!(!guard.contains(&room_b), "room_b should NOT be in ambient mode");
        drop(guard);

        // Disable ambient mode for room_a.
        ambient_rooms.lock().await.remove(&room_a);
        assert!(!ambient_rooms.lock().await.contains(&room_a), "room_a ambient mode should be off");
    }
}