//! Matrix bot run loop — connects to the homeserver and processes sync events. use crate::service::timer::TimerStore; use crate::services::Services; use crate::slog; use matrix_sdk::ruma::OwnedRoomId; use matrix_sdk::{Client, LoopCtrl, config::SyncSettings}; use std::collections::{HashMap, HashSet}; use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU64, Ordering}; use tokio::sync::Mutex as TokioMutex; use tokio::sync::{RwLock, watch}; use super::context::BotContext; use super::format::{format_startup_announcement, markdown_to_html}; use super::history::load_history; use super::messages::on_room_message; use super::verification::{on_room_verification_request, on_to_device_verification_request}; /// 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. #[allow(clippy::too_many_arguments)] pub async fn run_bot( config: super::super::config::BotConfig, services: Arc, watcher_rx: tokio::sync::broadcast::Receiver, watcher_rx_auto: tokio::sync::broadcast::Receiver, watcher_tx: tokio::sync::broadcast::Sender, shutdown_rx: watch::Receiver>, gateway_active_project: Option>>, gateway_projects_store: Option< Arc< RwLock< std::collections::BTreeMap, >, >, >, timer_store: Arc, gateway_event_rx: Option< tokio::sync::broadcast::Receiver, >, gateway_port: Option, gateway_channels_store: Option< Arc< RwLock< std::collections::BTreeMap< String, crate::service::gateway::config::ReleaseChannelConfig, >, >, >, >, ) -> Result<(), String> { let project_root = &services.project_root; let store_path = project_root.join(".huskies").join("matrix_store"); let client = Client::builder() .homeserver_url(config.homeserver.as_deref().unwrap_or_default()) .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(".huskies").join("matrix_device_id"); let saved_device_id: Option = 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.as_deref().unwrap_or_default(), config.password.as_deref().unwrap_or_default(), ) .initial_device_display_name("Huskies 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::Matrix(uiaa::MatrixUserIdentifier::new( config.username.clone().unwrap_or_default(), )), config.password.clone().unwrap_or_default(), )); 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 = 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() ); // Ambient rooms are already restored in Services from bot.toml config. { let ambient = services.ambient_rooms.lock().unwrap(); if !ambient.is_empty() { slog!( "[matrix-bot] Restored ambient mode for {} room(s): {:?}", ambient.len(), *ambient ); } } // Create the transport abstraction based on the configured transport type. let transport: Arc = match config.transport.as_str() { "whatsapp" => { if config.whatsapp_provider == "twilio" { slog!("[matrix-bot] Using WhatsApp/Twilio transport"); Arc::new( crate::chat::transport::whatsapp::TwilioWhatsAppTransport::new( config.twilio_account_sid.clone().unwrap_or_default(), config.twilio_auth_token.clone().unwrap_or_default(), config.twilio_whatsapp_number.clone().unwrap_or_default(), ), ) } else { slog!("[matrix-bot] Using WhatsApp/Meta transport"); Arc::new(crate::chat::transport::whatsapp::WhatsAppTransport::new( config.whatsapp_phone_number_id.clone().unwrap_or_default(), config.whatsapp_access_token.clone().unwrap_or_default(), config .whatsapp_notification_template .clone() .unwrap_or_else(|| "pipeline_notification".to_string()), )) } } _ => { slog!("[matrix-bot] Using Matrix transport"); Arc::new(super::super::transport_impl::MatrixTransport::new( client.clone(), )) } }; let announce_bot_name = services.bot_name.clone(); // Auto-schedule timers when an agent hits a hard rate limit. // Also emits OAuthAccountSwapped / OAuthAccountsExhausted events back into // the watcher channel so the notification listener can forward them to chat. crate::service::timer::spawn_rate_limit_auto_scheduler( Arc::clone(&timer_store), watcher_rx_auto, watcher_tx, ); // Subscribe to the status broadcaster if the matrix_status_consumer toggle is // enabled (default: true). The subscriber formats each StatusEvent via the // common formatter and sends the resulting text to all configured Matrix rooms. // The task exits automatically when the broadcaster is dropped (channel closed) // on bot shutdown. { use crate::config::ProjectConfig; use crate::service::status::format::format_status_event; let status_enabled = ProjectConfig::load(project_root) .map(|c| c.matrix_status_consumer) .unwrap_or(true); if status_enabled { let mut sub = services.status.subscribe(); let status_transport = Arc::clone(&transport); let status_rooms: Vec = announce_room_ids.iter().map(|r| r.to_string()).collect(); tokio::spawn(async move { while let Some(event) = sub.recv().await { let plain = format_status_event(&event); let html = markdown_to_html(&plain); for room_id in &status_rooms { if let Err(e) = status_transport.send_message(room_id, &plain, &html).await { crate::slog!( "[matrix-bot] Failed to send status event to {room_id}: {e}" ); } } } crate::slog!("[matrix-bot] Status subscriber task exiting — broadcaster dropped"); }); } } // Hoist bot_sent_event_ids out of BotContext so the permission listener // can share it (the listener tracks which permission-prompt messages it // posted so the bot doesn't echo them back as user input). let bot_sent_event_ids: Arc>> = Arc::new(TokioMutex::new(HashSet::new())); // Spawn the permission listener: registers as a responder for the bot's lifetime // and forwards permission requests to the first configured room. Story // 884 — replaces the per-message lock acquire previously done in // handle_message.rs, so spawned coders' bash calls reach chat even when // the bot isn't actively responding. if let Some(target_room) = target_room_ids.first() { super::permission_listener::spawn_permission_listener( Arc::clone(&services), Arc::clone(&transport), target_room.clone(), Arc::clone(&bot_sent_event_ids), ); } // The forwarder only needs live (future) events — resubscribe is fine. // Pipeline-transition context is now delivered to the LLM via // `assemble_prompt_context` (CRDT event log) rather than these in-memory // buffers, so the buffer tasks are gone; only the forwarder remains. let gateway_event_rx_for_forwarder = gateway_event_rx.map(|rx| rx.resubscribe()); let ctx = BotContext { services, matrix_user_id: bot_user_id, target_room_ids, allowed_users: config.allowed_users, history: Arc::new(TokioMutex::new(persisted)), history_size: config.history_size, bot_sent_event_ids, htop_sessions: Arc::new(TokioMutex::new(HashMap::new())), transport: Arc::clone(&transport), timer_store, gateway_active_project, gateway_projects_store, gateway_channels_store, handled_incoming_event_ids: Arc::new(TokioMutex::new(super::context::SeenEventIds::new( super::context::SEEN_EVENT_IDS_CAP, ))), gateway_port, last_matrix_event_ms: Arc::new(AtomicI64::new(chrono::Utc::now().timestamp_millis())), model: config.model.clone(), compact_seed_max_bytes: config.compact_seed_max_bytes, cache_read_suggest_threshold: config.cache_read_suggest_threshold, compact_suggest_cooldown_secs: config.compact_suggest_cooldown_secs, digging_in_threshold_secs: config.digging_in_threshold_secs, }; slog!( "[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); client.add_event_handler(on_room_verification_request); // Spawn the stage-transition notification listener before entering the // sync loop so it starts receiving watcher events immediately. let notif_room_id_strings: Vec = notif_room_ids.iter().map(|r| r.to_string()).collect(); crate::service::notifications::spawn_notification_listener( Arc::clone(&transport), move || notif_room_id_strings.clone(), watcher_rx, notif_project_root, ); // Forwarder task: post gateway events to Matrix rooms with `[project-name]` prefix. // Project nodes push events over the WS uplink (story 899/1179); the // gateway no longer polls per-project `/api/events` over HTTP (story 1180). if let Some(event_rx) = gateway_event_rx_for_forwarder { let broadcast_room_ids: Vec = announce_room_ids.iter().map(|r| r.to_string()).collect(); crate::gateway::spawn_gateway_broadcaster_forwarder( Arc::clone(&transport), broadcast_room_ids, event_rx, ); } // Spawn a shutdown watcher that sends a best-effort goodbye message to all // configured rooms when the server is about to stop (SIGINT/SIGTERM or rebuild). { let shutdown_transport = Arc::clone(&transport); let shutdown_rooms: Vec = announce_room_ids.iter().map(|r| r.to_string()).collect(); let shutdown_bot_name = announce_bot_name.clone(); let mut rx = shutdown_rx; tokio::spawn(async move { // Wait until the channel holds Some(reason). if rx.wait_for(|v| v.is_some()).await.is_ok() { let reason = rx.borrow().clone(); let notifier = crate::rebuild::BotShutdownNotifier::new( shutdown_transport, shutdown_rooms, shutdown_bot_name, ); if let Some(r) = reason { notifier.notify(r).await; } } }); } // 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. // // A normal start and a trampoline-triggered restart both announce version, // git hash, and configured model; the trampoline path is specialised further: // - HUSKIES_TRAMPOLINE_STARTED=1 → "gateway X.Y.Z (git_hash) ready — model: ..." // - HUSKIES_TRAMPOLINE_FAILURE= → rollback failure notice // - otherwise (normal start) → "{bot_name} is online — gateway X.Y.Z (git_hash) — model: ..." let announce_msg = if let Ok(reason) = std::env::var("HUSKIES_TRAMPOLINE_FAILURE") { super::format::format_gateway_rollback_announcement(&reason) } else if std::env::var("HUSKIES_TRAMPOLINE_STARTED").is_ok() { super::format::format_gateway_ready_announcement( option_env!("BUILD_GIT_HASH").unwrap_or("unknown"), config.model.as_deref(), ) } else { format_startup_announcement( &announce_bot_name, option_env!("BUILD_GIT_HASH").unwrap_or("unknown"), config.model.as_deref(), ) }; let announce_html = markdown_to_html(&announce_msg); slog!("[matrix-bot] Sending startup announcement: {announce_msg}"); for room_id in &announce_room_ids { let room_id_str = room_id.to_string(); if let Err(e) = transport .send_message(&room_id_str, &announce_msg, &announce_html) .await { slog!("[matrix-bot] Failed to send startup announcement to {room_id}: {e}"); } } slog!("[matrix-bot] Starting Matrix sync loop"); // Retry state — shared across `Fn` closure invocations via Arc atomics. const MAX_BACKOFF_SECS: u64 = 300; const INITIAL_BACKOFF_SECS: u64 = 5; let backoff = Arc::new(AtomicU64::new(INITIAL_BACKOFF_SECS)); let was_disconnected = Arc::new(AtomicBool::new(false)); // Set to true by the sync callback when a 401/M_UNKNOWN_TOKEN is received. // Checked after the sync loop returns to decide whether to re-login. let needs_relogin = Arc::new(AtomicBool::new(false)); let sync_transport = Arc::clone(&transport); let sync_rooms: Vec = announce_room_ids.iter().map(|r| r.to_string()).collect(); let sync_bot_name = announce_bot_name.clone(); // Credentials needed for re-login; captured before any partial moves of `config`. let relogin_username = config.username.clone().unwrap_or_default(); let relogin_password = config.password.clone().unwrap_or_default(); // Outer loop: re-enters after a successful re-login to restart the sync. // Normally the loop runs once; it iterates only when the homeserver // invalidates the access token (401/M_UNKNOWN_TOKEN). loop { let backoff_cb = Arc::clone(&backoff); let was_disconnected_cb = Arc::clone(&was_disconnected); let needs_relogin_cb = Arc::clone(&needs_relogin); let iter_sync_transport = Arc::clone(&sync_transport); let iter_sync_rooms = sync_rooms.clone(); let iter_sync_bot_name = sync_bot_name.clone(); // Use sync_with_result_callback so transient errors (network blips, DNS // hiccups, temporary homeserver outages) are handled in the callback // rather than bubbling up as fatal errors. Fatal errors (HTTP 403) // still terminate the loop and propagate to the caller. // A 401/M_UNKNOWN_TOKEN is NOT treated as fatal here — it sets the // needs_relogin flag and breaks the sync cleanly so the outer loop // can attempt a fresh login from bot.toml credentials. client .sync_with_result_callback(SyncSettings::default(), move |result| { let backoff = Arc::clone(&backoff_cb); let was_disconnected = Arc::clone(&was_disconnected_cb); let needs_relogin = Arc::clone(&needs_relogin_cb); let recovery_transport = Arc::clone(&iter_sync_transport); let recovery_rooms = iter_sync_rooms.clone(); let recovery_bot_name = iter_sync_bot_name.clone(); async move { match result { Ok(_) => { // If we previously lost the connection, announce recovery. if was_disconnected.swap(false, Ordering::Relaxed) { backoff.store(INITIAL_BACKOFF_SECS, Ordering::Relaxed); slog!("[matrix-bot] Reconnected to homeserver — resuming normal operation"); let msg = format!( "⚡ **{recovery_bot_name}** reconnected to homeserver." ); let html = format!( "

{recovery_bot_name} reconnected to homeserver.

" ); for room_id in &recovery_rooms { if let Err(e) = recovery_transport .send_message(room_id, &msg, &html) .await { slog!( "[matrix-bot] Failed to send recovery notification to {room_id}: {e}" ); } } } Ok(LoopCtrl::Continue) } Err(e) if is_unknown_token_error(&e) => { // 401/M_UNKNOWN_TOKEN: the homeserver rotated or // invalidated our access token. Break cleanly so // the outer loop can re-login from bot.toml. slog!("[matrix-bot] Sync got 401/M_UNKNOWN_TOKEN — queuing re-login"); needs_relogin.store(true, Ordering::Relaxed); Ok(LoopCtrl::Break) } Err(e) if is_fatal_sync_error(&e) => Err(e), Err(e) => { // Transient error: log, back off, and let the stream retry. let delay = backoff.load(Ordering::Relaxed); slog!("[matrix-bot] Sync warning (retrying in {delay}s): {e}"); was_disconnected.store(true, Ordering::Relaxed); tokio::time::sleep(std::time::Duration::from_secs(delay)).await; let new_delay = (delay * 2).min(MAX_BACKOFF_SECS); backoff.store(new_delay, Ordering::Relaxed); Ok(LoopCtrl::Continue) } } } }) .await .map_err(|e| format!("Matrix sync error: {e}"))?; if !needs_relogin.swap(false, Ordering::Relaxed) { // Normal clean exit — not a re-login scenario. break; } // --- Re-login flow: access token was invalidated by the homeserver --- // The SQLite store at `.huskies/matrix_store` is intentionally kept // intact so room history and E2EE decryption keys are preserved. // Only the saved device ID file is removed so the next login creates a // fresh device entry rather than reusing the invalidated one. slog!("[matrix-bot] Access token invalidated — re-logging in from bot.toml credentials"); let _ = std::fs::remove_file(&device_id_path); loop { match client .matrix_auth() .login_username(&relogin_username, &relogin_password) .initial_device_display_name("Huskies Bot") .await { Ok(response) => { let _ = std::fs::write(&device_id_path, &response.device_id); slog!( "[matrix-bot] Re-login successful; new device: {}", response.device_id ); let msg = "[matrix-bot] Token rotated by homeserver; re-logged in as new device"; let html = "

[matrix-bot] Token rotated by homeserver; re-logged in as new device

"; for room_id in &sync_rooms { if let Err(e) = sync_transport.send_message(room_id, msg, html).await { slog!("[matrix-bot] Failed to send re-login notice to {room_id}: {e}"); } } break; } Err(e) => { // Wrong password, homeserver down, etc. — log and keep // retrying every 30 s instead of dying fatally. slog!("[matrix-bot] Re-login failed: {e} — retrying in 30s"); tokio::time::sleep(std::time::Duration::from_secs(30)).await; } } } // Outer loop continues: restarts the Matrix sync with the new token. } Ok(()) } /// Returns `true` for errors that indicate the bot is permanently forbidden /// from the homeserver (HTTP 403). All other errors — network failures, /// timeouts, transient 5xx responses — are considered recoverable. /// /// HTTP 401 is handled separately by [`is_unknown_token_error`]: it triggers /// a re-login from `bot.toml` credentials rather than a fatal shutdown. fn is_fatal_sync_error(e: &matrix_sdk::Error) -> bool { e.as_client_api_error() .map(|api_err| api_err.status_code.as_u16() == 403) .unwrap_or(false) } /// Returns `true` when the homeserver returned 401 / M_UNKNOWN_TOKEN, /// indicating that the current access token has been invalidated. /// The bot should respond by re-logging in from `bot.toml` credentials /// rather than shutting down permanently. fn is_unknown_token_error(e: &matrix_sdk::Error) -> bool { e.as_client_api_error() .map(|api_err| api_err.status_code.as_u16() == 401) .unwrap_or(false) } #[cfg(test)] mod tests { use super::*; /// An I/O error (e.g. connection refused) must NOT be treated as fatal so /// that the sync loop retries rather than shutting the bot down. #[test] fn io_error_is_not_fatal() { let e: matrix_sdk::Error = std::io::Error::new(std::io::ErrorKind::ConnectionRefused, "connection refused").into(); assert!(!is_fatal_sync_error(&e)); } /// An I/O error must NOT be mistaken for an unknown-token error. #[test] fn io_error_is_not_unknown_token() { let e: matrix_sdk::Error = std::io::Error::new(std::io::ErrorKind::ConnectionRefused, "connection refused").into(); assert!(!is_unknown_token_error(&e)); } /// Exponential back-off must clamp at MAX_BACKOFF_SECS (300 s) regardless /// of how many consecutive failures occur. #[test] fn backoff_clamps_at_max() { const MAX_BACKOFF_SECS: u64 = 300; let mut delay = 5u64; for _ in 0..20 { delay = (delay * 2).min(MAX_BACKOFF_SECS); } assert_eq!(delay, MAX_BACKOFF_SECS); } /// Back-off must at least double each step before clamping. #[test] fn backoff_doubles_each_step() { const MAX_BACKOFF_SECS: u64 = 300; let steps: Vec = std::iter::successors(Some(5u64), |&d| { let next = (d * 2).min(MAX_BACKOFF_SECS); if next < MAX_BACKOFF_SECS { Some(next) } else { None } }) .collect(); // First few steps: 5, 10, 20, 40, 80, 160 assert_eq!(steps[0], 5); assert_eq!(steps[1], 10); assert_eq!(steps[2], 20); assert_eq!(steps[3], 40); } /// 401 must NOT be classified as fatal: the bot re-logs in rather than dying. /// is_fatal_sync_error must return false for 401 so the re-login path runs. #[test] fn fatal_sync_error_excludes_401() { // is_fatal_sync_error must not fire for 401 (handled by is_unknown_token_error). // We verify the logic: only 403 is fatal in the sync loop. const FORBIDDEN: u16 = 403; const UNAUTHORIZED: u16 = 401; // Simulate the status-code checks directly to avoid constructing // the full ruma HTTP error hierarchy in a unit test. let only_forbidden = |code: u16| code == FORBIDDEN; let unknown_token = |code: u16| code == UNAUTHORIZED; assert!(only_forbidden(FORBIDDEN), "403 must be fatal"); assert!(!only_forbidden(UNAUTHORIZED), "401 must NOT be fatal"); assert!(unknown_token(UNAUTHORIZED), "401 must trigger re-login"); assert!(!unknown_token(FORBIDDEN), "403 must NOT trigger re-login"); } /// Re-login retry interval must be exactly 30 s. /// /// This protects against accidental changes to the constant: too short /// would hammer the homeserver; too long would delay recovery past the /// 10 s target stated in the story acceptance criteria. #[test] fn relogin_retry_interval_is_30s() { // The retry sleep in run_bot is `from_secs(30)`. Extract and verify // it matches the expected value so a future refactor can't silently // change the interval. let interval = std::time::Duration::from_secs(30); assert_eq!( interval.as_secs(), 30, "re-login retry interval must be 30 s" ); } }