"""Run local shell commands from !input in the TUI."""
from __future__ import annotations
import subprocess
from dataclasses import dataclass
from pathlib import Path
DEFAULT_TIMEOUT = 30.0
MAX_OUTPUT_LINES = 200
@dataclass
class ShellResult:
"""Result of running a shell command."""
command: str
returncode: int
stdout: str
stderr: str
truncated: bool = False
def summary(self) -> str:
"""Return a short, chat-friendly summary."""
marker = "✓" if self.returncode == 0 else "✗"
lines = [f"{marker} $ {self.command}", ""]
if self.stdout:
lines.append(self.stdout)
if self.stderr:
if self.stdout:
lines.append("")
lines.append(f"--- stderr ---\n{self.stderr}")
lines.append("")
lines.append(f"exit code: {self.returncode}")
return "\n".join(lines)
def run_shell_command(raw: str, cwd: Path | str | None = None, timeout: float = DEFAULT_TIMEOUT) -> ShellResult:
"""Run a shell command from user input (without the leading !).
The command is passed to a real shell so pipes, redirections and globs work.
"""
command = raw[1:] if raw.startswith("!") else raw
command = command.strip()
if not command:
return ShellResult(command="", returncode=1, stdout="", stderr="empty command")
work_dir = Path(cwd or Path.cwd()).expanduser().resolve()
try:
proc = subprocess.run(
command,
shell=True,
cwd=work_dir,
capture_output=True,
text=True,
timeout=timeout,
)
stdout, stdout_truncated = _truncate(proc.stdout)
stderr, stderr_truncated = _truncate(proc.stderr)
return ShellResult(
command=command,
returncode=proc.returncode,
stdout=stdout,
stderr=stderr,
truncated=stdout_truncated or stderr_truncated,
)
except subprocess.TimeoutExpired:
return ShellResult(command=command, returncode=124, stdout="", stderr=f"timed out after {timeout}s", truncated=False)
except Exception as exc:
return ShellResult(command=command, returncode=1, stdout="", stderr=str(exc), truncated=False)
def _truncate(text: str) -> tuple[str, bool]:
"""Limit output to the last MAX_OUTPUT_LINES lines to avoid flooding the UI.
Returns the possibly-truncated text and a flag indicating whether truncation
actually happened.
"""
lines = text.splitlines()
if len(lines) <= MAX_OUTPUT_LINES:
return text, False
truncated = lines[-MAX_OUTPUT_LINES:]
return f"... [{len(lines) - MAX_OUTPUT_LINES} lines truncated]\n" + "\n".join(truncated), True