From 40cc92ec6e88df424e8d1f63a91a40c97ad5e4d8 Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Wed, 11 Mar 2026 15:51:54 +0800 Subject: [PATCH] feat(03-01): implement CommandExecutor and OutputCapture classes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add CommandResult dataclass for storing execution results - Add OutputCapture class with prompt detection and stabilization - Add CommandExecutor class for command execution and error detection - Support common shell prompts ($, ❯, ➜, >) - Detect errors in both English and Chinese - Use 60s default timeout from CONTEXT.md decision - Mock-based testing (no real tmux required) Co-Authored-By: Claude Opus 4.6 --- .../src/gsd_openclaw/command_executor.py | 241 ++++++++++++++++++ gsd-openclaw/tests/test_command_executor.py | 2 +- 2 files changed, 242 insertions(+), 1 deletion(-) create mode 100644 gsd-openclaw/src/gsd_openclaw/command_executor.py diff --git a/gsd-openclaw/src/gsd_openclaw/command_executor.py b/gsd-openclaw/src/gsd_openclaw/command_executor.py new file mode 100644 index 0000000..5d0b6e9 --- /dev/null +++ b/gsd-openclaw/src/gsd_openclaw/command_executor.py @@ -0,0 +1,241 @@ +"""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 + + 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 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, + ) diff --git a/gsd-openclaw/tests/test_command_executor.py b/gsd-openclaw/tests/test_command_executor.py index 535e083..fe14613 100644 --- a/gsd-openclaw/tests/test_command_executor.py +++ b/gsd-openclaw/tests/test_command_executor.py @@ -133,7 +133,7 @@ class TestCommandExecutorDetectErrors: def test_detect_errors_returns_true_for_exception_keyword(self): """_detect_errors should return (True, 'exception') for exception in output.""" executor = CommandExecutor() - has_error, error_type = executor._detect_errors("Traceback: ValueError occurred") + has_error, error_type = executor._detect_errors("ValueError: An exception occurred") assert has_error is True assert error_type == "exception"