input.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_ui/terminal/input.py
 
"""
Input layer for the OC Chat REPL (Phase 2 PR 2.A + 2.B).
 
Purpose
-------
:class:`PromptDriver` is the single entry point for reading the next
user line. At construction it tries to import ``prompt_toolkit``; if
that succeeds (and ``OC_CHAT_LEGACY_PROMPT`` is not set), it builds a
:class:`prompt_toolkit.PromptSession` with file-backed history,
context-aware tab completion (see
:class:`~oc_ui.terminal.completers.OCChatCompleter`), an
``Alt+Enter`` keybinding that inserts a literal newline, a
``bottom_toolbar`` showing live session stats, an ``rprompt`` showing
token count, and a ``Ctrl+P`` binding that opens a fuzzy command
palette. Otherwise it falls back to plain :func:`input` — same
prompt format, no behaviour change.
 
The driver runs **between turns only**. Rich's ``Live`` region owns
the terminal during streaming; ``prompt_toolkit`` owns it during
input. They never overlap, so the two libraries coexist without
fighting over the screen.
 
Responsibilities
----------------
- :class:`PromptDriver` — wraps prompt-toolkit (or legacy
  :func:`input`) and exposes a single ``read_line`` call.
- :func:`build_prompt_prefix` — formats the ``[<model>] You: `` string used by
  both paths. The session NAME is deliberately not in the typing line; it is
  shown on the bottom status toolbar instead (:func:`_build_bottom_toolbar`).
  Both surfaces report whichever session ``read_user_input`` is handed, so the
  caller must pass the CURRENT session — ``ctx.session``, never a loop-local
  bound at startup, which a slash command may since have replaced (plan 152 D7).
- :class:`TurnStats` — thread-safe shared state used to communicate
  token counts from the streaming thread to the rprompt.
 
Diagnostics
-----------
Domain: OC_UI.TERMINAL
Levels:
    L2 — lifecycle events
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- ``PromptDriver.read_line`` is blocking; the caller is responsible
  for backgrounding it if needed.
- **Chat prompts are MULTI-LINE (multiline-prompt fix, 2026-07-18):**
  when ``read_user_input`` is called without a ``prompt_text`` override
  (the main chat prompt), Enter inserts a newline and submission is
  **Ctrl+Enter** (arrives as ``c-j`` on Windows consoles — incl.
  Windows Terminal / VS Code via ConPTY), with ``Ctrl+J`` and
  ``Alt+Enter`` as the portable equivalents on terminals that cannot
  distinguish Ctrl+Enter from Enter. A multi-line paste therefore lands
  in the buffer as ONE prompt — nothing auto-submits. Up/Down move the
  cursor within a multi-line buffer and navigate history from the
  first/last line (prompt_toolkit ``auto_up``/``auto_down`` defaults).
- **Menu/startup prompts stay single-line:** any call passing
  ``prompt_text=`` keeps the classic Enter-submits behaviour, so
  numbered menus answer on a single keypress+Enter as before.
- ``TurnStats`` mutations are all guarded by an internal lock; reads
  return a consistent snapshot.
- The legacy ``input()`` fallback path remains SINGLE-line (Python's
  ``input`` cannot compose multi-line); it is otherwise functionally
  equivalent.
- ``OC_CHAT_LEGACY_PROMPT`` is honoured at construction; toggling it
  mid-session has no effect.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_ui.terminal.statusline (StatusBarState, render_status_line and token formatting for the bottom toolbar and rprompt)
- oc_ui.terminal.turn_layout (TURN_INDENT for the prompt prefix)
- oc_chat.core.cache_usage (lazy; cache hit ratio and prompt cache hit/miss split for turn stats)
- oc_chat.core.commands (lazy; COMMANDS for the Ctrl+P command palette)
- oc_chat.core.persistence (lazy; list_sessions for the Ctrl+P command palette)
- oc_chat.core.session (lazy; ChatSession type hint for the prompt prefix and read_user_input)
- oc_ui.terminal.completers (lazy; OCChatCompleter for tab completion)
- prompt_toolkit (lazy; PromptSession for the interactive prompt)
- prompt_toolkit.completion (lazy; FuzzyCompleter and WordCompleter for the Ctrl+P palette)
- prompt_toolkit.filters (lazy; Condition gating the multi-line chat key bindings)
- prompt_toolkit.formatted_text (lazy; FormattedText for the coloured chat prompt)
- prompt_toolkit.history (lazy; FileHistory for persistent input history)
- prompt_toolkit.key_binding (lazy; KeyBindings for the newline, submit, cancel and palette keys)
"""
from __future__ import annotations
 
import os
import sys
import time
from pathlib import Path
from threading import Lock
from typing import TYPE_CHECKING, Any, Callable
 
from oc_diagnostics import _dbg
 
from oc_ui.terminal.statusline import (
    StatusBarState,
    busy_hint_text,
    fmt_tokens_short,
    render_status_line,
)
from oc_ui.terminal.turn_layout import TURN_INDENT
 
if TYPE_CHECKING:
    from oc_chat.core.session import ChatSession
 
# Lazy import — keeps oc_ui independent of oc_chat at import time.
# (Full DI via UIContext is deferred to Phase C.)
def _cache_usage_fns():
    from oc_chat.core.cache_usage import cache_hit_ratio_pct, usage_prompt_cache_hit_miss
    return cache_hit_ratio_pct, usage_prompt_cache_hit_miss
 
 
# ── TurnStats: thread-safe shared state for rprompt ─────────────────────────
 
class TurnStats:
    """Thread-safe accumulator for in-flight turn token counts.
 
    ``PromptDriver.read_user_input`` reads these via
    :meth:`rprompt_text` to show a live ``→ NNN tok`` counter in the
    right prompt.  The main thread's ``run_turn`` writes them.
 
    Plan 147 D1: also the sole owner of turn/session wall-clock timing
    and :class:`~oc_ui.terminal.statusline.StatusBarState` production —
    see :meth:`build_status`. ``clock`` is injectable so tests never
    need ``sleep()``-based timing assertions.
 
    Thread safety: all public methods acquire a :class:`Lock`.
    """
 
    def __init__(self, clock: Callable[[], float] = time.monotonic) -> None:
        self._lock = Lock()
        self._clock = clock
        self._prompt_tokens: int = 0
        self._completion_tokens: int = 0
        self._turn_total: int = 0
        self._session_total: int = 0
        self._active: bool = False
        self._cache_hit: int = 0
        self._cache_miss: int = 0
        self._turn_iterations: int = 0
        self._turn_tool_calls: int = 0
        self._session_cache_pct: float | None = None
        self._session_cache_saved: int = 0
        # ── Plan 147 D1: wall-clock timing (monotonic; never sleep()-tested) ──
        self._session_start: float = clock()
        self._turn_start: float | None = None
        self._last_turn_elapsed_ms: int = 0
        self._last_bound_session_id: str | None = None
 
    def start_turn(self) -> None:
        """Call at the beginning of a chat turn (resets per-turn counters)."""
        with self._lock:
            self._prompt_tokens = 0
            self._completion_tokens = 0
            self._turn_total = 0
            self._cache_hit = 0
            self._cache_miss = 0
            self._turn_iterations = 0
            self._turn_tool_calls = 0
            self._session_cache_pct = None
            self._session_cache_saved = 0
            self._active = True
            self._turn_start = self._clock()
 
    def add_usage(self, usage: dict) -> None:
        """Accumulate token usage from a single iteration."""
        with self._lock:
            pt = usage.get("prompt_tokens", 0)
            ct = usage.get("completion_tokens", 0)
            self._prompt_tokens += pt
            self._completion_tokens += ct
            self._turn_total += usage.get("total_tokens", pt + ct)
            _, usage_prompt_cache_hit_miss = _cache_usage_fns()
            ch, cm = usage_prompt_cache_hit_miss(usage)
            self._cache_hit += ch
            self._cache_miss += cm
 
    def add_iteration(self) -> None:
        """Increment the per-turn iteration counter."""
        with self._lock:
            self._turn_iterations += 1
 
    def add_tool_calls(self, n: int) -> None:
        """Increment the per-turn tool-call counter by *n*."""
        with self._lock:
            self._turn_tool_calls += n
 
    def finish_turn(self, session_total: int, iterations: int = 0,
                    tool_calls: int = 0, session_cache_pct: float | None = None,
                    session_cache_saved: int = 0) -> None:
        """Call at the end of a turn; sets the session total and marks idle.
 
        Freezes :attr:`_last_turn_elapsed_ms` from this instance's OWN
        ``_turn_start`` (set in :meth:`start_turn`) — self-contained and
        independent of any duration a caller may separately compute for
        ``show_turn_summary`` (the D1 summary-ordering pin: that value is
        never derived from ``finish_turn`` state, nor vice versa).
        """
        with self._lock:
            self._session_total = session_total
            self._turn_iterations = iterations
            self._turn_tool_calls = tool_calls
            self._session_cache_pct = session_cache_pct
            self._session_cache_saved = session_cache_saved
            if self._turn_start is not None:
                self._last_turn_elapsed_ms = int((self._clock() - self._turn_start) * 1000)
            self._turn_start = None
            self._active = False
 
    def rebind_session(self, session_id: str) -> None:
        """Reset the session-elapsed clock when the active session identity changes.
 
        A no-op when *session_id* matches the last-bound id — called
        uniformly from every :func:`~oc_chat.core.commands.refresh_presenter_bindings`
        call site (including the harmless every-turn rebind), so it must
        stay idempotent rather than resetting unconditionally.
        """
        with self._lock:
            if session_id != self._last_bound_session_id:
                self._session_start = self._clock()
                self._last_bound_session_id = session_id
 
    def build_status(
        self, *, model: str, session_name: str, effort: str, session_tools: int,
        session_grant: "str | None" = None,
        session_cost_usd: "float | None" = None,
        unpriced_samples: int = 0,
        checkpoint_auto: bool = False,
    ) -> StatusBarState:
        """Produce a :class:`~oc_ui.terminal.statusline.StatusBarState` snapshot.
 
        *model* / *session_name* / *effort* / *session_tools* are read by
        the caller from the live ``ChatSession`` at call time (late-bound —
        never captured) since :class:`TurnStats` itself only tracks
        token/timing aggregates, not session identity fields. *session_grant*
        (plan 235) is read the same way from the process grant state, and
        defaults to ``None`` so every pre-235 caller is unchanged.
 
        *session_cost_usd* / *unpriced_samples* / *checkpoint_auto* (plan 237)
        follow the same rule: read at CALL time from the live session's cost
        fold and the process checkpoint latches, all defaulted so a pre-237
        caller produces a byte-identical snapshot.
        """
        with self._lock:
            now = self._clock()
            if self._active and self._turn_start is not None:
                turn_elapsed_ms = int((now - self._turn_start) * 1000)
            else:
                turn_elapsed_ms = self._last_turn_elapsed_ms
            session_elapsed_ms = int((now - self._session_start) * 1000)
            cache_hit_ratio_pct, _ = _cache_usage_fns()
            turn_cache_pct = cache_hit_ratio_pct(self._cache_hit, self._cache_miss)
            return StatusBarState(
                model=model,
                session_name=session_name,
                effort=effort,
                turn_tokens=self._turn_total,
                turn_iterations=self._turn_iterations,
                turn_tools=self._turn_tool_calls,
                turn_cache_hit_pct=turn_cache_pct,
                turn_cache_saved_tokens=self._cache_hit,
                turn_elapsed_ms=turn_elapsed_ms,
                session_tokens=self._session_total,
                session_tools=session_tools,
                session_cache_hit_pct=self._session_cache_pct,
                session_cache_saved_tokens=self._session_cache_saved,
                session_elapsed_ms=session_elapsed_ms,
                session_grant=session_grant,
                session_cost_usd=session_cost_usd,
                unpriced_samples=unpriced_samples,
                checkpoint_auto=checkpoint_auto,
            )
 
    def rprompt_text(self) -> str:
        """Return the right-prompt string for the current state.
 
        Returns ``"→ NNN tok"`` during an active turn, or
        ``"↻ NNNk / NNNk"`` (this-turn / session) between turns.
        """
        with self._lock:
            cache_hit_ratio_pct, _ = _cache_usage_fns()
            if self._active and self._turn_total > 0:
                line = f"\u2192 {fmt_tokens_short(self._turn_total)} tok"
                pct = cache_hit_ratio_pct(self._cache_hit, self._cache_miss)
                if pct is not None:
                    line += f" · {pct:.0f}% cache"
                return line
            if self._session_total > 0 or self._turn_total > 0:
                st = self._session_total or 0
                tt = self._turn_total or 0
                return f"\u21bb {fmt_tokens_short(tt)} / {fmt_tokens_short(st)}"
            return ""
 
    @property
    def active(self) -> bool:
        with self._lock:
            return self._active
 
    @property
    def totals(self) -> tuple[int, int]:
        """Return ``(turn_total, session_total)`` atomically.
 
        Kept for direct unit testing of the accumulator; ``_build_bottom_toolbar``
        renders through :meth:`build_status` and the shared statusline
        formatter instead (plan 147 D1/D3).
        """
        with self._lock:
            return (self._turn_total, self._session_total)
 
    @property
    def toolbar_data(self) -> tuple[int, int, int, int, float | None, float | None, int, int]:
        """Return ``(turn_total, session_total, turn_iterations, turn_tool_calls,
        turn_cache_pct, session_cache_pct, turn_cache_saved, session_cache_saved)`` atomically.
 
        Kept for direct unit testing of the accumulator; ``_build_bottom_toolbar``
        renders through :meth:`build_status` and the shared statusline
        formatter instead (plan 147 D1/D3).
        """
        with self._lock:
            cache_hit_ratio_pct, _ = _cache_usage_fns()
            turn_cache_pct = cache_hit_ratio_pct(self._cache_hit, self._cache_miss)
            return (
                self._turn_total,
                self._session_total,
                self._turn_iterations,
                self._turn_tool_calls,
                turn_cache_pct,
                self._session_cache_pct,
                self._cache_hit,
                self._session_cache_saved,
            )
 
 
# ── Prompt colour ───────────────────────────────────────────────────────────
# Applied only to the "[model | session] You: " prefix, not to typed text.
# Change _PROMPT_COLOR_PT / _PROMPT_COLOR_ANSI to restyle without hunting
# through logic.
_PROMPT_COLOR_PT   = "fg:#ff87af"       # prompt_toolkit style string (pink)
_PROMPT_COLOR_ANSI = "\033[38;5;211m"   # ANSI 256-colour equivalent
_ANSI_RESET        = "\033[0m"
 
 
# ── Prompt prefix ───────────────────────────────────────────────────────────
 
def build_prompt_prefix(session: ChatSession) -> str:
    """Return the input prompt prefix used by the REPL.
 
    Format:  ``\n   [<model>] You: ``
 
    The session name is intentionally omitted from the typing line — it added
    nothing there (and was truncated). The full session name is shown on the
    bottom status toolbar instead (see :func:`_build_bottom_toolbar`).
 
    Plan 218 T3: the line carries the same three-space turn indent every
    other turn-scoped line does. The prompt echo sits between the previous
    rule and the turn rule that the Enter keypress generates, so indenting
    it is what makes the question read as part of the turn it opens —
    the operator's "it would be even better if the You: line fell within
    the turn boundary". The driver gains ONLY this indent (P4).
    """
    return f"\n{TURN_INDENT}[{session.model}] You: "
 
 
# ── Single-key menu input ────────────────────────────────────────────────────
# Used by the startup/model/confirm menus so a bare ``1``/``2``/``y`` selects
# immediately without an Enter press. Engages only on an interactive TTY; under
# a pipe / test capture / redirected stdin there is no raw mode, so it degrades
# to a single line read (Enter required) — keeping tests and headless runs
# behaviour-identical to the old line-based menus.
 
def _getch_raw() -> str:
    """Block for one keypress and return it (no echo, no Enter needed).
 
    Windows multi-byte function keys (arrows, F-keys) are consumed and reported
    as the sentinel ``"\\x00"`` so the caller's valid-key filter ignores them.
    """
    if os.name == "nt":
        import msvcrt  # Windows-only; imported lazily.
 
        ch = msvcrt.getwch()
        if ch in ("\x00", "\xe0"):  # function-key prefix — consume the scan code
            msvcrt.getwch()
            return "\x00"
        return ch
    import termios
    import tty
 
    fd = sys.stdin.fileno()
    old = termios.tcgetattr(fd)
    try:
        tty.setcbreak(fd)
        return sys.stdin.read(1)
    finally:
        termios.tcsetattr(fd, termios.TCSADRAIN, old)
 
 
def read_single_key(prompt: str = "", *, valid: str = "", fallback=None) -> str:
    """Read a single keypress from an interactive TTY (no Enter required).
 
    Returns the pressed character lower-cased, or ``""`` for Enter/Return — a
    caller treats ``""`` as "accept the default / first option". When *valid* is
    non-empty, keys outside it (and not Enter) are ignored and the read repeats.
    ``Ctrl+C`` raises :class:`KeyboardInterrupt`; ``Ctrl+D`` raises
    :class:`EOFError` — the caller's existing handlers decide what that means.
 
    On a non-interactive stdin there is no raw single-key mode, so this degrades
    to one line read: *fallback* (e.g. a ``prompt_driver.read_user_input``
    closure) when given, else :func:`input`. The first character of the entered
    line is used (empty line -> ``""``).
    """
    func = "read_single_key"
    if prompt:
        sys.stdout.write(prompt)
        sys.stdout.flush()
 
    if not sys.stdin.isatty():
        # No raw mode available (pipe / pytest capture / redirect) — line read.
        if prompt:
            sys.stdout.write("\n")
            sys.stdout.flush()
        try:
            line = fallback() if fallback is not None else input("")
        except (EOFError, KeyboardInterrupt):
            raise
        if line is None:
            return ""
        line = line.strip()
        ch = line[0].lower() if line else ""
        return ch if (not valid or ch in valid) else ""
 
    while True:
        ch = _getch_raw()
        if ch in ("\r", "\n"):
            sys.stdout.write("\n")
            sys.stdout.flush()
            _dbg("OC_UI.TERMINAL", func, "single-key: Enter (default)", 4)
            return ""
        if ch == "\x03":
            raise KeyboardInterrupt
        if ch == "\x04":
            raise EOFError
        ch = ch.lower()
        if not valid or ch in valid:
            sys.stdout.write(ch + "\n")
            sys.stdout.flush()
            _dbg("OC_UI.TERMINAL", func, f"single-key: {ch!r}", 4)
            return ch
        # Unrecognised key — ignore and keep waiting.
 
 
# ── Toolbar helpers ─────────────────────────────────────────────────────────
 
def _build_bottom_toolbar(session, turn_stats: TurnStats | None) -> str:
    """Build the bottom-toolbar string shown at the prompt (idle regime, D3-a).
 
    Renders through the shared :func:`~oc_ui.terminal.statusline.render_status_line`
    formatter — the ONLY bar-string producer; :class:`RichPresenter` renders
    the same formatter for the streaming/dispatch regimes. Format::
 
        model: deepseek-v4-pro │ name: 141421 │ this turn: 1.2K · 3 iters · 2 tools · 75% cache · saved ~900 tok · 12.4s │ session: 430K · 40 tools · 68% cache · 18m 32s │ effort: high   Ctrl+D exit · Ctrl+P palette
    """
    import shutil
 
    session_tools = session.stats.tool_call_count if hasattr(session, "stats") else 0
    if turn_stats is not None:
        state = turn_stats.build_status(
            model=session.model,
            session_name=session.name,
            effort=getattr(session, "reasoning_effort", "high"),
            session_tools=session_tools,
        )
    else:
        # Degraded fallback (no TurnStats bound) — still renders through the
        # shared formatter with the data session.stats alone can provide.
        state = StatusBarState(
            model=session.model,
            session_name=session.name,
            effort=getattr(session, "reasoning_effort", "high"),
            turn_tokens=0, turn_iterations=0, turn_tools=0,
            turn_cache_hit_pct=None, turn_cache_saved_tokens=0, turn_elapsed_ms=0,
            session_tokens=session.stats.total_tokens if hasattr(session, "stats") else 0,
            session_tools=session_tools,
            session_cache_hit_pct=None, session_cache_saved_tokens=0, session_elapsed_ms=0,
        )
 
    width = max(20, shutil.get_terminal_size((80, 24)).columns)
    _segments, line = render_status_line(state, busy_label=None, target_width=width)
    hint = busy_hint_text(busy=False)
    if len(line) + len(hint) + 3 <= width:
        line = f"{line}   {hint}"
    return line
 
 
# ── PromptDriver ────────────────────────────────────────────────────────────
 
class PromptDriver:
    """Read the next user line.
 
    Resolves at construction whether ``prompt_toolkit`` is available.
    If yes (and the legacy escape hatch is not set), uses a
    ``PromptSession`` with history + completer + bottom_toolbar +
    rprompt + Ctrl+P palette.  Otherwise falls back to plain
    :func:`input`.
 
    Construction never raises on missing optional dep.
 
    Parameters
    ----------
    ctx:
        The live :class:`~oc_chat.core.commands.Context`.  The
        completer reads ``ctx.model_manager`` and the live
        ``COMMANDS`` registry.
    history_path:
        Override the default ``~/.oc_chat/history`` location.  Useful
        for tests.
    turn_stats:
        Shared :class:`TurnStats` for live rprompt updates.  ``None``
        disables the live token counter.
    """
 
    def __init__(self, ctx, history_path: Path | None = None,
                 turn_stats: TurnStats | None = None) -> None:
        func = "__init__"
        self._ctx = ctx
        self._turn_stats = turn_stats
        self._last_session = None
        self._cancelled = False
        self._pt: Any = None  # PromptSession when prompt_toolkit is available
        # Multiline-prompt fix (2026-07-18): True only while a CHAT prompt
        # (no prompt_text override) is being read — gates the Enter-inserts-
        # newline / Ctrl+Enter-submits bindings so menus keep Enter-submits.
        self._chat_multiline = False
 
        if os.getenv("OC_CHAT_LEGACY_PROMPT"):
            _dbg("OC_UI.TERMINAL", func, "PromptDriver: legacy prompt mode", 2)
            return
 
        try:
            from prompt_toolkit import PromptSession
            from prompt_toolkit.filters import Condition
            from prompt_toolkit.history import FileHistory
            from prompt_toolkit.key_binding import KeyBindings
            from oc_ui.terminal.completers import OCChatCompleter
        except ImportError:
            return
 
        if history_path is None:
            history_path = Path.home() / ".oc_chat" / "history"
        try:
            history_path.parent.mkdir(parents=True, exist_ok=True)
        except OSError as exc:
            print(f"[history disabled: {exc}]", file=sys.stderr)
            history_path = None
 
        bindings = KeyBindings()
 
        # ── Multiline chat prompt (2026-07-18 fix) ───────────────────────
        # Active ONLY for the main chat prompt (read_user_input without a
        # prompt_text override). Enter inserts a newline — so a multi-line
        # paste can never auto-submit line-by-line — and submission is
        # Ctrl+Enter (delivered as c-j by Windows consoles, incl. Windows
        # Terminal / VS Code via ConPTY) with Ctrl+J / Alt+Enter as the
        # portable equivalents. Up/Down keep prompt_toolkit's auto_up /
        # auto_down defaults: cursor movement inside a multi-line buffer,
        # history navigation from the first/last line.
        in_chat_prompt = Condition(lambda: self._chat_multiline)
 
        @bindings.add("enter", filter=in_chat_prompt)
        def _chat_newline(event):
            event.current_buffer.insert_text("\n")
 
        @bindings.add("c-j", filter=in_chat_prompt)  # Ctrl+Enter / Ctrl+J
        def _chat_submit(event):
            event.current_buffer.validate_and_handle()
 
        @bindings.add("escape", "enter", filter=in_chat_prompt)  # Alt+Enter
        def _chat_submit_alt(event):
            event.current_buffer.validate_and_handle()
 
        @bindings.add("escape", "enter", filter=~in_chat_prompt)
        def _insert_newline(event):
            # Menu/startup prompts keep the pre-fix Alt+Enter behaviour:
            # compose a manual newline inside a single-line prompt.
            event.current_buffer.insert_text("\n")
 
        # ── Ctrl+C — cancel current input (does not exit app) ────────────
        @bindings.add("c-c")
        def _cancel_input(event):
            """Reset the buffer, save partial input to history, and mark as cancelled."""
            self._cancelled = True
            text = event.current_buffer.text
            if text.strip() and self._pt is not None and getattr(self._pt, "history", None):
                self._pt.history.append_string(text)
            event.current_buffer.reset()
            event.app.exit()
 
        # ── Ctrl+P — fuzzy command palette (Phase 2.B) ──────────────────
        try:
            from prompt_toolkit.completion import FuzzyCompleter
        except ImportError:
            FuzzyCompleter = None  # type: ignore[assignment]
 
        if FuzzyCompleter is not None:
            @bindings.add("c-p")
            def _open_palette(event):
                """Open a fuzzy command-palette overlay.
 
                Temporarily swaps the buffer's completer for a
                ``FuzzyCompleter``, opens the completion menu, and
                restores the original completer immediately.
                The completion menu captures the swapped completer
                at open time, so it keeps using the fuzzy matcher
                until dismissed — after which normal Tab completion
                works again.
                """
                func = "_open_palette"
                from oc_chat.core.commands import COMMANDS
                from oc_chat.core.persistence import list_sessions
                palette_words: list[str] = sorted(COMMANDS.keys())
                try:
                    palette_words.extend(
                        f"/session load {s['name']}"
                        for s in list_sessions()
                    )
                except Exception as e:
                    # VALIDATION_ERROR — sessions dir absent / unreadable; degrade to verb-only palette
                    _dbg("OC_UI.TERMINAL", func, f"VALIDATION_ERROR: {type(e).__name__}: {e}", 3)
 
                from prompt_toolkit.completion import WordCompleter
                palette_completer = FuzzyCompleter(
                    WordCompleter(palette_words, WORD=True)
                )
 
                buf = event.app.current_buffer
                old_completer = buf.completer
                buf.completer = palette_completer
                buf.start_completion(select_first=False)
                # Restore immediately — the completion popup captured the
                # palette completer at start_completion time, so it keeps
                # working until dismissed.  Normal Tab after that uses the
                # original OCChatCompleter again.
                buf.completer = old_completer
 
        if ctx is not None:
            kwargs: dict[str, Any] = dict(
                completer=OCChatCompleter(ctx),
                multiline=False,
                mouse_support=False,
                key_bindings=bindings,
                complete_while_typing=False,  # Tab to complete; less noisy.
            )
        else:
            kwargs = dict(
                multiline=False,
                mouse_support=False,
                key_bindings=bindings,
                complete_while_typing=False,
            )
        if history_path is not None:
            kwargs["history"] = FileHistory(str(history_path))
 
        # Continuation prefix for the wrapped lines of a multi-line chat
        # prompt (plain text — NO_COLOR-safe; unused by single-line menus).
        kwargs["prompt_continuation"] = "  … "
 
        # ── bottom_toolbar & rprompt (Phase 2.B) ─────────────────────────
        kwargs["bottom_toolbar"] = lambda: _build_bottom_toolbar(
            self._last_session, self._turn_stats
        ) if self._last_session is not None else ""
 
        if self._turn_stats is not None:
            kwargs["rprompt"] = lambda: self._turn_stats.rprompt_text()
 
        # Plan 147 D3-a: without a periodic refresh the toolbar lambda only
        # re-evaluates on a keypress, so the idle session timer would appear
        # frozen between turns. 1.0s matches the status bar's coarsest tier
        # (whole seconds once >= 60s).
        kwargs["refresh_interval"] = 1.0
 
        try:
            self._pt = PromptSession(**kwargs)
        except Exception as exc:
            print(f"[prompt_toolkit disabled: {exc}]", file=sys.stderr)
            self._pt = None
 
    # ── Public API ────────────────────────────────────────────────────────
 
    @property
    def using_prompt_toolkit(self) -> bool:
        """``True`` iff the modern prompt is active.  Read-only."""
        return self._pt is not None
 
    def read_user_input(self, session: ChatSession, prompt_text: str | None = None) -> str | None:
        """Read the next user input.
 
        *prompt_text* overrides the default ``[model | session] You: ``
        prefix — useful for startup menus that need a custom prompt.
 
        The main CHAT prompt (``prompt_text is None``) is MULTI-LINE:
        Enter inserts a newline, Ctrl+Enter submits (``c-j`` on Windows
        consoles; ``Ctrl+J`` / ``Alt+Enter`` are the portable
        equivalents), so a multi-line paste arrives as ONE prompt.
        Menu/startup prompts (``prompt_text`` given) stay single-line
        with Enter submitting, as before. The returned text may contain
        newlines; the caller strips as needed.
 
        Returns ``None`` when the user cancels with ``Ctrl+C``.
        Re-raises :class:`EOFError` so the main loop's existing handler
        keeps working unchanged.
        """
        self._cancelled = False
        self._last_session = session
 
        is_chat = prompt_text is None
        # Colour applies only to the main chat prompt, not startup-menu overrides
        # or NO_COLOR sessions.
        is_chat_prompt = is_chat and not os.getenv("NO_COLOR")
 
        if prompt_text is not None:
            prompt = prompt_text
        else:
            prompt = build_prompt_prefix(session)
 
        if self._pt is not None:
            if is_chat_prompt:
                from prompt_toolkit.formatted_text import FormattedText
                pt_prompt = FormattedText([(_PROMPT_COLOR_PT, prompt)])
            else:
                pt_prompt = prompt
            self._chat_multiline = is_chat
            try:
                return self._pt.prompt(pt_prompt, multiline=is_chat)
            except EOFError:
                raise
            except KeyboardInterrupt:
                # If our custom c-c binding fired, _cancelled is True.
                if self._cancelled:
                    return None
                raise
            finally:
                self._chat_multiline = False
 
        # Legacy input() path
        if is_chat_prompt:
            display_prompt = f"{_PROMPT_COLOR_ANSI}{prompt}{_ANSI_RESET}"
        else:
            display_prompt = prompt
        try:
            return input(display_prompt)
        except KeyboardInterrupt:
            return None