From 98eff8229033c79db7e0ce3ea4ff8e1d77339a09 Mon Sep 17 00:00:00 2001 From: "ARC (arc-butler)" Date: Fri, 3 Jul 2026 12:16:49 +0000 Subject: [PATCH] feat(cli): add prompt-only session export with JSONL and Markdown support - Add hermes_cli/session_export.py with filter_user_prompts, render_prompt_only_jsonl, render_prompt_only_md, and render_full_session_md - Add --only user-prompts and --format jsonl|markdown flags to sessions export - Keep default behavior unchanged (full-session JSONL) - 18 unit tests for all render paths --- hermes_cli/main.py | 106 ++++++++-- hermes_cli/session_export.py | 140 +++++++++++++ tests/hermes_cli/test_session_export.py | 248 ++++++++++++++++++++++++ 3 files changed, 480 insertions(+), 14 deletions(-) create mode 100644 hermes_cli/session_export.py create mode 100644 tests/hermes_cli/test_session_export.py diff --git a/hermes_cli/main.py b/hermes_cli/main.py index 4400ee9a261d..73355a81a8ac 100644 --- a/hermes_cli/main.py +++ b/hermes_cli/main.py @@ -13374,13 +13374,27 @@ def cmd_computer_use(args): ) sessions_export = sessions_subparsers.add_parser( - "export", help="Export sessions to a JSONL file" + "export", help="Export sessions to JSONL (default) or Markdown" ) sessions_export.add_argument( - "output", help="Output JSONL file path (use - for stdout)" + "output", help="Output file path (use - for stdout)" ) sessions_export.add_argument("--source", help="Filter by source") sessions_export.add_argument("--session-id", help="Export a specific session") + sessions_export.add_argument( + "--only", + choices=["user-prompts"], + help=( + "Only export user-authored prompts. " + "Excludes assistant messages, tool calls, and system messages." + ), + ) + sessions_export.add_argument( + "--format", + choices=["jsonl", "markdown"], + default="jsonl", + help="Output format (default: jsonl)", + ) sessions_delete = sessions_subparsers.add_parser( "delete", help="Delete a specific session" @@ -13555,25 +13569,89 @@ def cmd_sessions(args): if not data: print(f"Session '{args.session_id}' not found.") return - line = _json.dumps(data, ensure_ascii=False) + "\n" - if args.output == "-": - sys.stdout.write(line) + if getattr(args, "only", None) == "user-prompts": + from hermes_cli.session_export import ( + render_prompt_only_jsonl, + render_prompt_only_md, + ) + + if args.format == "markdown": + output = render_prompt_only_md(data) + else: + output = render_prompt_only_jsonl(data) + if args.output == "-": + + sys.stdout.write(output) + else: + with open(args.output, "w", encoding="utf-8") as f: + f.write(output) + print(f"Exported 1 session to {args.output}") + elif args.format == "markdown": + from hermes_cli.session_export import render_full_session_md + + output = render_full_session_md(data) + if args.output == "-": + + sys.stdout.write(output) + else: + with open(args.output, "w", encoding="utf-8") as f: + f.write(output) + print(f"Exported 1 session to {args.output}") else: - with open(args.output, "w", encoding="utf-8") as f: - f.write(line) - print(f"Exported 1 session to {args.output}") + line = _json.dumps(data, ensure_ascii=False) + "\n" + if args.output == "-": + + sys.stdout.write(line) + else: + with open(args.output, "w", encoding="utf-8") as f: + f.write(line) + print(f"Exported 1 session to {args.output}") else: - sessions = db.export_all(source=args.source) - if args.output == "-": + only = getattr(args, "only", None) + fmt = args.format + if only == "user-prompts" or fmt == "markdown": + from hermes_cli.session_export import ( + render_prompt_only_jsonl, + render_prompt_only_md, + render_full_session_md, + ) + sessions = db.export_all(source=args.source) + output_lines: list[str] = [] for s in sessions: - sys.stdout.write(_json.dumps(s, ensure_ascii=False) + "\n") + if only == "user-prompts": + if fmt == "markdown": + output_lines.append(render_prompt_only_md(s)) + output_lines.append("") + else: + output_lines.append( + render_prompt_only_jsonl(s).rstrip("\n") + ) + else: + # --format markdown without --only + output_lines.append(render_full_session_md(s)) + output_lines.append("") + output = "\n".join(output_lines) + "\n" + if args.output == "-": + + sys.stdout.write(output) + else: + with open(args.output, "w", encoding="utf-8") as f: + f.write(output) + plural = "sessions" if len(sessions) != 1 else "session" + print(f"Exported {len(sessions)} {plural} to {args.output}") else: - with open(args.output, "w", encoding="utf-8") as f: + sessions = db.export_all(source=args.source) + if args.output == "-": + for s in sessions: - f.write(_json.dumps(s, ensure_ascii=False) + "\n") - print(f"Exported {len(sessions)} sessions to {args.output}") + sys.stdout.write(_json.dumps(s, ensure_ascii=False) + "\n") + else: + with open(args.output, "w", encoding="utf-8") as f: + for s in sessions: + f.write(_json.dumps(s, ensure_ascii=False) + "\n") + print(f"Exported {len(sessions)} sessions to {args.output}") elif action == "delete": resolved_session_id = db.resolve_session_id(args.session_id) diff --git a/hermes_cli/session_export.py b/hermes_cli/session_export.py new file mode 100644 index 000000000000..e93db9f2c0ab --- /dev/null +++ b/hermes_cli/session_export.py @@ -0,0 +1,140 @@ +"""Prompt-only and Markdown session export renderers. + +``filter_user_prompts`` extracts only user-authored messages from a +session dict. ``render_prompt_only_jsonl`` and ``render_prompt_only_md`` +serialise those prompts as JSONL or Markdown respectively. +""" + +from __future__ import annotations + +import json +from datetime import datetime, timezone +from typing import Any + + +def filter_user_prompts( + session_data: dict[str, Any], +) -> list[dict[str, Any]]: + """Return only user-authored messages from *session_data*. + + Each returned message dict includes the original row columns from + the ``messages`` table (*id*, *session_id*, *role*, *content*, + *timestamp*, *platform_message_id*, …). + """ + messages: list[dict[str, Any]] = session_data.get("messages", []) + return [m for m in messages if m.get("role") == "user"] + + +def _ts_to_iso(timestamp: float) -> str: + """Convert a Unix-epoch *timestamp* to ISO-8601 with Z suffix.""" + try: + return ( + datetime.fromtimestamp(timestamp, tz=timezone.utc) + .isoformat(timespec="seconds") + .replace("+00:00", "Z") + ) + except (OSError, ValueError, OverflowError): + return "" + + +def render_prompt_only_jsonl( + session_data: dict[str, Any], +) -> str: + """Render user prompts from *session_data* as newline-delimited JSON. + + Each line is a JSON object with:: + + { + "session_id": "", + "prompt_index": <1-based-int>, + "timestamp": "", + "prompt": "", + "message_id": , + "platform_message_id": "", + "event_id": "" + } + """ + prompts = filter_user_prompts(session_data) + lines: list[str] = [] + for idx, msg in enumerate(prompts, start=1): + record = { + "session_id": session_data.get("id"), + "prompt_index": idx, + "timestamp": _ts_to_iso(msg.get("timestamp", 0)), + "prompt": msg.get("content") or "", + "message_id": msg.get("id"), + "platform_message_id": msg.get("platform_message_id"), + "event_id": None, # reserved for future gateway event linkage + } + lines.append(json.dumps(record, ensure_ascii=False)) + return "\n".join(lines) + ("\n" if lines else "") + + +def render_prompt_only_md( + session_data: dict[str, Any], +) -> str: + """Render user prompts from *session_data* as Markdown. + + Output format:: + + # User prompts for session + + ## 1. + + + + ## 2. + + … + """ + prompts = filter_user_prompts(session_data) + sid = session_data.get("id", "unknown") + lines: list[str] = [f"# User prompts for session {sid}", ""] + for idx, msg in enumerate(prompts, start=1): + ts = _ts_to_iso(msg.get("timestamp", 0)) + lines.append(f"## {idx}. {ts}") + lines.append("") + lines.append((msg.get("content") or "").strip()) + lines.append("") + return "\n".join(lines) + + +def render_full_session_md( + session_data: dict[str, Any], +) -> str: + """Render a full session as Markdown (all messages in order). + + Suitable as a shared full-session Markdown renderer that other + export surfaces can reuse in the future. + + Output:: + + # Session: + + **User** + + + + **Assistant** + + + + … + """ + messages: list[dict[str, Any]] = session_data.get("messages", []) + sid = session_data.get("id", "unknown") + lines: list[str] = [f"# Session: {sid}", ""] + for msg in messages: + role = msg.get("role", "unknown").capitalize() + ts = _ts_to_iso(msg.get("timestamp", 0)) + lines.append(f"**{role}** {ts}") + lines.append("") + content = (msg.get("content") or "").strip() + if content: + lines.append(content) + tool_calls = msg.get("tool_calls") + if tool_calls: + lines.append("") + lines.append(f" *tool_calls:* `{json.dumps(tool_calls, ensure_ascii=False)}`") + lines.append("") + return "\n".join(lines) diff --git a/tests/hermes_cli/test_session_export.py b/tests/hermes_cli/test_session_export.py new file mode 100644 index 000000000000..4d80f84199ce --- /dev/null +++ b/tests/hermes_cli/test_session_export.py @@ -0,0 +1,248 @@ +"""Tests for ``hermes_cli.session_export`` — prompt-only and Markdown renderers.""" + +from __future__ import annotations + +import json +from datetime import datetime, timezone + +from hermes_cli.session_export import ( + filter_user_prompts, + render_full_session_md, + render_prompt_only_jsonl, + render_prompt_only_md, +) + +_FAKE_NOW = 1800000000.0 # Arbitrary stable timestamp for deterministic tests. + + +def _make_session( + session_id: str = "test-session-1", + messages: list[dict] | None = None, +) -> dict: + return { + "id": session_id, + "title": "Test session", + "source": "cli", + "created_at": _FAKE_NOW, + "updated_at": _FAKE_NOW, + "messages": messages or [], + } + + +def _user_msg( + content: str, + ts: float = _FAKE_NOW, + msg_id: int = 1, + platform_id: str | None = "pl-1", +) -> dict: + return { + "id": msg_id, + "role": "user", + "content": content, + "timestamp": ts, + "platform_message_id": platform_id, + } + + +def _assistant_msg( + content: str, + ts: float = _FAKE_NOW, + msg_id: int = 999, +) -> dict: + return { + "id": msg_id, + "role": "assistant", + "content": content, + "timestamp": ts, + } + + +def _tool_msg( + content: str, + ts: float = _FAKE_NOW, + msg_id: int = 888, +) -> dict: + return { + "id": msg_id, + "role": "tool", + "content": content, + "timestamp": ts, + } + + +def _system_msg( + content: str, + ts: float = _FAKE_NOW, + msg_id: int = 777, +) -> dict: + return { + "id": msg_id, + "role": "system", + "content": content, + "timestamp": ts, + } + + +# ── filter_user_prompts ────────────────────────────────────────────── + + +class TestFilterUserPrompts: + def test_empty_session(self): + assert filter_user_prompts(_make_session()) == [] + + def test_only_user_messages(self): + msgs = [_user_msg("hello"), _user_msg("world")] + result = filter_user_prompts(_make_session(messages=msgs)) + assert len(result) == 2 + assert all(m["role"] == "user" for m in result) + + def test_filters_out_non_user_roles(self): + msgs = [ + _user_msg("hi"), + _assistant_msg("response"), + _user_msg("follow-up"), + _tool_msg('{"result": "ok"}'), + _system_msg("system prompt"), + ] + result = filter_user_prompts(_make_session(messages=msgs)) + assert len(result) == 2 + assert [m["content"] for m in result] == ["hi", "follow-up"] + + def test_no_messages_key(self): + session = {"id": "s1"} + assert filter_user_prompts(session) == [] + + +# ── render_prompt_only_jsonl ────────────────────────────────────────── + + +class TestRenderPromptOnlyJsonl: + def test_empty_session(self): + result = render_prompt_only_jsonl(_make_session()) + assert result == "" + + def test_single_prompt(self): + msgs = [_user_msg("What is AI?", msg_id=1)] + result = render_prompt_only_jsonl(_make_session(messages=msgs)) + records = [json.loads(line) for line in result.strip().split("\n")] + assert len(records) == 1 + rec = records[0] + assert rec["session_id"] == "test-session-1" + assert rec["prompt_index"] == 1 + assert rec["prompt"] == "What is AI?" + assert rec["message_id"] == 1 + assert rec["platform_message_id"] == "pl-1" + + def test_multiple_prompts_sequential_index(self): + msgs = [ + _user_msg("first", msg_id=1), + _assistant_msg("ok"), + _user_msg("second", msg_id=2), + _user_msg("third", msg_id=3), + ] + result = render_prompt_only_jsonl(_make_session(messages=msgs)) + records = [json.loads(line) for line in result.strip().split("\n")] + assert len(records) == 3 + assert [r["prompt_index"] for r in records] == [1, 2, 3] + assert [r["prompt"] for r in records] == ["first", "second", "third"] + + def test_timestamp_iso_format(self): + ts = 1800000000.0 + expected_iso = ( + datetime.fromtimestamp(ts, tz=timezone.utc) + .isoformat(timespec="seconds") + .replace("+00:00", "Z") + ) + msgs = [_user_msg("prompt", ts=ts)] + result = render_prompt_only_jsonl(_make_session(messages=msgs)) + rec = json.loads(result.strip()) + assert rec["timestamp"] == expected_iso + + def test_jsonl_trailing_newline(self): + msgs = [_user_msg("a"), _user_msg("b")] + result = render_prompt_only_jsonl(_make_session(messages=msgs)) + assert result.endswith("\n") + assert result.count("\n") == 2 # two lines + trailing + + def test_null_fields_for_missing_data(self): + msgs = [{"id": None, "role": "user", "content": "test", "timestamp": 0}] + result = render_prompt_only_jsonl(_make_session(messages=msgs)) + rec = json.loads(result.strip()) + assert rec["message_id"] is None + assert rec["platform_message_id"] is None + + +# ── render_prompt_only_md ────────────────────────────────────────── + + +class TestRenderPromptOnlyMd: + def test_empty_session(self): + result = render_prompt_only_md(_make_session()) + assert result.startswith("# User prompts for session test-session-1") + + def test_single_prompt(self): + msgs = [_user_msg("Hello world", msg_id=1)] + result = render_prompt_only_md(_make_session(messages=msgs)) + assert "## 1." in result + assert "Hello world" in result + + def test_multiple_prompts(self): + msgs = [_user_msg("first", msg_id=1), _user_msg("second", msg_id=2)] + result = render_prompt_only_md(_make_session(messages=msgs)) + assert "## 1." in result + assert "## 2." in result + assert "first" in result + assert "second" in result + + def test_excludes_non_user_messages(self): + msgs = [ + _user_msg("user text"), + _assistant_msg("assistant text"), + _tool_msg("tool output"), + ] + result = render_prompt_only_md(_make_session(messages=msgs)) + assert "user text" in result + assert "assistant text" not in result + assert "tool output" not in result + + +# ── render_full_session_md ────────────────────────────────────────── + + +class TestRenderFullSessionMd: + def test_empty_session(self): + result = render_full_session_md(_make_session()) + assert result.startswith("# Session: test-session-1") + + def test_shows_all_roles(self): + msgs = [ + _user_msg("hello"), + _assistant_msg("hi there"), + _tool_msg("result: ok"), + ] + result = render_full_session_md(_make_session(messages=msgs)) + assert "hello" in result + assert "hi there" in result + assert "result: ok" in result + assert "**User**" in result + assert "**Assistant**" in result + assert "**Tool**" in result + + def test_labels_role_properly(self): + msgs = [_user_msg("test")] + result = render_full_session_md(_make_session(messages=msgs)) + assert "User" in result + + def test_handles_tool_calls_field(self): + msgs = [ + { + "id": 1, + "role": "assistant", + "content": "Let me search", + "timestamp": _FAKE_NOW, + "tool_calls": [{"name": "web_search", "args": {"q": "test"}}], + } + ] + result = render_full_session_md(_make_session(messages=msgs)) + assert "tool_calls" in result + assert "web_search" in result