core/cli: multiline \"\"\" input and /edit via EDITOR

Add triple-quote block messages in the REPL and /edit to compose a turn
in $EDITOR (temp file). Empty multiline or empty edit cancels.
This commit is contained in:
2026-07-15 13:29:00 +08:00
parent 3a0d9ac3cc
commit ec5e6ab0f4
9 changed files with 344 additions and 32 deletions
+2 -2
View File
@@ -114,15 +114,15 @@ Basedpyright `recommended`. Ruff includes `ANN` (private return types `ANN202` i
- G1: `ReasoningDeltaEvent`, `/stream`, `/verbose`
- G2: `/rename`, `/delete` (confirm), `/export md|json`
- G2.5: Click slash registry (`cli/slash.py`) + `awaitlet` for sync Click / async memory; auto `/help`; completer from `slash.list_commands`
- G3: multiline `"""``"""` input (`cli/input_text.py`); `/edit` via `$EDITOR` (`edit_text_in_editor`)
**Planned**
- **G3 — Input**: multiline (`"""``"""`) and/or `/edit` via `$EDITOR`
- **G4 — One-shot**: `plyngent chat -p/--prompt` (and stdin non-TTY); exit codes; non-interactive safety (`--yes` / deny destructive)
- **G5 — Hardening**: tool timeouts/defaults review, config error clarity, cancel edges, optional `--log-level`, no secrets in status/export
- **G6 — Docs**: real README, example TOML, CLAUDE/help accuracy
Milestone order: G3 → G4 → G5 → G6.
Milestone order: G4 → G5 → G6.
- **Phase H**: multi-tenant platform (`router/`, auth, sandboxed tools, web).
+48 -3
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
import contextlib
import os
import shlex
import subprocess
@@ -60,14 +61,24 @@ def ensure_config_file(path: Path) -> None:
_ = path.write_text(_MINIMAL_CONFIG, encoding="utf-8")
def open_in_editor(path: Path, *, editor: str | None = None) -> None:
"""Open ``path`` with ``EDITOR`` (supports values like ``codium --wait``)."""
def open_in_editor(
path: Path,
*,
editor: str | None = None,
ensure_exists: bool = True,
) -> None:
"""Open ``path`` with ``EDITOR`` (supports values like ``codium --wait``).
When ``ensure_exists`` is true (default), create a minimal config template
if the file is missing (used for ``plyngent config edit``).
"""
editor_cmd = editor if editor is not None else get_editor()
if editor_cmd is None:
msg = "EDITOR is not set"
raise click.ClickException(msg)
ensure_config_file(path)
if ensure_exists:
ensure_config_file(path)
try:
argv = [*shlex.split(editor_cmd, posix=os.name != "nt"), str(path)]
except ValueError as exc:
@@ -91,6 +102,40 @@ def open_in_editor(path: Path, *, editor: str | None = None) -> None:
raise click.ClickException(msg)
def edit_text_in_editor(initial: str = "", *, suffix: str = ".md") -> str | None:
"""Edit ``initial`` in ``$EDITOR``; return text or ``None`` if empty/cancelled.
Uses a temporary file. Does not create a config template.
"""
import tempfile
if get_editor() is None:
msg = "EDITOR is not set; cannot /edit"
raise click.ClickException(msg)
with tempfile.NamedTemporaryFile(
mode="w",
encoding="utf-8",
suffix=suffix,
prefix="plyngent-edit-",
delete=False,
) as handle:
path = Path(handle.name)
_ = handle.write(initial)
try:
open_in_editor(path, ensure_exists=False)
text = path.read_text(encoding="utf-8")
finally:
with contextlib.suppress(OSError):
path.unlink(missing_ok=True)
cleaned = text.rstrip("\n")
if not cleaned.strip():
return None
return cleaned
def prompt_edit_config(path: Path, *, reason: str | None = None) -> bool:
"""If ``EDITOR`` is set, ask whether to edit ``path``. Returns True if opened."""
if get_editor() is None:
+125
View File
@@ -0,0 +1,125 @@
from __future__ import annotations
from typing import TYPE_CHECKING
import click
if TYPE_CHECKING:
from collections.abc import Callable
_TRIPLE = '"""'
def parse_triple_quote_line(line: str) -> tuple[str, bool] | None:
"""If ``line`` starts a ``\"\"\"`` block, return ``(content_so_far, complete)``.
``complete`` is True when the closing marker is on the same line
(e.g. ``\"\"\"hello\"\"\"`` → ``(\"hello\", True)``).
"""
stripped = line.strip()
if not stripped.startswith(_TRIPLE):
return None
rest = stripped[len(_TRIPLE) :]
if rest.endswith(_TRIPLE) and len(rest) >= len(_TRIPLE):
inner = rest[: -len(_TRIPLE)]
return inner, True
return rest, False
def finish_triple_quote_line(line: str) -> tuple[str, bool]:
"""Parse a continuation line. Returns ``(content_piece, is_closer)``."""
stripped = line.strip()
if stripped == _TRIPLE:
return "", True
# Allow closing as trailing """ on a content line.
if line.rstrip().endswith(_TRIPLE):
without = line.rstrip()[: -len(_TRIPLE)]
# Don't treat a line that is only """ as content (handled above).
return without.rstrip("\n"), True
return line, False
def assemble_multiline(
opening_line: str,
*,
read_line: Callable[[], str],
echo: Callable[[str], None] | None = None,
) -> str | None:
"""Read a full triple-quoted message. Empty body → ``None`` (cancel)."""
parsed = parse_triple_quote_line(opening_line)
if parsed is None:
text = opening_line.strip()
return text or None
first, complete = parsed
if complete:
return first if first.strip() else None
parts: list[str] = []
if first:
parts.append(first)
if echo is not None:
echo(f"(multiline; end with {_TRIPLE})")
while True:
try:
line = read_line()
except EOFError, KeyboardInterrupt:
if echo is not None:
echo("")
echo("cancelled")
return None
piece, done = finish_triple_quote_line(line)
if done:
if piece:
parts.append(piece)
break
parts.append(piece)
text = "\n".join(parts)
return text if text.strip() else None
def _default_prompt() -> str:
return input("> ")
def _default_cont() -> str:
return input("... ")
def _default_echo(message: str) -> None:
click.echo(message)
def read_repl_entry(
*,
read_line: Callable[[], str] | None = None,
echo: Callable[[str], None] | None = None,
) -> str | None:
"""Read one REPL entry (slash line, single line, or multiline).
Returns:
``None`` — empty line or cancelled multiline (caller should re-prompt).
``str`` starting with ``/`` — slash command line (stripped).
other ``str`` — user message text (may contain newlines).
"""
_read = read_line if read_line is not None else _default_prompt
_echo = echo if echo is not None else _default_echo
try:
first = _read()
except EOFError:
raise
except KeyboardInterrupt:
_echo("")
return None
stripped = first.strip()
if not stripped:
return None
if stripped.startswith("/") and parse_triple_quote_line(stripped) is None:
return stripped
if parse_triple_quote_line(first) is not None:
cont_read = read_line if read_line is not None else _default_cont
return assemble_multiline(first, read_line=cont_read, echo=_echo)
return stripped
+21 -14
View File
@@ -4,6 +4,7 @@ from typing import TYPE_CHECKING
import click
from plyngent.cli.input_text import read_repl_entry
from plyngent.cli.readline_setup import setup_readline
from plyngent.cli.retry import run_user_text_with_retries
from plyngent.cli.slash import handle_slash
@@ -12,9 +13,13 @@ if TYPE_CHECKING:
from plyngent.cli.state import ReplState
def _read_line() -> str:
"""Blocking readline input (intentional for TTY REPL)."""
return input("> ").strip()
def _echo_user(text: str) -> None:
click.secho("user: ", fg="green", nl=False)
if "\n" in text:
click.echo()
click.echo(text)
else:
click.echo(text)
async def run_repl(state: ReplState) -> None:
@@ -27,26 +32,28 @@ async def run_repl(state: ReplState) -> None:
f"stream={'on' if state.agent.stream else 'off'} "
f"verbose={'on' if state.verbose else 'off'}"
)
click.echo("Type /help for commands. Empty line is ignored.")
click.echo('Type /help for commands. Multiline: """""". Empty line is ignored.')
while True:
try:
line = _read_line()
entry = read_repl_entry()
except EOFError:
click.echo()
break
except KeyboardInterrupt:
click.echo()
if entry is None:
continue
if not line:
continue
if line.startswith("/"):
cont = await handle_slash(state, line)
if entry.startswith("/"):
cont = await handle_slash(state, entry)
if not cont:
break
if state.pending_user_text is not None:
text = state.pending_user_text
state.pending_user_text = None
_echo_user(text)
_ = await run_user_text_with_retries(state.agent, text)
continue
click.secho("user: ", fg="green", nl=False)
click.echo(line)
_ = await run_user_text_with_retries(state.agent, line)
_echo_user(entry)
_ = await run_user_text_with_retries(state.agent, entry)
+34 -9
View File
@@ -30,15 +30,18 @@ _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
assistant/tool output is discarded but the user message stays (so /retry
works after resume, not only via readline history). Auto-retry: 10s/20s/30s.
Tab completes slash commands and some arguments (provider, model, tools,
stream, verbose, export). Use --session ID or /resume to continue a prior
chat after restart.
"""
HELP_FOOTER = (
"User messages are saved immediately. On API errors or Ctrl+C, partial\n"
"assistant/tool output is discarded but the user message stays (so /retry\n"
"works after resume, not only via readline history). Auto-retry: 10s/20s/30s.\n"
"\n"
"Tab completes slash commands and some arguments (provider, model, tools,\n"
"stream, verbose, export). Use --session ID or /resume to continue a prior\n"
"chat after restart.\n"
"\n"
'Multiline: start a message with """ then end a later line with """.\n'
"Long prompts: /edit opens $EDITOR.\n"
)
class ReplExitError(Exception):
@@ -268,6 +271,28 @@ def clear_cmd(state: ReplState) -> None:
click.echo("conversation cleared (in-memory only; DB history kept)")
@slash.command("edit")
@click.pass_obj
def edit_cmd(state: ReplState) -> None:
"""Compose a user message in ``$EDITOR``, then send it.
Opens a temporary buffer; save and quit the editor to submit.
Empty buffer cancels. Requires ``EDITOR`` (e.g. ``codium --wait``).
"""
from plyngent.cli.editor import edit_text_in_editor
try:
text = edit_text_in_editor("")
except click.ClickException as exc:
click.echo(f"error: {exc}")
return
if text is None:
click.echo("edit cancelled (empty)")
return
state.pending_user_text = text
click.echo(f"(edit) {len(text)} characters ready to send")
@slash.command("status")
@click.pass_obj
def status_cmd(state: ReplState) -> None:
+2
View File
@@ -31,6 +31,8 @@ class ReplState:
max_rounds: int = DEFAULT_MAX_ROUNDS
stream_enabled: bool = True
verbose: bool = False
# Set by /edit; REPL sends as the next user turn then clears.
pending_user_text: str | None = None
client: ChatClient = field(init=False)
agent: ChatAgent = field(init=False)
session_id: int | None = None
+37 -4
View File
@@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from pathlib import Path
import pytest
from click.testing import CliRunner
@@ -14,9 +14,6 @@ from plyngent.cli.editor import (
resolve_config_path,
)
if TYPE_CHECKING:
from pathlib import Path
def test_get_editor(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("EDITOR", raising=False)
@@ -66,6 +63,42 @@ def test_open_in_editor_missing_editor(tmp_path: Path, monkeypatch: pytest.Monke
open_in_editor(tmp_path / "x.toml", editor=None)
def test_edit_text_in_editor(monkeypatch: pytest.MonkeyPatch) -> None:
from plyngent.cli.editor import edit_text_in_editor
def fake_run(argv: list[str], check: bool = False) -> object:
del check
path = Path(argv[-1])
_ = path.write_text("hello from editor\n", encoding="utf-8")
class Result:
returncode: int = 0
return Result()
monkeypatch.setenv("EDITOR", "true")
monkeypatch.setattr("plyngent.cli.editor.subprocess.run", fake_run)
assert edit_text_in_editor("seed") == "hello from editor"
def test_edit_text_empty_cancels(monkeypatch: pytest.MonkeyPatch) -> None:
from plyngent.cli.editor import edit_text_in_editor
def fake_run(argv: list[str], check: bool = False) -> object:
del check
path = Path(argv[-1])
_ = path.write_text(" \n", encoding="utf-8")
class Result:
returncode: int = 0
return Result()
monkeypatch.setenv("EDITOR", "true")
monkeypatch.setattr("plyngent.cli.editor.subprocess.run", fake_run)
assert edit_text_in_editor("") is None
def test_prompt_edit_no_editor(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("EDITOR", raising=False)
assert prompt_edit_config(tmp_path / "p.toml") is False
+61
View File
@@ -0,0 +1,61 @@
from __future__ import annotations
from plyngent.cli.input_text import (
assemble_multiline,
finish_triple_quote_line,
parse_triple_quote_line,
read_repl_entry,
)
def test_parse_same_line_block() -> None:
assert parse_triple_quote_line('"""hello"""') == ("hello", True)
assert parse_triple_quote_line(' """hi""" ') == ("hi", True)
def test_parse_open_block() -> None:
assert parse_triple_quote_line('"""first') == ("first", False)
assert parse_triple_quote_line('"""') == ("", False)
def test_parse_not_block() -> None:
assert parse_triple_quote_line("hello") is None
assert parse_triple_quote_line("/help") is None
def test_finish_closer() -> None:
assert finish_triple_quote_line('"""') == ("", True)
assert finish_triple_quote_line('last line"""') == ("last line", True)
assert finish_triple_quote_line("middle") == ("middle", False)
def test_assemble_multiline() -> None:
lines = iter(["line two", '"""'])
text = assemble_multiline(
'"""line one',
read_line=lambda: next(lines),
)
assert text == "line one\nline two"
def test_assemble_empty_cancels() -> None:
lines = iter(['"""'])
assert assemble_multiline('"""', read_line=lambda: next(lines)) is None
def test_read_repl_entry_slash() -> None:
assert read_repl_entry(read_line=lambda: "/status") == "/status"
def test_read_repl_entry_simple() -> None:
assert read_repl_entry(read_line=lambda: " hi ") == "hi"
def test_read_repl_entry_empty() -> None:
assert read_repl_entry(read_line=lambda: " ") is None
def test_read_repl_entry_multiline() -> None:
seq = iter(['"""a', "b", '"""'])
text = read_repl_entry(read_line=lambda: next(seq), echo=lambda _s: None)
assert text == "a\nb"
+14
View File
@@ -134,6 +134,20 @@ async def test_quit(state: ReplState) -> None:
assert await handle_slash(state, "/quit") is False
async def test_edit_sets_pending(
state: ReplState,
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
monkeypatch.setattr(
"plyngent.cli.editor.edit_text_in_editor",
lambda *a, **k: "composed message",
)
assert await handle_slash(state, "/edit") is True
assert state.pending_user_text == "composed message"
assert "characters ready" in capsys.readouterr().out
async def test_new_and_sessions(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None:
first = state.session_id
assert await handle_slash(state, "/new other") is True