From f202e295d566d14a26fbc2a1e019d95960405bb5 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 20:24:58 -0700 Subject: [PATCH] refactor(hermes_cli): dispatch-table /save renderers, fold md export frontmatter/content helpers --- hermes_cli/session_export.py | 113 +++++++++--------- hermes_cli/session_export_md.py | 198 +++++++++++++------------------- 2 files changed, 135 insertions(+), 176 deletions(-) diff --git a/hermes_cli/session_export.py b/hermes_cli/session_export.py index 51548ce6f0..d44bab6324 100644 --- a/hermes_cli/session_export.py +++ b/hermes_cli/session_export.py @@ -1,8 +1,8 @@ """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. +CLI, dashboard, and slash-command surfaces all deal with the same session-shaped data (a session +dict with a ``messages`` list); filtering and human-readable rendering live here so each surface +only loads sessions and writes bytes. """ from __future__ import annotations @@ -16,10 +16,13 @@ from typing import Any, Dict, Iterable, Iterator, List, Literal, Optional, Tuple ExportFormat = Literal["jsonl", "markdown"] ExportOnly = Literal["user-prompts"] +_EXPORT_FORMATS = {"jsonl": "jsonl", "markdown": "markdown", "md": "markdown"} +_ONLY_ALIASES = {"user", "prompts", "user-prompts", "user_prompts"} + def normalize_export_format(fmt: str) -> ExportFormat: """Return the canonical export format name.""" - value = {"jsonl": "jsonl", "markdown": "markdown", "md": "markdown"}.get((fmt or "jsonl").strip().lower()) + value = _EXPORT_FORMATS.get((fmt or "jsonl").strip().lower()) if value is None: raise ValueError(f"Unsupported session export format: {fmt}") return value # type: ignore[return-value] @@ -29,7 +32,7 @@ def normalize_export_only(only: Optional[str]) -> Optional[ExportOnly]: """Return the canonical export filter name.""" if only is None: return None - if only.strip().lower() in {"user", "prompts", "user-prompts", "user_prompts"}: + if only.strip().lower() in _ONLY_ALIASES: return "user-prompts" raise ValueError(f"Unsupported session export filter: {only}") @@ -42,22 +45,33 @@ def render_sessions_export( ) -> 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. + ``fmt=jsonl`` with no filter keeps the legacy shape (one full session object per line); + ``only=user-prompts`` switches the unit to one prompt record per line for piping into + review / memory-ingestion / prompt-library tooling. """ session_list = list(sessions) export_format = normalize_export_format(fmt) - export_only = normalize_export_only(only) - - prompts_only = export_only == "user-prompts" + prompts_only = normalize_export_only(only) == "user-prompts" if export_format == "jsonl": rows = iter_user_prompt_records(session_list) if prompts_only else session_list lines = [json.dumps(row, ensure_ascii=False) for row in rows] return ("\n".join(lines) + "\n") if lines else "" if prompts_only: - return _render_user_prompts_markdown(session_list) - return _render_full_markdown(session_list) + lines = _render_sessions_markdown( + session_list, "User prompts export", + lambda session: f"User prompts for session {_heading_text(_session_id(session))}", + lambda session: f"Session {_heading_text(_session_id(session))}", + _append_prompt_records, + ) + if not session_list: + lines += ["_No user prompts found._", ""] + return _finish_markdown(lines) + return _finish_markdown(_render_sessions_markdown( + session_list, "Hermes sessions export", + lambda session: f"Session: {_heading_text(_session_title_or_id(session))}", + lambda session: f"Session: {_heading_text(_session_title_or_id(session))}", + _append_session_messages, + )) def export_record_count( @@ -88,27 +102,13 @@ def iter_user_prompt_records( "role": "user", "text": _message_text(message.get("content")), } - message_id = message.get("id") - if message_id is not None: + if (message_id := message.get("id")) is not None: record["message_id"] = message_id - event_id = message.get("platform_message_id") or message.get("event_id") - if event_id: + if event_id := message.get("platform_message_id") or message.get("event_id"): record["event_id"] = event_id yield record -def _render_user_prompts_markdown(sessions: List[Dict[str, Any]]) -> str: - lines = _render_sessions_markdown( - sessions, "User prompts export", - lambda session: f"User prompts for session {_heading_text(_session_id(session))}", - lambda session: f"Session {_heading_text(_session_id(session))}", - _append_prompt_records, - ) - if not sessions: - lines += ["_No user prompts found._", ""] - return _finish_markdown(lines) - - def _render_sessions_markdown(sessions, multi_title, single_heading, multi_heading, append_body) -> List[str]: """One session → its own H1 with body at H2; several → a shared H1, each session H2/H3.""" lines: List[str] = [] @@ -140,15 +140,6 @@ def _append_prompt_records( lines += [str(prompt.get("text") or ""), ""] -def _render_full_markdown(sessions: List[Dict[str, Any]]) -> str: - return _finish_markdown(_render_sessions_markdown( - sessions, "Hermes sessions export", - lambda session: f"Session: {_heading_text(_session_title_or_id(session))}", - lambda session: f"Session: {_heading_text(_session_title_or_id(session))}", - _append_session_messages, - )) - - def _append_session_messages( lines: List[str], session: Dict[str, Any], *, heading_level: int ) -> None: @@ -178,8 +169,7 @@ def _append_session_messages( 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)] + return [message for message in session.get("messages") or [] if isinstance(message, dict)] def _message_text(content: Any) -> str: @@ -220,7 +210,7 @@ def _session_metadata_lines(session: Dict[str, Any]) -> List[str]: if session.get(key): lines.append(f"- {label}: `{session[key]}`") if title := session.get("title"): - lines.append(f"- Title: {_inline_text(str(title))}") + lines.append(f"- Title: {' '.join(str(title).splitlines()).strip()}") if started := _format_timestamp(session.get("started_at")): lines.append(f"- Started: {started}") if (message_count := session.get("message_count")) is not None: @@ -240,10 +230,6 @@ 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: @@ -257,9 +243,7 @@ def _finish_markdown(lines: List[str]) -> str: return "\n".join(lines) + "\n" -# --------------------------------------------------------------------------- -# Current-session save helper (shared by CLI /save and gateway /save) -# --------------------------------------------------------------------------- +# --- Current-session save helper (shared by CLI /save and gateway /save) --- SAVE_FORMATS = ("json", "md", "html") @@ -282,29 +266,38 @@ Examples: /save md notes.md /save html session.html redact""" +_SAVE_FORMAT_ALIASES = {"json": "json", "snapshot": "json", "md": "md", "markdown": "md", "html": "html"} + def normalize_save_format(fmt: Optional[str]) -> str: """Map a user-typed /save format token to a canonical format.""" token = (fmt or "json").strip().lower() - canonical = {"json": "json", "snapshot": "json", "md": "md", "markdown": "md", "html": "html"} - if token not in canonical: + if token not in _SAVE_FORMAT_ALIASES: raise ValueError( f"Unknown format {token!r} — expected one of: json, md, html" ) - return canonical[token] + return _SAVE_FORMAT_ALIASES[token] + + +def _render_html_for_save(session: Dict[str, Any]) -> str: + from hermes_cli.session_export_html import generate_html_export + + return generate_html_export(session) + + +_SAVE_RENDERERS = { + "json": lambda session: json.dumps(session, indent=2, ensure_ascii=False, default=str), + "md": lambda session: render_sessions_export([session], fmt="markdown"), + "html": _render_html_for_save, +} def render_session_for_save(session: Dict[str, Any], fmt: str) -> str: """Render one exported session dict for /save.""" - if fmt == "json": - return json.dumps(session, indent=2, ensure_ascii=False, default=str) - if fmt == "md": - return render_sessions_export([session], fmt="markdown") - if fmt == "html": - from hermes_cli.session_export_html import generate_html_export - - return generate_html_export(session) - raise ValueError(f"Unknown save format: {fmt!r}") + renderer = _SAVE_RENDERERS.get(fmt) + if renderer is None: + raise ValueError(f"Unknown save format: {fmt!r}") + return renderer(session) def default_save_filename(session_id: str, fmt: str) -> str: diff --git a/hermes_cli/session_export_md.py b/hermes_cli/session_export_md.py index a06ccabd13..e5c2953455 100644 --- a/hermes_cli/session_export_md.py +++ b/hermes_cli/session_export_md.py @@ -1,8 +1,7 @@ """Markdown/QMD export helpers for Hermes sessions. -This module is intentionally filesystem-only: it formats already-exported SessionDB dictionaries and -writes them to user-selected export directories. It must not mutate state.db or call -delete/prune/archive APIs. +Filesystem-only: formats already-exported SessionDB dicts and writes them to user-selected export +directories. Must not mutate state.db or call delete/prune/archive APIs. """ from __future__ import annotations @@ -17,6 +16,8 @@ from typing import Any EXPORTER_VERSION = "hermes sessions export (md/qmd) v1" _SHA_LINE_RE = re.compile(r"- SHA256 of exported body: `([0-9a-f]{64})`") +_SHA_PLACEHOLDER = "__SHA256_PLACEHOLDER__" +_VERIFICATION_HEADING = "## Export verification" def _iso_timestamp(value: Any) -> str: @@ -29,46 +30,30 @@ def _iso_timestamp(value: Any) -> str: return datetime.fromtimestamp(ts, tz=timezone.utc).isoformat().replace("+00:00", "Z") -def _frontmatter_value(value: Any) -> str: - if value is None: - return "null" - if isinstance(value, bool): - return "true" if value else "false" - if isinstance(value, (int, float)) and not isinstance(value, bool): - return json.dumps(value, ensure_ascii=False) - if isinstance(value, list): - return json.dumps(value, ensure_ascii=False) - return json.dumps(str(value), ensure_ascii=False) - - def _frontmatter_line(key: str, value: Any) -> str: - return f"{key}: {_frontmatter_value(value)}" + if value is None: + shown = "null" + elif isinstance(value, bool): + shown = "true" if value else "false" + elif isinstance(value, (int, float, list)): + shown = json.dumps(value, ensure_ascii=False) + else: + shown = json.dumps(str(value), ensure_ascii=False) + return f"{key}: {shown}" + + +def _json_block(value: Any) -> str: + return "```json\n" + json.dumps(value, ensure_ascii=False, indent=2) + "\n```" def _message_heading(message: dict[str, Any]) -> str: role = str(message.get("role") or "message") - label = role.capitalize() name = message.get("name") or message.get("tool_name") - if role == "tool" and name: - label = f"Tool — {name}" + label = f"Tool — {name}" if role == "tool" and name else role.capitalize() timestamp = _iso_timestamp(message.get("created_at") or message.get("timestamp")) return f"### {label}{' — ' + timestamp if timestamp else ''}" -def _render_content(content: Any) -> str: - if content is None: - return "" - if isinstance(content, str): - return content.rstrip() - return "```json\n" + json.dumps(content, ensure_ascii=False, indent=2) + "\n```" - - -def _render_tool_calls(tool_calls: Any) -> str: - if not tool_calls: - return "" - return "\n\n## Tool calls\n\n```json\n" + json.dumps(tool_calls, ensure_ascii=False, indent=2) + "\n```" - - def _session_id(session: dict[str, Any]) -> str: return str(session.get("id") or session.get("session_id") or "unknown-session") @@ -87,8 +72,7 @@ def _message_count(session: dict[str, Any]) -> int: def _render_messages(session: dict[str, Any]) -> str: parts: list[str] = ["## Messages\n"] segments = _segments(session) - total_messages = _message_count(session) - if total_messages == 0: + if _message_count(session) == 0: parts.append("_No messages in this session._\n") return "\n".join(parts).rstrip() + "\n" @@ -98,99 +82,82 @@ def _render_messages(session: dict[str, Any]) -> str: parts.append(f"## Compression segment: {_session_id(segment)}\n") for message in list(segment.get("messages") or []): parts.append(_message_heading(message) + "\n") - rendered_content = _render_content(message.get("content")) - if rendered_content: - parts.append(rendered_content + "\n") - tool_calls = _render_tool_calls(message.get("tool_calls")) - if tool_calls: - parts.append(tool_calls + "\n") + content = message.get("content") + if content is not None: + rendered = content.rstrip() if isinstance(content, str) else _json_block(content) + if rendered: + parts.append(rendered + "\n") + if tool_calls := message.get("tool_calls"): + parts.append("\n\n## Tool calls\n\n" + _json_block(tool_calls) + "\n") parts.append("") return "\n".join(parts).rstrip() + "\n" def _export_body_without_hash(session: dict[str, Any], *, fmt: str, exported_at: float) -> str: session_id = _session_id(session) - title = session.get("title") or session_id - provider = session.get("billing_provider") or session.get("provider") - started_at = _iso_timestamp(session.get("started_at") or session.get("created_at")) - last_active = _iso_timestamp(session.get("last_active") or session.get("updated_at")) - ended_at = _iso_timestamp(session.get("ended_at")) exported_iso = _iso_timestamp(exported_at) message_count = _message_count(session) - frontmatter = [ - "---", - _frontmatter_line("session_id", session_id), - _frontmatter_line("title", session.get("title")), - _frontmatter_line("source", session.get("source")), - _frontmatter_line("created_at", started_at), - _frontmatter_line("updated_at", last_active), - _frontmatter_line("ended_at", ended_at), - _frontmatter_line("model", session.get("model")), - _frontmatter_line("provider", provider), - _frontmatter_line("cwd", session.get("cwd")), - _frontmatter_line("archived", bool(session.get("archived"))), - _frontmatter_line("message_count", message_count), - _frontmatter_line("tool_call_count", session.get("tool_call_count") or 0), + fields = [ + ("session_id", session_id), + ("title", session.get("title")), + ("source", session.get("source")), + ("created_at", _iso_timestamp(session.get("started_at") or session.get("created_at"))), + ("updated_at", _iso_timestamp(session.get("last_active") or session.get("updated_at"))), + ("ended_at", _iso_timestamp(session.get("ended_at"))), + ("model", session.get("model")), + ("provider", session.get("billing_provider") or session.get("provider")), + ("cwd", session.get("cwd")), + ("archived", bool(session.get("archived"))), + ("message_count", message_count), + ("tool_call_count", session.get("tool_call_count") or 0), ] if session.get("lineage_session_ids"): - frontmatter.append(_frontmatter_line("lineage_session_ids", session.get("lineage_session_ids"))) - frontmatter.extend([ - _frontmatter_line("format", fmt), - _frontmatter_line("exported_at", exported_iso), - _frontmatter_line("exporter", EXPORTER_VERSION), - "---", - "", - ]) + fields.append(("lineage_session_ids", session.get("lineage_session_ids"))) + fields += [("format", fmt), ("exported_at", exported_iso), ("exporter", EXPORTER_VERSION)] + frontmatter = ["---", *(_frontmatter_line(k, v) for k, v in fields), "---", ""] - parts = ["\n".join(frontmatter), f"# {title}\n"] - parts.append(f"Session ID: `{session_id}`\n") + parts = ["\n".join(frontmatter), f"# {session.get('title') or session_id}\n", f"Session ID: `{session_id}`\n"] if session.get("source"): parts.append(f"Source: `{session.get('source')}`\n") if session.get("cwd"): parts.append(f"Working directory: `{session.get('cwd')}`\n") - - parts.append(_render_messages(session)) - parts.append("## Export verification\n") - parts.append(f"- Session id: `{session_id}`") - parts.append(f"- Exported messages: `{message_count}`") - parts.append(f"- Source DB message count at export: `{session.get('message_count', message_count)}`") - parts.append(f"- Exported at: `{exported_iso}`") - parts.append("- SHA256 of exported body: `__SHA256_PLACEHOLDER__`") + parts += [ + _render_messages(session), + f"{_VERIFICATION_HEADING}\n", + f"- Session id: `{session_id}`", + f"- Exported messages: `{message_count}`", + f"- Source DB message count at export: `{session.get('message_count', message_count)}`", + f"- Exported at: `{exported_iso}`", + f"- SHA256 of exported body: `{_SHA_PLACEHOLDER}`", + ] return "\n".join(parts).rstrip() + "\n" -def _body_for_digest(text: str) -> str: - return _SHA_LINE_RE.sub("- SHA256 of exported body: `pending`", text) +def _check_fmt(fmt: str) -> None: + if fmt not in {"md", "qmd"}: + raise ValueError("fmt must be 'md' or 'qmd'") def render_session_markdown( session: dict[str, Any], *, fmt: str = "md", include_verification: bool = True ) -> str: """Render a SessionDB export dictionary as Markdown/QMD text.""" - if fmt not in {"md", "qmd"}: - raise ValueError("fmt must be 'md' or 'qmd'") - exported_at = time.time() - body = _export_body_without_hash(session, fmt=fmt, exported_at=exported_at) - digest_body = body.replace("`__SHA256_PLACEHOLDER__`", "`pending`") - digest = hashlib.sha256(digest_body.encode("utf-8")).hexdigest() - if include_verification: - return body.replace("__SHA256_PLACEHOLDER__", digest) - before_verification = body.split("\n## Export verification\n", 1)[0].rstrip() + "\n" - return before_verification + _check_fmt(fmt) + body = _export_body_without_hash(session, fmt=fmt, exported_at=time.time()) + if not include_verification: + return body.split(f"\n{_VERIFICATION_HEADING}\n", 1)[0].rstrip() + "\n" + # The digest covers the body with the SHA line set to `pending`, which is what verify recomputes. + digest_body = body.replace(f"`{_SHA_PLACEHOLDER}`", "`pending`") + return body.replace(_SHA_PLACEHOLDER, hashlib.sha256(digest_body.encode("utf-8")).hexdigest()) def safe_session_filename(session: dict[str, Any], *, fmt: str = "md") -> str: """Return a deterministic, path-safe filename for a session export.""" - if fmt not in {"md", "qmd"}: - raise ValueError("fmt must be 'md' or 'qmd'") - session_id = _session_id(session) + _check_fmt(fmt) title = str(session.get("title") or "session") - slug = re.sub(r"[^A-Za-z0-9._-]+", "-", title).strip(".-_").lower() - if not slug: - slug = "session" - slug = slug[:60] - return f"{session_id}-{slug}.{fmt}" + slug = re.sub(r"[^A-Za-z0-9._-]+", "-", title).strip(".-_").lower() or "session" + return f"{_session_id(session)}-{slug[:60]}.{fmt}" def file_sha256(path: Path | str) -> str: @@ -205,11 +172,10 @@ def verify_export_file(path: Path | str, session: dict[str, Any]) -> tuple[bool, match = _SHA_LINE_RE.search(text) if not match: return False, "sha256 marker missing" - actual = hashlib.sha256(_body_for_digest(text).encode("utf-8")).hexdigest() - if actual != match.group(1): + digest_body = _SHA_LINE_RE.sub("- SHA256 of exported body: `pending`", text) + if hashlib.sha256(digest_body.encode("utf-8")).hexdigest() != match.group(1): return False, "sha256 mismatch" - expected_count = _message_count(session) - if f"- Exported messages: `{expected_count}`" not in text: + if f"- Exported messages: `{_message_count(session)}`" not in text: return False, "message count mismatch" if f"- Session id: `{_session_id(session)}`" not in text: return False, "session id mismatch" @@ -217,11 +183,10 @@ def verify_export_file(path: Path | str, session: dict[str, Any]) -> tuple[bool, def redact_session_data(session: dict[str, Any]) -> dict[str, Any]: - """Return a deep copy of a session export dict with secrets redacted. + """Return a copy of a session export dict with secrets redacted. - Runs every message's content and tool-call arguments through the force-mode redaction pass - (``agent.redact.redact_sensitive_text``), so API keys, tokens, and credentials that appeared in - tool output never land in plaintext export files. + Every message's content and tool-call arguments go through the force-mode redaction pass + (``agent.redact.redact_sensitive_text``) so credentials in tool output never land in exports. """ from agent.redact import redact_sensitive_text @@ -236,18 +201,22 @@ def redact_session_data(session: dict[str, Any]) -> dict[str, Any]: redacted = dict(session) for key in ("messages", "segments"): - if key in redacted and redacted[key] is not None: + if redacted.get(key) is not None: redacted[key] = _clean(redacted[key]) return redacted +def _export_dir(output_dir: Path | str) -> Path: + out_dir = Path(output_dir).expanduser() + out_dir.mkdir(parents=True, exist_ok=True) + return out_dir + + def write_session_markdown( session: dict[str, Any], output_dir: Path | str, *, fmt: str = "md", force: bool = False ) -> Path: """Write a Markdown/QMD export file and return its path.""" - out_dir = Path(output_dir).expanduser() - out_dir.mkdir(parents=True, exist_ok=True) - path = out_dir / safe_session_filename(session, fmt=fmt) + path = _export_dir(output_dir) / safe_session_filename(session, fmt=fmt) if path.exists() and not force: raise FileExistsError(str(path)) path.write_text(render_session_markdown(session, fmt=fmt), encoding="utf-8") @@ -255,19 +224,16 @@ def write_session_markdown( def append_manifest_entry(output_dir: Path | str, session: dict[str, Any], path: Path | str, *, fmt: str) -> Path: - out_dir = Path(output_dir).expanduser() - out_dir.mkdir(parents=True, exist_ok=True) - export_path = Path(path) entry = { "session_id": _session_id(session), "lineage_session_ids": session.get("lineage_session_ids") or [_session_id(session)], - "path": str(export_path), + "path": str(Path(path)), "format": fmt, "message_count": _message_count(session), - "sha256": file_sha256(export_path), + "sha256": file_sha256(path), "exported_at": time.time(), } - manifest = out_dir / "manifest.jsonl" + manifest = _export_dir(output_dir) / "manifest.jsonl" with manifest.open("a", encoding="utf-8") as fh: fh.write(json.dumps(entry, ensure_ascii=False, sort_keys=True) + "\n") return manifest