diff --git a/src/plyngent/cli/slash.py b/src/plyngent/cli/slash.py index 9240b73..51c1419 100644 --- a/src/plyngent/cli/slash.py +++ b/src/plyngent/cli/slash.py @@ -2,23 +2,15 @@ from __future__ import annotations import os import shlex -from typing import TYPE_CHECKING, Any, cast, override +from typing import TYPE_CHECKING, Any, Literal, cast, override import awaitlet # pyright: ignore[reportMissingTypeStubs] import click from click.shell_completion import CompletionItem -from msgspec import UNSET from plyngent.cli.models_source import DEFAULT_MODELS_CACHE_TTL, model_choices_for_provider from plyngent.cli.retry import retry_pending_with_retries from plyngent.cli.selection import select_model, select_provider -from plyngent.lmproto.openai_compatible.model import ( - AssistantChatMessage, - AssistantFunctionToolCall, - DeveloperChatMessage, - SystemChatMessage, - UserChatMessage, -) from plyngent.runtime import ProviderNotSupportedError if TYPE_CHECKING: @@ -26,10 +18,7 @@ if TYPE_CHECKING: from plyngent.agent import ToolRegistry from plyngent.cli.state import ReplState, YoloMode - from plyngent.lmproto.openai_compatible.model import AnyChatMessage -_DEFAULT_HISTORY_LINES = 20 -_CONTENT_PREVIEW = 200 _COMPACT_PREVIEW = 400 _ON_OFF_CHOICES = ("on", "off") _YOLO_MODE_CHOICES = ("on", "off", "once") @@ -49,33 +38,77 @@ _TODO_ACTION_CHOICES = ( _ROUNDS_CHOICES = ("8", "16", "32", "64", "128") -class HistoryLimitType(click.ParamType[int]): - """``N`` (int >= 1) or the shortcut ``last`` (equivalent to ``1``).""" +class HistoryTargetType(click.ParamType[int | Literal["last"]]): + """``N`` (turn number >= 1) or the keyword ``last`` (most recent turn).""" - name: str = "history_limit" + name: str = "history_target" @override - def convert(self, value: object, param: click.Parameter | None, ctx: click.Context | None) -> int: - if isinstance(value, int): - if value < 1: - self.fail("must be >= 1", param, ctx) - return value - text = str(value).strip().lower() + def convert( + self, + value: object, + param: click.Parameter | None, + ctx: click.Context | None, + ) -> int | Literal["last"]: + text = value.strip().lower() if isinstance(value, str) else str(value) if text == "last": - return 1 + return "last" try: n = int(text, 10) except ValueError: - self.fail("expected an integer N or 'last'", param, ctx) + self.fail("expected a turn number N or 'last'", param, ctx) if n < 1: self.fail("must be >= 1", param, ctx) return n @override def shell_complete(self, ctx: click.Context, param: click.Parameter, incomplete: str) -> list[CompletionItem]: - del ctx, param - tokens = ("last", "1", "5", "10", "20") - return [CompletionItem(token) for token in tokens if incomplete == "" or token.startswith(incomplete.lower())] + del param + tokens = ["last"] + state = _repl_state(ctx) + if state is not None: + from plyngent.cli.transcript import build_transcript + + turns = build_transcript(state.agent.messages).turns + # Recent turn numbers are the ones a human is likely to jump back to. + tokens += [str(turn.ordinal) for turn in turns[-5:]] + token = incomplete.strip().lower() + return [CompletionItem(item) for item in tokens if item.startswith(token)] + + +HISTORY_TARGET = HistoryTargetType() + + +class HistoryRowNumberType(click.ParamType[int]): + """One-based display number of a transcript row (``/history --message N``).""" + + name: str = "history_message" + + @override + def convert(self, value: object, param: click.Parameter | None, ctx: click.Context | None) -> int: + try: + n = int(str(value), 10) + except ValueError: + self.fail("expected a message number N", param, ctx) + if n < 1: + self.fail("must be >= 1", param, ctx) + return n + + @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 [] + from plyngent.cli.transcript import build_transcript + + transcript = build_transcript(state.agent.messages) + token = incomplete.strip() + recent = range(max(1, transcript.messages - 4), transcript.messages + 1) + return [CompletionItem(str(n)) for n in recent if str(n).startswith(token)] + + +HISTORY_ROW_NUMBER = HistoryRowNumberType() HELP_FOOTER = ( @@ -1367,65 +1400,61 @@ def rounds_cmd(state: ReplState, n: int | None) -> None: @slash.command("history") -@click.argument("n", required=False, type=HistoryLimitType()) +@click.argument("target", required=False, type=HISTORY_TARGET) +@click.argument("count", required=False, type=click.IntRange(min=1), metavar="[N]") @click.option( - "--full", - "mode_full", - is_flag=True, - default=False, - help="Print full message bodies (markdown for assistant when TTY).", + "--message", + "message_no", + type=HISTORY_ROW_NUMBER, + default=None, + metavar="N", + help="Print message N (display number) instead of a turn.", +) +@click.option( + "-v", + "--verbose", + "verbose", + count=True, + help="Print every row of the turn(s); repeat (-vv) for full bodies.", ) @click.option( "--preview", "mode_preview", is_flag=True, default=False, - help="Force short previews even for a single message.", + help="One-line preview for every printed row.", ) @click.pass_obj def history_cmd( state: ReplState, - n: int | None, + target: int | Literal["last"] | None, + count: int | None, *, - mode_full: bool, + message_no: int | None, + verbose: int, mode_preview: bool, ) -> None: - """Show recent messages (default: last 20 as short previews). + """Show conversation history, grouped by turn. - ``/history last`` or ``/history 1`` shows the last message in full with Rich - markdown when available. Use ``--full`` for multiple messages in full; - ``--preview`` forces the short form even for a single message. + ``/history`` — the last turn: your message and the model's final answer + ``/history N`` — turn N (numbered from the first turn of this session) + ``/history last [N]`` — the last N turns (default 1) + ``/history --message N`` — one message (display number, not a DB row) + ``-v`` adds every intermediate row (tool calls/results, notices); ``-vv`` + prints full bodies for all of them; ``--preview`` collapses to one line each. """ - if mode_full and mode_preview: - msg = "use only one of --full / --preview" - raise click.UsageError(msg) - limit = _DEFAULT_HISTORY_LINES if n is None else n - messages = state.agent.messages - if not messages: - click.echo("(no messages in this session)") - return - start = max(0, len(messages) - limit) - slice_msgs = messages[start:] - # Full: --full, or single-message window (last / 1) unless --preview. - full = mode_full or (len(slice_msgs) == 1 and not mode_preview) - mode_tag = "full" if full else "preview" - click.echo(f"session={state.session_id} messages={len(messages)} showing={len(slice_msgs)} mode={mode_tag}") - for offset, message in enumerate(slice_msgs): - idx = start + offset - if full: - _print_history_message_full(idx, message) - else: - click.echo(_format_history_message(idx, message)) - if state.agent.pending_retry_text is not None: - pending = state.agent.pending_retry_text - if full: - click.secho("(pending retry) user:", fg="yellow") - click.echo(pending) - else: - click.secho( - f"(pending retry) user: {_preview_content(pending)}", - fg="yellow", - ) + from plyngent.cli import transcript as transcript_mod + + transcript_mod.echo_history( + state.agent.messages, + session_id=state.session_id, + pending_retry_text=state.agent.pending_retry_text, + target=target, + count=count, + message_no=message_no, + verbose=verbose, + preview=mode_preview, + ) @slash.command("todos") @@ -1519,98 +1548,6 @@ def retry_cmd(state: ReplState) -> None: _ = _await(retry_pending_with_retries(state.agent)) -def _preview_content(text: str | None) -> str: - if not text: - return "" - if len(text) <= _CONTENT_PREVIEW: - return text - return text[:_CONTENT_PREVIEW] + "…" - - -def _print_history_assistant_full(index: int, message: AssistantChatMessage) -> None: - from plyngent.cli.display import markdown_render_available, print_markdown - - click.secho(f"{index}. assistant:", fg="cyan") - content = message.content - has_text = isinstance(content, str) and content.strip() - if has_text and isinstance(content, str): - if markdown_render_available(): - print_markdown(content, label="") - else: - click.echo(content) - reasoning = message.reasoning_content - has_reason = isinstance(reasoning, str) and reasoning.strip() - if has_reason and isinstance(reasoning, str): - click.secho("reasoning:", fg="bright_black") - click.echo(reasoning) - tool_calls = message.tool_calls - has_tools = tool_calls is not UNSET and bool(tool_calls) - if has_tools and tool_calls is not UNSET: - for call in tool_calls: - if isinstance(call, AssistantFunctionToolCall): - click.secho( - f" tool_call {call.id}: {call.function.name}({call.function.arguments})", - fg="yellow", - ) - else: - click.secho(f" tool_call custom id={call.id}", fg="yellow") - if not (has_text or has_reason or has_tools): - click.echo("(empty)") - click.echo() - - -def _print_history_message_full(index: int, message: AnyChatMessage) -> None: - """Print one history message with full body; Rich markdown for assistant text.""" - if isinstance(message, UserChatMessage): - click.secho(f"{index}. user:", fg="green") - click.echo(message.content or "") - click.echo() - return - if isinstance(message, DeveloperChatMessage): - click.secho(f"{index}. developer:", fg="blue") - click.echo(message.content or "") - click.echo() - return - if isinstance(message, SystemChatMessage): - click.secho(f"{index}. system:", fg="bright_black") - click.echo(message.content or "") - click.echo() - return - if isinstance(message, AssistantChatMessage): - _print_history_assistant_full(index, message) - return - # ToolChatMessage (remaining AnyChatMessage arm) - click.secho(f"{index}. tool({message.tool_call_id}):", fg="magenta") - click.echo(message.content or "") - click.echo() - - -def _format_history_message(index: int, message: AnyChatMessage) -> str: - if isinstance(message, UserChatMessage): - return f"{index}. user: {_preview_content(message.content)}" - if isinstance(message, DeveloperChatMessage): - return f"{index}. developer: {_preview_content(message.content)}" - if isinstance(message, SystemChatMessage): - return f"{index}. system: {_preview_content(message.content)}" - if isinstance(message, AssistantChatMessage): - parts: list[str] = [] - if isinstance(message.content, str) and message.content: - parts.append(_preview_content(message.content)) - tool_calls = message.tool_calls - if tool_calls is not UNSET and tool_calls: - names: list[str] = [] - for call in tool_calls: - if isinstance(call, AssistantFunctionToolCall): - names.append(call.function.name) - else: - names.append("custom") - parts.append(f"tool_calls=[{', '.join(names)}]") - body = " ".join(parts) if parts else "(empty)" - return f"{index}. assistant: {body}" - # ToolChatMessage - return f"{index}. tool({message.tool_call_id}): {_preview_content(message.content)}" - - def _run_slash_argv(args: Sequence[str], state: ReplState) -> None: """Sync Click entrypoint; may call awaitlet() for async work.""" # standalone_mode=False → UsageError/ClickException instead of SystemExit. diff --git a/src/plyngent/cli/transcript.py b/src/plyngent/cli/transcript.py new file mode 100644 index 0000000..78cfc94 --- /dev/null +++ b/src/plyngent/cli/transcript.py @@ -0,0 +1,393 @@ +"""Turn-oriented transcript grouping/rendering for ``/history``. + +The REPL transcript is a flat list of chat messages, but what a human reviews is +a *turn*: their message, the tool-loop rounds it triggered, and the model's final +answer. This module groups the list into numbered rows and turns, and renders +either the two ends of a turn (default), every row of it (``-v``), or full bodies +for every printed row (``-vv``). + +Numbering is a **display ordinal** over the conversation rows (1-based, in the +order the session holds them): it is what ``/history`` prints and what +``/history --message N`` addresses. Database row numbers are deliberately never +shown or used — they are an implementation detail that shifts with ``/clear``. +Local-only rows (the injected system prompt) have no number. +""" + +from __future__ import annotations + +from dataclasses import dataclass, replace +from typing import TYPE_CHECKING, Literal + +import click +from msgspec import UNSET + +from plyngent.lmproto.openai_compatible.model import ( + AssistantChatMessage, + AssistantFunctionToolCall, + DeveloperChatMessage, + SystemChatMessage, + ToolChatMessage, + UserChatMessage, +) + +if TYPE_CHECKING: + from collections.abc import Sequence + + from plyngent.lmproto.openai_compatible.model import AnyChatMessage + +PREVIEW_CHARS = 200 +# ``-v`` expands a turn's rows; ``-vv`` also prints their full bodies. +MAX_VERBOSE = 2 + +# Detail policy: ``mixed`` = a turn's final assistant answer in full, everything +# else on one line; ``full``/``preview`` force the extremes. +type Detail = Literal["mixed", "full", "preview"] +# ``/history N`` addresses a turn by number; ``last`` counts from the end. +type TurnTarget = int | Literal["last"] + + +def _content(message: AnyChatMessage) -> str: + """Message body as text (``content`` may be UNSET/None for assistants).""" + return message.content if isinstance(message.content, str) else "" + + +def one_line(text: str | None, *, limit: int = PREVIEW_CHARS) -> str: + """Collapse *text* to one line: first line, truncated, may add ``(+N lines)``.""" + body = (text or "").strip() + if not body: + return "" + lines = body.splitlines() + head = lines[0].strip() + if len(head) > limit: + head = head[:limit] + "…" + if len(lines) > 1: + head += f" (+{len(lines) - 1} lines)" + return head + + +def _role(message: AnyChatMessage) -> str: + if isinstance(message, UserChatMessage): + return "user" + if isinstance(message, AssistantChatMessage): + return "assistant" + if isinstance(message, DeveloperChatMessage): + return "developer" + if isinstance(message, SystemChatMessage): + return "system" + assert isinstance(message, ToolChatMessage) + return f"tool({message.tool_call_id})" + + +_ROLE_COLOR: dict[str, str] = { + "user": "green", + "assistant": "cyan", + "developer": "blue", + "system": "bright_black", +} + + +def _role_color(message: AnyChatMessage) -> str: + return "magenta" if isinstance(message, ToolChatMessage) else _ROLE_COLOR[_role(message)] + + +def _tool_call_names(message: AssistantChatMessage) -> list[str]: + tool_calls = message.tool_calls + if tool_calls is UNSET or not tool_calls: + return [] + return [call.function.name if isinstance(call, AssistantFunctionToolCall) else "custom" for call in tool_calls] + + +@dataclass(frozen=True, slots=True) +class Row: + """One transcript row; ``number`` is the display ordinal (None = local-only).""" + + message: AnyChatMessage + number: int | None + + def prefix(self) -> str: + """``12. user:`` — always styled (local rows keep the role colour).""" + label = "local" if self.number is None else str(self.number) + return str(click.style(f"{label}. {_role(self.message)}:", fg=_role_color(self.message))) + + def preview(self) -> str: + message = self.message + names = _tool_call_names(message) if isinstance(message, AssistantChatMessage) else [] + suffix = f" tool_calls=[{', '.join(names)}]" if names else "" + text = one_line(_content(message)) + if not text and isinstance(message, AssistantChatMessage) and not names: + text = "(empty)" + return f"{text}{suffix}".strip() + + +@dataclass(frozen=True, slots=True) +class Turn: + """A user turn: the user row plus the rounds it triggered.""" + + ordinal: int + rows: tuple[Row, ...] + incomplete: bool = False + + @property + def user(self) -> Row: + return self.rows[0] + + @property + def last(self) -> Row: + return self.rows[-1] + + @property + def rounds(self) -> int: + """Model rounds that asked for tools (one tool batch each).""" + return sum( + 1 + for row in self.rows + if isinstance(row.message, AssistantChatMessage) and bool(_tool_call_names(row.message)) + ) + + def label(self) -> str: + numbers = [row.number for row in self.rows if row.number is not None] + parts = [f"msgs {numbers[0]}-{numbers[-1]}" if numbers else "local rows"] + if self.rounds: + parts.append(f"rounds={self.rounds}") + text = f"turn {self.ordinal} ({', '.join(parts)})" + if self.incomplete: + text += " [incomplete — /retry continues]" + return text + + +@dataclass(frozen=True, slots=True) +class Transcript: + """Numbered rows grouped by turn (plus rows that precede the first turn).""" + + prelude: tuple[Row, ...] + turns: tuple[Turn, ...] + + @property + def messages(self) -> int: + """Numbered rows (what ``/history --message N`` can address).""" + return sum(1 for turn in self.turns for row in turn.rows if row.number is not None) + sum( + 1 for row in self.prelude if row.number is not None + ) + + def turn(self, ordinal: int) -> Turn | None: + return next((turn for turn in self.turns if turn.ordinal == ordinal), None) + + def row(self, number: int) -> Row | None: + for row in self.prelude: + if row.number == number: + return row + for turn in self.turns: + for row in turn.rows: + if row.number == number: + return row + return None + + def turn_of(self, number: int) -> Turn | None: + return next((turn for turn in self.turns if any(row.number == number for row in turn.rows)), None) + + +def build_transcript(messages: Sequence[AnyChatMessage], *, pending_retry_text: str | None = None) -> Transcript: + """Group *messages* into numbered rows and turns. + + A turn starts at a user message and runs until the next one, so tool results, + developer notices/checkpoints and synthetic todo nags stay inside the turn + they belong to. Rows before the first user message (the injected system + prompt, a compact seed summary, a resume notice) form the prelude. + + ``pending_retry_text`` marks the last turn as incomplete when it matches that + turn's user message (``ChatAgent.pending_retry_text``). + """ + rows: list[Row] = [] + number = 0 + for message in messages: + if isinstance(message, SystemChatMessage): + # The system prompt is injected locally and never persisted. + rows.append(Row(message=message, number=None)) + continue + number += 1 + rows.append(Row(message=message, number=number)) + + prelude: list[Row] = [] + turns: list[Turn] = [] + current: list[Row] = [] + for row in rows: + if isinstance(row.message, UserChatMessage): + if current: + turns.append(Turn(ordinal=len(turns) + 1, rows=tuple(current))) + current = [row] + elif current: + current.append(row) + else: + prelude.append(row) + if current: + turns.append(Turn(ordinal=len(turns) + 1, rows=tuple(current))) + + if turns and pending_retry_text is not None: + user = turns[-1].user.message + if isinstance(user, UserChatMessage) and user.content == pending_retry_text: + turns[-1] = replace(turns[-1], incomplete=True) + return Transcript(prelude=tuple(prelude), turns=tuple(turns)) + + +def echo_row(row: Row, *, detail: Detail) -> None: + """Print one row: a one-line preview, or the full body for ``full``/``mixed``.""" + if detail == "preview": + click.echo(f"{row.prefix()} {row.preview()}") + return + if isinstance(row.message, AssistantChatMessage): + _echo_assistant(row, reasoning=detail == "full") + return + click.echo(row.prefix()) + click.echo(row.message.content or "") + click.echo() + + +def _echo_assistant(row: Row, *, reasoning: bool) -> None: + from plyngent.cli.display import markdown_render_available, print_markdown + + message = row.message + assert isinstance(message, AssistantChatMessage) + click.echo(row.prefix()) + content = message.content + if isinstance(content, str) and content.strip(): + if markdown_render_available(): + print_markdown(content, label="") + else: + click.echo(content) + if reasoning: + shown = message.reasoning_content + if isinstance(shown, str) and shown.strip(): + click.secho("reasoning:", fg="bright_black") + click.echo(shown) + names = _tool_call_names(message) + if names: + click.secho(f" tool_calls=[{', '.join(names)}]", fg="yellow") + if not names and not _content(message).strip(): + click.echo("(empty)") + click.echo() + + +def echo_turn(turn: Turn, *, verbose: bool, detail: Detail) -> None: + """Print one turn: its header, then every row (verbose) or just its ends.""" + click.secho(turn.label(), fg="yellow" if turn.incomplete else "bright_black") + if verbose: + rows = turn.rows + elif turn.last is turn.user: + rows = (turn.user,) + else: + rows = (turn.user, turn.last) + for row in rows: + echo_row(row, detail=_row_detail(row, turn=turn, detail=detail)) + + +def _row_detail(row: Row, *, turn: Turn, detail: Detail) -> Detail: + """Effective detail for *row*: mixed prints the turn's answer in full.""" + if detail != "mixed": + return detail + return "full" if row is turn.last and isinstance(row.message, AssistantChatMessage) else "preview" + + +def echo_prelude_rows(rows: Sequence[Row], *, detail: Detail) -> None: + """Print rows that precede the first turn (memory seed, restart notice, …). + + Local-only rows (the injected system prompt) are control-plane noise, so they + only appear in a full dump (``-vv``) — numbered head rows always do. + """ + for row in rows: + if row.number is None and detail != "full": + continue + echo_row(row, detail="preview" if detail == "mixed" else detail) + + +def _resolve_detail(*, verbose: int, preview: bool) -> Detail: + """Map ``-v`` / ``-vv`` / ``--preview`` onto the detail policy.""" + if verbose > MAX_VERBOSE: + msg = "use at most -vv" + raise click.UsageError(msg) + if verbose == MAX_VERBOSE and preview: + msg = "use only one of -vv / --preview" + raise click.UsageError(msg) + if verbose == MAX_VERBOSE: + return "full" + return "preview" if preview else "mixed" + + +def _echo_message(view: Transcript, head: str, number: int, *, detail: Detail) -> None: + """Print one addressed message in full (``--message``).""" + row = view.row(number) + if row is None: + click.echo(f"error: no message {number} (messages 1-{view.messages})") + return + body: Detail = "preview" if detail == "preview" else "full" + click.echo(f"{head} showing=message {number} mode={body}") + echo_row(row, detail=body) + owner = view.turn_of(number) + if owner is not None: + click.secho(f"(turn {owner.ordinal}; /history {owner.ordinal} -v to see it whole)", fg="bright_black") + + +def _select_turns( + view: Transcript, + *, + target: TurnTarget | None, + count: int | None, +) -> tuple[tuple[Turn, ...], str] | None: + """Resolve ``/history [N]`` / ``/history last [N]`` into turns to print.""" + if target is None or target == "last": + want = count if count is not None else 1 + selected = view.turns[-want:] + return selected, f"last {len(selected)} turn(s)" + turn = view.turn(target) + if turn is None: + click.echo(f"error: no turn {target} (turns 1-{len(view.turns)})") + return None + return (turn,), f"turn {target}" + + +def echo_history( + messages: Sequence[AnyChatMessage], + *, + session_id: int | None, + pending_retry_text: str | None = None, + target: TurnTarget | None = None, + count: int | None = None, + message_no: int | None = None, + verbose: int = 0, + preview: bool = False, +) -> None: + """Render ``/history``: a turn window, or one addressed message. + + ``target`` is a turn number or ``"last"``; ``count`` (only with ``"last"``) + is how many turns; ``message_no`` prints a single row instead. Numbering is + the display ordinal of this transcript, never a database row number. + """ + if message_no is not None and (target is not None or count is not None): + msg = "use --message N alone" + raise click.UsageError(msg) + if message_no is not None and verbose: + msg = "-v/-vv apply to turns: /history N" + raise click.UsageError(msg) + if count is not None and target != "last": + msg = "a count applies to 'last': /history last [N]" + raise click.UsageError(msg) + detail = _resolve_detail(verbose=verbose, preview=preview) + if not messages: + click.echo("(no messages in this session)") + return + view = build_transcript(messages, pending_retry_text=pending_retry_text) + head = f"session={session_id} turns={len(view.turns)} messages={view.messages}" + + if message_no is not None: + _echo_message(view, head, message_no, detail=detail) + return + selection = _select_turns(view, target=target, count=count) + if selection is None: + return + selected, showing = selection + click.echo(f"{head} showing={showing} mode={detail}") + if verbose >= 1 and view.prelude: + echo_prelude_rows(view.prelude, detail=detail) + if not selected: + click.echo("(no turns yet in this session)") + for turn in selected: + echo_turn(turn, verbose=verbose >= 1, detail=detail) diff --git a/tests/test_cli/test_history.py b/tests/test_cli/test_history.py new file mode 100644 index 0000000..9563e17 --- /dev/null +++ b/tests/test_cli/test_history.py @@ -0,0 +1,303 @@ +"""``/history``: turn grouping, display numbering, verbose levels, addressing.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest +import tomlkit + +from plyngent.agent import ChatAgent +from plyngent.cli.slash import complete_slash_args, handle_slash +from plyngent.cli.state import ReplState +from plyngent.cli.transcript import build_transcript, one_line +from plyngent.config.models import DatabaseConfig, OpenAIProvider +from plyngent.config.store import ConfigStore +from plyngent.lmproto.openai_compatible.model import ( + AssistantChatMessage, + AssistantFunctionTool, + AssistantFunctionToolCall, + DeveloperChatMessage, + SystemChatMessage, + ToolChatMessage, + UserChatMessage, +) +from plyngent.memory import MemoryStore + +if TYPE_CHECKING: + from collections.abc import AsyncIterator + from pathlib import Path + + from plyngent.lmproto.openai_compatible.model import AnyChatMessage + + +class StubClient: + """No model calls happen while rendering history.""" + + async def chat_completions(self, param: object, *, stream: bool = False) -> object: + del param, stream + raise AssertionError("history must not call the model") + + +@pytest.fixture +async def state(tmp_path: Path) -> AsyncIterator[ReplState]: + memory = await MemoryStore.open(DatabaseConfig()) + provider = OpenAIProvider(access_key_or_token="sk-test") + config = ConfigStore(path=tmp_path / "plyngent.toml", document=tomlkit.document()) + config.providers = {"local": provider} + st = ReplState( + config=config, + memory=memory, + workspace=tmp_path, + provider_name="local", + provider=provider, + model="gpt-test", + tools_enabled=False, + ) + st.client = StubClient() + st.agent = ChatAgent(st.client, model=st.model, memory=st.memory, session_id=None) + await st.new_session("t") + yield st + await memory.close() + + +def _tool_call(message_id: str, name: str, args: str = "{}") -> AssistantChatMessage: + call = AssistantFunctionToolCall(id=message_id, function=AssistantFunctionTool(name=name, arguments=args)) + return AssistantChatMessage(content="", tool_calls=[call]) + + +def _two_turns(*, system: bool = False) -> list[AnyChatMessage]: + """Two settled turns; turn 1 has two tool rounds, turn 2 one.""" + messages: list[AnyChatMessage] = [] + if system: + messages.append(SystemChatMessage(content="You are plyngent. " + "rules. " * 30)) + messages += [ + UserChatMessage(content="read src/a.py and fix the typo"), + _tool_call("c1", "read_file", '{"path": "src/a.py"}'), + ToolChatMessage(tool_call_id="c1", content="L1-30\n" + "print('hi')\n" * 25 + "print('bye')\n"), + DeveloperChatMessage(content="[DIRECTIVE CHECKPOINT band=1 tokens≈100000 source=estimate]\nplaybook…"), + _tool_call("c2", "edit_replace", '{"path": "src/a.py"}'), + ToolChatMessage(tool_call_id="c2", content="replaced 1 occurrence"), + AssistantChatMessage(content="Fixed the typo in `src/a.py`.\n\n- line 12 changed"), + UserChatMessage(content="now run the tests"), + _tool_call("c3", "run_argv", '{"argv": ["pytest", "-q"]}'), + ToolChatMessage(tool_call_id="c3", content="12 passed in 3.2s"), + AssistantChatMessage(content="Tests pass: **12 passed** in 3.2s."), + ] + return messages + + +async def test_default_shows_the_last_turns_two_ends( + state: ReplState, + capsys: pytest.CaptureFixture[str], +) -> None: + state.agent.messages = _two_turns() + assert await handle_slash(state, "/history") is True + out = capsys.readouterr().out + assert "turns=2 messages=11 showing=last 1 turn(s) mode=mixed" in out + assert "turn 2 (msgs 8-11, rounds=1)" in out + assert "8. user: now run the tests" in out + # The answer is printed in full; the round's tool rows are not. + assert "11. assistant:" in out + assert "Tests pass: **12 passed** in 3.2s." in out + assert "tool(c3)" not in out + assert "turn 1" not in out + + +async def test_verbose_expands_every_row_of_the_turn( + state: ReplState, + capsys: pytest.CaptureFixture[str], +) -> None: + state.agent.messages = _two_turns() + assert await handle_slash(state, "/history -v 1") is True + out = capsys.readouterr().out + assert "turn 1 (msgs 1-7, rounds=2)" in out + assert "2. assistant: tool_calls=[read_file]" in out + assert "3. tool(c1): L1-30 (+26 lines)" in out + assert "4. developer: [DIRECTIVE CHECKPOINT band=1" in out + assert "6. tool(c2): replaced 1 occurrence" in out + # The turn's final answer stays full even though the row detail is mixed. + assert "7. assistant:" in out + assert "- line 12 changed" in out + + +async def test_double_verbose_prints_full_bodies( + state: ReplState, + capsys: pytest.CaptureFixture[str], +) -> None: + state.agent.messages = _two_turns(system=True) + assert await handle_slash(state, "/history -vv 1") is True + out = capsys.readouterr().out + assert "mode=full" in out + # A tool result past the 200-char preview window is printed in full. + assert "3. tool(c1):" in out + assert "print('bye')" in out + # -vv is also what surfaces the local-only system prompt. + assert "local. system:" in out + assert "You are plyngent." in out + + +async def test_verbose_keeps_local_rows_out_of_the_way( + state: ReplState, + capsys: pytest.CaptureFixture[str], +) -> None: + state.agent.messages = _two_turns(system=True) + assert await handle_slash(state, "/history -v 1") is True + out = capsys.readouterr().out + assert "local. system:" not in out + # Numbering is over conversation rows only, so the user row is #1. + assert "1. user: read src/a.py and fix the typo" in out + + +async def test_preview_collapses_every_row( + state: ReplState, + capsys: pytest.CaptureFixture[str], +) -> None: + state.agent.messages = _two_turns() + assert await handle_slash(state, "/history last 2 --preview") is True + out = capsys.readouterr().out + assert "showing=last 2 turn(s) mode=preview" in out + assert "7. assistant: Fixed the typo in `src/a.py`." in out + assert "11. assistant: Tests pass: **12 passed** in 3.2s." in out + + +async def test_last_n_turns(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None: + state.agent.messages = _two_turns() + assert await handle_slash(state, "/history last 2") is True + out = capsys.readouterr().out + assert "turn 1 (msgs 1-7, rounds=2)" in out + assert "turn 2 (msgs 8-11, rounds=1)" in out + + +async def test_turn_out_of_range(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None: + state.agent.messages = _two_turns() + assert await handle_slash(state, "/history 9") is True + assert "error: no turn 9 (turns 1-2)" in capsys.readouterr().out + + +async def test_message_number_prints_one_row_in_full( + state: ReplState, + capsys: pytest.CaptureFixture[str], +) -> None: + state.agent.messages = _two_turns() + assert await handle_slash(state, "/history --message 3") is True + out = capsys.readouterr().out + assert "showing=message 3 mode=full" in out + assert "3. tool(c1):" in out + assert "print('bye')" in out + assert "(turn 1; /history 1 -v to see it whole)" in out + + +async def test_message_number_preview(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None: + state.agent.messages = _two_turns() + assert await handle_slash(state, "/history --message 3 --preview") is True + out = capsys.readouterr().out + assert "showing=message 3 mode=preview" in out + assert "3. tool(c1): L1-30 (+26 lines)" in out + + +async def test_message_out_of_range(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None: + state.agent.messages = _two_turns() + assert await handle_slash(state, "/history --message 99") is True + assert "error: no message 99 (messages 1-11)" in capsys.readouterr().out + + +@pytest.mark.parametrize( + ("command", "message"), + [ + ("/history 1 2", "a count applies to 'last'"), + ("/history --message 3 -v", "-v/-vv apply to turns"), + ("/history -vv --preview", "use only one of -vv / --preview"), + ("/history -vvv", "use at most -vv"), + ], +) +async def test_usage_errors( + state: ReplState, + capsys: pytest.CaptureFixture[str], + command: str, + message: str, +) -> None: + state.agent.messages = _two_turns() + assert await handle_slash(state, command) is True # the REPL keeps running + assert message in capsys.readouterr().err + + +async def test_incomplete_turn_is_marked(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None: + """A retryable turn is flagged; no duplicate ``(pending retry)`` line.""" + state.agent.messages = [ + *_two_turns(), + UserChatMessage(content="also update the changelog"), + _tool_call("c4", "read_file", '{"path": "CHANGELOG.md"}'), + ToolChatMessage(tool_call_id="c4", content="L1-4\n# Changelog"), + ] + assert state.agent.pending_retry_text == "also update the changelog" + assert await handle_slash(state, "/history") is True + out = capsys.readouterr().out + assert "turn 3 (msgs 12-14, rounds=1) [incomplete — /retry continues]" in out + assert "pending retry" not in out + + +async def test_no_turns_yet_shows_the_prelude(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None: + """A compacted session starts with a seed assistant row, no user turn.""" + state.agent.messages = [ + SystemChatMessage(content="You are plyngent."), + AssistantChatMessage(content="Summary of session 4: …"), + ] + assert await handle_slash(state, "/history") is True + assert "(no turns yet in this session)" in capsys.readouterr().out + assert await handle_slash(state, "/history -v") is True + out = capsys.readouterr().out + assert "1. assistant: Summary of session 4: …" in out + assert "local system" not in out.replace(".", " ") + + +async def test_empty_history(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None: + state.agent.messages = [] + assert await handle_slash(state, "/history") is True + assert "(no messages in this session)" in capsys.readouterr().out + + +def test_turn_and_row_numbering() -> None: + messages = _two_turns() + view = build_transcript(messages) + assert len(view.turns) == 2 + assert view.turns[0].ordinal == 1 + assert view.messages == 11 + assert view.turn(2) is not None + assert view.turn(3) is None + row = view.row(11) + assert row is not None + assert view.turn_of(11) is not None + assert view.row(12) is None + + +def test_one_line_preview() -> None: + assert one_line(None) == "" + assert one_line("single") == "single" + assert one_line("first\nsecond\nthird") == "first (+2 lines)" + assert one_line("x" * 250).endswith("…") + assert len(one_line("x" * 250)) == 201 + + +def test_row_prefix_is_styled_once() -> None: + """Rows style their own prefix; renderers must not wrap it a second time.""" + view = build_transcript(_two_turns()) + row = view.row(1) + assert row is not None + prefix = row.prefix() + assert prefix.count("\x1b[32m") == 1 # green user, exactly one style open + assert prefix.startswith("\x1b[32m") + assert "1. user:" in prefix + local = build_transcript([SystemChatMessage(content="sys prompt")]).prelude[0] + assert "local. system:" in local.prefix() + + +async def test_slash_completion_offers_last_and_verbose( + state: ReplState, + capsys: pytest.CaptureFixture[str], +) -> None: + state.agent.messages = _two_turns() + assert complete_slash_args(state, "/history", "la") == ["last"] + assert "--verbose" in complete_slash_args(state, "/history", "--v") + assert "2" in complete_slash_args(state, "/history", "2") + capsys.readouterr() diff --git a/tests/test_cli/test_readline_setup.py b/tests/test_cli/test_readline_setup.py index f94bd5b..6a3ce31 100644 --- a/tests/test_cli/test_readline_setup.py +++ b/tests/test_cli/test_readline_setup.py @@ -186,7 +186,8 @@ def test_complete_slash_args_from_registry(tmp_path: object) -> None: assert "push" in complete_slash_args(state, "/todos", "p") assert "pop" in complete_slash_args(state, "/todos", "p") assert "--persist" in complete_slash_args(state, "/model", "--p") - assert "--full" in complete_slash_args(state, "/history", "--f") + assert "--verbose" in complete_slash_args(state, "/history", "--v") + assert complete_slash_args(state, "/history", "la") == ["last"] assert "32" in complete_slash_args(state, "/rounds", "3") from plyngent.agent.todo_stack import TodoStack diff --git a/tests/test_cli/test_repl_commands.py b/tests/test_cli/test_repl_commands.py index 37ce306..194f96c 100644 --- a/tests/test_cli/test_repl_commands.py +++ b/tests/test_cli/test_repl_commands.py @@ -134,8 +134,8 @@ async def test_help_history_no_fake_options(state: ReplState, capsys: pytest.Cap assert await handle_slash(state, "/help history") is True out = capsys.readouterr().out assert "Usage: /history" in out - # Real flags for full/preview are documented; Click's meta --help is not. - assert "--full" in out + # Real flags are documented; Click's meta --help is not. + assert "--verbose" in out assert "--preview" in out assert " --help" not in out @@ -498,70 +498,6 @@ async def test_resume(state: ReplState) -> None: assert await handle_slash(state, f"/resume {sid}") is True -async def test_history(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None: - from plyngent.lmproto.openai_compatible.model import AssistantChatMessage, UserChatMessage - - state.agent.messages = [ - UserChatMessage(content="hello"), - AssistantChatMessage(content="hi there"), - ] - assert await handle_slash(state, "/history") is True - out = capsys.readouterr().out - assert "mode=preview" in out - assert "user: hello" in out - assert "assistant: hi there" in out - - -async def test_history_last_full(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None: - from plyngent.lmproto.openai_compatible.model import AssistantChatMessage, UserChatMessage - - long_body = "# Title\n\n" + ("paragraph text. " * 20) - state.agent.messages = [ - UserChatMessage(content="hello"), - AssistantChatMessage(content=long_body), - ] - assert await handle_slash(state, "/history last") is True - out = capsys.readouterr().out - assert "mode=full" in out - assert "assistant:" in out - # Full mode should not use the 200-char preview ellipsis on the body alone - assert long_body[:50] in out or "Title" in out - - -async def test_history_one_full(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None: - from plyngent.lmproto.openai_compatible.model import UserChatMessage - - state.agent.messages = [UserChatMessage(content="only me")] - assert await handle_slash(state, "/history 1") is True - out = capsys.readouterr().out - assert "mode=full" in out - assert "only me" in out - - -async def test_history_one_preview_flag(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None: - from plyngent.lmproto.openai_compatible.model import UserChatMessage - - state.agent.messages = [UserChatMessage(content="x" * 250)] - assert await handle_slash(state, "/history 1 --preview") is True - out = capsys.readouterr().out - assert "mode=preview" in out - assert "…" in out - - -async def test_history_full_flag(state: ReplState, capsys: pytest.CaptureFixture[str]) -> None: - from plyngent.lmproto.openai_compatible.model import AssistantChatMessage, UserChatMessage - - body = "full " * 100 - state.agent.messages = [ - UserChatMessage(content="u"), - AssistantChatMessage(content=body), - ] - assert await handle_slash(state, "/history 2 --full") is True - out = capsys.readouterr().out - assert "mode=full" in out - assert body[:40] in out - - async def test_rounds(state: ReplState) -> None: assert await handle_slash(state, "/rounds 40") is True assert state.max_rounds == 40