diff --git a/hermes_cli/session_export.py b/hermes_cli/session_export.py new file mode 100644 index 00000000000..031c21034ed --- /dev/null +++ b/hermes_cli/session_export.py @@ -0,0 +1,317 @@ +"""Shared renderers for session export commands. + +The CLI, dashboard, and slash-command surfaces all deal with the same +session-shaped data: a session dict with a ``messages`` list. Keep filtering +and human-readable rendering here so each surface only has to load sessions +and write bytes. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from html import escape as html_escape +import json +from typing import Any, Dict, Iterable, Iterator, List, Literal, Optional, Tuple + + +ExportFormat = Literal["jsonl", "markdown"] +ExportOnly = Literal["user-prompts"] + + +def normalize_export_format(fmt: str) -> ExportFormat: + """Return the canonical export format name.""" + value = (fmt or "jsonl").strip().lower() + if value == "md": + value = "markdown" + if value not in {"jsonl", "markdown"}: + raise ValueError(f"Unsupported session export format: {fmt}") + return value # type: ignore[return-value] + + +def normalize_export_only(only: Optional[str]) -> Optional[ExportOnly]: + """Return the canonical export filter name.""" + if only is None: + return None + value = only.strip().lower() + if value in {"user", "prompts", "user-prompts", "user_prompts"}: + return "user-prompts" + raise ValueError(f"Unsupported session export filter: {only}") + + +def render_sessions_export( + sessions: Iterable[Dict[str, Any]], + *, + fmt: str = "jsonl", + only: Optional[str] = None, +) -> str: + """Render exported sessions in a stable, reusable format. + + ``fmt=jsonl`` with no filter intentionally preserves the legacy shape: + one full session object per line. ``only=user-prompts`` switches the unit + of export to one prompt record per line so the output is easy to pipe into + review, memory-ingestion, or prompt-library tooling. + """ + session_list = list(sessions) + export_format = normalize_export_format(fmt) + export_only = normalize_export_only(only) + + if export_format == "jsonl": + return _render_jsonl(session_list, only=export_only) + return _render_markdown(session_list, only=export_only) + + +def export_record_count( + sessions: Iterable[Dict[str, Any]], *, only: Optional[str] = None +) -> Tuple[int, str]: + """Return ``(count, noun)`` for status messages after an export.""" + session_list = list(sessions) + export_only = normalize_export_only(only) + if export_only == "user-prompts": + return sum(1 for _ in iter_user_prompt_records(session_list)), "prompt" + return len(session_list), "session" + + +def iter_user_prompt_records( + sessions: Iterable[Dict[str, Any]] +) -> Iterator[Dict[str, Any]]: + """Yield one normalized record for each user-authored prompt.""" + for session in sessions: + session_id = str(session.get("id") or session.get("session_id") or "") + index = 0 + for message in _messages(session): + if message.get("role") != "user": + continue + index += 1 + record: Dict[str, Any] = { + "session_id": session_id, + "index": index, + "created_at": _format_timestamp(message.get("timestamp")), + "role": "user", + "text": _message_text(message.get("content")), + } + message_id = message.get("id") + if message_id is not None: + record["message_id"] = message_id + event_id = message.get("platform_message_id") or message.get("event_id") + if event_id: + record["event_id"] = event_id + yield record + + +def _render_jsonl( + sessions: List[Dict[str, Any]], *, only: Optional[ExportOnly] +) -> str: + if only == "user-prompts": + rows = iter_user_prompt_records(sessions) + else: + rows = iter(sessions) + lines = [json.dumps(row, ensure_ascii=False) for row in rows] + return ("\n".join(lines) + "\n") if lines else "" + + +def _render_markdown( + sessions: List[Dict[str, Any]], *, only: Optional[ExportOnly] +) -> str: + if only == "user-prompts": + return _render_user_prompts_markdown(sessions) + return _render_full_markdown(sessions) + + +def _render_user_prompts_markdown(sessions: List[Dict[str, Any]]) -> str: + lines: List[str] = [] + if len(sessions) == 1: + session = sessions[0] + lines.append(f"# User prompts for session {_heading_text(_session_id(session))}") + lines.extend(_session_metadata_lines(session)) + lines.append("") + _append_prompt_records(lines, session, heading_level=2) + else: + lines.append("# User prompts export") + lines.append("") + for session in sessions: + lines.append(f"## Session {_heading_text(_session_id(session))}") + lines.extend(_session_metadata_lines(session)) + lines.append("") + _append_prompt_records(lines, session, heading_level=3) + if not sessions: + lines.append("_No user prompts found._") + lines.append("") + return _finish_markdown(lines) + + +def _append_prompt_records( + lines: List[str], session: Dict[str, Any], *, heading_level: int +) -> None: + prompts = list(iter_user_prompt_records([session])) + if not prompts: + lines.append("_No user prompts found._") + lines.append("") + return + marker = "#" * heading_level + for prompt in prompts: + timestamp = prompt.get("created_at") or "timestamp unavailable" + lines.append(f"{marker} {prompt['index']}. {timestamp}") + message_id = prompt.get("message_id") + if message_id is not None: + lines.append(f"Message ID: `{message_id}`") + lines.append("") + lines.append(str(prompt.get("text") or "")) + lines.append("") + + +def _render_full_markdown(sessions: List[Dict[str, Any]]) -> str: + lines: List[str] = [] + if len(sessions) == 1: + session = sessions[0] + lines.append(f"# Session: {_heading_text(_session_title_or_id(session))}") + lines.extend(_session_metadata_lines(session)) + lines.append("") + _append_session_messages(lines, session, heading_level=2) + else: + lines.append("# Hermes sessions export") + lines.append("") + for session in sessions: + lines.append(f"## Session: {_heading_text(_session_title_or_id(session))}") + lines.extend(_session_metadata_lines(session)) + lines.append("") + _append_session_messages(lines, session, heading_level=3) + return _finish_markdown(lines) + + +def _append_session_messages( + lines: List[str], session: Dict[str, Any], *, heading_level: int +) -> None: + marker = "#" * heading_level + visible_messages = [ + message for message in _messages(session) if message.get("role") != "system" + ] + if not visible_messages: + lines.append("_No messages found._") + lines.append("") + return + + for message in visible_messages: + role = str(message.get("role") or "unknown") + timestamp = _format_timestamp(message.get("timestamp")) + suffix = f" - {timestamp}" if timestamp else "" + if role == "tool": + tool_name = str(message.get("tool_name") or message.get("name") or "tool") + lines.append(f"{marker} Tool: {_heading_text(tool_name)}{suffix}") + lines.append("") + lines.append(f"
{html_escape(tool_name)}") + lines.append("") + lines.append(_fenced_text(_message_text(message.get("content")))) + lines.append("") + lines.append("
") + lines.append("") + continue + + label = { + "user": "User", + "assistant": "Assistant", + }.get(role, role.title()) + lines.append(f"{marker} {label}{suffix}") + lines.append("") + lines.append(_message_text(message.get("content"))) + lines.append("") + + +def _messages(session: Dict[str, Any]) -> List[Dict[str, Any]]: + messages = session.get("messages") or [] + return [message for message in messages if isinstance(message, dict)] + + +def _message_text(content: Any) -> str: + if content is None: + return "" + if isinstance(content, str): + return content + if isinstance(content, list): + parts = [_content_part_text(part) for part in content] + return "\n".join(part for part in parts if part) + if isinstance(content, dict): + for key in ("text", "content"): + value = content.get(key) + if isinstance(value, str): + return value + return json.dumps(content, ensure_ascii=False, sort_keys=True) + return str(content) + + +def _content_part_text(part: Any) -> str: + if isinstance(part, str): + return part + if isinstance(part, dict): + for key in ("text", "content"): + value = part.get(key) + if isinstance(value, str): + return value + return json.dumps(part, ensure_ascii=False, sort_keys=True) + return str(part) + + +def _format_timestamp(value: Any) -> Optional[str]: + if value is None: + return None + if isinstance(value, (int, float)): + return ( + datetime.fromtimestamp(float(value), tz=timezone.utc) + .isoformat(timespec="seconds") + .replace("+00:00", "Z") + ) + if isinstance(value, datetime): + dt = value if value.tzinfo else value.replace(tzinfo=timezone.utc) + return dt.astimezone(timezone.utc).isoformat(timespec="seconds").replace( + "+00:00", "Z" + ) + return str(value) + + +def _session_metadata_lines(session: Dict[str, Any]) -> List[str]: + lines: List[str] = [f"- Session ID: `{_session_id(session)}`"] + source = session.get("source") + if source: + lines.append(f"- Source: `{source}`") + model = session.get("model") + if model: + lines.append(f"- Model: `{model}`") + title = session.get("title") + if title: + lines.append(f"- Title: {_inline_text(str(title))}") + started = _format_timestamp(session.get("started_at")) + if started: + lines.append(f"- Started: {started}") + message_count = session.get("message_count") + if message_count is not None: + lines.append(f"- Messages: {message_count}") + return lines + + +def _session_id(session: Dict[str, Any]) -> str: + return str(session.get("id") or session.get("session_id") or "unknown") + + +def _session_title_or_id(session: Dict[str, Any]) -> str: + title = str(session.get("title") or "").strip() + return title or _session_id(session) + + +def _heading_text(value: str) -> str: + return " ".join(str(value).splitlines()).strip() or "unknown" + + +def _inline_text(value: str) -> str: + return " ".join(value.splitlines()).strip() + + +def _fenced_text(text: str, *, language: str = "text") -> str: + fence = "```" + while fence in text: + fence += "`" + return f"{fence}{language}\n{text}\n{fence}" + + +def _finish_markdown(lines: List[str]) -> str: + while lines and lines[-1] == "": + lines.pop() + return "\n".join(lines) + "\n" diff --git a/tests/hermes_cli/test_session_export.py b/tests/hermes_cli/test_session_export.py new file mode 100644 index 00000000000..c737ccd6861 --- /dev/null +++ b/tests/hermes_cli/test_session_export.py @@ -0,0 +1,309 @@ +import json +import sys + +from hermes_cli.session_export import export_record_count, render_sessions_export + + +def _sample_session(): + return { + "id": "sess-123", + "source": "cli", + "model": "test/model", + "title": "Debug auth flow", + "started_at": 1700000000, + "message_count": 5, + "messages": [ + { + "id": 1, + "role": "system", + "content": "hidden system context", + "timestamp": 1700000000, + }, + { + "id": 2, + "role": "user", + "content": "Why is login broken?", + "timestamp": 1700000001, + "platform_message_id": "evt-2", + }, + { + "id": 3, + "role": "assistant", + "content": "I will inspect the auth middleware.", + "timestamp": 1700000002, + }, + { + "id": 4, + "role": "tool", + "tool_name": "read_file", + "content": "def redirect_after_login(): pass", + "timestamp": 1700000003, + }, + { + "id": 5, + "role": "user", + "content": [{"type": "text", "text": "Only show me the prompts."}], + "timestamp": 1700000004, + }, + ], + } + + +def test_default_jsonl_preserves_full_session_shape(): + session = _sample_session() + + rendered = render_sessions_export([session]) + + assert [json.loads(line) for line in rendered.splitlines()] == [session] + + +def test_prompt_only_jsonl_emits_one_record_per_user_prompt(): + rendered = render_sessions_export( + [_sample_session()], + only="user-prompts", + ) + + records = [json.loads(line) for line in rendered.splitlines()] + + assert records == [ + { + "session_id": "sess-123", + "index": 1, + "created_at": "2023-11-14T22:13:21Z", + "role": "user", + "text": "Why is login broken?", + "message_id": 2, + "event_id": "evt-2", + }, + { + "session_id": "sess-123", + "index": 2, + "created_at": "2023-11-14T22:13:24Z", + "role": "user", + "text": "Only show me the prompts.", + "message_id": 5, + }, + ] + + +def test_prompt_only_markdown_excludes_assistant_tool_and_system_content(): + rendered = render_sessions_export( + [_sample_session()], + fmt="markdown", + only="user-prompts", + ) + + assert "# User prompts for session sess-123" in rendered + assert "## 1. 2023-11-14T22:13:21Z" in rendered + assert "Why is login broken?" in rendered + assert "Only show me the prompts." in rendered + assert "I will inspect the auth middleware." not in rendered + assert "def redirect_after_login" not in rendered + assert "hidden system context" not in rendered + + +def test_full_markdown_renderer_collapses_tool_output_and_filters_system(): + rendered = render_sessions_export([_sample_session()], fmt="markdown") + + assert "# Session: Debug auth flow" in rendered + assert "## User - 2023-11-14T22:13:21Z" in rendered + assert "## Assistant - 2023-11-14T22:13:22Z" in rendered + assert "
read_file" in rendered + assert "```text\ndef redirect_after_login(): pass\n```" in rendered + assert "hidden system context" not in rendered + + +def test_export_record_count_switches_unit_for_prompt_only_exports(): + assert export_record_count([_sample_session()]) == (1, "session") + assert export_record_count([_sample_session()], only="user-prompts") == ( + 2, + "prompt", + ) + + +def test_sessions_export_cli_prompt_only_stdout(monkeypatch, capsys): + import hermes_cli.main as main_mod + import hermes_state + + captured = {} + + class FakeDB: + def resolve_session_id(self, session_id): + captured["resolved_from"] = session_id + return "sess-123" + + def export_session(self, session_id): + captured["exported"] = session_id + return _sample_session() + + def close(self): + captured["closed"] = True + + monkeypatch.setattr(hermes_state, "SessionDB", lambda: FakeDB()) + monkeypatch.setattr( + sys, + "argv", + ["hermes", "sessions", "export", "-", "--session-id", "sess", "--only", "user-prompts"], + ) + + main_mod.main() + + output = capsys.readouterr().out + records = [json.loads(line) for line in output.splitlines()] + assert [record["text"] for record in records] == [ + "Why is login broken?", + "Only show me the prompts.", + ] + assert captured == { + "resolved_from": "sess", + "exported": "sess-123", + "closed": True, + } + + +def test_sessions_export_cli_prompt_only_markdown_file(monkeypatch, capsys, tmp_path): + import hermes_cli.main as main_mod + import hermes_state + + class FakeDB: + def resolve_session_id(self, _session_id): + return "sess-123" + + def export_session(self, _session_id): + return _sample_session() + + def close(self): + pass + + output_path = tmp_path / "prompts.md" + monkeypatch.setattr(hermes_state, "SessionDB", lambda: FakeDB()) + monkeypatch.setattr( + sys, + "argv", + [ + "hermes", + "sessions", + "export", + str(output_path), + "--session-id", + "sess", + "--format", + "markdown", + "--only", + "user-prompts", + ], + ) + + main_mod.main() + + assert f"Exported 2 prompts to {output_path}" in capsys.readouterr().out + content = output_path.read_text(encoding="utf-8") + assert "# User prompts for session sess-123" in content + assert "Why is login broken?" in content + assert "I will inspect the auth middleware." not in content + + +def test_sessions_export_rejects_full_session_markdown(monkeypatch, capsys): + import hermes_cli.main as main_mod + import hermes_state + + class FakeDB: + def export_all(self, source=None): + return [_sample_session()] + + def close(self): + pass + + monkeypatch.setattr(hermes_state, "SessionDB", lambda: FakeDB()) + monkeypatch.setattr( + sys, + "argv", + ["hermes", "sessions", "export", "-", "--format", "markdown"], + ) + + main_mod.main() + + assert ( + "Error: --format markdown currently requires --only user-prompts." + in capsys.readouterr().out + ) + + +def test_sessions_export_prompts_cli_stdout(monkeypatch, capsys): + import hermes_cli.main as main_mod + import hermes_state + + captured = {} + + class FakeDB: + def resolve_session_id(self, session_id): + captured["resolved_from"] = session_id + return "sess-123" + + def export_session(self, session_id): + captured["exported"] = session_id + return _sample_session() + + def close(self): + captured["closed"] = True + + monkeypatch.setattr(hermes_state, "SessionDB", lambda: FakeDB()) + monkeypatch.setattr( + sys, + "argv", + ["hermes", "sessions", "export-prompts", "sess"], + ) + + main_mod.main() + + output = capsys.readouterr().out + records = [json.loads(line) for line in output.splitlines()] + assert [record["text"] for record in records] == [ + "Why is login broken?", + "Only show me the prompts.", + ] + assert captured == { + "resolved_from": "sess", + "exported": "sess-123", + "closed": True, + } + + +def test_sessions_export_prompts_cli_markdown_file(monkeypatch, capsys, tmp_path): + import hermes_cli.main as main_mod + import hermes_state + + class FakeDB: + def resolve_session_id(self, _session_id): + return "sess-123" + + def export_session(self, _session_id): + return _sample_session() + + def close(self): + pass + + output_path = tmp_path / "prompts.md" + monkeypatch.setattr(hermes_state, "SessionDB", lambda: FakeDB()) + monkeypatch.setattr( + sys, + "argv", + [ + "hermes", + "sessions", + "export-prompts", + "sess", + "--format", + "md", + "--output", + str(output_path), + ], + ) + + main_mod.main() + + assert f"Exported 2 prompts to {output_path}" in capsys.readouterr().out + content = output_path.read_text(encoding="utf-8") + assert "# User prompts for session sess-123" in content + assert "Why is login broken?" in content + assert "I will inspect the auth middleware." not in content