mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-08 13:12:08 +00:00
feat: add prompt-only session export
This commit is contained in:
Xue Li
Teknium
parent
9a322726ae
commit
b598f8e69b
2 changed files with 626 additions and 0 deletions
317
hermes_cli/session_export.py
Normal file
317
hermes_cli/session_export.py
Normal file
|
|
@ -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"<details><summary>{html_escape(tool_name)}</summary>")
|
||||
lines.append("")
|
||||
lines.append(_fenced_text(_message_text(message.get("content"))))
|
||||
lines.append("")
|
||||
lines.append("</details>")
|
||||
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"
|
||||
309
tests/hermes_cli/test_session_export.py
Normal file
309
tests/hermes_cli/test_session_export.py
Normal file
|
|
@ -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 "<details><summary>read_file</summary>" 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
|
||||
Loading…
Add table
Add a link
Reference in a new issue