Newer
Older
navi-1 / clients / terminal / tui / shell_runner.py
"""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