""" Main command router and orchestrator for OpenClaw skill invocation. This module provides the central command routing system that: - Receives chat messages via handle_message() - Routes commands through session management and command execution - Handles approval workflows and progress updates - Returns formatted Feishu card responses """ from __future__ import annotations import re from typing import Any, Dict, Optional, Tuple, TYPE_CHECKING if TYPE_CHECKING: from pathlib import Path from gsd_openclaw.tmux_session import TmuxSession, TmuxSessionManager from gsd_openclaw.command_executor import CommandExecutor, CommandResult from gsd_openclaw.approval_handler import ApprovalHandler from gsd_openclaw.progress_monitor import ProgressMonitor from gsd_openclaw.response_formatter import FeishuResponseFormatter, CardButton from gsd_openclaw.user_context import UserContextManager class Orchestrator: """Central command router for OpenClaw skill invocation. This class provides the main entry point for handling GSD commands from chat messages, integrating session management, command execution, approval handling, and response formatting. Attributes: tmux_manager: TmuxSessionManager for session lifecycle. formatter: FeishuResponseFormatter for card creation. executor: CommandExecutor for running commands. approval_handler: ApprovalHandler for interactive approvals. progress_monitor: ProgressMonitor for progress updates. """ # Regex pattern for /gsd:* command parsing COMMAND_PATTERN = re.compile(r'^/gsd:(\S+)\s*(.*)$') # Maximum characters for card content (Feishu limit) MAX_OUTPUT_LENGTH = 2000 # Commands that require long-running sessions (no timeout) LONG_COMMANDS = {'execute-phase', 'plan-phase', 'discuss-phase', 'new-project'} def __init__( self, tmux_manager: Optional[TmuxSessionManager] = None, formatter: Optional[FeishuResponseFormatter] = None, users_root: Optional["Path"] = None ): """Initialize Orchestrator. Args: tmux_manager: Optional TmuxSessionManager instance. If not provided, a new one will be created. formatter: Optional FeishuResponseFormatter instance. If not provided, a new one will be created. users_root: Optional custom root for user contexts (useful for testing). """ self.tmux_manager = tmux_manager or TmuxSessionManager() self.formatter = formatter or FeishuResponseFormatter() self.executor = CommandExecutor() self.approval_handler = ApprovalHandler(self.formatter) self.progress_monitor = ProgressMonitor(self.formatter) self.user_context_managers: Dict[str, UserContextManager] = {} self._users_root = users_root def _get_user_context_manager(self, user_id: str) -> UserContextManager: """Get or create UserContextManager for a user. Args: user_id: The user identifier. Returns: UserContextManager instance for the user. """ if user_id not in self.user_context_managers: self.user_context_managers[user_id] = UserContextManager(user_id, self._users_root) return self.user_context_managers[user_id] def _is_long_command(self, command: str) -> bool: """Check if command requires no timeout. Args: command: The command name (e.g., 'execute-phase', 'status'). Returns: True if command is in LONG_COMMANDS, False otherwise. """ return command in self.LONG_COMMANDS def _handle_new_project(self, user_id: str, project_name: str) -> Dict[str, Any]: """Handle /gsd:new-project command. Pre-creates a tmux session before executing the command, then updates user context to track the new project. Args: user_id: ID of the user sending the command. project_name: Name of the project to create. Returns: Feishu Card JSON dict as response. """ # Validate project name if not project_name or not project_name.strip(): return self.formatter.format_error( title="Project name required", content="Please provide a project name: /gsd:new-project " ) project_name = project_name.strip() try: # Create session first session = self.tmux_manager.create_session(project_name) # Build and execute the new-project command full_command = f"/gsd:new-project {project_name}" # Execute with no timeout for long-running commands result = self.executor.execute(session, full_command, timeout=None) # Update user context user_ctx = self._get_user_context_manager(user_id) user_ctx.set_current_project(project_name) # Truncate output if needed output = result.output if len(output) > self.MAX_OUTPUT_LENGTH: output = output[-self.MAX_OUTPUT_LENGTH:] # Return appropriate response if result.has_error: return self.formatter.format_error( title="Project creation failed", content=output ) else: return self.formatter.format_success( title="Project created", content=f"Project '{project_name}' created successfully.\n\n{output}" ) except Exception as e: return self.formatter.format_error( title="Project creation failed", content=str(e)[-self.MAX_OUTPUT_LENGTH:] ) def _handle_use_project(self, user_id: str, project_name: str) -> Dict[str, Any]: """Handle /gsd:use command. Switches the user's current project context. Shows error with "Create New Project" button if the project doesn't exist. Args: user_id: ID of the user sending the command. project_name: Name of the project to switch to. Returns: Feishu Card JSON dict as response. """ # Validate project name if not project_name or not project_name.strip(): return self.formatter.format_error( title="Project name required", content="Please provide a project name: /gsd:use " ) project_name = project_name.strip() try: # Check if project session exists session = self.tmux_manager.get_session(project_name) if session is not None: # Project exists - update user context and return success user_ctx = self._get_user_context_manager(user_id) user_ctx.set_current_project(project_name) return self.formatter.format_success( title="Project switched", content=f"Switched to project '{project_name}'" ) else: # Project doesn't exist - return error with Create button create_button = CardButton( text="Create New Project", value={"action": "new-project", "name": project_name}, button_type="primary" ) return self.formatter.format_error( title="Project not found", content=f"Project '{project_name}' does not exist", buttons=[create_button] ) except Exception as e: return self.formatter.format_error( title="Switch failed", content=str(e)[-self.MAX_OUTPUT_LENGTH:] ) def parse_command(self, message: str) -> Optional[Tuple[str, str]]: """Parse a message for /gsd:* command pattern. Args: message: The message text to parse. Returns: Tuple of (command, args) if matched, None otherwise. """ match = self.COMMAND_PATTERN.match(message.strip()) if match: command = match.group(1) args = match.group(2).strip() return (command, args) return None def handle_message( self, user_id: str, message: str, channel: str, context: Dict[str, Any] ) -> Dict[str, Any]: """Handle an incoming chat message. This is the main entry point for processing GSD commands. It handles: 1. Command parsing 2. Session management (auto-create if needed) 3. Command execution 4. Approval detection 5. Response formatting Args: user_id: ID of the user sending the message. message: The message text. channel: The channel where the message was sent. context: Additional context including project name. Returns: Feishu Card JSON dict as response. """ try: # Parse command parsed = self.parse_command(message) if parsed is None: return self.formatter.format_info( title="无效命令", content="请使用 /gsd:* 格式的命令。例如: /gsd:status" ) command, args = parsed # Route special commands if command == 'new-project': return self._handle_new_project(user_id, args) if command == 'use': return self._handle_use_project(user_id, args) # Generic commands - get project from user context first user_ctx = self._get_user_context_manager(user_id) project_name = user_ctx.get_current_project() # Fall back to context param or default if not project_name: project_name = context.get("project", "default") # Get or create session (auto-create per CONTEXT.md) session = self.tmux_manager.get_session(project_name) if session is None: session = self.tmux_manager.create_session(project_name) # Build the full command full_command = f"/gsd:{command}" if args: full_command = f"{full_command} {args}" # Determine timeout based on command type timeout = None if self._is_long_command(command) else self.executor.DEFAULT_TIMEOUT # Execute command with adaptive timeout result = self.executor.execute(session, full_command, timeout=timeout) # Truncate output if needed output = result.output if len(output) > self.MAX_OUTPUT_LENGTH: output = output[-self.MAX_OUTPUT_LENGTH:] # Check for approval request approval_context = self.approval_handler.detect_approval_request(result.output) if approval_context: return self.approval_handler.create_approval_card( context=approval_context, session_id=session.session_id, project_name=project_name ) # Return appropriate response based on result if result.has_error: return self.formatter.format_error( title="命令执行失败", content=output ) else: return self.formatter.format_success( title="执行成功", content=output ) except Exception as e: return self.formatter.format_error( title="命令执行失败", content=str(e)[-self.MAX_OUTPUT_LENGTH:] ) def handle_callback(self, callback_data: Dict[str, Any]) -> Dict[str, Any]: """Handle a button callback from a card. This processes various actions including: - approve/reject: For approval workflows - new-project: For creating new projects from error cards Args: callback_data: Dictionary containing: - action: "approve", "reject", or "new-project" - For approvals: approval_id, project - For new-project: name, user_id Returns: Feishu Card JSON dict as confirmation response. """ try: action = callback_data.get("action", "") # Handle new-project action from error card button if action == "new-project": project_name = callback_data.get("name", "") user_id = callback_data.get("user_id", "") if not project_name: return self.formatter.format_error( title="Invalid Request", content="Project name is required" ) return self._handle_new_project(user_id, project_name) # Handle approval/reject actions approval_id = callback_data.get("approval_id", "") project_name = callback_data.get("project", "default") # Handle callback via approval handler result = self.approval_handler.handle_callback(callback_data) if result.startswith("error:"): return self.formatter.format_error( title="操作失败", content=result ) # Get session session = self.tmux_manager.get_session(project_name) if session is None: return self.formatter.format_error( title="会话不存在", content="无法找到对应的会话" ) # Send response to tmux if result == "approved": response_text = "y" # Standard confirmation self.executor.send_command(session.tmux_session.active_pane, response_text) return self.formatter.format_success( title="已确认", content="操作已继续执行" ) elif result == "rejected": response_text = "n" # Standard rejection self.executor.send_command(session.tmux_session.active_pane, response_text) return self.formatter.format_info( title="已取消", content="操作已取消" ) else: return self.formatter.format_error( title="未知结果", content=f"处理结果: {result}" ) except Exception as e: return self.formatter.format_error( title="操作失败", content=str(e)[-self.MAX_OUTPUT_LENGTH:] )