Files
wehub-resource-sync 4b6817381b
CI (OpenClaw E2E) / openclaw test (push) Has been cancelled
CI / coverage-report (push) Has been cancelled
CI / test-kubernetes (push) Has been cancelled
CI / should-run-thorough (push) Has been cancelled
CI / test-thorough (cloudwatch-demo) (push) Has been cancelled
CI / test-thorough (flink-ecs) (push) Has been cancelled
CI / test-thorough (upstream-lambda) (push) Has been cancelled
CI / test-thorough (prefect-ecs-fargate) (push) Has been cancelled
Release / build-binaries (zip, opensre.exe, onefile, windows-latest, windows-x64) (push) Has been cancelled
Benchmark image — build + push to ECR (any adapter) / build + push (push) Has been cancelled
CI / quality (ubuntu-latest) (push) Has been cancelled
CI / test (tools-runtime) (push) Has been cancelled
CI / test (e2e-general) (push) Has been cancelled
CI / test (cli-runtime) (push) Has been cancelled
CI / test (e2e-provider-and-openclaw) (push) Has been cancelled
CI / test (integrations-and-misc) (push) Has been cancelled
Release / verify (push) Has been cancelled
Release / build-python-dist (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, macos-15-intel, darwin-x64) (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, macos-latest, darwin-arm64) (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04, linux-x64) (push) Has been cancelled
Release / publish-release (push) Has been cancelled
Release / publish-main-release (push) Has been cancelled
Interactive Shell Live (PR + post-merge) / turn-checks (no-LLM) (push) Has been cancelled
CodeQL / Analyze (python) (push) Has been cancelled
Interactive Shell Live (PR + post-merge) / turn-live shard ${{ matrix.shard_index }} (push) Has been cancelled
Release / prepare (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04-arm, linux-arm64) (push) Has been cancelled
Synthetic Deterministic Tests / Synthetic offline (deterministic) (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 13:10:45 +08:00

286 lines
8.7 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Interactive choice helpers for TTY-first REPL flows.
Inline menus render in the terminal scrollback (below the submitted command),
not as a separate prompt-toolkit full-screen dialog — important when the REPL
already runs under asyncio.
Each menu erases itself on exit (selection or Esc) so nested menus never
pile up — only the result output and the next level appear on screen.
"""
from __future__ import annotations
import os
import shutil
import sys
from typing import Literal
from rich.console import Console
from rich.markup import escape
import platform.terminal.theme as ui_theme
from surfaces.interactive_shell.ui.components.key_reader import read_key_unix, read_key_windows
_HINT = "↑↓/j/k/Tab Enter/Space Esc/q"
CRUMB_SEP = " "
# Blank line after the submitted slash line before the menu header (all pickers).
_MENU_LEADING_LINES = 1
_TERMINAL_NEWLINE = "\r\n"
MenuAction = Literal["up", "down", "enter", "cancel", "eof", "ignore"]
def repl_tty_interactive() -> bool:
"""Return True when stdin/stdout support an interactive picker UI."""
return bool(sys.stdin.isatty() and sys.stdout.isatty())
def ensure_tty_column_zero() -> None:
"""Reset the cursor column before Rich output when a TTY is active."""
if repl_tty_interactive():
reset_tty_column()
def prepare_repl_output_line() -> None:
"""Begin Rich output on a new line after inline menu I/O."""
if repl_tty_interactive():
sys.stdout.write(_TERMINAL_NEWLINE)
reset_tty_column()
def repl_section_break(console: Console) -> None:
"""Blank line + dim rule between an inline menu step and Rich output."""
prepare_repl_output_line()
console.print()
console.rule(characters="─", style=str(ui_theme.DIM))
console.print()
# ── raw key reader ───────────────────────────────────────────────────────────
def _read_action() -> MenuAction:
"""Map a raw keypress to a menu action.
Delegates terminal I/O to :mod:`key_reader` and applies
choice_menu-specific overrides: Tab → ``"down"``,
right-arrow → ``"enter"``, left-arrow → ``"ignore"``.
"""
key = read_key_windows() if os.name == "nt" else read_key_unix()
if key == "tab":
return "down"
if key == "right":
return "enter"
if key == "left":
return "ignore"
return key # type: ignore[return-value]
def read_menu_action() -> MenuAction:
"""Read one normalized inline-menu action from stdin."""
return _read_action()
# ── rendering helpers ────────────────────────────────────────────────────────
def _cols() -> int:
return max(40, shutil.get_terminal_size(fallback=(80, 24)).columns)
def menu_columns() -> int:
"""Return the current terminal width floor used by inline menus."""
return _cols()
def _rule(width: int) -> str:
return "─" * width
def _pad(sym: str, label: str, width: int) -> str:
content = f" {sym} {label}"
pad = width - len(content)
return content + (" " * pad if pad > 0 else "")
def _menu_height(crumb: str, labels: list[str]) -> int:
# leading, title, [crumb], rule, blank, choices, blank, hint
return _MENU_LEADING_LINES + 1 + (1 if crumb else 0) + 1 + 1 + len(labels) + 1 + 1
def write_menu_line(text: str = "") -> None:
"""Write one inline-menu line at column zero even while the terminal is in raw mode."""
if text:
sys.stdout.write(f"\r{text}{_TERMINAL_NEWLINE}")
return
sys.stdout.write(_TERMINAL_NEWLINE)
def _erase_menu_block(height: int) -> None:
if height:
sys.stdout.write(f"\r\x1b[{height}A\r\x1b[J")
reset_tty_column()
def reset_tty_column() -> None:
"""Return the cursor to column zero after inline menu I/O.
Menu rows are padded to the terminal width, so the cursor often ends on a
high column. Rich output that follows must start at column zero or tables
render as a diagonal block of leading whitespace.
"""
sys.stdout.write("\r")
sys.stdout.flush()
def erase_menu_lines(height: int) -> None:
"""Erase a previously-rendered inline menu block."""
_erase_menu_block(height)
def _draw_menu(
*,
title: str,
crumb: str,
labels: list[str],
index: int,
erase_lines: int,
) -> None:
out = sys.stdout
w = _cols()
if erase_lines:
_erase_menu_block(erase_lines)
for _ in range(_MENU_LEADING_LINES):
write_menu_line()
# title
write_menu_line(f"{ui_theme.PROMPT_ACCENT_ANSI}{title}{ui_theme.ANSI_RESET}")
# breadcrumb path
if crumb:
write_menu_line(f"{ui_theme.DIM_COUNTER_ANSI}{crumb}{ui_theme.ANSI_RESET}")
# separator below header
write_menu_line(f"{ui_theme.DIM_COUNTER_ANSI}{_rule(w)}{ui_theme.ANSI_RESET}")
write_menu_line()
# choices
for i, label in enumerate(labels):
here = i == index
sym = ">" if here else " "
padded = _pad(sym, label, w)
if here:
write_menu_line(f"{ui_theme.MENU_SELECTION_ROW_ANSI}{padded}{ui_theme.ANSI_RESET}")
else:
write_menu_line(f"{ui_theme.DIM_COUNTER_ANSI}{padded}{ui_theme.ANSI_RESET}")
write_menu_line()
write_menu_line(f"{ui_theme.DIM_COUNTER_ANSI}{_HINT}{ui_theme.ANSI_RESET}")
out.flush()
def _erase_menu(crumb: str, labels: list[str]) -> None:
"""Move cursor up to the start of this menu block and wipe it."""
height = _menu_height(crumb, labels)
_erase_menu_block(height)
sys.stdout.flush()
# ── picker loop ──────────────────────────────────────────────────────────────
def _pick(
*,
title: str,
crumb: str,
labels: list[str],
initial_index: int = 0,
) -> int | None:
"""Draw an inline menu, let user navigate, erase on exit. Returns index or None."""
if not labels:
return None
idx = initial_index % len(labels)
height = _menu_height(crumb, labels)
first = True
while True:
_draw_menu(
title=title,
crumb=crumb,
labels=labels,
index=idx,
erase_lines=0 if first else height,
)
first = False
action = _read_action()
if action == "enter":
_erase_menu(crumb, labels)
return idx
if action in ("cancel", "eof"):
_erase_menu(crumb, labels)
return None
if action == "ignore":
continue
if action == "up":
idx = (idx - 1) % len(labels)
elif action == "down":
idx = (idx + 1) % len(labels)
# ── public API ───────────────────────────────────────────────────────────────
def repl_choose_one(
*,
title: str,
choices: list[tuple[str, str]],
breadcrumb: str = "",
initial_value: str | None = None,
) -> str | None:
"""Show an inline erasing arrow-key menu; return selected value or None on Esc.
``breadcrumb`` is a slash-separated path shown dimly below the title, e.g.
``/model set``. Only call when :func:`repl_tty_interactive` is True.
"""
from surfaces.interactive_shell.ui.components.cpr_stdin import drain_stale_cpr_bytes
if not choices or not repl_tty_interactive():
return None
drain_stale_cpr_bytes()
crumb = breadcrumb
labels = [label for _value, label in choices]
initial_index = 0
if initial_value is not None:
for index, (value, _label) in enumerate(choices):
if value == initial_value:
initial_index = index
break
picked = _pick(title=title, crumb=crumb, labels=labels, initial_index=initial_index)
if picked is None:
return None
value = choices[picked][0]
return value if isinstance(value, str) else None
def print_valid_choice_list(
console: Console,
*,
title: str,
choices: list[str],
) -> None:
"""Print one choice per line for scan-friendly fallback/error messaging."""
if not choices:
return
console.print(f"[{ui_theme.SECONDARY}]{title}[/]")
for choice in choices:
console.print(f"[{ui_theme.SECONDARY}] - {escape(choice)}[/]")
__all__ = [
"CRUMB_SEP",
"erase_menu_lines",
"menu_columns",
"print_valid_choice_list",
"read_menu_action",
"repl_choose_one",
"ensure_tty_column_zero",
"prepare_repl_output_line",
"repl_section_break",
"repl_tty_interactive",
"reset_tty_column",
"write_menu_line",
]