""" Rhodawk AI — Multi-Channel Notification Engine ================================================ Fire-and-forget notifications across Telegram (and extensible to Slack/PagerDuty). All dispatches use tenacity retry logic and never block the audit loop. MINOR BUG FIX: Telegram/Slack URLs are no longer captured at module load time. They are resolved dynamically at dispatch time, so rotating credentials at runtime (without a process restart) takes effect immediately. """ import os import threading import requests from tenacity import retry, stop_after_attempt, wait_exponential def _get_telegram_creds() -> tuple[str, str]: """Resolve Telegram credentials at dispatch time, not module load time.""" return os.getenv("TELEGRAM_BOT_TOKEN", ""), os.getenv("TELEGRAM_CHAT_ID", "") def _get_slack_url() -> str: """Resolve Slack webhook URL at dispatch time, not module load time.""" return os.getenv("SLACK_WEBHOOK_URL", "") @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10)) def _post_telegram(payload: dict): token, _ = _get_telegram_creds() url = f"https://api.telegram.org/bot{token}/sendMessage" resp = requests.post(url, json=payload, timeout=10) resp.raise_for_status() @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10)) def _post_slack(payload: dict): slack_url = _get_slack_url() resp = requests.post(slack_url, json=payload, timeout=10) resp.raise_for_status() def _dispatch(message: str, level: str = "INFO"): token, chat_id = _get_telegram_creds() if token and chat_id: try: _post_telegram({ "chat_id": chat_id, "text": message, "parse_mode": "Markdown", "disable_web_page_preview": True, }) except Exception: pass slack_url = _get_slack_url() if slack_url: color_map = {"INFO": "#36a64f", "WARN": "#ffa500", "ERROR": "#ff0000", "CRITICAL": "#8b0000"} try: _post_slack({ "attachments": [{ "color": color_map.get(level, "#36a64f"), "text": message.replace("*", ""), "mrkdwn_in": ["text"], }] }) except Exception: pass def notify(message: str, level: str = "INFO"): """Non-blocking dispatch. Spawns a daemon thread — never blocks audit loop.""" threading.Thread(target=_dispatch, args=(message, level), daemon=True).start() def notify_audit_start(repo: str): notify(f"🚀 *Rhodawk AI*\n\nAutonomous audit initiated on `{repo}`.", "INFO") def notify_test_failed(test_path: str): notify(f"⚠️ *Test Failed*\n`{test_path}`\nDispatching Aider agent...", "WARN") def notify_sast_blocked(test_path: str, reason: str): notify(f"🛡️ *SAST Gate BLOCKED PR*\n`{test_path}`\nReason: `{reason}`\nHuman review required.", "CRITICAL") def notify_pr_created(test_path: str, pr_url: str): notify(f"✅ *Auto-Heal PR Generated*\n`{test_path}`\n[View PR]({pr_url})\nAwaiting human review.", "INFO") def notify_patch_failed(test_path: str): notify(f"🔴 *Patch Failed*\n`{test_path}`\nAider returned non-zero exit.", "ERROR") def notify_audit_complete(metrics: dict): notify( f"🎯 *Audit Complete*\n" f"Scanned: `{metrics['total']}` | Green: `{metrics['done']}` | " f"PRs: `{metrics['prs_created']}` | SAST Blocked: `{metrics['sast_blocked']}`", "INFO", ) def notify_chain_integrity(valid: bool, summary: str): if valid: notify(f"🔒 *Audit Chain Verified*\n{summary}", "INFO") else: notify(f"🚨 *CHAIN INTEGRITY VIOLATION*\n{summary}", "CRITICAL")