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.

agent/trace_upload.py

@@ -0,0 +1,401 @@
+"""Upload a Hermes session transcript to Hugging Face as an agent trace.
+
+Ported from qwibitai/nanoclaw#2648 ("/upload-trace"), adapted to Hermes'
+architecture. NanoClaw could push its Claude Code ``.jsonl`` files as-is
+because its runner *is* Claude Code. 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.** Like NanoClaw's runner-handled command, 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 the slash
+  handlers can echo it straight back to the user (matches the NanoClaw
+  contract). 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,
+    )

hermes_cli/main.py

@@ -299,6 +299,7 @@ from hermes_cli.subcommands.skills import build_skills_parser
 from hermes_cli.subcommands.pairing import build_pairing_parser
 from hermes_cli.subcommands.plugins import build_plugins_parser
 from hermes_cli.subcommands.mcp import build_mcp_parser
+from hermes_cli.subcommands.trace import build_trace_parser
 from hermes_cli.subcommands.claw import build_claw_parser
 
 
@@ -4081,6 +4082,13 @@ def cmd_status(args):
     show_status(args)
 
 
+def cmd_trace(args):
+    """Upload a session transcript to Hugging Face Agent Trace Viewer."""
+    from hermes_cli.trace import run_trace
+
+    run_trace(args)
+
+
 def cmd_cron(args):
     """Cron job management."""
     from hermes_cli.cron import cron_command
@@ -11500,6 +11508,11 @@ def main():
     # =========================================================================
     build_skills_parser(subparsers, cmd_skills=cmd_skills)
 
+    # =========================================================================
+    # trace command  (parser built in hermes_cli/subcommands/trace.py)
+    # =========================================================================
+    build_trace_parser(subparsers, cmd_trace=cmd_trace)
+
     # =========================================================================
     # bundles command — skill bundles (alias /<name> for multiple skills)
     # =========================================================================

hermes_cli/subcommands/trace.py

@@ -0,0 +1,40 @@
+"""``hermes trace`` subcommand parser."""
+
+from __future__ import annotations
+
+from typing import Callable
+
+
+def build_trace_parser(subparsers, *, cmd_trace: Callable) -> None:
+    """Attach the ``trace`` subcommand to ``subparsers``."""
+    trace_parser = subparsers.add_parser(
+        "trace",
+        help="Upload session transcripts to Hugging Face Agent Trace Viewer",
+        description=(
+            "Export a Hermes session transcript as Claude Code JSONL and upload "
+            "it to a private Hugging Face dataset for the Agent Trace Viewer."
+        ),
+    )
+    trace_sub = trace_parser.add_subparsers(dest="trace_command")
+
+    upload = trace_sub.add_parser(
+        "upload",
+        aliases=["up"],
+        help="Upload a session transcript",
+    )
+    upload.add_argument(
+        "session_id",
+        nargs="?",
+        help="Session id to upload (default: most recent session)",
+    )
+    upload.add_argument(
+        "--public",
+        action="store_true",
+        help="Create/update a public trace dataset instead of private",
+    )
+    upload.add_argument(
+        "--no-redact",
+        action="store_true",
+        help="Upload without secret redaction; only use after manual review",
+    )
+    trace_parser.set_defaults(func=cmd_trace)

hermes_cli/trace.py

@@ -0,0 +1,71 @@
+"""``hermes trace`` CLI subcommand — upload a session transcript to Hugging Face.
+
+Thin CLI wrapper over :mod:`agent.trace_upload`. The heavy lifting (session
+load, Claude Code JSONL conversion, HF upload) lives there so the command
+handler stays small and deterministic.
+
+Ported from qwibitai/nanoclaw#2648.
+"""
+
+from __future__ import annotations
+
+import sys
+
+
+def _latest_session_id() -> str | None:
+    """Return the most recently active session id, or None.
+
+    Used when the user runs ``hermes trace upload`` without an explicit id
+    (the common case from the shell — they mean "the last thing I did").
+    """
+    try:
+        from hermes_state import SessionDB
+        db = SessionDB()
+        sessions = db.list_sessions_rich(limit=1, order_by_last_active=True)
+        if sessions:
+            return sessions[0].get("id")
+    except Exception:
+        pass
+    return None
+
+
+def run_trace(args) -> None:
+    """Dispatch ``hermes trace <subcommand>``."""
+    sub = getattr(args, "trace_command", None)
+    if sub in ("upload", "up"):
+        _run_upload(args)
+        return
+    # No/unknown subcommand → brief usage.
+    print("Usage: hermes trace upload [SESSION_ID] [--public] [--no-redact]")
+    print("Upload a session transcript to your private Hugging Face traces dataset,")
+    print("viewable in the HF Agent Trace Viewer (https://huggingface.co/docs/hub/agent-traces).")
+
+
+def _run_upload(args) -> None:
+    from hermes_cli.env_loader import load_hermes_dotenv
+    from hermes_cli.config import get_env_path, get_project_root
+
+    # Load .env so HF_TOKEN is visible even outside a running session.
+    env_path = get_env_path()
+    try:
+        load_hermes_dotenv(
+            hermes_home=env_path.parent,
+            project_env=get_project_root() / ".env",
+        )
+    except Exception:
+        pass
+
+    from agent.trace_upload import upload_session_trace
+
+    session_id = getattr(args, "session_id", None) or _latest_session_id()
+    if not session_id:
+        print("No session found to upload. Pass a SESSION_ID explicitly.", file=sys.stderr)
+        sys.exit(1)
+
+    status = upload_session_trace(
+        session_id,
+        cwd="",
+        redact=not getattr(args, "no_redact", False),
+        private=not getattr(args, "public", False),
+    )
+    print(status)

tests/agent/test_trace_upload.py

@@ -0,0 +1,262 @@
+"""Tests for agent.trace_upload — Hugging Face session-trace upload.
+
+Ported from qwibitai/nanoclaw#2648. Covers the Claude Code JSONL converter,
+HF token resolution, the no-token message path, and the upload path with a
+mocked ``HfApi`` (verifying repo id, file path, and content without touching
+the network).
+"""
+
+import json
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from agent import trace_upload
+from agent.trace_upload import (
+    build_trace_jsonl,
+    upload_session_trace,
+    _resolve_hf_token,
+    _do_upload,
+)
+
+
+# ---------------------------------------------------------------------------
+# Converter
+# ---------------------------------------------------------------------------
+
+def _sample_messages():
+    return [
+        {"role": "system", "content": "you are hermes"},
+        {"role": "user", "content": "list files"},
+        {"role": "assistant", "content": "Listing.", "tool_calls": [
+            {"id": "call_1", "function": {"name": "terminal", "arguments": '{"command": "ls"}'}},
+        ]},
+        {"role": "tool", "tool_call_id": "call_1", "tool_name": "terminal", "content": "a.txt\nb.txt"},
+        {"role": "assistant", "content": "Two files."},
+    ]
+
+
+def test_converter_skips_system_and_counts_lines():
+    jsonl = build_trace_jsonl(_sample_messages(), session_id="s1", model="m")
+    lines = [json.loads(x) for x in jsonl.strip().split("\n")]
+    assert len(lines) == 4  # system dropped
+    assert all(o["sessionId"] == "s1" for o in lines)
+
+
+def test_converter_links_turns_as_linked_list():
+    jsonl = build_trace_jsonl(_sample_messages(), session_id="s1")
+    lines = [json.loads(x) for x in jsonl.strip().split("\n")]
+    prev = None
+    for o in lines:
+        assert o["parentUuid"] == prev
+        prev = o["uuid"]
+
+
+def test_converter_emits_tool_use_and_tool_result():
+    jsonl = build_trace_jsonl(_sample_messages(), session_id="s1", model="m")
+    lines = [json.loads(x) for x in jsonl.strip().split("\n")]
+    # line 0 user, line 1 assistant (text + tool_use), line 2 tool_result, line 3 assistant
+    assert lines[0]["type"] == "user"
+    assert lines[1]["type"] == "assistant"
+    blocks = lines[1]["message"]["content"]
+    assert any(b.get("type") == "text" for b in blocks)
+    tool_use = [b for b in blocks if b.get("type") == "tool_use"]
+    assert tool_use and tool_use[0]["name"] == "terminal"
+    assert tool_use[0]["input"] == {"command": "ls"}
+    # tool result rides on a user turn
+    assert lines[2]["type"] == "user"
+    tr = lines[2]["message"]["content"][0]
+    assert tr["type"] == "tool_result"
+    assert tr["tool_use_id"] == "call_1"
+    assert "a.txt" in tr["content"]
+
+
+def test_converter_redacts_secrets_by_default():
+    msgs = [{"role": "user", "content": "key OPENAI_API_KEY=sk-abc123def456ghi789jklmno end"}]
+    jsonl = build_trace_jsonl(msgs, session_id="s1", redact=True)
+    assert "sk-abc123def456ghi789jklmno" not in jsonl
+
+
+def test_converter_refuses_unredacted_passthrough_when_redactor_fails(monkeypatch):
+    def boom(_text, *, force=False):
+        raise RuntimeError("redactor unavailable")
+
+    monkeypatch.setattr("agent.redact.redact_sensitive_text", boom)
+    msgs = [{"role": "user", "content": "OPENAI_API_KEY=sk-abc123def456ghi789jklmno"}]
+
+    with pytest.raises(trace_upload.TraceRedactionError):
+        build_trace_jsonl(msgs, session_id="s1", redact=True)
+
+
+def test_upload_blocks_when_redactor_fails(monkeypatch):
+    monkeypatch.setenv("HF_TOKEN", "hf_test")
+
+    def boom(_text, *, force=False):
+        raise RuntimeError("redactor unavailable")
+
+    monkeypatch.setattr("agent.redact.redact_sensitive_text", boom)
+    with patch.object(trace_upload, "load_session_messages", return_value=(_sample_messages(), {})), \
+         patch.object(trace_upload, "_do_upload") as upload_mock:
+        msg = upload_session_trace("s1")
+
+    assert "Trace upload blocked" in msg
+    upload_mock.assert_not_called()
+
+
+def test_converter_keeps_secrets_when_redact_disabled():
+    secret = "sk-abc123def456ghi789jklmno"
+    msgs = [{"role": "user", "content": f"key OPENAI_API_KEY={secret} end"}]
+    jsonl = build_trace_jsonl(msgs, session_id="s1", redact=False)
+    assert secret in jsonl
+
+
+def test_converter_image_placeholder():
+    msgs = [{"role": "user", "content": [
+        {"type": "text", "text": "look"},
+        {"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}},
+    ]}]
+    jsonl = build_trace_jsonl(msgs, session_id="s1")
+    line = json.loads(jsonl.strip())
+    assert any("image omitted" in b.get("text", "") for b in line["message"]["content"])
+    assert "AAAA" not in jsonl
+
+
+def test_converter_empty_messages_returns_empty():
+    assert build_trace_jsonl([], session_id="s1") == ""
+
+
+def test_converter_handles_dict_tool_arguments():
+    msgs = [{"role": "assistant", "content": "", "tool_calls": [
+        {"id": "c", "function": {"name": "f", "arguments": {"already": "dict"}}},
+    ]}]
+    jsonl = build_trace_jsonl(msgs, session_id="s1")
+    line = json.loads(jsonl.strip())
+    tu = [b for b in line["message"]["content"] if b.get("type") == "tool_use"][0]
+    assert tu["input"] == {"already": "dict"}
+
+
+# ---------------------------------------------------------------------------
+# Token resolution
+# ---------------------------------------------------------------------------
+
+def test_resolve_token_prefers_hf_token(monkeypatch):
+    monkeypatch.setenv("HF_TOKEN", "hf_primary")
+    monkeypatch.setenv("HUGGINGFACE_TOKEN", "hf_secondary")
+    assert _resolve_hf_token() == "hf_primary"
+
+
+def test_resolve_token_falls_back(monkeypatch):
+    monkeypatch.delenv("HF_TOKEN", raising=False)
+    monkeypatch.delenv("HUGGINGFACE_HUB_TOKEN", raising=False)
+    monkeypatch.delenv("HUGGING_FACE_HUB_TOKEN", raising=False)
+    monkeypatch.setenv("HUGGINGFACE_TOKEN", "hf_fallback")
+    assert _resolve_hf_token() == "hf_fallback"
+
+
+def test_resolve_token_none(monkeypatch):
+    for v in ("HF_TOKEN", "HUGGINGFACE_HUB_TOKEN", "HUGGING_FACE_HUB_TOKEN", "HUGGINGFACE_TOKEN"):
+        monkeypatch.delenv(v, raising=False)
+    assert _resolve_hf_token() is None
+
+
+# ---------------------------------------------------------------------------
+# Top-level upload entry point
+# ---------------------------------------------------------------------------
+
+def test_upload_no_session_id():
+    assert "No active session" in upload_session_trace("")
+
+
+def test_upload_no_token(monkeypatch):
+    for v in ("HF_TOKEN", "HUGGINGFACE_HUB_TOKEN", "HUGGING_FACE_HUB_TOKEN", "HUGGINGFACE_TOKEN"):
+        monkeypatch.delenv(v, raising=False)
+    msg = upload_session_trace("some_session")
+    assert "no Hugging Face token" in msg
+
+
+def test_upload_empty_transcript(monkeypatch):
+    monkeypatch.setenv("HF_TOKEN", "hf_test")
+    with patch.object(trace_upload, "load_session_messages", return_value=([], {})):
+        msg = upload_session_trace("s1")
+    assert "No transcript" in msg
+
+
+def test_upload_happy_path_mocked(monkeypatch):
+    """Full upload path with a mocked HfApi — verifies repo id / path / content."""
+    pytest.importorskip("huggingface_hub")  # optional dep; runtime degrades gracefully
+    monkeypatch.setenv("HF_TOKEN", "hf_test")
+    messages = _sample_messages()
+
+    fake_api = MagicMock()
+    fake_api.whoami.return_value = {"name": "alice"}
+
+    with patch.object(trace_upload, "load_session_messages",
+                      return_value=(messages, {"model": "claude-x"})), \
+         patch("huggingface_hub.HfApi", return_value=fake_api):
+        msg = upload_session_trace("20260531_abc", cwd="/tmp")
+
+    # Returned a viewer URL
+    assert "huggingface.co/datasets/alice/hermes-traces" in msg
+
+    # Created private dataset repo
+    fake_api.create_repo.assert_called_once()
+    _, kwargs = fake_api.create_repo.call_args
+    assert kwargs["repo_id"] == "alice/hermes-traces"
+    assert kwargs["repo_type"] == "dataset"
+    assert kwargs["private"] is True
+
+    # Uploaded the JSONL to sessions/<id>.jsonl
+    fake_api.upload_file.assert_called_once()
+    _, ukwargs = fake_api.upload_file.call_args
+    assert ukwargs["path_in_repo"] == "sessions/20260531_abc.jsonl"
+    assert ukwargs["repo_id"] == "alice/hermes-traces"
+    body = ukwargs["path_or_fileobj"]
+    if isinstance(body, bytes):
+        body = body.decode("utf-8")
+    # Content is valid Claude Code JSONL
+    first = json.loads(body.strip().split("\n")[0])
+    assert first["type"] in ("user", "assistant")
+    assert first["sessionId"] == "20260531_abc"
+
+
+def test_upload_public_flag(monkeypatch):
+    pytest.importorskip("huggingface_hub")  # optional dep
+    monkeypatch.setenv("HF_TOKEN", "hf_test")
+    fake_api = MagicMock()
+    fake_api.whoami.return_value = {"name": "bob"}
+    with patch.object(trace_upload, "load_session_messages",
+                      return_value=(_sample_messages(), {})), \
+         patch("huggingface_hub.HfApi", return_value=fake_api):
+        upload_session_trace("s1", private=False)
+    _, kwargs = fake_api.create_repo.call_args
+    assert kwargs["private"] is False
+
+
+def test_upload_whoami_failure(monkeypatch):
+    pytest.importorskip("huggingface_hub")  # optional dep
+    monkeypatch.setenv("HF_TOKEN", "hf_bad")
+    fake_api = MagicMock()
+    fake_api.whoami.side_effect = Exception("401 unauthorized")
+    with patch.object(trace_upload, "load_session_messages",
+                      return_value=(_sample_messages(), {})), \
+         patch("huggingface_hub.HfApi", return_value=fake_api):
+        msg = upload_session_trace("s1")
+    assert "token was rejected" in msg
+
+
+def test_do_upload_missing_huggingface_hub(monkeypatch):
+    """If huggingface_hub import fails, return a clear install hint."""
+    # Disable lazy-install so the import path deterministically fails here
+    # instead of attempting a real pip install in CI.
+    monkeypatch.setenv("HERMES_DISABLE_LAZY_INSTALLS", "1")
+    import builtins
+    real_import = builtins.__import__
+
+    def fake_import(name, *args, **kwargs):
+        if name == "huggingface_hub":
+            raise ImportError("no module")
+        return real_import(name, *args, **kwargs)
+
+    monkeypatch.setattr(builtins, "__import__", fake_import)
+    msg = _do_upload("{}\n", token="t", session_id="s1")
+    assert "huggingface_hub" in msg

tools/lazy_deps.py

@@ -180,6 +180,7 @@ LAZY_DEPS: dict[str, tuple[str, ...]] = {
     # call site uses prompt=False so it can never raise a blocking input()
     # prompt mid-session (#40490).
     "tool.vision": ("Pillow==12.2.0",),
+    "tool.trace_upload": ("huggingface-hub==1.2.3",),
 }