feat: display previous messages when resuming a session in CLI
When resuming a session via --continue or --resume, show a compact recap of the previous conversation inside a Rich panel before the input prompt. This gives users immediate visual context about what was discussed. Changes: - Add _preload_resumed_session() to load session history early (in run(), before banner) so _init_agent() doesn't need a separate DB round-trip - Add _display_resumed_history() that renders a formatted recap panel: * User messages shown with gold bullet (truncated at 300 chars) * Assistant responses shown with green diamond (truncated at 200 chars / 3 lines) * Tool calls collapsed to count + tool names * System messages and tool results hidden * <REASONING_SCRATCHPAD> blocks stripped from display * Pure-reasoning messages (no visible output) skipped entirely * Capped at last 10 exchanges with 'N earlier messages' indicator * Dim/muted styling distinguishes recap from active conversation - Add display.resume_display config option: 'full' (default) or 'minimal' - Store resume_display as instance variable (like compact) for testability - 27 new tests covering all display scenarios, config, and edge cases Closes #719
This commit is contained in:
214
cli.py
214
cli.py
@@ -193,6 +193,7 @@ def load_cli_config() -> Dict[str, Any]:
|
||||
"toolsets": ["all"],
|
||||
"display": {
|
||||
"compact": False,
|
||||
"resume_display": "full",
|
||||
},
|
||||
"clarify": {
|
||||
"timeout": 120, # Seconds to wait for a clarify answer before auto-proceeding
|
||||
@@ -1008,6 +1009,8 @@ class HermesCLI:
|
||||
self.compact = compact if compact is not None else CLI_CONFIG["display"].get("compact", False)
|
||||
# tool_progress: "off", "new", "all", "verbose" (from config.yaml display section)
|
||||
self.tool_progress_mode = CLI_CONFIG["display"].get("tool_progress", "all")
|
||||
# resume_display: "full" (show history) | "minimal" (one-liner only)
|
||||
self.resume_display = CLI_CONFIG["display"].get("resume_display", "full")
|
||||
self.verbose = verbose if verbose is not None else (self.tool_progress_mode == "verbose")
|
||||
|
||||
# Configuration - priority: CLI args > env vars > config file
|
||||
@@ -1266,8 +1269,11 @@ class HermesCLI:
|
||||
except Exception as e:
|
||||
logger.debug("SQLite session store not available: %s", e)
|
||||
|
||||
# If resuming, validate the session exists and load its history
|
||||
if self._resumed and self._session_db:
|
||||
# If resuming, validate the session exists and load its history.
|
||||
# _preload_resumed_session() may have already loaded it (called from
|
||||
# run() for immediate display). In that case, conversation_history
|
||||
# is non-empty and we skip the DB round-trip.
|
||||
if self._resumed and self._session_db and not self.conversation_history:
|
||||
session_meta = self._session_db.get_session(self.session_id)
|
||||
if not session_meta:
|
||||
_cprint(f"\033[1;31mSession not found: {self.session_id}{_RST}")
|
||||
@@ -1371,7 +1377,202 @@ class HermesCLI:
|
||||
self._show_tool_availability_warnings()
|
||||
|
||||
self.console.print()
|
||||
|
||||
|
||||
def _preload_resumed_session(self) -> bool:
|
||||
"""Load a resumed session's history from the DB early (before first chat).
|
||||
|
||||
Called from run() so the conversation history is available for display
|
||||
before the user sends their first message. Sets
|
||||
``self.conversation_history`` and prints the one-liner status. Returns
|
||||
True if history was loaded, False otherwise.
|
||||
|
||||
The corresponding block in ``_init_agent()`` checks whether history is
|
||||
already populated and skips the DB round-trip.
|
||||
"""
|
||||
if not self._resumed or not self._session_db:
|
||||
return False
|
||||
|
||||
session_meta = self._session_db.get_session(self.session_id)
|
||||
if not session_meta:
|
||||
self.console.print(
|
||||
f"[bold red]Session not found: {self.session_id}[/]"
|
||||
)
|
||||
self.console.print(
|
||||
"[dim]Use a session ID from a previous CLI run "
|
||||
"(hermes sessions list).[/]"
|
||||
)
|
||||
return False
|
||||
|
||||
restored = self._session_db.get_messages_as_conversation(self.session_id)
|
||||
if restored:
|
||||
self.conversation_history = restored
|
||||
msg_count = len([m for m in restored if m.get("role") == "user"])
|
||||
title_part = ""
|
||||
if session_meta.get("title"):
|
||||
title_part = f' "{session_meta["title"]}"'
|
||||
self.console.print(
|
||||
f"[#DAA520]↻ Resumed session [bold]{self.session_id}[/bold]"
|
||||
f"{title_part} "
|
||||
f"({msg_count} user message{'s' if msg_count != 1 else ''}, "
|
||||
f"{len(restored)} total messages)[/]"
|
||||
)
|
||||
else:
|
||||
self.console.print(
|
||||
f"[#DAA520]Session {self.session_id} found but has no "
|
||||
f"messages. Starting fresh.[/]"
|
||||
)
|
||||
return False
|
||||
|
||||
# Re-open the session (clear ended_at so it's active again)
|
||||
try:
|
||||
self._session_db._conn.execute(
|
||||
"UPDATE sessions SET ended_at = NULL, end_reason = NULL "
|
||||
"WHERE id = ?",
|
||||
(self.session_id,),
|
||||
)
|
||||
self._session_db._conn.commit()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return True
|
||||
|
||||
def _display_resumed_history(self):
|
||||
"""Render a compact recap of previous conversation messages.
|
||||
|
||||
Uses Rich markup with dim/muted styling so the recap is visually
|
||||
distinct from the active conversation. Caps the display at the
|
||||
last ``MAX_DISPLAY_EXCHANGES`` user/assistant exchanges and shows
|
||||
an indicator for earlier hidden messages.
|
||||
"""
|
||||
if not self.conversation_history:
|
||||
return
|
||||
|
||||
# Check config: resume_display setting
|
||||
if self.resume_display == "minimal":
|
||||
return
|
||||
|
||||
MAX_DISPLAY_EXCHANGES = 10 # max user+assistant pairs to show
|
||||
MAX_USER_LEN = 300 # truncate user messages
|
||||
MAX_ASST_LEN = 200 # truncate assistant text
|
||||
MAX_ASST_LINES = 3 # max lines of assistant text
|
||||
|
||||
def _strip_reasoning(text: str) -> str:
|
||||
"""Remove <REASONING_SCRATCHPAD>...</REASONING_SCRATCHPAD> blocks
|
||||
from displayed text (reasoning model internal thoughts)."""
|
||||
import re
|
||||
cleaned = re.sub(
|
||||
r"<REASONING_SCRATCHPAD>.*?</REASONING_SCRATCHPAD>\s*",
|
||||
"", text, flags=re.DOTALL,
|
||||
)
|
||||
# Also strip unclosed reasoning tags at the end
|
||||
cleaned = re.sub(
|
||||
r"<REASONING_SCRATCHPAD>.*$",
|
||||
"", cleaned, flags=re.DOTALL,
|
||||
)
|
||||
return cleaned.strip()
|
||||
|
||||
# Collect displayable entries (skip system, tool-result messages)
|
||||
entries = [] # list of (role, display_text)
|
||||
for msg in self.conversation_history:
|
||||
role = msg.get("role", "")
|
||||
content = msg.get("content")
|
||||
tool_calls = msg.get("tool_calls") or []
|
||||
|
||||
if role == "system":
|
||||
continue
|
||||
if role == "tool":
|
||||
continue
|
||||
|
||||
if role == "user":
|
||||
text = "" if content is None else str(content)
|
||||
# Handle multimodal content (list of dicts)
|
||||
if isinstance(content, list):
|
||||
parts = []
|
||||
for part in content:
|
||||
if isinstance(part, dict) and part.get("type") == "text":
|
||||
parts.append(part.get("text", ""))
|
||||
elif isinstance(part, dict) and part.get("type") == "image_url":
|
||||
parts.append("[image]")
|
||||
text = " ".join(parts)
|
||||
if len(text) > MAX_USER_LEN:
|
||||
text = text[:MAX_USER_LEN] + "..."
|
||||
entries.append(("user", text))
|
||||
|
||||
elif role == "assistant":
|
||||
text = "" if content is None else str(content)
|
||||
text = _strip_reasoning(text)
|
||||
parts = []
|
||||
if text:
|
||||
lines = text.splitlines()
|
||||
if len(lines) > MAX_ASST_LINES:
|
||||
text = "\n".join(lines[:MAX_ASST_LINES]) + " ..."
|
||||
if len(text) > MAX_ASST_LEN:
|
||||
text = text[:MAX_ASST_LEN] + "..."
|
||||
parts.append(text)
|
||||
if tool_calls:
|
||||
tc_count = len(tool_calls)
|
||||
# Extract tool names
|
||||
names = []
|
||||
for tc in tool_calls:
|
||||
fn = tc.get("function", {})
|
||||
name = fn.get("name", "unknown") if isinstance(fn, dict) else "unknown"
|
||||
if name not in names:
|
||||
names.append(name)
|
||||
names_str = ", ".join(names[:4])
|
||||
if len(names) > 4:
|
||||
names_str += ", ..."
|
||||
noun = "call" if tc_count == 1 else "calls"
|
||||
parts.append(f"[{tc_count} tool {noun}: {names_str}]")
|
||||
if not parts:
|
||||
# Skip pure-reasoning messages that have no visible output
|
||||
continue
|
||||
entries.append(("assistant", " ".join(parts)))
|
||||
|
||||
if not entries:
|
||||
return
|
||||
|
||||
# Determine if we need to truncate
|
||||
skipped = 0
|
||||
if len(entries) > MAX_DISPLAY_EXCHANGES * 2:
|
||||
skipped = len(entries) - MAX_DISPLAY_EXCHANGES * 2
|
||||
entries = entries[skipped:]
|
||||
|
||||
# Build the display using Rich
|
||||
from rich.panel import Panel
|
||||
from rich.text import Text
|
||||
|
||||
lines = Text()
|
||||
if skipped:
|
||||
lines.append(
|
||||
f" ... {skipped} earlier messages ...\n\n",
|
||||
style="dim italic",
|
||||
)
|
||||
|
||||
for i, (role, text) in enumerate(entries):
|
||||
if role == "user":
|
||||
lines.append(" ● You: ", style="dim bold #DAA520")
|
||||
# Show first line inline, indent rest
|
||||
msg_lines = text.splitlines()
|
||||
lines.append(msg_lines[0] + "\n", style="dim")
|
||||
for ml in msg_lines[1:]:
|
||||
lines.append(f" {ml}\n", style="dim")
|
||||
else:
|
||||
lines.append(" ◆ Hermes: ", style="dim bold #8FBC8F")
|
||||
msg_lines = text.splitlines()
|
||||
lines.append(msg_lines[0] + "\n", style="dim")
|
||||
for ml in msg_lines[1:]:
|
||||
lines.append(f" {ml}\n", style="dim")
|
||||
if i < len(entries) - 1:
|
||||
lines.append("") # small gap
|
||||
|
||||
panel = Panel(
|
||||
lines,
|
||||
title="[dim #DAA520]Previous Conversation[/]",
|
||||
border_style="dim #8B8682",
|
||||
padding=(0, 1),
|
||||
)
|
||||
self.console.print(panel)
|
||||
|
||||
def _try_attach_clipboard_image(self) -> bool:
|
||||
"""Check clipboard for an image and attach it if found.
|
||||
|
||||
@@ -2948,6 +3149,13 @@ class HermesCLI:
|
||||
def run(self):
|
||||
"""Run the interactive CLI loop with persistent input at bottom."""
|
||||
self.show_banner()
|
||||
|
||||
# If resuming a session, load history and display it immediately
|
||||
# so the user has context before typing their first message.
|
||||
if self._resumed:
|
||||
if self._preload_resumed_session():
|
||||
self._display_resumed_history()
|
||||
|
||||
self.console.print("[#FFF8DC]Welcome to Hermes Agent! Type your message or /help for commands.[/]")
|
||||
self.console.print()
|
||||
|
||||
|
||||
@@ -92,6 +92,7 @@ DEFAULT_CONFIG = {
|
||||
"display": {
|
||||
"compact": False,
|
||||
"personality": "kawaii",
|
||||
"resume_display": "full", # "full" (show previous messages) | "minimal" (one-liner only)
|
||||
},
|
||||
|
||||
# Text-to-speech configuration
|
||||
|
||||
488
tests/test_resume_display.py
Normal file
488
tests/test_resume_display.py
Normal file
@@ -0,0 +1,488 @@
|
||||
"""Tests for session resume history display — _display_resumed_history() and
|
||||
_preload_resumed_session().
|
||||
|
||||
Verifies that resuming a session shows a compact recap of the previous
|
||||
conversation with correct formatting, truncation, and config behavior.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
from io import StringIO
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
||||
|
||||
|
||||
def _make_cli(config_overrides=None, env_overrides=None, **kwargs):
|
||||
"""Create a HermesCLI instance with minimal mocking."""
|
||||
import cli as _cli_mod
|
||||
from cli import HermesCLI
|
||||
|
||||
_clean_config = {
|
||||
"model": {
|
||||
"default": "anthropic/claude-opus-4.6",
|
||||
"base_url": "https://openrouter.ai/api/v1",
|
||||
"provider": "auto",
|
||||
},
|
||||
"display": {"compact": False, "tool_progress": "all", "resume_display": "full"},
|
||||
"agent": {},
|
||||
"terminal": {"env_type": "local"},
|
||||
}
|
||||
if config_overrides:
|
||||
for k, v in config_overrides.items():
|
||||
if isinstance(v, dict) and k in _clean_config and isinstance(_clean_config[k], dict):
|
||||
_clean_config[k].update(v)
|
||||
else:
|
||||
_clean_config[k] = v
|
||||
|
||||
clean_env = {"LLM_MODEL": "", "HERMES_MAX_ITERATIONS": ""}
|
||||
if env_overrides:
|
||||
clean_env.update(env_overrides)
|
||||
with (
|
||||
patch("cli.get_tool_definitions", return_value=[]),
|
||||
patch.dict("os.environ", clean_env, clear=False),
|
||||
patch.dict(_cli_mod.__dict__, {"CLI_CONFIG": _clean_config}),
|
||||
):
|
||||
return HermesCLI(**kwargs)
|
||||
|
||||
|
||||
# ── Sample conversation histories for tests ──────────────────────────
|
||||
|
||||
|
||||
def _simple_history():
|
||||
"""Two-turn conversation: user → assistant → user → assistant."""
|
||||
return [
|
||||
{"role": "system", "content": "You are a helpful assistant."},
|
||||
{"role": "user", "content": "What is Python?"},
|
||||
{"role": "assistant", "content": "Python is a high-level programming language."},
|
||||
{"role": "user", "content": "How do I install it?"},
|
||||
{"role": "assistant", "content": "You can install Python from python.org."},
|
||||
]
|
||||
|
||||
|
||||
def _tool_call_history():
|
||||
"""Conversation with tool calls and tool results."""
|
||||
return [
|
||||
{"role": "system", "content": "system prompt"},
|
||||
{"role": "user", "content": "Search for Python tutorials"},
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": None,
|
||||
"tool_calls": [
|
||||
{
|
||||
"id": "call_1",
|
||||
"type": "function",
|
||||
"function": {"name": "web_search", "arguments": '{"query":"python tutorials"}'},
|
||||
},
|
||||
{
|
||||
"id": "call_2",
|
||||
"type": "function",
|
||||
"function": {"name": "web_extract", "arguments": '{"urls":["https://example.com"]}'},
|
||||
},
|
||||
],
|
||||
},
|
||||
{"role": "tool", "tool_call_id": "call_1", "content": "Found 5 results..."},
|
||||
{"role": "tool", "tool_call_id": "call_2", "content": "Page content..."},
|
||||
{"role": "assistant", "content": "Here are some great Python tutorials I found."},
|
||||
]
|
||||
|
||||
|
||||
def _large_history(n_exchanges=15):
|
||||
"""Build a history with many exchanges to test truncation."""
|
||||
msgs = [{"role": "system", "content": "system prompt"}]
|
||||
for i in range(n_exchanges):
|
||||
msgs.append({"role": "user", "content": f"Question #{i + 1}: What is item {i + 1}?"})
|
||||
msgs.append({"role": "assistant", "content": f"Answer #{i + 1}: Item {i + 1} is great."})
|
||||
return msgs
|
||||
|
||||
|
||||
def _multimodal_history():
|
||||
"""Conversation with multimodal (image) content."""
|
||||
return [
|
||||
{"role": "system", "content": "system prompt"},
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"type": "text", "text": "What's in this image?"},
|
||||
{"type": "image_url", "image_url": {"url": "https://example.com/cat.jpg"}},
|
||||
],
|
||||
},
|
||||
{"role": "assistant", "content": "I see a cat in the image."},
|
||||
]
|
||||
|
||||
|
||||
# ── Tests for _display_resumed_history ───────────────────────────────
|
||||
|
||||
|
||||
class TestDisplayResumedHistory:
|
||||
"""_display_resumed_history() renders a Rich panel with conversation recap."""
|
||||
|
||||
def _capture_display(self, cli_obj):
|
||||
"""Run _display_resumed_history and capture the Rich console output."""
|
||||
buf = StringIO()
|
||||
cli_obj.console.file = buf
|
||||
cli_obj._display_resumed_history()
|
||||
return buf.getvalue()
|
||||
|
||||
def test_simple_history_shows_user_and_assistant(self):
|
||||
cli = _make_cli()
|
||||
cli.conversation_history = _simple_history()
|
||||
output = self._capture_display(cli)
|
||||
|
||||
assert "You:" in output
|
||||
assert "Hermes:" in output
|
||||
assert "What is Python?" in output
|
||||
assert "Python is a high-level programming language." in output
|
||||
assert "How do I install it?" in output
|
||||
|
||||
def test_system_messages_hidden(self):
|
||||
cli = _make_cli()
|
||||
cli.conversation_history = _simple_history()
|
||||
output = self._capture_display(cli)
|
||||
|
||||
assert "You are a helpful assistant" not in output
|
||||
|
||||
def test_tool_messages_hidden(self):
|
||||
cli = _make_cli()
|
||||
cli.conversation_history = _tool_call_history()
|
||||
output = self._capture_display(cli)
|
||||
|
||||
# Tool result content should NOT appear
|
||||
assert "Found 5 results" not in output
|
||||
assert "Page content" not in output
|
||||
|
||||
def test_tool_calls_shown_as_summary(self):
|
||||
cli = _make_cli()
|
||||
cli.conversation_history = _tool_call_history()
|
||||
output = self._capture_display(cli)
|
||||
|
||||
assert "2 tool calls" in output
|
||||
assert "web_search" in output
|
||||
assert "web_extract" in output
|
||||
|
||||
def test_long_user_message_truncated(self):
|
||||
cli = _make_cli()
|
||||
long_text = "A" * 500
|
||||
cli.conversation_history = [
|
||||
{"role": "user", "content": long_text},
|
||||
{"role": "assistant", "content": "OK."},
|
||||
]
|
||||
output = self._capture_display(cli)
|
||||
|
||||
# Should have truncation indicator and NOT contain the full 500 chars
|
||||
assert "..." in output
|
||||
assert "A" * 500 not in output
|
||||
# The 300-char truncated text is present but may be line-wrapped by
|
||||
# Rich's panel renderer, so check the total A count in the output
|
||||
a_count = output.count("A")
|
||||
assert 200 <= a_count <= 310 # roughly 300 chars (±panel padding)
|
||||
|
||||
def test_long_assistant_message_truncated(self):
|
||||
cli = _make_cli()
|
||||
long_text = "B" * 400
|
||||
cli.conversation_history = [
|
||||
{"role": "user", "content": "Tell me a lot."},
|
||||
{"role": "assistant", "content": long_text},
|
||||
]
|
||||
output = self._capture_display(cli)
|
||||
|
||||
assert "..." in output
|
||||
assert "B" * 400 not in output
|
||||
|
||||
def test_multiline_assistant_truncated(self):
|
||||
cli = _make_cli()
|
||||
multi = "\n".join([f"Line {i}" for i in range(20)])
|
||||
cli.conversation_history = [
|
||||
{"role": "user", "content": "Show me lines."},
|
||||
{"role": "assistant", "content": multi},
|
||||
]
|
||||
output = self._capture_display(cli)
|
||||
|
||||
# First 3 lines should be there
|
||||
assert "Line 0" in output
|
||||
assert "Line 1" in output
|
||||
assert "Line 2" in output
|
||||
# Line 19 should NOT be there (truncated after 3 lines)
|
||||
assert "Line 19" not in output
|
||||
|
||||
def test_large_history_shows_truncation_indicator(self):
|
||||
cli = _make_cli()
|
||||
cli.conversation_history = _large_history(n_exchanges=15)
|
||||
output = self._capture_display(cli)
|
||||
|
||||
# Should show "earlier messages" indicator
|
||||
assert "earlier messages" in output
|
||||
# Last question should still be visible
|
||||
assert "Question #15" in output
|
||||
|
||||
def test_multimodal_content_handled(self):
|
||||
cli = _make_cli()
|
||||
cli.conversation_history = _multimodal_history()
|
||||
output = self._capture_display(cli)
|
||||
|
||||
assert "What's in this image?" in output
|
||||
assert "[image]" in output
|
||||
|
||||
def test_empty_history_no_output(self):
|
||||
cli = _make_cli()
|
||||
cli.conversation_history = []
|
||||
output = self._capture_display(cli)
|
||||
|
||||
assert output.strip() == ""
|
||||
|
||||
def test_minimal_config_suppresses_display(self):
|
||||
cli = _make_cli(config_overrides={"display": {"resume_display": "minimal"}})
|
||||
# resume_display is captured as an instance variable during __init__
|
||||
assert cli.resume_display == "minimal"
|
||||
cli.conversation_history = _simple_history()
|
||||
output = self._capture_display(cli)
|
||||
|
||||
assert output.strip() == ""
|
||||
|
||||
def test_panel_has_title(self):
|
||||
cli = _make_cli()
|
||||
cli.conversation_history = _simple_history()
|
||||
output = self._capture_display(cli)
|
||||
|
||||
assert "Previous Conversation" in output
|
||||
|
||||
def test_assistant_with_no_content_no_tools_skipped(self):
|
||||
"""Assistant messages with no visible output (e.g. pure reasoning)
|
||||
are skipped in the recap."""
|
||||
cli = _make_cli()
|
||||
cli.conversation_history = [
|
||||
{"role": "user", "content": "Hello"},
|
||||
{"role": "assistant", "content": None},
|
||||
]
|
||||
output = self._capture_display(cli)
|
||||
|
||||
# The assistant entry should be skipped, only the user message shown
|
||||
assert "You:" in output
|
||||
assert "Hermes:" not in output
|
||||
|
||||
def test_only_system_messages_no_output(self):
|
||||
cli = _make_cli()
|
||||
cli.conversation_history = [
|
||||
{"role": "system", "content": "You are helpful."},
|
||||
]
|
||||
output = self._capture_display(cli)
|
||||
|
||||
assert output.strip() == ""
|
||||
|
||||
def test_reasoning_scratchpad_stripped(self):
|
||||
"""<REASONING_SCRATCHPAD> blocks should be stripped from display."""
|
||||
cli = _make_cli()
|
||||
cli.conversation_history = [
|
||||
{"role": "user", "content": "Think about this"},
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": (
|
||||
"<REASONING_SCRATCHPAD>\nLet me think step by step.\n"
|
||||
"</REASONING_SCRATCHPAD>\n\nThe answer is 42."
|
||||
),
|
||||
},
|
||||
]
|
||||
output = self._capture_display(cli)
|
||||
|
||||
assert "REASONING_SCRATCHPAD" not in output
|
||||
assert "Let me think step by step" not in output
|
||||
assert "The answer is 42" in output
|
||||
|
||||
def test_pure_reasoning_message_skipped(self):
|
||||
"""Assistant messages that are only reasoning should be skipped."""
|
||||
cli = _make_cli()
|
||||
cli.conversation_history = [
|
||||
{"role": "user", "content": "Hello"},
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": "<REASONING_SCRATCHPAD>\nJust thinking...\n</REASONING_SCRATCHPAD>",
|
||||
},
|
||||
{"role": "assistant", "content": "Hi there!"},
|
||||
]
|
||||
output = self._capture_display(cli)
|
||||
|
||||
assert "Just thinking" not in output
|
||||
assert "Hi there!" in output
|
||||
|
||||
def test_assistant_with_text_and_tool_calls(self):
|
||||
"""When an assistant message has both text content AND tool_calls."""
|
||||
cli = _make_cli()
|
||||
cli.conversation_history = [
|
||||
{"role": "user", "content": "Do something complex"},
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": "Let me search for that.",
|
||||
"tool_calls": [
|
||||
{
|
||||
"id": "call_1",
|
||||
"type": "function",
|
||||
"function": {"name": "terminal", "arguments": '{"command":"ls"}'},
|
||||
}
|
||||
],
|
||||
},
|
||||
]
|
||||
output = self._capture_display(cli)
|
||||
|
||||
assert "Let me search for that." in output
|
||||
assert "1 tool call" in output
|
||||
assert "terminal" in output
|
||||
|
||||
|
||||
# ── Tests for _preload_resumed_session ──────────────────────────────
|
||||
|
||||
|
||||
class TestPreloadResumedSession:
|
||||
"""_preload_resumed_session() loads session from DB early."""
|
||||
|
||||
def test_returns_false_when_not_resumed(self):
|
||||
cli = _make_cli()
|
||||
assert cli._preload_resumed_session() is False
|
||||
|
||||
def test_returns_false_when_no_session_db(self):
|
||||
cli = _make_cli(resume="test_session_id")
|
||||
cli._session_db = None
|
||||
assert cli._preload_resumed_session() is False
|
||||
|
||||
def test_returns_false_when_session_not_found(self):
|
||||
cli = _make_cli(resume="nonexistent_session")
|
||||
mock_db = MagicMock()
|
||||
mock_db.get_session.return_value = None
|
||||
cli._session_db = mock_db
|
||||
|
||||
buf = StringIO()
|
||||
cli.console.file = buf
|
||||
result = cli._preload_resumed_session()
|
||||
|
||||
assert result is False
|
||||
output = buf.getvalue()
|
||||
assert "Session not found" in output
|
||||
|
||||
def test_returns_false_when_session_has_no_messages(self):
|
||||
cli = _make_cli(resume="empty_session")
|
||||
mock_db = MagicMock()
|
||||
mock_db.get_session.return_value = {"id": "empty_session", "title": None}
|
||||
mock_db.get_messages_as_conversation.return_value = []
|
||||
cli._session_db = mock_db
|
||||
|
||||
buf = StringIO()
|
||||
cli.console.file = buf
|
||||
result = cli._preload_resumed_session()
|
||||
|
||||
assert result is False
|
||||
output = buf.getvalue()
|
||||
assert "no messages" in output
|
||||
|
||||
def test_loads_session_successfully(self):
|
||||
cli = _make_cli(resume="good_session")
|
||||
messages = _simple_history()
|
||||
mock_db = MagicMock()
|
||||
mock_db.get_session.return_value = {"id": "good_session", "title": "Test Session"}
|
||||
mock_db.get_messages_as_conversation.return_value = messages
|
||||
cli._session_db = mock_db
|
||||
|
||||
buf = StringIO()
|
||||
cli.console.file = buf
|
||||
result = cli._preload_resumed_session()
|
||||
|
||||
assert result is True
|
||||
assert cli.conversation_history == messages
|
||||
output = buf.getvalue()
|
||||
assert "Resumed session" in output
|
||||
assert "good_session" in output
|
||||
assert "Test Session" in output
|
||||
assert "2 user messages" in output
|
||||
|
||||
def test_reopens_session_in_db(self):
|
||||
cli = _make_cli(resume="reopen_session")
|
||||
messages = [{"role": "user", "content": "hi"}]
|
||||
mock_db = MagicMock()
|
||||
mock_db.get_session.return_value = {"id": "reopen_session", "title": None}
|
||||
mock_db.get_messages_as_conversation.return_value = messages
|
||||
mock_conn = MagicMock()
|
||||
mock_db._conn = mock_conn
|
||||
cli._session_db = mock_db
|
||||
|
||||
buf = StringIO()
|
||||
cli.console.file = buf
|
||||
cli._preload_resumed_session()
|
||||
|
||||
# Should have executed UPDATE to clear ended_at
|
||||
mock_conn.execute.assert_called_once()
|
||||
call_args = mock_conn.execute.call_args
|
||||
assert "ended_at = NULL" in call_args[0][0]
|
||||
mock_conn.commit.assert_called_once()
|
||||
|
||||
def test_singular_user_message_grammar(self):
|
||||
"""1 user message should say 'message' not 'messages'."""
|
||||
cli = _make_cli(resume="one_msg_session")
|
||||
messages = [
|
||||
{"role": "user", "content": "hello"},
|
||||
{"role": "assistant", "content": "hi"},
|
||||
]
|
||||
mock_db = MagicMock()
|
||||
mock_db.get_session.return_value = {"id": "one_msg_session", "title": None}
|
||||
mock_db.get_messages_as_conversation.return_value = messages
|
||||
mock_db._conn = MagicMock()
|
||||
cli._session_db = mock_db
|
||||
|
||||
buf = StringIO()
|
||||
cli.console.file = buf
|
||||
cli._preload_resumed_session()
|
||||
|
||||
output = buf.getvalue()
|
||||
assert "1 user message," in output
|
||||
assert "1 user messages" not in output
|
||||
|
||||
|
||||
# ── Integration: _init_agent skips when preloaded ────────────────────
|
||||
|
||||
|
||||
class TestInitAgentSkipsPreloaded:
|
||||
"""_init_agent() should skip DB load when history is already populated."""
|
||||
|
||||
def test_init_agent_skips_db_when_preloaded(self):
|
||||
"""If conversation_history is already set, _init_agent should not
|
||||
reload from the DB."""
|
||||
cli = _make_cli(resume="preloaded_session")
|
||||
cli.conversation_history = _simple_history()
|
||||
|
||||
mock_db = MagicMock()
|
||||
cli._session_db = mock_db
|
||||
|
||||
# _init_agent will fail at credential resolution (no real API key),
|
||||
# but the session-loading block should be skipped entirely
|
||||
with patch.object(cli, "_ensure_runtime_credentials", return_value=False):
|
||||
cli._init_agent()
|
||||
|
||||
# get_messages_as_conversation should NOT have been called
|
||||
mock_db.get_messages_as_conversation.assert_not_called()
|
||||
|
||||
|
||||
# ── Config default tests ─────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestResumeDisplayConfig:
|
||||
"""resume_display config option defaults and behavior."""
|
||||
|
||||
def test_default_config_has_resume_display(self):
|
||||
"""DEFAULT_CONFIG in hermes_cli/config.py includes resume_display."""
|
||||
from hermes_cli.config import DEFAULT_CONFIG
|
||||
display = DEFAULT_CONFIG.get("display", {})
|
||||
assert "resume_display" in display
|
||||
assert display["resume_display"] == "full"
|
||||
|
||||
def test_cli_defaults_have_resume_display(self):
|
||||
"""cli.py load_cli_config defaults include resume_display."""
|
||||
import cli as _cli_mod
|
||||
from cli import load_cli_config
|
||||
|
||||
with (
|
||||
patch("pathlib.Path.exists", return_value=False),
|
||||
patch.dict("os.environ", {"LLM_MODEL": ""}, clear=False),
|
||||
):
|
||||
config = load_cli_config()
|
||||
|
||||
display = config.get("display", {})
|
||||
assert display.get("resume_display") == "full"
|
||||
Reference in New Issue
Block a user