mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-09 13:21:42 +00:00
0e04d14209
* feat(trace): upload sessions to HF Agent Trace Viewer Salvage trace upload as a smaller CLI-first feature: deterministic Claude Code JSONL export, fail-closed redaction, lazy Hugging Face dependency, and no gateway slash-command wiring. * chore(trace): drop external porting references from docstrings Describe the trace-upload design in Hermes' own terms. * feat(sessions): fold trace upload into 'sessions export --format trace' Integrates the HF Agent Trace Viewer exporter (PR #36145) onto the unified export surface instead of a separate 'hermes trace' subcommand: - --format trace: Claude Code JSONL to stdout/file, or one <id>.trace.jsonl per session for filtered bulk export; defaults to the most recent session when no --session-id/filters given. - --upload pushes to the user's private HF traces dataset (--public to opt out of private); reads HF_TOKEN with guided setup when missing. - traces are secret-redacted by default (force mode); --no-redact opts out after review; redaction failure blocks export (fail closed). - hermes_cli/trace.py + subcommands/trace.py removed; agent/trace_upload.py is the single engine. Docs EN + zh-Hans; 4 new CLI tests.
398 lines
14 KiB
Python
398 lines
14 KiB
Python
"""Upload a Hermes session transcript to Hugging Face as an agent trace.
|
|
|
|
Hermes stores sessions in its own SQLite store (``hermes_state.SessionDB``),
|
|
so we reconstruct the conversation and emit it in the **Claude Code JSONL**
|
|
shape — one of the three formats the Hugging Face Agent Trace Viewer
|
|
auto-detects (Claude Code / Codex / Pi). No dataset-side preprocessing is
|
|
needed; the Hub tags the dataset ``agent-traces`` and opens it in the viewer.
|
|
|
|
Docs: https://huggingface.co/docs/hub/agent-traces
|
|
|
|
Design notes
|
|
------------
|
|
* **Zero LLM turn.** This is a deterministic export — it never spends a
|
|
model call. The ``hermes trace upload`` subcommand calls
|
|
:func:`upload_session_trace` directly.
|
|
* **Private by default.** Traces can contain prompts, tool output, local
|
|
paths, and secrets. The dataset is created private and every text body
|
|
is passed through Hermes' secret redactor (``force=True``) unless the
|
|
caller explicitly opts out with ``redact=False``.
|
|
* **Never raises.** Returns a user-facing status string so command
|
|
handlers can echo it straight back to the user. Programmatic callers
|
|
that need the URL can use :func:`build_trace_jsonl` + :func:`_do_upload`
|
|
directly.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
DEFAULT_DATASET_NAME = "hermes-traces"
|
|
_HERMES_VERSION = "hermes-agent"
|
|
_REDACTION_BLOCKED_MESSAGE = (
|
|
"Trace upload blocked: secret redaction failed, so the transcript may "
|
|
"still contain credentials or other sensitive data. Fix the redactor or "
|
|
"rerun with --no-redact only after manually reviewing the transcript."
|
|
)
|
|
|
|
|
|
class TraceRedactionError(RuntimeError):
|
|
"""Raised when a trace cannot be safely redacted before upload."""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Conversion: Hermes OpenAI-format messages -> Claude Code JSONL
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _now_iso() -> str:
|
|
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
|
|
|
|
|
|
def _redact(text: Any, enabled: bool) -> Any:
|
|
"""Redact secrets from a string body when redaction is enabled.
|
|
|
|
Non-strings pass through untouched. Uses Hermes' shared redactor with
|
|
``force=True`` so an upload always scrubs known secret shapes even if
|
|
the user disabled log redaction globally.
|
|
"""
|
|
if not enabled or not isinstance(text, str) or not text:
|
|
return text
|
|
try:
|
|
from agent.redact import redact_sensitive_text
|
|
return redact_sensitive_text(text, force=True)
|
|
except Exception as exc:
|
|
logger.warning("Trace upload redaction failed; refusing upload", exc_info=True)
|
|
raise TraceRedactionError(_REDACTION_BLOCKED_MESSAGE) from exc
|
|
|
|
|
|
def _content_to_blocks(content: Any, redact: bool) -> List[Dict[str, Any]]:
|
|
"""Normalize a message ``content`` field into Anthropic content blocks."""
|
|
if content is None:
|
|
return []
|
|
if isinstance(content, str):
|
|
return [{"type": "text", "text": _redact(content, redact)}]
|
|
if isinstance(content, list):
|
|
blocks: List[Dict[str, Any]] = []
|
|
for part in content:
|
|
if isinstance(part, dict):
|
|
ptype = part.get("type")
|
|
if ptype == "text":
|
|
blocks.append({"type": "text", "text": _redact(part.get("text", ""), redact)})
|
|
elif ptype in ("image_url", "image"):
|
|
# Keep a placeholder; the viewer renders text turns and we
|
|
# don't want to inline base64 blobs into a trace.
|
|
blocks.append({"type": "text", "text": "[image omitted]"})
|
|
else:
|
|
blocks.append({"type": "text", "text": _redact(json.dumps(part), redact)})
|
|
else:
|
|
blocks.append({"type": "text", "text": _redact(str(part), redact)})
|
|
return blocks
|
|
return [{"type": "text", "text": _redact(json.dumps(content), redact)}]
|
|
|
|
|
|
def _tool_calls_to_blocks(tool_calls: Any, redact: bool) -> List[Dict[str, Any]]:
|
|
"""Convert OpenAI tool_calls into Anthropic ``tool_use`` content blocks."""
|
|
blocks: List[Dict[str, Any]] = []
|
|
if not isinstance(tool_calls, list):
|
|
return blocks
|
|
for tc in tool_calls:
|
|
if not isinstance(tc, dict):
|
|
continue
|
|
fn = tc.get("function") or {}
|
|
name = fn.get("name") or tc.get("name") or "tool"
|
|
raw_args = fn.get("arguments")
|
|
if isinstance(raw_args, str):
|
|
try:
|
|
parsed = json.loads(raw_args) if raw_args.strip() else {}
|
|
except (json.JSONDecodeError, ValueError):
|
|
parsed = {"_raw": raw_args}
|
|
elif isinstance(raw_args, dict):
|
|
parsed = raw_args
|
|
else:
|
|
parsed = {}
|
|
if redact:
|
|
try:
|
|
parsed = json.loads(_redact(json.dumps(parsed), redact))
|
|
except (json.JSONDecodeError, ValueError):
|
|
logger.warning("Trace upload redacted tool arguments are not valid JSON; refusing upload")
|
|
raise TraceRedactionError(_REDACTION_BLOCKED_MESSAGE)
|
|
blocks.append({
|
|
"type": "tool_use",
|
|
"id": tc.get("id") or f"toolu_{uuid.uuid4().hex[:16]}",
|
|
"name": name,
|
|
"input": parsed,
|
|
})
|
|
return blocks
|
|
|
|
|
|
def build_trace_jsonl(
|
|
messages: List[Dict[str, Any]],
|
|
*,
|
|
session_id: str,
|
|
model: str = "",
|
|
cwd: str = "",
|
|
redact: bool = True,
|
|
) -> str:
|
|
"""Render Hermes conversation messages as Claude Code JSONL text.
|
|
|
|
Each non-system message becomes one JSONL line in the Claude Code
|
|
transcript shape the HF Agent Trace Viewer auto-detects:
|
|
|
|
* ``user`` / ``tool`` -> ``{"type": "user", "message": {...}}``
|
|
* ``assistant`` -> ``{"type": "assistant", "message": {...}}``
|
|
with ``content`` blocks (text + ``tool_use``).
|
|
|
|
Tool results are emitted as user turns carrying a ``tool_result``
|
|
block keyed by ``tool_call_id`` — the same way Claude Code records
|
|
them. Turns are linked via ``uuid`` / ``parentUuid``.
|
|
"""
|
|
lines: List[str] = []
|
|
parent: Optional[str] = None
|
|
base_ts = _now_iso()
|
|
git_branch = ""
|
|
try:
|
|
import subprocess
|
|
if cwd:
|
|
r = subprocess.run(
|
|
["git", "rev-parse", "--abbrev-ref", "HEAD"],
|
|
capture_output=True, text=True, timeout=3, cwd=cwd,
|
|
)
|
|
if r.returncode == 0:
|
|
git_branch = r.stdout.strip()
|
|
except Exception:
|
|
git_branch = ""
|
|
|
|
def _common(turn_uuid: str) -> Dict[str, Any]:
|
|
return {
|
|
"parentUuid": parent,
|
|
"isSidechain": False,
|
|
"userType": "external",
|
|
"cwd": cwd or os.getcwd(),
|
|
"sessionId": session_id,
|
|
"version": _HERMES_VERSION,
|
|
"gitBranch": git_branch,
|
|
"uuid": turn_uuid,
|
|
"timestamp": base_ts,
|
|
}
|
|
|
|
for msg in messages:
|
|
role = msg.get("role")
|
|
if role == "system":
|
|
continue
|
|
turn_uuid = str(uuid.uuid4())
|
|
|
|
if role == "assistant":
|
|
blocks = _content_to_blocks(msg.get("content"), redact)
|
|
blocks.extend(_tool_calls_to_blocks(msg.get("tool_calls"), redact))
|
|
if not blocks:
|
|
blocks = [{"type": "text", "text": ""}]
|
|
entry = _common(turn_uuid)
|
|
entry["type"] = "assistant"
|
|
entry["message"] = {
|
|
"role": "assistant",
|
|
"model": model or "unknown",
|
|
"content": blocks,
|
|
}
|
|
lines.append(json.dumps(entry, ensure_ascii=False))
|
|
parent = turn_uuid
|
|
continue
|
|
|
|
if role == "tool":
|
|
tool_use_id = msg.get("tool_call_id") or msg.get("tool_name") or "tool"
|
|
result_content = _redact(
|
|
msg.get("content") if isinstance(msg.get("content"), str)
|
|
else json.dumps(msg.get("content")),
|
|
redact,
|
|
)
|
|
entry = _common(turn_uuid)
|
|
entry["type"] = "user"
|
|
entry["message"] = {
|
|
"role": "user",
|
|
"content": [{
|
|
"type": "tool_result",
|
|
"tool_use_id": tool_use_id,
|
|
"content": result_content,
|
|
}],
|
|
}
|
|
lines.append(json.dumps(entry, ensure_ascii=False))
|
|
parent = turn_uuid
|
|
continue
|
|
|
|
# Default: user (and any unknown role) -> user turn.
|
|
content = msg.get("content")
|
|
if isinstance(content, str):
|
|
message_content: Any = _redact(content, redact)
|
|
else:
|
|
message_content = _content_to_blocks(content, redact)
|
|
entry = _common(turn_uuid)
|
|
entry["type"] = "user"
|
|
entry["message"] = {"role": "user", "content": message_content}
|
|
lines.append(json.dumps(entry, ensure_ascii=False))
|
|
parent = turn_uuid
|
|
|
|
return "\n".join(lines) + ("\n" if lines else "")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Upload
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _resolve_hf_token() -> Optional[str]:
|
|
"""Return the user's Hugging Face token from the usual env vars."""
|
|
for var in ("HF_TOKEN", "HUGGINGFACE_HUB_TOKEN", "HUGGING_FACE_HUB_TOKEN", "HUGGINGFACE_TOKEN"):
|
|
val = os.getenv(var)
|
|
if val and val.strip():
|
|
return val.strip()
|
|
return None
|
|
|
|
|
|
_NO_TOKEN_MESSAGE = (
|
|
"Can't upload — no Hugging Face token is available. To set it up:\n"
|
|
"\n"
|
|
"1. Create a token with WRITE access at https://huggingface.co/settings/tokens\n"
|
|
" (New token -> type \"Write\" -> copy it).\n"
|
|
"2. Add it to your environment as HF_TOKEN (e.g. in ~/.hermes/.env):\n"
|
|
" HF_TOKEN=hf_xxxxxxxxxxxxxxxxxxxx\n"
|
|
"3. Run /upload-trace again (or `hermes trace upload`)."
|
|
)
|
|
|
|
|
|
def _do_upload(
|
|
jsonl: str,
|
|
*,
|
|
token: str,
|
|
session_id: str,
|
|
dataset_name: str = DEFAULT_DATASET_NAME,
|
|
private: bool = True,
|
|
) -> str:
|
|
"""Create (idempotently) the private dataset and push the trace file.
|
|
|
|
Returns a user-facing status string. Never raises.
|
|
"""
|
|
try:
|
|
from tools import lazy_deps
|
|
lazy_deps.ensure("tool.trace_upload", prompt=False)
|
|
except Exception:
|
|
# lazy-install unavailable/declined — fall through to the import,
|
|
# which surfaces the install hint below if the package is missing.
|
|
pass
|
|
try:
|
|
from huggingface_hub import HfApi
|
|
except ImportError:
|
|
return ("Hugging Face upload needs the `huggingface_hub` package "
|
|
"(`pip install huggingface_hub`).")
|
|
|
|
api = HfApi(token=token)
|
|
try:
|
|
who = api.whoami()
|
|
user = who.get("name") if isinstance(who, dict) else None
|
|
except Exception as e:
|
|
logger.warning("HF whoami failed: %s", e)
|
|
return ("Your Hugging Face token was rejected (whoami failed). "
|
|
"Make sure it has WRITE access and isn't expired.")
|
|
if not user:
|
|
return "Could not resolve your Hugging Face username from the token."
|
|
|
|
repo_id = f"{user}/{dataset_name}"
|
|
try:
|
|
api.create_repo(
|
|
repo_id=repo_id, repo_type="dataset", private=private, exist_ok=True,
|
|
)
|
|
except Exception as e:
|
|
logger.warning("HF create_repo failed for %s: %s", repo_id, e)
|
|
return f"Could not create/access dataset {repo_id}: {e}"
|
|
|
|
path_in_repo = f"sessions/{session_id}.jsonl"
|
|
try:
|
|
api.upload_file(
|
|
path_or_fileobj=jsonl.encode("utf-8"),
|
|
path_in_repo=path_in_repo,
|
|
repo_id=repo_id,
|
|
repo_type="dataset",
|
|
commit_message=f"add session trace {session_id}",
|
|
)
|
|
except Exception as e:
|
|
logger.warning("HF upload_file failed for %s: %s", repo_id, e)
|
|
return f"Upload to Hugging Face failed: {e}"
|
|
|
|
return (f"Uploaded -> https://huggingface.co/datasets/{repo_id}/blob/main/{path_in_repo}\n"
|
|
f"View in the trace viewer: https://huggingface.co/datasets/{repo_id}")
|
|
|
|
|
|
def load_session_messages(
|
|
session_id: str, db_path=None
|
|
) -> Tuple[List[Dict[str, Any]], Dict[str, Any]]:
|
|
"""Load a session's conversation + metadata from the SQLite store.
|
|
|
|
Returns ``(messages, meta)``. ``meta`` is ``{}`` when the session row is
|
|
missing (messages may still be present for a live, untitled session).
|
|
"""
|
|
from hermes_state import SessionDB
|
|
db = SessionDB(db_path=db_path) if db_path else SessionDB()
|
|
resolved = db.resolve_session_id(session_id) or session_id
|
|
meta = db.get_session(resolved) or {}
|
|
messages = db.get_messages_as_conversation(resolved)
|
|
return messages, meta
|
|
|
|
|
|
def upload_session_trace(
|
|
session_id: str,
|
|
*,
|
|
model: str = "",
|
|
cwd: str = "",
|
|
redact: bool = True,
|
|
private: bool = True,
|
|
dataset_name: str = DEFAULT_DATASET_NAME,
|
|
db_path=None,
|
|
token: Optional[str] = None,
|
|
) -> str:
|
|
"""Top-level entry point used by the CLI/gateway/subcommand.
|
|
|
|
Loads the session, converts it to Claude Code JSONL, and uploads it to
|
|
the user's private ``{user}/hermes-traces`` dataset. Returns a
|
|
user-facing status string and never raises.
|
|
"""
|
|
if not session_id:
|
|
return "No active session to upload."
|
|
|
|
token = token or _resolve_hf_token()
|
|
if not token:
|
|
return _NO_TOKEN_MESSAGE
|
|
|
|
try:
|
|
messages, meta = load_session_messages(session_id, db_path=db_path)
|
|
except Exception as e:
|
|
logger.warning("Failed to load session %s for trace upload: %s", session_id, e)
|
|
return f"Could not load session {session_id}: {e}"
|
|
|
|
if not messages:
|
|
return "No transcript to upload for this session yet."
|
|
|
|
resolved_model = model or meta.get("model") or ""
|
|
try:
|
|
jsonl = build_trace_jsonl(
|
|
messages,
|
|
session_id=session_id,
|
|
model=resolved_model,
|
|
cwd=cwd,
|
|
redact=redact,
|
|
)
|
|
except TraceRedactionError:
|
|
return _REDACTION_BLOCKED_MESSAGE
|
|
if not jsonl.strip():
|
|
return "No transcript content to upload for this session."
|
|
|
|
return _do_upload(
|
|
jsonl,
|
|
token=token,
|
|
session_id=session_id,
|
|
dataset_name=dataset_name,
|
|
private=private,
|
|
)
|