diff --git a/CHANGELOG.md b/CHANGELOG.md index cdc0a181..7b064076 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,7 @@ GitHub Releases page; `0.8.0` is the new starting line. ## Unreleased +- Reduce the shell prompt session to a compatibility façade over deep `prompting/` modules (config, keybindings, completion menus, narrow shell-facing methods), with Unicode cell-width coverage and documented module ownership in the architecture guide. - Render image/audio/video and unknown content parts as payload-free labels in the live view, roll nested subagent activity (up to 16 levels) under the correct root tool card, and stop provider-remapped tool results from starting replay user turns. - **Behavior change:** `!` shell commands now run through the detected configured shell (` -c`, PowerShell `-command` on Windows) instead of the implicit `/bin/sh`/`cmd.exe`, with separate 1 MiB stdout/stderr caps and cancellation cleanup; existing `cmd.exe`-syntax commands on Windows may need updating. - Distinguish background auto-trigger grace expiry from user input activity with typed prompt events, and bound streamed tool-argument label scans to 1 KiB growth boundaries. diff --git a/docs/en/customization/architecture.md b/docs/en/customization/architecture.md index 38fc2ad4..0025d6b7 100644 --- a/docs/en/customization/architecture.md +++ b/docs/en/customization/architecture.md @@ -231,6 +231,33 @@ events include `StepBegin`, `StepRetry`, `StepInterrupted`, `ToolExecutionStarte The shell can run with a working directory inside its subtree, so `src/pythinker_code/ui/` is a candidate for a focused nested guide on prompt, visualization, and component layout. +### Shell prompt deep modules + +`src/pythinker_code/ui/shell/prompt.py` is a compatibility façade: it keeps the public +`CustomPromptSession` surface and re-exports legacy underscore helper names, while the +implementation lives in the `src/pythinker_code/ui/shell/prompting/` deep modules. + +| Path | Owns | +| --- | --- | +| `prompting/frame.py` | One-frame snapshot flow: `PromptFrameCollector` samples every render delegate exactly once per frame into an immutable `PromptFrame`. | +| `prompting/state.py` | Reducer-style prompt state: UI events flow through the reducer; render code reads state, never mutates it. | +| `prompting/lifecycle.py` | Async task ownership and closers: startup/shutdown symmetry for session-owned background tasks. | +| `prompting/renderer.py` | Scene row allocation and rendering within the terminal height/width budget (cell-width measured). | +| `prompting/footer.py` | `FooterViewModel` plus content precedence and cell-width truncation, consumed by both toolbar adapters (pythinker and card styles). | +| `prompting/toasts.py`, `git_status.py`, `history.py`, `clipboard.py` | Session-owned resources: toast queue, Git status snapshots, prompt history persistence, clipboard integration. | +| `prompting/config.py` | `PromptConfig` / `PromptProviders`: constructor-time wiring of callbacks and providers. | +| `prompting/keybindings.py` | `build_prompt_key_bindings` over the `PromptController` protocol. | +| `prompting/completion/` | Canonical `CompletionContext`, slash-command completion, workspace-indexed `@` file mentions, and completion menus. | + +Workspace and Git snapshots publish asynchronously under a generation counter: each refresh +owns its generation, and results from a superseded generation are discarded rather than +rendered stale. + +Compatibility policy: legacy underscore helper names re-exported from `prompt.py` delegate to +the session modules and remain import-compatible, but private implementation helpers are not +supported extension interfaces and may change without notice; extend via the documented +public surfaces (`CustomPromptSession`, `PromptConfig`/`PromptProviders`, tool renderers). + ## ACP server | Path | Purpose | Key entry points and interfaces | diff --git a/src/pythinker_code/ui/shell/__init__.py b/src/pythinker_code/ui/shell/__init__.py index 3939b3f2..01054805 100644 --- a/src/pythinker_code/ui/shell/__init__.py +++ b/src/pythinker_code/ui/shell/__init__.py @@ -2081,8 +2081,7 @@ def _maybe_present_pending_approvals(self) -> None: def _get_default_buffer_text_and_cursor(self) -> tuple[str, int]: if self._prompt_session is None: return "", 0 - buf = self._prompt_session._session.default_buffer # pyright: ignore[reportPrivateUsage] - return buf.text, buf.cursor_position + return self._prompt_session.input_state() def _activate_prompt_approval_modal(self) -> None: if self._prompt_session is None: @@ -2101,7 +2100,7 @@ def _activate_prompt_approval_modal(self) -> None: current_request, on_response=self._handle_prompt_approval_response, buffer_state_provider=self._get_default_buffer_text_and_cursor, - text_expander=self._prompt_session._get_placeholder_manager().serialize_for_history, # pyright: ignore[reportPrivateUsage] + text_expander=self._prompt_session.serialize_for_history, ) self._prompt_session.attach_modal(self._approval_modal) else: diff --git a/src/pythinker_code/ui/shell/prompt.py b/src/pythinker_code/ui/shell/prompt.py index 9330ed34..e6357c00 100644 --- a/src/pythinker_code/ui/shell/prompt.py +++ b/src/pythinker_code/ui/shell/prompt.py @@ -10,7 +10,6 @@ from collections import deque from collections.abc import Awaitable, Callable, Sequence from contextvars import Token -from dataclasses import dataclass from datetime import datetime from enum import Enum from hashlib import md5 @@ -21,8 +20,7 @@ from prompt_toolkit.application import Application from prompt_toolkit.application.current import get_app_or_none from prompt_toolkit.buffer import Buffer -from prompt_toolkit.completion import Completion, merge_completers -from prompt_toolkit.data_structures import Point +from prompt_toolkit.completion import merge_completers from prompt_toolkit.document import Document from prompt_toolkit.filters import Condition, has_completions from prompt_toolkit.formatted_text import ( @@ -32,8 +30,7 @@ to_formatted_text, ) from prompt_toolkit.history import InMemoryHistory -from prompt_toolkit.key_binding import KeyBindings, KeyPressEvent -from prompt_toolkit.keys import Keys +from prompt_toolkit.key_binding import KeyPressEvent from prompt_toolkit.layout.containers import ( ConditionalContainer, DynamicContainer, @@ -42,7 +39,7 @@ Window, WindowRenderInfo, ) -from prompt_toolkit.layout.controls import BufferControl, UIContent, UIControl +from prompt_toolkit.layout.controls import BufferControl, UIContent from prompt_toolkit.layout.dimension import Dimension from prompt_toolkit.layout.margins import Margin from prompt_toolkit.layout.menus import CompletionsMenu @@ -99,9 +96,11 @@ from pythinker_code.ui.shell.prompting.clipboard import ( is_media_clipboard_available as is_media_clipboard_available, ) -from pythinker_code.ui.shell.prompting.completion.context import ( - CompletionKind, - parse_completion_context, +from pythinker_code.ui.shell.prompting.completion.menus import ( + LocalFileMentionMenuControl, + SlashCommandMenuControl, + truncate_to_width, + wrap_to_width, ) from pythinker_code.ui.shell.prompting.completion.slash import ( InputHighlightLexer, @@ -114,6 +113,11 @@ HostFileMentionCompleter, WorkspaceIndex, ) +from pythinker_code.ui.shell.prompting.config import ( + BgTaskCounts, + PromptConfig, + PromptProviders, +) from pythinker_code.ui.shell.prompting.git_status import ( bind_git_status_index, current_git_snapshot, @@ -123,6 +127,7 @@ HistoryEntry, redact_history_secrets, ) +from pythinker_code.ui.shell.prompting.keybindings import build_prompt_key_bindings from pythinker_code.ui.shell.prompting.lifecycle import PromptLifecycle from pythinker_code.ui.shell.prompting.state import ( BufferObserved, @@ -240,37 +245,9 @@ def _prompt_rule(columns: int) -> str: return "─" * max(0, columns - 1) -def _truncate_to_width(text: str, width: int) -> str: - if width <= 0: - return "" - - total = 0 - chars: list[str] = [] - for ch in text: - ch_width = get_cwidth(ch) - if total + ch_width > width: - break - chars.append(ch) - total += ch_width - - if total == get_cwidth(text): - return text + (" " * max(0, width - total)) - - ellipsis = "..." - ellipsis_width = get_cwidth(ellipsis) - if width <= ellipsis_width: - return "." * width - - available = width - ellipsis_width - total = 0 - chars = [] - for ch in text: - ch_width = get_cwidth(ch) - if total + ch_width > available: - break - chars.append(ch) - total += ch_width - return "".join(chars) + ellipsis + (" " * max(0, width - total - ellipsis_width)) +# Compatibility aliases: the width helpers moved to +# pythinker_code.ui.shell.prompting.completion.menus with the menu controls. +_truncate_to_width = truncate_to_width def _formatted_text_display_rows(fragments: FormattedText, columns: int) -> list[FormattedText]: @@ -455,63 +432,7 @@ def _prompt_preamble_max_rows(terminal_rows: int | None) -> int: return PromptSceneBudget(terminal_rows=terminal_rows).preamble_rows -def _wrap_to_width(text: str, width: int, *, max_lines: int | None = None) -> list[str]: - if width <= 0: - return [] - - words = text.split() - if not words: - return [""] - - lines: list[str] = [] - current_words: list[str] = [] - current_width = 0 - index = 0 - - while index < len(words): - word = words[index] - word_width = get_cwidth(word) - separator_width = 1 if current_words else 0 - - if current_words and current_width + separator_width + word_width <= width: - current_words.append(word) - current_width += separator_width + word_width - index += 1 - continue - - if not current_words and word_width <= width: - current_words.append(word) - current_width = word_width - index += 1 - continue - - if not current_words and word_width > width: - current_words.append(_truncate_to_width(word, width).rstrip()) - current_width = get_cwidth(current_words[0]) - index += 1 - - lines.append(" ".join(current_words)) - current_words = [] - current_width = 0 - - if max_lines is not None and len(lines) == max_lines: - remaining = " ".join(words[index:]) - if remaining: - prefix = f"{lines[-1]} " if lines[-1] else "" - lines[-1] = _truncate_to_width(prefix + remaining, width).rstrip() - return lines - - if current_words: - line = " ".join(current_words) - if max_lines is not None and len(lines) + 1 > max_lines: - if lines: - lines[-1] = _truncate_to_width(f"{lines[-1]} {line}", width).rstrip() - else: - lines.append(_truncate_to_width(line, width).rstrip()) - else: - lines.append(line) - - return lines +_wrap_to_width = wrap_to_width def _find_prompt_float_container(layout_container: object) -> FloatContainer | None: @@ -622,565 +543,6 @@ def _walk(node: object) -> bool: return _walk(root) -class SlashCommandMenuControl(UIControl): - """Render slash command completions as a full-width menu that matches the shell UI.""" - - _MAX_EXPANDED_META_LINES = 3 - # One blank line is reserved above the list as breathing room from the input - # row, so the menu reads as its own region rather than crowding what's typed. - _GAP_LINES = 1 - # A persistent footer block at the bottom: a blank separator line plus the - # navigation legend (which folds in the overflow count when the list scrolls). - # The separator gives the legend the same breathing room as the top gap. - _FOOTER_LINES = 2 - _FOOTER_LEGEND = "Enter to select · ↑/↓ to navigate · Esc to cancel" - - def __init__( - self, - *, - left_padding: Callable[[], int], - scroll_offset: int = 1, - ) -> None: - self._left_padding = left_padding - self._scroll_offset = scroll_offset - - def has_focus(self) -> bool: - return False - - def preferred_width(self, max_available_width: int) -> int | None: - return max_available_width - - def preferred_height( - self, - width: int, - max_available_height: int, - wrap_lines: bool, - get_line_prefix: Callable[..., AnyFormattedText] | None, - ) -> int | None: - app = get_app_or_none() - complete_state = ( - getattr(app.current_buffer, "complete_state", None) if app is not None else None - ) - if complete_state is None: - return 0 - completions = complete_state.completions - if not completions: - return 0 - selected_index = complete_state.complete_index - if selected_index is None: - content_height = len(completions) - else: - menu_width = max(0, width - self._left_padding()) - marker_width = 2 - command_width = self._command_column_width(completions, menu_width, marker_width) - gap_width = 3 if menu_width > command_width + 6 else 1 - meta_width = max(0, menu_width - marker_width - command_width - gap_width) - selected_meta_lines = self._selected_meta_lines( - completions[selected_index].display_meta_text, - meta_width, - ) - content_height = (len(completions) - 1) + len(selected_meta_lines) - # Reserve the gap line above the list and the footer line below it. When - # the list is taller than the space the window allows, the window caps the - # height and create_content lays the list out within whatever rows remain. - chrome = self._GAP_LINES + self._FOOTER_LINES - return min(max_available_height, content_height + chrome) - - def create_content(self, width: int, height: int) -> UIContent: - app = get_app_or_none() - complete_state = ( - getattr(app.current_buffer, "complete_state", None) if app is not None else None - ) - if complete_state is None or not complete_state.completions: - return UIContent() - - completions = complete_state.completions - selected_index = complete_state.complete_index - match_prefix_len = self._match_prefix_len(app) - - menu_width = max(0, width - self._left_padding()) - marker_width = 2 - command_width = self._command_column_width(completions, menu_width, marker_width) - gap_width = 3 if menu_width > command_width + 6 else 1 - meta_width = max(0, menu_width - marker_width - command_width - gap_width) - - total_rows = max(1, height) - # The gap line above the list and the footer line below it are always - # present, so the list itself lays out within the remaining rows. - item_rows = max(1, total_rows - self._GAP_LINES - self._FOOTER_LINES) - - rendered_lines: list[FormattedText] = [self._blank_line()] - cursor_y = 0 - - if selected_index is None: - # Pre-highlight index 0 even before the user navigates: pressing - # Enter accepts the first completion, so the visual state should - # match that behavior. Without this the menu looks ambiguous (no - # row highlighted) but Enter still commits the top row. - shown = min(len(completions), item_rows) - for index in range(shown): - rendered_lines.append( - self._render_single_line_item( - width=width, - completion=completions[index], - marker_width=marker_width, - command_width=command_width, - meta_width=meta_width, - gap_width=gap_width, - is_current=index == 0, - match_prefix_len=match_prefix_len, - ) - ) - cursor_y = 1 if shown else 0 - hidden = len(completions) - shown - else: - selected_meta_lines = self._selected_meta_lines( - completions[selected_index].display_meta_text, - meta_width, - ) - start, end = self._visible_window_bounds( - completion_count=len(completions), - selected_index=selected_index, - available_rows=item_rows, - selected_item_height=len(selected_meta_lines), - ) - for index in range(start, end + 1): - completion = completions[index] - if index == selected_index: - cursor_y = len(rendered_lines) - rendered_lines.extend( - self._render_selected_item_lines( - width=width, - completion=completion, - marker_width=marker_width, - command_width=command_width, - meta_width=meta_width, - gap_width=gap_width, - meta_lines=selected_meta_lines, - match_prefix_len=match_prefix_len, - ) - ) - continue - rendered_lines.append( - self._render_single_line_item( - width=width, - completion=completion, - marker_width=marker_width, - command_width=command_width, - meta_width=meta_width, - gap_width=gap_width, - is_current=False, - match_prefix_len=match_prefix_len, - ) - ) - hidden = len(completions) - (end - start + 1) - - rendered_lines.append(self._blank_line()) - rendered_lines.append( - self._render_footer_line( - width=width, marker_width=marker_width, hidden_count=max(0, hidden) - ) - ) - return UIContent( - get_line=lambda i: rendered_lines[i], - line_count=len(rendered_lines), - cursor_position=Point(x=0, y=cursor_y), - ) - - def _blank_line(self) -> FormattedText: - return FormattedText([("class:slash-completion-menu", "")]) - - def _render_footer_line( - self, *, width: int, marker_width: int, hidden_count: int - ) -> FormattedText: - # Persistent navigation legend, rendered in the dim meta style and aligned - # under the command column. When the list scrolled, the count of hidden - # entries leads so it survives truncation on narrow terminals. - indent = self._left_padding() + marker_width - text = self._FOOTER_LEGEND - if hidden_count > 0: - text = f"+{hidden_count} more · {text}" - body = _truncate_to_width(text, max(0, width - indent)) - trailing = max(0, width - indent - get_cwidth(body)) - fragments: FormattedText = FormattedText() - fragments.append(("class:slash-completion-menu", " " * indent)) - fragments.append(("class:slash-completion-menu.meta", body)) - fragments.append(("class:slash-completion-menu", " " * trailing)) - return fragments - - def _match_prefix_len(self, app: Any) -> int: - document = getattr(getattr(app, "current_buffer", None), "document", None) - if not isinstance(document, Document): - return 0 - context = parse_completion_context(document, allow_file=False) - if context.kind is not CompletionKind.SLASH_COMMAND: - return 0 - return len(context.token[1:]) - - def _selected_meta_lines(self, text: str, meta_width: int) -> list[str]: - lines = _wrap_to_width( - text, - meta_width, - max_lines=self._MAX_EXPANDED_META_LINES, - ) - return lines or [""] - - def _visible_window_bounds( - self, - *, - completion_count: int, - selected_index: int, - available_rows: int, - selected_item_height: int, - ) -> tuple[int, int]: - selected_item_height = min(selected_item_height, available_rows) - remaining_rows = max(0, available_rows - selected_item_height) - - before = min(self._scroll_offset, selected_index, remaining_rows) - remaining_rows -= before - after = min(completion_count - selected_index - 1, remaining_rows) - remaining_rows -= after - - extra_before = min(selected_index - before, remaining_rows) - before += extra_before - remaining_rows -= extra_before - - extra_after = min(completion_count - selected_index - 1 - after, remaining_rows) - after += extra_after - - return selected_index - before, selected_index + after - - def _command_column_width( - self, - completions: Sequence[Completion], - menu_width: int, - marker_width: int, - ) -> int: - if menu_width <= 0: - return 0 - longest = max((get_cwidth(c.display_text) for c in completions), default=0) - preferred = longest + 2 - usable_width = max(0, menu_width - marker_width) - minimum = min(usable_width, 18) - maximum = max(minimum, min(28, usable_width // 2)) - return max(minimum, min(preferred, maximum)) - - def _render_command_text( - self, - text: str, - *, - width: int, - base_style: str, - is_current: bool, - match_prefix_len: int, - ) -> FormattedText: - display = _truncate_to_width(text, width) - if match_prefix_len <= 0: - return FormattedText([(base_style, display)]) - - # Match highlighting for the slash popup: the leading slash stays in the - # normal command style; the typed command prefix is emphasized. - match_end = min(len(text), 1 + match_prefix_len) - match_style = ( - "class:slash-completion-menu.command.match.current" - if is_current - else "class:slash-completion-menu.command.match" - ) - fragments: FormattedText = FormattedText() - for index, ch in enumerate(display): - style = match_style if 0 < index < match_end and index < len(text) else base_style - fragments.append((style, ch)) - return fragments - - def _render_single_line_item( - self, - *, - width: int, - completion: Completion, - marker_width: int, - command_width: int, - meta_width: int, - gap_width: int, - is_current: bool, - match_prefix_len: int, - ) -> FormattedText: - padding_width = max(0, width - marker_width - command_width - meta_width - gap_width) - left_padding = min(self._left_padding(), padding_width) - trailing_width = max( - 0, - width - left_padding - marker_width - command_width - gap_width - meta_width, - ) - - command_style = ( - "class:slash-completion-menu.command.current" - if is_current - else "class:slash-completion-menu.command" - ) - meta_style = ( - "class:slash-completion-menu.meta.current" - if is_current - else "class:slash-completion-menu.meta" - ) - marker_style = ( - "class:slash-completion-menu.marker.current" - if is_current - else "class:slash-completion-menu.marker" - ) - marker = f"{TRANSCRIPT_PROMPT_MARKER} " if is_current else " " - - # When a row is selected, use the row.current background for the - # gap and trailing padding so the highlight reads as a contiguous bar - # rather than a fragmented set of pieces. - gap_style = ( - "class:slash-completion-menu.row.current" - if is_current - else "class:slash-completion-menu" - ) - fragments: FormattedText = FormattedText() - fragments.append(("class:slash-completion-menu", " " * left_padding)) - fragments.append((marker_style, marker.ljust(marker_width))) - fragments.extend( - self._render_command_text( - completion.display_text, - width=command_width, - base_style=command_style, - is_current=is_current, - match_prefix_len=match_prefix_len, - ) - ) - fragments.append((gap_style, " " * gap_width)) - fragments.append((meta_style, _truncate_to_width(completion.display_meta_text, meta_width))) - fragments.append((gap_style, " " * trailing_width)) - return fragments - - def _render_selected_item_lines( - self, - *, - width: int, - completion: Completion, - marker_width: int, - command_width: int, - meta_width: int, - gap_width: int, - meta_lines: Sequence[str], - match_prefix_len: int, - ) -> list[FormattedText]: - lines = [ - self._render_single_line_item( - width=width, - completion=Completion( - text=completion.text, - start_position=completion.start_position, - display=completion.display, - display_meta=meta_lines[0], - ), - marker_width=marker_width, - command_width=command_width, - meta_width=meta_width, - gap_width=gap_width, - is_current=True, - match_prefix_len=match_prefix_len, - ) - ] - - continuation_prefix = ( - " " * self._left_padding() + " " * marker_width + " " * command_width + " " * gap_width - ) - continuation_trailing = max( - 0, - width - get_cwidth(continuation_prefix) - meta_width, - ) - for meta_line in meta_lines[1:]: - fragments: FormattedText = FormattedText() - fragments.append(("class:slash-completion-menu", continuation_prefix)) - fragments.append( - ( - "class:slash-completion-menu.meta.current", - _truncate_to_width(meta_line, meta_width), - ) - ) - fragments.append(("class:slash-completion-menu", " " * continuation_trailing)) - lines.append(fragments) - - return lines - - -class LocalFileMentionMenuControl(UIControl): - """Render `@` file completions as a clean inline, two-column menu.""" - - _MIN_DETAIL_WIDTH = 16 - _MAX_NAME_WIDTH = 32 - - def __init__( - self, - *, - left_padding: Callable[[], int], - scroll_offset: int = 1, - ) -> None: - self._left_padding = left_padding - self._scroll_offset = scroll_offset - - def has_focus(self) -> bool: - return False - - def preferred_width(self, max_available_width: int) -> int | None: - return max_available_width - - def preferred_height( - self, - width: int, - max_available_height: int, - wrap_lines: bool, - get_line_prefix: Callable[..., AnyFormattedText] | None, - ) -> int | None: - app = get_app_or_none() - complete_state = ( - getattr(app.current_buffer, "complete_state", None) if app is not None else None - ) - if complete_state is None or not complete_state.completions: - return 0 - # Reserve the final row for the position counter. - return min(max_available_height, len(complete_state.completions) + 1) - - def create_content(self, width: int, height: int) -> UIContent: - app = get_app_or_none() - complete_state = ( - getattr(app.current_buffer, "complete_state", None) if app is not None else None - ) - if complete_state is None or not complete_state.completions or height <= 0: - return UIContent() - - completions = complete_state.completions - selected_index = complete_state.complete_index or 0 - selected_index = max(0, min(selected_index, len(completions) - 1)) - show_count = height > 1 - item_rows = max(1, height - (1 if show_count else 0)) - start, end = self._visible_window_bounds( - completion_count=len(completions), - selected_index=selected_index, - available_rows=item_rows, - ) - - menu_width = max(0, width - self._left_padding()) - marker_width = 2 - gap_width = 4 if menu_width >= 48 else 2 - detail_enabled = menu_width >= marker_width + gap_width + self._MIN_DETAIL_WIDTH + 12 - if detail_enabled: - name_width = min( - self._MAX_NAME_WIDTH, - max(12, (menu_width - marker_width - gap_width) // 2), - ) - detail_width = max(0, menu_width - marker_width - name_width - gap_width) - else: - name_width = max(0, menu_width - marker_width) - detail_width = 0 - gap_width = 0 - - rendered_lines: list[FormattedText] = [] - selected_line_index = 0 - for index in range(start, end + 1): - if index == selected_index: - selected_line_index = len(rendered_lines) - rendered_lines.append( - self._render_item_line( - width=width, - completion=completions[index], - is_current=index == selected_index, - marker_width=marker_width, - name_width=name_width, - gap_width=gap_width, - detail_width=detail_width, - ) - ) - - if show_count: - rendered_lines.append( - self._render_count_line( - width=width, - selected_index=selected_index, - total=len(completions), - marker_width=marker_width, - ) - ) - - return UIContent( - get_line=lambda i: rendered_lines[i], - line_count=len(rendered_lines), - cursor_position=Point(x=0, y=selected_line_index), - ) - - def _visible_window_bounds( - self, - *, - completion_count: int, - selected_index: int, - available_rows: int, - ) -> tuple[int, int]: - visible_rows = min(completion_count, max(1, available_rows)) - max_start = max(0, completion_count - visible_rows) - start = min(max(0, selected_index - self._scroll_offset), max_start) - return start, start + visible_rows - 1 - - def _render_item_line( - self, - *, - width: int, - completion: Completion, - is_current: bool, - marker_width: int, - name_width: int, - gap_width: int, - detail_width: int, - ) -> FormattedText: - left_padding = min(self._left_padding(), width) - name = completion.display_text or completion.text - detail = (completion.text or name).rstrip("/") - marker = "→ " if is_current else " " - marker_style = ( - "class:file-completion-menu.marker.current" - if is_current - else "class:file-completion-menu.marker" - ) - name_style = ( - "class:file-completion-menu.name.current" - if is_current - else "class:file-completion-menu.name" - ) - detail_style = ( - "class:file-completion-menu.detail.current" - if is_current - else "class:file-completion-menu.detail" - ) - - fragments: FormattedText = FormattedText() - fragments.append(("class:file-completion-menu", " " * left_padding)) - fragments.append((marker_style, marker.ljust(marker_width))) - fragments.append((name_style, _truncate_to_width(name, name_width))) - if detail_width > 0: - fragments.append(("class:file-completion-menu", " " * gap_width)) - fragments.append((detail_style, _truncate_to_width(detail, detail_width))) - used_width = left_padding + marker_width + name_width + gap_width + detail_width - if used_width < width: - fragments.append(("class:file-completion-menu", " " * (width - used_width))) - return fragments - - def _render_count_line( - self, - *, - width: int, - selected_index: int, - total: int, - marker_width: int, - ) -> FormattedText: - left_padding = min(self._left_padding() + marker_width, width) - label = f"({selected_index + 1}/{total})" - fragments: FormattedText = FormattedText() - fragments.append(("class:file-completion-menu", " " * left_padding)) - count_text = _truncate_to_width(label, max(0, width - left_padding)) - fragments.append(("class:file-completion-menu.count", count_text)) - return fragments - - LocalFileMentionCompleter = HostFileMentionCompleter @@ -1313,12 +675,6 @@ def _truncate_right(text: str, max_cols: int) -> str: return "".join(chars) + ellipsis -@dataclass(frozen=True, slots=True) -class BgTaskCounts: - bash: int = 0 - agent: int = 0 - - @runtime_checkable class AgentStatusProvider(Protocol): """Optional protocol for delegates that render always-visible agent status. @@ -1420,7 +776,32 @@ def __init__( resolve_segments, ) - _statusline_cfg = statusline_config or StatusLineConfig() + prompt_config = PromptConfig( + model_capabilities=model_capabilities, + model_name=model_name, + thinking=thinking, + thinking_effort=thinking_effort, + agent_mode_slash_commands=agent_mode_slash_commands, + shell_mode_slash_commands=shell_mode_slash_commands, + history_enabled=history_enabled, + statusline_config=statusline_config, + sticky_input=sticky_input, + ) + providers = PromptProviders( + status_provider=status_provider, + status_block_provider=status_block_provider, + fast_refresh_provider=fast_refresh_provider, + background_task_count_provider=background_task_count_provider, + update_notice_provider=update_notice_provider, + editor_command_provider=editor_command_provider, + turn_recaps_provider=turn_recaps_provider, + plan_mode_toggle_callback=plan_mode_toggle_callback, + thinking_effort_cycle_callback=thinking_effort_cycle_callback, + ) + self._prompt_config = prompt_config + self._prompt_providers = providers + + _statusline_cfg = prompt_config.statusline_config or StatusLineConfig() self._statusline_layout = resolve_segments(_statusline_cfg) self._statusline_runner: StatusLineCommandRunner | None = None self._lifecycle = PromptLifecycle() @@ -1442,7 +823,7 @@ def __init__( str(HostPath.cwd()).encode(encoding="utf-8"), usedforsecurity=False ).hexdigest() self._history_file = (history_dir / work_dir_id).with_suffix(".jsonl") - self._history_enabled = history_enabled and not _env_truthy( + self._history_enabled = prompt_config.history_enabled and not _env_truthy( "PYTHINKER_DISABLE_PROMPT_HISTORY" ) if self._history_enabled: @@ -1465,23 +846,25 @@ def __init__( self._git_status_token: Token[GitStatusIndex | None] | None = None self._toast_token: Token[ToastManager | None] | None = None self._clipboard_token: Token[ClipboardAdapter | None] | None = None - self._status_provider = status_provider - self._status_block_provider = status_block_provider - self._fast_refresh_provider = fast_refresh_provider - self._background_task_count_provider = background_task_count_provider - self._update_notice_provider = update_notice_provider + self._status_provider = providers.status_provider + self._status_block_provider = providers.status_block_provider + self._fast_refresh_provider = providers.fast_refresh_provider + self._background_task_count_provider = providers.background_task_count_provider + self._update_notice_provider = providers.update_notice_provider self._prompt_frame_update_notice: str | None = None self._prompt_footer_row_budget = 0 - self._editor_command_provider = editor_command_provider - self._turn_recaps_provider = turn_recaps_provider - self._plan_mode_toggle_callback = plan_mode_toggle_callback - self._thinking_effort_cycle_callback = thinking_effort_cycle_callback - self._model_capabilities = model_capabilities - self._model_name = model_name + self._editor_command_provider = providers.editor_command_provider + self._turn_recaps_provider = providers.turn_recaps_provider + self._plan_mode_toggle_callback = providers.plan_mode_toggle_callback + self._thinking_effort_cycle_callback = providers.thinking_effort_cycle_callback + self._model_capabilities = prompt_config.model_capabilities + self._model_name = prompt_config.model_name self._last_history_content: str | None = None self._mode: PromptMode = PromptMode.AGENT - self._thinking = thinking - self._thinking_effort = thinking_effort or ("high" if thinking else "off") + self._thinking = prompt_config.thinking + self._thinking_effort = prompt_config.thinking_effort or ( + "high" if prompt_config.thinking else "off" + ) self._placeholder_manager = PromptPlaceholderManager() # Keep the old attribute for test compatibility and for any external imports. self._attachment_cache = self._placeholder_manager.attachment_cache @@ -1499,7 +882,7 @@ def __init__( # fossilizes above the stream. Cleared on attach/detach. See # _input_card_hidden_pre_stream. self._turn_starting: bool = False - self._sticky_input = sticky_input + self._sticky_input = prompt_config.sticky_input self._latest_todos: tuple[TodoDisplayItem, ...] = () self._modal_delegates: list[RunningPromptDelegate] = [] self._shortcut_help_open = False @@ -1578,349 +961,11 @@ def __init__( arg_suggestions=self._slash_arg_suggestions, ) - # Build key bindings - _kb = KeyBindings() - - def _accept_completion(buff: Buffer) -> None: - """Accept the current or first completion, suppressing re-completion.""" - state = buff.complete_state - if state is None: - return - completion = state.current_completion - if completion is None: - if not state.completions: - return - completion = state.completions[0] - self._suppress_auto_completion = True - try: - buff.apply_completion(completion) - finally: - self._suppress_auto_completion = False - - def _is_slash_completion() -> bool: - """True when the active completion menu is for a slash command.""" - buff = self._session.default_buffer - return bool( - buff.complete_state - and buff.complete_state.completions - and self._slash_completion_active(buff.document) - ) - - _slash_completion_filter = has_completions & Condition(_is_slash_completion) - _non_slash_completion_filter = has_completions & ~Condition(_is_slash_completion) - - @_kb.add("enter", filter=_slash_completion_filter) - def _(event: KeyPressEvent) -> None: - """Slash command completion: accept and submit in one step.""" - _accept_completion(event.current_buffer) - event.current_buffer.validate_and_handle() - - @_kb.add("escape", eager=True, filter=_slash_completion_filter) - def _(event: KeyPressEvent) -> None: - """Slash completion: Escape discards a draft command, or dismisses - the argument menu when there is no draft command to remove.""" - buffer = event.current_buffer - if not _discard_slash_command(buffer): - # Slash-argument completion (e.g. "/model gpt"): the eager - # binding swallowed Escape but there is no root command to - # strip, so dismiss the completion menu explicitly instead of - # leaving it open. - buffer.cancel_completion() - event.app.invalidate() - - @_kb.add("enter", filter=_non_slash_completion_filter) - def _(event: KeyPressEvent) -> None: - """Non-slash completion (file mentions, etc.): accept only.""" - _accept_completion(event.current_buffer) - - def _has_slash_suggestion() -> bool: - buff = self._session.default_buffer - return bool(buff.suggestion and buff.suggestion.text) - - @_kb.add("tab", filter=Condition(_has_slash_suggestion)) - def _(event: KeyPressEvent) -> None: - """Slash command ghost suggestion: Tab completes the word inline.""" - suggestion = event.current_buffer.suggestion - if suggestion and suggestion.text: - event.current_buffer.insert_text(suggestion.text) - - @_kb.add("?", eager=True) - def _(event: KeyPressEvent) -> None: - """Toggle a compact shortcuts popup when the input row is empty.""" - if self._active_prompt_delegate() is not None: - event.current_buffer.insert_text("?") - return - if event.current_buffer.text.strip(): - event.current_buffer.insert_text("?") - return - self.toggle_shortcut_help() - - @_kb.add("c-x", eager=True) - def _(event: KeyPressEvent) -> None: - if self._active_prompt_delegate() is not None: - return - self.toggle_mode() - from pythinker_code.telemetry import track - - track("shortcut_mode_switch", to_mode=self._mode.value) - - @_kb.add("s-tab", eager=True) - def _(event: KeyPressEvent) -> None: - """Cycle thinking effort with Shift+Tab.""" - if self._active_prompt_delegate() is not None: - return - if self._thinking_effort_cycle_callback is not None: - - async def _cycle() -> None: - assert self._thinking_effort_cycle_callback is not None - new_level = await self._thinking_effort_cycle_callback() - from pythinker_code.telemetry import track - - if new_level is None: - message = ( - "Current model uses native reasoning" - if self._uses_native_thinking() - else "Current model does not support thinking" - ) - toast( - message, - topic="thinking_level", - duration=3.0, - immediate=True, - ) - else: - self._thinking_effort = new_level - self._thinking = new_level != "off" - track("shortcut_thinking_cycle", level=new_level) - toast( - f"Thinking level: {new_level}", - topic="thinking_level", - duration=3.0, - immediate=True, - ) - event.app.invalidate() - - event.app.create_background_task(_cycle()) - event.app.invalidate() - - @_kb.add("escape", "enter", eager=True) - @_kb.add("c-j", eager=True) - def _(event: KeyPressEvent) -> None: - """Insert a newline when Alt-Enter or Ctrl-J is pressed.""" - from pythinker_code.telemetry import track - - track("shortcut_newline") - event.current_buffer.insert_text("\n") - - @_kb.add("c-o", eager=True) - def _(event: KeyPressEvent) -> None: - """Expand active transcript content, or open current buffer in external editor.""" - if self._active_prompt_delegate() is not None: - if self._should_handle_running_prompt_key("c-o"): - self._handle_running_prompt_key("c-o", event) - return - - from pythinker_code.telemetry import track - - track("shortcut_editor") - self._open_in_external_editor(event) - - @_kb.add("c-l", eager=True) - def _(event: KeyPressEvent) -> None: - """Erase and fully repaint the screen (recovery from console damage).""" - self._hard_repaint(event) - - def _has_staged_suggestion_prefill() -> bool: - return bool(getattr(self, "_staged_suggestion_prefill", None)) - - @_kb.add("escape", "s", eager=True, filter=Condition(_has_staged_suggestion_prefill)) - def _(event: KeyPressEvent) -> None: - """Accept the latest agent suggestion into the prompt buffer.""" - if self.accept_staged_suggestion_prefill(): - from pythinker_code.telemetry import track - - track("suggestion_accepted") - event.app.invalidate() - - @_kb.add( - "up", - eager=True, - filter=Condition(lambda: self._should_handle_running_prompt_key("up")), - ) - def _(event: KeyPressEvent) -> None: - self._handle_running_prompt_key("up", event) - - @_kb.add( - "down", - eager=True, - filter=Condition(lambda: self._should_handle_running_prompt_key("down")), - ) - def _(event: KeyPressEvent) -> None: - self._handle_running_prompt_key("down", event) - - @_kb.add( - "left", - eager=True, - filter=Condition(lambda: self._should_handle_running_prompt_key("left")), - ) - def _(event: KeyPressEvent) -> None: - self._handle_running_prompt_key("left", event) - - @_kb.add( - "right", - eager=True, - filter=Condition(lambda: self._should_handle_running_prompt_key("right")), - ) - def _(event: KeyPressEvent) -> None: - self._handle_running_prompt_key("right", event) - - @_kb.add( - "tab", - eager=True, - filter=Condition(lambda: self._should_handle_running_prompt_key("tab")), - ) - def _(event: KeyPressEvent) -> None: - self._handle_running_prompt_key("tab", event) - - @_kb.add( - "enter", - eager=True, - filter=Condition(lambda: self._should_handle_running_prompt_key("enter")), - ) - def _(event: KeyPressEvent) -> None: - self._handle_running_prompt_key("enter", event) - - @_kb.add( - "space", - eager=True, - filter=Condition(lambda: self._should_handle_running_prompt_key("space")), - ) - def _(event: KeyPressEvent) -> None: - self._handle_running_prompt_key("space", event) - - @_kb.add( - "c-s", - eager=True, - filter=Condition(lambda: self._should_handle_running_prompt_key("c-s")), - ) - def _(event: KeyPressEvent) -> None: - self._handle_running_prompt_key("c-s", event) - - @_kb.add( - "c-e", - eager=True, - filter=Condition(lambda: self._should_handle_running_prompt_key("c-e")), - ) - def _(event: KeyPressEvent) -> None: - self._handle_running_prompt_key("c-e", event) - - @_kb.add( - "c-t", - eager=True, - filter=Condition(lambda: self._should_handle_running_prompt_key("c-t")), - ) - def _(event: KeyPressEvent) -> None: - self._handle_running_prompt_key("c-t", event) - - @_kb.add( - "c-c", - eager=True, - filter=Condition(lambda: self._should_handle_running_prompt_key("c-c")), - ) - def _(event: KeyPressEvent) -> None: - self._handle_running_prompt_key("c-c", event) - - @_kb.add( - "c-d", - eager=True, - filter=Condition(lambda: self._should_handle_running_prompt_key("c-d")), - ) - def _(event: KeyPressEvent) -> None: - self._handle_running_prompt_key("c-d", event) - - @_kb.add( - "escape", - eager=True, - filter=Condition(lambda: self._should_handle_running_prompt_key("escape")), - ) - def _(event: KeyPressEvent) -> None: - self._handle_running_prompt_key("escape", event) - - @_kb.add( - "escape", - eager=True, - filter=Condition(lambda: self._shortcut_help_open), - ) - def _(event: KeyPressEvent) -> None: - self.close_shortcut_help() - - @_kb.add( - "1", - eager=True, - filter=Condition(lambda: self._should_handle_running_prompt_key("1")), - ) - def _(event: KeyPressEvent) -> None: - self._handle_running_prompt_key("1", event) - - @_kb.add( - "2", - eager=True, - filter=Condition(lambda: self._should_handle_running_prompt_key("2")), - ) - def _(event: KeyPressEvent) -> None: - self._handle_running_prompt_key("2", event) - - @_kb.add( - "3", - eager=True, - filter=Condition(lambda: self._should_handle_running_prompt_key("3")), - ) - def _(event: KeyPressEvent) -> None: - self._handle_running_prompt_key("3", event) - - @_kb.add( - "4", - eager=True, - filter=Condition(lambda: self._should_handle_running_prompt_key("4")), - ) - def _(event: KeyPressEvent) -> None: - self._handle_running_prompt_key("4", event) - - @_kb.add( - "5", - eager=True, - filter=Condition(lambda: self._should_handle_running_prompt_key("5")), - ) - def _(event: KeyPressEvent) -> None: - self._handle_running_prompt_key("5", event) - - @_kb.add( - "6", - eager=True, - filter=Condition(lambda: self._should_handle_running_prompt_key("6")), - ) - def _(event: KeyPressEvent) -> None: - self._handle_running_prompt_key("6", event) - - @_kb.add(Keys.BracketedPaste, eager=True) - def _(event: KeyPressEvent) -> None: - self._handle_bracketed_paste(event) - - if clipboard_available or media_clipboard_available: - - @_kb.add("c-v", eager=True) - def _(event: KeyPressEvent) -> None: - from pythinker_code.telemetry import track - - track("shortcut_paste") - if self._try_paste_media(event): - return - if clipboard_available: - clipboard_text = self._clipboard_adapter.paste_text(event.app.clipboard) - if clipboard_text is None: - return - self._insert_pasted_text(event.current_buffer, clipboard_text) - event.app.invalidate() + # Build key bindings (moved to prompting.keybindings; the session + # implements PromptController implicitly and passes itself). + self._clipboard_text_available = clipboard_available + self._media_clipboard_available = media_clipboard_available + _kb = build_prompt_key_bindings(self) # Only use PyperclipClipboard when pyperclip actually works. # PromptSession built-in keybindings (ctrl-k, ctrl-w, ctrl-y) @@ -3464,6 +2509,32 @@ def running_prompt_accepts_submission(self) -> bool: return False return delegate.running_prompt_accepts_submission() + def input_text(self) -> str: + """Current text of the input buffer ("" when the session is not built yet).""" + buffer = getattr(getattr(self, "_session", None), "default_buffer", None) + return buffer.text if buffer is not None else "" + + def input_state(self) -> tuple[str, int]: + """Current input buffer text and cursor position.""" + buffer = getattr(getattr(self, "_session", None), "default_buffer", None) + if buffer is None: + return "", 0 + return buffer.text, buffer.cursor_position + + def clear_input(self) -> None: + """Clear the input buffer when it holds text.""" + buffer = getattr(getattr(self, "_session", None), "default_buffer", None) + if buffer is not None and buffer.text: + buffer.set_document(Document(), bypass_readonly=True) + + def build_user_input(self, command: str) -> UserInput: + """Resolve *command* (placeholders, mode) into a submission-ready ``UserInput``.""" + return self._build_user_input(command) + + def serialize_for_history(self, command: str) -> str: + """Expand placeholder tokens in *command* into their history/audit form.""" + return self._get_placeholder_manager().serialize_for_history(command) + def _on_workspace_snapshot_published(self) -> None: """Re-run file completion when a fresh workspace snapshot lands mid-menu.""" app = self._session.app diff --git a/src/pythinker_code/ui/shell/prompting/__init__.py b/src/pythinker_code/ui/shell/prompting/__init__.py index 8be50b2d..17573cb6 100644 --- a/src/pythinker_code/ui/shell/prompting/__init__.py +++ b/src/pythinker_code/ui/shell/prompting/__init__.py @@ -1,6 +1,11 @@ """Deep prompt rendering and session-resource modules.""" from pythinker_code.ui.shell.prompting.clipboard import ClipboardAdapter +from pythinker_code.ui.shell.prompting.config import ( + BgTaskCounts, + PromptConfig, + PromptProviders, +) from pythinker_code.ui.shell.prompting.footer import ( FooterContent, FooterViewModel, @@ -21,6 +26,10 @@ PromptHistoryStatus, PromptHistoryStore, ) +from pythinker_code.ui.shell.prompting.keybindings import ( + PromptController, + build_prompt_key_bindings, +) from pythinker_code.ui.shell.prompting.renderer import ( PromptSceneAllocation, PromptSceneBudget, @@ -29,12 +38,16 @@ from pythinker_code.ui.shell.prompting.toasts import ToastManager, ToastSnapshot __all__ = ( + "BgTaskCounts", "ClipboardAdapter", "FrozenFragments", "FooterContent", "FooterViewModel", "GitSnapshot", "GitStatusIndex", + "PromptConfig", + "PromptController", + "PromptProviders", "PromptFrame", "PromptFrameCollector", "PromptHistoryError", @@ -46,6 +59,7 @@ "ToastSnapshot", "allocate_prompt_scene_rows", "background_task_summary", + "build_prompt_key_bindings", "freeze_fragments", "select_footer_content", "truncate_footer_left", diff --git a/src/pythinker_code/ui/shell/prompting/completion/__init__.py b/src/pythinker_code/ui/shell/prompting/completion/__init__.py index 4f71153c..f70938e4 100644 --- a/src/pythinker_code/ui/shell/prompting/completion/__init__.py +++ b/src/pythinker_code/ui/shell/prompting/completion/__init__.py @@ -5,6 +5,12 @@ CompletionKind, parse_completion_context, ) +from pythinker_code.ui.shell.prompting.completion.menus import ( + LocalFileMentionMenuControl, + SlashCommandMenuControl, + truncate_to_width, + wrap_to_width, +) from pythinker_code.ui.shell.prompting.completion.slash import ( InputHighlightLexer, SlashCommandAutoSuggest, @@ -24,10 +30,14 @@ "HostFileMentionCompleter", "InputHighlightLexer", "LocalFileMentionCompleter", + "LocalFileMentionMenuControl", "SlashCommandAutoSuggest", + "SlashCommandMenuControl", "SlashCommandCompleter", "WorkspaceEntry", "WorkspaceIndex", "WorkspaceSnapshot", "parse_completion_context", + "truncate_to_width", + "wrap_to_width", ) diff --git a/src/pythinker_code/ui/shell/prompting/completion/menus.py b/src/pythinker_code/ui/shell/prompting/completion/menus.py new file mode 100644 index 00000000..70d0b222 --- /dev/null +++ b/src/pythinker_code/ui/shell/prompting/completion/menus.py @@ -0,0 +1,687 @@ +"""Completion-menu rendering controls for the prompt (slash and file mentions). + +The prompt_toolkit *layout installation* for these controls stays in +``pythinker_code.ui.shell.prompt`` because the facade owns the +``PromptSession`` object; this module owns only the rendering. +""" + +from __future__ import annotations + +from collections.abc import Callable, Sequence +from typing import Any + +from prompt_toolkit.completion import Completion +from prompt_toolkit.data_structures import Point +from prompt_toolkit.document import Document +from prompt_toolkit.formatted_text import AnyFormattedText, FormattedText +from prompt_toolkit.layout.controls import UIContent, UIControl +from prompt_toolkit.utils import get_cwidth + +from pythinker_code.ui.shell.glyphs import TRANSCRIPT_PROMPT_MARKER +from pythinker_code.ui.shell.prompting.completion.context import ( + CompletionKind, + parse_completion_context, +) + + +def _resolve_app() -> Any: + """Resolve the active application through the prompt facade module. + + Repository tests monkeypatch ``pythinker_code.ui.shell.prompt.get_app_or_none`` + to drive these controls; routing the lookup through that module keeps the + monkeypatch surface intact after the rendering moved here. + """ + from pythinker_code.ui.shell import prompt as shell_prompt + + return shell_prompt.get_app_or_none() + + +def truncate_to_width(text: str, width: int) -> str: + if width <= 0: + return "" + + total = 0 + chars: list[str] = [] + for ch in text: + ch_width = get_cwidth(ch) + if total + ch_width > width: + break + chars.append(ch) + total += ch_width + + if total == get_cwidth(text): + return text + (" " * max(0, width - total)) + + ellipsis = "..." + ellipsis_width = get_cwidth(ellipsis) + if width <= ellipsis_width: + return "." * width + + available = width - ellipsis_width + total = 0 + chars = [] + for ch in text: + ch_width = get_cwidth(ch) + if total + ch_width > available: + break + chars.append(ch) + total += ch_width + return "".join(chars) + ellipsis + (" " * max(0, width - total - ellipsis_width)) + + +def wrap_to_width(text: str, width: int, *, max_lines: int | None = None) -> list[str]: + if width <= 0: + return [] + + words = text.split() + if not words: + return [""] + + lines: list[str] = [] + current_words: list[str] = [] + current_width = 0 + index = 0 + + while index < len(words): + word = words[index] + word_width = get_cwidth(word) + separator_width = 1 if current_words else 0 + + if current_words and current_width + separator_width + word_width <= width: + current_words.append(word) + current_width += separator_width + word_width + index += 1 + continue + + if not current_words and word_width <= width: + current_words.append(word) + current_width = word_width + index += 1 + continue + + if not current_words and word_width > width: + current_words.append(truncate_to_width(word, width).rstrip()) + current_width = get_cwidth(current_words[0]) + index += 1 + + lines.append(" ".join(current_words)) + current_words = [] + current_width = 0 + + if max_lines is not None and len(lines) == max_lines: + remaining = " ".join(words[index:]) + if remaining: + prefix = f"{lines[-1]} " if lines[-1] else "" + lines[-1] = truncate_to_width(prefix + remaining, width).rstrip() + return lines + + if current_words: + line = " ".join(current_words) + if max_lines is not None and len(lines) + 1 > max_lines: + if lines: + lines[-1] = truncate_to_width(f"{lines[-1]} {line}", width).rstrip() + else: + lines.append(truncate_to_width(line, width).rstrip()) + else: + lines.append(line) + + return lines + + +class SlashCommandMenuControl(UIControl): + """Render slash command completions as a full-width menu that matches the shell UI.""" + + _MAX_EXPANDED_META_LINES = 3 + # One blank line is reserved above the list as breathing room from the input + # row, so the menu reads as its own region rather than crowding what's typed. + _GAP_LINES = 1 + # A persistent footer block at the bottom: a blank separator line plus the + # navigation legend (which folds in the overflow count when the list scrolls). + # The separator gives the legend the same breathing room as the top gap. + _FOOTER_LINES = 2 + _FOOTER_LEGEND = "Enter to select · ↑/↓ to navigate · Esc to cancel" + + def __init__( + self, + *, + left_padding: Callable[[], int], + scroll_offset: int = 1, + ) -> None: + self._left_padding = left_padding + self._scroll_offset = scroll_offset + + def has_focus(self) -> bool: + return False + + def preferred_width(self, max_available_width: int) -> int | None: + return max_available_width + + def preferred_height( + self, + width: int, + max_available_height: int, + wrap_lines: bool, + get_line_prefix: Callable[..., AnyFormattedText] | None, + ) -> int | None: + app = _resolve_app() + complete_state = ( + getattr(app.current_buffer, "complete_state", None) if app is not None else None + ) + if complete_state is None: + return 0 + completions = complete_state.completions + if not completions: + return 0 + selected_index = complete_state.complete_index + if selected_index is None: + content_height = len(completions) + else: + menu_width = max(0, width - self._left_padding()) + marker_width = 2 + command_width = self._command_column_width(completions, menu_width, marker_width) + gap_width = 3 if menu_width > command_width + 6 else 1 + meta_width = max(0, menu_width - marker_width - command_width - gap_width) + selected_meta_lines = self._selected_meta_lines( + completions[selected_index].display_meta_text, + meta_width, + ) + content_height = (len(completions) - 1) + len(selected_meta_lines) + # Reserve the gap line above the list and the footer line below it. When + # the list is taller than the space the window allows, the window caps the + # height and create_content lays the list out within whatever rows remain. + chrome = self._GAP_LINES + self._FOOTER_LINES + return min(max_available_height, content_height + chrome) + + def create_content(self, width: int, height: int) -> UIContent: + app = _resolve_app() + complete_state = ( + getattr(app.current_buffer, "complete_state", None) if app is not None else None + ) + if complete_state is None or not complete_state.completions: + return UIContent() + + completions = complete_state.completions + selected_index = complete_state.complete_index + match_prefix_len = self._match_prefix_len(app) + + menu_width = max(0, width - self._left_padding()) + marker_width = 2 + command_width = self._command_column_width(completions, menu_width, marker_width) + gap_width = 3 if menu_width > command_width + 6 else 1 + meta_width = max(0, menu_width - marker_width - command_width - gap_width) + + total_rows = max(1, height) + # The gap line above the list and the footer line below it are always + # present, so the list itself lays out within the remaining rows. + item_rows = max(1, total_rows - self._GAP_LINES - self._FOOTER_LINES) + + rendered_lines: list[FormattedText] = [self._blank_line()] + cursor_y = 0 + + if selected_index is None: + # Pre-highlight index 0 even before the user navigates: pressing + # Enter accepts the first completion, so the visual state should + # match that behavior. Without this the menu looks ambiguous (no + # row highlighted) but Enter still commits the top row. + shown = min(len(completions), item_rows) + for index in range(shown): + rendered_lines.append( + self._render_single_line_item( + width=width, + completion=completions[index], + marker_width=marker_width, + command_width=command_width, + meta_width=meta_width, + gap_width=gap_width, + is_current=index == 0, + match_prefix_len=match_prefix_len, + ) + ) + cursor_y = 1 if shown else 0 + hidden = len(completions) - shown + else: + selected_meta_lines = self._selected_meta_lines( + completions[selected_index].display_meta_text, + meta_width, + ) + start, end = self._visible_window_bounds( + completion_count=len(completions), + selected_index=selected_index, + available_rows=item_rows, + selected_item_height=len(selected_meta_lines), + ) + for index in range(start, end + 1): + completion = completions[index] + if index == selected_index: + cursor_y = len(rendered_lines) + rendered_lines.extend( + self._render_selected_item_lines( + width=width, + completion=completion, + marker_width=marker_width, + command_width=command_width, + meta_width=meta_width, + gap_width=gap_width, + meta_lines=selected_meta_lines, + match_prefix_len=match_prefix_len, + ) + ) + continue + rendered_lines.append( + self._render_single_line_item( + width=width, + completion=completion, + marker_width=marker_width, + command_width=command_width, + meta_width=meta_width, + gap_width=gap_width, + is_current=False, + match_prefix_len=match_prefix_len, + ) + ) + hidden = len(completions) - (end - start + 1) + + rendered_lines.append(self._blank_line()) + rendered_lines.append( + self._render_footer_line( + width=width, marker_width=marker_width, hidden_count=max(0, hidden) + ) + ) + return UIContent( + get_line=lambda i: rendered_lines[i], + line_count=len(rendered_lines), + cursor_position=Point(x=0, y=cursor_y), + ) + + def _blank_line(self) -> FormattedText: + return FormattedText([("class:slash-completion-menu", "")]) + + def _render_footer_line( + self, *, width: int, marker_width: int, hidden_count: int + ) -> FormattedText: + # Persistent navigation legend, rendered in the dim meta style and aligned + # under the command column. When the list scrolled, the count of hidden + # entries leads so it survives truncation on narrow terminals. + indent = self._left_padding() + marker_width + text = self._FOOTER_LEGEND + if hidden_count > 0: + text = f"+{hidden_count} more · {text}" + body = truncate_to_width(text, max(0, width - indent)) + trailing = max(0, width - indent - get_cwidth(body)) + fragments: FormattedText = FormattedText() + fragments.append(("class:slash-completion-menu", " " * indent)) + fragments.append(("class:slash-completion-menu.meta", body)) + fragments.append(("class:slash-completion-menu", " " * trailing)) + return fragments + + def _match_prefix_len(self, app: Any) -> int: + document = getattr(getattr(app, "current_buffer", None), "document", None) + if not isinstance(document, Document): + return 0 + context = parse_completion_context(document, allow_file=False) + if context.kind is not CompletionKind.SLASH_COMMAND: + return 0 + return len(context.token[1:]) + + def _selected_meta_lines(self, text: str, meta_width: int) -> list[str]: + lines = wrap_to_width( + text, + meta_width, + max_lines=self._MAX_EXPANDED_META_LINES, + ) + return lines or [""] + + def _visible_window_bounds( + self, + *, + completion_count: int, + selected_index: int, + available_rows: int, + selected_item_height: int, + ) -> tuple[int, int]: + selected_item_height = min(selected_item_height, available_rows) + remaining_rows = max(0, available_rows - selected_item_height) + + before = min(self._scroll_offset, selected_index, remaining_rows) + remaining_rows -= before + after = min(completion_count - selected_index - 1, remaining_rows) + remaining_rows -= after + + extra_before = min(selected_index - before, remaining_rows) + before += extra_before + remaining_rows -= extra_before + + extra_after = min(completion_count - selected_index - 1 - after, remaining_rows) + after += extra_after + + return selected_index - before, selected_index + after + + def _command_column_width( + self, + completions: Sequence[Completion], + menu_width: int, + marker_width: int, + ) -> int: + if menu_width <= 0: + return 0 + longest = max((get_cwidth(c.display_text) for c in completions), default=0) + preferred = longest + 2 + usable_width = max(0, menu_width - marker_width) + minimum = min(usable_width, 18) + maximum = max(minimum, min(28, usable_width // 2)) + return max(minimum, min(preferred, maximum)) + + def _render_command_text( + self, + text: str, + *, + width: int, + base_style: str, + is_current: bool, + match_prefix_len: int, + ) -> FormattedText: + display = truncate_to_width(text, width) + if match_prefix_len <= 0: + return FormattedText([(base_style, display)]) + + # Match highlighting for the slash popup: the leading slash stays in the + # normal command style; the typed command prefix is emphasized. + match_end = min(len(text), 1 + match_prefix_len) + match_style = ( + "class:slash-completion-menu.command.match.current" + if is_current + else "class:slash-completion-menu.command.match" + ) + fragments: FormattedText = FormattedText() + for index, ch in enumerate(display): + style = match_style if 0 < index < match_end and index < len(text) else base_style + fragments.append((style, ch)) + return fragments + + def _render_single_line_item( + self, + *, + width: int, + completion: Completion, + marker_width: int, + command_width: int, + meta_width: int, + gap_width: int, + is_current: bool, + match_prefix_len: int, + ) -> FormattedText: + padding_width = max(0, width - marker_width - command_width - meta_width - gap_width) + left_padding = min(self._left_padding(), padding_width) + trailing_width = max( + 0, + width - left_padding - marker_width - command_width - gap_width - meta_width, + ) + + command_style = ( + "class:slash-completion-menu.command.current" + if is_current + else "class:slash-completion-menu.command" + ) + meta_style = ( + "class:slash-completion-menu.meta.current" + if is_current + else "class:slash-completion-menu.meta" + ) + marker_style = ( + "class:slash-completion-menu.marker.current" + if is_current + else "class:slash-completion-menu.marker" + ) + marker = f"{TRANSCRIPT_PROMPT_MARKER} " if is_current else " " + + # When a row is selected, use the row.current background for the + # gap and trailing padding so the highlight reads as a contiguous bar + # rather than a fragmented set of pieces. + gap_style = ( + "class:slash-completion-menu.row.current" + if is_current + else "class:slash-completion-menu" + ) + fragments: FormattedText = FormattedText() + fragments.append(("class:slash-completion-menu", " " * left_padding)) + fragments.append((marker_style, marker.ljust(marker_width))) + fragments.extend( + self._render_command_text( + completion.display_text, + width=command_width, + base_style=command_style, + is_current=is_current, + match_prefix_len=match_prefix_len, + ) + ) + fragments.append((gap_style, " " * gap_width)) + fragments.append((meta_style, truncate_to_width(completion.display_meta_text, meta_width))) + fragments.append((gap_style, " " * trailing_width)) + return fragments + + def _render_selected_item_lines( + self, + *, + width: int, + completion: Completion, + marker_width: int, + command_width: int, + meta_width: int, + gap_width: int, + meta_lines: Sequence[str], + match_prefix_len: int, + ) -> list[FormattedText]: + lines = [ + self._render_single_line_item( + width=width, + completion=Completion( + text=completion.text, + start_position=completion.start_position, + display=completion.display, + display_meta=meta_lines[0], + ), + marker_width=marker_width, + command_width=command_width, + meta_width=meta_width, + gap_width=gap_width, + is_current=True, + match_prefix_len=match_prefix_len, + ) + ] + + continuation_prefix = ( + " " * self._left_padding() + " " * marker_width + " " * command_width + " " * gap_width + ) + continuation_trailing = max( + 0, + width - get_cwidth(continuation_prefix) - meta_width, + ) + for meta_line in meta_lines[1:]: + fragments: FormattedText = FormattedText() + fragments.append(("class:slash-completion-menu", continuation_prefix)) + fragments.append( + ( + "class:slash-completion-menu.meta.current", + truncate_to_width(meta_line, meta_width), + ) + ) + fragments.append(("class:slash-completion-menu", " " * continuation_trailing)) + lines.append(fragments) + + return lines + + +class LocalFileMentionMenuControl(UIControl): + """Render `@` file completions as a clean inline, two-column menu.""" + + _MIN_DETAIL_WIDTH = 16 + _MAX_NAME_WIDTH = 32 + + def __init__( + self, + *, + left_padding: Callable[[], int], + scroll_offset: int = 1, + ) -> None: + self._left_padding = left_padding + self._scroll_offset = scroll_offset + + def has_focus(self) -> bool: + return False + + def preferred_width(self, max_available_width: int) -> int | None: + return max_available_width + + def preferred_height( + self, + width: int, + max_available_height: int, + wrap_lines: bool, + get_line_prefix: Callable[..., AnyFormattedText] | None, + ) -> int | None: + app = _resolve_app() + complete_state = ( + getattr(app.current_buffer, "complete_state", None) if app is not None else None + ) + if complete_state is None or not complete_state.completions: + return 0 + # Reserve the final row for the position counter. + return min(max_available_height, len(complete_state.completions) + 1) + + def create_content(self, width: int, height: int) -> UIContent: + app = _resolve_app() + complete_state = ( + getattr(app.current_buffer, "complete_state", None) if app is not None else None + ) + if complete_state is None or not complete_state.completions or height <= 0: + return UIContent() + + completions = complete_state.completions + selected_index = complete_state.complete_index or 0 + selected_index = max(0, min(selected_index, len(completions) - 1)) + show_count = height > 1 + item_rows = max(1, height - (1 if show_count else 0)) + start, end = self._visible_window_bounds( + completion_count=len(completions), + selected_index=selected_index, + available_rows=item_rows, + ) + + menu_width = max(0, width - self._left_padding()) + marker_width = 2 + gap_width = 4 if menu_width >= 48 else 2 + detail_enabled = menu_width >= marker_width + gap_width + self._MIN_DETAIL_WIDTH + 12 + if detail_enabled: + name_width = min( + self._MAX_NAME_WIDTH, + max(12, (menu_width - marker_width - gap_width) // 2), + ) + detail_width = max(0, menu_width - marker_width - name_width - gap_width) + else: + name_width = max(0, menu_width - marker_width) + detail_width = 0 + gap_width = 0 + + rendered_lines: list[FormattedText] = [] + selected_line_index = 0 + for index in range(start, end + 1): + if index == selected_index: + selected_line_index = len(rendered_lines) + rendered_lines.append( + self._render_item_line( + width=width, + completion=completions[index], + is_current=index == selected_index, + marker_width=marker_width, + name_width=name_width, + gap_width=gap_width, + detail_width=detail_width, + ) + ) + + if show_count: + rendered_lines.append( + self._render_count_line( + width=width, + selected_index=selected_index, + total=len(completions), + marker_width=marker_width, + ) + ) + + return UIContent( + get_line=lambda i: rendered_lines[i], + line_count=len(rendered_lines), + cursor_position=Point(x=0, y=selected_line_index), + ) + + def _visible_window_bounds( + self, + *, + completion_count: int, + selected_index: int, + available_rows: int, + ) -> tuple[int, int]: + visible_rows = min(completion_count, max(1, available_rows)) + max_start = max(0, completion_count - visible_rows) + start = min(max(0, selected_index - self._scroll_offset), max_start) + return start, start + visible_rows - 1 + + def _render_item_line( + self, + *, + width: int, + completion: Completion, + is_current: bool, + marker_width: int, + name_width: int, + gap_width: int, + detail_width: int, + ) -> FormattedText: + left_padding = min(self._left_padding(), width) + name = completion.display_text or completion.text + detail = (completion.text or name).rstrip("/") + marker = "→ " if is_current else " " + marker_style = ( + "class:file-completion-menu.marker.current" + if is_current + else "class:file-completion-menu.marker" + ) + name_style = ( + "class:file-completion-menu.name.current" + if is_current + else "class:file-completion-menu.name" + ) + detail_style = ( + "class:file-completion-menu.detail.current" + if is_current + else "class:file-completion-menu.detail" + ) + + fragments: FormattedText = FormattedText() + fragments.append(("class:file-completion-menu", " " * left_padding)) + fragments.append((marker_style, marker.ljust(marker_width))) + fragments.append((name_style, truncate_to_width(name, name_width))) + if detail_width > 0: + fragments.append(("class:file-completion-menu", " " * gap_width)) + fragments.append((detail_style, truncate_to_width(detail, detail_width))) + used_width = left_padding + marker_width + name_width + gap_width + detail_width + if used_width < width: + fragments.append(("class:file-completion-menu", " " * (width - used_width))) + return fragments + + def _render_count_line( + self, + *, + width: int, + selected_index: int, + total: int, + marker_width: int, + ) -> FormattedText: + left_padding = min(self._left_padding() + marker_width, width) + label = f"({selected_index + 1}/{total})" + fragments: FormattedText = FormattedText() + fragments.append(("class:file-completion-menu", " " * left_padding)) + count_text = truncate_to_width(label, max(0, width - left_padding)) + fragments.append(("class:file-completion-menu.count", count_text)) + return fragments diff --git a/src/pythinker_code/ui/shell/prompting/config.py b/src/pythinker_code/ui/shell/prompting/config.py new file mode 100644 index 00000000..0f5272a3 --- /dev/null +++ b/src/pythinker_code/ui/shell/prompting/config.py @@ -0,0 +1,55 @@ +"""Immutable constructor data and callable providers for the prompt session. + +``CustomPromptSession.__init__`` keeps its keyword signature and builds these +bundles internally; they group the constructor surface into constructor data +(``PromptConfig``) and dynamic callable providers (``PromptProviders``). +""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable, Sequence +from dataclasses import dataclass +from typing import Any + +from prompt_toolkit.formatted_text import AnyFormattedText + +from pythinker_code.config import StatusLineConfig +from pythinker_code.llm import ModelCapability +from pythinker_code.soul import StatusSnapshot +from pythinker_code.utils.slashcmd import SlashCommand + + +@dataclass(frozen=True, slots=True) +class BgTaskCounts: + bash: int = 0 + agent: int = 0 + + +@dataclass(frozen=True, slots=True) +class PromptConfig: + """Immutable constructor data for one prompt session.""" + + model_capabilities: set[ModelCapability] + model_name: str | None + thinking: bool + thinking_effort: str | None + agent_mode_slash_commands: Sequence[SlashCommand[Any]] + shell_mode_slash_commands: Sequence[SlashCommand[Any]] + history_enabled: bool + statusline_config: StatusLineConfig | None + sticky_input: bool + + +@dataclass(frozen=True, slots=True) +class PromptProviders: + """Callable providers the prompt session samples while rendering.""" + + status_provider: Callable[[], StatusSnapshot] + status_block_provider: Callable[[int], AnyFormattedText | None] | None + fast_refresh_provider: Callable[[], bool] | None + background_task_count_provider: Callable[[], BgTaskCounts] | None + update_notice_provider: Callable[[], str | None] | None + editor_command_provider: Callable[[], str] + turn_recaps_provider: Callable[[], bool] + plan_mode_toggle_callback: Callable[[], Awaitable[bool]] | None + thinking_effort_cycle_callback: Callable[[], Awaitable[str | None]] | None diff --git a/src/pythinker_code/ui/shell/prompting/keybindings.py b/src/pythinker_code/ui/shell/prompting/keybindings.py new file mode 100644 index 00000000..b4c27698 --- /dev/null +++ b/src/pythinker_code/ui/shell/prompting/keybindings.py @@ -0,0 +1,420 @@ +# pyright: reportPrivateUsage=false +# ``PromptController`` deliberately mirrors the session's protected surface so the +# extracted binding block stays byte-for-byte behavioral with the facade internals. +"""Prompt key-binding construction, extracted from the prompt session facade. + +``build_prompt_key_bindings`` owns the whole prompt_toolkit ``KeyBindings`` +assembly. ``PromptController`` is the internal interface the bindings need from +the session; ``CustomPromptSession`` implements it implicitly and passes itself. +""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable +from typing import Protocol + +from prompt_toolkit import PromptSession +from prompt_toolkit.buffer import Buffer +from prompt_toolkit.document import Document +from prompt_toolkit.filters import Condition, has_completions +from prompt_toolkit.key_binding import KeyBindings, KeyBindingsBase, KeyPressEvent +from prompt_toolkit.keys import Keys + +from pythinker_code.ui.shell.prompting.clipboard import ClipboardAdapter +from pythinker_code.ui.shell.prompting.completion.slash import discard_slash_command +from pythinker_code.ui.shell.prompting.state import PromptMode, RunningPromptDelegate +from pythinker_code.ui.shell.prompting.toasts import toast + + +class PromptController(Protocol): + """Exactly the session surface the prompt key bindings dispatch into.""" + + _session: PromptSession[str] + _suppress_auto_completion: bool + _mode: PromptMode + _thinking: bool + _thinking_effort: str + _thinking_effort_cycle_callback: Callable[[], Awaitable[str | None]] | None + _shortcut_help_open: bool + _clipboard_adapter: ClipboardAdapter + _clipboard_text_available: bool + _media_clipboard_available: bool + + def _slash_completion_active(self, document: Document) -> bool: ... + + def _active_prompt_delegate(self) -> RunningPromptDelegate | None: ... + + def _uses_native_thinking(self) -> bool: ... + + def _should_handle_running_prompt_key(self, key: str) -> bool: ... + + def _handle_running_prompt_key(self, key: str, event: KeyPressEvent) -> None: ... + + def _open_in_external_editor(self, event: KeyPressEvent) -> None: ... + + def _hard_repaint(self, event: KeyPressEvent) -> None: ... + + def _handle_bracketed_paste(self, event: KeyPressEvent) -> None: ... + + def _try_paste_media(self, event: KeyPressEvent) -> bool: ... + + def _insert_pasted_text(self, buffer: Buffer, text: str) -> None: ... + + def toggle_shortcut_help(self) -> None: ... + + def toggle_mode(self) -> None: ... + + def close_shortcut_help(self) -> None: ... + + def accept_staged_suggestion_prefill(self) -> bool: ... + + +def build_prompt_key_bindings(controller: PromptController) -> KeyBindingsBase: + """Build the prompt's full key-binding set against *controller*.""" + clipboard_available = controller._clipboard_text_available + media_clipboard_available = controller._media_clipboard_available + + _kb = KeyBindings() + + def _accept_completion(buff: Buffer) -> None: + """Accept the current or first completion, suppressing re-completion.""" + state = buff.complete_state + if state is None: + return + completion = state.current_completion + if completion is None: + if not state.completions: + return + completion = state.completions[0] + controller._suppress_auto_completion = True + try: + buff.apply_completion(completion) + finally: + controller._suppress_auto_completion = False + + def _is_slash_completion() -> bool: + """True when the active completion menu is for a slash command.""" + buff = controller._session.default_buffer + return bool( + buff.complete_state + and buff.complete_state.completions + and controller._slash_completion_active(buff.document) + ) + + _slash_completion_filter = has_completions & Condition(_is_slash_completion) + _non_slash_completion_filter = has_completions & ~Condition(_is_slash_completion) + + @_kb.add("enter", filter=_slash_completion_filter) + def _(event: KeyPressEvent) -> None: + """Slash command completion: accept and submit in one step.""" + _accept_completion(event.current_buffer) + event.current_buffer.validate_and_handle() + + @_kb.add("escape", eager=True, filter=_slash_completion_filter) + def _(event: KeyPressEvent) -> None: + """Slash completion: Escape discards a draft command, or dismisses + the argument menu when there is no draft command to remove.""" + buffer = event.current_buffer + if not discard_slash_command(buffer): + # Slash-argument completion (e.g. "/model gpt"): the eager + # binding swallowed Escape but there is no root command to + # strip, so dismiss the completion menu explicitly instead of + # leaving it open. + buffer.cancel_completion() + event.app.invalidate() + + @_kb.add("enter", filter=_non_slash_completion_filter) + def _(event: KeyPressEvent) -> None: + """Non-slash completion (file mentions, etc.): accept only.""" + _accept_completion(event.current_buffer) + + def _has_slash_suggestion() -> bool: + buff = controller._session.default_buffer + return bool(buff.suggestion and buff.suggestion.text) + + @_kb.add("tab", filter=Condition(_has_slash_suggestion)) + def _(event: KeyPressEvent) -> None: + """Slash command ghost suggestion: Tab completes the word inline.""" + suggestion = event.current_buffer.suggestion + if suggestion and suggestion.text: + event.current_buffer.insert_text(suggestion.text) + + @_kb.add("?", eager=True) + def _(event: KeyPressEvent) -> None: + """Toggle a compact shortcuts popup when the input row is empty.""" + if controller._active_prompt_delegate() is not None: + event.current_buffer.insert_text("?") + return + if event.current_buffer.text.strip(): + event.current_buffer.insert_text("?") + return + controller.toggle_shortcut_help() + + @_kb.add("c-x", eager=True) + def _(event: KeyPressEvent) -> None: + if controller._active_prompt_delegate() is not None: + return + controller.toggle_mode() + from pythinker_code.telemetry import track + + track("shortcut_mode_switch", to_mode=controller._mode.value) + + @_kb.add("s-tab", eager=True) + def _(event: KeyPressEvent) -> None: + """Cycle thinking effort with Shift+Tab.""" + if controller._active_prompt_delegate() is not None: + return + if controller._thinking_effort_cycle_callback is not None: + + async def _cycle() -> None: + assert controller._thinking_effort_cycle_callback is not None + new_level = await controller._thinking_effort_cycle_callback() + from pythinker_code.telemetry import track + + if new_level is None: + message = ( + "Current model uses native reasoning" + if controller._uses_native_thinking() + else "Current model does not support thinking" + ) + toast( + message, + topic="thinking_level", + duration=3.0, + immediate=True, + ) + else: + controller._thinking_effort = new_level + controller._thinking = new_level != "off" + track("shortcut_thinking_cycle", level=new_level) + toast( + f"Thinking level: {new_level}", + topic="thinking_level", + duration=3.0, + immediate=True, + ) + event.app.invalidate() + + event.app.create_background_task(_cycle()) + event.app.invalidate() + + @_kb.add("escape", "enter", eager=True) + @_kb.add("c-j", eager=True) + def _(event: KeyPressEvent) -> None: + """Insert a newline when Alt-Enter or Ctrl-J is pressed.""" + from pythinker_code.telemetry import track + + track("shortcut_newline") + event.current_buffer.insert_text("\n") + + @_kb.add("c-o", eager=True) + def _(event: KeyPressEvent) -> None: + """Expand active transcript content, or open current buffer in external editor.""" + if controller._active_prompt_delegate() is not None: + if controller._should_handle_running_prompt_key("c-o"): + controller._handle_running_prompt_key("c-o", event) + return + + from pythinker_code.telemetry import track + + track("shortcut_editor") + controller._open_in_external_editor(event) + + @_kb.add("c-l", eager=True) + def _(event: KeyPressEvent) -> None: + """Erase and fully repaint the screen (recovery from console damage).""" + controller._hard_repaint(event) + + def _has_staged_suggestion_prefill() -> bool: + return bool(getattr(controller, "_staged_suggestion_prefill", None)) + + @_kb.add("escape", "s", eager=True, filter=Condition(_has_staged_suggestion_prefill)) + def _(event: KeyPressEvent) -> None: + """Accept the latest agent suggestion into the prompt buffer.""" + if controller.accept_staged_suggestion_prefill(): + from pythinker_code.telemetry import track + + track("suggestion_accepted") + event.app.invalidate() + + @_kb.add( + "up", + eager=True, + filter=Condition(lambda: controller._should_handle_running_prompt_key("up")), + ) + def _(event: KeyPressEvent) -> None: + controller._handle_running_prompt_key("up", event) + + @_kb.add( + "down", + eager=True, + filter=Condition(lambda: controller._should_handle_running_prompt_key("down")), + ) + def _(event: KeyPressEvent) -> None: + controller._handle_running_prompt_key("down", event) + + @_kb.add( + "left", + eager=True, + filter=Condition(lambda: controller._should_handle_running_prompt_key("left")), + ) + def _(event: KeyPressEvent) -> None: + controller._handle_running_prompt_key("left", event) + + @_kb.add( + "right", + eager=True, + filter=Condition(lambda: controller._should_handle_running_prompt_key("right")), + ) + def _(event: KeyPressEvent) -> None: + controller._handle_running_prompt_key("right", event) + + @_kb.add( + "tab", + eager=True, + filter=Condition(lambda: controller._should_handle_running_prompt_key("tab")), + ) + def _(event: KeyPressEvent) -> None: + controller._handle_running_prompt_key("tab", event) + + @_kb.add( + "enter", + eager=True, + filter=Condition(lambda: controller._should_handle_running_prompt_key("enter")), + ) + def _(event: KeyPressEvent) -> None: + controller._handle_running_prompt_key("enter", event) + + @_kb.add( + "space", + eager=True, + filter=Condition(lambda: controller._should_handle_running_prompt_key("space")), + ) + def _(event: KeyPressEvent) -> None: + controller._handle_running_prompt_key("space", event) + + @_kb.add( + "c-s", + eager=True, + filter=Condition(lambda: controller._should_handle_running_prompt_key("c-s")), + ) + def _(event: KeyPressEvent) -> None: + controller._handle_running_prompt_key("c-s", event) + + @_kb.add( + "c-e", + eager=True, + filter=Condition(lambda: controller._should_handle_running_prompt_key("c-e")), + ) + def _(event: KeyPressEvent) -> None: + controller._handle_running_prompt_key("c-e", event) + + @_kb.add( + "c-t", + eager=True, + filter=Condition(lambda: controller._should_handle_running_prompt_key("c-t")), + ) + def _(event: KeyPressEvent) -> None: + controller._handle_running_prompt_key("c-t", event) + + @_kb.add( + "c-c", + eager=True, + filter=Condition(lambda: controller._should_handle_running_prompt_key("c-c")), + ) + def _(event: KeyPressEvent) -> None: + controller._handle_running_prompt_key("c-c", event) + + @_kb.add( + "c-d", + eager=True, + filter=Condition(lambda: controller._should_handle_running_prompt_key("c-d")), + ) + def _(event: KeyPressEvent) -> None: + controller._handle_running_prompt_key("c-d", event) + + @_kb.add( + "escape", + eager=True, + filter=Condition(lambda: controller._should_handle_running_prompt_key("escape")), + ) + def _(event: KeyPressEvent) -> None: + controller._handle_running_prompt_key("escape", event) + + @_kb.add( + "escape", + eager=True, + filter=Condition(lambda: controller._shortcut_help_open), + ) + def _(event: KeyPressEvent) -> None: + controller.close_shortcut_help() + + @_kb.add( + "1", + eager=True, + filter=Condition(lambda: controller._should_handle_running_prompt_key("1")), + ) + def _(event: KeyPressEvent) -> None: + controller._handle_running_prompt_key("1", event) + + @_kb.add( + "2", + eager=True, + filter=Condition(lambda: controller._should_handle_running_prompt_key("2")), + ) + def _(event: KeyPressEvent) -> None: + controller._handle_running_prompt_key("2", event) + + @_kb.add( + "3", + eager=True, + filter=Condition(lambda: controller._should_handle_running_prompt_key("3")), + ) + def _(event: KeyPressEvent) -> None: + controller._handle_running_prompt_key("3", event) + + @_kb.add( + "4", + eager=True, + filter=Condition(lambda: controller._should_handle_running_prompt_key("4")), + ) + def _(event: KeyPressEvent) -> None: + controller._handle_running_prompt_key("4", event) + + @_kb.add( + "5", + eager=True, + filter=Condition(lambda: controller._should_handle_running_prompt_key("5")), + ) + def _(event: KeyPressEvent) -> None: + controller._handle_running_prompt_key("5", event) + + @_kb.add( + "6", + eager=True, + filter=Condition(lambda: controller._should_handle_running_prompt_key("6")), + ) + def _(event: KeyPressEvent) -> None: + controller._handle_running_prompt_key("6", event) + + @_kb.add(Keys.BracketedPaste, eager=True) + def _(event: KeyPressEvent) -> None: + controller._handle_bracketed_paste(event) + + if clipboard_available or media_clipboard_available: + + @_kb.add("c-v", eager=True) + def _(event: KeyPressEvent) -> None: + from pythinker_code.telemetry import track + + track("shortcut_paste") + if controller._try_paste_media(event): + return + if clipboard_available: + clipboard_text = controller._clipboard_adapter.paste_text(event.app.clipboard) + if clipboard_text is None: + return + controller._insert_pasted_text(event.current_buffer, clipboard_text) + event.app.invalidate() + + return _kb diff --git a/src/pythinker_code/ui/shell/visualize/_interactive.py b/src/pythinker_code/ui/shell/visualize/_interactive.py index 753be311..b440cace 100644 --- a/src/pythinker_code/ui/shell/visualize/_interactive.py +++ b/src/pythinker_code/ui/shell/visualize/_interactive.py @@ -345,9 +345,7 @@ def _start_btw(self, question: str) -> None: self._btw_modal = modal self._prompt_session.attach_modal(modal) # Now safe to clear — buffer is hidden by modal - buf = self._prompt_session._session.default_buffer # pyright: ignore[reportPrivateUsage] - if buf.text: - buf.set_document(Document(), bypass_readonly=True) + self._prompt_session.clear_input() self._btw_refresh_task = asyncio.create_task(self._btw_refresh_loop()) self._btw_run_task = asyncio.create_task(self._run_btw(question)) @@ -1067,8 +1065,7 @@ def should_handle_running_prompt_key(self, key: str) -> bool: # Only intercept when buffer is empty — otherwise let prompt_toolkit # handle ↑ for cursor movement / history navigation. if key == "up" and self._queued_messages: - buf = self._prompt_session._session.default_buffer # pyright: ignore[reportPrivateUsage] - return not buf.text.strip() + return not self._prompt_session.input_text().strip() # Ctrl+S: immediate steer return key == "c-s" @@ -1121,7 +1118,7 @@ def handle_running_prompt_key(self, key: str, event: KeyPressEvent) -> None: buf = event.current_buffer text = buf.text.strip() if text: - steer_input = self._prompt_session._build_user_input(text) # pyright: ignore[reportPrivateUsage] + steer_input = self._prompt_session.build_user_input(text) self._clear_buffer(buf) self.handle_immediate_steer(steer_input) elif self._queued_messages: @@ -1186,8 +1183,8 @@ def _on_question_panel_state_changed(self) -> None: panel, on_advance=self._advance_question, on_invalidate=self._flush_prompt_refresh, - buffer_text_provider=lambda: self._prompt_session._session.default_buffer.text, # pyright: ignore[reportPrivateUsage] - text_expander=self._prompt_session._get_placeholder_manager().serialize_for_history, # pyright: ignore[reportPrivateUsage] + buffer_text_provider=self._prompt_session.input_text, + text_expander=self._prompt_session.serialize_for_history, ) self._prompt_session.attach_modal(self._question_modal) else: diff --git a/tests/ui_and_conv/test_modal_lifecycle.py b/tests/ui_and_conv/test_modal_lifecycle.py index 30127862..c3fd4d36 100644 --- a/tests/ui_and_conv/test_modal_lifecycle.py +++ b/tests/ui_and_conv/test_modal_lifecycle.py @@ -604,6 +604,15 @@ def invalidate(self) -> None: def _get_placeholder_manager(self) -> _FakePlaceholderManager: return _FakePlaceholderManager() + def serialize_for_history(self, command: str) -> str: + return _FakePlaceholderManager.serialize_for_history(command) + + def input_text(self) -> str: + return "" + + def input_state(self) -> tuple[str, int]: + return "", 0 + view = _PromptLiveView( StatusUpdate(), prompt_session=cast(Any, _PromptSession()), @@ -647,6 +656,15 @@ def invalidate(self) -> None: def _get_placeholder_manager(self) -> _FakePlaceholderManager: return _FakePlaceholderManager() + def serialize_for_history(self, command: str) -> str: + return _FakePlaceholderManager.serialize_for_history(command) + + def input_text(self) -> str: + return "" + + def input_state(self) -> tuple[str, int]: + return "", 0 + view = _PromptLiveView( StatusUpdate(), prompt_session=cast(Any, _PromptSession()), @@ -701,6 +719,15 @@ def invalidate(self) -> None: def _get_placeholder_manager(self) -> _FakePlaceholderManager: return _FakePlaceholderManager() + def serialize_for_history(self, command: str) -> str: + return _FakePlaceholderManager.serialize_for_history(command) + + def input_text(self) -> str: + return "" + + def input_state(self) -> tuple[str, int]: + return "", 0 + view = _PromptLiveView( StatusUpdate(), prompt_session=cast(Any, _PromptSession()), @@ -1063,6 +1090,15 @@ def invalidate(self) -> None: def _get_placeholder_manager(self) -> _FakePlaceholderManager: return _FakePlaceholderManager() + def serialize_for_history(self, command: str) -> str: + return _FakePlaceholderManager.serialize_for_history(command) + + def input_text(self) -> str: + return "" + + def input_state(self) -> tuple[str, int]: + return "", 0 + shell._prompt_session = _PromptSession() # type: ignore[attr-defined] # Send both requests @@ -1168,6 +1204,9 @@ async def test_prompt_live_view_question_does_not_affect_should_handle_key() -> "detach_modal": lambda self, d: None, "invalidate": lambda self: None, "_get_placeholder_manager": lambda self: _FakePlaceholderManager(), + "serialize_for_history": lambda self, command: command, + "input_text": lambda self: "", + "input_state": lambda self: ("", 0), }, )(), ), @@ -1207,6 +1246,9 @@ async def test_prompt_live_view_render_body_no_awaiting_other_hint() -> None: "detach_modal": lambda self, d: None, "invalidate": lambda self: None, "_get_placeholder_manager": lambda self: _FakePlaceholderManager(), + "serialize_for_history": lambda self, command: command, + "input_text": lambda self: "", + "input_state": lambda self: ("", 0), }, )(), ), diff --git a/tests/ui_and_conv/test_prompt_height_budget.py b/tests/ui_and_conv/test_prompt_height_budget.py index 18ea685a..6c077f70 100644 --- a/tests/ui_and_conv/test_prompt_height_budget.py +++ b/tests/ui_and_conv/test_prompt_height_budget.py @@ -1,10 +1,13 @@ from __future__ import annotations +import math +from dataclasses import replace from types import SimpleNamespace from typing import Literal import pytest from prompt_toolkit.formatted_text import FormattedText +from prompt_toolkit.utils import get_cwidth from pythinker_code.ui.shell import prompt as shell_prompt from pythinker_code.ui.shell.prompt import CustomPromptSession, PromptMode @@ -14,6 +17,8 @@ PromptSceneBudget, allocate_prompt_scene_rows, freeze_fragments, + truncate_footer_left, + truncate_footer_right, ) Scene = Literal[ @@ -478,3 +483,108 @@ def test_scene_budget_is_zero_safe(terminal_rows: int) -> None: ) assert budget.preamble_rows == max(0, terminal_rows - 12) + + +# Unicode / terminal-capability cases. Row and column measurements must use terminal +# CELL width (prompt_toolkit ``get_cwidth``), not codepoint counts: combining marks are +# zero cells, emoji and CJK are two cells, RTL text is one cell per letter. +_UNICODE_SAMPLES = ( + ("combining_marks", "e\u0301" * 30), + ("emoji", "\U0001f642" * 30), + ("cjk_wide", "漢字端末幅測定" * 8), + ("rtl", "مرحبا بالعالم اختبار" * 3), + ("wide_key_labels", "⌘K 漢🙂 " * 8), + ("ascii_glyphs", "[tool] running... -> ok " * 4), +) + + +def _cell_width(text: str) -> int: + return sum(max(0, get_cwidth(character)) for character in text) + + +@pytest.mark.parametrize(("case", "text"), _UNICODE_SAMPLES, ids=lambda value: str(value)) +@pytest.mark.parametrize("width", (20, 40, 80)) +def test_display_rows_measure_unicode_in_terminal_cells( + case: str, + text: str, + width: int, +) -> None: + rows = shell_prompt._formatted_text_display_rows(FormattedText([("", text)]), width) + + # Every rendered row fits the terminal width when measured in cells; a + # codepoint-based split would overflow rows containing wide characters. + for row in rows: + assert _cell_width("".join(fragment[1] for fragment in row)) <= width + # Cell accounting requires at least ceil(total_cells / width) rows, and a + # codepoint count would demand more rows than cells allow for combining marks. + total_cells = _cell_width(text) + assert len(rows) >= math.ceil(total_cells / width) + if case == "combining_marks": + # 30 base letters + 30 zero-width combining marks is 30 cells, not 60. + assert total_cells == 30 + assert len(rows) == math.ceil(30 / width) + if case == "cjk_wide": + # Even widths pack wide chars exactly: two cells per char, no spare cell. + assert len(rows) == math.ceil(total_cells / width) + + +@pytest.mark.parametrize(("case", "text"), _UNICODE_SAMPLES, ids=lambda value: str(value)) +@pytest.mark.parametrize(("width", "height"), ((20, 6), (40, 10))) +def test_unicode_scenes_stay_within_height_budget( + case: str, + text: str, + width: int, + height: int, + monkeypatch: pytest.MonkeyPatch, +) -> None: + del case + session = _session_for_scene( + "body_pinned", + width=width, + height=height, + card_style=False, + monkeypatch=monkeypatch, + ) + frame = session._current_prompt_frame + assert frame is not None + session._current_prompt_frame = replace( + frame, + interactive_body=_text(text), + pinned_tail=_text(f"⏺ {text}"), + ) + + message = session._render_agent_prompt_message() + message_rows = len(shell_prompt._formatted_text_display_rows(message, width)) if message else 0 + monkeypatch.setattr( + shell_prompt, + "get_app_or_none", + lambda: SimpleNamespace( + output=SimpleNamespace(get_size=lambda: SimpleNamespace(columns=width, rows=height)) + ), + ) + footer = session._fit_toolbar_to_terminal(FormattedText([("", text)]), width) + footer_rows = len(shell_prompt._formatted_text_display_rows(footer, width)) if footer else 0 + + assert message_rows + footer_rows <= height + + +@pytest.mark.parametrize("ascii_only", (False, True), ids=("unicode", "ascii")) +@pytest.mark.parametrize("width", (6, 11, 24)) +def test_footer_truncation_counts_wide_labels_in_cells( + ascii_only: bool, + width: int, +) -> None: + text = "⌘K 漢字🙂 model: qwen3.6-35b" + + right = truncate_footer_right(text, width, ascii_only=ascii_only) + left = truncate_footer_left(text, width, ascii_only=ascii_only) + + assert _cell_width(right) <= width + assert _cell_width(left) <= width + ellipsis = "..." if ascii_only else "…" + assert right.endswith(ellipsis) + assert left.startswith(ellipsis) + if ascii_only: + # ASCII glyph mode must never introduce non-ASCII ellipsis characters. + assert "…" not in right + assert "…" not in left diff --git a/tests/ui_and_conv/test_prompt_public_contract.py b/tests/ui_and_conv/test_prompt_public_contract.py index 05ed4ed1..8f928d68 100644 --- a/tests/ui_and_conv/test_prompt_public_contract.py +++ b/tests/ui_and_conv/test_prompt_public_contract.py @@ -55,6 +55,18 @@ } +# Narrow façade methods that external shell callers (ui/shell/__init__.py, +# ui/shell/visualize/_interactive.py) use instead of reaching into private +# prompt internals. +PROMPT_SESSION_FACADE_METHODS = ( + "input_text", + "input_state", + "clear_input", + "build_user_input", + "serialize_for_history", +) + + def test_prompt_module_preserves_repository_import_surface() -> None: """Names found by searching prompt imports and ``shell_prompt`` references stay exported.""" for name, imported in PUBLIC_PROMPT_CONTRACT.items(): @@ -81,6 +93,44 @@ def test_custom_prompt_session_keyword_initialization_contract( assert prompt_session._session.app.max_render_postpone_time == pytest.approx(1 / 30) +def test_prompt_session_exposes_facade_methods(prompt_session: CustomPromptSession) -> None: + for name in PROMPT_SESSION_FACADE_METHODS: + assert callable(getattr(prompt_session, name)) + + +@pytest.mark.asyncio +async def test_input_facade_reads_and_clears_default_buffer( + prompt_session: CustomPromptSession, +) -> None: + assert prompt_session.input_state() == ("", 0) + prompt_session._session.default_buffer.set_document( + shell_prompt.Document("hello", 5), bypass_readonly=True + ) + assert prompt_session.input_text() == "hello" + assert prompt_session.input_state() == ("hello", 5) + prompt_session.clear_input() + assert prompt_session.input_state() == ("", 0) + + +def test_input_facade_tolerates_partially_constructed_session() -> None: + bare = CustomPromptSession.__new__(CustomPromptSession) + assert bare.input_text() == "" + assert bare.input_state() == ("", 0) + bare.clear_input() # must not raise + + +def test_serialize_for_history_passes_plain_text_through( + prompt_session: CustomPromptSession, +) -> None: + assert prompt_session.serialize_for_history("plain text") == "plain text" + + +def test_build_user_input_returns_user_input(prompt_session: CustomPromptSession) -> None: + user_input = prompt_session.build_user_input("hello world") + assert isinstance(user_input, UserInput) + assert user_input.command == "hello world" + + class _CountingRunningPrompt(_DummyRunningPrompt): def __init__(self) -> None: self.calls: Counter[str] = Counter() diff --git a/tests/ui_and_conv/test_shell_task_slash.py b/tests/ui_and_conv/test_shell_task_slash.py index e5989395..9207624a 100644 --- a/tests/ui_and_conv/test_shell_task_slash.py +++ b/tests/ui_and_conv/test_shell_task_slash.py @@ -233,6 +233,15 @@ def invalidate(self) -> None: def _get_placeholder_manager(self) -> _FakePlaceholderManager: return _FakePlaceholderManager() + def serialize_for_history(self, command: str) -> str: + return _FakePlaceholderManager.serialize_for_history(command) + + def input_text(self) -> str: + return "" + + def input_state(self) -> tuple[str, int]: + return "", 0 + shell._prompt_session = _PromptSession() # type: ignore[attr-defined] request = ApprovalRequest( @@ -301,6 +310,15 @@ def invalidate(self) -> None: def _get_placeholder_manager(self) -> _FakePlaceholderManager: return _FakePlaceholderManager() + def serialize_for_history(self, command: str) -> str: + return _FakePlaceholderManager.serialize_for_history(command) + + def input_text(self) -> str: + return "" + + def input_state(self) -> tuple[str, int]: + return "", 0 + shell._prompt_session = _PromptSession() # type: ignore[attr-defined] request_one = ApprovalRequest( @@ -377,6 +395,15 @@ def invalidate(self) -> None: def _get_placeholder_manager(self) -> _FakePlaceholderManager: return _FakePlaceholderManager() + def serialize_for_history(self, command: str) -> str: + return _FakePlaceholderManager.serialize_for_history(command) + + def input_text(self) -> str: + return "" + + def input_state(self) -> tuple[str, int]: + return "", 0 + shell._prompt_session = _PromptSession() # type: ignore[attr-defined] request_one = ApprovalRequest( @@ -724,6 +751,15 @@ def invalidate(self) -> None: def _get_placeholder_manager(self) -> _FakePlaceholderManager: return _FakePlaceholderManager() + def serialize_for_history(self, command: str) -> str: + return _FakePlaceholderManager.serialize_for_history(command) + + def input_text(self) -> str: + return "" + + def input_state(self) -> tuple[str, int]: + return "", 0 + shell._prompt_session = _PromptSession() # type: ignore[attr-defined] request = ApprovalRequest( @@ -802,6 +838,15 @@ def invalidate(self) -> None: def _get_placeholder_manager(self) -> _FakePlaceholderManager: return _FakePlaceholderManager() + def serialize_for_history(self, command: str) -> str: + return _FakePlaceholderManager.serialize_for_history(command) + + def input_text(self) -> str: + return "" + + def input_state(self) -> tuple[str, int]: + return "", 0 + shell._prompt_session = _PromptSession() # type: ignore[attr-defined] request = ApprovalRequest(