diff --git a/cli.py b/cli.py index f0ba6fc991..7712fc95d3 100644 --- a/cli.py +++ b/cli.py @@ -4912,6 +4912,40 @@ class HermesCLI: flush_tool_summary() print() + + def _handle_recap_command(self) -> None: + """Show a compact recap of recent activity in this session. + + Inspired by Claude Code's ``/recap`` (v2.1.114, April 2026) — useful + when running multiple sessions simultaneously and returning to one + after a while. Purely local; no LLM call, no token cost, no cache + invalidation. + """ + try: + from hermes_cli.session_recap import build_recap + except Exception as exc: # pragma: no cover - defensive + print(f" (recap unavailable: {exc})") + return + + title = None + try: + if self._session_db and self.session_id: + row = self._session_db.get_session(self.session_id) + if row: + title = row.get("title") or None + except Exception: + title = None + + text = build_recap( + self.conversation_history or [], + session_title=title, + session_id=self.session_id, + platform="cli", + ) + print() + for line in text.splitlines(): + print(line) + print() def _notify_session_boundary(self, event_type: str) -> None: """Fire a session-boundary plugin hook (on_session_finalize or on_session_reset). @@ -6364,6 +6398,8 @@ class HermesCLI: pass elif canonical == "history": self.show_history() + elif canonical == "recap": + self._handle_recap_command() elif canonical == "title": parts = cmd_original.split(maxsplit=1) if len(parts) > 1: diff --git a/gateway/run.py b/gateway/run.py index 88196d6927..f7604e1748 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -4630,6 +4630,9 @@ class GatewayRunner: if event.get_command() == "status": return await self._handle_status_command(event) + if event.get_command() == "recap": + return await self._handle_recap_command(event) + # Resolve the command once for all early-intercept checks below. from hermes_cli.commands import ( ACTIVE_SESSION_BYPASS_COMMANDS as _DEDICATED_HANDLERS, @@ -5004,6 +5007,9 @@ class GatewayRunner: if canonical == "status": return await self._handle_status_command(event) + if canonical == "recap": + return await self._handle_recap_command(event) + if canonical == "agents": return await self._handle_agents_command(event) @@ -6846,6 +6852,34 @@ class GatewayRunner: return "\n".join(lines) + async def _handle_recap_command(self, event: MessageEvent) -> str: + """Handle /recap command — compact summary of recent session activity. + + Inspired by Claude Code's ``/recap`` (v2.1.114, April 2026). Purely + local: reads the transcript from SessionStore and builds a text + summary with no LLM call. Safe for prompt-cache integrity. + """ + from hermes_cli.session_recap import build_recap + + source = event.source + session_entry = self.session_store.get_or_create_session(source) + history = self.session_store.load_transcript(session_entry.session_id) or [] + + title = None + try: + if self._session_db: + title = self._session_db.get_session_title(session_entry.session_id) + except Exception: + title = None + + platform_name = source.platform.value if source and source.platform else None + return build_recap( + history, + session_title=title, + session_id=session_entry.session_id, + platform=platform_name, + ) + async def _handle_agents_command(self, event: MessageEvent) -> str: """Handle /agents command - list active agents and running tasks.""" from tools.process_registry import format_uptime_short, process_registry diff --git a/hermes_cli/commands.py b/hermes_cli/commands.py index 41b1dad500..589475cc9a 100644 --- a/hermes_cli/commands.py +++ b/hermes_cli/commands.py @@ -68,6 +68,7 @@ COMMAND_REGISTRY: list[CommandDef] = [ cli_only=True), CommandDef("history", "Show conversation history", "Session", cli_only=True), + CommandDef("recap", "Summarize recent activity in this session", "Session"), CommandDef("save", "Save the current conversation", "Session", cli_only=True), CommandDef("retry", "Retry the last message (resend to agent)", "Session"), @@ -319,6 +320,7 @@ ACTIVE_SESSION_BYPASS_COMMANDS: frozenset[str] = frozenset( "new", "profile", "queue", + "recap", "restart", "status", "steer", diff --git a/hermes_cli/session_recap.py b/hermes_cli/session_recap.py new file mode 100644 index 0000000000..d67f737d79 --- /dev/null +++ b/hermes_cli/session_recap.py @@ -0,0 +1,316 @@ +"""Session recap — summarize what's happened in the current session. + +Inspired by Claude Code's `/recap` command (v2.1.114, April 2026), which +shows a one-line summary of what happened while a terminal was unfocused +so users juggling multiple sessions can re-orient quickly. + +Source: https://code.claude.com/docs/en/whats-new/2026-w17 + +Differences from Claude Code: + - Pure local computation from the in-memory conversation history. No + LLM call, no auxiliary model, no prompt-cache invalidation. A + recap should be instant and free. + - Works unchanged on CLI and every gateway platform (Telegram, + Discord, Slack, …) because both call into the same ``build_recap`` + helper. Claude Code only shows this on the CLI. + - Tailored to hermes-agent's tool vocabulary (``terminal``, ``patch``, + ``write_file``, ``delegate_task``, ``browser_*``, ``web_*``) — the + recap surfaces which classes of work were most active. +""" +from __future__ import annotations + +import os +from collections import Counter +from typing import Any, Iterable, List, Mapping, Optional, Sequence, Tuple + +# How many recent user/assistant turns we consider "recent activity". +_RECENT_TURN_WINDOW = 20 + +# How many characters of the latest user prompt to show. +_PROMPT_PREVIEW_CHARS = 140 + +# How many characters of the latest assistant text to show. +_ASSISTANT_PREVIEW_CHARS = 200 + +# How many recently-touched files to list. +_MAX_FILES_LISTED = 5 + +# Tool names that identify a file-editing action and the argument key that +# holds the path. +_FILE_EDIT_TOOLS: Mapping[str, str] = { + "write_file": "path", + "patch": "path", + "read_file": "path", + "skill_manage": "file_path", + "skill_view": "file_path", +} + + +def _coerce_text(value: Any) -> str: + """Flatten assistant/user ``content`` into a plain string. + + Content can be a string or a list of content blocks (for multimodal + or reasoning models). We concatenate every text-like block and + ignore the rest. + """ + if value is None: + return "" + if isinstance(value, str): + return value + if isinstance(value, list): + parts: List[str] = [] + for block in value: + if isinstance(block, str): + parts.append(block) + continue + if isinstance(block, Mapping): + text = block.get("text") + if isinstance(text, str) and text: + parts.append(text) + return "\n".join(parts) + return str(value) + + +def _tool_call_name_and_args(tool_call: Any) -> Tuple[str, Mapping[str, Any]]: + """Extract ``(name, arguments_dict)`` from a tool_call entry. + + ``arguments`` may be a JSON string or a dict depending on provider. + Return an empty dict if it cannot be parsed. + """ + if not isinstance(tool_call, Mapping): + return "", {} + fn = tool_call.get("function") or {} + if not isinstance(fn, Mapping): + return "", {} + name = str(fn.get("name") or "") or "" + raw_args = fn.get("arguments") + if isinstance(raw_args, Mapping): + return name, raw_args + if isinstance(raw_args, str) and raw_args: + try: + import json + + parsed = json.loads(raw_args) + if isinstance(parsed, Mapping): + return name, parsed + except Exception: + return name, {} + return name, {} + + +def _iter_assistant_tool_calls( + messages: Sequence[Mapping[str, Any]], +) -> Iterable[Tuple[str, Mapping[str, Any]]]: + for msg in messages: + if not isinstance(msg, Mapping): + continue + if msg.get("role") != "assistant": + continue + tool_calls = msg.get("tool_calls") or [] + if not isinstance(tool_calls, list): + continue + for tc in tool_calls: + name, args = _tool_call_name_and_args(tc) + if name: + yield name, args + + +def _count_visible_turns( + messages: Sequence[Mapping[str, Any]], +) -> Tuple[int, int, int]: + """Return ``(user_turn_count, assistant_turn_count, tool_message_count)``.""" + users = assistants = tools = 0 + for msg in messages: + if not isinstance(msg, Mapping): + continue + role = msg.get("role") + if role == "user": + users += 1 + elif role == "assistant": + assistants += 1 + elif role == "tool": + tools += 1 + return users, assistants, tools + + +def _latest_user_prompt( + messages: Sequence[Mapping[str, Any]], +) -> Optional[str]: + for msg in reversed(messages): + if isinstance(msg, Mapping) and msg.get("role") == "user": + text = _coerce_text(msg.get("content")).strip() + if text: + return text + return None + + +def _latest_assistant_text( + messages: Sequence[Mapping[str, Any]], +) -> Optional[str]: + for msg in reversed(messages): + if not isinstance(msg, Mapping): + continue + if msg.get("role") != "assistant": + continue + text = _coerce_text(msg.get("content")).strip() + if text: + return text + return None + + +def _recent_window( + messages: Sequence[Mapping[str, Any]], window: int = _RECENT_TURN_WINDOW +) -> List[Mapping[str, Any]]: + """Return the tail slice of ``messages`` covering at most ``window`` + user+assistant turns (tool messages ride along inside the window). + + Iterating from the end, we count user and assistant messages and + keep everything from the first message that falls within the window. + """ + count = 0 + cut = 0 + for i in range(len(messages) - 1, -1, -1): + msg = messages[i] + if isinstance(msg, Mapping) and msg.get("role") in ("user", "assistant"): + count += 1 + if count >= window: + cut = i + break + else: + return list(messages) + return list(messages[cut:]) + + +def _shortened_path(path: str) -> str: + """Show a path relative to cwd when possible, otherwise with ~ expansion.""" + if not path: + return path + try: + abs_path = os.path.abspath(os.path.expanduser(path)) + cwd = os.getcwd() + if abs_path == cwd: + return "." + if abs_path.startswith(cwd + os.sep): + return abs_path[len(cwd) + 1 :] + home = os.path.expanduser("~") + if abs_path.startswith(home + os.sep): + return "~/" + abs_path[len(home) + 1 :] + return abs_path + except Exception: + return path + + +def _summarise_tool_activity( + tool_calls: Sequence[Tuple[str, Mapping[str, Any]]], +) -> Tuple[List[Tuple[str, int]], List[str]]: + """Return ``(tool_counts_sorted, recently_edited_files)``. + + ``tool_counts_sorted`` is descending by count, keeping the full list + so callers can truncate for display. ``recently_edited_files`` lists + distinct paths (most recent first) from file-editing tools. + """ + counter: Counter[str] = Counter() + files_seen: List[str] = [] + files_set: set[str] = set() + # Walk in reverse so "most recent first" drops out of order-preserved iteration. + for name, args in reversed(list(tool_calls)): + counter[name] += 1 + arg_key = _FILE_EDIT_TOOLS.get(name) + if arg_key: + path = args.get(arg_key) + if isinstance(path, str) and path and path not in files_set: + files_set.add(path) + files_seen.append(_shortened_path(path)) + # Restore "reverse of reverse" for correct counts; Counter ignores order + # so only files_seen needed the reversal. Fix ordering: currently + # files_seen is newest→oldest which is what we want for display. + tool_counts = sorted(counter.items(), key=lambda kv: (-kv[1], kv[0])) + return tool_counts, files_seen + + +def _truncate(text: str, limit: int) -> str: + text = " ".join(text.split()) # collapse newlines for a compact one-liner + if len(text) <= limit: + return text + return text[: limit - 1].rstrip() + "…" + + +def build_recap( + messages: Sequence[Mapping[str, Any]], + *, + session_title: Optional[str] = None, + session_id: Optional[str] = None, + platform: Optional[str] = None, +) -> str: + """Build a multi-line recap of recent activity. + + Inputs: + messages: the full conversation history as a list of + chat-completion-style dicts (``role``, ``content``, + ``tool_calls``, …). + session_title: optional human title (from SessionDB). + session_id: optional session id. + platform: optional hint (``"cli"``, ``"telegram"``, …). Does not + change behavior today but is accepted for forward compat. + + The output is plain text designed to render well in both a terminal + (with 80-col wrapping) and a gateway message bubble. + """ + _ = platform # reserved for future use + lines: List[str] = [] + + header_bits: List[str] = ["Session recap"] + if session_title: + header_bits.append(f"— {session_title}") + elif session_id: + header_bits.append(f"— {session_id[:8]}") + lines.append(" ".join(header_bits)) + + if not messages: + lines.append(" (nothing to recap — no messages yet)") + return "\n".join(lines) + + users, assistants, tool_msgs = _count_visible_turns(messages) + window = _recent_window(messages) + win_users, win_assistants, _ = _count_visible_turns(window) + + scope = ( + f"{win_users} user turn{'s' if win_users != 1 else ''} / " + f"{win_assistants} assistant repl{'ies' if win_assistants != 1 else 'y'}" + ) + if (users, assistants) != (win_users, win_assistants): + scope += f" (of {users}/{assistants} total)" + lines.append(f" Recent: {scope}, {tool_msgs} tool result{'s' if tool_msgs != 1 else ''}") + + tool_calls = list(_iter_assistant_tool_calls(window)) + tool_counts, files = _summarise_tool_activity(tool_calls) + if tool_counts: + top = ", ".join(f"{name}×{count}" for name, count in tool_counts[:5]) + extra = len(tool_counts) - 5 + if extra > 0: + top += f" (+{extra} more)" + lines.append(f" Tools used: {top}") + if files: + shown = files[:_MAX_FILES_LISTED] + extra = len(files) - len(shown) + entry = ", ".join(shown) + if extra > 0: + entry += f" (+{extra} more)" + lines.append(f" Files touched: {entry}") + + latest_user = _latest_user_prompt(window) + if latest_user: + lines.append(f" Last ask: {_truncate(latest_user, _PROMPT_PREVIEW_CHARS)}") + + latest_reply = _latest_assistant_text(window) + if latest_reply: + lines.append(f" Last reply: {_truncate(latest_reply, _ASSISTANT_PREVIEW_CHARS)}") + + if len(lines) == 2: + # Only the header + scope line — nothing substantive to show. + lines.append(" (no assistant activity yet in this window)") + + return "\n".join(lines) + + +__all__ = ["build_recap"] diff --git a/tests/hermes_cli/test_session_recap.py b/tests/hermes_cli/test_session_recap.py new file mode 100644 index 0000000000..3998c06c61 --- /dev/null +++ b/tests/hermes_cli/test_session_recap.py @@ -0,0 +1,180 @@ +"""Unit tests for hermes_cli.session_recap.""" +from __future__ import annotations + +import json + +import pytest + +from hermes_cli.session_recap import build_recap + + +def _user(text): + return {"role": "user", "content": text} + + +def _assistant(text=None, tool_calls=None): + msg = {"role": "assistant", "content": text} + if tool_calls: + msg["tool_calls"] = tool_calls + return msg + + +def _tool_call(name, args): + return { + "id": f"call_{name}", + "type": "function", + "function": {"name": name, "arguments": json.dumps(args)}, + } + + +def _tool_result(content="ok"): + return {"role": "tool", "content": content} + + +def test_empty_history(): + out = build_recap([]) + assert "Session recap" in out + assert "nothing to recap" in out + + +def test_header_shows_title_when_provided(): + out = build_recap([_user("hello")], session_title="Refactor the adapter") + assert "Refactor the adapter" in out.splitlines()[0] + + +def test_header_shows_short_id_when_no_title(): + out = build_recap([_user("hello")], session_id="abcdef1234567890") + assert "abcdef12" in out.splitlines()[0] + + +def test_counts_recent_turns(): + msgs = [ + _user("one"), + _assistant("first reply"), + _user("two"), + _assistant("second reply"), + ] + out = build_recap(msgs) + assert "2 user turn" in out + assert "assistant repl" in out + + +def test_last_ask_and_reply_are_surfaced(): + msgs = [ + _user("old question"), + _assistant("old answer"), + _user("summarise the docs"), + _assistant("here is the summary of the docs you asked for"), + ] + out = build_recap(msgs) + assert "summarise the docs" in out + assert "summary of the docs" in out + + +def test_tool_counts_and_files(): + msgs = [ + _user("edit the readme and run tests"), + _assistant( + tool_calls=[ + _tool_call("read_file", {"path": "README.md"}), + _tool_call("patch", {"path": "README.md"}), + ] + ), + _tool_result(), + _tool_result(), + _assistant( + tool_calls=[ + _tool_call("terminal", {"command": "pytest"}), + ] + ), + _tool_result("tests ok"), + _assistant("All green."), + ] + out = build_recap(msgs) + assert "patch×1" in out + assert "terminal×1" in out + assert "read_file×1" in out + # README.md should appear (may include cwd-relative prefix stripping). + assert "README.md" in out + + +def test_tool_preview_length_truncates_long_user_prompt(): + long = "x " * 500 + out = build_recap([_user(long)]) + ask_line = [l for l in out.splitlines() if "Last ask" in l][0] + assert len(ask_line) < 300 # truncated with ellipsis + assert "…" in ask_line + + +def test_respects_recent_window(): + # 30 turns of user+assistant; only the most recent 20 should be summarised. + msgs = [] + for i in range(30): + msgs.append(_user(f"question {i}")) + msgs.append(_assistant(f"answer {i}")) + out = build_recap(msgs) + # We scoped to the 20-turn window but show "of 30/30 total". + assert "of 30/30 total" in out + + +def test_multimodal_content_blocks_flattened(): + msgs = [ + { + "role": "user", + "content": [ + {"type": "text", "text": "check this file"}, + {"type": "image_url", "image_url": {"url": "..."}}, + ], + }, + _assistant("Looked at your image."), + ] + out = build_recap(msgs) + assert "check this file" in out + assert "Looked at your image" in out + + +def test_handles_arguments_as_dict_not_string(): + # Some providers return arguments already as a dict. + msgs = [ + _user("go"), + { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "type": "function", + "function": { + "name": "patch", + "arguments": {"path": "foo.py"}, + }, + } + ], + }, + ] + out = build_recap(msgs) + assert "patch×1" in out + assert "foo.py" in out + + +def test_no_assistant_activity_hint(): + out = build_recap([_user("just sent my first message")]) + assert "no assistant activity" in out or "Last ask" in out + + +def test_tool_message_count_reported(): + msgs = [ + _user("go"), + _assistant(tool_calls=[_tool_call("read_file", {"path": "a"})]), + _tool_result(), + _tool_result(), + _assistant("done"), + ] + out = build_recap(msgs) + assert "2 tool result" in out + + +def test_ignores_non_mapping_entries_gracefully(): + msgs = [None, "stray", _user("hi"), _assistant("hello")] + # Should not raise. + out = build_recap(msgs) + assert "Session recap" in out diff --git a/website/docs/reference/slash-commands.md b/website/docs/reference/slash-commands.md index ef566cd5ba..e4f6f861de 100644 --- a/website/docs/reference/slash-commands.md +++ b/website/docs/reference/slash-commands.md @@ -24,6 +24,7 @@ Type `/` in the CLI to open the autocomplete menu. Built-in commands are case-in | `/new` (alias: `/reset`) | Start a new session (fresh session ID + history) | | `/clear` | Clear screen and start a new session | | `/history` | Show conversation history | +| `/recap` | Compact summary of recent session activity (turn counts, tools used, files touched, last ask/reply). Purely local; no LLM call or token cost. Works in CLI and on every gateway platform. Inspired by Claude Code's `/recap`. | | `/save` | Save the current conversation | | `/retry` | Retry the last message (resend to agent) | | `/undo` | Remove the last user/assistant exchange |