completers.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_ui/terminal/completers.py
 
"""
Context-aware tab-completion for the OC Chat REPL (Phase 2 PR 2.A).
 
Purpose
-------
The :class:`OCChatCompleter` understands the *shape* of partial input
and picks completions from the appropriate live registry — never
hardcoded. It is plugged into ``prompt_toolkit.PromptSession`` by
:class:`~oc_ui.terminal.input.PromptDriver`.
 
This module imports ``prompt_toolkit`` at top-level — only modules
that need completion functionality should import it, and they should
only do so after the optional dependency has been verified to be
available. The :class:`PromptDriver` handles that gating; tests use
``importorskip``.
 
Completion sources (in order of input shape):
 
================================== =====================================
Input shape                         Source
================================== =====================================
``/<partial>``                      ``oc_chat.core.commands.COMMANDS``
``/model <p>`` / ``/use <p>``       ``ctx.model_manager.known_models()``
``/session load <p>``               ``list_sessions()`` names
``/session switch <p>``             ``list_sessions()`` names
``/session save <p>``               ``list_sessions()`` names + free
``/session new <p>``                free (no completion)
``/context <p>`` / ``/read <p>``    paths via ``PathCompleter``
``/ui reasoning <p>``               ``(\"show\", \"dim\", \"hide\")``
``/effort <p>``                     ``REASONING_EFFORT_LEVELS``
``/help``, free text, etc.          no completions
================================== =====================================
 
Responsibilities
----------------
- :class:`OCChatCompleter` — top-level completer that routes the
  prompt-toolkit ``Document`` / ``CompleteEvent`` to one of the
  per-shape sub-completers.
- Per-verb completion logic for ``/model``, ``/session``, ``/read``,
  ``/context``, ``/ui``, ``/effort``.
 
Diagnostics
-----------
Domain: OC_UI.TERMINAL
Levels:
    L2 — lifecycle events
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- ``get_completions`` is a generator and must not block.
- All registry reads happen at call time so completions stay fresh
  as the user adds models / sessions during a REPL run.
- Unknown verbs yield no completions rather than raising.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- prompt_toolkit.completion (interactive REPL input with completion and history)
- prompt_toolkit.document (interactive REPL input with completion and history)
- oc_chat.core.commands (lazy; COMMANDS, COMMAND_HELP and UI_REASONING_VALUES for verb and choice completion)
- oc_chat.core.config (lazy; REASONING_EFFORT_LEVELS for /effort completion)
- oc_chat.core.persistence (lazy; list_sessions for session name completion)
"""
from __future__ import annotations
 
from typing import Iterable
 
from oc_diagnostics import _dbg
 
from prompt_toolkit.completion import (
    CompleteEvent, Completer, Completion, PathCompleter,
)
from prompt_toolkit.document import Document
 
 
class OCChatCompleter(Completer):
    """Tab completion driven by the live :class:`Context`."""
 
    def __init__(self, ctx) -> None:
        self._ctx = ctx
        self._path_completer = PathCompleter(expanduser=True)
        # Cache list_sessions per call to ``get_completions`` so that
        # multiple keystrokes in the same prompt invocation don't repeat
        # the disk walk.  Reset by ``_refresh_caches``.
        self._sessions_cache: list[str] | None = None
 
    # ── Public Completer API ──────────────────────────────────────────────
 
    def get_completions(self, document: Document,
                        complete_event: CompleteEvent) -> Iterable[Completion]:
        func = "get_completions"
        text = document.text_before_cursor
        _dbg("OC_UI.TERMINAL", func, f"text={text!r}", 4)
        # Refresh the per-prompt caches once at the start of each
        # completion request.
        self._refresh_caches()
 
        # Free-text chat (not starting with /) → no completions.
        if not text.startswith("/"):
            return
 
        # Tokenise on spaces to figure out which "slot" we're in.
        head, _, tail = text.partition(" ")
        if " " not in text:
            # Still in the verb itself.
            yield from self._complete_verb(text)
            return
 
        # We're past the first space.  Dispatch by verb.
        verb = head.lower()
 
        # ``get_word_before_cursor(WORD=False)`` splits on non-alphanumeric
        # chars, so ``deepseek-v4-pro`` would become just ``v4``.  For
        # model and session names (which may contain hyphens) we use
        # ``_last_token`` instead.  For subcommands and simple choices
        # (which are plain alphanumeric tokens) ``get_word_before_cursor``
        # is correct and handles trailing-space → empty-string properly.
        word_pt = document.get_word_before_cursor(WORD=False)
 
        if verb in ("/model", "/use"):
            word = self._last_token(text)
            yield from self._complete_model(word)
        elif verb == "/effort":
            yield from self._complete_choices(word_pt, self._effort_values())
        elif verb == "/context" or verb == "/read":
            yield from self._delegate_path_completion(document, complete_event)
        elif verb == "/session":
            yield from self._complete_session(text, word_pt, document, complete_event)
        elif verb == "/ui":
            yield from self._complete_ui(text, word_pt)
        # Other verbs: no further completions.
 
    # ── Helpers ───────────────────────────────────────────────────────────
 
    @staticmethod
    def _last_token(text: str) -> str:
        """Return the last whitespace-delimited token in *text*.
 
        Used instead of ``get_word_before_cursor(WORD=False)`` for values
        that may contain hyphens (model names, session names).
        """
        return text.split()[-1] if text.split() else ""
 
    def _refresh_caches(self) -> None:
        self._sessions_cache = None  # lazy — fetched on first use
 
    def _sessions(self) -> list[str]:
        func = "_sessions"
        if self._sessions_cache is None:
            try:
                from oc_chat.core.persistence import list_sessions
                self._sessions_cache = [s["name"] for s in list_sessions()]
            except Exception as e:
                # VALIDATION_ERROR — sessions dir absent / unreadable; degrade to empty list
                _dbg("OC_UI.TERMINAL", func, f"VALIDATION_ERROR: {type(e).__name__}: {e}", 3)
                self._sessions_cache = []
        return self._sessions_cache
 
    def _verbs(self) -> list[str]:
        from oc_chat.core.commands import COMMANDS
        return sorted(COMMANDS.keys())
 
    def _models(self) -> list[str]:
        func = "_models"
        try:
            return list(self._ctx.model_manager.known_models())
        except Exception as e:
            # VALIDATION_ERROR — model_manager not yet initialised; degrade to empty list
            _dbg("OC_UI.TERMINAL", func, f"VALIDATION_ERROR: {type(e).__name__}: {e}", 3)
            return []
 
    def _effort_values(self) -> tuple[str, ...]:
        func = "_effort_values"
        try:
            from oc_chat.core.config import REASONING_EFFORT_LEVELS
            return tuple(REASONING_EFFORT_LEVELS)
        except ImportError as e:
            # VALIDATION_ERROR — config not importable; fall back to canonical levels
            _dbg("OC_UI.TERMINAL", func, f"VALIDATION_ERROR: {type(e).__name__}: {e}", 3)
            return ("low", "medium", "high")
 
    def _reasoning_values(self) -> tuple[str, ...]:
        """Plan 147 D5: lazily import the canonical tuple (same pattern as
        :meth:`_effort_values` / :meth:`_verbs`) instead of hand-duplicating
        it — a deferred import avoids the module-load-time cycle a top-level
        ``from oc_chat.core.commands import ...`` would risk."""
        func = "_reasoning_values"
        try:
            from oc_chat.core.commands import UI_REASONING_VALUES
            return tuple(UI_REASONING_VALUES)
        except ImportError as e:
            # VALIDATION_ERROR — commands not importable; fall back to canonical values
            _dbg("OC_UI.TERMINAL", func, f"VALIDATION_ERROR: {type(e).__name__}: {e}", 3)
            return ("show", "dim", "hide")
 
    # ── Per-shape completers ──────────────────────────────────────────────
 
    def _complete_verb(self, partial: str) -> Iterable[Completion]:
        for verb in self._verbs():
            if verb.startswith(partial):
                yield Completion(verb, start_position=-len(partial),
                                 display_meta=self._verb_help(verb))
 
    def _verb_help(self, verb: str) -> str:
        func = "_verb_help"
        try:
            from oc_chat.core.commands import COMMAND_HELP
            line = COMMAND_HELP.get(verb, "")
            # COMMAND_HELP entries look like:  "/verb args  - description"
            if "  - " in line:
                return line.split("  - ", 1)[1].strip()
        except (ImportError, KeyError) as e:
            # VALIDATION_ERROR — commands module/key absent; degrade to no help text
            _dbg("OC_UI.TERMINAL", func, f"VALIDATION_ERROR: {type(e).__name__}: {e}", 3)
        return ""
 
    def _complete_model(self, word: str) -> Iterable[Completion]:
        for name in self._models():
            if name.startswith(word):
                yield Completion(name, start_position=-len(word))
 
    def _complete_choices(self, word: str,
                          choices: Iterable[str]) -> Iterable[Completion]:
        for choice in choices:
            if choice.startswith(word.lower()):
                yield Completion(choice, start_position=-len(word))
 
    def _complete_session(self, text: str, word: str,
                          document: Document, complete_event,
                          ) -> Iterable[Completion]:
        # Parse:  /session <sub> [name]
        tokens = text.split()
        if len(tokens) <= 1:
            # Only the verb so far.  Suggest subcommands.
            yield from self._complete_choices(word, _SESSION_SUBS)
            return
        sub = tokens[1].lower() if len(tokens) >= 2 else ""
        if sub not in _SESSION_SUBS_THAT_TAKE_NAMES:
            return
 
        # Determine the partial name to match.  If there are more than 2
        # tokens the user has typed (part of) a session name after the
        # subcommand — use the last token.  Otherwise they've just typed
        # the subcommand and a space (``/session load ``) so match with
        # an empty prefix.
        if len(tokens) > 2:
            name_word = self._last_token(text)
        else:
            name_word = ""
        for name in self._sessions():
            if name.startswith(name_word):
                yield Completion(name, start_position=-len(name_word))
 
    def _complete_ui(self, text: str, word: str) -> Iterable[Completion]:
        tokens = text.split()
        if len(tokens) <= 1:
            yield from self._complete_choices(word, _UI_SUBS)
            return
        sub = tokens[1].lower() if len(tokens) >= 2 else ""
        if sub == "reasoning":
            yield from self._complete_choices(word, self._reasoning_values())
 
    def _delegate_path_completion(self, document: Document, complete_event,
                                  ) -> Iterable[Completion]:
        # Build a sub-Document containing only the path-arg text so the
        # PathCompleter sees it as if it were the only input.  The
        # original cursor position becomes the cursor at end-of-arg.
        text = document.text_before_cursor
        # Position right after the verb + first space.
        first_space = text.find(" ")
        if first_space < 0:
            return
        arg = text[first_space + 1:]
        sub_doc = Document(text=arg, cursor_position=len(arg))
        for c in self._path_completer.get_completions(sub_doc, complete_event):
            yield c
 
 
# ── Constants used by the dispatcher ─────────────────────────────────────────
 
_SESSION_SUBS = (
    "new", "list", "current", "save", "load", "export", "switch", "clear",
)
_SESSION_SUBS_THAT_TAKE_NAMES = ("load", "switch", "save")
 
_UI_SUBS = ("reasoning",)