"""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)