mirror of
https://github.com/NCBM/plyngent.git
synced 2026-07-23 05:55:16 +08:00
core/cli: readline Tab completion and persistent history
Complete slash commands and provider/model/tools args; store history under platformdirs user data.
This commit is contained in:
@@ -73,6 +73,7 @@ Click app + readline REPL. Entry: `plyngent` / `python -m plyngent`.
|
|||||||
- **`plyngent chat`**: provider/model selection (flags or interactive), SQLite sessions via config `[database]`, `/help` slash commands.
|
- **`plyngent chat`**: provider/model selection (flags or interactive), SQLite sessions via config `[database]`, `/help` slash commands.
|
||||||
- **`plyngent providers`**: list config providers.
|
- **`plyngent providers`**: list config providers.
|
||||||
- Tools default on (`--tools` / `--no-tools`); workspace defaults to cwd.
|
- Tools default on (`--tools` / `--no-tools`); workspace defaults to cwd.
|
||||||
|
- Readline: Tab completion for slash commands/args; history file under platformdirs user data (`repl_history`).
|
||||||
|
|
||||||
### Composition utility: `Forward` descriptor
|
### Composition utility: `Forward` descriptor
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,102 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import atexit
|
||||||
|
import contextlib
|
||||||
|
from typing import TYPE_CHECKING
|
||||||
|
|
||||||
|
from platformdirs import user_data_path
|
||||||
|
|
||||||
|
if TYPE_CHECKING:
|
||||||
|
from collections.abc import Callable
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
from plyngent.cli.state import ReplState
|
||||||
|
|
||||||
|
HISTORY_FILE_NAME = "repl_history"
|
||||||
|
DEFAULT_HISTORY_LENGTH = 1000
|
||||||
|
|
||||||
|
SLASH_COMMANDS: tuple[str, ...] = (
|
||||||
|
"/help",
|
||||||
|
"/quit",
|
||||||
|
"/exit",
|
||||||
|
"/clear",
|
||||||
|
"/sessions",
|
||||||
|
"/new",
|
||||||
|
"/resume",
|
||||||
|
"/provider",
|
||||||
|
"/model",
|
||||||
|
"/tools",
|
||||||
|
)
|
||||||
|
|
||||||
|
_TOOLS_ARGS: tuple[str, ...] = ("on", "off")
|
||||||
|
|
||||||
|
|
||||||
|
def history_path() -> Path:
|
||||||
|
return user_data_path("plyngent", ensure_exists=True) / HISTORY_FILE_NAME
|
||||||
|
|
||||||
|
|
||||||
|
def filter_prefix(prefix: str, candidates: list[str]) -> list[str]:
|
||||||
|
"""Return candidates that start with ``prefix`` (or all if prefix empty)."""
|
||||||
|
if not prefix:
|
||||||
|
return list(candidates)
|
||||||
|
return [c for c in candidates if c.startswith(prefix)]
|
||||||
|
|
||||||
|
|
||||||
|
def build_completer(state: ReplState) -> Callable[[str, int], str | None]:
|
||||||
|
"""Return a readline completer bound to the current REPL state."""
|
||||||
|
|
||||||
|
def completer(text: str, state_index: int) -> str | None:
|
||||||
|
import readline
|
||||||
|
|
||||||
|
buffer = readline.get_line_buffer()
|
||||||
|
begidx = readline.get_begidx()
|
||||||
|
# Completing the first token (command).
|
||||||
|
if begidx == 0 or (begidx > 0 and buffer[:begidx].strip() == ""):
|
||||||
|
options = filter_prefix(text, list(SLASH_COMMANDS))
|
||||||
|
else:
|
||||||
|
head = buffer[:begidx].strip()
|
||||||
|
command = head.split()[0] if head else ""
|
||||||
|
options = _argument_options(state, command, text)
|
||||||
|
if state_index < len(options):
|
||||||
|
return options[state_index]
|
||||||
|
return None
|
||||||
|
|
||||||
|
return completer
|
||||||
|
|
||||||
|
|
||||||
|
def _argument_options(state: ReplState, command: str, text: str) -> list[str]:
|
||||||
|
if command == "/provider":
|
||||||
|
return filter_prefix(text, sorted(state.config.providers.keys()))
|
||||||
|
if command == "/model":
|
||||||
|
return filter_prefix(text, sorted(state.provider.models.keys()))
|
||||||
|
if command == "/tools":
|
||||||
|
return filter_prefix(text, list(_TOOLS_ARGS))
|
||||||
|
if command == "/resume":
|
||||||
|
# Session ids are numeric; no sync list without await — skip.
|
||||||
|
return []
|
||||||
|
return []
|
||||||
|
|
||||||
|
|
||||||
|
def setup_readline(state: ReplState) -> None:
|
||||||
|
"""Configure Tab completion and persistent history when readline is available."""
|
||||||
|
try:
|
||||||
|
import readline
|
||||||
|
except ImportError:
|
||||||
|
return
|
||||||
|
|
||||||
|
readline.parse_and_bind("tab: complete")
|
||||||
|
# Treat path-like chars as part of a token so /help completes as one word.
|
||||||
|
readline.set_completer_delims(" \t\n")
|
||||||
|
readline.set_completer(build_completer(state))
|
||||||
|
|
||||||
|
hist = history_path()
|
||||||
|
with contextlib.suppress(FileNotFoundError, OSError):
|
||||||
|
readline.read_history_file(str(hist))
|
||||||
|
readline.set_history_length(DEFAULT_HISTORY_LENGTH)
|
||||||
|
|
||||||
|
def _save_history() -> None:
|
||||||
|
with contextlib.suppress(OSError):
|
||||||
|
_ = hist.parent.mkdir(parents=True, exist_ok=True)
|
||||||
|
readline.write_history_file(str(hist))
|
||||||
|
|
||||||
|
_ = atexit.register(_save_history)
|
||||||
@@ -1,18 +1,13 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import contextlib
|
|
||||||
import inspect
|
import inspect
|
||||||
from collections.abc import Awaitable, Callable
|
from collections.abc import Awaitable, Callable
|
||||||
from typing import TYPE_CHECKING
|
from typing import TYPE_CHECKING
|
||||||
|
|
||||||
import click
|
import click
|
||||||
|
|
||||||
with contextlib.suppress(ImportError):
|
|
||||||
import readline as _readline
|
|
||||||
|
|
||||||
_ = _readline
|
|
||||||
|
|
||||||
from plyngent.cli.display import render_events
|
from plyngent.cli.display import render_events
|
||||||
|
from plyngent.cli.readline_setup import setup_readline
|
||||||
from plyngent.cli.selection import select_model, select_provider
|
from plyngent.cli.selection import select_model, select_provider
|
||||||
from plyngent.runtime import ProviderNotSupportedError
|
from plyngent.runtime import ProviderNotSupportedError
|
||||||
|
|
||||||
@@ -30,6 +25,8 @@ Commands:
|
|||||||
/provider [name] Show or switch provider
|
/provider [name] Show or switch provider
|
||||||
/model [id] Show or switch model
|
/model [id] Show or switch model
|
||||||
/tools [on|off] Show or toggle tools
|
/tools [on|off] Show or toggle tools
|
||||||
|
|
||||||
|
Tab completes slash commands and some arguments (provider, model, tools).
|
||||||
"""
|
"""
|
||||||
|
|
||||||
type SlashHandler = Callable[[], None | Awaitable[None]]
|
type SlashHandler = Callable[[], None | Awaitable[None]]
|
||||||
@@ -154,7 +151,8 @@ def _read_line() -> str:
|
|||||||
|
|
||||||
|
|
||||||
async def run_repl(state: ReplState) -> None:
|
async def run_repl(state: ReplState) -> None:
|
||||||
"""Interactive chat loop with readline editing."""
|
"""Interactive chat loop with readline editing, history, and Tab completion."""
|
||||||
|
setup_readline(state)
|
||||||
click.echo(
|
click.echo(
|
||||||
f"plyngent chat provider={state.provider_name} model={state.model} "
|
f"plyngent chat provider={state.provider_name} model={state.model} "
|
||||||
f"session={state.session_id} tools={'on' if state.tools_enabled else 'off'}"
|
f"session={state.session_id} tools={'on' if state.tools_enabled else 'off'}"
|
||||||
|
|||||||
@@ -0,0 +1,98 @@
|
|||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import tomlkit
|
||||||
|
|
||||||
|
from plyngent.cli.readline_setup import (
|
||||||
|
SLASH_COMMANDS,
|
||||||
|
build_completer,
|
||||||
|
filter_prefix,
|
||||||
|
history_path,
|
||||||
|
)
|
||||||
|
from plyngent.cli.state import ReplState
|
||||||
|
from plyngent.config.models import ModelConfig, OpenAICompatibleProvider, OpenAIProvider
|
||||||
|
from plyngent.config.store import ConfigStore
|
||||||
|
|
||||||
|
|
||||||
|
def _minimal_state(tmp_path: object) -> ReplState:
|
||||||
|
from pathlib import Path
|
||||||
|
from unittest.mock import MagicMock
|
||||||
|
|
||||||
|
assert isinstance(tmp_path, Path)
|
||||||
|
provider = OpenAICompatibleProvider(
|
||||||
|
access_key_or_token="sk",
|
||||||
|
url="https://example.com/v1",
|
||||||
|
models={"alpha": ModelConfig(), "beta": ModelConfig()},
|
||||||
|
)
|
||||||
|
config = ConfigStore(path=tmp_path / "plyngent.toml", document=tomlkit.document())
|
||||||
|
config.providers = {
|
||||||
|
"local": OpenAIProvider(access_key_or_token="sk"),
|
||||||
|
"remote": provider,
|
||||||
|
}
|
||||||
|
# Avoid real client/network: build ReplState pieces manually via object.__new__
|
||||||
|
state = object.__new__(ReplState)
|
||||||
|
state.config = config
|
||||||
|
state.provider = provider
|
||||||
|
state.provider_name = "remote"
|
||||||
|
state.model = "alpha"
|
||||||
|
state.tools_enabled = True
|
||||||
|
state.memory = MagicMock()
|
||||||
|
state.workspace = tmp_path
|
||||||
|
state.session_id = None
|
||||||
|
state.client = MagicMock()
|
||||||
|
state.agent = MagicMock()
|
||||||
|
return state
|
||||||
|
|
||||||
|
|
||||||
|
def test_filter_prefix() -> None:
|
||||||
|
assert filter_prefix("/he", ["/help", "/quit"]) == ["/help"]
|
||||||
|
assert filter_prefix("", ["a", "b"]) == ["a", "b"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_history_path_under_user_data() -> None:
|
||||||
|
path = history_path()
|
||||||
|
assert path.name == "repl_history"
|
||||||
|
assert "plyngent" in str(path)
|
||||||
|
|
||||||
|
|
||||||
|
def test_completer_commands(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: "/")
|
||||||
|
monkeypatch.setattr(readline, "get_begidx", lambda: 0)
|
||||||
|
|
||||||
|
found: list[str] = []
|
||||||
|
index = 0
|
||||||
|
while True:
|
||||||
|
item = completer("/", index)
|
||||||
|
if item is None:
|
||||||
|
break
|
||||||
|
found.append(item)
|
||||||
|
index += 1
|
||||||
|
assert "/help" in found
|
||||||
|
assert set(found) <= set(SLASH_COMMANDS)
|
||||||
|
|
||||||
|
|
||||||
|
def test_completer_provider_args(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: "/provider ")
|
||||||
|
monkeypatch.setattr(readline, "get_begidx", lambda: len("/provider "))
|
||||||
|
|
||||||
|
first = completer("r", 0)
|
||||||
|
assert first == "remote"
|
||||||
Reference in New Issue
Block a user