diff --git a/src/plyngent/cli/readline_setup.py b/src/plyngent/cli/readline_setup.py index 75a9f79..9179eac 100644 --- a/src/plyngent/cli/readline_setup.py +++ b/src/plyngent/cli/readline_setup.py @@ -15,9 +15,6 @@ if TYPE_CHECKING: HISTORY_FILE_NAME = "repl_history" DEFAULT_HISTORY_LENGTH = 1000 -_ON_OFF_ARGS: tuple[str, ...] = ("on", "off") -_EXPORT_ARGS: tuple[str, ...] = ("md", "json") - def slash_commands() -> list[str]: """Slash command tokens for Tab completion (with leading /).""" @@ -38,11 +35,18 @@ def filter_prefix(prefix: str, candidates: list[str]) -> list[str]: def build_completer(state: ReplState) -> Callable[[str, int], str | None]: - """Return a readline completer bound to the current REPL state.""" + """Return a readline completer bound to the current REPL state. + + Command names come from the Click slash registry; argument values come from + each parameter's :meth:`~click.ParamType.shell_complete` via + :func:`plyngent.cli.slash.complete_slash_args`. + """ def completer(text: str, state_index: int) -> str | None: import readline + from plyngent.cli.slash import complete_slash_args + buffer = readline.get_line_buffer() begidx = readline.get_begidx() # Completing the first token (command). @@ -51,7 +55,7 @@ def build_completer(state: ReplState) -> Callable[[str, int], str | None]: else: head = buffer[:begidx].strip() command = head.split()[0] if head else "" - options = _argument_options(state, command, text) + options = complete_slash_args(state, command, text) if state_index < len(options): return options[state_index] return None @@ -59,23 +63,6 @@ def build_completer(state: ReplState) -> Callable[[str, int], str | None]: return completer -def _argument_options(state: ReplState, command: str, text: str) -> list[str]: - candidates: list[str] = [] - if command == "/help": - # Second token: command name without leading slash (also accept /name). - candidates = [n.removeprefix("/") for n in slash_commands()] - text = text.lstrip("/") - elif command == "/provider": - candidates = sorted(state.config.providers.keys()) - elif command == "/model": - candidates = sorted(state.provider.models.keys()) - elif command in {"/tools", "/stream", "/verbose"}: - candidates = list(_ON_OFF_ARGS) - elif command == "/export": - candidates = list(_EXPORT_ARGS) - return filter_prefix(text, candidates) - - def bind_tab_complete(readline_mod: object) -> None: """Bind Tab to completion for GNU readline and libedit/editline.""" parse = getattr(readline_mod, "parse_and_bind", None) diff --git a/src/plyngent/cli/slash.py b/src/plyngent/cli/slash.py index f67bd05..4310908 100644 --- a/src/plyngent/cli/slash.py +++ b/src/plyngent/cli/slash.py @@ -1,10 +1,11 @@ from __future__ import annotations import shlex -from typing import TYPE_CHECKING, Any, override +from typing import TYPE_CHECKING, Any, cast, override import awaitlet # pyright: ignore[reportMissingTypeStubs] import click +from click.shell_completion import CompletionItem from msgspec import UNSET from plyngent.cli.retry import retry_pending_with_retries @@ -26,6 +27,8 @@ if TYPE_CHECKING: _DEFAULT_HISTORY_LINES = 20 _CONTENT_PREVIEW = 200 _COMPACT_PREVIEW = 400 +_ON_OFF_CHOICES = ("on", "off") +_EXPORT_FORMAT_CHOICES = ("md", "json") HELP_FOOTER = """\ User messages are saved immediately. On API errors or Ctrl+C, partial @@ -42,6 +45,16 @@ class ReplExitError(Exception): """Signal that the REPL should leave (not a process exit).""" +def _filter_choices(incomplete: str, choices: Sequence[str]) -> list[CompletionItem]: + return [CompletionItem(c) for c in choices if c.startswith(incomplete)] + + +def _repl_state(ctx: click.Context | None) -> ReplState | None: + if ctx is None or ctx.obj is None: + return None + return cast("ReplState", ctx.obj) + + class OnOffParam(click.ParamType[bool]): """Accept on/off (and common synonyms); convert to bool.""" @@ -59,10 +72,95 @@ class OnOffParam(click.ParamType[bool]): msg = "expected on or off" raise click.BadParameter(msg, ctx=ctx, param=param) + @override + def shell_complete(self, ctx: click.Context, param: click.Parameter, incomplete: str) -> list[CompletionItem]: + del ctx, param + return _filter_choices(incomplete, _ON_OFF_CHOICES) + ON_OFF = OnOffParam() +class ExportFormatParam(click.ParamType[str]): + """First token of /export: md|json (or a path if not a format).""" + + name: str = "export_format" + + @override + def convert(self, value: Any, param: click.Parameter | None, ctx: click.Context | None) -> str: + del param, ctx + return str(value) + + @override + def shell_complete(self, ctx: click.Context, param: click.Parameter, incomplete: str) -> list[CompletionItem]: + del ctx, param + return _filter_choices(incomplete, _EXPORT_FORMAT_CHOICES) + + +EXPORT_FORMAT = ExportFormatParam() + + +class ProviderNameParam(click.ParamType[str]): + name: str = "provider" + + @override + def convert(self, value: Any, param: click.Parameter | None, ctx: click.Context | None) -> str: + del param, ctx + return str(value) + + @override + def shell_complete(self, ctx: click.Context, param: click.Parameter, incomplete: str) -> list[CompletionItem]: + del param + state = _repl_state(ctx) + if state is None: + return [] + return _filter_choices(incomplete, sorted(state.config.providers.keys())) + + +PROVIDER_NAME = ProviderNameParam() + + +class ModelIdParam(click.ParamType[str]): + name: str = "model" + + @override + def convert(self, value: Any, param: click.Parameter | None, ctx: click.Context | None) -> str: + del param, ctx + return str(value) + + @override + def shell_complete(self, ctx: click.Context, param: click.Parameter, incomplete: str) -> list[CompletionItem]: + del param + state = _repl_state(ctx) + if state is None: + return [] + return _filter_choices(incomplete, sorted(state.provider.models.keys())) + + +MODEL_ID = ModelIdParam() + + +class SlashCommandNameParam(click.ParamType[str]): + """Command name for ``/help ``.""" + + name: str = "slash_command" + + @override + def convert(self, value: Any, param: click.Parameter | None, ctx: click.Context | None) -> str: + del param, ctx + return str(value) + + @override + def shell_complete(self, ctx: click.Context, param: click.Parameter, incomplete: str) -> list[CompletionItem]: + del param + names = sorted(slash.list_commands(ctx)) + token = incomplete.lstrip("/") + return _filter_choices(token, names) + + +SLASH_COMMAND_NAME = SlashCommandNameParam() + + class SlashGroup(click.Group): """Click group for REPL slash commands (no process-level ownership). @@ -104,6 +202,27 @@ def slash_command_names() -> list[str]: return sorted(f"/{name}" for name in slash.list_commands(ctx)) +def complete_slash_args(state: ReplState, command: str, incomplete: str) -> list[str]: + """Tab-complete arguments for ``command`` (e.g. ``/stream``) from ParamTypes. + + Uses the first :class:`click.Argument` on the registered command whose type + implements :meth:`~click.ParamType.shell_complete` with candidates. + """ + name = command.lstrip("/").lower() + ctx = click.Context(slash, obj=state) + cmd = slash.get_command(ctx, name) + if cmd is None: + return [] + with click.Context(cmd, info_name=name, parent=ctx, obj=state) as sub: + for param in cmd.params: + if not isinstance(param, click.Argument): + continue + items = param.type.shell_complete(sub, param, incomplete) + if items: + return [item.value for item in items] + return [] + + def _await[T](awaitable: Awaitable[T]) -> T: # Greenlet parks until the awaitable completes on the running loop. return awaitlet.awaitlet(awaitable) @@ -113,7 +232,7 @@ def _await[T](awaitable: Awaitable[T]) -> T: @slash.command("help") -@click.argument("command", required=False) +@click.argument("command", required=False, type=SLASH_COMMAND_NAME) @click.pass_context def help_cmd(ctx: click.Context, command: str | None) -> None: """Show this help, or help for one command.""" @@ -265,7 +384,7 @@ def delete_cmd(state: ReplState, session_id: int | None) -> None: @slash.command("export") -@click.argument("parts", nargs=-1) +@click.argument("parts", nargs=-1, type=EXPORT_FORMAT) @click.pass_obj def export_cmd(state: ReplState, parts: tuple[str, ...]) -> None: """Export session transcript from DB: /export [md|json] [path].""" @@ -371,7 +490,7 @@ def compact_cmd(state: ReplState, name: str | None) -> None: @slash.command("provider") -@click.argument("name", required=False) +@click.argument("name", required=False, type=PROVIDER_NAME) @click.pass_obj def provider_cmd(state: ReplState, name: str | None) -> None: """Show or switch provider.""" @@ -389,7 +508,7 @@ def provider_cmd(state: ReplState, name: str | None) -> None: @slash.command("model") -@click.argument("model_id", required=False) +@click.argument("model_id", required=False, type=MODEL_ID) @click.pass_obj def model_cmd(state: ReplState, model_id: str | None) -> None: """Show or switch model.""" diff --git a/tests/test_cli/test_readline_setup.py b/tests/test_cli/test_readline_setup.py index 9af4e2f..d0a43dc 100644 --- a/tests/test_cli/test_readline_setup.py +++ b/tests/test_cli/test_readline_setup.py @@ -142,3 +142,34 @@ def test_completer_help_commands(tmp_path: object, monkeypatch: object) -> None: index += 1 assert "compact" in found assert "clear" in found + + +def test_completer_stream_on_off(tmp_path: object, monkeypatch: object) -> None: + import readline + from pathlib import Path + + import pytest + + assert isinstance(tmp_path, Path) + assert isinstance(monkeypatch, pytest.MonkeyPatch) + + state = _minimal_state(tmp_path) + completer = build_completer(state) + monkeypatch.setattr(readline, "get_line_buffer", lambda: "/stream ") + monkeypatch.setattr(readline, "get_begidx", lambda: len("/stream ")) + + assert completer("o", 0) == "on" + assert completer("o", 1) == "off" + + +def test_complete_slash_args_from_registry(tmp_path: object) -> None: + from pathlib import Path + + from plyngent.cli.slash import complete_slash_args + + assert isinstance(tmp_path, Path) + state = _minimal_state(tmp_path) + assert complete_slash_args(state, "/provider", "r") == ["remote"] + assert complete_slash_args(state, "/model", "a") == ["alpha"] + assert complete_slash_args(state, "/export", "j") == ["json"] + assert complete_slash_args(state, "/help", "st") == ["status", "stream"]