feat(03-01): implement CommandExecutor and OutputCapture classes
- 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 <noreply@anthropic.com>
This commit is contained in:
@@ -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,
|
||||
)
|
||||
Reference in New Issue
Block a user