Newer
Older
navi-1 / clients / terminal / tui / screens / terminals_picker.py
"""Terminals picker modal screen for Navi Code TUI.

Lists the session's open persistent terminals (background dev servers,
watchers, …) and lets the user force-close one. Source: a REST seed on open
(``api.list_terminals``) kept live by the chat model's terminals state — the
app refreshes the picker while it is open as terminal_opened/closed events
arrive. ``Up/Down`` navigate, ``Enter`` closes the highlighted terminal,
``Escape`` cancels.
"""

from __future__ import annotations

from rich.text import Text
from textual import events
from textual.app import ComposeResult
from textual.containers import Container
from textual.screen import ModalScreen
from textual.widgets import ListItem, ListView, Static

from clients.terminal import api


def _row(summary: dict) -> Text:
    name = summary.get("name", "?")
    description = summary.get("description", "") or ""
    status = summary.get("status", "idle")
    pid = summary.get("pid")
    uptime = summary.get("uptime_seconds")
    busy = status == "busy"
    # ▶ running process, ■ stopped — a terminal-native play/stop metaphor
    # instead of coloured circles.
    icon = "▶" if busy else "■"

    line = Text()
    line.append(f"{icon} ", style="bold")
    line.append(str(name), style="bold")

    meta: list[str] = []
    if description:
        meta.append(str(description))
    if pid is not None:
        meta.append(f"pid {pid}")
    if uptime is not None:
        meta.append(f"{int(uptime)}s")
    if meta:
        line.append("\n")
        line.append("  " + "  ·  ".join(meta), style="dim")
    return line


class TerminalsPickerScreen(ModalScreen[str | None]):
    """Pick an open terminal to close, or cancel."""

    DEFAULT_CSS = """
    TerminalsPickerScreen {
        align: center middle;
    }
    TerminalsPickerScreen > Container {
        width: 70;
        height: auto;
        max-height: 24;
        border: thick $tui-primary;
        background: $tui-surface;
        padding: 0 0 1 0;
    }
    TerminalsPickerScreen .title {
        text-style: bold;
        color: $tui-primary;
        background: $tui-panel;
        padding: 1;
        height: auto;
        text-align: center;
    }
    TerminalsPickerScreen ListView {
        height: auto;
        max-height: 16;
        border: none;
        background: $tui-surface;
        padding: 0;
        margin: 0;
    }
    TerminalsPickerScreen ListItem {
        color: $tui-text;
        background: transparent;
        height: auto;
        padding: 0 1;
    }
    TerminalsPickerScreen ListItem.--highlight {
        background: $tui-selection;
        color: $tui-background;
    }
    TerminalsPickerScreen .empty {
        color: $tui-text-dim;
        text-align: center;
        padding: 1;
    }
    """

    BINDINGS = [
        ("escape", "dismiss_cancel", "Cancel"),
    ]

    def __init__(self, session_id: str) -> None:
        super().__init__()
        self._session_id = session_id
        # name → summary dict (the rows we render). Built from the REST seed
        # and then kept live by refresh_from_model as events arrive.
        self._terminals: dict[str, dict] = {}
        self._order: list[str] = []
        self._list_items: list[ListItem] = []

    @property
    def session_id(self) -> str:
        return self._session_id

    def compose(self) -> ComposeResult:
        with Container():
            yield Static("Open terminals — Delete to close, Esc to cancel", classes="title")
            yield ListView(id="terminals-list")

    def on_mount(self) -> None:
        self.run_worker(self._load())

    async def _load(self) -> None:
        try:
            summaries = await api.list_terminals(self._session_id)
        except Exception as exc:
            self.app.query_one("ChatPanel").handle_ws_event(
                {"type": "error", "message": f"Failed to list terminals: {exc}"}
            )
            self.dismiss(None)
            return
        self._set_from_summaries(summaries)
        self._render_list()

    def _set_from_summaries(self, summaries: list[dict]) -> None:
        self._terminals = {}
        self._order = []
        for s in summaries:
            name = s.get("name")
            if name:
                self._terminals[name] = s
                self._order.append(name)

    def refresh_from_model(self, terminals: dict[str, dict]) -> None:
        """Live-refresh the list from the chat model's terminals state (called
        by the app as terminal_opened/closed events arrive while the modal is
        open). Preserves the highlight where possible."""
        list_view = self.query_one("#terminals-list", ListView)
        keep = list_view.index if list_view.is_mounted else None

        new_order = [name for name, t in terminals.items() if not t.get("closed")]
        # Preserve original order for survivors, append newly opened at the end.
        survivors = [n for n in self._order if n in new_order]
        added = [n for n in new_order if n not in self._order]
        self._order = survivors + added
        self._terminals = {
            name: {
                "name": name,
                "description": terminals[name].get("description", ""),
                "pid": terminals[name].get("pid"),
                "status": terminals[name].get("status", "idle"),
                "uptime_seconds": terminals[name].get("uptime_seconds"),
            }
            for name in self._order
        }
        self._render_list()
        if keep is not None and self._order and 0 <= keep < len(self._order):
            list_view.index = keep

    def _render_list(self) -> None:
        list_view = self.query_one("#terminals-list", ListView)
        list_view.clear()
        self._list_items = []

        if not self._order:
            list_view.append(ListItem(Static("No open terminals", classes="empty")))
            list_view.index = 0
            return

        for name in self._order:
            item = ListItem(Static(_row(self._terminals[name])))
            self._list_items.append(item)
            list_view.append(item)
        list_view.index = 0

    def _select_highlighted(self) -> str | None:
        list_view = self.query_one("#terminals-list", ListView)
        idx = list_view.index
        if idx is None or not self._order or not (0 <= idx < len(self._order)):
            return None
        return self._order[idx]

    async def _close_highlighted(self) -> None:
        name = self._select_highlighted()
        if name is None:
            return
        try:
            await api.close_terminal(self._session_id, name)
        except Exception as exc:
            self.app.query_one("ChatPanel").handle_ws_event(
                {"type": "error", "message": f"Failed to close terminal '{name}': {exc}"}
            )
            return
        # The server emits terminal_closed → the app refreshes this modal via
        # refresh_from_model. If no event arrives (e.g. already gone), drop it
        # locally so the list updates immediately.
        self._order = [n for n in self._order if n != name]
        self._terminals.pop(name, None)
        self._render_list()

    def on_key(self, event: events.Key) -> None:
        list_view = self.query_one("#terminals-list", ListView)
        if event.key == "down":
            list_view.action_cursor_down()
            event.stop()
            event.prevent_default()
        elif event.key == "up":
            list_view.action_cursor_up()
            event.stop()
            event.prevent_default()
        elif event.key in ("delete", "backspace"):
            # Delete closes the highlighted terminal — more intuitive than Enter
            # (Enter commonly means "open/select", Delete means "remove/close").
            # No-op on an empty list.
            if self._order:
                self.run_worker(self._close_highlighted())
            event.stop()
            event.prevent_default()
        elif event.key == "escape":
            self.dismiss(None)
            event.stop()
            event.prevent_default()

    def action_dismiss_cancel(self) -> None:
        self.dismiss(None)