From 3233c2d206118775a4b8392eb07e545ddc696a44 Mon Sep 17 00:00:00 2001 From: elkaix Date: Mon, 20 Jul 2026 23:45:54 -0400 Subject: [PATCH 1/7] refactor(tui): scope prompt resources to each session --- src/pythinker_code/ui/shell/prompt.py | 536 ++++++------------ .../ui/shell/prompting/__init__.py | 18 +- .../ui/shell/prompting/clipboard.py | 111 ++++ .../ui/shell/prompting/git_status.py | 407 +++++++++++++ .../ui/shell/prompting/history.py | 276 +++++++++ .../ui/shell/prompting/toasts.py | 182 ++++++ src/pythinker_code/ui/shell/slash.py | 46 ++ tests/ui_and_conv/test_prompt_history.py | 90 +++ tests/ui_and_conv/test_prompt_resources.py | 167 ++++++ tests/ui_and_conv/test_prompt_tips.py | 70 --- 10 files changed, 1483 insertions(+), 420 deletions(-) create mode 100644 src/pythinker_code/ui/shell/prompting/clipboard.py create mode 100644 src/pythinker_code/ui/shell/prompting/git_status.py create mode 100644 src/pythinker_code/ui/shell/prompting/history.py create mode 100644 src/pythinker_code/ui/shell/prompting/toasts.py create mode 100644 tests/ui_and_conv/test_prompt_resources.py diff --git a/src/pythinker_code/ui/shell/prompt.py b/src/pythinker_code/ui/shell/prompt.py index d230dc2d..c2457948 100644 --- a/src/pythinker_code/ui/shell/prompt.py +++ b/src/pythinker_code/ui/shell/prompt.py @@ -2,16 +2,14 @@ import asyncio import contextlib -import json import os import random -import re import shlex -import subprocess import sys import time 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 @@ -23,7 +21,6 @@ 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.clipboard.pyperclip import PyperclipClipboard from prompt_toolkit.completion import Completion, merge_completers from prompt_toolkit.data_structures import Point from prompt_toolkit.document import Document @@ -51,7 +48,7 @@ from prompt_toolkit.layout.menus import CompletionsMenu from prompt_toolkit.patch_stdout import patch_stdout from prompt_toolkit.utils import get_cwidth -from pydantic import BaseModel, ValidationError +from pydantic import BaseModel from pythinker_host import get_current_host from pythinker_host.path import HostPath @@ -74,12 +71,32 @@ sanitize_surrogates, ) from pythinker_code.ui.shell.prompting import ( + ClipboardAdapter, FrozenFragments, + GitSnapshot, + GitStatusIndex, PromptFrame, PromptFrameCollector, + PromptHistoryError, + PromptHistoryStore, PromptSceneBudget, + ToastManager, + ToastSnapshot, allocate_prompt_scene_rows, ) +from pythinker_code.ui.shell.prompting.clipboard import ( + bind_clipboard_adapter, + reset_clipboard_adapter, +) +from pythinker_code.ui.shell.prompting.clipboard import ( + grab_media_from_clipboard as grab_media_from_clipboard, +) +from pythinker_code.ui.shell.prompting.clipboard import ( + is_clipboard_available as is_clipboard_available, +) +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, @@ -95,6 +112,17 @@ HostFileMentionCompleter, WorkspaceIndex, ) +from pythinker_code.ui.shell.prompting.git_status import ( + bind_git_status_index, + current_git_snapshot, + reset_git_status_index, +) +from pythinker_code.ui.shell.prompting.history import ( + HistoryEntry, + ensure_private_history_path, + load_history_entries, + redact_history_secrets, +) from pythinker_code.ui.shell.prompting.lifecycle import PromptLifecycle from pythinker_code.ui.shell.prompting.state import ( BufferObserved, @@ -120,6 +148,13 @@ TurnStarting, transition, ) +from pythinker_code.ui.shell.prompting.toasts import ( + bind_toast_manager, + bootstrap_toast_queues, + current_toast, + reset_toast_manager, + toast, +) from pythinker_code.ui.shell.spacing import ( PREAMBLE_EARLIER_OUTPUT_HIDDEN_HINT, ensure_prompt_newline, @@ -130,11 +165,6 @@ from pythinker_code.ui.theme import get_prompt_style, get_toolbar_colors, thinking_dot_style from pythinker_code.ui.theme import get_tui_tokens as _get_tui_tokens from pythinker_code.ui.tui_config import is_card_style -from pythinker_code.utils.clipboard import ( - grab_media_from_clipboard, - is_clipboard_available, - is_media_clipboard_available, -) from pythinker_code.utils.logging import logger from pythinker_code.utils.slashcmd import SlashCommand from pythinker_code.wire.types import ContentPart, TextPart @@ -1164,35 +1194,7 @@ def _render_count_line( LocalFileMentionCompleter = HostFileMentionCompleter -class _HistoryEntry(BaseModel): - content: str - - -_HISTORY_SECRET_PATTERNS: tuple[tuple[re.Pattern[str], str], ...] = ( - ( - re.compile(r"(?i)\b((?:authorization\s*:\s*)?(?:bearer|basic)\s+)[A-Za-z0-9._~+/=-]{8,}"), - r"\1[REDACTED]", - ), - ( - re.compile( - r"(?i)([\"']?(?:api[_-]?key|token|secret|password|access[_-]?token|" - r"refresh[_-]?token|id[_-]?token|session[_-]?token)[\"']?\s*[:=]\s*[\"'])" - r"([^\"'\r\n]{8,})([\"'])" - ), - r"\1[REDACTED]\3", - ), - ( - re.compile( - r"(?i)\b(api[_-]?key|token|secret|password|access[_-]?token|" - r"refresh[_-]?token|id[_-]?token|session[_-]?token)(\s*[:=]\s*)([^\s'\"&]{8,})" - ), - r"\1\2[REDACTED]", - ), - (re.compile(r"\b(sk-[A-Za-z0-9][A-Za-z0-9_-]{16,})\b"), "[REDACTED]"), - (re.compile(r"\b(?:gh[opusr]_[A-Za-z0-9_]{20,}|github_pat_[A-Za-z0-9_]{20,})\b"), "[REDACTED]"), - (re.compile(r"\b(AKIA[0-9A-Z]{16})\b"), "[REDACTED]"), - (re.compile(r"\b(AIza[0-9A-Za-z_-]{20,})\b"), "[REDACTED]"), -) +_HistoryEntry = HistoryEntry def _env_truthy(name: str) -> bool: @@ -1200,56 +1202,11 @@ def _env_truthy(name: str) -> bool: def _redact_history_secrets(text: str) -> str: - redacted = text - for pattern, replacement in _HISTORY_SECRET_PATTERNS: - redacted = pattern.sub(replacement, redacted) - return redacted - - -def _ensure_private_history_path(path: Path) -> None: - with contextlib.suppress(OSError): - os.chmod(path.parent, 0o700) - if path.exists(): - with contextlib.suppress(OSError): - os.chmod(path, 0o600) - - -def _load_history_entries(history_file: Path) -> list[_HistoryEntry]: - entries: list[_HistoryEntry] = [] - if not history_file.exists(): - return entries - - try: - with history_file.open(encoding="utf-8") as f: - for raw_line in f: - line = raw_line.strip() - if not line: - continue - try: - record = json.loads(line) - except json.JSONDecodeError: - logger.warning( - "Failed to parse user history line; skipping: {line}", - line=line, - ) - continue - try: - entry = _HistoryEntry.model_validate(record) - entries.append(entry) - except ValidationError: - logger.warning( - "Failed to validate user history entry; skipping: {line}", - line=line, - ) - continue - except OSError as exc: - logger.warning( - "Failed to load user history file: {file} ({error})", - file=history_file, - error=exc, - ) + return redact_history_secrets(text) - return entries + +_ensure_private_history_path = ensure_private_history_path +_load_history_entries = load_history_entries class PromptUIState(Enum): @@ -1279,131 +1236,20 @@ def __bool__(self) -> bool: # ponytail: 2s quiet threshold — silent dev servers drop to idle refresh _BG_QUIET_THRESHOLD_S = 2.0 -_GIT_BRANCH_TTL = 5.0 -_GIT_STATUS_TTL = 15.0 _TIP_ROTATE_INTERVAL = 30.0 _MAX_CWD_COLS = 30 _MAX_BRANCH_COLS = 22 -@dataclass -class _GitBranchState: - timestamp: float = 0.0 - branch: str | None = None - proc: subprocess.Popen[str] | None = None - - -@dataclass -class _GitStatusState: - timestamp: float = 0.0 - dirty: bool = False - ahead: int = 0 - behind: int = 0 - proc: subprocess.Popen[str] | None = None - - -_git_branch_state = _GitBranchState() -_git_status_state = _GitStatusState() - -_GIT_STATUS_AB_RE = re.compile(r"\[(?:ahead (\d+))?(?:, )?(?:behind (\d+))?\]") - - def _get_git_branch() -> str | None: - """Return the current git branch name via a non-blocking cached subprocess.""" - state = _git_branch_state - now = time.monotonic() - - # Collect result if a previously launched process has finished - if state.proc is not None: - returncode = state.proc.poll() - if returncode is not None: - try: - stdout, _ = state.proc.communicate() - new_branch = stdout.strip() or None - # Branch changed — discard any in-flight status subprocess so it cannot - # write stale results for the old branch, then force an immediate refresh. - if new_branch != state.branch: - if _git_status_state.proc is not None: - with contextlib.suppress(Exception): - _git_status_state.proc.terminate() - _git_status_state.proc = None - _git_status_state.timestamp = 0.0 - state.branch = new_branch - except Exception: - state.branch = None - state.proc = None - - # Launch a new process when the TTL has expired and nothing is running - if state.timestamp + _GIT_BRANCH_TTL <= now and state.proc is None: - state.timestamp = now - try: - state.proc = subprocess.Popen( - ["git", "branch", "--show-current"], - stdout=subprocess.PIPE, - stderr=subprocess.DEVNULL, - text=True, - encoding="utf-8", - errors="replace", - ) - except Exception: - state.branch = None - - return state.branch + """Return the active session's cached branch without blocking on I/O.""" + return current_git_snapshot().branch def _get_git_status() -> tuple[bool, int, int]: - """Return (dirty, ahead, behind) via a non-blocking cached subprocess. - - Runs ``git status --porcelain -b`` (includes untracked files so newly created - files show as dirty). TTL is longer than the branch check because file-tree - scanning is expensive. - """ - state = _git_status_state - now = time.monotonic() - - if state.proc is not None: - returncode = state.proc.poll() - if returncode is not None: - try: - stdout, _ = state.proc.communicate() - dirty = False - ahead = 0 - behind = 0 - for line in stdout.splitlines(): - if line.startswith("## "): - m = _GIT_STATUS_AB_RE.search(line) - if m: - ahead = int(m.group(1) or 0) - behind = int(m.group(2) or 0) - elif line.strip(): - dirty = True - state.dirty = dirty - state.ahead = ahead - state.behind = behind - except Exception: - pass - state.proc = None - elif now - state.timestamp > _GIT_STATUS_TTL: - # Subprocess is stuck (e.g. OS pipe buffer full from many untracked files). - # Terminate it so the toolbar is not permanently frozen; retry after next TTL. - with contextlib.suppress(Exception): - state.proc.terminate() - state.proc = None - state.timestamp = now # delay next spawn by one full TTL - - if state.timestamp + _GIT_STATUS_TTL <= now and state.proc is None: - state.timestamp = now - with contextlib.suppress(Exception): - state.proc = subprocess.Popen( - ["git", "status", "--porcelain", "-b"], - stdout=subprocess.PIPE, - stderr=subprocess.DEVNULL, - text=True, - encoding="utf-8", - errors="replace", - ) - - return state.dirty, state.ahead, state.behind + """Return the active session's cached dirty/ahead/behind state.""" + snapshot = current_git_snapshot() + return snapshot.dirty, snapshot.ahead, snapshot.behind def _format_git_badge(branch: str, dirty: bool, ahead: int, behind: int) -> str: @@ -1423,55 +1269,9 @@ def _format_git_badge(branch: str, dirty: bool, ahead: int, behind: int) -> str: return f"{branch} [{' '.join(parts)}]" -_GIT_DIFFSTAT_TTL = 15.0 - - -@dataclass -class _GitDiffStatState: - timestamp: float = 0.0 - added: int = 0 - removed: int = 0 - proc: subprocess.Popen[str] | None = None - - -_git_diffstat_state = _GitDiffStatState() - - def _get_git_diffstat() -> tuple[int, int] | None: - """Return (added, removed) working-tree line counts via a non-blocking cached - subprocess. None when not a repo / no changes.""" - from pythinker_code.ui.shell.statusline import parse_shortstat - - state = _git_diffstat_state - now = time.monotonic() - if state.proc is not None: - returncode = state.proc.poll() - if returncode is not None: - try: - stdout, _ = state.proc.communicate() - state.added, state.removed = parse_shortstat(stdout) - except Exception: - logger.debug("git diff --shortstat read/parse failed", exc_info=True) - state.proc = None - elif now - state.timestamp > _GIT_DIFFSTAT_TTL: - with contextlib.suppress(Exception): - state.proc.terminate() - state.proc = None - state.timestamp = now - if state.timestamp + _GIT_DIFFSTAT_TTL <= now and state.proc is None: - state.timestamp = now - with contextlib.suppress(Exception): - state.proc = subprocess.Popen( - ["git", "--no-optional-locks", "diff", "--shortstat"], - stdout=subprocess.PIPE, - stderr=subprocess.DEVNULL, - text=True, - encoding="utf-8", - errors="replace", - ) - if state.added == 0 and state.removed == 0: - return None - return state.added, state.removed + """Return the active session's cached working-tree line counts.""" + return current_git_snapshot().diffstat def _shorten_cwd(path: str) -> str: @@ -1527,16 +1327,6 @@ def _truncate_right(text: str, max_cols: int) -> str: return "".join(chars) + ellipsis -@dataclass(slots=True) -class _ToastEntry: - topic: str | None - """There can be only one toast of each non-None topic in the queue.""" - message: str - expires_at: float - style: str = "" - """Optional prompt_toolkit style for the rendered line; "" uses the default toast style.""" - - @dataclass(frozen=True, slots=True) class BgTaskCounts: bash: int = 0 @@ -1568,48 +1358,8 @@ class PinnedStatusTailProvider(Protocol): def render_pinned_status_tail(self, columns: int) -> AnyFormattedText: ... -_toast_queues: dict[Literal["left", "right"], deque[_ToastEntry]] = { - "left": deque(), - "right": deque(), -} -"""The queue of toasts to show, including the one currently being shown (the first one).""" - - -def toast( - message: str, - duration: float = 5.0, - topic: str | None = None, - immediate: bool = False, - position: Literal["left", "right"] = "left", - style: str = "", -) -> None: - queue = _toast_queues[position] - duration = max(duration, _IDLE_REFRESH_INTERVAL) - entry = _ToastEntry( - topic=topic, - message=message, - expires_at=time.monotonic() + duration, - style=style, - ) - if topic is not None: - # Remove existing toasts with the same topic - for existing in list(queue): - if existing.topic == topic: - queue.remove(existing) - if immediate: - queue.appendleft(entry) - else: - queue.append(entry) - - -def _current_toast(position: Literal["left", "right"] = "left") -> _ToastEntry | None: - queue = _toast_queues[position] - now = time.monotonic() - while queue and queue[0].expires_at <= now: - queue.popleft() - if not queue: - return None - return queue[0] +_toast_queues = bootstrap_toast_queues +_current_toast = current_toast def _build_toolbar_tips(clipboard_available: bool) -> list[str]: @@ -1711,7 +1461,24 @@ def __init__( ) if self._history_enabled: history_dir.mkdir(parents=True, exist_ok=True) - _ensure_private_history_path(self._history_file) + self._history_store = PromptHistoryStore( + self._history_file, + enabled=self._history_enabled, + ) + self._lifecycle.register_closer("prompt history", self._history_store.aclose) + self._toast_manager = ToastManager() + self._lifecycle.register_closer("toast manager", self._toast_manager.aclose) + self._clipboard_adapter = ClipboardAdapter() + self._lifecycle.register_closer("clipboard adapter", self._clipboard_adapter.aclose) + self._git_status_index = GitStatusIndex( + get_current_host(), + self._lifecycle, + on_publish=self.invalidate, + ) + self._lifecycle.register_closer("Git status index", self._git_status_index.aclose) + 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 @@ -1755,12 +1522,12 @@ def __init__( self._last_ui_state: PromptUIState = PromptUIState.NORMAL_INPUT self._suspended_buffer_document: Document | None = None self._prompt_state = PromptState(mode=self._mode) - clipboard_available = is_clipboard_available() - media_clipboard_available = is_media_clipboard_available() + clipboard_available = self._clipboard_adapter.is_text_available() + media_clipboard_available = self._clipboard_adapter.is_media_available() self._tips = _build_toolbar_tips(clipboard_available or media_clipboard_available) self._tip_rotation_index: int = random.randrange(len(self._tips)) if self._tips else 0 - history_entries = _load_history_entries(self._history_file) if self._history_enabled else [] + history_entries = self._history_store.load() history = InMemoryHistory() for entry in history_entries: history.append_string(entry.content) @@ -2163,20 +1930,17 @@ def _(event: KeyPressEvent) -> None: if self._try_paste_media(event): return if clipboard_available: - try: - clipboard_data = event.app.clipboard.get_data() - except Exception: - return - if clipboard_data is None: # type: ignore[reportUnnecessaryComparison] + clipboard_text = self._clipboard_adapter.paste_text(event.app.clipboard) + if clipboard_text is None: return - self._insert_pasted_text(event.current_buffer, clipboard_data.text) + self._insert_pasted_text(event.current_buffer, clipboard_text) event.app.invalidate() # Only use PyperclipClipboard when pyperclip actually works. # PromptSession built-in keybindings (ctrl-k, ctrl-w, ctrl-y) # use clipboard without error handling, so a broken clipboard # object would crash the UI. - clipboard = PyperclipClipboard() if clipboard_available else None + clipboard = self._clipboard_adapter.create_text_clipboard(available=clipboard_available) self._session = PromptSession[str]( message=self._render_message, @@ -3388,8 +3152,13 @@ async def _refresh() -> None: self._statusline_runner.start() def __enter__(self) -> CustomPromptSession: - self._start() - return self + self._bind_resource_bindings() + try: + self._start() + return self + except BaseException: + self._reset_resource_bindings() + raise def __exit__(self, *_: object) -> None: if self._status_refresh_task is not None and not self._status_refresh_task.done(): @@ -3397,17 +3166,51 @@ def __exit__(self, *_: object) -> None: self._status_refresh_task = None if self._statusline_runner is not None: self._statusline_runner.cancel() + self._reset_resource_bindings() async def __aenter__(self) -> CustomPromptSession: - self._start() - return self + self._bind_resource_bindings() + try: + self._start() + return self + except BaseException: + self._reset_resource_bindings() + raise + + def _bind_resource_bindings(self) -> None: + self._git_status_token = bind_git_status_index(self._git_status_index) + self._toast_token = bind_toast_manager(self._toast_manager) + self._clipboard_token = bind_clipboard_adapter(self._clipboard_adapter) async def __aexit__(self, *_: object) -> None: await self.aclose() async def aclose(self) -> None: - await self._lifecycle.aclose() - self._status_refresh_task = None + try: + await self._lifecycle.aclose() + self._status_refresh_task = None + finally: + self._reset_resource_bindings() + + def _reset_resource_bindings(self) -> None: + # Partially constructed sessions (lifecycle tests) may never have bound tokens. + clipboard_token = getattr(self, "_clipboard_token", None) + self._clipboard_token = None + if clipboard_token is not None: + reset_clipboard_adapter(clipboard_token) + toast_token = getattr(self, "_toast_token", None) + self._toast_token = None + if toast_token is not None: + reset_toast_manager(toast_token) + git_status_token = getattr(self, "_git_status_token", None) + self._git_status_token = None + if git_status_token is not None: + reset_git_status_index(git_status_token) + + @property + def prompt_history_store(self) -> PromptHistoryStore: + """Expose this session's history store to the shell slash command.""" + return self._history_store def _get_placeholder_manager(self) -> PromptPlaceholderManager: manager = getattr(self, "_placeholder_manager", None) @@ -3439,7 +3242,12 @@ def _try_paste_media(self, event: KeyPressEvent) -> bool: Returns True if any media content was inserted. """ try: - result = grab_media_from_clipboard() + clipboard_adapter = getattr(self, "_clipboard_adapter", None) + result = ( + clipboard_adapter.paste_media() + if clipboard_adapter is not None + else grab_media_from_clipboard() + ) except Exception: # ImageGrab.grabclipboard() may fail on headless Linux if the # real xclip cannot connect to an X server. Silently ignore so @@ -3774,16 +3582,14 @@ def _append_history_entry(self, text: str) -> None: if entry.content == self._last_history_content: return + history_store = getattr(self, "_history_store", None) + if not isinstance(history_store, PromptHistoryStore): + history_store = PromptHistoryStore(self._history_file) + self._history_store = history_store try: - self._history_file.parent.mkdir(parents=True, exist_ok=True) - _ensure_private_history_path(self._history_file) - fd = os.open(self._history_file, os.O_APPEND | os.O_CREAT | os.O_WRONLY, 0o600) - with os.fdopen(fd, "a", encoding="utf-8") as f: - f.write(entry.model_dump_json(ensure_ascii=False) + "\n") - with contextlib.suppress(OSError): - os.chmod(self._history_file, 0o600) - self._last_history_content = entry.content - except OSError as exc: + if history_store.append(entry.content): + self._last_history_content = entry.content + except PromptHistoryError as exc: logger.warning( "Failed to append user history entry: {file} ({error})", file=self._history_file, @@ -3817,6 +3623,32 @@ def _append_update_notice(self, fragments: list[tuple[str, str]], columns: int) style = f"fg:{tokens.warning or 'ansiyellow'} bold" fragments.extend([("", "\n"), (style, line)]) + def _prompt_git_snapshot(self, root: HostPath) -> GitSnapshot: + index = getattr(self, "_git_status_index", None) + if isinstance(index, GitStatusIndex): + snapshot = index.snapshot(root) + index.request_refresh(root) + return snapshot + branch = _get_git_branch() + dirty, ahead, behind = _get_git_status() if branch else (False, 0, 0) + diffstat = _get_git_diffstat() + added, removed = diffstat if diffstat is not None else (0, 0) + return GitSnapshot( + root=root.canonical(), + branch=branch, + dirty=dirty, + ahead=ahead, + behind=behind, + added=added, + removed=removed, + ) + + def _prompt_toast(self, position: Literal["left", "right"]) -> ToastSnapshot | None: + manager = getattr(self, "_toast_manager", None) + if isinstance(manager, ToastManager): + return manager.current(position) + return _current_toast(position) + def _fit_toolbar_to_terminal(self, fragments: FormattedText, columns: int) -> FormattedText: app = get_app_or_none() size = app.output.get_size() if app is not None else None @@ -3894,7 +3726,8 @@ def _render_bottom_toolbar(self) -> FormattedText: # CWD (truncated from left) + git branch with status badge # Degrade gracefully on narrow terminals: full → cwd-only → truncated cwd → skip try: - cwd = _truncate_left(_shorten_cwd(str(HostPath.cwd())), _MAX_CWD_COLS) + git_root = HostPath.cwd() + cwd = _truncate_left(_shorten_cwd(str(git_root)), _MAX_CWD_COLS) except OSError: # CWD no longer exists (e.g. external drive unplugged). Ask # prompt_toolkit to exit; the raised exception will propagate out @@ -3902,11 +3735,16 @@ def _render_bottom_toolbar(self) -> FormattedText: # crash report with session info and exits cleanly. app.exit(exception=CwdLostError()) return FormattedText([]) - branch = _get_git_branch() + git_snapshot = self._prompt_git_snapshot(git_root) + branch = git_snapshot.branch if branch: - dirty, ahead, behind = _get_git_status() branch = _truncate_right(branch, _MAX_BRANCH_COLS) - badge = _format_git_badge(branch, dirty, ahead, behind) + badge = _format_git_badge( + branch, + git_snapshot.dirty, + git_snapshot.ahead, + git_snapshot.behind, + ) cwd_text = f"{cwd} {badge}" else: cwd_text = cwd @@ -3957,7 +3795,7 @@ def _render_bottom_toolbar(self) -> FormattedText: right_text = self._render_right_span(status) right_width = _display_width(right_text) - left_toast = _current_toast("left") + left_toast = self._prompt_toast("left") if left_toast is not None: max_left = max(0, columns - right_width - 2) if max_left > 0: @@ -4010,22 +3848,23 @@ def _build_statusline_context(self, columns: int) -> StatusLineContext: rate_out_sampler.reset() try: - cwd_text = _truncate_left(_shorten_cwd(str(HostPath.cwd())), _MAX_CWD_COLS) + git_root = HostPath.cwd() + cwd_text = _truncate_left(_shorten_cwd(str(git_root)), _MAX_CWD_COLS) except OSError as exc: raise CwdLostError() from exc git_info: GitInfo | None = None - branch = _get_git_branch() + git_snapshot = self._prompt_git_snapshot(git_root) + branch = git_snapshot.branch if branch: - dirty, ahead, behind = _get_git_status() git_info = GitInfo( branch=_truncate_right(branch, _MAX_BRANCH_COLS), - dirty=dirty, - ahead=ahead, - behind=behind, + dirty=git_snapshot.dirty, + ahead=git_snapshot.ahead, + behind=git_snapshot.behind, ) - diff = _get_git_diffstat() + diff = git_snapshot.diffstat diff_added, diff_removed = diff if diff is not None else (None, None) thinking_effort = getattr(self, "_thinking_effort", None) @@ -4138,7 +3977,7 @@ def _render_card_bottom_toolbar(self, columns: int) -> FormattedText: fragments.append((tc.bg_tasks, bg_summary)) left_width = _display_width(bg_summary) else: - left_toast = _current_toast("left") + left_toast = self._prompt_toast("left") if left_toast is not None: left_text = _truncate_right(left_toast.message, max_left_width) fragments.append((left_toast.style or secondary_style, left_text)) @@ -4169,13 +4008,12 @@ def _get_one_rotating_tip(self) -> str | None: return None return self._tips[self._tip_rotation_index % len(self._tips)] - @staticmethod - def _render_right_span(status: StatusSnapshot) -> str: - current_toast = _current_toast("right") - if current_toast is None: + def _render_right_span(self, status: StatusSnapshot) -> str: + right_toast = self._prompt_toast("right") + if right_toast is None: return format_context_status( status.context_usage, status.context_tokens, status.max_context_tokens, ) - return current_toast.message + return right_toast.message diff --git a/src/pythinker_code/ui/shell/prompting/__init__.py b/src/pythinker_code/ui/shell/prompting/__init__.py index f62b473c..c6718807 100644 --- a/src/pythinker_code/ui/shell/prompting/__init__.py +++ b/src/pythinker_code/ui/shell/prompting/__init__.py @@ -1,23 +1,39 @@ -"""Immutable prompt-render frame capture.""" +"""Deep prompt rendering and session-resource modules.""" +from pythinker_code.ui.shell.prompting.clipboard import ClipboardAdapter from pythinker_code.ui.shell.prompting.frame import ( FrozenFragments, PromptFrame, PromptFrameCollector, freeze_fragments, ) +from pythinker_code.ui.shell.prompting.git_status import GitSnapshot, GitStatusIndex +from pythinker_code.ui.shell.prompting.history import ( + PromptHistoryError, + PromptHistoryStatus, + PromptHistoryStore, +) from pythinker_code.ui.shell.prompting.renderer import ( PromptSceneAllocation, PromptSceneBudget, allocate_prompt_scene_rows, ) +from pythinker_code.ui.shell.prompting.toasts import ToastManager, ToastSnapshot __all__ = ( + "ClipboardAdapter", "FrozenFragments", + "GitSnapshot", + "GitStatusIndex", "PromptFrame", "PromptFrameCollector", + "PromptHistoryError", + "PromptHistoryStatus", + "PromptHistoryStore", "PromptSceneAllocation", "PromptSceneBudget", + "ToastManager", + "ToastSnapshot", "allocate_prompt_scene_rows", "freeze_fragments", ) diff --git a/src/pythinker_code/ui/shell/prompting/clipboard.py b/src/pythinker_code/ui/shell/prompting/clipboard.py new file mode 100644 index 00000000..f6d96c9d --- /dev/null +++ b/src/pythinker_code/ui/shell/prompting/clipboard.py @@ -0,0 +1,111 @@ +"""Session-owned adapter for prompt clipboard capabilities and paste operations.""" + +from __future__ import annotations + +from collections.abc import Callable +from contextvars import ContextVar, Token + +from prompt_toolkit.clipboard import Clipboard +from prompt_toolkit.clipboard.pyperclip import PyperclipClipboard + +from pythinker_code.utils.clipboard import ( + ClipboardResult, +) +from pythinker_code.utils.clipboard import ( + grab_media_from_clipboard as utility_grab_media_from_clipboard, +) +from pythinker_code.utils.clipboard import ( + is_clipboard_available as utility_is_clipboard_available, +) +from pythinker_code.utils.clipboard import ( + is_media_clipboard_available as utility_is_media_clipboard_available, +) + + +class ClipboardAdapter: + """Wrap clipboard probing and paste operations for one prompt session.""" + + def __init__( + self, + *, + text_capability: Callable[[], bool] = utility_is_clipboard_available, + media_capability: Callable[[], bool] = utility_is_media_clipboard_available, + media_grabber: Callable[[], ClipboardResult | None] = utility_grab_media_from_clipboard, + text_clipboard_factory: Callable[[], Clipboard] = PyperclipClipboard, + ) -> None: + self._text_capability = text_capability + self._media_capability = media_capability + self._media_grabber = media_grabber + self._text_clipboard_factory = text_clipboard_factory + + def is_text_available(self) -> bool: + return self._text_capability() + + def is_media_available(self) -> bool: + return self._media_capability() + + def create_text_clipboard(self, *, available: bool | None = None) -> Clipboard | None: + """Create the prompt-toolkit clipboard only after a successful probe.""" + if not (self.is_text_available() if available is None else available): + return None + return self._text_clipboard_factory() + + def paste_text(self, clipboard: Clipboard) -> str | None: + """Return pasted text, preserving the prompt's silent failure behavior.""" + try: + data = clipboard.get_data() + except Exception: + return None + return data.text + + def paste_media(self) -> ClipboardResult | None: + """Read images and file paths from the platform clipboard once.""" + return self._media_grabber() + + async def aclose(self) -> None: + """Close hook for uniform prompt lifecycle ownership.""" + + +_active_adapter: ContextVar[ClipboardAdapter | None] = ContextVar( + "prompt_clipboard_adapter", + default=None, +) + + +def bind_clipboard_adapter(adapter: ClipboardAdapter) -> Token[ClipboardAdapter | None]: + """Bind ``adapter`` to the current prompt-session context.""" + return _active_adapter.set(adapter) + + +def reset_clipboard_adapter(token: Token[ClipboardAdapter | None]) -> None: + """Restore the clipboard binding that preceded ``token``.""" + _active_adapter.reset(token) + + +def is_clipboard_available() -> bool: + adapter = _active_adapter.get() + return adapter.is_text_available() if adapter is not None else utility_is_clipboard_available() + + +def is_media_clipboard_available() -> bool: + adapter = _active_adapter.get() + return ( + adapter.is_media_available() + if adapter is not None + else utility_is_media_clipboard_available() + ) + + +def grab_media_from_clipboard() -> ClipboardResult | None: + adapter = _active_adapter.get() + return adapter.paste_media() if adapter is not None else utility_grab_media_from_clipboard() + + +__all__ = ( + "ClipboardAdapter", + "bind_clipboard_adapter", + "grab_media_from_clipboard", + "is_clipboard_available", + "is_media_clipboard_available", + "reset_clipboard_adapter", +) diff --git a/src/pythinker_code/ui/shell/prompting/git_status.py b/src/pythinker_code/ui/shell/prompting/git_status.py new file mode 100644 index 00000000..fff68be6 --- /dev/null +++ b/src/pythinker_code/ui/shell/prompting/git_status.py @@ -0,0 +1,407 @@ +"""Session-owned, non-blocking Git status snapshots for the prompt footer.""" + +from __future__ import annotations + +import asyncio +import re +import time +from collections.abc import Callable +from contextvars import ContextVar, Token +from dataclasses import dataclass +from typing import Literal + +from pythinker_host import AsyncReadable, Host, HostProcess +from pythinker_host.path import HostPath + +from pythinker_code.ui.shell.prompting.lifecycle import PromptLifecycle +from pythinker_code.utils.logging import logger + +_GIT_TIMEOUT_SECONDS = 5.0 +_MAX_GIT_OUTPUT_BYTES = 2 * 1024 * 1024 +_REFRESH_INTERVAL_SECONDS = 5.0 +_AHEAD_BEHIND_RE = re.compile(r"\[(?:ahead (\d+))?(?:, )?(?:behind (\d+))?\]") +_INSERTIONS_RE = re.compile(r"(\d+) insertion") +_DELETIONS_RE = re.compile(r"(\d+) deletion") + +type GitDegradedReason = Literal[ + "detached-head", + "not-a-repository", + "command-failed", + "timeout", + "output-limit", +] + + +@dataclass(frozen=True, slots=True) +class GitSnapshot: + """Immutable Git information belonging to exactly one requested root.""" + + root: HostPath + repository_root: HostPath | None = None + branch: str | None = None + dirty: bool = False + ahead: int = 0 + behind: int = 0 + added: int = 0 + removed: int = 0 + refreshed_at: float = 0.0 + degraded: bool = False + degraded_reason: GitDegradedReason | None = None + + @property + def diffstat(self) -> tuple[int, int] | None: + if self.added == 0 and self.removed == 0: + return None + return self.added, self.removed + + +@dataclass(frozen=True, slots=True) +class _BranchCache: + branch: str | None + + +@dataclass(frozen=True, slots=True) +class _StatusCache: + dirty: bool + ahead: int + behind: int + + +@dataclass(frozen=True, slots=True) +class _DiffCache: + added: int + removed: int + + +@dataclass(frozen=True, slots=True) +class _CommandFailure: + reason: GitDegradedReason + + +class GitStatusIndex: + """Own asynchronous Git refreshes and root-isolated immutable caches.""" + + def __init__( + self, + host: Host, + lifecycle: PromptLifecycle, + *, + refresh_interval: float = _REFRESH_INTERVAL_SECONDS, + on_publish: Callable[[], None] | None = None, + ) -> None: + self._host = host + self._lifecycle = lifecycle + self._refresh_interval = max(0.0, refresh_interval) + self._on_publish = on_publish + self._active_root_key: str | None = None + self._generation = 0 + self._aliases: dict[str, str] = {} + self._repository_roots: dict[str, HostPath | None] = {} + self._branches: dict[str, _BranchCache] = {} + self._statuses: dict[str, _StatusCache] = {} + self._diffs: dict[str, _DiffCache] = {} + self._refreshed_at: dict[str, float] = {} + self._degraded: dict[str, GitDegradedReason | None] = {} + self._refresh_task: asyncio.Task[None] | None = None + self._tasks: set[asyncio.Task[None]] = set() + self._closed = False + + def snapshot(self, root: HostPath) -> GitSnapshot: + """Read cached data for ``root`` synchronously, without filesystem I/O.""" + canonical_root = root.canonical() + root_key = str(canonical_root) + cache_key = self._aliases.get(root_key, root_key) + branch = self._branches.get(cache_key) + status = self._statuses.get(cache_key) + diff = self._diffs.get(cache_key) + reason = self._degraded.get(root_key) + return GitSnapshot( + root=canonical_root, + repository_root=self._repository_roots.get(root_key), + branch=branch.branch if branch is not None else None, + dirty=status.dirty if status is not None else False, + ahead=status.ahead if status is not None else 0, + behind=status.behind if status is not None else 0, + added=diff.added if diff is not None else 0, + removed=diff.removed if diff is not None else 0, + refreshed_at=self._refreshed_at.get(root_key, 0.0), + degraded=reason is not None, + degraded_reason=reason, + ) + + def request_refresh(self, root: HostPath) -> None: + """Schedule a bounded refresh for ``root`` if its snapshot is stale.""" + if self._closed: + return + canonical_root = root.canonical() + root_key = str(canonical_root) + if root_key != self._active_root_key: + self._generation += 1 + self._active_root_key = root_key + if self._refresh_task is not None and not self._refresh_task.done(): + self._refresh_task.cancel() + self._refresh_task = None + + refreshed_at = self._refreshed_at.get(root_key, 0.0) + if time.monotonic() - refreshed_at <= self._refresh_interval: + return + if self._refresh_task is not None and not self._refresh_task.done(): + return + + refresh = self._refresh(canonical_root, root_key, self._generation) + try: + task = self._lifecycle.create_task(refresh) + except RuntimeError as exc: + logger.debug("Git refresh was not scheduled: error={!r}", exc) + return + self._refresh_task = task + self._tasks.add(task) + task.add_done_callback(self._tasks.discard) + + async def aclose(self) -> None: + """Cancel and await all Git work owned by this index.""" + if self._closed: + return + self._closed = True + tasks = tuple(self._tasks) + for task in tasks: + if not task.done(): + task.cancel() + if tasks: + results = await asyncio.gather(*tasks, return_exceptions=True) + for result in results: + if isinstance(result, BaseException) and not isinstance( + result, asyncio.CancelledError + ): + logger.warning("Git refresh failed during shutdown: error={!r}", result) + self._refresh_task = None + + async def _refresh(self, root: HostPath, root_key: str, generation: int) -> None: + repository = await self._run_git(root, "rev-parse", "--show-toplevel") + if isinstance(repository, _CommandFailure): + self._publish_degraded(root, root_key, generation, repository.reason) + return + + repository_root = HostPath(repository.strip()).canonical() + repository_key = str(repository_root) + branch = await self._run_git(repository_root, "symbolic-ref", "--quiet", "--short", "HEAD") + if isinstance(branch, _CommandFailure): + self._publish_degraded(root, root_key, generation, branch.reason) + return + if not branch.strip(): + self._publish_degraded(root, root_key, generation, "detached-head") + return + + status_output = await self._run_git( + repository_root, + "-c", + "core.quotepath=false", + "status", + "--porcelain", + "-b", + ) + if isinstance(status_output, _CommandFailure): + self._publish_degraded(root, root_key, generation, status_output.reason) + return + diff_output = await self._run_git( + repository_root, + "--no-optional-locks", + "diff", + "--shortstat", + ) + if isinstance(diff_output, _CommandFailure): + self._publish_degraded(root, root_key, generation, diff_output.reason) + return + + dirty, ahead, behind = self._parse_status(status_output) + added, removed = self._parse_diffstat(diff_output) + if not self._can_publish(root_key, generation): + return + self._aliases[root_key] = repository_key + self._repository_roots[root_key] = repository_root + self._branches[repository_key] = _BranchCache(branch=branch.strip()) + self._statuses[repository_key] = _StatusCache( + dirty=dirty, + ahead=ahead, + behind=behind, + ) + self._diffs[repository_key] = _DiffCache(added=added, removed=removed) + self._refreshed_at[root_key] = time.monotonic() + self._degraded[root_key] = None + self._notify_publish() + + def _publish_degraded( + self, + root: HostPath, + root_key: str, + generation: int, + reason: GitDegradedReason, + ) -> None: + if not self._can_publish(root_key, generation): + return + self._aliases[root_key] = root_key + self._repository_roots[root_key] = None + self._branches[root_key] = _BranchCache(branch=None) + self._statuses[root_key] = _StatusCache(dirty=False, ahead=0, behind=0) + self._diffs[root_key] = _DiffCache(added=0, removed=0) + self._refreshed_at[root_key] = time.monotonic() + self._degraded[root_key] = reason + logger.debug("Git status is degraded: root={} reason={}", root, reason) + self._notify_publish() + + def _can_publish(self, root_key: str, generation: int) -> bool: + return ( + not self._closed + and generation == self._generation + and root_key == self._active_root_key + ) + + def _notify_publish(self) -> None: + if self._on_publish is None: + return + try: + self._on_publish() + except Exception as exc: + logger.warning("Git publish notification failed: error={!r}", exc) + + @staticmethod + def _parse_status(output: str) -> tuple[bool, int, int]: + dirty = False + ahead = 0 + behind = 0 + for line in output.splitlines(): + if line.startswith("## "): + match = _AHEAD_BEHIND_RE.search(line) + if match is not None: + ahead = int(match.group(1) or 0) + behind = int(match.group(2) or 0) + elif line.strip(): + dirty = True + return dirty, ahead, behind + + @staticmethod + def _parse_diffstat(output: str) -> tuple[int, int]: + insertions = _INSERTIONS_RE.search(output) + deletions = _DELETIONS_RE.search(output) + return ( + int(insertions.group(1)) if insertions is not None else 0, + int(deletions.group(1)) if deletions is not None else 0, + ) + + async def _run_git(self, root: HostPath, *args: str) -> str | _CommandFailure: + process: HostProcess | None = None + try: + async with asyncio.timeout(_GIT_TIMEOUT_SECONDS): + process = await self._host.exec("git", *args, cwd=str(root)) + stdout = await self._read_streams(process) + returncode = await process.wait() + except asyncio.CancelledError: + if process is not None: + await self._stop_process(process) + raise + except TimeoutError: + if process is not None: + await self._stop_process(process) + logger.warning("Git command timed out: root={} command={!r}", root, args) + return _CommandFailure("timeout") + except _OutputLimitError: + if process is not None: + await self._stop_process(process) + logger.warning("Git command exceeded output limit: root={} command={!r}", root, args) + return _CommandFailure("output-limit") + except Exception as exc: + if process is not None: + await self._stop_process(process) + logger.warning( + "Git command could not run: root={} command={!r} error={!r}", + root, + args, + exc, + ) + return _CommandFailure("command-failed") + + if returncode != 0: + if args == ("rev-parse", "--show-toplevel"): + return _CommandFailure("not-a-repository") + if args == ("symbolic-ref", "--quiet", "--short", "HEAD") and returncode == 1: + return _CommandFailure("detached-head") + return _CommandFailure("command-failed") + encoding = "utf-8" + return stdout.decode(encoding=encoding, errors="replace") + + async def _read_streams(self, process: HostProcess) -> bytes: + stdout_reader = asyncio.create_task(self._read_bounded(process.stdout, keep=True)) + stderr_reader = asyncio.create_task(self._read_bounded(process.stderr, keep=False)) + try: + stdout, _stderr = await asyncio.gather(stdout_reader, stderr_reader) + except BaseException: + for reader in (stdout_reader, stderr_reader): + reader.cancel() + await asyncio.gather(stdout_reader, stderr_reader, return_exceptions=True) + raise + return stdout + + @staticmethod + async def _read_bounded(stream: AsyncReadable, *, keep: bool) -> bytes: + chunks = bytearray() + total = 0 + while True: + chunk = await stream.read(65536) + if not chunk: + return bytes(chunks) + total += len(chunk) + if total > _MAX_GIT_OUTPUT_BYTES: + raise _OutputLimitError + if keep: + chunks.extend(chunk) + + @staticmethod + async def _stop_process(process: HostProcess) -> None: + try: + await process.kill() + except Exception as exc: + logger.warning("Git process could not be killed: error={!r}", exc) + try: + await process.wait() + except Exception as exc: + logger.warning("Git process cleanup failed: error={!r}", exc) + + +class _OutputLimitError(RuntimeError): + pass + + +_active_index: ContextVar[GitStatusIndex | None] = ContextVar( + "prompt_git_status_index", + default=None, +) + + +def bind_git_status_index(index: GitStatusIndex) -> Token[GitStatusIndex | None]: + """Bind ``index`` to the current prompt-session context.""" + return _active_index.set(index) + + +def reset_git_status_index(token: Token[GitStatusIndex | None]) -> None: + """Restore the Git status binding that preceded ``token``.""" + _active_index.reset(token) + + +def current_git_snapshot(root: HostPath | None = None) -> GitSnapshot: + """Compatibility facade for synchronous access to the active session index.""" + requested_root = (root or HostPath.cwd()).canonical() + index = _active_index.get() + if index is None: + return GitSnapshot(root=requested_root) + snapshot = index.snapshot(requested_root) + index.request_refresh(requested_root) + return snapshot + + +__all__ = ( + "GitSnapshot", + "GitStatusIndex", + "bind_git_status_index", + "current_git_snapshot", + "reset_git_status_index", +) diff --git a/src/pythinker_code/ui/shell/prompting/history.py b/src/pythinker_code/ui/shell/prompting/history.py new file mode 100644 index 00000000..3cd09be6 --- /dev/null +++ b/src/pythinker_code/ui/shell/prompting/history.py @@ -0,0 +1,276 @@ +"""Bounded, locked persistence for prompt input history.""" + +from __future__ import annotations + +import contextlib +import json +import os +import re +from collections.abc import Collection +from dataclasses import dataclass +from pathlib import Path +from typing import Any, cast + +from pydantic import BaseModel, ValidationError + +from pythinker_code.utils.io import file_lock +from pythinker_code.utils.logging import logger + +_DEFAULT_MAX_ENTRIES = 1000 +_MAX_SERIALIZED_RECORD_BYTES = 256 * 1024 +_ROTATE_AT_BYTES = 10 * 1024 * 1024 +_LOCK_TIMEOUT_SECONDS = 1.0 +_DEFAULT_SENSITIVE_COMMANDS = frozenset({"login", "logout", "setup"}) + +_HISTORY_SECRET_PATTERNS: tuple[tuple[re.Pattern[str], str], ...] = ( + ( + re.compile(r"(?i)\b((?:authorization\s*:\s*)?(?:bearer|basic)\s+)[A-Za-z0-9._~+/=-]{8,}"), + r"\1[REDACTED]", + ), + ( + re.compile( + r"(?i)([\"']?(?:api[_-]?key|token|secret|password|access[_-]?token|" + r"refresh[_-]?token|id[_-]?token|session[_-]?token)[\"']?\s*[:=]\s*[\"'])" + r"([^\"'\r\n]{8,})([\"'])" + ), + r"\1[REDACTED]\3", + ), + ( + re.compile( + r"(?i)\b(api[_-]?key|token|secret|password|access[_-]?token|" + r"refresh[_-]?token|id[_-]?token|session[_-]?token)(\s*[:=]\s*)([^\s'\"&]{8,})" + ), + r"\1\2[REDACTED]", + ), + (re.compile(r"\b(sk-[A-Za-z0-9][A-Za-z0-9_-]{16,})\b"), "[REDACTED]"), + (re.compile(r"\b(?:gh[opusr]_[A-Za-z0-9_]{20,}|github_pat_[A-Za-z0-9_]{20,})\b"), "[REDACTED]"), + (re.compile(r"\b(AKIA[0-9A-Z]{16})\b"), "[REDACTED]"), + (re.compile(r"\b(AIza[0-9A-Za-z_-]{20,})\b"), "[REDACTED]"), +) + + +class HistoryEntry(BaseModel): + content: str + + +@dataclass(frozen=True, slots=True) +class PromptHistoryStatus: + """Current prompt-history paths, counts, and storage sizes.""" + + enabled: bool + entries: int + path: Path + rotated_path: Path + size_bytes: int + rotated_size_bytes: int + + +class PromptHistoryError(RuntimeError): + """A locked prompt-history mutation could not be completed.""" + + +class PromptHistoryStore: + """Own one workspace's JSONL prompt history.""" + + def __init__( + self, + path: Path, + *, + enabled: bool = True, + max_entries: int = _DEFAULT_MAX_ENTRIES, + sensitive_commands: Collection[str] = _DEFAULT_SENSITIVE_COMMANDS, + ) -> None: + self.path = path + self.rotated_path = path.with_name(path.name + ".1") + self.enabled = enabled + self.max_entries = max(1, max_entries) + self.sensitive_commands = frozenset(command.lower() for command in sensitive_commands) + self._last_content: str | None = None + if enabled: + ensure_private_history_path(path) + + def load(self) -> list[HistoryEntry]: + """Load at most ``max_entries`` recent records using tail reads.""" + if not self.enabled: + return [] + current_lines = self._tail_lines(self.path, self.max_entries) + remaining = self.max_entries - len(current_lines) + rotated_lines = self._tail_lines(self.rotated_path, remaining) if remaining > 0 else [] + entries: list[HistoryEntry] = [] + for source, lines in ( + (self.rotated_path, rotated_lines), + (self.path, current_lines), + ): + entries.extend(self._parse_lines(source, lines)) + entries = entries[-self.max_entries :] + self._last_content = entries[-1].content if entries else None + return entries + + def append(self, text: str) -> bool: + """Append a redacted record; return true only when it was persisted.""" + if not self.enabled: + return False + content = redact_history_secrets(text.strip()) + if not content or self._is_sensitive_command(content) or content == self._last_content: + return False + entry = HistoryEntry(content=content) + serialized = entry.model_dump_json(ensure_ascii=False) + "\n" + encoding = "utf-8" + serialized_bytes = serialized.encode(encoding=encoding, errors="replace") + if len(serialized_bytes) > _MAX_SERIALIZED_RECORD_BYTES: + logger.warning( + "Prompt history record exceeds the size limit; skipping: bytes={}", + len(serialized_bytes), + ) + return False + + try: + self.path.parent.mkdir(parents=True, exist_ok=True) + with file_lock(self.path, timeout=_LOCK_TIMEOUT_SECONDS): + self._rotate_if_needed(len(serialized_bytes)) + ensure_private_history_path(self.path) + fd = os.open(self.path, os.O_APPEND | os.O_CREAT | os.O_WRONLY, 0o600) + with os.fdopen(fd, "a", encoding=encoding) as stream: + stream.write(serialized) + with contextlib.suppress(OSError): + os.chmod(self.path, 0o600) + self._last_content = content + return True + except OSError as exc: + message = f"Could not append prompt history at {self.path}: {exc}" + raise PromptHistoryError(message) from exc + + def clear(self) -> PromptHistoryStatus: + """Remove current and rotated records under the store lock.""" + try: + with file_lock(self.path, timeout=_LOCK_TIMEOUT_SECONDS): + for candidate in (self.path, self.rotated_path): + with contextlib.suppress(FileNotFoundError): + candidate.unlink() + if self.path.exists() or self.rotated_path.exists(): + raise OSError("history files still exist after removal") + self._last_content = None + except OSError as exc: + message = f"Could not clear prompt history at {self.path}: {exc}" + raise PromptHistoryError(message) from exc + return self.status() + + def status(self) -> PromptHistoryStatus: + """Report the current paths, retained count, and file sizes.""" + entries = len(self.load()) if self.enabled else 0 + return PromptHistoryStatus( + enabled=self.enabled, + entries=entries, + path=self.path, + rotated_path=self.rotated_path, + size_bytes=self._file_size(self.path), + rotated_size_bytes=self._file_size(self.rotated_path), + ) + + async def aclose(self) -> None: + """Close hook for uniform prompt lifecycle ownership.""" + + def _rotate_if_needed(self, incoming_bytes: int) -> None: + if self._file_size(self.path) + incoming_bytes <= _ROTATE_AT_BYTES: + return + if self.path.exists(): + os.replace(self.path, self.rotated_path) + with contextlib.suppress(OSError): + os.chmod(self.rotated_path, 0o600) + + def _is_sensitive_command(self, content: str) -> bool: + command = content.lstrip().removeprefix("/").split(maxsplit=1)[0].lower() + return command in self.sensitive_commands + + @staticmethod + def _file_size(path: Path) -> int: + try: + return path.stat().st_size + except FileNotFoundError: + return 0 + except OSError as exc: + logger.warning("Failed to inspect prompt history file: file={} error={!r}", path, exc) + return 0 + + @staticmethod + def _tail_lines(path: Path, limit: int) -> list[tuple[int, bytes]]: + if limit <= 0: + return [] + try: + with path.open("rb") as stream: + stream.seek(0, os.SEEK_END) + position = stream.tell() + chunks: list[bytes] = [] + newline_count = 0 + while position > 0 and newline_count <= limit: + read_size = min(64 * 1024, position) + position -= read_size + stream.seek(position) + chunk = stream.read(read_size) + chunks.append(chunk) + newline_count += chunk.count(b"\n") + data = b"".join(reversed(chunks)) + if position > 0: + first_newline = data.find(b"\n") + data = data[first_newline + 1 :] if first_newline >= 0 else b"" + lines = data.splitlines()[-limit:] + total = len(lines) + return [(-(total - index), line) for index, line in enumerate(lines)] + except FileNotFoundError: + return [] + except OSError as exc: + logger.warning("Failed to load prompt history file: file={} error={!r}", path, exc) + return [] + + @staticmethod + def _parse_lines(path: Path, lines: list[tuple[int, bytes]]) -> list[HistoryEntry]: + entries: list[HistoryEntry] = [] + encoding = "utf-8" + for line_number, raw_line in lines: + if not raw_line.strip() or len(raw_line) > _MAX_SERIALIZED_RECORD_BYTES: + continue + try: + decoded = raw_line.decode(encoding=encoding, errors="replace") + loaded = json.loads(decoded) + if not isinstance(loaded, dict): + raise ValueError("record is not an object") + record = cast(dict[str, Any], loaded) + entries.append(HistoryEntry.model_validate(record)) + except (ValidationError, ValueError): + logger.warning( + "Failed to parse prompt history record; skipping: file={} line_from_end={}", + path, + line_number, + ) + return entries + + +def redact_history_secrets(text: str) -> str: + redacted = text + for pattern, replacement in _HISTORY_SECRET_PATTERNS: + redacted = pattern.sub(replacement, redacted) + return redacted + + +def ensure_private_history_path(path: Path) -> None: + with contextlib.suppress(OSError): + os.chmod(path.parent, 0o700) + if path.exists(): + with contextlib.suppress(OSError): + os.chmod(path, 0o600) + + +def load_history_entries(history_file: Path) -> list[HistoryEntry]: + """Compatibility facade for loading one prompt-history path.""" + return PromptHistoryStore(history_file).load() + + +__all__ = ( + "HistoryEntry", + "PromptHistoryError", + "PromptHistoryStatus", + "PromptHistoryStore", + "ensure_private_history_path", + "load_history_entries", + "redact_history_secrets", +) diff --git a/src/pythinker_code/ui/shell/prompting/toasts.py b/src/pythinker_code/ui/shell/prompting/toasts.py new file mode 100644 index 00000000..0ace6d19 --- /dev/null +++ b/src/pythinker_code/ui/shell/prompting/toasts.py @@ -0,0 +1,182 @@ +"""Session-owned toast queues with a compatibility facade.""" + +from __future__ import annotations + +import time +from collections import deque +from contextvars import ContextVar, Token +from dataclasses import dataclass +from typing import Literal + +ToastPosition = Literal["left", "right"] + +_MINIMUM_DURATION_SECONDS = 1.0 +_MAX_BOOTSTRAP_TOASTS = 100 + + +@dataclass(frozen=True, slots=True) +class ToastSnapshot: + """One immutable toast entry owned by a prompt session.""" + + message: str + position: ToastPosition + style: str + topic: str | None + expires_at: float + + +class ToastManager: + """Own, expire, and isolate toast entries for one prompt session.""" + + def __init__(self, *, adopt_bootstrap: bool = True) -> None: + self._queues: dict[ToastPosition, deque[ToastSnapshot]] = { + "left": deque(), + "right": deque(), + } + self._closed = False + if adopt_bootstrap: + self._adopt_bootstrap() + + def add( + self, + message: str, + *, + duration: float = 5.0, + topic: str | None = None, + immediate: bool = False, + position: ToastPosition = "left", + style: str = "", + ) -> None: + if self._closed: + return + entry = ToastSnapshot( + message=message, + position=position, + style=style, + topic=topic, + expires_at=time.monotonic() + max(duration, _MINIMUM_DURATION_SECONDS), + ) + self._add_entry(entry, immediate=immediate) + + def current(self, position: ToastPosition = "left") -> ToastSnapshot | None: + """Return the current unexpired toast for ``position``.""" + queue = self._queues[position] + now = time.monotonic() + while queue and queue[0].expires_at <= now: + queue.popleft() + return queue[0] if queue else None + + def snapshots(self, position: ToastPosition) -> tuple[ToastSnapshot, ...]: + """Return all unexpired entries in display order.""" + self.current(position) + return tuple(self._queues[position]) + + async def aclose(self) -> None: + """Close this manager without affecting another session's entries.""" + if self._closed: + return + self._closed = True + for queue in self._queues.values(): + queue.clear() + + def _add_entry(self, entry: ToastSnapshot, *, immediate: bool) -> None: + queue = self._queues[entry.position] + if entry.topic is not None: + self._queues[entry.position] = queue = deque( + existing for existing in queue if existing.topic != entry.topic + ) + if immediate: + queue.appendleft(entry) + else: + queue.append(entry) + + def _adopt_bootstrap(self) -> None: + for queue in bootstrap_toast_queues.values(): + while queue: + self._add_entry(queue.popleft(), immediate=False) + + +bootstrap_toast_queues: dict[ToastPosition, deque[ToastSnapshot]] = { + "left": deque(maxlen=_MAX_BOOTSTRAP_TOASTS), + "right": deque(maxlen=_MAX_BOOTSTRAP_TOASTS), +} + +_active_manager: ContextVar[ToastManager | None] = ContextVar( + "prompt_toast_manager", + default=None, +) + + +def bind_toast_manager(manager: ToastManager) -> Token[ToastManager | None]: + """Bind ``manager`` to the current prompt-session context.""" + return _active_manager.set(manager) + + +def reset_toast_manager(token: Token[ToastManager | None]) -> None: + """Restore the toast binding that preceded ``token``.""" + _active_manager.reset(token) + + +def toast( + message: str, + duration: float = 5.0, + topic: str | None = None, + immediate: bool = False, + position: ToastPosition = "left", + style: str = "", +) -> None: + """Route a toast to the bound session, or retain it for the next session.""" + manager = _active_manager.get() + if manager is not None: + manager.add( + message, + duration=duration, + topic=topic, + immediate=immediate, + position=position, + style=style, + ) + return + + queue = bootstrap_toast_queues[position] + if topic is not None: + retained = (entry for entry in queue if entry.topic != topic) + bootstrap_toast_queues[position] = queue = deque( + retained, + maxlen=_MAX_BOOTSTRAP_TOASTS, + ) + entry = ToastSnapshot( + message=message, + position=position, + style=style, + topic=topic, + expires_at=time.monotonic() + max(duration, _MINIMUM_DURATION_SECONDS), + ) + if immediate: + queue.appendleft(entry) + else: + queue.append(entry) + + +def current_toast(position: ToastPosition = "left") -> ToastSnapshot | None: + """Read the current toast from the active session or bootstrap queue.""" + manager = _active_manager.get() + if manager is not None: + return manager.current(position) + queue = bootstrap_toast_queues[position] + now = time.monotonic() + while queue and queue[0].expires_at <= now: + queue.popleft() + return queue[0] if queue else None + + +__all__ = ( + "ToastManager", + "ToastPosition", + "ToastSnapshot", + "bind_toast_manager", + "bootstrap_toast_queues", + "current_toast", + "reset_toast_manager", + "toast", +) diff --git a/src/pythinker_code/ui/shell/slash.py b/src/pythinker_code/ui/shell/slash.py index 1556660b..2e14d171 100644 --- a/src/pythinker_code/ui/shell/slash.py +++ b/src/pythinker_code/ui/shell/slash.py @@ -65,6 +65,7 @@ def exit(app: Shell, args: str): # Ordered first-token hints for slash commands with fixed subcommands (ghost text + menu). _THEME_ARGS: tuple[str, ...] = ("current", "doctor", "tokens", "code", "dark", "light", "auto") +_PROMPT_HISTORY_ARGS: tuple[str, ...] = ("status", "clear") def slash_command_arg_suggestions() -> dict[str, tuple[str, ...]]: @@ -72,6 +73,7 @@ def slash_command_arg_suggestions() -> dict[str, tuple[str, ...]]: return { "theme": _THEME_ARGS, "color": _THEME_ARGS, + "prompt-history": _PROMPT_HISTORY_ARGS, } @@ -172,6 +174,50 @@ def version(app: Shell, args: str): console.print(f"pythinker, version {VERSION}") +@registry.command(name="prompt-history", available_during_task=True) +@shell_mode_registry.command(name="prompt-history") +def prompt_history(app: Shell, args: str) -> None: + """Inspect or clear prompt history. Usage: /prompt-history status|clear""" + from pythinker_code.ui.shell.prompt import CustomPromptSession + from pythinker_code.ui.shell.prompting.history import PromptHistoryError + from pythinker_code.ui.theme import get_tui_tokens + + tokens = get_tui_tokens() + action = args.strip().lower() + if action not in _PROMPT_HISTORY_ARGS: + console.print(f"[{tokens.error}]Usage: /prompt-history status|clear[/]") + return + + prompt_session = getattr(app, "_prompt_session", None) + if not isinstance(prompt_session, CustomPromptSession): + console.print( + f"[{tokens.error}]Prompt history is unavailable because no shell session is active.[/]" + ) + return + store = prompt_session.prompt_history_store + if action == "status": + status = store.status() + state = "enabled" if status.enabled else "disabled" + total_size = status.size_bytes + status.rotated_size_bytes + console.print( + f"Prompt history: [{tokens.info}]{state}[/] · {status.entries} entries · " + f"{total_size} bytes\n" + f"[{tokens.muted}]current: {_rich_escape(status.path)}\n" + f"rotated: {_rich_escape(status.rotated_path)}[/]" + ) + return + + try: + status = store.clear() + except PromptHistoryError as exc: + console.print( + f"[{tokens.error}]Failed to clear prompt history: {_rich_escape(exc)}. " + "Close other Pythinker sessions using this workspace and try again.[/]" + ) + return + console.print(f"[{tokens.success}]Prompt history cleared: {status.entries} entries remain.[/]") + + @registry.command(available_during_task=True) @shell_mode_registry.command def agents(app: Shell, args: str): diff --git a/tests/ui_and_conv/test_prompt_history.py b/tests/ui_and_conv/test_prompt_history.py index 6e9f1da7..905c4e39 100644 --- a/tests/ui_and_conv/test_prompt_history.py +++ b/tests/ui_and_conv/test_prompt_history.py @@ -2,11 +2,16 @@ import json import stat +from types import SimpleNamespace +from typing import Any, cast from PIL import Image from pythinker_code.ui.shell import prompt as shell_prompt +from pythinker_code.ui.shell import slash as shell_slash from pythinker_code.ui.shell.placeholders import AttachmentCache, PromptPlaceholderManager +from pythinker_code.ui.shell.prompting.history import PromptHistoryError, PromptHistoryStore +from pythinker_code.ui.shell.slash import registry, slash_command_arg_suggestions def _make_prompt_session( @@ -138,3 +143,88 @@ def test_append_history_entry_restricts_file_permissions(tmp_path) -> None: mode = stat.S_IMODE(prompt_session._history_file.stat().st_mode) assert mode == 0o600 + + +def test_prompt_history_store_loads_only_configured_tail(tmp_path) -> None: + store = PromptHistoryStore(tmp_path / "history.jsonl", max_entries=2) + + assert store.append("one") + assert store.append("two") + assert store.append("three") + + assert [entry.content for entry in store.load()] == ["two", "three"] + + +def test_prompt_history_store_excludes_credential_commands(tmp_path) -> None: + store = PromptHistoryStore(tmp_path / "history.jsonl") + + assert store.append("/login api-key") is False + assert store.append("logout openai") is False + assert not store.path.exists() + + +def test_prompt_history_store_skips_oversized_records(tmp_path) -> None: + store = PromptHistoryStore(tmp_path / "history.jsonl") + + assert store.append("x" * (256 * 1024)) is False + assert not store.path.exists() + + +def test_prompt_history_store_clear_confirms_both_files_are_removed(tmp_path) -> None: + store = PromptHistoryStore(tmp_path / "history.jsonl") + assert store.append("kept") + encoding = "utf-8" + store.rotated_path.write_text('{"content":"older"}\n', encoding=encoding) + + status = store.clear() + + assert status.entries == 0 + assert not store.path.exists() + assert not store.rotated_path.exists() + + +def test_prompt_history_slash_command_has_only_supported_argument_suggestions() -> None: + command = registry.find_command("prompt-history") + + assert command is not None + assert command.name == "prompt-history" + assert slash_command_arg_suggestions()["prompt-history"] == ("status", "clear") + + +def test_prompt_history_clear_reports_success_only_after_confirmed_removal( + tmp_path, + monkeypatch, +) -> None: + store = PromptHistoryStore(tmp_path / "history.jsonl") + assert store.append("kept") + prompt_session = object.__new__(shell_prompt.CustomPromptSession) + cast(Any, prompt_session)._history_store = store + app = SimpleNamespace(_prompt_session=prompt_session) + messages: list[str] = [] + monkeypatch.setattr(shell_slash.console, "print", lambda message: messages.append(str(message))) + command = registry.find_command("prompt-history") + assert command is not None + + command.func(cast(Any, app), "clear") + + assert any("Prompt history cleared" in message for message in messages) + assert not store.path.exists() + + +def test_prompt_history_clear_failure_never_claims_success(monkeypatch) -> None: + class _FailingStore: + def clear(self) -> None: + raise PromptHistoryError("lock timed out") + + prompt_session = object.__new__(shell_prompt.CustomPromptSession) + cast(Any, prompt_session)._history_store = _FailingStore() + app = SimpleNamespace(_prompt_session=prompt_session) + messages: list[str] = [] + monkeypatch.setattr(shell_slash.console, "print", lambda message: messages.append(str(message))) + command = registry.find_command("prompt-history") + assert command is not None + + command.func(cast(Any, app), "clear") + + assert any("Failed to clear prompt history" in message for message in messages) + assert all("Prompt history cleared" not in message for message in messages) diff --git a/tests/ui_and_conv/test_prompt_resources.py b/tests/ui_and_conv/test_prompt_resources.py new file mode 100644 index 00000000..f0902d6b --- /dev/null +++ b/tests/ui_and_conv/test_prompt_resources.py @@ -0,0 +1,167 @@ +from __future__ import annotations + +import asyncio +from collections.abc import Mapping +from typing import Any, cast + +import pytest +from pythinker_host import Host, HostProcess +from pythinker_host.path import HostPath + +from pythinker_code.ui.shell.prompting.git_status import GitStatusIndex +from pythinker_code.ui.shell.prompting.lifecycle import PromptLifecycle +from pythinker_code.ui.shell.prompting.toasts import ( + ToastManager, + bind_toast_manager, + reset_toast_manager, + toast, +) + + +class _FakeProcess: + def __init__(self, stdout: str, returncode: int = 0) -> None: + self.stdout = asyncio.StreamReader() + self.stderr = asyncio.StreamReader() + encoding = "utf-8" + self.stdout.feed_data(stdout.encode(encoding=encoding)) + self.stdout.feed_eof() + self.stderr.feed_eof() + self.returncode = returncode + self.pid = 1 + + async def wait(self) -> int: + return self.returncode + + async def kill(self) -> None: + self.returncode = -9 + + +class _FakeGitHost: + name = "fake-git" + + def __init__(self, branches: Mapping[str, str]) -> None: + self.branches = branches + + async def exec( + self, + *args: str, + env: Mapping[str, str] | None = None, + cwd: str | None = None, + ) -> HostProcess: + del env + assert cwd is not None + branch = self.branches[cwd] + command = args[1:] + if command == ("rev-parse", "--show-toplevel"): + output = f"{cwd}\n" + elif command == ("symbolic-ref", "--quiet", "--short", "HEAD"): + output = f"{branch}\n" + elif command[-3:] == ("status", "--porcelain", "-b"): + output = f"## {branch}...origin/{branch} [ahead 1]\n M tracked.py\n" + elif command == ("--no-optional-locks", "diff", "--shortstat"): + output = " 1 file changed, 3 insertions(+), 1 deletion(-)\n" + else: + raise AssertionError(f"unexpected Git command: {args!r}") + return cast(HostProcess, _FakeProcess(output)) + + +async def _wait_for_snapshot(index: GitStatusIndex, root: HostPath) -> None: + for _ in range(100): + if index.snapshot(root).refreshed_at > 0: + return + await asyncio.sleep(0) + raise AssertionError(f"Git snapshot was not published for {root}") + + +@pytest.mark.asyncio +async def test_prompt_resources_are_isolated_across_sessions() -> None: + root_a = HostPath("/workspace/a") + root_b = HostPath("/workspace/b") + root_c = HostPath("/workspace/c") + host = cast( + Host, + _FakeGitHost( + { + str(root_a): "branch-a", + str(root_b): "branch-b", + str(root_c): "branch-c", + } + ), + ) + lifecycle_a = PromptLifecycle() + lifecycle_b = PromptLifecycle() + git_a = GitStatusIndex(host, lifecycle_a, refresh_interval=0) + git_b = GitStatusIndex(host, lifecycle_b, refresh_interval=0) + lifecycle_a.register_closer("Git A", git_a.aclose) + lifecycle_b.register_closer("Git B", git_b.aclose) + toasts_a = ToastManager(adopt_bootstrap=False) + toasts_b = ToastManager(adopt_bootstrap=False) + lifecycle_a.register_closer("toasts A", toasts_a.aclose) + lifecycle_b.register_closer("toasts B", toasts_b.aclose) + + git_a.request_refresh(root_a) + git_b.request_refresh(root_b) + await _wait_for_snapshot(git_a, root_a) + await _wait_for_snapshot(git_b, root_b) + + assert git_a.snapshot(root_a).branch == "branch-a" + assert git_a.snapshot(root_b).branch is None + assert git_b.snapshot(root_b).branch == "branch-b" + assert git_b.snapshot(root_a).branch is None + + token_a = bind_toast_manager(toasts_a) + toast("session A only", topic="session") + reset_toast_manager(token_a) + token_b = bind_toast_manager(toasts_b) + try: + assert toasts_b.current() is None + toast("session B only", topic="session") + finally: + reset_toast_manager(token_b) + toast_a = toasts_a.current() + toast_b = toasts_b.current() + assert toast_a is not None + assert toast_a.message == "session A only" + assert toast_b is not None + assert toast_b.message == "session B only" + + await lifecycle_a.aclose() + assert toasts_a.current() is None + assert toasts_b.current() is not None + git_b.request_refresh(root_c) + await _wait_for_snapshot(git_b, root_c) + assert git_b.snapshot(root_c).branch == "branch-c" + + await lifecycle_b.aclose() + + +@pytest.mark.asyncio +async def test_stale_git_result_cannot_replace_new_root_snapshot() -> None: + root_a = HostPath("/workspace/a") + root_b = HostPath("/workspace/b") + host = cast( + Host, + _FakeGitHost( + { + str(root_a): "branch-a", + str(root_b): "branch-b", + } + ), + ) + lifecycle = PromptLifecycle() + index = GitStatusIndex(host, lifecycle, refresh_interval=0) + lifecycle.register_closer("Git", index.aclose) + + index.request_refresh(root_a) + await _wait_for_snapshot(index, root_a) + index.request_refresh(root_b) + await _wait_for_snapshot(index, root_b) + + internal = cast(Any, index) + internal._publish_degraded(root_a, str(root_a), 1, "command-failed") + + assert index.snapshot(root_a).branch == "branch-a" + assert index.snapshot(root_a).degraded is False + assert index.snapshot(root_b).branch == "branch-b" + assert index.snapshot(root_b).degraded is False + await lifecycle.aclose() diff --git a/tests/ui_and_conv/test_prompt_tips.py b/tests/ui_and_conv/test_prompt_tips.py index 048df56e..f4d6fccc 100644 --- a/tests/ui_and_conv/test_prompt_tips.py +++ b/tests/ui_and_conv/test_prompt_tips.py @@ -2,11 +2,9 @@ import asyncio import os -import time from collections.abc import Callable from types import SimpleNamespace from typing import Any, cast -from unittest.mock import MagicMock import pytest from prompt_toolkit.completion import Completion @@ -16,7 +14,6 @@ from pythinker_code.ui.shell import prompt as shell_prompt from pythinker_code.ui.shell.glyphs import SPINNER_FRAMES from pythinker_code.ui.shell.prompt import ( - _GIT_STATUS_TTL, PROMPT_SYMBOL, BgTaskCounts, CustomPromptSession, @@ -27,10 +24,6 @@ _build_toolbar_tips, _display_width, _format_git_badge, - _get_git_branch, - _get_git_status, - _git_branch_state, - _git_status_state, _PromptRightPaddingMargin, _shorten_cwd, _toast_queues, @@ -885,69 +878,6 @@ def test_toolbar_line2_right_toast_replaces_context(monkeypatch: Any) -> None: assert "context:" not in lines[2] -# ── Fix #4 regression: branch change invalidates in-flight status subprocess ── - - -def test_git_branch_change_terminates_in_flight_status_proc(monkeypatch: Any) -> None: - """Regression: switching branches must discard any in-flight status subprocess - so stale results from the old branch are never applied to the new branch.""" - mock_branch_proc = MagicMock() - mock_branch_proc.poll.return_value = 0 # process completed - mock_branch_proc.communicate.return_value = ("feature-branch\n", "") - - mock_status_proc = MagicMock() - - # Simulate: branch proc has a result ready; status proc is still in-flight. - monkeypatch.setattr(_git_branch_state, "branch", "main") - monkeypatch.setattr(_git_branch_state, "proc", mock_branch_proc) - monkeypatch.setattr(_git_branch_state, "timestamp", float("inf")) # TTL fresh, won't re-launch - monkeypatch.setattr(_git_status_state, "proc", mock_status_proc) - monkeypatch.setattr(_git_status_state, "timestamp", float("inf")) # TTL fresh - - _get_git_branch() - - mock_status_proc.terminate.assert_called_once() - assert _git_status_state.proc is None - assert _git_status_state.timestamp == 0.0 - assert _git_branch_state.branch == "feature-branch" - - -def test_git_status_stuck_subprocess_terminated_after_ttl(monkeypatch: Any) -> None: - """Regression: a subprocess that never exits (pipe buffer deadlock) must be - terminated after TTL to prevent the toolbar from being permanently frozen.""" - mock_proc = MagicMock() - mock_proc.poll.return_value = None # subprocess never finishes (deadlocked) - - spawn_time = time.monotonic() - _GIT_STATUS_TTL - 1.0 # spawned > TTL ago - monkeypatch.setattr(_git_status_state, "proc", mock_proc) - monkeypatch.setattr(_git_status_state, "timestamp", spawn_time) - monkeypatch.setattr(_git_status_state, "dirty", True) # stale value preserved - - result = _get_git_status() - - # Must have been terminated - mock_proc.terminate.assert_called_once() - assert _git_status_state.proc is None - # timestamp reset to ~now so next spawn is delayed by one full TTL - assert time.monotonic() - _git_status_state.timestamp < 2.0 - # Stale cached values are still returned (better than crashing) - assert result == (True, 0, 0) - - -def test_git_status_recent_subprocess_not_terminated(monkeypatch: Any) -> None: - """A subprocess that is still within TTL must not be terminated prematurely.""" - mock_proc = MagicMock() - mock_proc.poll.return_value = None # not finished yet but within TTL - - monkeypatch.setattr(_git_status_state, "proc", mock_proc) - monkeypatch.setattr(_git_status_state, "timestamp", time.monotonic() - 1.0) # only 1s old - - _get_git_status() - - mock_proc.terminate.assert_not_called() - assert _git_status_state.proc is mock_proc # unchanged - - def test_git_status_not_called_when_branch_is_none(monkeypatch: Any) -> None: """When not in a git repo (branch=None), _get_git_status must not be called. From d4b818888999b0de6ea979e691f093bf00fbe731 Mon Sep 17 00:00:00 2001 From: elkaix Date: Tue, 21 Jul 2026 00:19:14 -0400 Subject: [PATCH 2/7] refactor(tui): unify footer data and capability caching --- src/pythinker_code/ui/shell/prompt.py | 348 +++++++++++------- .../ui/shell/prompting/__init__.py | 14 + .../ui/shell/prompting/footer.py | 122 ++++++ src/pythinker_code/ui/shell/statusline.py | 24 +- src/pythinker_code/ui/theme/registry.py | 10 +- tests/ui_and_conv/test_footer_view_model.py | 228 ++++++++++++ 6 files changed, 609 insertions(+), 137 deletions(-) create mode 100644 src/pythinker_code/ui/shell/prompting/footer.py create mode 100644 tests/ui_and_conv/test_footer_view_model.py diff --git a/src/pythinker_code/ui/shell/prompt.py b/src/pythinker_code/ui/shell/prompt.py index c2457948..af4f7f1e 100644 --- a/src/pythinker_code/ui/shell/prompt.py +++ b/src/pythinker_code/ui/shell/prompt.py @@ -72,6 +72,7 @@ ) from pythinker_code.ui.shell.prompting import ( ClipboardAdapter, + FooterViewModel, FrozenFragments, GitSnapshot, GitStatusIndex, @@ -83,6 +84,10 @@ ToastManager, ToastSnapshot, allocate_prompt_scene_rows, + background_task_summary, + select_footer_content, + truncate_footer_left, + truncate_footer_right, ) from pythinker_code.ui.shell.prompting.clipboard import ( bind_clipboard_adapter, @@ -324,17 +329,7 @@ def create_margin( def _background_task_summary(counts: BgTaskCounts) -> str | None: - total = counts.bash + counts.agent - if total <= 0: - return None - noun = "background task" if total == 1 else "background tasks" - parts: list[str] = [] - if counts.bash: - parts.append(f"{counts.bash} bash") - if counts.agent: - parts.append(f"{counts.agent} agent") - detail = f" ({', '.join(parts)})" if parts else "" - return f"{total} {noun} running{detail} · /task to view" + return background_task_summary(bash=counts.bash, agent=counts.agent) or None def _append_footer_hint_fragments( @@ -1957,6 +1952,7 @@ def _(event: KeyPressEvent) -> None: lexer=self._input_highlight_lexer, ) self._current_prompt_frame: PromptFrame | None = None + self._current_footer_view_model: FooterViewModel | None = None self._prompt_frame_collector = self._make_prompt_frame_collector() def _capture_prompt_frame(app: Application[str]) -> None: @@ -1965,6 +1961,11 @@ def _capture_prompt_frame(app: Application[str]) -> None: columns=size.columns, terminal_rows=size.rows, ) + try: + self._current_footer_view_model = self._build_footer_view_model(size.columns) + except CwdLostError as exc: + self._current_footer_view_model = None + app.exit(exception=exc) self._session.app.before_render.add_handler(_capture_prompt_frame) self._session.app.after_render.add_handler(self._clear_prompt_frame_snapshot) @@ -2237,15 +2238,7 @@ def _render_shell_prompt_message(self) -> FormattedText: # (not just on the agent path) so a mode switch or resize cannot leave # _fit_toolbar_to_terminal clipping against a stale agent-mode value. self._prompt_footer_row_budget = frame.terminal_rows - # Snapshot the update notice for this frame too. The agent path caches it - # in _render_agent_prompt_message; without the same refresh here, - # _append_update_notice would replay a stale agent-mode notice (or the - # initial None, suppressing a live notice) once a frame is captured. - provider = cast( - Callable[[], str | None] | None, - getattr(self, "_update_notice_provider", None), - ) - self._prompt_frame_update_notice = provider() if callable(provider) else None + self._prompt_frame_update_notice = self._update_notice_for_render() fragments: FormattedText = FormattedText() if getattr(self, "_shortcut_help_open", False): @@ -2385,6 +2378,7 @@ def _clear_prompt_frame_snapshot(self, _app: Application[str] | None = None) -> notice together, so a later render can never read a snapshot captured for a stale frame/mode.""" self._current_prompt_frame = None + self._current_footer_view_model = None self._prompt_frame_update_notice = None def _prompt_frame_for_render(self, *, columns: int | None = None) -> PromptFrame: @@ -2488,11 +2482,7 @@ def _render_agent_prompt_message(self) -> FormattedText: columns = frame.columns fragments: FormattedText = FormattedText() - provider = cast( - Callable[[], str | None] | None, - getattr(self, "_update_notice_provider", None), - ) - update_notice = provider() if callable(provider) else None + update_notice = self._update_notice_for_render() self._prompt_frame_update_notice = update_notice footer_rows = 3 + (1 if update_notice else 0) @@ -3121,6 +3111,12 @@ def _start(self) -> None: async def _refresh() -> None: try: while True: + git_index = getattr(self, "_git_status_index", None) + if isinstance(git_index, GitStatusIndex): + # A lost CWD is reported explicitly at the render boundary. + with contextlib.suppress(OSError): + git_index.request_refresh(HostPath.cwd()) + app = self._app_for_repaint() if app is not None: app.invalidate() @@ -3596,7 +3592,12 @@ def _append_history_entry(self, text: str) -> None: error=exc, ) - def _append_update_notice(self, fragments: list[tuple[str, str]], columns: int) -> None: + def _append_update_notice( + self, + fragments: list[tuple[str, str]], + columns: int, + footer: FooterViewModel | None = None, + ) -> None: """Append a persistent yellow 'update available' line as the *last* footer row — below the status/clock line — so it sits fully clear of the prompt input box instead of glued to it. Call this last, after the status lines @@ -3604,9 +3605,12 @@ def _append_update_notice(self, fragments: list[tuple[str, str]], columns: int) none) and adds no trailing newline, so it never leaves a blank row at the bottom. No-op when no update is pending; style-agnostic across both toolbar layouts.""" - if getattr(self, "_current_prompt_frame", None) is not None and hasattr( + if footer is not None: + text = footer.update_notice + elif getattr(self, "_current_prompt_frame", None) is not None and hasattr( self, "_prompt_frame_update_notice" ): + # Reuse the notice sampled for this frame; never re-sample mid-frame. text = self._prompt_frame_update_notice else: provider = cast( @@ -3616,7 +3620,11 @@ def _append_update_notice(self, fragments: list[tuple[str, str]], columns: int) text = provider() if callable(provider) else None if not text: return - line = _truncate_right(text, max(0, columns - 1)) + line = truncate_footer_right( + text, + max(0, columns - 1), + ascii_only=footer.status.ascii_only if footer is not None else False, + ) if not line: return tokens = _get_tui_tokens() @@ -3626,9 +3634,7 @@ def _append_update_notice(self, fragments: list[tuple[str, str]], columns: int) def _prompt_git_snapshot(self, root: HostPath) -> GitSnapshot: index = getattr(self, "_git_status_index", None) if isinstance(index, GitStatusIndex): - snapshot = index.snapshot(root) - index.request_refresh(root) - return snapshot + return index.snapshot(root) branch = _get_git_branch() dirty, ahead, behind = _get_git_status() if branch else (False, 0, 0) diffstat = _get_git_diffstat() @@ -3673,6 +3679,11 @@ def _render_bottom_toolbar(self) -> FormattedText: app = get_app_or_none() assert app is not None columns = app.output.get_size().columns + try: + footer = self._footer_view_model_for_render(columns) + except CwdLostError as exc: + app.exit(exception=exc) + return FormattedText([]) # Pythinker footer dispatch. Mirrors components/footer.ts layout while # reusing the existing data sources so we never lose information vs @@ -3680,7 +3691,15 @@ def _render_bottom_toolbar(self) -> FormattedText: from pythinker_code.ui.tui_config import is_card_style if is_card_style(): - return self._render_card_bottom_toolbar(columns) + return self._render_card_bottom_toolbar(footer) + + return self._render_legacy_bottom_toolbar(footer) + + def _render_legacy_bottom_toolbar(self, footer: FooterViewModel) -> FormattedText: + """Render legacy footer chrome over one immutable footer snapshot.""" + from pythinker_code.ui.shell.statusline import format_git_badge + + columns = footer.status.columns fragments: list[tuple[str, str]] = [] tc = get_toolbar_colors() @@ -3697,14 +3716,14 @@ def _render_bottom_toolbar(self) -> FormattedText: self._last_tip_rotate_time = now # Status flags: yolo / auto / plan - status = self._status_provider() - if status.yolo_enabled: + ctx = footer.status + if ctx.flags.yolo: fragments.extend([(tc.yolo_label, "yolo"), ("", " ")]) remaining -= 6 # "yolo" = 4, " " = 2 - if status.auto_enabled: + if ctx.flags.auto: fragments.extend([(tc.auto_label, "auto"), ("", " ")]) remaining -= 6 # "auto" = 4, " " = 2 - if status.plan_mode: + if ctx.flags.plan: fragments.extend([(tc.plan_label, "plan"), ("", " ")]) remaining -= 6 @@ -3725,26 +3744,10 @@ def _render_bottom_toolbar(self) -> FormattedText: # CWD (truncated from left) + git branch with status badge # Degrade gracefully on narrow terminals: full → cwd-only → truncated cwd → skip - try: - git_root = HostPath.cwd() - cwd = _truncate_left(_shorten_cwd(str(git_root)), _MAX_CWD_COLS) - except OSError: - # CWD no longer exists (e.g. external drive unplugged). Ask - # prompt_toolkit to exit; the raised exception will propagate out - # of prompt_async() into the Shell's event router which prints a - # crash report with session info and exits cleanly. - app.exit(exception=CwdLostError()) - return FormattedText([]) - git_snapshot = self._prompt_git_snapshot(git_root) - branch = git_snapshot.branch - if branch: - branch = _truncate_right(branch, _MAX_BRANCH_COLS) - badge = _format_git_badge( - branch, - git_snapshot.dirty, - git_snapshot.ahead, - git_snapshot.behind, - ) + cwd = ctx.cwd or "" + git_info = ctx.git + if git_info is not None: + badge = format_git_badge(git_info, ascii_only=ctx.ascii_only) cwd_text = f"{cwd} {badge}" else: cwd_text = cwd @@ -3753,7 +3756,11 @@ def _render_bottom_toolbar(self) -> FormattedText: cwd_text = cwd # drop badge cwd_w = _display_width(cwd_text) if cwd_w > remaining - 2: - cwd_text = _truncate_right(cwd, max(0, remaining - 2)) + cwd_text = truncate_footer_right( + cwd, + max(0, remaining - 2), + ascii_only=ctx.ascii_only, + ) cwd_w = _display_width(cwd_text) if cwd_text and remaining >= cwd_w + 2: fragments.extend([(tc.cwd, cwd_text), ("", " ")]) @@ -3762,15 +3769,13 @@ def _render_bottom_toolbar(self) -> FormattedText: # Active background task counts (bash + agent, each rendered as its own # badge). Order matters: bash renders first; if there isn't room for the # agent badge too, drop agent and keep bash. - bg_counts = ( - self._background_task_count_provider() - if self._background_task_count_provider - else BgTaskCounts() - ) - for kind_label, kind_count in (("bash", bg_counts.bash), ("agent", bg_counts.agent)): + for kind_label, kind_count in ( + ("bash", ctx.background_bash), + ("agent", ctx.background_agent), + ): if kind_count <= 0: continue - bg_text = f"◇ {kind_label}: {kind_count}" + bg_text = f"{'*' if ctx.ascii_only else '◇'} {kind_label}: {kind_count}" bg_width = _display_width(bg_text) if remaining < bg_width + 2: break @@ -3792,30 +3797,52 @@ def _render_bottom_toolbar(self) -> FormattedText: # ── line 2: toast (left) + context (right) — always rendered ────── fragments.append(("", "\n")) - right_text = self._render_right_span(status) + usable = max(0, columns - 1) + right_text = self._render_right_span(footer) right_width = _display_width(right_text) + if right_width > usable: + right_text = truncate_footer_left( + right_text, + usable, + ascii_only=ctx.ascii_only, + ) + right_width = _display_width(right_text) - left_toast = self._prompt_toast("left") - if left_toast is not None: - max_left = max(0, columns - right_width - 2) + left_content = select_footer_content(footer) + if left_content is not None: + max_left = max(0, usable - right_width - 1) if max_left > 0: - left_text = left_toast.message + left_text = left_content.text if _display_width(left_text) > max_left: - left_text = _truncate_right(left_text, max_left) + left_text = truncate_footer_right( + left_text, + max_left, + ascii_only=ctx.ascii_only, + ) left_width = _display_width(left_text) - fragments.append((left_toast.style or secondary_style, left_text)) + left_style = { + "background": tc.bg_tasks, + "toast": left_content.style or secondary_style, + }.get(left_content.kind, tc.tip) + fragments.append((left_style, left_text)) else: left_width = 0 else: left_width = 0 - fragments.append(("", " " * max(0, columns - left_width - right_width))) + fragments.append(("", " " * max(0, usable - left_width - right_width))) fragments.append((secondary_style, right_text)) - self._append_update_notice(fragments, columns) + self._append_update_notice(fragments, columns, footer) return self._fit_toolbar_to_terminal(FormattedText(fragments), columns) - def _build_statusline_context(self, columns: int) -> StatusLineContext: + def _build_statusline_context( + self, + columns: int, + *, + status: StatusSnapshot | None = None, + background_counts: BgTaskCounts | None = None, + ) -> StatusLineContext: from pythinker_code.ui.shell.statusline import ( GitInfo, RateSampler, @@ -3825,11 +3852,12 @@ def _build_statusline_context(self, columns: int) -> StatusLineContext: from pythinker_code.ui.terminal_capabilities import ascii_glyphs_enabled cfg = getattr(self, "_statusline_cfg", None) or StatusLineConfig() - status = self._status_provider() + status = status if status is not None else self._status_provider() + background_counts = background_counts or self._background_task_counts() now = time.monotonic() self._statusline_frame = getattr(self, "_statusline_frame", 0) + 1 - working = self._has_background_tasks() + working = background_counts.bash > 0 or background_counts.agent > 0 # Samplers may be missing when a session is constructed without __init__ # (test helpers do this); fall back to fresh ones so rendering is robust. @@ -3847,9 +3875,14 @@ def _build_statusline_context(self, columns: int) -> StatusLineContext: rate_in_sampler.reset() rate_out_sampler.reset() + ascii_only = ascii_glyphs_enabled() try: git_root = HostPath.cwd() - cwd_text = _truncate_left(_shorten_cwd(str(git_root)), _MAX_CWD_COLS) + cwd_text = truncate_footer_left( + _shorten_cwd(str(git_root)), + _MAX_CWD_COLS, + ascii_only=ascii_only, + ) except OSError as exc: raise CwdLostError() from exc @@ -3858,7 +3891,11 @@ def _build_statusline_context(self, columns: int) -> StatusLineContext: branch = git_snapshot.branch if branch: git_info = GitInfo( - branch=_truncate_right(branch, _MAX_BRANCH_COLS), + branch=truncate_footer_right( + branch, + _MAX_BRANCH_COLS, + ascii_only=ascii_only, + ), dirty=git_snapshot.dirty, ahead=git_snapshot.ahead, behind=git_snapshot.behind, @@ -3877,7 +3914,7 @@ def _build_statusline_context(self, columns: int) -> StatusLineContext: columns=columns, working=working, frame=self._statusline_frame, - model_name=self._model_name, + model_name=getattr(self, "_model_name", None), provider_label=None, effort=effort, rate_in=rate_in, @@ -3898,12 +3935,79 @@ def _build_statusline_context(self, columns: int) -> StatusLineContext: plan=status.plan_mode, ), limits=None, - ascii_only=ascii_glyphs_enabled(), + ascii_only=ascii_only, style=cfg.style if cfg.enabled else "plain", bar_width=cfg.bar_width, + context_usage=status.context_usage, + background_bash=background_counts.bash, + background_agent=background_counts.agent, + ) + + def _build_footer_view_model(self, columns: int) -> FooterViewModel: + """Sample every dynamic footer provider exactly once for one frame.""" + from pythinker_code.extensions import footer_statuses + from pythinker_code.ui.shell.statusline import StatusLineCommandRunner + + cfg = getattr(self, "_statusline_cfg", None) or StatusLineConfig() + status = self._status_provider() + background_counts = self._background_task_counts() + runner = getattr(self, "_statusline_runner", None) + command_line = "" + if ( + cfg.enabled + and "command" in cfg.segments + and isinstance(runner, StatusLineCommandRunner) + ): + command_line = runner.current_line + + left_toast = self._prompt_toast("left") + toast_snapshot = left_toast if left_toast is not None else self._prompt_toast("right") + update_provider = cast( + Callable[[], str | None] | None, + getattr(self, "_update_notice_provider", None), + ) + return FooterViewModel( + status=self._build_statusline_context( + columns, + status=status, + background_counts=background_counts, + ), + command_line=command_line, + extension_statuses=tuple(sorted(footer_statuses().items())), + background_summary=background_task_summary( + bash=background_counts.bash, + agent=background_counts.agent, + ), + toast=toast_snapshot, + update_notice=update_provider() if callable(update_provider) else None, ) - def _render_card_bottom_toolbar(self, columns: int) -> FormattedText: + def _update_notice_for_render(self) -> str | None: + """Read the update notice for message rendering without building a footer. + + Message rendering must stay independent of the footer providers so + partially constructed sessions (tests, shell mode) can render; prefer + the per-frame footer snapshot when one exists. + """ + footer = getattr(self, "_current_footer_view_model", None) + if footer is not None: + return footer.update_notice + provider = cast( + Callable[[], str | None] | None, + getattr(self, "_update_notice_provider", None), + ) + return provider() if callable(provider) else None + + def _footer_view_model_for_render(self, columns: int) -> FooterViewModel: + current = getattr(self, "_current_footer_view_model", None) + if current is not None and current.status.columns == columns: + return current + footer = self._build_footer_view_model(columns) + if getattr(self, "_current_prompt_frame", None) is not None: + self._current_footer_view_model = footer + return footer + + def _render_card_bottom_toolbar(self, footer: FooterViewModel) -> FormattedText: """Pythinker two-line footer (statusline v2). Line 1 + line-2 right are assembled from the segment registry; the @@ -3911,12 +4015,12 @@ def _render_card_bottom_toolbar(self, columns: int) -> FormattedText: precedence from the legacy footer. """ from pythinker_code.config import StatusLineConfig - from pythinker_code.extensions import footer_statuses from pythinker_code.ui.shell.statusline import ( DEFAULT_STATUSLINE_SEGMENTS, assemble_footer, ) + columns = footer.status.columns cfg = getattr(self, "_statusline_cfg", None) or StatusLineConfig() tc = get_toolbar_colors() tokens = _get_tui_tokens() @@ -3926,16 +4030,8 @@ def _render_card_bottom_toolbar(self, columns: int) -> FormattedText: fragments.append((self._prompt_separator_style(tc.separator), _prompt_rule(columns))) fragments.append(("", "\n")) - try: - ctx = self._build_statusline_context(columns) - except CwdLostError as exc: - app = get_app_or_none() - if app is not None: - app.exit(exception=exc) - return FormattedText([]) - segments = list(cfg.segments) if cfg.enabled else list(DEFAULT_STATUSLINE_SEGMENTS) - line1, line2_right = assemble_footer(ctx, segments) + line1, line2_right = assemble_footer(footer.status, segments) fragments.extend(line1) fragments.append(("", "\n")) @@ -3946,48 +4042,34 @@ def _render_card_bottom_toolbar(self, columns: int) -> FormattedText: right_text = "".join(t for _, t in line2_right) right_width = _display_width(right_text) if right_width > usable: - right_text = _truncate_left(right_text, usable) + right_text = truncate_footer_left( + right_text, + usable, + ascii_only=footer.status.ascii_only, + ) line2_right = [(secondary_style, right_text)] right_width = _display_width(right_text) max_left_width = max(0, usable - right_width - 1) - command_line = "" - runner = getattr(self, "_statusline_runner", None) - if cfg.enabled and "command" in segments and runner is not None: - command_line = runner.current_line - ext = footer_statuses() - if command_line: - command_line = _truncate_right(command_line, max_left_width) - fragments.append((tc.tip, command_line)) - left_width = _display_width(command_line) - elif ext: - ordered = sorted(ext.items()) - ext_line = " ".join(f"{k}:{v}" for k, v in ordered) - ext_line = _truncate_right(ext_line, max_left_width) - fragments.append((tc.tip, ext_line)) - left_width = _display_width(ext_line) - elif ( - bg_summary := _background_task_summary( - self._background_task_count_provider() - if self._background_task_count_provider - else BgTaskCounts() + left_content = select_footer_content(footer) + if left_content is not None: + left_text = truncate_footer_right( + left_content.text, + max_left_width, + ascii_only=footer.status.ascii_only, ) - ) is not None: - bg_summary = _truncate_right(bg_summary, max_left_width) - fragments.append((tc.bg_tasks, bg_summary)) - left_width = _display_width(bg_summary) + left_style = { + "background": tc.bg_tasks, + "toast": left_content.style or secondary_style, + }.get(left_content.kind, tc.tip) + fragments.append((left_style, left_text)) + left_width = _display_width(left_text) else: - left_toast = self._prompt_toast("left") - if left_toast is not None: - left_text = _truncate_right(left_toast.message, max_left_width) - fragments.append((left_toast.style or secondary_style, left_text)) - left_width = _display_width(left_text) - else: - left_width = 0 + left_width = 0 fragments.append(("", " " * max(0, usable - left_width - right_width))) fragments.extend(line2_right) - self._append_update_notice(fragments, columns) + self._append_update_notice(fragments, columns, footer) return self._fit_toolbar_to_terminal(FormattedText(fragments), columns) def _get_two_rotating_tips(self) -> str | None: @@ -4008,12 +4090,22 @@ def _get_one_rotating_tip(self) -> str | None: return None return self._tips[self._tip_rotation_index % len(self._tips)] - def _render_right_span(self, status: StatusSnapshot) -> str: - right_toast = self._prompt_toast("right") - if right_toast is None: + def _render_right_span(self, footer: FooterViewModel) -> str: + if footer.toast is None or footer.toast.position != "right": + status = footer.status return format_context_status( status.context_usage, status.context_tokens, status.max_context_tokens, ) - return right_toast.message + return footer.toast.message + + +# Compatibility surface kept for tests that still import the legacy footer +# helpers (tests/ui_and_conv/test_prompt_tips.py); rendering now goes through +# pythinker_code.ui.shell.prompting.footer. +_LEGACY_FOOTER_HELPERS = ( + _background_task_summary, + _format_git_badge, + _truncate_left, +) diff --git a/src/pythinker_code/ui/shell/prompting/__init__.py b/src/pythinker_code/ui/shell/prompting/__init__.py index c6718807..8be50b2d 100644 --- a/src/pythinker_code/ui/shell/prompting/__init__.py +++ b/src/pythinker_code/ui/shell/prompting/__init__.py @@ -1,6 +1,14 @@ """Deep prompt rendering and session-resource modules.""" from pythinker_code.ui.shell.prompting.clipboard import ClipboardAdapter +from pythinker_code.ui.shell.prompting.footer import ( + FooterContent, + FooterViewModel, + background_task_summary, + select_footer_content, + truncate_footer_left, + truncate_footer_right, +) from pythinker_code.ui.shell.prompting.frame import ( FrozenFragments, PromptFrame, @@ -23,6 +31,8 @@ __all__ = ( "ClipboardAdapter", "FrozenFragments", + "FooterContent", + "FooterViewModel", "GitSnapshot", "GitStatusIndex", "PromptFrame", @@ -35,5 +45,9 @@ "ToastManager", "ToastSnapshot", "allocate_prompt_scene_rows", + "background_task_summary", "freeze_fragments", + "select_footer_content", + "truncate_footer_left", + "truncate_footer_right", ) diff --git a/src/pythinker_code/ui/shell/prompting/footer.py b/src/pythinker_code/ui/shell/prompting/footer.py new file mode 100644 index 00000000..5f393fc7 --- /dev/null +++ b/src/pythinker_code/ui/shell/prompting/footer.py @@ -0,0 +1,122 @@ +"""Immutable footer snapshots and theme-neutral content policy.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Literal + +from prompt_toolkit.utils import get_cwidth + +from pythinker_code.ui.shell.prompting.toasts import ToastSnapshot +from pythinker_code.ui.shell.statusline import StatusLineContext + +FooterContentKind = Literal["command", "extension", "background", "toast"] + + +@dataclass(frozen=True, slots=True) +class FooterViewModel: + """All dynamic data consumed while rendering one prompt footer.""" + + status: StatusLineContext + command_line: str + extension_statuses: tuple[tuple[str, str], ...] + background_summary: str + toast: ToastSnapshot | None + update_notice: str | None + + +@dataclass(frozen=True, slots=True) +class FooterContent: + """The winning left-side footer content after applying shared precedence.""" + + kind: FooterContentKind + text: str + style: str = "" + + +def background_task_summary(*, bash: int, agent: int) -> str: + """Return the stable user-facing summary for active background work.""" + total = bash + agent + if total <= 0: + return "" + noun = "background task" if total == 1 else "background tasks" + parts: list[str] = [] + if bash: + parts.append(f"{bash} bash") + if agent: + parts.append(f"{agent} agent") + detail = f" ({', '.join(parts)})" if parts else "" + return f"{total} {noun} running{detail} · /task to view" + + +def select_footer_content(model: FooterViewModel) -> FooterContent | None: + """Choose left content using command, extension, background, toast precedence.""" + if model.command_line: + return FooterContent("command", model.command_line) + if model.extension_statuses: + text = " ".join(f"{key}:{value}" for key, value in model.extension_statuses) + return FooterContent("extension", text) + if model.background_summary: + return FooterContent("background", model.background_summary) + if model.toast is not None and model.toast.position == "left": + return FooterContent("toast", model.toast.message, model.toast.style) + return None + + +def _display_width(text: str) -> int: + return sum(get_cwidth(character) for character in text) + + +def truncate_footer_right(text: str, width: int, *, ascii_only: bool) -> str: + """Fit footer text from the right with a capability-safe ellipsis.""" + if width <= 0: + return "" + if _display_width(text) <= width: + return text + ellipsis = "..." if ascii_only else "…" + ellipsis_width = _display_width(ellipsis) + if width <= ellipsis_width: + return "." * width + budget = width - ellipsis_width + chars: list[str] = [] + used = 0 + for character in text: + char_width = get_cwidth(character) + if used + char_width > budget: + break + chars.append(character) + used += char_width + return "".join(chars) + ellipsis + + +def truncate_footer_left(text: str, width: int, *, ascii_only: bool) -> str: + """Fit footer text from the left with a capability-safe ellipsis.""" + if width <= 0: + return "" + if _display_width(text) <= width: + return text + ellipsis = "..." if ascii_only else "…" + ellipsis_width = _display_width(ellipsis) + if width <= ellipsis_width: + return "." * width + budget = width - ellipsis_width + chars: list[str] = [] + used = 0 + for character in reversed(text): + char_width = get_cwidth(character) + if used + char_width > budget: + break + chars.append(character) + used += char_width + return ellipsis + "".join(reversed(chars)) + + +__all__ = ( + "FooterContent", + "FooterContentKind", + "FooterViewModel", + "background_task_summary", + "select_footer_content", + "truncate_footer_left", + "truncate_footer_right", +) diff --git a/src/pythinker_code/ui/shell/statusline.py b/src/pythinker_code/ui/shell/statusline.py index 7a4021a3..ddff7a06 100644 --- a/src/pythinker_code/ui/shell/statusline.py +++ b/src/pythinker_code/ui/shell/statusline.py @@ -186,6 +186,9 @@ class StatusLineContext: ascii_only: bool style: str # "fancy" | "plain" bar_width: int + context_usage: float = 0.0 + background_bash: int = 0 + background_agent: int = 0 @dataclass(frozen=True, slots=True) @@ -496,13 +499,22 @@ def width(frags: list[StyleFragment]) -> int: if "context" in line2_ids and "tokens" in line2_ids: line2_ids = [seg for seg in line2_ids if seg != "tokens"] line2_parts = rendered(line2_ids) - line2: list[StyleFragment] = [] - for i, (_seg, seg_frags) in enumerate(line2_parts): - if i: - line2.append((_style(ctx, colors.dim), sep_bar)) - line2.extend(seg_frags) - return joined(line1_parts), line2 + def joined_line2(parts: list[tuple[str, list[StyleFragment]]]) -> list[StyleFragment]: + line: list[StyleFragment] = [] + for index, (_seg, seg_frags) in enumerate(parts): + if index: + line.append((_style(ctx, colors.dim), sep_bar)) + line.extend(seg_frags) + return line + + while width(joined_line2(line2_parts)) > ctx.columns and len(line2_parts) > 1: + victim = max(line2_parts, key=lambda part: SEGMENT_REGISTRY[part[0]].drop_priority) + if SEGMENT_REGISTRY[victim[0]].drop_priority == 0: + break + line2_parts.remove(victim) + + return joined(line1_parts), joined_line2(line2_parts) SEGMENT_REGISTRY: dict[str, SegmentSpec] = { diff --git a/src/pythinker_code/ui/theme/registry.py b/src/pythinker_code/ui/theme/registry.py index f9f6ace0..54578baf 100644 --- a/src/pythinker_code/ui/theme/registry.py +++ b/src/pythinker_code/ui/theme/registry.py @@ -13,7 +13,7 @@ from pythinker_code.ui.color_utils import blend, parse_hex_color, to_hex_color from pythinker_code.ui.terminal_capabilities import color_depth, colors_disabled -from .capabilities import get_terminal_capabilities +from .capabilities import TerminalCapabilities, get_terminal_capabilities from .palettes import THEME_SPECS from .resolver import StyleResolver from .spec import ( @@ -72,9 +72,13 @@ def get_theme_spec(theme: ThemeName | None = None) -> ThemeSpec: return THEME_SPECS[_mode(theme)] -@lru_cache(maxsize=4) +@lru_cache(maxsize=16) +def _resolver_for(theme: ThemeName, caps: TerminalCapabilities) -> StyleResolver: + return StyleResolver(THEME_SPECS[_mode(theme)], caps) + + def get_resolver(theme: ThemeName = "dark") -> StyleResolver: - return StyleResolver(THEME_SPECS[_mode(theme)], get_terminal_capabilities()) + return _resolver_for(theme, get_terminal_capabilities()) def active_resolver() -> StyleResolver: diff --git a/tests/ui_and_conv/test_footer_view_model.py b/tests/ui_and_conv/test_footer_view_model.py new file mode 100644 index 00000000..15598d98 --- /dev/null +++ b/tests/ui_and_conv/test_footer_view_model.py @@ -0,0 +1,228 @@ +"""Parity tests for the immutable prompt footer view model.""" + +from __future__ import annotations + +from dataclasses import replace +from typing import Any, NoReturn + +import pytest +from prompt_toolkit.formatted_text import FormattedText + +from pythinker_code.config import StatusLineConfig +from pythinker_code.ui.shell.prompt import CustomPromptSession, PromptMode +from pythinker_code.ui.shell.prompting.footer import ( + FooterViewModel, + background_task_summary, + select_footer_content, +) +from pythinker_code.ui.shell.prompting.toasts import ToastSnapshot +from pythinker_code.ui.shell.statusline import StatusFlags, StatusLineContext +from pythinker_code.ui.theme import registry as theme_registry +from pythinker_code.ui.theme.capabilities import TerminalCapabilities + + +def _status(columns: int, *, ascii_only: bool = False) -> StatusLineContext: + return StatusLineContext( + columns=columns, + working=True, + frame=0, + model_name="test-model", + provider_label=None, + effort=None, + rate_in=None, + rate_out=None, + session_cost_usd=0.0, + cost_budget_usd=None, + context_tokens=0, + max_context_tokens=0, + elapsed_s=61.0, + clock="12:34", + cwd="~/project", + git=None, + diff_added=None, + diff_removed=None, + flags=StatusFlags(yolo=False, auto=False, plan=False), + limits=None, + ascii_only=ascii_only, + style="plain" if ascii_only else "fancy", + bar_width=8, + background_bash=1, + ) + + +def _toast(message: str = "toast-only") -> ToastSnapshot: + return ToastSnapshot( + message=message, + position="left", + style="bold", + topic=None, + expires_at=float("inf"), + ) + + +def _model( + columns: int, + *, + command: str = "", + extensions: tuple[tuple[str, str], ...] = (), + background: str = "", + toast: ToastSnapshot | None = None, + update_notice: str | None = "Update available", + ascii_only: bool = False, +) -> FooterViewModel: + status = _status(columns, ascii_only=ascii_only) + status = replace( + status, + working=bool(background), + background_bash=1 if background else 0, + ) + return FooterViewModel( + status=status, + command_line=command, + extension_statuses=extensions, + background_summary=background, + toast=toast, + update_notice=update_notice, + ) + + +def _session() -> Any: + session = object.__new__(CustomPromptSession) + session._mode = PromptMode.AGENT + session._model_name = "test-model" + session._tips = [] + session._tip_rotation_index = 0 + session._last_tip_rotate_time = float("inf") + session._prompt_footer_row_budget = 20 + session._statusline_cfg = StatusLineConfig(segments=["command"]) + + def unexpected_provider() -> NoReturn: + raise AssertionError("render adapter sampled a live provider") + + session._status_provider = unexpected_provider + session._background_task_count_provider = unexpected_provider + session._update_notice_provider = unexpected_provider + return session + + +def _text(fragments: FormattedText) -> str: + return "".join(fragment[1] for fragment in fragments) + + +@pytest.mark.parametrize("width", [20, 40, 80, 120]) +@pytest.mark.parametrize( + ("model_kwargs", "expected_kind"), + [ + ( + { + "command": "custom-command", + "extensions": (("long-extension", "x" * 100),), + "background": "background-only", + "toast": _toast(), + }, + "command", + ), + ( + { + "extensions": (("long-extension", "x" * 100),), + "background": "background-only", + "toast": _toast(), + }, + "extension", + ), + ({"background": background_task_summary(bash=1, agent=2), "toast": _toast()}, "background"), + ({"toast": _toast()}, "toast"), + ], +) +def test_legacy_and_card_adapters_share_left_policy_and_width_safety( + monkeypatch: pytest.MonkeyPatch, + width: int, + model_kwargs: dict[str, object], + expected_kind: str, +) -> None: + monkeypatch.setenv("NO_COLOR", "1") + session = _session() + model = _model(width, ascii_only=True, **model_kwargs) # type: ignore[arg-type] + + selected = select_footer_content(model) + assert selected is not None and selected.kind == expected_kind + + legacy = session._render_legacy_bottom_toolbar(model) + card = session._render_card_bottom_toolbar(model) + assert select_footer_content(model) == selected + + for rendered in (legacy, card): + rows = _text(rendered).splitlines() + assert all(len(row) <= width for row in rows) + assert "Update available"[: max(0, width - 1)] in rows[-1] + assert "…" not in _text(rendered) + + # Lower-precedence policy fields cannot leak into either adapter's left row. + if expected_kind == "command": + assert "long-extension:" not in _text(legacy) + assert "long-extension:" not in _text(card) + assert "background-only" not in _text(card) + assert "toast-only" not in _text(card) + elif expected_kind == "extension": + assert "background-only" not in _text(card) + assert "toast-only" not in _text(card) + elif expected_kind == "background": + assert "toast-only" not in _text(legacy) + assert "toast-only" not in _text(card) + + +def test_footer_view_model_is_immutable_and_width_specific() -> None: + model = _model(80, command="cached") + narrower = replace(model, status=replace(model.status, columns=40)) + + assert model.status.columns == 80 + assert narrower.status.columns == 40 + with pytest.raises(AttributeError): + model.command_line = "changed" # type: ignore[misc] + + +def test_footer_view_model_is_built_once_per_prompt_frame() -> None: + session = _session() + model = _model(80, command="cached") + calls = 0 + + def build(columns: int) -> FooterViewModel: + nonlocal calls + calls += 1 + assert columns == 80 + return model + + session._current_prompt_frame = object() + session._current_footer_view_model = None + session._build_footer_view_model = build + + assert session._footer_view_model_for_render(80) is model + assert session._footer_view_model_for_render(80) is model + assert calls == 1 + + +def test_resolver_cache_includes_terminal_capabilities(monkeypatch: pytest.MonkeyPatch) -> None: + first_caps = TerminalCapabilities( + color_enabled=True, + truecolor=True, + color_256=True, + dumb=False, + ) + second_caps = TerminalCapabilities( + color_enabled=False, + truecolor=False, + color_256=False, + dumb=True, + ) + assert isinstance(hash(first_caps), int) + assert first_caps != second_caps + + theme_registry._resolver_for.cache_clear() + monkeypatch.setattr(theme_registry, "get_terminal_capabilities", lambda: first_caps) + first = theme_registry.get_resolver("dark") + assert theme_registry.get_resolver("dark") is first + + monkeypatch.setattr(theme_registry, "get_terminal_capabilities", lambda: second_caps) + second = theme_registry.get_resolver("dark") + assert second is not first + theme_registry._resolver_for.cache_clear() From 14534e4c2e9b2e95057be996e9ebd86435bd1a84 Mon Sep 17 00:00:00 2001 From: elkaix Date: Tue, 21 Jul 2026 00:19:42 -0400 Subject: [PATCH 3/7] docs(changelog): record session-scoped prompt resources and unified footer --- CHANGELOG.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 52674811..d972a9e1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,8 @@ GitHub Releases page; `0.8.0` is the new starting line. ## Unreleased +- Scope prompt Git status, toasts, history, and clipboard to each shell session with a new `/prompt-history status|clear` command, bounded locked history storage, and cross-session isolation. +- Unify the shell footer behind one per-frame view model shared by the legacy and card renderers, and key theme style-resolver caching by terminal capabilities. - Fix a rare queued-follow-up "ghost card": echoing a drained queued command now commits through the scrollback handoff (which hides the input card before the terminal teardown erases the prompt), so the input-card border can no longer fossilize into scrollback above the echoed command under heavy load. - Freeze the public shell prompt compatibility contract with constructor and rendering coverage. - Unify slash and file-mention completion behind one canonical completion context, adding quoted `@"path with spaces"` file mentions. From be35a742c4bc41df3c1efcb4db5e61ec26cd38fc Mon Sep 17 00:00:00 2001 From: elkaix Date: Tue, 21 Jul 2026 04:41:34 -0400 Subject: [PATCH 4/7] fix(tui): preserve simultaneous right-side toast and address review findings - Carry left and right toasts separately in FooterViewModel so a right-side toast is no longer dropped when a left toast is active (regression from the footer unification). - Log clipboard read failures at debug before returning None. - Drop the unused is_clipboard_available re-export from prompt.py. - Split side-effecting append() calls out of assert statements in history tests. --- src/pythinker_code/ui/shell/prompt.py | 12 +++---- .../ui/shell/prompting/clipboard.py | 4 ++- .../ui/shell/prompting/footer.py | 4 +++ tests/ui_and_conv/test_footer_view_model.py | 33 +++++++++++++++++++ tests/ui_and_conv/test_prompt_history.py | 24 +++++++++----- 5 files changed, 60 insertions(+), 17 deletions(-) diff --git a/src/pythinker_code/ui/shell/prompt.py b/src/pythinker_code/ui/shell/prompt.py index af4f7f1e..7eaf58e9 100644 --- a/src/pythinker_code/ui/shell/prompt.py +++ b/src/pythinker_code/ui/shell/prompt.py @@ -96,9 +96,6 @@ from pythinker_code.ui.shell.prompting.clipboard import ( grab_media_from_clipboard as grab_media_from_clipboard, ) -from pythinker_code.ui.shell.prompting.clipboard import ( - is_clipboard_available as is_clipboard_available, -) from pythinker_code.ui.shell.prompting.clipboard import ( is_media_clipboard_available as is_media_clipboard_available, ) @@ -3960,8 +3957,6 @@ def _build_footer_view_model(self, columns: int) -> FooterViewModel: ): command_line = runner.current_line - left_toast = self._prompt_toast("left") - toast_snapshot = left_toast if left_toast is not None else self._prompt_toast("right") update_provider = cast( Callable[[], str | None] | None, getattr(self, "_update_notice_provider", None), @@ -3978,8 +3973,9 @@ def _build_footer_view_model(self, columns: int) -> FooterViewModel: bash=background_counts.bash, agent=background_counts.agent, ), - toast=toast_snapshot, + toast=self._prompt_toast("left"), update_notice=update_provider() if callable(update_provider) else None, + toast_right=self._prompt_toast("right"), ) def _update_notice_for_render(self) -> str | None: @@ -4091,14 +4087,14 @@ def _get_one_rotating_tip(self) -> str | None: return self._tips[self._tip_rotation_index % len(self._tips)] def _render_right_span(self, footer: FooterViewModel) -> str: - if footer.toast is None or footer.toast.position != "right": + if footer.toast_right is None: status = footer.status return format_context_status( status.context_usage, status.context_tokens, status.max_context_tokens, ) - return footer.toast.message + return footer.toast_right.message # Compatibility surface kept for tests that still import the legacy footer diff --git a/src/pythinker_code/ui/shell/prompting/clipboard.py b/src/pythinker_code/ui/shell/prompting/clipboard.py index f6d96c9d..0f9452b1 100644 --- a/src/pythinker_code/ui/shell/prompting/clipboard.py +++ b/src/pythinker_code/ui/shell/prompting/clipboard.py @@ -20,6 +20,7 @@ from pythinker_code.utils.clipboard import ( is_media_clipboard_available as utility_is_media_clipboard_available, ) +from pythinker_code.utils.logging import logger class ClipboardAdapter: @@ -54,7 +55,8 @@ def paste_text(self, clipboard: Clipboard) -> str | None: """Return pasted text, preserving the prompt's silent failure behavior.""" try: data = clipboard.get_data() - except Exception: + except Exception as exc: + logger.debug("Clipboard text read failed: error={!r}", exc) return None return data.text diff --git a/src/pythinker_code/ui/shell/prompting/footer.py b/src/pythinker_code/ui/shell/prompting/footer.py index 5f393fc7..4fecd224 100644 --- a/src/pythinker_code/ui/shell/prompting/footer.py +++ b/src/pythinker_code/ui/shell/prompting/footer.py @@ -23,6 +23,10 @@ class FooterViewModel: background_summary: str toast: ToastSnapshot | None update_notice: str | None + # Left- and right-positioned toasts can be active at once; the right toast is + # rendered by the right span, so it must be carried separately from ``toast`` + # (which feeds left-side content precedence) or it would be dropped. + toast_right: ToastSnapshot | None = None @dataclass(frozen=True, slots=True) diff --git a/tests/ui_and_conv/test_footer_view_model.py b/tests/ui_and_conv/test_footer_view_model.py index 15598d98..759bcfb5 100644 --- a/tests/ui_and_conv/test_footer_view_model.py +++ b/tests/ui_and_conv/test_footer_view_model.py @@ -171,6 +171,39 @@ def test_legacy_and_card_adapters_share_left_policy_and_width_safety( assert "toast-only" not in _text(card) +def test_left_and_right_toasts_both_render(monkeypatch: pytest.MonkeyPatch) -> None: + """Regression: a left toast must not drop a simultaneously active right toast.""" + monkeypatch.setenv("NO_COLOR", "1") + session = _session() + left = ToastSnapshot( + message="left-toast", + position="left", + style="bold", + topic=None, + expires_at=float("inf"), + ) + right = ToastSnapshot( + message="right-toast", + position="right", + style="bold", + topic=None, + expires_at=float("inf"), + ) + model = FooterViewModel( + status=_status(120, ascii_only=True), + command_line="", + extension_statuses=(), + background_summary="", + toast=left, + update_notice=None, + toast_right=right, + ) + + rendered = _text(session._render_legacy_bottom_toolbar(model)) + assert "left-toast" in rendered + assert "right-toast" in rendered + + def test_footer_view_model_is_immutable_and_width_specific() -> None: model = _model(80, command="cached") narrower = replace(model, status=replace(model.status, columns=40)) diff --git a/tests/ui_and_conv/test_prompt_history.py b/tests/ui_and_conv/test_prompt_history.py index 905c4e39..9345ce56 100644 --- a/tests/ui_and_conv/test_prompt_history.py +++ b/tests/ui_and_conv/test_prompt_history.py @@ -148,9 +148,12 @@ def test_append_history_entry_restricts_file_permissions(tmp_path) -> None: def test_prompt_history_store_loads_only_configured_tail(tmp_path) -> None: store = PromptHistoryStore(tmp_path / "history.jsonl", max_entries=2) - assert store.append("one") - assert store.append("two") - assert store.append("three") + appended_one = store.append("one") + appended_two = store.append("two") + appended_three = store.append("three") + assert appended_one + assert appended_two + assert appended_three assert [entry.content for entry in store.load()] == ["two", "three"] @@ -158,21 +161,25 @@ def test_prompt_history_store_loads_only_configured_tail(tmp_path) -> None: def test_prompt_history_store_excludes_credential_commands(tmp_path) -> None: store = PromptHistoryStore(tmp_path / "history.jsonl") - assert store.append("/login api-key") is False - assert store.append("logout openai") is False + login_appended = store.append("/login api-key") + logout_appended = store.append("logout openai") + assert login_appended is False + assert logout_appended is False assert not store.path.exists() def test_prompt_history_store_skips_oversized_records(tmp_path) -> None: store = PromptHistoryStore(tmp_path / "history.jsonl") - assert store.append("x" * (256 * 1024)) is False + oversized_appended = store.append("x" * (256 * 1024)) + assert oversized_appended is False assert not store.path.exists() def test_prompt_history_store_clear_confirms_both_files_are_removed(tmp_path) -> None: store = PromptHistoryStore(tmp_path / "history.jsonl") - assert store.append("kept") + kept_appended = store.append("kept") + assert kept_appended encoding = "utf-8" store.rotated_path.write_text('{"content":"older"}\n', encoding=encoding) @@ -196,7 +203,8 @@ def test_prompt_history_clear_reports_success_only_after_confirmed_removal( monkeypatch, ) -> None: store = PromptHistoryStore(tmp_path / "history.jsonl") - assert store.append("kept") + kept_appended = store.append("kept") + assert kept_appended prompt_session = object.__new__(shell_prompt.CustomPromptSession) cast(Any, prompt_session)._history_store = store app = SimpleNamespace(_prompt_session=prompt_session) From 251531fb949f82d0a3883fee8414b87ea99d4c9c Mon Sep 17 00:00:00 2001 From: elkaix Date: Tue, 21 Jul 2026 04:56:25 -0400 Subject: [PATCH 5/7] fix(tui): drop dead history re-export aliases from prompt facade Remove the unused _ensure_private_history_path / _load_history_entries aliases (and their now-unused imports); history I/O lives in PromptHistoryStore and nothing references these shims. --- src/pythinker_code/ui/shell/prompt.py | 6 ------ 1 file changed, 6 deletions(-) diff --git a/src/pythinker_code/ui/shell/prompt.py b/src/pythinker_code/ui/shell/prompt.py index 7eaf58e9..4c6c0ca2 100644 --- a/src/pythinker_code/ui/shell/prompt.py +++ b/src/pythinker_code/ui/shell/prompt.py @@ -121,8 +121,6 @@ ) from pythinker_code.ui.shell.prompting.history import ( HistoryEntry, - ensure_private_history_path, - load_history_entries, redact_history_secrets, ) from pythinker_code.ui.shell.prompting.lifecycle import PromptLifecycle @@ -1197,10 +1195,6 @@ def _redact_history_secrets(text: str) -> str: return redact_history_secrets(text) -_ensure_private_history_path = ensure_private_history_path -_load_history_entries = load_history_entries - - class PromptUIState(Enum): NORMAL_INPUT = "normal_input" MODAL_HIDDEN_INPUT = "modal_hidden_input" From 0501693bb8fc356d6cfaa79580cea6c9bbe45956 Mon Sep 17 00:00:00 2001 From: elkaix Date: Tue, 21 Jul 2026 05:44:04 -0400 Subject: [PATCH 6/7] fix(tui): drop non-functional sync prompt-session context manager CustomPromptSession.__enter__ scheduled a lifecycle task via asyncio.create_task (which requires a running loop) and __exit__ never drove the lifecycle's async closers, so a synchronous `with` either raised on entry or leaked history/toast/clipboard/index resources on exit. Production uses `async with` exclusively; remove the broken sync context manager so misuse fails loudly instead of silently leaking. Also assert footer rows stay within terminal display width (not code-point length) and add wide/combining Unicode coverage, so a wide glyph overflowing the row can no longer pass the width guard unnoticed. --- src/pythinker_code/ui/shell/prompt.py | 21 ++++----------- tests/ui_and_conv/test_footer_view_model.py | 29 +++++++++++++++++++-- 2 files changed, 32 insertions(+), 18 deletions(-) diff --git a/src/pythinker_code/ui/shell/prompt.py b/src/pythinker_code/ui/shell/prompt.py index 4c6c0ca2..9330ed34 100644 --- a/src/pythinker_code/ui/shell/prompt.py +++ b/src/pythinker_code/ui/shell/prompt.py @@ -3138,22 +3138,11 @@ async def _refresh() -> None: if self._statusline_runner is not None: self._statusline_runner.start() - def __enter__(self) -> CustomPromptSession: - self._bind_resource_bindings() - try: - self._start() - return self - except BaseException: - self._reset_resource_bindings() - raise - - def __exit__(self, *_: object) -> None: - if self._status_refresh_task is not None and not self._status_refresh_task.done(): - self._status_refresh_task.cancel() - self._status_refresh_task = None - if self._statusline_runner is not None: - self._statusline_runner.cancel() - self._reset_resource_bindings() + # Only the async context manager is supported: ``_start()`` schedules a + # lifecycle task via ``asyncio.create_task`` (needs a running loop) and + # teardown is inherently async (``_lifecycle.aclose()`` awaits task + # cancellation and each registered closer). A synchronous ``with`` could + # never run its closers, so it is intentionally omitted — use ``async with``. async def __aenter__(self) -> CustomPromptSession: self._bind_resource_bindings() diff --git a/tests/ui_and_conv/test_footer_view_model.py b/tests/ui_and_conv/test_footer_view_model.py index 759bcfb5..556a6d25 100644 --- a/tests/ui_and_conv/test_footer_view_model.py +++ b/tests/ui_and_conv/test_footer_view_model.py @@ -9,7 +9,7 @@ from prompt_toolkit.formatted_text import FormattedText from pythinker_code.config import StatusLineConfig -from pythinker_code.ui.shell.prompt import CustomPromptSession, PromptMode +from pythinker_code.ui.shell.prompt import CustomPromptSession, PromptMode, _display_width from pythinker_code.ui.shell.prompting.footer import ( FooterViewModel, background_task_summary, @@ -153,7 +153,9 @@ def test_legacy_and_card_adapters_share_left_policy_and_width_safety( for rendered in (legacy, card): rows = _text(rendered).splitlines() - assert all(len(row) <= width for row in rows) + # Production truncation reserves terminal columns by display width, so the + # width guard must measure display width, not code-point count. + assert all(_display_width(row) <= width for row in rows) assert "Update available"[: max(0, width - 1)] in rows[-1] assert "…" not in _text(rendered) @@ -171,6 +173,29 @@ def test_legacy_and_card_adapters_share_left_policy_and_width_safety( assert "toast-only" not in _text(card) +def test_footer_rows_respect_display_width_with_wide_and_combining_chars( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Wide/combining glyphs must be width-accounted so no rendered row overflows. + + ``len()`` would mis-measure both directions — a CJK glyph occupies two columns + but counts as one code point, while a combining mark occupies zero columns but + counts as one — so only a display-width guard catches terminal overflow here. + """ + monkeypatch.setenv("NO_COLOR", "1") + session = _session() + wide_command = "全角指令" + "é" * 3 + for width in (40, 80, 120): + model = _model(width, command=wide_command, ascii_only=False) + assert select_footer_content(model) is not None + legacy = session._render_legacy_bottom_toolbar(model) + card = session._render_card_bottom_toolbar(model) + for rendered in (legacy, card): + rows = _text(rendered).splitlines() + assert all(_display_width(row) <= width for row in rows) + assert "Update available"[: max(0, width - 1)] in rows[-1] + + def test_left_and_right_toasts_both_render(monkeypatch: pytest.MonkeyPatch) -> None: """Regression: a left toast must not drop a simultaneously active right toast.""" monkeypatch.setenv("NO_COLOR", "1") From d0be57c448722a377fc6bc4b2b429e07a692ef08 Mon Sep 17 00:00:00 2001 From: elkaix Date: Tue, 21 Jul 2026 06:03:26 -0400 Subject: [PATCH 7/7] test(tui): force footer width regression across a display-width boundary The wide/combining-char footer test used broad terminal widths, so the wide command never approached the width boundary and a len()-based guard could still have passed. Derive the width from the fixture so the left region must truncate: a len()-based renderer would under-truncate and overflow the row, while display-width accounting keeps every row within bounds and emits the capability-safe ellipsis (now asserted). --- tests/ui_and_conv/test_footer_view_model.py | 31 ++++++++++++++------- 1 file changed, 21 insertions(+), 10 deletions(-) diff --git a/tests/ui_and_conv/test_footer_view_model.py b/tests/ui_and_conv/test_footer_view_model.py index 556a6d25..587965a4 100644 --- a/tests/ui_and_conv/test_footer_view_model.py +++ b/tests/ui_and_conv/test_footer_view_model.py @@ -184,16 +184,27 @@ def test_footer_rows_respect_display_width_with_wide_and_combining_chars( """ monkeypatch.setenv("NO_COLOR", "1") session = _session() - wide_command = "全角指令" + "é" * 3 - for width in (40, 80, 120): - model = _model(width, command=wide_command, ascii_only=False) - assert select_footer_content(model) is not None - legacy = session._render_legacy_bottom_toolbar(model) - card = session._render_card_bottom_toolbar(model) - for rendered in (legacy, card): - rows = _text(rendered).splitlines() - assert all(_display_width(row) <= width for row in rows) - assert "Update available"[: max(0, width - 1)] in rows[-1] + # A run of CJK glyphs (each two columns, one code point) plus a zero-width + # combining mark: display width far exceeds the code-point count. + wide_command = "全角指令文字幅測試漢字表示幅検証" + "é" + # Derive a terminal narrower than the command's display width so the left + # region must truncate. A len()-based guard would treat the command as fitting + # and under-truncate, overflowing the row; only display-width accounting keeps + # the row within bounds and emits the capability-safe ellipsis. + width = _display_width(wide_command) - 6 + model = _model(width, command=wide_command, ascii_only=False) + selected = select_footer_content(model) + assert selected is not None and selected.kind == "command" + legacy = session._render_legacy_bottom_toolbar(model) + card = session._render_card_bottom_toolbar(model) + for rendered in (legacy, card): + text = _text(rendered) + rows = text.splitlines() + assert all(_display_width(row) <= width for row in rows) + # Truncation must have fired at this boundary (regression guard: a + # len()-based renderer would not have needed to truncate here). + assert "…" in text + assert "Update available"[: max(0, width - 1)] in rows[-1] def test_left_and_right_toasts_both_render(monkeypatch: pytest.MonkeyPatch) -> None: