starmen1265
b817b457e9
- Add LONG_COMMANDS set with execute-phase, plan-phase, discuss-phase, new-project - Add _is_long_command method to detect long-running commands - Add TestLongCommands test class with 5 test cases - Long commands will use no timeout in execute() Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
264 lines
8.6 KiB
Python
264 lines
8.6 KiB
Python
"""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,
|
||
)
|