"""Command execution and output capture for tmux sessions. This module provides the foundational I/O layer for all GSD command bridging: - Send commands to tmux sessions - Capture output with completion detection - Identify errors in command output """ from __future__ import annotations import re import time from dataclasses import dataclass from typing import List, Optional, Tuple, Pattern import libtmux @dataclass class CommandResult: """Result of a command execution. Attributes: output: The captured output from the command. completed: True if the command completed (prompt detected or output stabilized). has_error: True if error keywords were detected in output. error_type: The type of error detected, if any. """ output: str completed: bool has_error: bool error_type: Optional[str] class OutputCapture: """Captures output from tmux panes with completion detection. This class provides functionality to capture output from a tmux pane and detect when command execution has completed, either by detecting a shell prompt or by observing stable output. """ # Regex patterns for common shell prompts # These patterns match lines that are ONLY a prompt (no other content) PROMPT_PATTERNS: List[Pattern] = [ re.compile(r'^\$\s*$'), # Simple $ prompt re.compile(r'^❯\s*$'), # Starship/Powerline prompt re.compile(r'^➜\s*$'), # Agnoster prompt re.compile(r'^>\s*$'), # Simple > prompt ] def _is_prompt(self, line: str) -> bool: """Check if a line is a shell prompt. Args: line: The line to check. Returns: True if the line matches a known prompt pattern. """ stripped = line.strip() if not stripped: return False for pattern in self.PROMPT_PATTERNS: if pattern.match(stripped): return True return False def capture_until_complete( self, pane: libtmux.Pane, timeout: float = 60.0, poll_interval: float = 0.1, stable_count: int = 3 ) -> Tuple[str, bool]: """Capture output until command completes or timeout. Completion is detected when either: 1. A shell prompt appears in the output (command finished) 2. Output stabilizes (same line count for N consecutive polls) Args: pane: The tmux pane to capture output from. timeout: Maximum time to wait for completion (seconds). poll_interval: Time between capture polls (seconds). stable_count: Number of consecutive polls with same output to consider stable. Returns: Tuple of (captured_output, completed). - captured_output: The captured output as a string. - completed: True if completion was detected, False if timeout. """ start_time = time.time() last_output_hash: Optional[str] = None stability_counter = 0 captured_lines: List[str] = [] while True: elapsed = time.time() - start_time if elapsed >= timeout: # Timeout reached return '\n'.join(captured_lines), False # Capture current pane content try: current_lines = pane.capture_pane() except Exception: # If capture fails, wait and retry time.sleep(poll_interval) continue # Store captured lines captured_lines = list(current_lines) if current_lines else [] # Check for prompt in last few lines for line in captured_lines[-5:]: # Check last 5 lines for prompt if self._is_prompt(line): return '\n'.join(captured_lines), True # Check for output stabilization (same content for N consecutive polls) current_output = '\n'.join(captured_lines) current_hash = hash(current_output) if current_hash == last_output_hash: stability_counter += 1 if stability_counter >= stable_count: return '\n'.join(captured_lines), True else: stability_counter = 0 last_output_hash = current_hash # Wait before next poll time.sleep(poll_interval) class CommandExecutor: """Executes commands in tmux sessions and captures output. This class provides the main interface for sending commands to tmux sessions and capturing their output with error detection. """ # Keywords that indicate errors in command output ERROR_KEYWORDS: List[str] = [ 'error', 'failed', 'exception', 'traceback', '错误', '失败', ] DEFAULT_TIMEOUT: float = 60.0 # Commands that require no timeout (long-running operations) LONG_COMMANDS: set = {'execute-phase', 'plan-phase', 'discuss-phase', 'new-project'} def __init__(self, output_capture: Optional[OutputCapture] = None): """Initialize CommandExecutor. Args: output_capture: Optional OutputCapture instance for dependency injection. """ self.output_capture = output_capture or OutputCapture() def send_command(self, pane: libtmux.Pane, command: str) -> None: """Send a command to a tmux pane. Args: pane: The tmux pane to send the command to. command: The command string to execute. """ pane.send_keys(command, enter=True) def _detect_errors(self, output: str) -> Tuple[bool, Optional[str]]: """Detect error keywords in output. Args: output: The output to analyze for errors. Returns: Tuple of (has_error, error_type). - has_error: True if any error keyword was found. - error_type: The first matching error keyword, or None. """ output_lower = output.lower() # Check each keyword and return the first match for keyword in self.ERROR_KEYWORDS: keyword_lower = keyword.lower() # For ASCII keywords, use word boundary matching # For non-ASCII (e.g., Chinese), use simple substring match if keyword.isascii(): if re.search(rf'\b{re.escape(keyword_lower)}\b', output_lower): return True, keyword else: if keyword in output: return True, keyword return False, None def _is_long_command(self, command: str) -> bool: """Check if command requires no timeout. Long commands like execute-phase, plan-phase, discuss-phase, and new-project can take an extended time to complete and should not have a timeout limit. Args: command: The full command string (e.g., '/gsd:execute-phase'). Returns: True if command is in LONG_COMMANDS, False otherwise. """ # Extract command name from /gsd:command format match = re.match(r'^/gsd:(\S+)', command) if match: command_name = match.group(1) return command_name in self.LONG_COMMANDS return False def execute( self, session, # TmuxSession type - avoiding circular import command: str, timeout: Optional[float] = None, poll_interval: float = 0.1 ) -> CommandResult: """Execute a command in a tmux session and capture the result. Args: session: The TmuxSession to execute the command in. command: The command string to execute. timeout: Maximum time to wait for completion (seconds). Defaults to DEFAULT_TIMEOUT (60.0). poll_interval: Time between capture polls (seconds). Returns: CommandResult containing output, completion status, and error info. """ timeout = timeout if timeout is not None else self.DEFAULT_TIMEOUT # Get the active pane from the session pane = session.tmux_session.active_pane # Send the command self.send_command(pane, command) # Capture output until completion output, completed = self.output_capture.capture_until_complete( pane, timeout=timeout, poll_interval=poll_interval ) # Detect errors in output has_error, error_type = self._detect_errors(output) return CommandResult( output=output, completed=completed, has_error=has_error, error_type=error_type, )