diff --git a/CHANGELOG.md b/CHANGELOG.md index ff105080..5a92066a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,7 @@ GitHub Releases page; `0.8.0` is the new starting line. ## Unreleased - Freeze the public shell prompt compatibility contract with constructor and rendering coverage. +- Keep the interactive prompt scene within the terminal height with a priority-ordered row allocator, and shut prompt background tasks/processes down with an awaited lifecycle. - Add xAI Grok OAuth login (browser loopback and device-code). - Add GitHub Copilot device-code OAuth login for individual github.com accounts. - Add DigitalOcean Gradient AI browser OAuth login with dynamically discovered Inference Routers; router-discovery failures (unauthorized, outage, malformed, empty) are now reported distinctly instead of silently yielding no models. diff --git a/src/pythinker_code/ui/shell/__init__.py b/src/pythinker_code/ui/shell/__init__.py index ff514d82..775d6472 100644 --- a/src/pythinker_code/ui/shell/__init__.py +++ b/src/pythinker_code/ui/shell/__init__.py @@ -931,7 +931,7 @@ def _bg_task_counts() -> BgTaskCounts: _bg_cache.time = now return _bg_cache.counts - with CustomPromptSession( + async with CustomPromptSession( status_provider=lambda: self.soul.status, status_block_provider=_mcp_status_block, fast_refresh_provider=_mcp_status_loading, diff --git a/src/pythinker_code/ui/shell/prompt.py b/src/pythinker_code/ui/shell/prompt.py index 2a560df8..4f8b24d1 100644 --- a/src/pythinker_code/ui/shell/prompt.py +++ b/src/pythinker_code/ui/shell/prompt.py @@ -88,6 +88,31 @@ PromptSceneBudget, allocate_prompt_scene_rows, ) +from pythinker_code.ui.shell.prompting.lifecycle import PromptLifecycle +from pythinker_code.ui.shell.prompting.state import ( + BufferObserved, + Invalidate, + ModalAttached, + ModalDetached, + ModalState, + ModeChanged, + PromptEffect, + PromptEvent, + PromptMode, + PromptPhase, + PromptState, + RestoreDocument, + RunningDelegateAttached, + RunningDelegateDetached, + RunningPromptDelegate, + SelectCompleter, + SetEraseWhenDone, + ShortcutHelpToggled, + SuspendDocument, + TurnCleared, + TurnStarting, + transition, +) from pythinker_code.ui.shell.spacing import ( PREAMBLE_EARLIER_OUTPUT_HIDDEN_HINT, ensure_prompt_newline, @@ -143,23 +168,6 @@ def _pythinker_unraisable_hook(unraisable: Any) -> None: _ORIGINAL_UNRAISABLE_HOOK(unraisable) -def _is_prompt_toolkit_empty_exception_context(context: dict[str, Any]) -> bool: - """Return true for prompt_toolkit's unhelpful ``Exception None`` report. - - prompt_toolkit prints ``Unhandled exception in event loop`` and blocks on - ``Press ENTER to continue`` even when asyncio only supplied a diagnostic - context with no exception object. That message has no traceback or useful - recovery action for users, so Pythinker logs it instead of surfacing a modal - terminal pause. - """ - if context.get("exception") is not None: - return False - message = str(context.get("message") or "") - if not message: - return True - return message.startswith(("Task was destroyed but it is pending", "Future exception")) - - # Python 3.14 can report prompt_toolkit's already-cancelled key-timeout coroutine as an # unraisable KeyError("__import__") during interpreter/module teardown. The RuntimeWarning filters # above catch the normal warning path; this hook catches the shutdown-only unraisable path while @@ -1875,17 +1883,6 @@ def _load_history_entries(history_file: Path) -> list[_HistoryEntry]: return entries -class PromptMode(Enum): - AGENT = "agent" - SHELL = "shell" - - def toggle(self) -> PromptMode: - return PromptMode.SHELL if self == PromptMode.AGENT else PromptMode.AGENT - - def __str__(self) -> str: - return self.value - - class PromptUIState(Enum): NORMAL_INPUT = "normal_input" MODAL_HIDDEN_INPUT = "modal_hidden_input" @@ -2171,26 +2168,6 @@ class _ToastEntry: """Optional prompt_toolkit style for the rendered line; "" uses the default toast style.""" -class RunningPromptDelegate(Protocol): - """Protocol for components that can take over the bottom prompt area.""" - - modal_priority: int - - def render_running_prompt_body(self, columns: int) -> AnyFormattedText: ... - - def running_prompt_placeholder(self) -> AnyFormattedText | None: ... - - def running_prompt_allows_text_input(self) -> bool: ... - - def running_prompt_hides_input_buffer(self) -> bool: ... - - def running_prompt_accepts_submission(self) -> bool: ... - - def should_handle_running_prompt_key(self, key: str) -> bool: ... - - def handle_running_prompt_key(self, key: str, event: KeyPressEvent) -> None: ... - - @dataclass(frozen=True, slots=True) class BgTaskCounts: bash: int = 0 @@ -2341,11 +2318,15 @@ def __init__( _statusline_cfg = statusline_config or StatusLineConfig() self._statusline_layout = resolve_segments(_statusline_cfg) self._statusline_runner: StatusLineCommandRunner | None = None + self._lifecycle = PromptLifecycle() if self._statusline_layout.show_command and _statusline_cfg.command: self._statusline_runner = StatusLineCommandRunner( command=_statusline_cfg.command, timeout_ms=_statusline_cfg.command_timeout_ms, ) + self._lifecycle.register_closer( + "statusline command runner", self._statusline_runner.stop + ) self._statusline_cfg = _statusline_cfg self._statusline_started_at = time.monotonic() self._rate_in_sampler = RateSampler() @@ -2404,6 +2385,7 @@ def __init__( self._slash_menu_control: SlashCommandMenuControl | None = None 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() self._tips = _build_toolbar_tips(clipboard_available or media_clipboard_available) @@ -2534,21 +2516,16 @@ def _(event: KeyPressEvent) -> None: if event.current_buffer.text.strip(): event.current_buffer.insert_text("?") return - self._shortcut_help_open = not self._shortcut_help_open - event.app.invalidate() + self.toggle_shortcut_help() @_kb.add("c-x", eager=True) def _(event: KeyPressEvent) -> None: if self._active_prompt_delegate() is not None: return - self._mode = self._mode.toggle() + self.toggle_mode() from pythinker_code.telemetry import track track("shortcut_mode_switch", to_mode=self._mode.value) - # Apply mode-specific settings - self._apply_mode(event) - # Redraw UI - event.app.invalidate() @_kb.add("s-tab", eager=True) def _(event: KeyPressEvent) -> None: @@ -2738,8 +2715,7 @@ def _(event: KeyPressEvent) -> None: filter=Condition(lambda: self._shortcut_help_open), ) def _(event: KeyPressEvent) -> None: - self._shortcut_help_open = False - event.app.invalidate() + self.close_shortcut_help() @_kb.add( "1", @@ -2863,7 +2839,6 @@ def _capture_prompt_frame(app: Application[str]) -> None: and not delegate.running_prompt_allows_text_input() ) ) - self._install_prompt_exception_filter() self._install_slash_completion_menu() self._install_prompt_buffer_visibility() self._apply_mode() @@ -2874,6 +2849,7 @@ def _capture_prompt_frame(app: Application[str]) -> None: def _(buffer: Buffer) -> None: self._last_input_activity_time = time.monotonic() self._input_activity_event.set() + self._dispatch(BufferObserved(buffer.document)) if buffer.complete_while_typing() and not self._suppress_auto_completion: buffer.start_completion() @@ -2899,22 +2875,6 @@ def _(buffer: Buffer) -> None: self._status_refresh_task: asyncio.Task[None] | None = None - def _install_prompt_exception_filter(self) -> None: - """Avoid prompt_toolkit's blocking ``Exception None`` terminal pause.""" - app = self._session.app - original_handler = app._handle_exception # pyright: ignore[reportPrivateUsage] - - def _handle_exception(loop: asyncio.AbstractEventLoop, context: dict[str, Any]) -> None: - if _is_prompt_toolkit_empty_exception_context(context): - logger.debug( - "Suppressed prompt_toolkit empty exception context: {context}", - context={k: repr(v) for k, v in context.items()}, - ) - return - original_handler(loop, context) - - app._handle_exception = _handle_exception # pyright: ignore[reportPrivateUsage] - def _install_slash_completion_menu(self) -> None: float_container = _find_prompt_float_container(self._session.layout.container) if not isinstance(float_container, FloatContainer): @@ -3373,33 +3333,7 @@ def invalidate(self) -> None: app.invalidate() def _sync_prompt_ui_state(self) -> None: - new_state = self._active_ui_state() - old_state = getattr(self, "_last_ui_state", PromptUIState.NORMAL_INPUT) - buffer = self._session.default_buffer - - if ( - old_state != PromptUIState.MODAL_HIDDEN_INPUT - and new_state == PromptUIState.MODAL_HIDDEN_INPUT - ): - if self._suspended_buffer_document is None and buffer.text: - self._suspended_buffer_document = buffer.document - buffer.set_document(Document(), bypass_readonly=True) - elif ( - old_state == PromptUIState.MODAL_HIDDEN_INPUT - and new_state != PromptUIState.MODAL_HIDDEN_INPUT - and self._suspended_buffer_document is not None - ): - if not buffer.text: - buffer.set_document(self._suspended_buffer_document, bypass_readonly=True) - else: - # Buffer was externally modified (e.g. approval inline feedback). - # Don't overwrite the new content, but log that the old input is lost. - logger.debug( - "Dropping suspended buffer document because buffer was modified externally" - ) - self._suspended_buffer_document = None - - self._last_ui_state = new_state + self._last_ui_state = self._active_ui_state() def _render_agent_prompt_message(self) -> FormattedText: frame = self._prompt_frame_for_render() @@ -4032,9 +3966,9 @@ def _render_agent_prompt_label(self) -> FormattedText: """Render the prompt label (empty — cursor starts at column 0).""" return FormattedText([("", " ")]) - def __enter__(self) -> CustomPromptSession: + def _start(self) -> None: if self._status_refresh_task is not None and not self._status_refresh_task.done(): - return self + return async def _refresh() -> None: try: @@ -4065,18 +3999,32 @@ async def _refresh() -> None: # graceful exit pass - self._status_refresh_task = asyncio.create_task(_refresh()) + self._status_refresh_task = self._lifecycle.create_task(_refresh()) if self._statusline_runner is not None: self._statusline_runner.start() + + def __enter__(self) -> CustomPromptSession: + self._start() return self - def __exit__(self, *_) -> None: + 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() + async def __aenter__(self) -> CustomPromptSession: + self._start() + return self + + async def __aexit__(self, *_: object) -> None: + await self.aclose() + + async def aclose(self) -> None: + await self._lifecycle.aclose() + self._status_refresh_task = None + def _get_placeholder_manager(self) -> PromptPlaceholderManager: manager = getattr(self, "_placeholder_manager", None) if manager is None: @@ -4202,6 +4150,102 @@ async def wait_for_input_activity(self) -> None: await self._input_activity_event.wait() self._input_activity_event.clear() + def _prompt_reducer_state(self) -> PromptState: + """Return the authoritative reducer state, bootstrapping legacy sessions once. + + A few integrations historically populated these attributes on partially + constructed sessions. Bootstrap those sessions at this boundary without + rebuilding reducer-owned state from mutable facade projections on every + dispatch. + """ + existing = getattr(self, "_prompt_state", None) + if isinstance(existing, PromptState): + return existing + + running = getattr(self, "_running_prompt_delegate", None) + turn_starting = getattr(self, "_turn_starting", False) + phase = ( + PromptPhase.RUNNING + if running is not None + else PromptPhase.TURN_STARTING + if turn_starting + else PromptPhase.IDLE + ) + return PromptState( + mode=getattr(self, "_mode", PromptMode.AGENT), + phase=phase, + running_delegate=running, + modal_stack=tuple( + ModalState( + delegate=delegate, + priority=delegate.modal_priority, + hides_input=delegate.running_prompt_hides_input_buffer(), + ) + for delegate in getattr(self, "_modal_delegates", ()) + ), + suspended_document=getattr(self, "_suspended_buffer_document", None), + shortcut_help_open=getattr(self, "_shortcut_help_open", False), + running_previous_mode=getattr( + self, + "_running_prompt_previous_mode", + None, + ), + ) + + def _apply_reducer_state(self, state: PromptState) -> None: + self._prompt_state = state + self._mode = state.mode + self._turn_starting = state.phase is PromptPhase.TURN_STARTING + self._running_prompt_delegate = state.running_delegate + self._running_prompt_previous_mode = state.running_previous_mode + self._modal_delegates = [modal.delegate for modal in state.modal_stack] + self._suspended_buffer_document = state.suspended_document + self._shortcut_help_open = state.shortcut_help_open + + def _apply_prompt_effect(self, effect: PromptEffect) -> None: + session = getattr(self, "_session", None) + buffer = getattr(session, "default_buffer", None) + if isinstance(effect, SelectCompleter): + if buffer is not None: + attribute = ( + "_shell_mode_completer" + if effect.mode is PromptMode.SHELL + else "_agent_mode_completer" + ) + completer = getattr(self, attribute, None) + if completer is not None: + buffer.completer = completer + elif isinstance(effect, SetEraseWhenDone): + app = getattr(session, "app", None) + if app is not None: + app.erase_when_done = effect.erase_when_done + elif isinstance(effect, SuspendDocument): + if buffer is not None and buffer.text: + buffer.set_document(Document(), bypass_readonly=True) + elif isinstance(effect, RestoreDocument) and buffer is not None and not buffer.text: + buffer.set_document(effect.document, bypass_readonly=True) + + def _dispatch(self, event: PromptEvent, *, invalidate_noop: bool = False) -> None: + result = transition(self._prompt_reducer_state(), event) + self._apply_reducer_state(result.state) + should_invalidate = invalidate_noop + for effect in result.effects: + if isinstance(effect, Invalidate): + should_invalidate = True + else: + self._apply_prompt_effect(effect) + if should_invalidate: + self.invalidate() + + def toggle_mode(self) -> None: + self._dispatch(ModeChanged(self._prompt_reducer_state().mode.toggle())) + + def toggle_shortcut_help(self) -> None: + self._dispatch(ShortcutHelpToggled()) + + def close_shortcut_help(self) -> None: + self._dispatch(ShortcutHelpToggled(open=False)) + def mark_turn_starting(self) -> None: """Collapse the input card immediately, before the delegate attaches. @@ -4211,11 +4255,7 @@ def mark_turn_starting(self) -> None: the stream). Superseded by the delegate once :meth:`attach_running_prompt` runs; cleared there and on detach. """ - # Idempotent: a repeat call (e.g. two dispatches before an attach) must - # not cost an extra repaint. - if not self._turn_starting: - self._turn_starting = True - self.invalidate() + self._dispatch(TurnStarting()) def clear_turn_starting(self) -> None: """Drop the pre-attach turn-starting hint without an attach/detach. @@ -4225,52 +4265,30 @@ def clear_turn_starting(self) -> None: path that occurred before the running-prompt delegate ever attached — without reaching into the private ``_turn_starting`` attribute. """ - self._turn_starting = False - self.invalidate() + self._dispatch(TurnCleared(), invalidate_noop=not hasattr(self, "_session")) def attach_running_prompt(self, delegate: RunningPromptDelegate) -> None: - current = getattr(self, "_running_prompt_delegate", None) - if current is delegate: - return - if current is None: - self._running_prompt_previous_mode = self._mode - self._running_prompt_delegate = delegate - # The delegate is the source of truth now; drop the pre-attach hint. - self._turn_starting = False - self._mode = PromptMode.AGENT - self._apply_mode() - self.invalidate() + self._dispatch(RunningDelegateAttached(delegate)) def detach_running_prompt(self, delegate: RunningPromptDelegate) -> None: - if getattr(self, "_running_prompt_delegate", None) is not delegate: - return - previous_mode = getattr(self, "_running_prompt_previous_mode", None) - self._running_prompt_delegate = None - self._running_prompt_previous_mode = None - self._turn_starting = False - if previous_mode is not None: - self._mode = previous_mode - self._apply_mode() - self.invalidate() + self._dispatch(RunningDelegateDetached(delegate)) def attach_modal(self, delegate: RunningPromptDelegate) -> None: - modal_delegates: list[RunningPromptDelegate] | None = getattr( - self, "_modal_delegates", None + buffer = getattr(getattr(self, "_session", None), "default_buffer", None) + document = buffer.document if buffer is not None else Document() + self._dispatch( + ModalAttached( + delegate=delegate, + priority=delegate.modal_priority, + hides_input=delegate.running_prompt_hides_input_buffer(), + document=document, + ) ) - if modal_delegates is None: - modal_delegates = [] - self._modal_delegates = modal_delegates - if delegate in modal_delegates: - return - modal_delegates.append(delegate) - self.invalidate() def detach_modal(self, delegate: RunningPromptDelegate) -> None: - modal_delegates = getattr(self, "_modal_delegates", None) - if not modal_delegates or delegate not in modal_delegates: - return - modal_delegates.remove(delegate) - self.invalidate() + buffer = getattr(getattr(self, "_session", None), "default_buffer", None) + document = buffer.document if buffer is not None else Document() + self._dispatch(ModalDetached(delegate, document)) def running_prompt_accepts_submission(self) -> bool: delegate = self._active_prompt_delegate() @@ -4288,7 +4306,11 @@ async def _prompt_once(self, *, append_history: bool | None) -> UserInput: self._staged_suggestion_prefill = None with patch_stdout(raw=True): command = str( - await self._session.prompt_async(placeholder=placeholder, default=default) + await self._session.prompt_async( + placeholder=placeholder, + default=default, + set_exception_handler=False, + ) ).strip() command = command.replace("\x00", "") # just in case null bytes are somehow inserted # Sanitize UTF-16 surrogates that may come from Windows clipboard diff --git a/src/pythinker_code/ui/shell/prompting/lifecycle.py b/src/pythinker_code/ui/shell/prompting/lifecycle.py new file mode 100644 index 00000000..69c47602 --- /dev/null +++ b/src/pythinker_code/ui/shell/prompting/lifecycle.py @@ -0,0 +1,77 @@ +"""Awaited lifetime management for prompt-session background resources.""" + +from __future__ import annotations + +import asyncio +from collections.abc import Awaitable, Callable, Coroutine +from typing import Any + +from pythinker_code.utils.logging import logger + +AsyncCloser = Callable[[], Awaitable[None]] + + +class PromptLifecycle: + """Own tasks and asynchronous cleanup belonging to one prompt session.""" + + def __init__(self) -> None: + self._tasks: list[asyncio.Task[None]] = [] + self._closers: list[tuple[str, AsyncCloser]] = [] + self._closed = False + self._close_complete = asyncio.Event() + + def create_task(self, coro: Coroutine[Any, Any, None]) -> asyncio.Task[None]: + """Create lifecycle-owned work, or explicitly refuse it after shutdown.""" + if self._closed: + coro.close() + raise RuntimeError("prompt lifecycle is closed") + task = asyncio.create_task(coro) + self._tasks.append(task) + return task + + def register_closer(self, name: str, async_closer: AsyncCloser) -> None: + """Register an async resource closer to run during shutdown.""" + if self._closed: + raise RuntimeError(f"prompt lifecycle is closed; cannot register closer {name!r}") + self._closers.append((name, async_closer)) + + async def aclose(self) -> None: + """Cancel all work, await it, then close resources in reverse order.""" + if self._closed: + await self._close_complete.wait() + return + self._closed = True + try: + for task in self._tasks: + if not task.done(): + task.cancel() + results = await asyncio.gather(*self._tasks, return_exceptions=True) + for task, result in zip(self._tasks, results, strict=True): + if isinstance(result, BaseException) and not isinstance( + result, asyncio.CancelledError + ): + logger.warning( + "Prompt lifecycle task failed during shutdown: task={} error={!r}", + task.get_name(), + result, + ) + + for name, closer in reversed(self._closers): + try: + await closer() + except asyncio.CancelledError: + # Cancellation here means aclose() itself was cancelled (e.g. + # a wait_for timeout), not a closer's own internal cancel — + # propagate it instead of swallowing and closing on regardless. + logger.warning( + "Prompt lifecycle aclose cancelled while closing resource={}", name + ) + raise + except Exception as exc: + logger.warning( + "Prompt lifecycle resource failed during shutdown: resource={} error={!r}", + name, + exc, + ) + finally: + self._close_complete.set() diff --git a/src/pythinker_code/ui/shell/prompting/state.py b/src/pythinker_code/ui/shell/prompting/state.py new file mode 100644 index 00000000..a2295e2b --- /dev/null +++ b/src/pythinker_code/ui/shell/prompting/state.py @@ -0,0 +1,322 @@ +"""Pure state transitions for the interactive shell prompt.""" + +from __future__ import annotations + +from dataclasses import dataclass, replace +from enum import Enum +from typing import Protocol + +from prompt_toolkit.document import Document +from prompt_toolkit.formatted_text import AnyFormattedText +from prompt_toolkit.key_binding import KeyPressEvent + + +class PromptMode(Enum): + AGENT = "agent" + SHELL = "shell" + + def toggle(self) -> PromptMode: + return PromptMode.SHELL if self is PromptMode.AGENT else PromptMode.AGENT + + def __str__(self) -> str: + return self.value + + +class RunningPromptDelegate(Protocol): + """A component that can take over the bottom prompt area.""" + + modal_priority: int + + def render_running_prompt_body(self, columns: int) -> AnyFormattedText: ... + + def running_prompt_placeholder(self) -> AnyFormattedText | None: ... + + def running_prompt_allows_text_input(self) -> bool: ... + + def running_prompt_hides_input_buffer(self) -> bool: ... + + def running_prompt_accepts_submission(self) -> bool: ... + + def should_handle_running_prompt_key(self, key: str) -> bool: ... + + def handle_running_prompt_key(self, key: str, event: KeyPressEvent) -> None: ... + + +class PromptPhase(Enum): + IDLE = "idle" + TURN_STARTING = "turn_starting" + RUNNING = "running" + + +@dataclass(frozen=True, slots=True) +class PromptState: + mode: PromptMode = PromptMode.AGENT + phase: PromptPhase = PromptPhase.IDLE + running_delegate: RunningPromptDelegate | None = None + modal_stack: tuple[ModalState, ...] = () + suspended_document: Document | None = None + shortcut_help_open: bool = False + running_previous_mode: PromptMode | None = None + + def __post_init__(self) -> None: + is_running = self.phase is PromptPhase.RUNNING + has_delegate = self.running_delegate is not None + if is_running != has_delegate: + raise ValueError("a running prompt must have exactly one running delegate") + + +@dataclass(frozen=True, slots=True) +class ModalState: + """Reducer-owned snapshot of modal properties used for prompt transitions.""" + + delegate: RunningPromptDelegate + priority: int + hides_input: bool + + +@dataclass(frozen=True, slots=True) +class TurnStarting: + pass + + +@dataclass(frozen=True, slots=True) +class TurnCleared: + pass + + +@dataclass(frozen=True, slots=True) +class RunningDelegateAttached: + delegate: RunningPromptDelegate + + +@dataclass(frozen=True, slots=True) +class RunningDelegateDetached: + delegate: RunningPromptDelegate + + +@dataclass(frozen=True, slots=True) +class ModalAttached: + delegate: RunningPromptDelegate + priority: int + hides_input: bool + document: Document + + +@dataclass(frozen=True, slots=True) +class ModalDetached: + delegate: RunningPromptDelegate + document: Document + + +@dataclass(frozen=True, slots=True) +class ModeChanged: + mode: PromptMode + + +@dataclass(frozen=True, slots=True) +class ShortcutHelpToggled: + open: bool | None = None + + +@dataclass(frozen=True, slots=True) +class BufferObserved: + document: Document + + +type PromptEvent = ( + TurnStarting + | TurnCleared + | RunningDelegateAttached + | RunningDelegateDetached + | ModalAttached + | ModalDetached + | ModeChanged + | ShortcutHelpToggled + | BufferObserved +) + + +@dataclass(frozen=True, slots=True) +class SelectCompleter: + mode: PromptMode + + +@dataclass(frozen=True, slots=True) +class SetEraseWhenDone: + erase_when_done: bool + + +@dataclass(frozen=True, slots=True) +class SuspendDocument: + document: Document + + +@dataclass(frozen=True, slots=True) +class RestoreDocument: + document: Document + + +@dataclass(frozen=True, slots=True) +class Invalidate: + pass + + +type PromptEffect = ( + SelectCompleter | SetEraseWhenDone | SuspendDocument | RestoreDocument | Invalidate +) + + +@dataclass(frozen=True, slots=True) +class PromptTransition: + state: PromptState + effects: tuple[PromptEffect, ...] = () + + +def _active_modal(stack: tuple[ModalState, ...]) -> ModalState | None: + if not stack: + return None + return max(enumerate(stack), key=lambda item: (item[1].priority, item[0]))[1] + + +def _mode_effects(mode: PromptMode) -> tuple[PromptEffect, ...]: + return SelectCompleter(mode), SetEraseWhenDone(mode is PromptMode.AGENT), Invalidate() + + +def _suspend_restore_effects( + old_active: ModalState | None, + new_active: ModalState | None, + suspended: Document | None, + document: Document, +) -> tuple[Document | None, tuple[PromptEffect, ...]]: + """Compute the suspended-document/effects transition shared by modal + attach and detach: suspend the live input when a hides-input modal takes + over, restore it when the last such modal leaves.""" + old_hides_input = old_active is not None and old_active.hides_input + new_hides_input = new_active is not None and new_active.hides_input + if not old_hides_input and new_hides_input and document.text: + if suspended is None: + return document, (SuspendDocument(document),) + return suspended, () + if old_hides_input and not new_hides_input and suspended is not None: + effects: tuple[PromptEffect, ...] = ( + (RestoreDocument(suspended),) if not document.text else () + ) + return None, effects + return suspended, () + + +def transition(state: PromptState, event: PromptEvent) -> PromptTransition: + """Return the next prompt state and ordered facade effects without doing I/O.""" + if isinstance(event, TurnStarting): + if state.phase is not PromptPhase.IDLE: + return PromptTransition(state) + return PromptTransition(replace(state, phase=PromptPhase.TURN_STARTING), (Invalidate(),)) + + if isinstance(event, TurnCleared): + if state.phase is not PromptPhase.TURN_STARTING: + return PromptTransition(state) + return PromptTransition(replace(state, phase=PromptPhase.IDLE), (Invalidate(),)) + + if isinstance(event, RunningDelegateAttached): + if state.running_delegate is event.delegate: + return PromptTransition(state) + previous_mode = state.running_previous_mode + if state.running_delegate is None: + previous_mode = state.mode + next_state = replace( + state, + mode=PromptMode.AGENT, + phase=PromptPhase.RUNNING, + running_delegate=event.delegate, + running_previous_mode=previous_mode, + ) + return PromptTransition(next_state, _mode_effects(PromptMode.AGENT)) + + if isinstance(event, RunningDelegateDetached): + if state.running_delegate is not event.delegate: + return PromptTransition(state) + mode = state.running_previous_mode or state.mode + next_state = replace( + state, + mode=mode, + phase=PromptPhase.IDLE, + running_delegate=None, + running_previous_mode=None, + ) + return PromptTransition(next_state, _mode_effects(mode)) + + if isinstance(event, ModalAttached): + if any(modal.delegate is event.delegate for modal in state.modal_stack): + return PromptTransition(state) + old_active = _active_modal(state.modal_stack) + stack = ( + *state.modal_stack, + ModalState(event.delegate, event.priority, event.hides_input), + ) + new_active = _active_modal(stack) + suspended, effects = _suspend_restore_effects( + old_active, new_active, state.suspended_document, event.document + ) + next_state = replace( + state, + modal_stack=stack, + suspended_document=suspended, + shortcut_help_open=False, + ) + return PromptTransition(next_state, (*effects, Invalidate())) + + if isinstance(event, ModalDetached): + if not any(modal.delegate is event.delegate for modal in state.modal_stack): + return PromptTransition(state) + old_active = _active_modal(state.modal_stack) + stack = tuple(modal for modal in state.modal_stack if modal.delegate is not event.delegate) + new_active = _active_modal(stack) + suspended, effects = _suspend_restore_effects( + old_active, new_active, state.suspended_document, event.document + ) + next_state = replace(state, modal_stack=stack, suspended_document=suspended) + return PromptTransition(next_state, (*effects, Invalidate())) + + if isinstance(event, ModeChanged): + if state.mode is event.mode: + return PromptTransition(state) + return PromptTransition(replace(state, mode=event.mode), _mode_effects(event.mode)) + + if isinstance(event, ShortcutHelpToggled): + opened = not state.shortcut_help_open if event.open is None else event.open + if opened is state.shortcut_help_open: + return PromptTransition(state) + return PromptTransition(replace(state, shortcut_help_open=opened), (Invalidate(),)) + + # BufferObserved is the only remaining event type; clearing a stale + # suspended document once the buffer is externally repopulated. + if state.suspended_document is not None and event.document.text: + return PromptTransition(replace(state, suspended_document=None)) + return PromptTransition(state) + + +__all__ = ( + "BufferObserved", + "Invalidate", + "ModalAttached", + "ModalDetached", + "ModeChanged", + "ModalState", + "PromptEffect", + "PromptEvent", + "PromptMode", + "PromptPhase", + "PromptState", + "PromptTransition", + "RestoreDocument", + "RunningDelegateAttached", + "RunningDelegateDetached", + "RunningPromptDelegate", + "SelectCompleter", + "SetEraseWhenDone", + "ShortcutHelpToggled", + "SuspendDocument", + "TurnCleared", + "TurnStarting", + "transition", +) diff --git a/tests/ui_and_conv/test_prompt_lifecycle.py b/tests/ui_and_conv/test_prompt_lifecycle.py new file mode 100644 index 00000000..84bba8ed --- /dev/null +++ b/tests/ui_and_conv/test_prompt_lifecycle.py @@ -0,0 +1,196 @@ +"""Prompt-session task and process lifetime regression tests.""" + +from __future__ import annotations + +import asyncio +from typing import Any, cast + +import pytest + +from pythinker_code.ui.shell.prompt import CustomPromptSession +from pythinker_code.ui.shell.prompting.lifecycle import PromptLifecycle +from pythinker_code.ui.shell.statusline import StatusLineCommandRunner + + +@pytest.mark.asyncio +async def test_aclose_cancels_status_refresh_during_sleep() -> None: + session = object.__new__(CustomPromptSession) + session._lifecycle = PromptLifecycle() + session._status_refresh_task = None + session._statusline_runner = None + session._fast_refresh_provider = None + session._app_for_repaint = lambda: None + session._active_prompt_delegate = lambda: None + session._has_background_tasks = lambda: False + + session._start() + task = cast(asyncio.Task[None], session._status_refresh_task) + await asyncio.sleep(0) + await session.aclose() + + assert task.done() + assert session._status_refresh_task is None + + +@pytest.mark.asyncio +async def test_lifecycle_closes_placeholder_tasks_and_resources_in_reverse_order() -> None: + lifecycle = PromptLifecycle() + tasks_started = asyncio.Event() + closed: list[str] = [] + + async def placeholder() -> None: + tasks_started.set() + await asyncio.Event().wait() + + async def close_named(name: str) -> None: + closed.append(name) + + tasks = [lifecycle.create_task(placeholder()) for _ in range(3)] + lifecycle.register_closer("first", lambda: close_named("first")) + lifecycle.register_closer("second", lambda: close_named("second")) + await tasks_started.wait() + + await lifecycle.aclose() + + assert all(task.done() for task in tasks) + assert closed == ["second", "first"] + + +@pytest.mark.asyncio +async def test_prompt_startup_cancelled_before_delegate_attach() -> None: + lifecycle = PromptLifecycle() + startup_reached_wait = asyncio.Event() + delegate: list[object] = [] + + async def startup() -> None: + startup_reached_wait.set() + await asyncio.Event().wait() + delegate.append(object()) + + task = lifecycle.create_task(startup()) + await startup_reached_wait.wait() + await lifecycle.aclose() + + assert task.done() + assert delegate == [] + + +@pytest.mark.asyncio +async def test_repeated_aclose_is_idempotent_and_refuses_new_work() -> None: + lifecycle = PromptLifecycle() + close_calls = 0 + + async def close_resource() -> None: + nonlocal close_calls + close_calls += 1 + + lifecycle.register_closer("resource", close_resource) + await lifecycle.aclose() + await lifecycle.aclose() + + assert close_calls == 1 + + async def refused() -> None: + await asyncio.sleep(0) + + with pytest.raises(RuntimeError, match="closed"): + lifecycle.create_task(refused()) + + +@pytest.mark.asyncio +async def test_aclose_cancellation_propagates_and_halts_remaining_closers() -> None: + # If aclose() itself is cancelled (e.g. a wait_for timeout) while awaiting a + # closer, the cancellation must propagate — not be swallowed so shutdown + # keeps closing the remaining resources past the caller's deadline. + lifecycle = PromptLifecycle() + earlier_closer_ran = False + + async def blocking_closer() -> None: + await asyncio.Event().wait() + + async def earlier_closer() -> None: + nonlocal earlier_closer_ran + earlier_closer_ran = True + + # Closers run in reverse registration order, so "blocking" runs first. + lifecycle.register_closer("earlier", earlier_closer) + lifecycle.register_closer("blocking", blocking_closer) + + with pytest.raises(asyncio.TimeoutError): + await asyncio.wait_for(lifecycle.aclose(), timeout=0.05) + + assert earlier_closer_ran is False + + +class _BlockingStdout: + def __init__(self) -> None: + self.read_started = asyncio.Event() + self._reads = 0 + + async def read(self, _limit: int) -> bytes: + self._reads += 1 + if self._reads == 1: + self.read_started.set() + await asyncio.Event().wait() + return b"" + + +class _FakeProcess: + def __init__(self) -> None: + self.stdout = _BlockingStdout() + self.returncode: int | None = None + self.kill_calls = 0 + self.wait_calls = 0 + + def kill(self) -> None: + if self.returncode is not None: + raise ProcessLookupError + self.kill_calls += 1 + self.returncode = -9 + + async def wait(self) -> int: + self.wait_calls += 1 + assert self.returncode is not None + return self.returncode + + +@pytest.mark.asyncio +async def test_aclose_cancels_statusline_read_and_reaps_child_once( + monkeypatch: pytest.MonkeyPatch, +) -> None: + process = _FakeProcess() + + async def create_process(*_args: Any, **_kwargs: Any) -> asyncio.subprocess.Process: + return cast(asyncio.subprocess.Process, process) + + monkeypatch.setattr(asyncio, "create_subprocess_exec", create_process) + lifecycle = PromptLifecycle() + runner = StatusLineCommandRunner(command="status-command", timeout_ms=1000) + lifecycle.register_closer("statusline command runner", runner.stop) + runner.start() + await process.stdout.read_started.wait() + + await asyncio.wait_for(lifecycle.aclose(), timeout=1) + + assert runner.is_running is False + assert process.kill_calls <= 1 + assert process.wait_calls <= 1 + + +@pytest.mark.asyncio +async def test_prompt_once_disables_prompt_toolkit_exception_handler() -> None: + prompt_session = object.__new__(CustomPromptSession) + prompt_session._running_prompt_delegate = None + prompt_session._tip_rotation_index = 0 + captured: dict[str, Any] = {} + + class _Session: + async def prompt_async(self, **kwargs: Any) -> str: + captured.update(kwargs) + return "hello" + + prompt_session._session = cast(Any, _Session()) + cast(Any, prompt_session)._build_user_input = lambda command: command + + assert await prompt_session._prompt_once(append_history=False) == "hello" + assert captured["set_exception_handler"] is False diff --git a/tests/ui_and_conv/test_prompt_state.py b/tests/ui_and_conv/test_prompt_state.py new file mode 100644 index 00000000..943be32a --- /dev/null +++ b/tests/ui_and_conv/test_prompt_state.py @@ -0,0 +1,398 @@ +from __future__ import annotations + +from dataclasses import FrozenInstanceError +from typing import Any, cast + +import pytest +from prompt_toolkit.document import Document + +from pythinker_code.ui.shell.prompt import CustomPromptSession +from pythinker_code.ui.shell.prompting.state import ( + BufferObserved, + Invalidate, + ModalAttached, + ModalDetached, + ModalState, + ModeChanged, + PromptMode, + PromptPhase, + PromptState, + RestoreDocument, + RunningDelegateAttached, + RunningDelegateDetached, + SelectCompleter, + SetEraseWhenDone, + ShortcutHelpToggled, + SuspendDocument, + TurnCleared, + TurnStarting, + transition, +) + + +class _Delegate: + def __init__(self, *, priority: int = 0, hides_input: bool = False) -> None: + self.modal_priority = priority + self._hides_input = hides_input + self.visibility_queries = 0 + + def running_prompt_hides_input_buffer(self) -> bool: + self.visibility_queries += 1 + return self._hides_input + + def render_running_prompt_body(self, columns: int) -> str: + return str(columns) + + def running_prompt_placeholder(self) -> None: + return None + + def running_prompt_allows_text_input(self) -> bool: + return not self._hides_input + + def running_prompt_accepts_submission(self) -> bool: + return False + + def should_handle_running_prompt_key(self, key: str) -> bool: + return False + + def handle_running_prompt_key(self, key: str, event: Any) -> None: + return None + + +def _modal_attached(delegate: _Delegate, document: Document | None = None) -> ModalAttached: + return ModalAttached( + delegate=delegate, + priority=delegate.modal_priority, + hides_input=delegate._hides_input, + document=document or Document(), + ) + + +def test_state_and_transition_are_frozen() -> None: + result = transition(PromptState(), TurnStarting()) + + with pytest.raises(FrozenInstanceError): + result.state.phase = PromptPhase.IDLE # type: ignore[misc] + with pytest.raises(FrozenInstanceError): + result.effects = () # type: ignore[misc] + + +@pytest.mark.parametrize( + ("phase", "delegate"), + [ + (PromptPhase.IDLE, _Delegate()), + (PromptPhase.TURN_STARTING, _Delegate()), + (PromptPhase.RUNNING, None), + ], +) +def test_phase_and_running_delegate_invariant( + phase: PromptPhase, delegate: _Delegate | None +) -> None: + with pytest.raises(ValueError, match="exactly one"): + PromptState(phase=phase, running_delegate=delegate) + + +def test_turn_starting_and_clear_transition_table() -> None: + idle = PromptState() + starting = transition(idle, TurnStarting()) + assert starting.state.phase is PromptPhase.TURN_STARTING + assert starting.effects == (Invalidate(),) + + repeated_start = transition(starting.state, TurnStarting()) + assert repeated_start.state is starting.state + assert repeated_start.effects == () + + cleared = transition(starting.state, TurnCleared()) + assert cleared.state.phase is PromptPhase.IDLE + assert cleared.effects == (Invalidate(),) + + for state in (idle, PromptState(phase=PromptPhase.RUNNING, running_delegate=_Delegate())): + stale_clear = transition(state, TurnCleared()) + assert stale_clear.state is state + assert stale_clear.effects == () + + running = PromptState(phase=PromptPhase.RUNNING, running_delegate=_Delegate()) + stale_start = transition(running, TurnStarting()) + assert stale_start.state is running + assert stale_start.effects == () + + +def test_running_delegate_transition_table_and_effect_order() -> None: + first = _Delegate() + replacement = _Delegate() + stale = _Delegate() + idle = PromptState(mode=PromptMode.SHELL) + + attached = transition(idle, RunningDelegateAttached(first)) + assert attached.state.phase is PromptPhase.RUNNING + assert attached.state.running_delegate is first + assert attached.state.running_previous_mode is PromptMode.SHELL + assert attached.effects == ( + SelectCompleter(PromptMode.AGENT), + SetEraseWhenDone(True), + Invalidate(), + ) + + duplicate = transition(attached.state, RunningDelegateAttached(first)) + assert duplicate.state is attached.state + assert duplicate.effects == () + + stale_detach = transition(attached.state, RunningDelegateDetached(stale)) + assert stale_detach.state is attached.state + assert stale_detach.effects == () + + replaced = transition(attached.state, RunningDelegateAttached(replacement)) + assert replaced.state.running_delegate is replacement + assert replaced.state.running_previous_mode is PromptMode.SHELL + assert replaced.effects == ( + SelectCompleter(PromptMode.AGENT), + SetEraseWhenDone(True), + Invalidate(), + ) + + detached = transition(replaced.state, RunningDelegateDetached(replacement)) + assert detached.state == PromptState(mode=PromptMode.SHELL) + assert detached.effects == ( + SelectCompleter(PromptMode.SHELL), + SetEraseWhenDone(False), + Invalidate(), + ) + + +def test_running_delegate_attaches_from_turn_starting() -> None: + delegate = _Delegate() + starting = PromptState(mode=PromptMode.SHELL, phase=PromptPhase.TURN_STARTING) + + attached = transition(starting, RunningDelegateAttached(delegate)) + + assert attached.state.phase is PromptPhase.RUNNING + assert attached.state.running_delegate is delegate + assert attached.state.running_previous_mode is PromptMode.SHELL + assert attached.effects == ( + SelectCompleter(PromptMode.AGENT), + SetEraseWhenDone(True), + Invalidate(), + ) + + +def test_mode_changed_transition_table_and_effect_order() -> None: + agent = PromptState() + unchanged = transition(agent, ModeChanged(PromptMode.AGENT)) + assert unchanged.state is agent + assert unchanged.effects == () + + changed = transition(agent, ModeChanged(PromptMode.SHELL)) + assert changed.state.mode is PromptMode.SHELL + assert changed.effects == ( + SelectCompleter(PromptMode.SHELL), + SetEraseWhenDone(False), + Invalidate(), + ) + + +@pytest.mark.parametrize("phase", [PromptPhase.IDLE, PromptPhase.TURN_STARTING]) +def test_mode_changed_preserves_non_running_phase(phase: PromptPhase) -> None: + state = PromptState(phase=phase) + + changed = transition(state, ModeChanged(PromptMode.SHELL)) + + assert changed.state.phase is phase + assert changed.state.mode is PromptMode.SHELL + assert changed.effects == ( + SelectCompleter(PromptMode.SHELL), + SetEraseWhenDone(False), + Invalidate(), + ) + + +def test_mode_changed_while_running_preserves_delegate_and_previous_mode() -> None: + delegate = _Delegate() + state = PromptState( + phase=PromptPhase.RUNNING, + running_delegate=delegate, + running_previous_mode=PromptMode.SHELL, + ) + + changed = transition(state, ModeChanged(PromptMode.SHELL)) + + assert changed.state.phase is PromptPhase.RUNNING + assert changed.state.running_delegate is delegate + assert changed.state.running_previous_mode is PromptMode.SHELL + assert changed.effects == ( + SelectCompleter(PromptMode.SHELL), + SetEraseWhenDone(False), + Invalidate(), + ) + + unchanged = transition(changed.state, ModeChanged(PromptMode.SHELL)) + assert unchanged.state is changed.state + assert unchanged.effects == () + + +def test_shortcut_help_transition_table() -> None: + closed = PromptState() + opened = transition(closed, ShortcutHelpToggled()) + assert opened.state.shortcut_help_open is True + assert opened.effects == (Invalidate(),) + + already_open = transition(opened.state, ShortcutHelpToggled(open=True)) + assert already_open.state is opened.state + assert already_open.effects == () + + closed_again = transition(opened.state, ShortcutHelpToggled(open=False)) + assert closed_again.state.shortcut_help_open is False + assert closed_again.effects == (Invalidate(),) + + already_closed = transition(closed_again.state, ShortcutHelpToggled(open=False)) + assert already_closed.state is closed_again.state + assert already_closed.effects == () + + +def test_modal_priority_latest_tie_break_duplicate_and_purity() -> None: + visible_high = _Delegate(priority=20) + hidden_low = _Delegate(priority=10, hides_input=True) + hidden_tie = _Delegate(priority=20, hides_input=True) + document = Document("draft") + + visible = transition(PromptState(), _modal_attached(visible_high, document)) + low = transition(visible.state, _modal_attached(hidden_low, document)) + assert not any(isinstance(effect, SuspendDocument) for effect in low.effects) + + tie = transition(low.state, _modal_attached(hidden_tie, document)) + assert tie.effects == (SuspendDocument(document), Invalidate()) + assert tuple(modal.delegate for modal in tie.state.modal_stack) == ( + visible_high, + hidden_low, + hidden_tie, + ) + assert visible_high.visibility_queries == 0 + assert hidden_low.visibility_queries == 0 + assert hidden_tie.visibility_queries == 0 + + duplicate = transition(tie.state, _modal_attached(hidden_tie, document)) + assert duplicate.state is tie.state + assert duplicate.effects == () + + +def test_hidden_modal_suspends_and_restores_document_in_effect_order() -> None: + modal = _Delegate(priority=1, hides_input=True) + document = Document("draft", cursor_position=5) + attached = transition(PromptState(), _modal_attached(modal, document)) + assert attached.state.suspended_document == document + assert attached.effects == (SuspendDocument(document), Invalidate()) + + detached = transition(attached.state, ModalDetached(modal, Document())) + assert detached.effects == (RestoreDocument(document), Invalidate()) + assert detached.state.suspended_document is None + + stale = transition(detached.state, ModalDetached(modal, Document())) + assert stale.state is detached.state + assert stale.effects == () + + +def test_hidden_modal_with_empty_document_does_not_suspend() -> None: + modal = _Delegate(priority=1, hides_input=True) + + attached = transition(PromptState(), _modal_attached(modal, Document())) + + assert attached.state.suspended_document is None + assert attached.effects == (Invalidate(),) + + +def test_hidden_modal_detach_drops_suspension_when_buffer_is_non_empty() -> None: + modal = _Delegate(priority=1, hides_input=True) + attached = transition(PromptState(), _modal_attached(modal, Document("original"))) + + detached = transition(attached.state, ModalDetached(modal, Document("replacement"))) + + assert detached.state.suspended_document is None + assert detached.effects == (Invalidate(),) + + +def test_modal_active_visibility_transitions_in_both_directions() -> None: + hidden = _Delegate(priority=10, hides_input=True) + visible = _Delegate(priority=20) + draft = Document("draft") + + hidden_active = transition(PromptState(), _modal_attached(hidden, draft)) + visible_active = transition(hidden_active.state, _modal_attached(visible, Document())) + assert visible_active.effects == (RestoreDocument(draft), Invalidate()) + assert visible_active.state.suspended_document is None + + hidden_revealed = transition( + visible_active.state, + ModalDetached(visible, draft), + ) + assert hidden_revealed.effects == (SuspendDocument(draft), Invalidate()) + assert hidden_revealed.state.suspended_document == draft + + +def test_detaching_inactive_modal_keeps_active_visibility() -> None: + hidden_low = _Delegate(priority=10, hides_input=True) + visible_high = _Delegate(priority=20) + document = Document("draft") + low = transition(PromptState(), _modal_attached(hidden_low, document)) + high = transition(low.state, _modal_attached(visible_high, Document())) + + detached = transition(high.state, ModalDetached(hidden_low, document)) + + assert detached.state.modal_stack == (ModalState(visible_high, 20, False),) + assert detached.state.suspended_document is None + assert detached.effects == (Invalidate(),) + + +def test_modal_attach_closes_shortcut_help() -> None: + state = transition(PromptState(), ShortcutHelpToggled(open=True)).state + attached = transition(state, _modal_attached(_Delegate())) + assert attached.state.shortcut_help_open is False + assert attached.effects == (Invalidate(),) + + +def test_buffer_observed_transition_table_prevents_stale_restore() -> None: + modal = _Delegate(priority=1, hides_input=True) + state = transition(PromptState(), _modal_attached(modal, Document("original"))).state + + empty = transition(state, BufferObserved(Document())) + assert empty.state is state + assert empty.effects == () + + observed = transition(state, BufferObserved(Document("replacement"))) + assert observed.state.suspended_document is None + assert observed.effects == () + + repeated = transition(observed.state, BufferObserved(Document("replacement"))) + assert repeated.state is observed.state + assert repeated.effects == () + + detached = transition(observed.state, ModalDetached(modal, Document("replacement"))) + assert detached.effects == (Invalidate(),) + assert not any(isinstance(effect, RestoreDocument) for effect in detached.effects) + + +def test_session_dispatch_keeps_reducer_modal_snapshot_authoritative() -> None: + delegate = _Delegate(priority=1, hides_input=False) + session = object.__new__(CustomPromptSession) + facade = cast(Any, session) + facade._prompt_state = PromptState(modal_stack=(ModalState(delegate, 1, False),)) + facade._modal_delegates = [delegate] + facade._mode = PromptMode.AGENT + facade._shortcut_help_open = False + facade._turn_starting = False + facade._running_prompt_delegate = None + facade._running_prompt_previous_mode = None + facade._suspended_buffer_document = None + invalidations = 0 + + def invalidate() -> None: + nonlocal invalidations + invalidations += 1 + + facade.invalidate = invalidate + delegate._hides_input = True + + session.toggle_shortcut_help() + + assert session._prompt_state.modal_stack == (ModalState(delegate, 1, False),) + assert session._prompt_state.suspended_document is None + assert delegate.visibility_queries == 0 + assert invalidations == 1 diff --git a/tests/ui_and_conv/test_prompt_tips.py b/tests/ui_and_conv/test_prompt_tips.py index f7f72885..048df56e 100644 --- a/tests/ui_and_conv/test_prompt_tips.py +++ b/tests/ui_and_conv/test_prompt_tips.py @@ -123,22 +123,6 @@ def test_other_unraisable_exceptions_are_not_filtered() -> None: assert not shell_prompt._is_prompt_toolkit_keyprocessor_shutdown_noise(unraisable) -def test_prompt_toolkit_empty_exception_context_is_filtered() -> None: - assert shell_prompt._is_prompt_toolkit_empty_exception_context({"exception": None}) - assert shell_prompt._is_prompt_toolkit_empty_exception_context( - {"exception": None, "message": "Task was destroyed but it is pending!"} - ) - - -def test_prompt_toolkit_real_exception_context_is_not_filtered() -> None: - assert not shell_prompt._is_prompt_toolkit_empty_exception_context( - {"exception": RuntimeError("boom")} - ) - assert not shell_prompt._is_prompt_toolkit_empty_exception_context( - {"exception": None, "message": "unexpected loop failure"} - ) - - class _DummyRunningPrompt: modal_priority = 10 diff --git a/tests/ui_and_conv/test_shell_run_placeholders.py b/tests/ui_and_conv/test_shell_run_placeholders.py index 8264eeb8..d1223d4e 100644 --- a/tests/ui_and_conv/test_shell_run_placeholders.py +++ b/tests/ui_and_conv/test_shell_run_placeholders.py @@ -40,6 +40,12 @@ def __enter__(self) -> _FakePromptSession: def __exit__(self, exc_type, exc, tb) -> bool: return False + async def __aenter__(self) -> _FakePromptSession: + return self + + async def __aexit__(self, exc_type, exc, tb) -> bool: + return False + async def prompt_next(self) -> UserInput: self.prompt_calls += 1 response = _FakePromptSession.responses.popleft() diff --git a/tests/ui_and_conv/test_statusline.py b/tests/ui_and_conv/test_statusline.py index d5fe6b59..2963ffc6 100644 --- a/tests/ui_and_conv/test_statusline.py +++ b/tests/ui_and_conv/test_statusline.py @@ -229,6 +229,27 @@ async def test_command_runner_lifecycle_start_stop(): assert runner.is_running is False +@pytest.mark.asyncio +async def test_command_runner_repeated_start_and_stop_are_idempotent(monkeypatch): + runner = StatusLineCommandRunner(command="echo hi", timeout_ms=5000) + refresh_started = asyncio.Event() + + async def blocking_refresh(): + refresh_started.set() + await asyncio.Event().wait() + + monkeypatch.setattr(runner, "refresh_once", blocking_refresh) + runner.start() + first_task = runner._task + runner.start() + await refresh_started.wait() + + assert runner._task is first_task + await runner.stop() + await runner.stop() + assert runner.is_running is False + + # --------------------------------------------------------------------------- # Footer render integration # ---------------------------------------------------------------------------