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

115 lines
4.5 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.
"""Prompt text, hint, placeholder, and submitted-turn rendering."""
from __future__ import annotations
from prompt_toolkit.application.current import get_app_or_none
from prompt_toolkit.formatted_text import ANSI
from rich.console import Console
from rich.text import Text
from platform.terminal import theme as ui_theme
from surfaces.interactive_shell.runtime import Session
from surfaces.interactive_shell.ui.banner.banner_state import integration_display_name
from surfaces.interactive_shell.ui.input_prompt.completion import completion_preview_hint_ansi
from surfaces.interactive_shell.ui.input_prompt.layout import _short_meta, _terminal_columns
_PROMPT_RULE_CHAR = "─"
_DEFAULT_PLACEHOLDER_TEXT = "Type a message, /command, or paste an alert"
_DEFAULT_PLACEHOLDER_ANSI = ANSI(
f"{ui_theme.ANSI_DIM}{_DEFAULT_PLACEHOLDER_TEXT}{ui_theme.ANSI_RESET}"
)
def _prompt_rule_line(width: int) -> str:
return _PROMPT_RULE_CHAR * max(width, 1)
def _prompt_rule_ansi() -> str:
return (
f"{ui_theme.PROMPT_FRAME_ANSI}{_prompt_rule_line(_terminal_columns())}{ui_theme.ANSI_RESET}"
)
def _prompt_turn_number(session: Session) -> int:
"""1-based index for the turn about to be entered or just submitted."""
return len(session.history) + 1
def _prompt_counter_text(session: Session) -> str:
return f"[{_prompt_turn_number(session)}] "
def _prompt_prefix_text(session: Session) -> str:
return f"{_prompt_counter_text(session)} "
def _prompt_line_ansi(session: Session) -> ANSI:
counter = _prompt_counter_text(session)
prefix = f"{ui_theme.DIM_COUNTER_ANSI}{counter}{ui_theme.ANSI_RESET}"
return ANSI(f"{prefix}{ui_theme.PROMPT_ACCENT_ANSI}{ui_theme.ANSI_RESET} ")
def _prompt_message(session: Session) -> ANSI:
"""Top border rule plus cursor line: the top two rows of the input box."""
return ANSI(f"{_prompt_rule_ansi()}\n{_prompt_line_ansi(session).value}")
def render_submitted_prompt(console: Console, session: Session, text: str) -> None:
"""Render the submitted user turn above the streamed assistant response."""
lines = text.splitlines() or [""]
continuation_prefix = " " * len(_prompt_prefix_text(session))
rendered = Text()
counter = _prompt_counter_text(session)
# Rich's Style.parse() reads the bare str value of a _LazyRichStyle (""),
# so resolve to a concrete string at the call site to keep palette colors.
rendered.append(counter, style=str(ui_theme.DIM))
rendered.append(" ", style=f"bold {ui_theme.HIGHLIGHT}")
rendered.append(lines[0], style=str(ui_theme.TEXT))
for line in lines[1:]:
rendered.append("\n")
rendered.append(continuation_prefix, style=str(ui_theme.DIM))
rendered.append(line, style=str(ui_theme.TEXT))
console.print(rendered)
def resolve_prompt_prefix_ansi(*, inline_spinner: str, idle_hint: str) -> str:
"""Choose the prompt's top context line: spinner, completion preview, or idle hint."""
if inline_spinner:
return inline_spinner
preview = completion_preview_hint_ansi()
return preview or idle_hint
def resolve_idle_hint_ansi(session: Session) -> str:
"""Dim hint line above the prompt rule: shortcuts plus connected integrations."""
parts = ["/ for commands", "↑↓ history"]
if session.configured_integrations_known and session.configured_integrations:
max_shown = 4
names = [integration_display_name(name) for name in sorted(session.configured_integrations)]
shown = names[:max_shown]
overflow = len(names) - len(shown)
integration_segment = " · ".join(shown)
if overflow:
integration_segment += f" +{overflow}"
parts.append(integration_segment)
app = get_app_or_none()
if app is not None and app.current_buffer.text:
parts.append("esc to clear")
hint = " · ".join(parts)
return f"{ui_theme.DIM_ANSI}{hint}{ui_theme.ANSI_RESET}"
def resolve_prompt_placeholder(session: Session) -> ANSI:
"""Contextual ghost text when the input buffer is empty."""
parts: list[str] = []
if session.terminal.trust_mode:
parts.append("trust on")
running = session.task_registry.running_count()
if running:
parts.append(f"{running} task{'s' if running != 1 else ''} running")
if session.resumed_from_name:
parts.append(f"resumed: {_short_meta(session.resumed_from_name, max_len=32)}")
if parts:
return ANSI(f"{ui_theme.ANSI_DIM}{' · '.join(parts)}{ui_theme.ANSI_RESET}")
return _DEFAULT_PLACEHOLDER_ANSI