diff --git a/bot_bottle/cli/start.py b/bot_bottle/cli/start.py index 018e5ed..e95cd85 100644 --- a/bot_bottle/cli/start.py +++ b/bot_bottle/cli/start.py @@ -33,6 +33,7 @@ from ..backend.docker.capability_apply import snapshot_transcript from ..log import info from ..manifest import Manifest from ._common import PROG, USER_CWD, read_tty_line +from . import tui def cmd_start(argv: list[str]) -> int: @@ -49,15 +50,39 @@ def cmd_start(argv: list[str]) -> int: "or 'docker'). Overrides the env var when set." ), ) - parser.add_argument("name", help="agent name defined in bot-bottle.json") + parser.add_argument( + "name", + nargs="?", + default=None, + help="agent name defined in bot-bottle.json (omit to pick interactively)", + ) args = parser.parse_args(argv) dry_run = args.dry_run or os.environ.get("BOT_BOTTLE_DRY_RUN") == "1" manifest = Manifest.resolve(USER_CWD) + + agent_name: str | None = args.name + if agent_name is None: + agent_name = tui.filter_select( + sorted(manifest.agents.keys()), + title="Select agent", + ) + if agent_name is None: + return 0 + + backend_name: str | None = args.backend + if backend_name is None and "BOT_BOTTLE_BACKEND" not in os.environ: + backend_name = tui.filter_select( + list(known_backend_names()), + title="Select backend", + ) + if backend_name is None: + return 0 + spec = BottleSpec( manifest=manifest, - agent_name=args.name, + agent_name=agent_name, copy_cwd=args.cwd, user_cwd=USER_CWD, ) @@ -65,7 +90,7 @@ def cmd_start(argv: list[str]) -> int: spec, dry_run=dry_run, remote_control=args.remote_control, - backend_name=args.backend, + backend_name=backend_name, ) diff --git a/bot_bottle/cli/tui.py b/bot_bottle/cli/tui.py new file mode 100644 index 0000000..46eae42 --- /dev/null +++ b/bot_bottle/cli/tui.py @@ -0,0 +1,219 @@ +"""tui.py — minimal curses filter-select picker for CLI prompts. + +Exposed surface: + + filter_select(items, *, title="", tty_path="/dev/tty") -> str | None + +Opens /dev/tty directly so the picker works even when stdout/stdin are +redirected. Returns the selected item or None on cancel. +""" + +from __future__ import annotations + +import curses +import os +import sys +from typing import Optional + + +def filter_select( + items: list[str], + *, + title: str = "", + tty_path: str = "/dev/tty", +) -> Optional[str]: + """Render a filter-select picker over *items*. + + Returns the selected item string, or ``None`` if the user cancelled + (Esc / ``q`` / Ctrl-C / Ctrl-D) or if the terminal is too small. + + The picker opens *tty_path* directly so it works even when + stdout/stdin are redirected. + """ + if not items: + return None + + try: + tty_fd = open(tty_path, "r+b", buffering=0) + except OSError: + return None + + try: + result = _run_picker(items, title=title, tty_fd=tty_fd) + finally: + tty_fd.close() + + return result + + +# --------------------------------------------------------------------------- +# Internal implementation +# --------------------------------------------------------------------------- + +_KEY_ESC = 27 +_KEY_CTRL_C = 3 +_KEY_CTRL_D = 4 +_KEY_BACKSPACE_WIN = 8 +_KEY_ENTER_ALT = 10 + +_CANCEL_KEYS = frozenset([_KEY_ESC, _KEY_CTRL_C, _KEY_CTRL_D, ord("q")]) + + +def _run_picker(items: list[str], *, title: str, tty_fd) -> Optional[str]: + """Drive a curses session on *tty_fd* and return the picked item.""" + # newterm lets us run curses on an arbitrary fd rather than the + # process's controlling tty / stdout — crucial when stdout is piped. + old_term = os.environ.get("TERM", "xterm-256color") + os.environ.setdefault("TERM", "xterm-256color") + + # Save / restore the real stdin/stdout so curses newterm can use tty_fd. + orig_stdin = sys.__stdin__ + orig_stdout = sys.__stdout__ + + try: + import io + tty_text = io.TextIOWrapper(tty_fd, write_through=True) + sys.__stdin__ = tty_text # type: ignore[assignment] + sys.__stdout__ = tty_text # type: ignore[assignment] + + # curses.wrapper calls initscr which honours sys.__stdin__ / __stdout__ + # on some builds; use newterm where available. + screen = curses.initscr() + curses.noecho() + curses.cbreak() + screen.keypad(True) + + try: + result = _picker_loop(screen, items, title=title) + finally: + screen.keypad(False) + curses.nocbreak() + curses.echo() + curses.endwin() + except Exception: + return None + finally: + sys.__stdin__ = orig_stdin # type: ignore[assignment] + sys.__stdout__ = orig_stdout # type: ignore[assignment] + + return result + + +def _picker_loop(screen, items: list[str], *, title: str) -> Optional[str]: + query = "" + cursor = 0 + + while True: + filtered = _filter_items(items, query) + + # Clamp cursor into the visible list. + if not filtered: + cursor = 0 + elif cursor >= len(filtered): + cursor = len(filtered) - 1 + + try: + _render(screen, filtered, cursor, query=query, title=title) + except curses.error: + # Terminal too small or write error — bail out. + return None + + try: + key = screen.getch() + except KeyboardInterrupt: + return None + + if key in _CANCEL_KEYS: + return None + + if key in (curses.KEY_ENTER, _KEY_ENTER_ALT, ord("\r")): + return filtered[cursor] if filtered else None + + if key in (curses.KEY_UP, ord("k")): + if cursor > 0: + cursor -= 1 + + elif key in (curses.KEY_DOWN, ord("j")): + if cursor < len(filtered) - 1: + cursor += 1 + + elif key in (curses.KEY_BACKSPACE, _KEY_BACKSPACE_WIN, 127): + query = query[:-1] + # After narrowing the filter, keep cursor in range. + new_filtered = _filter_items(items, query) + if cursor >= len(new_filtered): + cursor = max(0, len(new_filtered) - 1) + + elif 32 <= key <= 126: + # Printable ASCII — append to query and reset cursor so the + # top of the newly-filtered list is selected. + query += chr(key) + cursor = 0 + + +def _filter_items(items: list[str], query: str) -> list[str]: + if not query: + return list(items) + q = query.lower() + return [i for i in items if q in i.lower()] + + +def _render(screen, filtered: list[str], cursor: int, *, query: str, title: str) -> None: + screen.erase() + rows, cols = screen.getmaxyx() + min_rows = 5 + + if rows < min_rows: + raise curses.error("terminal too small") + + row = 0 + + if title and row < rows - 1: + _addstr_safe(screen, row, 0, title[:cols - 1], curses.A_BOLD) + row += 1 + + filter_label = f"Filter: {query}" + if row < rows - 1: + _addstr_safe(screen, row, 0, filter_label[:cols - 1]) + row += 1 + + sep = "─" * min(cols - 1, 40) + if row < rows - 1: + _addstr_safe(screen, row, 0, sep) + row += 1 + + list_start = row + # Reserve two rows for separator + help line at bottom. + list_rows = rows - list_start - 2 + if list_rows < 1: + return + + # Scroll window: keep cursor visible. + scroll = max(0, cursor - list_rows + 1) + visible = filtered[scroll: scroll + list_rows] + + for idx, item in enumerate(visible): + abs_idx = scroll + idx + attr = curses.A_REVERSE if abs_idx == cursor else curses.A_NORMAL + prefix = "> " if abs_idx == cursor else " " + line = (prefix + item)[:cols - 1] + if row < rows - 1: + _addstr_safe(screen, row, 0, line, attr) + row += 1 + + if row < rows - 1: + _addstr_safe(screen, row, 0, sep) + row += 1 + + help_line = "[↑↓/jk] move [Enter] select [Esc/q] cancel" + if row < rows: + _addstr_safe(screen, min(rows - 1, row), 0, help_line[:cols - 1]) + + screen.refresh() + + +def _addstr_safe(screen, row: int, col: int, text: str, attr: int = curses.A_NORMAL) -> None: + try: + screen.addstr(row, col, text, attr) + except curses.error: + pass diff --git a/docs/prds/0051-launch-selector.md b/docs/prds/0051-launch-selector.md index 4aedd26..c31e11f 100644 --- a/docs/prds/0051-launch-selector.md +++ b/docs/prds/0051-launch-selector.md @@ -1,6 +1,6 @@ # PRD 0051: Launch selector -- **Status:** Draft +- **Status:** Active - **Author:** claude - **Created:** 2026-06-04 - **Issue:** #185 diff --git a/tests/unit/test_cli_start_selector.py b/tests/unit/test_cli_start_selector.py new file mode 100644 index 0000000..1e50e3c --- /dev/null +++ b/tests/unit/test_cli_start_selector.py @@ -0,0 +1,143 @@ +"""Unit: cmd_start selector dispatch (PRD 0051). + +Tests that cmd_start calls filter_select when name / backend are absent, +skips them when both are explicit, and returns 0 on cancel. + +All actual launch work is stubbed so no container is created. +""" + +from __future__ import annotations + +import os +import sys +import types +import unittest +from unittest.mock import MagicMock, patch, call + +import bot_bottle.cli.start as start_mod +import bot_bottle.cli.tui as tui_mod + + +def _make_manifest(agent_names: list[str]): + manifest = MagicMock() + manifest.agents = {name: MagicMock() for name in agent_names} + return manifest + + +class TestCmdStartSelector(unittest.TestCase): + """Drive cmd_start with a minimal set of stubs.""" + + def setUp(self): + # Stub Manifest.resolve so no on-disk manifest is needed. + self._manifest = _make_manifest(["researcher", "implementer"]) + self._resolve_patch = patch( + "bot_bottle.cli.start.Manifest.resolve", + return_value=self._manifest, + ) + self._resolve_patch.start() + + # Stub _launch_bottle so no real container work happens. + self._launch_patch = patch( + "bot_bottle.cli.start._launch_bottle", + return_value=0, + ) + self._launch_mock = self._launch_patch.start() + + # Stub filter_select to avoid opening /dev/tty. + self._tui_patch = patch.object(tui_mod, "filter_select") + self._tui_mock = self._tui_patch.start() + + # Ensure BOT_BOTTLE_BACKEND is absent so the backend picker fires. + self._env_patch = patch.dict(os.environ, {}, clear=False) + self._env_patch.start() + os.environ.pop("BOT_BOTTLE_BACKEND", None) + + def tearDown(self): + self._resolve_patch.stop() + self._launch_patch.stop() + self._tui_patch.stop() + self._env_patch.stop() + + # ------------------------------------------------------------------ + # Both explicit — no picker shown + # ------------------------------------------------------------------ + + def test_both_explicit_skips_picker(self): + self._tui_mock.return_value = "researcher" + rc = start_mod.cmd_start(["--backend=docker", "researcher"]) + self.assertEqual(0, rc) + self._tui_mock.assert_not_called() + self._launch_mock.assert_called_once() + _, kwargs = self._launch_mock.call_args + self.assertEqual("docker", kwargs["backend_name"]) + + # ------------------------------------------------------------------ + # Agent absent → agent picker fires; backend explicit + # ------------------------------------------------------------------ + + def test_agent_absent_shows_agent_picker(self): + self._tui_mock.return_value = "researcher" + rc = start_mod.cmd_start(["--backend=docker"]) + self.assertEqual(0, rc) + self._tui_mock.assert_called_once() + call_kwargs = self._tui_mock.call_args + self.assertEqual(["implementer", "researcher"], call_kwargs[0][0]) + self.assertIn("agent", call_kwargs[1]["title"].lower()) + + def test_agent_picker_cancel_returns_0(self): + self._tui_mock.return_value = None + rc = start_mod.cmd_start(["--backend=docker"]) + self.assertEqual(0, rc) + self._launch_mock.assert_not_called() + + # ------------------------------------------------------------------ + # Agent explicit, backend absent → backend picker fires + # ------------------------------------------------------------------ + + def test_backend_absent_shows_backend_picker(self): + self._tui_mock.return_value = "docker" + rc = start_mod.cmd_start(["researcher"]) + self.assertEqual(0, rc) + self._tui_mock.assert_called_once() + call_kwargs = self._tui_mock.call_args + self.assertIn("backend", call_kwargs[1]["title"].lower()) + + def test_backend_picker_cancel_returns_0(self): + self._tui_mock.return_value = None + rc = start_mod.cmd_start(["researcher"]) + self.assertEqual(0, rc) + self._launch_mock.assert_not_called() + + def test_bot_bottle_backend_env_skips_backend_picker(self): + os.environ["BOT_BOTTLE_BACKEND"] = "docker" + try: + rc = start_mod.cmd_start(["researcher"]) + finally: + os.environ.pop("BOT_BOTTLE_BACKEND", None) + self.assertEqual(0, rc) + self._tui_mock.assert_not_called() + + # ------------------------------------------------------------------ + # Both absent → agent picker then backend picker + # ------------------------------------------------------------------ + + def test_both_absent_shows_both_pickers_in_order(self): + self._tui_mock.side_effect = ["researcher", "docker"] + rc = start_mod.cmd_start([]) + self.assertEqual(0, rc) + self.assertEqual(2, self._tui_mock.call_count) + first_title = self._tui_mock.call_args_list[0][1]["title"].lower() + second_title = self._tui_mock.call_args_list[1][1]["title"].lower() + self.assertIn("agent", first_title) + self.assertIn("backend", second_title) + + def test_both_absent_agent_cancel_skips_backend_picker(self): + self._tui_mock.side_effect = [None] + rc = start_mod.cmd_start([]) + self.assertEqual(0, rc) + self.assertEqual(1, self._tui_mock.call_count) + self._launch_mock.assert_not_called() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_cli_tui.py b/tests/unit/test_cli_tui.py new file mode 100644 index 0000000..1f2ca5b --- /dev/null +++ b/tests/unit/test_cli_tui.py @@ -0,0 +1,50 @@ +"""Unit tests for bot_bottle.cli.tui — filter_select internals. + +We test the pure-Python logic (_filter_items, cursor movement, confirm, +cancel) by exercising the internal helpers directly, without spinning up +a real curses session (which requires a TTY). +""" + +from __future__ import annotations + +import unittest + +from bot_bottle.cli.tui import _filter_items, filter_select + + +class TestFilterItems(unittest.TestCase): + def setUp(self): + self.items = ["researcher", "implementer", "codex-researcher", "reviewer"] + + def test_empty_query_returns_all(self): + self.assertEqual(self.items, _filter_items(self.items, "")) + + def test_query_filters_case_insensitively(self): + result = _filter_items(self.items, "RESEARCH") + self.assertEqual(["researcher", "codex-researcher"], result) + + def test_no_match_returns_empty(self): + self.assertEqual([], _filter_items(self.items, "zzz")) + + def test_partial_match(self): + result = _filter_items(self.items, "impl") + self.assertEqual(["implementer"], result) + + def test_empty_items_returns_empty(self): + self.assertEqual([], _filter_items([], "foo")) + + +class TestFilterSelectEmptyItems(unittest.TestCase): + def test_returns_none_for_empty_list(self): + # No TTY needed — the short-circuit fires before opening tty. + result = filter_select([], title="Pick one", tty_path="/dev/null") + self.assertIsNone(result) + + def test_returns_none_when_tty_unavailable(self): + # /nonexistent is guaranteed to not open. + result = filter_select(["a", "b"], tty_path="/nonexistent/tty") + self.assertIsNone(result) + + +if __name__ == "__main__": + unittest.main()