import os import sys import click from slack_bolt import App from slack_bolt.adapter.socket_mode import SocketModeHandler from dotenv import load_dotenv import re import subprocess # Add project root to path sys.path.append(os.getcwd()) from health.utils.logging_config import setup_logger from health.utils.env_loader import load_env_with_extras from slack_bot.dispatcher import MessageDispatcher from slack_bot.shell.help_text import SHELL_HELP_TEXT from health.utils.time_utils import get_current_time_str logger = setup_logger(__name__) # Load environment variables load_env_with_extras() @click.command() @click.option("--model", default=None, help="Gemini model to use (overrides .env)") def main(model: str): """Butler Slack Bot - Shell Edition.""" # Guard: require explicit opt-in to prevent accidental startup if os.environ.get("SHELL_BOT_ENABLED", "").lower() != "true": logger.error("Shell Bot is disabled on this machine. Set SHELL_BOT_ENABLED=true in .env to enable.") sys.exit(1) # Override model in environment only if specified via CLI if model: os.environ["GEMINI_MODEL"] = model logger.info(f"Using model from CLI: {model}") # Get the actual model being used (from CLI or .env) actual_model = os.environ.get("GEMINI_MODEL", "gemini-2.0-flash-exp") logger.info(f"Starting Shell Bot with Gemini model: {actual_model}") # Initialize Slack App with SHELL-specific tokens bot_token = os.environ.get("SHELL_SLACK_BOT_TOKEN") app_token = os.environ.get("SHELL_SLACK_APP_TOKEN") if not bot_token or not app_token: logger.error("Missing SHELL_SLACK_BOT_TOKEN or SHELL_SLACK_APP_TOKEN in .env") sys.exit(1) app = App(token=bot_token) # --- SHELL CONFIGURATION --- MAX_TRACKED_MESSAGES = 1000 # NOTE: We can't easily inject dynamic time into the *static* dispatcher init here # if we want it to update per-request without recreating the dispatcher. # However, ShellBot is often single-session. # Better approach: We'll recreate the dispatcher or update the prompt inside handle_message? # Actually, MessageDispatcher init takes system_instruction. # We can just Init it here with STARTUP time, which is better than nothing, # but ideally we want per-request time. # The simplest fix for now: Inject it at module level (restart required for day change anyway). SHELL_SYSTEM_PROMPT = f"""You are a Shell Assistant, an expert in zsh and system administration. Current Time: {get_current_time_str()} Your Goal: Help the user execute commands and understand their system. Capabilities: - You have a persistent zsh session available via the `execute_shell` tool. - You can query files, check processes, and manage the system. CRITICAL ANTI-HALLUCINATION RULES: 1. **TOOL-FIRST POLICY**: When user asks about CURRENT system state (e.g., "what files are on my desktop", "what processes are running", "system load"), you MUST call `execute_shell` FIRST. NEVER answer based on conversation history or memory. 2. **ALWAYS PREFER EXECUTION**: If user asks for a command (e.g., "list files"), EXECUTE IT using `execute_shell("ls -la")`. DO NOT just explain it. 3. **100% OUTPUT FIDELITY**: Report EXACTLY what the terminal returns. If `execute_shell` returns empty string, error, or loading indicator, say "The command returned no output" or "Command failed with error: X". NEVER invent, guess, or assume what the output "should" look like. 4. **EXPLICIT UNCERTAINTY**: If you haven't executed a command in the current turn, DO NOT provide specific file names, process lists, or system state details. Guidelines: 5. **Numbered Shortcuts**: Provide numbered lists (1, 2, 3). If user replies with a number, EXECUTE that option immediately. 6. **Be Concise**: Summarize long outputs, but show relevant parts. 7. **Safety**: Warn before destructive commands (rm -rf), but respect user's decision. """ # Only include the shell tool from slack_bot.tools.registry import TOOLS_SCHEMA SHELL_TOOLS = [t for t in TOOLS_SCHEMA if t["function"]["name"] == "execute_shell"] # Initialize Dispatcher with Shell Identity dispatcher = MessageDispatcher( bot_token=bot_token, system_instruction=SHELL_SYSTEM_PROMPT, tools=SHELL_TOOLS ) # Message deduplication: Track processed message IDs processed_messages = set() MAX_TRACKED_MESSAGES = 1000 # Message Handler @app.event("message") def handle_message(message, say): try: # Deduplicate using client_msg_id or event_ts msg_id = message.get("client_msg_id") or message.get("ts") if msg_id in processed_messages: logger.debug(f"Skipping duplicate message: {msg_id}") return processed_messages.add(msg_id) if len(processed_messages) > MAX_TRACKED_MESSAGES: to_remove = list(processed_messages)[:MAX_TRACKED_MESSAGES // 2] for old_id in to_remove: processed_messages.discard(old_id) text = message.get("text", "") channel = message["channel"] user = message["user"] files = message.get("files", []) # --- HANDLE HELP COMMAND --- if text.strip().lower() in ["help", "help?", "帮助", "menu"]: say(SHELL_HELP_TEXT) return # Generate Request ID import uuid import time request_id = str(uuid.uuid4())[:8] logger.info(f"[{request_id}] 📥 ShellBot received: {text}") # --- FAILSAFE: CONFIG SWITCHING --- gmode_match = re.match(r"^### gmode (.*?) ###$", text.strip()) if gmode_match: target_config = gmode_match.group(1) logger.warning(f"[{request_id}] 🚨 Received FAILSAFE cmd: {target_config}") try: # Execute switch script script_path = os.path.join(os.getcwd(), "scripts", "select_gemini.py") result = subprocess.run( [sys.executable, script_path, target_config], capture_output=True, text=True ) if result.returncode == 0: # Special handling for 'check' and 'check-detailed' commands - show output, don't restart if target_config in ["check", "check-detailed"]: output = result.stdout + result.stderr say(f"🔍 **Health Check Results**:\n```\n{output}\n```") return say(f"♻️ **System Reset**: Switched to `{target_config}`. Restarting process...") # Allow message to be sent time.sleep(1) # Self-Restart logger.warning("RESTARTING PROCESS via os.execv...") os.execv(sys.executable, [sys.executable] + sys.argv) else: say(f"❌ Command Failed:\n```\n{result.stderr}\n```") return except Exception as e: logger.error(f"Failsafe Error: {e}") say(f"❌ Failsafe Error: {e}") return # ---------------------------------- # --- FAILSAFE: BOT COMMANDS --- bot_match = re.match(r"^### bot (.*?) ###$", text.strip()) if bot_match: bot_args = bot_match.group(1).split() logger.warning(f"[{request_id}] 🚨 Received FAILSAFE cmd: bot {' '.join(bot_args)}") try: # Execute bot_manager.py script_path = os.path.join(os.getcwd(), "scripts", "bot_manager.py") result = subprocess.run( [sys.executable, script_path] + bot_args, capture_output=True, text=True ) output = result.stdout + "\n" + result.stderr if result.returncode == 0: say(f"🤖 **Bot Manager Output**:\n```\n{output.strip()}\n```") else: say(f"❌ Command Failed:\n```\n{output.strip()}\n```") return except Exception as e: logger.error(f"Failsafe Error: {e}") say(f"❌ Failsafe Error: {e}") return # ---------------------------------- # Send temporary "Processing..." message ack_response = say(f"⏳ ShellBot is thinking...") ack_ts = ack_response["ts"] # Dispatch dispatcher.dispatch(text, channel, user, response_ts=ack_ts, request_id=request_id, files=files) except Exception as e: logger.error(f"Error handling message: {e}", exc_info=True) try: say(f"🐛 ShellBot Error: {e}") except Exception: pass # Start Socket Mode try: handler = SocketModeHandler(app, app_token) logger.info("⚡️ Shell Bot is connected to Slack!") handler.start() except Exception as e: logger.error(f"Failed to start SocketModeHandler: {e}") sys.exit(1) if __name__ == "__main__": main()