mirror of
https://github.com/NCBM/plyngent.git
synced 2026-11-07 07:17:54 +08:00
NCBM
76e5b8ac0f
/history counted raw message rows, so /history 1 printed whatever row happened to be last (usually a tool result in a tool-heavy session) and "one row means full body" was a proxy for "show me that one thing". A turn is what a human reviews: /history now prints the last turn's two ends (the user message and the model's answer, markdown for the latter), -v prints every row of the selected turn(s), -vv prints full bodies plus the local system row, and --preview collapses every row to one line. N addresses turn N, last [N] the last N turns, --message N a single row. Numbering is a display ordinal over conversation rows (local-only rows print as local) and deliberately never a database row number - seq is an implementation detail that shifts with /clear. Grouping and rendering live in the new cli/transcript.py.
304 lines
12 KiB
Python
304 lines
12 KiB
Python
"""``/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()
|