executor.py

"""
executor.py — Run shell commands and capture output
"""
import subprocess
import time
from dataclasses import dataclass
from typing import Optional

# Commands that modify shell state and cannot work in a subprocess
SHELL_STATE_COMMANDS = ("cd ", "export ", "source ", ". ", "pushd ", "popd", "cd\n", "cd")


@dataclass
class ExecutionResult:
    command: str
    stdout: str
    stderr: str
    exit_code: int
    elapsed_seconds: float


class Executor:
    def run(self, command: str, timeout: int = 30) -> ExecutionResult:
        """
        Run command via subprocess with shell=True.
        Captures stdout and stderr separately.
        On timeout: kills process, sets exit_code=-1.
        Never raises on command failure — always returns ExecutionResult.
        """
        start = time.monotonic()
        try:
            proc = subprocess.run(
                command,
                shell=True,
                capture_output=True,
                text=True,
                timeout=timeout,
            )
            elapsed = time.monotonic() - start
            return ExecutionResult(
                command=command,
                stdout=proc.stdout,
                stderr=proc.stderr,
                exit_code=proc.returncode,
                elapsed_seconds=elapsed,
            )
        except subprocess.TimeoutExpired:
            elapsed = time.monotonic() - start
            return ExecutionResult(
                command=command,
                stdout="",
                stderr=f"Command timed out after {timeout}s",
                exit_code=-1,
                elapsed_seconds=elapsed,
            )

    def warn_if_shell_state(self, command: str) -> Optional[str]:
        """
        Return a warning string if the command modifies shell state
        (e.g. cd, export, source) which cannot take effect in a subprocess.
        Returns None if no warning needed.
        """
        stripped = command.strip()
        for prefix in SHELL_STATE_COMMANDS:
            if stripped == prefix.strip() or stripped.startswith(prefix):
                return (
                    f"Warning: '{stripped.split()[0]}' affects shell state and will NOT "
                    f"change your current session. Run it directly in your shell instead."
                )
        return None