feat: Add /export command for session export (Markdown/JSON) to CLI and Gateway

This commit is contained in:
ag9920
2026-04-09 10:11:15 +08:00
committed by Teknium
parent 883b57cbd8
commit 23ad35b7fa
7 changed files with 173 additions and 0 deletions
+41
View File
@@ -9342,6 +9342,45 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
return True return True
def _handle_export_command(self, cmd: str):
"""Handle /export command to export session history to file."""
if not self._session_db:
_cprint(" Session database not available.")
return
parts = cmd.split()
format_type = "markdown"
filename = f"session_{self.session_id}.md"
if len(parts) > 1:
arg = parts[1].lower()
if arg in ["json", "md", "markdown"]:
format_type = "json" if arg == "json" else "markdown"
filename = f"session_{self.session_id}.{'json' if format_type == 'json' else 'md'}"
else:
filename = arg
if len(parts) > 2:
filename = parts[2]
export_data = self._session_db.export_session(self.session_id)
if not export_data:
_cprint(f" Failed to export session {self.session_id}.")
return
try:
with open(filename, "w", encoding="utf-8") as f:
if format_type == "json":
import json
json.dump(export_data, f, indent=2, ensure_ascii=False)
else:
from hermes_state import SessionDB
f.write(SessionDB.format_session_as_markdown(export_data))
_cprint(f" Session exported successfully to {filename}")
except Exception as e:
_cprint(f" Error writing export file: {e}")
def save_conversation(self): def save_conversation(self):
"""Save the current conversation to a JSON snapshot under ~/.hermes/sessions/saved/. """Save the current conversation to a JSON snapshot under ~/.hermes/sessions/saved/.
@@ -11129,6 +11168,8 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
self._handle_branch_command(cmd_original) self._handle_branch_command(cmd_original)
elif canonical == "save": elif canonical == "save":
self.save_conversation() self.save_conversation()
elif canonical == "export":
self._handle_export_command(cmd_original)
elif canonical == "cron": elif canonical == "cron":
self._handle_cron_command(cmd_original) self._handle_cron_command(cmd_original)
elif canonical == "suggestions": elif canonical == "suggestions":
+3
View File
@@ -16912,6 +16912,9 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
else: else:
return getattr(_blueprint_result, "text", "") or None return getattr(_blueprint_result, "text", "") or None
if canonical == "export":
return await self._handle_export_command(event)
if canonical == "retry": if canonical == "retry":
return await self._handle_retry_command(event) return await self._handle_retry_command(event)
+64
View File
@@ -4531,6 +4531,70 @@ class GatewaySlashCommandsMixin:
return await self._telegram_topic_root_status_message(source) return await self._telegram_topic_root_status_message(source)
async def _handle_export_command(self, event: MessageEvent) -> str:
"""Handle /export command — export session history and send it as a document."""
source = event.source
session_entry = self.session_store.get_or_create_session(source)
session_id = session_entry.session_id
if not self._session_db:
return "Session database not available."
args = event.get_command_args().split()
format_type = "markdown"
filename = f"session_{session_id}.md"
if len(args) > 0:
arg = args[0].lower()
if arg in ["json", "md", "markdown"]:
format_type = "json" if arg == "json" else "markdown"
filename = f"session_{session_id}.{'json' if format_type == 'json' else 'md'}"
else:
filename = arg
if len(args) > 1:
filename = args[1]
export_data = self._session_db.export_session(session_id)
if not export_data:
return f"Failed to export session {session_id}."
import json
import tempfile
import os
# Write to a temp file, then send it via adapter
temp_dir = tempfile.mkdtemp(prefix="hermes_export_")
temp_path = os.path.join(temp_dir, filename)
try:
with open(temp_path, "w", encoding="utf-8") as f:
if format_type == "json":
json.dump(export_data, f, indent=2, ensure_ascii=False)
else:
from hermes_state import SessionDB
f.write(SessionDB.format_session_as_markdown(export_data))
adapter = self.get_adapter(source.platform)
if adapter:
await adapter.send_document(
chat_id=source.chat_id,
file_path=temp_path,
caption=f"Here is your exported session: {filename}",
file_name=filename
)
return "Export complete."
else:
return "Platform adapter not found to send the document."
except Exception as e:
return f"Error exporting session: {e}"
finally:
try:
os.remove(temp_path)
os.rmdir(temp_dir)
except Exception:
pass
async def _handle_title_command(self, event: MessageEvent) -> str: async def _handle_title_command(self, event: MessageEvent) -> str:
"""Handle /title command — set or show the current session's title.""" """Handle /title command — set or show the current session's title."""
source = event.source source = event.source
+2
View File
@@ -156,6 +156,8 @@ COMMAND_REGISTRY: list[CommandDef] = [
cli_only=True), cli_only=True),
CommandDef("save", "Save the current conversation", "Session", CommandDef("save", "Save the current conversation", "Session",
cli_only=True), cli_only=True),
CommandDef("export", "Export the current session history to a file", "Session",
args_hint="[format] [filename]"),
CommandDef("retry", "Retry the last message (resend to agent)", "Session"), CommandDef("retry", "Retry the last message (resend to agent)", "Session"),
CommandDef("prompt", "Compose your next prompt in $EDITOR (markdown), then send it", "Session", CommandDef("prompt", "Compose your next prompt in $EDITOR (markdown), then send it", "Session",
cli_only=True, args_hint="[initial text]", aliases=("compose",)), cli_only=True, args_hint="[initial text]", aliases=("compose",)),
+36
View File
@@ -291,6 +291,42 @@ class SessionPortabilityMixin:
base["messages"] = [msg for seg in segments for msg in (seg.get("messages") or [])] base["messages"] = [msg for seg in segments for msg in (seg.get("messages") or [])]
return base return base
@staticmethod
def format_session_as_markdown(export_data: Dict[str, Any]) -> str:
"""Format exported session data as a readable Markdown document."""
lines = []
session_id = export_data.get("id", "Unknown")
lines.append(f"# Session Export: {session_id}\n")
title = export_data.get("title")
if title:
lines.append(f"**Title**: {title}")
created_at = export_data.get("created_at")
if created_at:
lines.append(f"**Started**: {created_at}")
if title or created_at:
lines.append("\n---\n")
for msg in export_data.get("messages", []):
role = msg.get("role", "unknown").capitalize()
content = msg.get("content", "")
if role == "User":
lines.append(f"## {role}\n\n{content}\n")
elif role == "Assistant":
reasoning = msg.get("reasoning", "")
if reasoning:
lines.append(f"## {role} (Reasoning)\n\n{reasoning}\n")
lines.append(f"## {role}\n\n{content}\n")
elif role == "Tool":
lines.append(f"### Tool Result\n\n```\n{content}\n```\n")
else:
lines.append(f"## {role}\n\n{content}\n")
return "\n".join(lines)
def export_all(self, source: str = None) -> List[Dict[str, Any]]: def export_all(self, source: str = None) -> List[Dict[str, Any]]:
""" """
Export all sessions (with messages) as a list of dicts. Export all sessions (with messages) as a list of dicts.
+6
View File
@@ -47,6 +47,12 @@ def _completions(completer: SlashCommandCompleter, text: str):
class TestCommandRegistry: class TestCommandRegistry:
def test_export_command_registered(self):
cmd = resolve_command("export")
assert cmd is not None
assert cmd.name == "export"
assert cmd.args_hint == "[format] [filename]"
def test_no_duplicate_canonical_names(self): def test_no_duplicate_canonical_names(self):
names = [cmd.name for cmd in COMMAND_REGISTRY] names = [cmd.name for cmd in COMMAND_REGISTRY]
assert len(names) == len(set(names)), f"Duplicate names: {[n for n in names if names.count(n) > 1]}" assert len(names) == len(set(names)), f"Duplicate names: {[n for n in names if names.count(n) > 1]}"
+21
View File
@@ -1056,6 +1056,27 @@ class TestDeleteAndExport:
def test_format_session_as_markdown(self, db):
db.create_session(session_id="s1", source="cli", model="test")
db.append_message("s1", role="user", content="Hello")
db.append_message("s1", role="assistant", content="Hi")
db.append_message("s1", role="assistant", content="Wait", reasoning="Thinking...")
db.append_message("s1", role="tool", content="Tool Result")
db.set_session_title("s1", "Test Title")
export = db.export_session("s1")
md_output = db.format_session_as_markdown(export)
assert "# Session Export: s1" in md_output
assert "**Title**: Test Title" in md_output
assert "## User\n\nHello" in md_output
assert "## Assistant\n\nHi" in md_output
assert "## Assistant (Reasoning)\n\nThinking..." in md_output
assert "## Assistant\n\nWait" in md_output
assert "### Tool Result\n\n```\nTool Result\n```" in md_output
def test_export_nonexistent(self, db):
assert db.export_session("nope") is None