Compare commits

...

3 Commits

Author SHA1 Message Date
Ben
db9a9d8e18 feat(debug): support /debug [nous|local] in the CLI/TUI slash command
The --nous flag was only wired into the argparse `hermes debug share`
subcommand. The /debug slash command (classic CLI + TUI, both via
process_command -> _handle_debug_command) built a hardcoded args
namespace with no `nous` attribute, so it always took the default
paste.rs path.

Pass cmd_original through to _handle_debug_command and parse an optional
destination word:

  /debug         -> public paste (default, unchanged)
  /debug nous    -> Nous-internal S3
  /debug local   -> stdout, no upload

local wins over nous (never touches the network); unknown words fall
back to the default. Add args_hint="[nous|local]" so help/autocomplete
surface it. New TestDebugSlashCommand covers the parsing + dispatch.
2026-06-09 16:13:13 +10:00
Ben
086dd4c28b feat(debug): drop dead confirm step from --nous upload (stateless NAS)
NAS PR #349 (merged) ships a stateless presigned-PUT endpoint: the only
route is POST /api/diagnostics/upload-url, and the object's existence in S3
is the only state. There is no /api/diagnostics/confirm route — confirming
live against the merged preview returns 404.

The client's confirm_upload() therefore fired a guaranteed-404 request on
every --nous upload (harmless, since errors were swallowed, but dead).
Remove it and simplify share_to_nous() to the 2-step mint + PUT flow that
matches the shipped contract. Drop the corresponding TestConfirmUpload class
and confirm assertions; add a test that the share succeeds even when the
response carries no id (we no longer depend on it).

The separately-flagged cross-repo requirement from #349's review --
sizeBytes is now REQUIRED and signed into the presigned URL's ContentLength
-- was already satisfied: share_to_nous() sends len(bundle) as sizeBytes and
urllib sets a matching Content-Length on the PUT. Verified against the live
merged preview (missing sizeBytes -> 400 invalid_body; present -> 503 dark).

Tested: pytest tests/hermes_cli/test_diagnostics_upload.py tests/hermes_cli/test_debug.py -> 95 passed.
2026-06-09 12:10:14 +10:00
Ben Barclay
bb6474cc51 feat(debug): add --nous flag to upload diagnostics to Nous S3
`hermes debug share --nous` uploads the (force-redacted) debug bundle to
Nous-internal S3 storage via a presigned URL minted by the Nous account
service, instead of a public paste. The bundle is private — viewable only
by Nous staff / allowlisted mods through a Google-OAuth-gated viewer — and
auto-deletes after 14 days. The paste.rs path is unchanged and remains the
default.

- hermes_cli/diagnostics_upload.py (new): stdlib-urllib NAS client —
  request_upload_url(), put_bundle(), confirm_upload() (best-effort),
  share_to_nous() orchestrator. Base URL via HERMES_DIAGNOSTICS_BASE_URL
  (default https://portal.nousresearch.com).
- hermes_cli/debug.py: extract collect_share_bundle() from build_debug_share()
  so the Nous path reuses the exact same redaction/collection (paste.rs
  behaviour unchanged); add build_nous_bundle() producing the gzipped
  {"format":"hermes-debug-share/1","redacted":...,"files":...} envelope the
  discord-support viewer parses; add the --nous run path with a privacy
  notice and a clean fallback (suggest --local) on failure.
- hermes_cli/main.py: add the --nous flag + help/epilog entry on
  `debug share`.
- tests: test_diagnostics_upload.py (new) mocks urllib; test_debug.py adds
  bundle/Nous coverage. 97 passing.
2026-06-09 12:10:13 +10:00
8 changed files with 825 additions and 73 deletions

2
cli.py
View File

@@ -7272,7 +7272,7 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
elif canonical == "copy":
self._handle_copy_command(cmd_original)
elif canonical == "debug":
self._handle_debug_command()
self._handle_debug_command(cmd_original)
elif canonical == "update":
if self._handle_update_command():
return False

View File

@@ -2090,12 +2090,25 @@ class CLICommandsMixin:
else:
_cprint(f" {_ACCENT}{feature_name} set to {label} (session only){_RST}")
def _handle_debug_command(self):
"""Handle /debug — upload debug report + logs and print paste URLs."""
def _handle_debug_command(self, cmd_original: str = ""):
"""Handle /debug — upload debug report + logs and print share URLs.
Accepts optional destination words after the command:
- ``/debug`` → upload to the public paste service (default)
- ``/debug nous`` → upload to Nous-internal storage (private, staff-only)
- ``/debug local`` → render the report to stdout, no upload
``nous`` and ``local`` are mutually exclusive; if both are given,
``local`` wins (it never touches the network).
"""
from hermes_cli.debug import run_debug_share
from types import SimpleNamespace
args = SimpleNamespace(lines=200, expire=7, local=False)
words = {w.lower() for w in cmd_original.split()[1:]}
local = "local" in words
nous = "nous" in words and not local
args = SimpleNamespace(lines=200, expire=7, local=local, nous=nous)
run_debug_share(args)
def _handle_update_command(self) -> bool:

View File

@@ -217,7 +217,8 @@ COMMAND_REGISTRY: list[CommandDef] = [
cli_only=True, args_hint="<path>"),
CommandDef("update", "Update Hermes Agent to the latest version", "Info"),
CommandDef("version", "Show Hermes Agent version", "Info", aliases=("v",)),
CommandDef("debug", "Upload debug report (system info + logs) and get shareable links", "Info"),
CommandDef("debug", "Upload debug report (system info + logs) and get shareable links", "Info",
args_hint="[nous|local]"),
# Exit
CommandDef("quit", "Exit the CLI (use --delete to also remove session history)", "Exit",

View File

@@ -9,8 +9,16 @@ Currently supports:
``~/.hermes/logs/*.log`` are not leaked into
the public paste service. Pass ``--no-redact``
to disable.
Pass ``--nous`` to upload instead to Nous-internal
storage (AWS S3) via a signed URL minted by the
Nous account service: the bundle is private
(viewable only by Nous staff / allowlisted mods via
a Google-login-gated viewer) and auto-deletes after
14 days, rather than going to a public paste.
"""
import datetime
import gzip
import io
import json
import logging
@@ -581,6 +589,97 @@ def collect_debug_report(
return buf.getvalue()
# ---------------------------------------------------------------------------
# Shared bundle collection (used by both the paste.rs and Nous-S3 paths)
# ---------------------------------------------------------------------------
# Bundle format identifier embedded in the Nous-S3 JSON envelope. The
# discord-support viewer keys off this string to parse the bundle.
_NOUS_BUNDLE_FORMAT = "hermes-debug-share/1"
def collect_share_bundle(
log_lines: int = 200,
redact: bool = True,
) -> dict[str, str]:
"""Collect the debug report + full logs as a label→text mapping.
Returns ``{"report": ..., "agent.log": ..., "gateway.log": ...,
"desktop.log": ...}`` where each value is the already-redacted (when
``redact`` is True) text that would be uploaded. Keys for logs that are
absent/empty are simply omitted.
This is the single source of collection + redaction shared by both
destinations: the paste.rs path (:func:`build_debug_share`) and the
Nous-S3 path (``--nous``). Centralising it guarantees the Nous bundle is
built from the *same* force-redacted snapshots as the public paste path —
redaction is the safety boundary, so the Nous path must never see raw
logs.
The dump header is prepended to each full log (mirroring the historical
paste behaviour) so every file is self-contained, and the redaction
banner is prepended when ``redact`` is True.
"""
dump_text = _capture_dump()
log_snapshots = _capture_default_log_snapshots(log_lines, redact=redact)
report = collect_debug_report(
log_lines=log_lines,
dump_text=dump_text,
log_snapshots=log_snapshots,
)
agent_log = log_snapshots["agent"].full_text
gateway_log = log_snapshots["gateway"].full_text
desktop_log = log_snapshots["desktop"].full_text
# Prepend dump header to each full log so every file is self-contained.
if agent_log:
agent_log = dump_text + "\n\n--- full agent.log ---\n" + agent_log
if gateway_log:
gateway_log = dump_text + "\n\n--- full gateway.log ---\n" + gateway_log
if desktop_log:
desktop_log = dump_text + "\n\n--- full desktop.log ---\n" + desktop_log
# Visible banner so reviewers know redaction was applied at upload time.
if redact:
report = _REDACTION_BANNER + report
if agent_log:
agent_log = _REDACTION_BANNER + agent_log
if gateway_log:
gateway_log = _REDACTION_BANNER + gateway_log
if desktop_log:
desktop_log = _REDACTION_BANNER + desktop_log
bundle: dict[str, str] = {"report": report}
if agent_log:
bundle["agent.log"] = agent_log
if gateway_log:
bundle["gateway.log"] = gateway_log
if desktop_log:
bundle["desktop.log"] = desktop_log
return bundle
def build_nous_bundle(bundle: dict[str, str], redact: bool = True) -> bytes:
"""Gzip-compress a :func:`collect_share_bundle` mapping into the Nous envelope.
The JSON shape is what the discord-support viewer (Repo 3) parses::
{"format": "hermes-debug-share/1",
"redacted": <bool>,
"created": <iso8601>,
"files": {"report": ..., "agent.log": ..., ...}}
"""
created = datetime.datetime.now(datetime.timezone.utc).isoformat()
envelope = {
"format": _NOUS_BUNDLE_FORMAT,
"redacted": bool(redact),
"created": created,
"files": bundle,
}
return gzip.compress(json.dumps(envelope).encode("utf-8"))
# ---------------------------------------------------------------------------
# CLI entry points
# ---------------------------------------------------------------------------
@@ -620,45 +719,18 @@ def build_debug_share(
"""
_best_effort_sweep_expired_pastes()
# Capture dump once — prepended to every paste for context.
# The dump is already redacted at extract time via dump.py:_redact;
# log_snapshots are redacted by _capture_default_log_snapshots when
# redact=True so credentials never reach the public paste service.
dump_text = _capture_dump()
log_snapshots = _capture_default_log_snapshots(log_lines, redact=redact)
# Collect the report + full logs (force-redacted when redact=True) via the
# shared collector so the paste.rs and Nous-S3 paths build identical,
# identically-redacted bundles. The dump header + redaction banner are
# applied inside collect_share_bundle.
bundle = collect_share_bundle(log_lines=log_lines, redact=redact)
if redact:
logger.info(
"hermes debug share: applied force-mode redaction to log snapshots before upload"
)
report = collect_debug_report(
log_lines=log_lines,
dump_text=dump_text,
log_snapshots=log_snapshots,
)
agent_log = log_snapshots["agent"].full_text
gateway_log = log_snapshots["gateway"].full_text
desktop_log = log_snapshots["desktop"].full_text
# Prepend dump header to each full log so every paste is self-contained.
if agent_log:
agent_log = dump_text + "\n\n--- full agent.log ---\n" + agent_log
if gateway_log:
gateway_log = dump_text + "\n\n--- full gateway.log ---\n" + gateway_log
if desktop_log:
desktop_log = dump_text + "\n\n--- full desktop.log ---\n" + desktop_log
# Visible banner so reviewers reading the public paste know redaction
# was applied at upload time. Banner is omitted under --no-redact.
if redact:
report = _REDACTION_BANNER + report
if agent_log:
agent_log = _REDACTION_BANNER + agent_log
if gateway_log:
gateway_log = _REDACTION_BANNER + gateway_log
if desktop_log:
desktop_log = _REDACTION_BANNER + desktop_log
report = bundle["report"]
urls: dict[str, str] = {}
failures: list[str] = []
@@ -667,11 +739,8 @@ def build_debug_share(
urls["Report"] = upload_to_pastebin(report, expiry_days=expiry)
# 2-4. Full logs (optional — failures are collected, not raised)
for label, content in (
("agent.log", agent_log),
("gateway.log", gateway_log),
("desktop.log", desktop_log),
):
for label in ("agent.log", "gateway.log", "desktop.log"):
content = bundle.get(label)
if not content:
continue
try:
@@ -696,43 +765,23 @@ def run_debug_share(args):
log_lines = getattr(args, "lines", 200)
expiry = getattr(args, "expire", 7)
local_only = getattr(args, "local", False)
nous = getattr(args, "nous", False)
redact = not getattr(args, "no_redact", False)
if local_only:
# Local-only path never uploads — render the report to stdout and bail
# before any network I/O. Mirrors the upload path's collection logic.
# before any network I/O. Reuses the shared collector so the rendered
# output matches exactly what would be uploaded.
_best_effort_sweep_expired_pastes()
print("Collecting debug report...")
dump_text = _capture_dump()
log_snapshots = _capture_default_log_snapshots(log_lines, redact=redact)
report = collect_debug_report(
log_lines=log_lines,
dump_text=dump_text,
log_snapshots=log_snapshots,
)
agent_log = log_snapshots["agent"].full_text
gateway_log = log_snapshots["gateway"].full_text
desktop_log = log_snapshots["desktop"].full_text
if agent_log:
agent_log = dump_text + "\n\n--- full agent.log ---\n" + agent_log
if gateway_log:
gateway_log = dump_text + "\n\n--- full gateway.log ---\n" + gateway_log
if desktop_log:
desktop_log = dump_text + "\n\n--- full desktop.log ---\n" + desktop_log
if redact:
report = _REDACTION_BANNER + report
if agent_log:
agent_log = _REDACTION_BANNER + agent_log
if gateway_log:
gateway_log = _REDACTION_BANNER + gateway_log
if desktop_log:
desktop_log = _REDACTION_BANNER + desktop_log
print(report)
for title, body in (
("FULL agent.log", agent_log),
("FULL gateway.log", gateway_log),
("FULL desktop.log", desktop_log),
bundle = collect_share_bundle(log_lines=log_lines, redact=redact)
print(bundle["report"])
for title, label in (
("FULL agent.log", "agent.log"),
("FULL gateway.log", "gateway.log"),
("FULL desktop.log", "desktop.log"),
):
body = bundle.get(label)
if body:
print(f"\n\n{'=' * 60}")
print(title)
@@ -740,6 +789,10 @@ def run_debug_share(args):
print(body)
return
if nous:
_run_debug_share_nous(log_lines=log_lines, redact=redact)
return
print(_PRIVACY_NOTICE)
print("Collecting debug report...")
print("Uploading...")
@@ -773,6 +826,80 @@ def run_debug_share(args):
print(f"\nShare these links with the Hermes team for support.")
_NOUS_PRIVACY_NOTICE = """\
⚠️ --nous: This uploads your debug bundle to Nous-INTERNAL storage (AWS S3),
NOT a public paste service. The following is included:
• System info (OS, Python/Hermes version, provider, which API keys are
configured — NOT the actual keys)
• Full agent.log, gateway.log, and desktop.log (up to 512 KB each — likely
contains conversation content, tool outputs, and file paths)
• The bundle is viewable only by Nous staff (and allowlisted Discord mods)
via a Google-login-gated viewer.
• It is NOT a public paste — there is no public URL to the contents.
• It auto-deletes after 14 days.
"""
def _run_debug_share_nous(*, log_lines: int, redact: bool) -> None:
"""Handle ``hermes debug share --nous``: upload the bundle to Nous-S3.
Collects the same force-redacted bundle as the paste path, gzips it into
the Nous envelope, requests a signed URL from NAS, uploads, and prints the
private viewer link. On any failure falls back to a clear error that
suggests ``--local``.
"""
from hermes_cli.diagnostics_upload import share_to_nous
print(_NOUS_PRIVACY_NOTICE)
if not redact:
print(
"⚠️ --no-redact is set: secrets in your logs will NOT be redacted "
"before upload.\n"
)
print("Collecting debug report...")
_best_effort_sweep_expired_pastes()
bundle = collect_share_bundle(log_lines=log_lines, redact=redact)
if redact:
logger.info(
"hermes debug share --nous: applied force-mode redaction before upload"
)
blob = build_nous_bundle(bundle, redact=redact)
print("Uploading to Nous diagnostics storage...")
try:
res = share_to_nous(blob)
except Exception as exc:
print(
f"\nNous upload failed: {exc}\n"
"\nThe Nous diagnostics service may be unavailable or not yet "
"provisioned.\n"
"Run `hermes debug share --local` to print the report instead, "
"or `hermes debug share` to upload to a public paste service.\n",
file=sys.stderr,
)
sys.exit(1)
view_url = res.get("viewUrl") or res.get("view_url")
print("\nDebug bundle uploaded to Nous (private):")
if view_url:
print(f" View URL {view_url}")
else:
print(f" (no view URL returned; upload id: {res.get('id', '?')})")
expires_at = res.get("expiresAt") or res.get("expires_at")
if expires_at:
print(f"\n⏱ Auto-deletes at {expires_at} (14-day retention).")
else:
print("\n⏱ Auto-deletes after 14 days.")
print(
"\nShare this private link with the Nous team — only Nous staff "
"(via Google login) can open it."
)
def run_debug_delete(args):
"""Delete one or more paste URLs uploaded by /debug."""
urls = getattr(args, "urls", [])
@@ -823,6 +950,8 @@ def run_debug(args):
print(" --lines N Number of log lines to include (default: 200)")
print(" --expire N Paste expiry in days (default: 7)")
print(" --local Print report locally instead of uploading")
print(" --nous Upload to Nous-internal storage (private, staff-only,")
print(" auto-deletes in 14 days) instead of a public paste")
print(" --no-redact Disable upload-time secret redaction (default: redact)")
print()
print("Options (delete):")

View File

@@ -0,0 +1,138 @@
"""Client for uploading ``hermes debug share`` bundles to Nous-internal S3.
This is the opt-in (``--nous``) destination for ``hermes debug share``.
Unlike the public paste.rs path, bundles uploaded here go to a Nous-owned
S3 bucket via a short-lived signed URL minted by the Nous account service
(NAS). The bucket auto-expires objects after 14 days, and the contents are
only viewable by Nous staff (and allowlisted Discord mods) through a
Google-OAuth-gated viewer.
Flow:
1. POST {NAS_BASE}/api/diagnostics/upload-url → {uploadUrl, viewUrl, id, ...}
(the request body carries ``sizeBytes``; NAS signs it into the presigned
URL's ``ContentLength``, so the PUT must send exactly that many bytes)
2. PUT <uploadUrl> (the gzipped bundle, Content-Type application/gzip)
NAS is stateless — the object's existence in S3 is the only state, so there is
no confirm/callback step.
Uses stdlib ``urllib`` only, matching ``debug.py`` style — no third-party deps.
"""
import json
import os
import urllib.request
# Base URL of the Nous account service that mints the signed upload URL.
# Overridable via env so the feature can be pointed at staging / a local dev
# NAS instance during testing.
NAS_BASE = os.environ.get(
"HERMES_DIAGNOSTICS_BASE_URL", "https://portal.nousresearch.com"
)
# Network timeout for each request (seconds). The upload itself can be larger
# (a gzipped log bundle), so the PUT gets a more generous window.
_REQUEST_TIMEOUT = 30
_UPLOAD_TIMEOUT = 120
_USER_AGENT = "hermes-agent/debug-share"
def request_upload_url(
content_type: str = "application/gzip",
size_bytes: int | None = None,
) -> dict:
"""Ask NAS to mint a presigned PUT URL for a diagnostics bundle.
POSTs a small JSON body to ``{NAS_BASE}/api/diagnostics/upload-url`` and
returns the parsed JSON response, expected to contain at least
``uploadUrl``, ``viewUrl`` and ``id`` (plus optional ``expiresAt`` /
``uploadExpiresInSeconds``).
Raises on non-2xx responses or unparseable JSON.
"""
payload: dict = {"contentType": content_type}
if size_bytes is not None:
payload["sizeBytes"] = int(size_bytes)
data = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(
f"{NAS_BASE}/api/diagnostics/upload-url",
data=data,
method="POST",
headers={
"Content-Type": "application/json",
"Accept": "application/json",
"User-Agent": _USER_AGENT,
},
)
with urllib.request.urlopen(req, timeout=_REQUEST_TIMEOUT) as resp:
status = getattr(resp, "status", None)
if status is None:
status = resp.getcode()
if not (200 <= status < 300):
raise RuntimeError(
f"diagnostics upload-url request failed: HTTP {status}"
)
body = resp.read().decode("utf-8")
try:
result = json.loads(body)
except (ValueError, json.JSONDecodeError) as exc:
raise RuntimeError(
f"diagnostics upload-url returned non-JSON response: {body[:200]}"
) from exc
if not isinstance(result, dict) or not result.get("uploadUrl"):
raise RuntimeError(
"diagnostics upload-url response missing 'uploadUrl': "
f"{body[:200]}"
)
return result
def put_bundle(
upload_url: str,
data: bytes,
content_type: str = "application/gzip",
) -> None:
"""PUT the gzipped *data* bundle to a presigned *upload_url*.
Sets the ``Content-Type`` header (must match what NAS pinned when signing
the URL, otherwise S3 rejects the signature). Raises on non-2xx.
"""
req = urllib.request.Request(
upload_url,
data=data,
method="PUT",
headers={
"Content-Type": content_type,
"User-Agent": _USER_AGENT,
},
)
with urllib.request.urlopen(req, timeout=_UPLOAD_TIMEOUT) as resp:
status = getattr(resp, "status", None)
if status is None:
status = resp.getcode()
if not (200 <= status < 300):
raise RuntimeError(f"diagnostics bundle PUT failed: HTTP {status}")
def share_to_nous(report_bundle: bytes) -> dict:
"""Orchestrate the full Nous-S3 upload of a gzipped *report_bundle*.
Two steps: mint a presigned PUT URL (sending the exact ``sizeBytes`` NAS
signs into the URL's ``ContentLength``), then PUT the bundle. NAS is
stateless — the object's existence in S3 is the only state, so there is no
confirm/callback step. Returns the dict from :func:`request_upload_url`
(which carries ``viewUrl`` / ``id`` / expiry metadata) so the caller can
print the viewer link. Raises on any failure of either step.
"""
size_bytes = len(report_bundle)
info = request_upload_url(
content_type="application/gzip", size_bytes=size_bytes
)
put_bundle(info["uploadUrl"], report_bundle, content_type="application/gzip")
return info

View File

@@ -29,6 +29,7 @@ Examples:
hermes debug share --expire 30 Keep paste for 30 days
hermes debug share --local Print report locally (no upload)
hermes debug share --no-redact Disable upload-time secret redaction
hermes debug share --nous Upload to Nous-internal storage (private)
hermes debug delete <url> Delete a previously uploaded paste
""",
)
@@ -64,6 +65,17 @@ Examples:
"into the public paste service."
),
)
share_parser.add_argument(
"--nous",
action="store_true",
help=(
"Upload the debug bundle to Nous-internal storage (AWS S3) instead "
"of a public paste service. The bundle is private — viewable only "
"by Nous staff (and allowlisted Discord mods) via a Google-login-"
"gated viewer — and auto-deletes after 14 days. Still force-redacts "
"secrets unless --no-redact is also passed."
),
)
delete_parser = debug_sub.add_parser(
"delete",
help="Delete a paste uploaded by 'hermes debug share'",

View File

@@ -491,6 +491,7 @@ class TestRunDebugShare:
args.lines = 50
args.expire = 7
args.local = False
args.nous = False
with patch("hermes_cli.dump.run_dump"), \
patch("hermes_cli.debug._sweep_expired_pastes", return_value=(0, 0)) as mock_sweep, \
@@ -509,6 +510,7 @@ class TestRunDebugShare:
args.lines = 50
args.expire = 7
args.local = False
args.nous = False
with patch("hermes_cli.dump.run_dump"), \
patch(
@@ -529,6 +531,7 @@ class TestRunDebugShare:
args.lines = 50
args.expire = 7
args.local = True
args.nous = False
with patch("hermes_cli.dump.run_dump"):
run_debug_share(args)
@@ -546,6 +549,7 @@ class TestRunDebugShare:
args.lines = 50
args.expire = 7
args.local = False
args.nous = False
call_count = [0]
uploaded_content = []
@@ -599,6 +603,7 @@ class TestRunDebugShare:
args.lines = 50
args.expire = 7
args.local = False
args.nous = False
uploaded_content = []
@@ -644,6 +649,7 @@ class TestRunDebugShare:
args.lines = 50
args.expire = 7
args.local = False
args.nous = False
call_count = [0]
def _mock_upload(content, expiry_days=7):
@@ -668,6 +674,7 @@ class TestRunDebugShare:
args.lines = 50
args.expire = 7
args.local = False
args.nous = False
call_count = [0]
def _mock_upload(content, expiry_days=7):
@@ -694,6 +701,7 @@ class TestRunDebugShare:
args.lines = 50
args.expire = 7
args.local = False
args.nous = False
with patch("hermes_cli.dump.run_dump"), \
patch("hermes_cli.debug.upload_to_pastebin",
@@ -742,6 +750,7 @@ class TestRunDebugShareRedaction:
args.lines = 50
args.expire = 7
args.local = False
args.nous = False
args.no_redact = False
captured: list[str] = []
@@ -772,6 +781,7 @@ class TestRunDebugShareRedaction:
args.lines = 50
args.expire = 7
args.local = False
args.nous = False
args.no_redact = False
captured: list[str] = []
@@ -800,6 +810,7 @@ class TestRunDebugShareRedaction:
args.lines = 50
args.expire = 7
args.local = False
args.nous = False
args.no_redact = True
captured: list[str] = []
@@ -850,6 +861,7 @@ class TestRunDebug:
args.lines = 200
args.expire = 7
args.local = True
args.nous = False
with patch("hermes_cli.dump.run_dump"):
run_debug(args)
@@ -1228,6 +1240,7 @@ class TestShareIncludesAutoDelete:
args.lines = 50
args.expire = 7
args.local = False
args.nous = False
with patch("hermes_cli.dump.run_dump"), \
patch("hermes_cli.debug.upload_to_pastebin",
@@ -1250,6 +1263,7 @@ class TestShareIncludesAutoDelete:
args.lines = 50
args.expire = 7
args.local = False
args.nous = False
with patch("hermes_cli.dump.run_dump"), \
patch("hermes_cli.debug.upload_to_pastebin",
@@ -1267,6 +1281,7 @@ class TestShareIncludesAutoDelete:
args.lines = 50
args.expire = 7
args.local = True
args.nous = False
with patch("hermes_cli.dump.run_dump"):
run_debug_share(args)
@@ -1380,3 +1395,220 @@ class TestBuildDebugShare:
), patch("hermes_cli.debug._schedule_auto_delete"):
with pytest.raises(RuntimeError, match="all paste services down"):
build_debug_share(log_lines=50, redact=True)
# ---------------------------------------------------------------------------
# Shared bundle collection + Nous-S3 path
# ---------------------------------------------------------------------------
class TestCollectShareBundle:
def test_returns_report_and_logs(self, hermes_home):
from hermes_cli.debug import collect_share_bundle
with patch("hermes_cli.dump.run_dump"):
bundle = collect_share_bundle(log_lines=50, redact=True)
assert "report" in bundle
assert "agent.log" in bundle
assert "gateway.log" in bundle
assert "desktop.log" in bundle
# Banner is prepended under redact=True.
assert "redacted at upload time" in bundle["report"]
assert "session started" in bundle["agent.log"]
def test_no_redact_omits_banner(self, hermes_home):
from hermes_cli.debug import collect_share_bundle
with patch("hermes_cli.dump.run_dump"):
bundle = collect_share_bundle(log_lines=50, redact=False)
assert "redacted at upload time" not in bundle["report"]
def test_redaction_keeps_secrets_out(self, hermes_home):
from hermes_cli.debug import collect_share_bundle
secret = "sk-proj-abcdefghijklmnopqrstuvwxyz1234567890"
(hermes_home / "logs" / "agent.log").write_text(
f"line one\nOPENAI_API_KEY={secret}\nline three\n"
)
with patch("hermes_cli.dump.run_dump"):
redacted = collect_share_bundle(log_lines=50, redact=True)
unredacted = collect_share_bundle(log_lines=50, redact=False)
# Sanity: without redaction the secret is present in the bundle.
assert secret in "\n".join(unredacted.values())
# With redaction it must be scrubbed everywhere.
assert secret not in "\n".join(redacted.values())
def test_build_debug_share_uses_collector(self, hermes_home):
# build_debug_share must produce the same report text the collector does
# (i.e. the refactor preserved paste.rs behaviour).
from hermes_cli.debug import build_debug_share, collect_share_bundle
with patch("hermes_cli.dump.run_dump"):
expected = collect_share_bundle(log_lines=50, redact=True)["report"]
uploaded = []
def _upload(content, expiry_days=7):
uploaded.append(content)
return "https://paste.rs/x"
with patch("hermes_cli.dump.run_dump"), patch(
"hermes_cli.debug.upload_to_pastebin", side_effect=_upload
), patch("hermes_cli.debug._schedule_auto_delete"):
result = build_debug_share(log_lines=50, redact=True)
assert result.urls["Report"] == "https://paste.rs/x"
# The report uploaded should match the collector's report.
assert uploaded[0] == expected
class TestBuildNousBundle:
def test_envelope_shape_and_gzip(self, hermes_home):
import gzip
import json as _json
from hermes_cli.debug import build_nous_bundle
files = {"report": "hello", "agent.log": "log line"}
blob = build_nous_bundle(files, redact=True)
# It's gzip — magic bytes.
assert blob[:2] == b"\x1f\x8b"
envelope = _json.loads(gzip.decompress(blob).decode())
assert envelope["format"] == "hermes-debug-share/1"
assert envelope["redacted"] is True
assert envelope["files"] == files
assert "created" in envelope
def test_redacted_false_recorded(self):
import gzip
import json as _json
from hermes_cli.debug import build_nous_bundle
blob = build_nous_bundle({"report": "x"}, redact=False)
envelope = _json.loads(gzip.decompress(blob).decode())
assert envelope["redacted"] is False
class TestRunDebugShareNous:
def _args(self, **over):
class _A:
lines = 50
expire = 7
local = False
nous = True
no_redact = False
a = _A()
for k, v in over.items():
setattr(a, k, v)
return a
def test_nous_success_prints_view_url(self, hermes_home, capsys):
from hermes_cli.debug import run_debug_share
res = {
"id": "id-1",
"viewUrl": "https://support.example.com/diagnostics/id-1",
"expiresAt": "2026-06-20T00:00:00Z",
}
with patch("hermes_cli.dump.run_dump"), patch(
"hermes_cli.diagnostics_upload.share_to_nous", return_value=res
) as share:
run_debug_share(self._args())
out = capsys.readouterr().out
assert "Nous-INTERNAL" in out
assert "https://support.example.com/diagnostics/id-1" in out
assert "2026-06-20T00:00:00Z" in out
# The blob passed to share_to_nous must be gzip bytes.
blob = share.call_args[0][0]
assert isinstance(blob, (bytes, bytearray)) and blob[:2] == b"\x1f\x8b"
def test_nous_failure_suggests_local(self, hermes_home, capsys):
from hermes_cli.debug import run_debug_share
with patch("hermes_cli.dump.run_dump"), patch(
"hermes_cli.diagnostics_upload.share_to_nous",
side_effect=RuntimeError("service down"),
):
with pytest.raises(SystemExit) as exc:
run_debug_share(self._args())
assert exc.value.code == 1
err = capsys.readouterr().err
assert "Nous upload failed" in err
assert "--local" in err
def test_nous_does_not_touch_pastebin(self, hermes_home):
from hermes_cli.debug import run_debug_share
res = {"id": "id-1", "viewUrl": "https://v"}
with patch("hermes_cli.dump.run_dump"), patch(
"hermes_cli.diagnostics_upload.share_to_nous", return_value=res
), patch("hermes_cli.debug.upload_to_pastebin") as paste:
run_debug_share(self._args())
paste.assert_not_called()
class TestDebugSlashCommand:
"""`/debug [nous|local]` parsing in the CLI/TUI handler.
The classic CLI and the TUI slash worker both dispatch through
``HermesCLI.process_command`` → ``_handle_debug_command(cmd_original)``,
which parses an optional destination word and builds the args namespace
handed to ``run_debug_share``.
"""
def _handler(self):
from hermes_cli.cli_commands_mixin import CLICommandsMixin
class _Stub(CLICommandsMixin):
pass
return _Stub()._handle_debug_command
def _captured(self, cmd_original):
captured = {}
def _fake_run(args):
captured.update(vars(args))
with patch("hermes_cli.debug.run_debug_share", _fake_run):
self._handler()(cmd_original)
return captured
def test_bare_debug_defaults_to_paste(self):
c = self._captured("/debug")
assert c["nous"] is False and c["local"] is False
assert c["lines"] == 200 and c["expire"] == 7
def test_nous_word_sets_nous(self):
c = self._captured("/debug nous")
assert c["nous"] is True and c["local"] is False
def test_local_word_sets_local(self):
c = self._captured("/debug local")
assert c["local"] is True and c["nous"] is False
def test_word_parsing_is_case_insensitive(self):
c = self._captured("/debug NOUS")
assert c["nous"] is True
def test_local_wins_over_nous(self):
# local never touches the network, so it takes precedence.
c = self._captured("/debug nous local")
assert c["local"] is True and c["nous"] is False
def test_unknown_word_falls_back_to_default(self):
c = self._captured("/debug paste")
assert c["nous"] is False and c["local"] is False
def test_no_arg_default_keyword(self):
# Calling with no cmd_original (legacy callers) must still work.
c = self._captured("")
assert c["nous"] is False and c["local"] is False

View File

@@ -0,0 +1,227 @@
"""Tests for ``hermes_cli.diagnostics_upload`` — the Nous-S3 upload client.
All network I/O is mocked at ``urllib.request.urlopen``; no real requests
are made.
"""
import io
import json
import urllib.error
from unittest.mock import MagicMock, patch
import pytest
def _resp(*, status=200, body=b""):
"""Build a context-manager mock mimicking ``urllib.request.urlopen``."""
m = MagicMock()
m.status = status
m.getcode.return_value = status
m.read.return_value = body
m.__enter__ = lambda s: s
m.__exit__ = MagicMock(return_value=False)
return m
# ---------------------------------------------------------------------------
# request_upload_url
# ---------------------------------------------------------------------------
class TestRequestUploadUrl:
def test_happy_path_posts_json_and_returns_dict(self):
from hermes_cli.diagnostics_upload import request_upload_url
payload = {
"success": True,
"id": "abc-123",
"uploadUrl": "https://bucket.s3.amazonaws.com/uploads/abc-123.json.gz?sig",
"viewUrl": "https://support.example.com/diagnostics/abc-123",
"uploadExpiresInSeconds": 900,
}
resp = _resp(status=200, body=json.dumps(payload).encode())
with patch(
"hermes_cli.diagnostics_upload.urllib.request.urlopen",
return_value=resp,
) as urlopen:
result = request_upload_url(content_type="application/gzip", size_bytes=512)
assert result == payload
# The request object passed to urlopen carries our JSON body + headers.
req = urlopen.call_args[0][0]
assert req.method == "POST"
assert req.full_url.endswith("/api/diagnostics/upload-url")
sent = json.loads(req.data.decode())
assert sent["contentType"] == "application/gzip"
assert sent["sizeBytes"] == 512
# urllib lower-cases header keys.
assert req.headers["Content-type"] == "application/json"
def test_non_2xx_raises(self):
from hermes_cli.diagnostics_upload import request_upload_url
resp = _resp(status=500, body=b"boom")
with patch(
"hermes_cli.diagnostics_upload.urllib.request.urlopen",
return_value=resp,
):
with pytest.raises(RuntimeError):
request_upload_url()
def test_missing_upload_url_raises(self):
from hermes_cli.diagnostics_upload import request_upload_url
resp = _resp(status=200, body=json.dumps({"id": "x"}).encode())
with patch(
"hermes_cli.diagnostics_upload.urllib.request.urlopen",
return_value=resp,
):
with pytest.raises(RuntimeError):
request_upload_url()
def test_non_json_raises(self):
from hermes_cli.diagnostics_upload import request_upload_url
resp = _resp(status=200, body=b"<html>not json</html>")
with patch(
"hermes_cli.diagnostics_upload.urllib.request.urlopen",
return_value=resp,
):
with pytest.raises(RuntimeError):
request_upload_url()
def test_base_url_env_override(self, monkeypatch):
# NAS_BASE is read at import time; re-import the module under the
# patched env to confirm the override is honoured.
import importlib
monkeypatch.setenv("HERMES_DIAGNOSTICS_BASE_URL", "https://staging.example.com")
import hermes_cli.diagnostics_upload as mod
mod = importlib.reload(mod)
try:
assert mod.NAS_BASE == "https://staging.example.com"
resp = _resp(
status=200,
body=json.dumps({"uploadUrl": "u", "id": "i", "viewUrl": "v"}).encode(),
)
with patch(
"hermes_cli.diagnostics_upload.urllib.request.urlopen",
return_value=resp,
) as urlopen:
mod.request_upload_url()
req = urlopen.call_args[0][0]
assert req.full_url == "https://staging.example.com/api/diagnostics/upload-url"
finally:
monkeypatch.delenv("HERMES_DIAGNOSTICS_BASE_URL", raising=False)
importlib.reload(mod)
# ---------------------------------------------------------------------------
# put_bundle
# ---------------------------------------------------------------------------
class TestPutBundle:
def test_put_sends_exact_body_and_content_type(self):
from hermes_cli.diagnostics_upload import put_bundle
data = b"\x1f\x8b\x08gzipped-bytes"
resp = _resp(status=200, body=b"")
with patch(
"hermes_cli.diagnostics_upload.urllib.request.urlopen",
return_value=resp,
) as urlopen:
put_bundle("https://bucket.s3.amazonaws.com/uploads/x.json.gz?sig", data)
req = urlopen.call_args[0][0]
assert req.method == "PUT"
# PUT body must be the bundle bytes, unchanged.
assert req.data == data
assert req.headers["Content-type"] == "application/gzip"
def test_custom_content_type(self):
from hermes_cli.diagnostics_upload import put_bundle
resp = _resp(status=204, body=b"")
with patch(
"hermes_cli.diagnostics_upload.urllib.request.urlopen",
return_value=resp,
) as urlopen:
put_bundle("https://u", b"data", content_type="application/json")
req = urlopen.call_args[0][0]
assert req.headers["Content-type"] == "application/json"
def test_non_2xx_raises(self):
from hermes_cli.diagnostics_upload import put_bundle
resp = _resp(status=403, body=b"AccessDenied")
with patch(
"hermes_cli.diagnostics_upload.urllib.request.urlopen",
return_value=resp,
):
with pytest.raises(RuntimeError):
put_bundle("https://u", b"data")
def test_http_error_propagates(self):
from hermes_cli.diagnostics_upload import put_bundle
err = urllib.error.HTTPError("https://u", 500, "err", {}, io.BytesIO(b""))
with patch(
"hermes_cli.diagnostics_upload.urllib.request.urlopen",
side_effect=err,
):
with pytest.raises(urllib.error.HTTPError):
put_bundle("https://u", b"data")
# ---------------------------------------------------------------------------
# share_to_nous (orchestration)
# ---------------------------------------------------------------------------
class TestShareToNous:
def test_orchestrates_request_then_put(self):
from hermes_cli import diagnostics_upload as mod
info = {
"id": "id-9",
"uploadUrl": "https://bucket/uploads/id-9.json.gz?sig",
"viewUrl": "https://support/diagnostics/id-9",
"expiresAt": "2026-06-20T00:00:00Z",
}
blob = b"gzipped-bundle"
with patch.object(mod, "request_upload_url", return_value=info) as req, \
patch.object(mod, "put_bundle") as put:
result = mod.share_to_nous(blob)
assert result == info
req.assert_called_once()
# request was told the real byte size (NAS signs it into ContentLength)
assert req.call_args.kwargs["size_bytes"] == len(blob)
# PUT got the signed URL + the exact blob
put.assert_called_once_with(
info["uploadUrl"], blob, content_type="application/gzip"
)
def test_put_failure_propagates(self):
from hermes_cli import diagnostics_upload as mod
info = {"id": "id-9", "uploadUrl": "https://u", "viewUrl": "v"}
with patch.object(mod, "request_upload_url", return_value=info), \
patch.object(mod, "put_bundle", side_effect=RuntimeError("PUT failed")):
with pytest.raises(RuntimeError):
mod.share_to_nous(b"data")
def test_share_succeeds_without_id_in_response(self):
from hermes_cli import diagnostics_upload as mod
# NAS is stateless and there is no confirm step, so the share must
# succeed regardless of whether the response carries an ``id``.
info = {"uploadUrl": "https://u", "viewUrl": "v"} # no id
with patch.object(mod, "request_upload_url", return_value=info), \
patch.object(mod, "put_bundle") as put:
result = mod.share_to_nous(b"data")
assert result == info
put.assert_called_once()