src/cli_agent/interactive_input.py - master - Watashicuvu/agentic-tools

src/cli_agent/interactive_input.py
187 строк6 KB
Якуб
cli fixes
14 апр 2026, 18:32
14 апр 2026, 18:326202d02
"""Интерактивный ввод команд с автодополнением и быстрым выбором.

Функционал:
1. Автодополнение команд при нажатии Tab (начинаешь с / → предлагаются варианты)
2. Быстрый выбор /approve / /reject через Page Up / Page Down
3. Возможность редактировать reject reason при выборе /reject

Использует prompt_toolkit для продвинутой обработки ввода.
"""

import asyncio
import sys
import logging
from typing import List, Optional, Tuple

from prompt_toolkit import PromptSession
from prompt_toolkit.completion import Completer, Completion
from prompt_toolkit.history import InMemoryHistory

logger = logging.getLogger(__name__)


# Доступные команды
APPROVAL_COMMANDS = ["/approve", "/reject"]
CLI_COMMANDS = [
    "/plan", "/execute", "/review", "/explore", "/explain", "/arch",
    "/status", "/approve", "/reject", "/help",
]


class CommandCompleter(Completer):
    """Автодополнение команд начинающихся с /."""

    def __init__(self, commands: List[str]):
        self.commands = commands

    def get_completions(self, document, complete_event):
        text = document.text_before_cursor
        if text.startswith("/"):
            first_word = text.split()[0] if text.split() else text
            for cmd in self.commands:
                if cmd.startswith(first_word):
                    yield Completion(
                        cmd,
                        start_position=-len(first_word),
                        display=cmd,
                    )


async def interactive_approval_prompt(
    message: str = "Command",
    timeout: int = 300,
) -> Tuple[str, Optional[str]]:
    """Интерактивный prompt для выбора approve/reject.

    Поддерживает:
    - Tab для автодополнения команд
    - Page Up / Page Down / Up / Down для быстрого выбора
    - Редактирование reject reason

    Returns:
        Tuple (command, reason) где command = "approve" или "reject"
    """
    selected = [0]  # 0 = approve, 1 = reject
    history = InMemoryHistory()

    # Создаём сессию с автодополнением
    session = PromptSession(
        completer=CommandCompleter(APPROVAL_COMMANDS),
        complete_while_typing=True,
        history=history,
    )

    # Показываем меню
    from rich.console import Console
    console = Console()

    console.print("\n[yellow]Выберите действие:[/yellow]")
    console.print("  [green]▶ 0. /approve[/green] — Одобрить выполнение")
    console.print("  [red]  1. /reject <причина>[/red] — Вернуть на доработку")
    console.print("[dim]  0/1 — быстрый выбор | Enter — подтвердить | Tab — автодополнение[/dim]\n")

    try:
        result = await asyncio.wait_for(
            session.prompt_async(message="> "),
            timeout=timeout
        )
    except asyncio.TimeoutError:
        return ("reject", "timeout")
    except (KeyboardInterrupt, EOFError):
        return ("reject", "interrupt")

    text = result.strip()

    # Быстрый выбор по номеру
    if text == "0":
        return ("approve", None)
    if text == "1":
        return ("reject", None)

    # Команды
    if text == "/approve":
        return ("approve", None)

    if text.startswith("/reject"):
        reason = text[8:].strip() or "Не указано"
        return ("reject", reason)

    if text == "":
        # Пустой ввод — используем выбранный по умолчанию
        return ("approve", None)

    console.print("[yellow]⚠ Введите `0`, `1`, `/approve` или `/reject <причина>`[/yellow]")
    return ("unknown", None)


async def simple_approval_input(
    console,
    timeout: int = 300,
) -> Tuple[str, Optional[str]]:
    """Простой ввод с подсказкой (fallback если prompt_toolkit недоступен).

    Returns:
        Tuple (command, reason)
    """
    console.print("\n[yellow]Выберите действие:[/yellow]")
    console.print("  [green]0. /approve[/green] — Одобрить выполнение")
    console.print("  [red]1. /reject <причина>[/red] — Вернуть на доработку")
    console.print("[dim]  0/1 или команда | Enter — подтвердить[/dim]")

    try:
        def _blocking_read():
            return console.input("\n[yellow]>[/yellow] ").strip()

        cmd = await asyncio.to_thread(_blocking_read)
    except EOFError:
        return ("reject", "eof")

    # Быстрый выбор по номеру
    if cmd == "0":
        return ("approve", None)
    if cmd == "1":
        return ("reject", None)

    if cmd.lower() in ['exit', 'quit']:
        return ("reject", "user_exit")

    if cmd == '/approve':
        return ("approve", None)

    if cmd.startswith('/reject'):
        reason = cmd[8:].strip() or "Не указано"
        return ("reject", reason)

    if cmd == "":
        return ("approve", None)

    console.print("[yellow]⚠ Введите `0`, `1`, `/approve` или `/reject <причина>`[/yellow]")
    return ("unknown", None)


async def cli_input_with_completion(prompt: str = "You: ") -> str:
    """Ввод с автодополнением команд CLI.

    Поддерживает:
    - Tab для автодополнения команд начинающихся с /
    - Стрелки для навигации по истории ввода

    Returns:
        Введённая строка
    """
    # Создаём сессию с автодополнением
    session = PromptSession(
        completer=CommandCompleter(CLI_COMMANDS),
        complete_while_typing=True,
        history=InMemoryHistory(),
    )

    try:
        from rich.console import Console
        console = Console()
        console.print(f"\n[yellow]{prompt}[/yellow]", end="")

        result = await session.prompt_async()
        return result.strip()
    except (KeyboardInterrupt, EOFError):
        return ""