Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
106 changes: 92 additions & 14 deletions hermes_cli/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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)
Expand Down
140 changes: 140 additions & 0 deletions hermes_cli/session_export.py
Original file line number Diff line number Diff line change
@@ -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": "<session-id>",
"prompt_index": <1-based-int>,
"timestamp": "<ISO-8601>",
"prompt": "<raw-user-text>",
"message_id": <int-or-null>,
"platform_message_id": "<str-or-null>",
"event_id": "<str-or-null>"
}
"""
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 <id>

## 1. <ISO-timestamp>

<prompt-text>

## 2. <ISO-timestamp>

…
"""
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: <id>

**User** <timestamp>

<content>

**Assistant** <timestamp>

<content>

…
"""
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)
Loading