File size: 37,079 Bytes

1d733c0

"""Tests for ``hermes debug`` CLI command and debug utilities."""

import os
import sys
import urllib.error
from pathlib import Path
from unittest.mock import MagicMock, patch, call

import pytest

# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------

@pytest.fixture
def hermes_home(tmp_path, monkeypatch):
    """Set up an isolated HERMES_HOME with minimal logs."""
    home = tmp_path / ".hermes"
    home.mkdir()
    monkeypatch.setenv("HERMES_HOME", str(home))

    # Create log files
    logs_dir = home / "logs"
    logs_dir.mkdir()
    (logs_dir / "agent.log").write_text(
        "2026-04-12 17:00:00 INFO agent: session started\n"
        "2026-04-12 17:00:01 INFO tools.terminal: running ls\n"
        "2026-04-12 17:00:02 WARNING agent: high token usage\n"
    )
    (logs_dir / "errors.log").write_text(
        "2026-04-12 17:00:05 ERROR gateway.run: connection lost\n"
    )
    (logs_dir / "gateway.log").write_text(
        "2026-04-12 17:00:10 INFO gateway.run: started\n"
    )

    return home


# ---------------------------------------------------------------------------
# Unit tests for upload helpers
# ---------------------------------------------------------------------------

class TestUploadPasteRs:
    """Test paste.rs upload path."""

    def test_upload_paste_rs_success(self):
        from hermes_cli.debug import _upload_paste_rs

        mock_resp = MagicMock()
        mock_resp.read.return_value = b"https://paste.rs/abc123\n"
        mock_resp.__enter__ = lambda s: s
        mock_resp.__exit__ = MagicMock(return_value=False)

        with patch("hermes_cli.debug.urllib.request.urlopen", return_value=mock_resp):
            url = _upload_paste_rs("hello world")

        assert url == "https://paste.rs/abc123"

    def test_upload_paste_rs_bad_response(self):
        from hermes_cli.debug import _upload_paste_rs

        mock_resp = MagicMock()
        mock_resp.read.return_value = b"<html>error</html>"
        mock_resp.__enter__ = lambda s: s
        mock_resp.__exit__ = MagicMock(return_value=False)

        with patch("hermes_cli.debug.urllib.request.urlopen", return_value=mock_resp):
            with pytest.raises(ValueError, match="Unexpected response"):
                _upload_paste_rs("test")

    def test_upload_paste_rs_network_error(self):
        from hermes_cli.debug import _upload_paste_rs

        with patch(
            "hermes_cli.debug.urllib.request.urlopen",
            side_effect=urllib.error.URLError("connection refused"),
        ):
            with pytest.raises(urllib.error.URLError):
                _upload_paste_rs("test")


class TestUploadDpasteCom:
    """Test dpaste.com fallback upload path."""

    def test_upload_dpaste_com_success(self):
        from hermes_cli.debug import _upload_dpaste_com

        mock_resp = MagicMock()
        mock_resp.read.return_value = b"https://dpaste.com/ABCDEFG\n"
        mock_resp.__enter__ = lambda s: s
        mock_resp.__exit__ = MagicMock(return_value=False)

        with patch("hermes_cli.debug.urllib.request.urlopen", return_value=mock_resp):
            url = _upload_dpaste_com("hello world", expiry_days=7)

        assert url == "https://dpaste.com/ABCDEFG"


class TestUploadToPastebin:
    """Test the combined upload with fallback."""

    def test_tries_paste_rs_first(self):
        from hermes_cli.debug import upload_to_pastebin

        with patch("hermes_cli.debug._upload_paste_rs",
                    return_value="https://paste.rs/test") as prs:
            url = upload_to_pastebin("content")

        assert url == "https://paste.rs/test"
        prs.assert_called_once()

    def test_falls_back_to_dpaste_com(self):
        from hermes_cli.debug import upload_to_pastebin

        with patch("hermes_cli.debug._upload_paste_rs",
                    side_effect=Exception("down")), \
             patch("hermes_cli.debug._upload_dpaste_com",
                    return_value="https://dpaste.com/TEST") as dp:
            url = upload_to_pastebin("content")

        assert url == "https://dpaste.com/TEST"
        dp.assert_called_once()

    def test_raises_when_both_fail(self):
        from hermes_cli.debug import upload_to_pastebin

        with patch("hermes_cli.debug._upload_paste_rs",
                    side_effect=Exception("err1")), \
             patch("hermes_cli.debug._upload_dpaste_com",
                    side_effect=Exception("err2")):
            with pytest.raises(RuntimeError, match="Failed to upload"):
                upload_to_pastebin("content")


# ---------------------------------------------------------------------------
# Log reading
# ---------------------------------------------------------------------------

class TestCaptureLogSnapshot:
    """Test _capture_log_snapshot for log reading and truncation."""

    def test_reads_small_file(self, hermes_home):
        from hermes_cli.debug import _capture_log_snapshot

        snap = _capture_log_snapshot("agent", tail_lines=10)
        assert snap.full_text is not None
        assert "session started" in snap.full_text
        assert "session started" in snap.tail_text

    def test_returns_none_for_missing(self, tmp_path, monkeypatch):
        home = tmp_path / ".hermes"
        home.mkdir()
        monkeypatch.setenv("HERMES_HOME", str(home))

        from hermes_cli.debug import _capture_log_snapshot
        snap = _capture_log_snapshot("agent", tail_lines=10)
        assert snap.full_text is None
        assert snap.tail_text == "(file not found)"

    def test_empty_primary_reports_file_empty(self, hermes_home):
        """Empty primary (no .1 fallback) surfaces as '(file empty)', not missing."""
        (hermes_home / "logs" / "agent.log").write_text("")

        from hermes_cli.debug import _capture_log_snapshot
        snap = _capture_log_snapshot("agent", tail_lines=10)
        assert snap.full_text is None
        assert snap.tail_text == "(file empty)"

    def test_race_truncate_after_resolve_reports_empty(self, hermes_home, monkeypatch):
        """If the log is truncated between resolve and stat, say 'empty', not 'missing'."""
        log_path = hermes_home / "logs" / "agent.log"
        from hermes_cli import debug

        monkeypatch.setattr(debug, "_resolve_log_path", lambda _name: log_path)
        log_path.write_text("")

        snap = debug._capture_log_snapshot("agent", tail_lines=10)
        assert snap.path == log_path
        assert snap.full_text is None
        assert snap.tail_text == "(file empty)"

    def test_truncates_large_file(self, hermes_home):
        """Files larger than max_bytes get tail-truncated."""
        from hermes_cli.debug import _capture_log_snapshot

        # Write a file larger than 1KB
        big_content = "x" * 100 + "\n"
        (hermes_home / "logs" / "agent.log").write_text(big_content * 200)

        snap = _capture_log_snapshot("agent", tail_lines=10, max_bytes=1024)
        assert snap.full_text is not None
        assert "truncated" in snap.full_text

    def test_keeps_first_line_when_truncation_on_boundary(self, hermes_home):
        """When truncation lands on a line boundary, keep the first full line."""
        from hermes_cli.debug import _capture_log_snapshot

        # File must exceed the initial chunk_size (8192) used by the
        # backward-reading loop so the truncation path actually fires.
        line = "A" * 99 + "\n"  # 100 bytes per line
        num_lines = 200  # 20000 bytes
        (hermes_home / "logs" / "agent.log").write_text(line * num_lines)

        # max_bytes = 1000 = 100 * 10 → cut at byte 20000 - 1000 = 19000,
        # and byte 19000 - 1 is '\n'.  Boundary hit → keep all 10 lines.
        snap = _capture_log_snapshot("agent", tail_lines=5, max_bytes=1000)
        assert snap.full_text is not None
        assert "truncated" in snap.full_text
        raw = snap.full_text.split("\n", 1)[1]
        kept = [l for l in raw.strip().splitlines() if l.startswith("A")]
        assert len(kept) == 10

    def test_drops_partial_when_truncation_mid_line(self, hermes_home):
        """When truncation lands mid-line, drop the partial fragment."""
        from hermes_cli.debug import _capture_log_snapshot

        line = "A" * 99 + "\n"  # 100 bytes per line
        num_lines = 200  # 20000 bytes
        (hermes_home / "logs" / "agent.log").write_text(line * num_lines)

        # max_bytes = 950 doesn't divide evenly into 100 → mid-line cut.
        snap = _capture_log_snapshot("agent", tail_lines=5, max_bytes=950)
        assert snap.full_text is not None
        assert "truncated" in snap.full_text
        raw = snap.full_text.split("\n", 1)[1]
        kept = [l for l in raw.strip().splitlines() if l.startswith("A")]
        # 950 / 100 = 9.5 → 9 complete lines after dropping partial
        assert len(kept) == 9

    def test_unknown_log_returns_none(self, hermes_home):
        from hermes_cli.debug import _capture_log_snapshot
        snap = _capture_log_snapshot("nonexistent", tail_lines=10)
        assert snap.full_text is None

    def test_falls_back_to_rotated_file(self, hermes_home):
        """When gateway.log doesn't exist, falls back to gateway.log.1."""
        from hermes_cli.debug import _capture_log_snapshot

        logs_dir = hermes_home / "logs"
        # Remove the primary (if any) and create a .1 rotation
        (logs_dir / "gateway.log").unlink(missing_ok=True)
        (logs_dir / "gateway.log.1").write_text(
            "2026-04-12 10:00:00 INFO gateway.run: rotated content\n"
        )

        snap = _capture_log_snapshot("gateway", tail_lines=10)
        assert snap.full_text is not None
        assert "rotated content" in snap.full_text

    def test_prefers_primary_over_rotated(self, hermes_home):
        """Primary log is used when it exists, even if .1 also exists."""
        from hermes_cli.debug import _capture_log_snapshot

        logs_dir = hermes_home / "logs"
        (logs_dir / "gateway.log").write_text("primary content\n")
        (logs_dir / "gateway.log.1").write_text("rotated content\n")

        snap = _capture_log_snapshot("gateway", tail_lines=10)
        assert "primary content" in snap.full_text
        assert "rotated" not in snap.full_text

    def test_falls_back_when_primary_empty(self, hermes_home):
        """Empty primary log falls back to .1 rotation."""
        from hermes_cli.debug import _capture_log_snapshot

        logs_dir = hermes_home / "logs"
        (logs_dir / "agent.log").write_text("")
        (logs_dir / "agent.log.1").write_text("rotated agent data\n")

        snap = _capture_log_snapshot("agent", tail_lines=10)
        assert snap.full_text is not None
        assert "rotated agent data" in snap.full_text


# ---------------------------------------------------------------------------
# Debug report collection
# ---------------------------------------------------------------------------

class TestCollectDebugReport:
    """Test the debug report builder."""

    def test_report_includes_dump_output(self, hermes_home):
        from hermes_cli.debug import collect_debug_report

        with patch("hermes_cli.dump.run_dump") as mock_dump:
            mock_dump.side_effect = lambda args: print(
                "--- hermes dump ---\nversion: 0.8.0\n--- end dump ---"
            )
            report = collect_debug_report(log_lines=50)

        assert "--- hermes dump ---" in report
        assert "version: 0.8.0" in report

    def test_report_includes_agent_log(self, hermes_home):
        from hermes_cli.debug import collect_debug_report

        with patch("hermes_cli.dump.run_dump"):
            report = collect_debug_report(log_lines=50)

        assert "--- agent.log" in report
        assert "session started" in report

    def test_report_includes_errors_log(self, hermes_home):
        from hermes_cli.debug import collect_debug_report

        with patch("hermes_cli.dump.run_dump"):
            report = collect_debug_report(log_lines=50)

        assert "--- errors.log" in report
        assert "connection lost" in report

    def test_report_includes_gateway_log(self, hermes_home):
        from hermes_cli.debug import collect_debug_report

        with patch("hermes_cli.dump.run_dump"):
            report = collect_debug_report(log_lines=50)

        assert "--- gateway.log" in report

    def test_missing_logs_handled(self, tmp_path, monkeypatch):
        home = tmp_path / ".hermes"
        home.mkdir()
        monkeypatch.setenv("HERMES_HOME", str(home))

        from hermes_cli.debug import collect_debug_report

        with patch("hermes_cli.dump.run_dump"):
            report = collect_debug_report(log_lines=50)

        assert "(file not found)" in report


# ---------------------------------------------------------------------------
# CLI entry point — run_debug_share
# ---------------------------------------------------------------------------

class TestRunDebugShare:
    """Test the run_debug_share CLI handler."""

    def test_share_sweeps_expired_pastes(self, hermes_home, capsys):
        """Slash-command path should sweep old pending deletes before uploading."""
        from hermes_cli.debug import run_debug_share

        args = MagicMock()
        args.lines = 50
        args.expire = 7
        args.local = False

        with patch("hermes_cli.dump.run_dump"), \
             patch("hermes_cli.debug._sweep_expired_pastes", return_value=(0, 0)) as mock_sweep, \
             patch("hermes_cli.debug.upload_to_pastebin",
                    return_value="https://paste.rs/test"):
            run_debug_share(args)

        mock_sweep.assert_called_once()
        assert "Debug report uploaded" in capsys.readouterr().out

    def test_share_survives_sweep_failure(self, hermes_home, capsys):
        """Expired-paste cleanup is best-effort and must not block sharing."""
        from hermes_cli.debug import run_debug_share

        args = MagicMock()
        args.lines = 50
        args.expire = 7
        args.local = False

        with patch("hermes_cli.dump.run_dump"), \
             patch(
                 "hermes_cli.debug._sweep_expired_pastes",
                 side_effect=RuntimeError("offline"),
             ), \
             patch("hermes_cli.debug.upload_to_pastebin",
                    return_value="https://paste.rs/test"):
            run_debug_share(args)

        assert "https://paste.rs/test" in capsys.readouterr().out

    def test_local_flag_prints_full_logs(self, hermes_home, capsys):
        """--local prints the report plus full log contents."""
        from hermes_cli.debug import run_debug_share

        args = MagicMock()
        args.lines = 50
        args.expire = 7
        args.local = True

        with patch("hermes_cli.dump.run_dump"):
            run_debug_share(args)

        out = capsys.readouterr().out
        assert "--- agent.log" in out
        assert "FULL agent.log" in out
        assert "FULL gateway.log" in out

    def test_share_uploads_three_pastes(self, hermes_home, capsys):
        """Successful share uploads report + agent.log + gateway.log."""
        from hermes_cli.debug import run_debug_share

        args = MagicMock()
        args.lines = 50
        args.expire = 7
        args.local = False

        call_count = [0]
        uploaded_content = []
        def _mock_upload(content, expiry_days=7):
            call_count[0] += 1
            uploaded_content.append(content)
            return f"https://paste.rs/paste{call_count[0]}"

        with patch("hermes_cli.dump.run_dump") as mock_dump, \
             patch("hermes_cli.debug.upload_to_pastebin",
                    side_effect=_mock_upload):
            mock_dump.side_effect = lambda a: print("--- hermes dump ---\nversion: test\n--- end dump ---")
            run_debug_share(args)

        out = capsys.readouterr().out
        # Should have 3 uploads: report, agent.log, gateway.log
        assert call_count[0] == 3
        assert "paste.rs/paste1" in out  # Report
        assert "paste.rs/paste2" in out  # agent.log
        assert "paste.rs/paste3" in out  # gateway.log
        assert "Report" in out
        assert "agent.log" in out
        assert "gateway.log" in out

        # Each log paste should start with the dump header
        agent_paste = uploaded_content[1]
        assert "--- hermes dump ---" in agent_paste
        assert "--- full agent.log ---" in agent_paste
        gateway_paste = uploaded_content[2]
        assert "--- hermes dump ---" in gateway_paste
        assert "--- full gateway.log ---" in gateway_paste

    def test_share_keeps_report_and_full_log_on_same_snapshot(self, hermes_home, capsys):
        """A mid-run rotation must not make full agent.log older than the report."""
        from hermes_cli.debug import run_debug_share, collect_debug_report as real_collect_debug_report

        logs_dir = hermes_home / "logs"
        (logs_dir / "agent.log").write_text(
            "2026-04-22 12:00:00 INFO agent: newest line\n"
        )
        (logs_dir / "agent.log.1").write_text(
            "2026-04-10 12:00:00 INFO agent: old rotated line\n"
        )

        args = MagicMock()
        args.lines = 50
        args.expire = 7
        args.local = False

        uploaded_content = []

        def _mock_upload(content, expiry_days=7):
            uploaded_content.append(content)
            return f"https://paste.rs/paste{len(uploaded_content)}"

        def _wrapped_collect_debug_report(*, log_lines=200, dump_text="", log_snapshots=None):
            report = real_collect_debug_report(
                log_lines=log_lines,
                dump_text=dump_text,
                log_snapshots=log_snapshots,
            )
            # Simulate the live log rotating after the report is built but
            # before the old implementation would have re-read agent.log for
            # standalone upload.
            (logs_dir / "agent.log").write_text("")
            (logs_dir / "agent.log.1").write_text(
                "2026-04-10 12:00:00 INFO agent: old rotated line\n"
            )
            return report

        with patch("hermes_cli.dump.run_dump"), \
             patch("hermes_cli.debug.collect_debug_report", side_effect=_wrapped_collect_debug_report), \
             patch("hermes_cli.debug.upload_to_pastebin", side_effect=_mock_upload):
            run_debug_share(args)

        report_paste = uploaded_content[0]
        agent_paste = uploaded_content[1]
        assert "2026-04-22 12:00:00 INFO agent: newest line" in report_paste
        assert "2026-04-22 12:00:00 INFO agent: newest line" in agent_paste
        assert "old rotated line" not in agent_paste

    def test_share_skips_missing_logs(self, tmp_path, monkeypatch, capsys):
        """Only uploads logs that exist."""
        home = tmp_path / ".hermes"
        home.mkdir()
        monkeypatch.setenv("HERMES_HOME", str(home))

        from hermes_cli.debug import run_debug_share

        args = MagicMock()
        args.lines = 50
        args.expire = 7
        args.local = False

        call_count = [0]
        def _mock_upload(content, expiry_days=7):
            call_count[0] += 1
            return f"https://paste.rs/paste{call_count[0]}"

        with patch("hermes_cli.dump.run_dump"), \
             patch("hermes_cli.debug.upload_to_pastebin",
                    side_effect=_mock_upload):
            run_debug_share(args)

        out = capsys.readouterr().out
        # Only the report should be uploaded (no log files exist)
        assert call_count[0] == 1
        assert "Report" in out

    def test_share_continues_on_log_upload_failure(self, hermes_home, capsys):
        """Log upload failure doesn't stop the report from being shared."""
        from hermes_cli.debug import run_debug_share

        args = MagicMock()
        args.lines = 50
        args.expire = 7
        args.local = False

        call_count = [0]
        def _mock_upload(content, expiry_days=7):
            call_count[0] += 1
            if call_count[0] > 1:
                raise RuntimeError("upload failed")
            return "https://paste.rs/report"

        with patch("hermes_cli.dump.run_dump"), \
             patch("hermes_cli.debug.upload_to_pastebin",
                    side_effect=_mock_upload):
            run_debug_share(args)

        out = capsys.readouterr().out
        assert "Report" in out
        assert "paste.rs/report" in out
        assert "failed to upload" in out

    def test_share_exits_on_report_upload_failure(self, hermes_home, capsys):
        """If the main report fails to upload, exit with code 1."""
        from hermes_cli.debug import run_debug_share

        args = MagicMock()
        args.lines = 50
        args.expire = 7
        args.local = False

        with patch("hermes_cli.dump.run_dump"), \
             patch("hermes_cli.debug.upload_to_pastebin",
                    side_effect=RuntimeError("all failed")):
            with pytest.raises(SystemExit) as exc_info:
                run_debug_share(args)

        assert exc_info.value.code == 1
        out = capsys.readouterr()
        assert "all failed" in out.err


# ---------------------------------------------------------------------------
# run_debug router
# ---------------------------------------------------------------------------

class TestRunDebug:
    def test_no_subcommand_shows_usage(self, capsys):
        from hermes_cli.debug import run_debug

        args = MagicMock()
        args.debug_command = None

        run_debug(args)

        out = capsys.readouterr().out
        assert "hermes debug" in out
        assert "share" in out
        assert "delete" in out

    def test_share_subcommand_routes(self, hermes_home):
        from hermes_cli.debug import run_debug

        args = MagicMock()
        args.debug_command = "share"
        args.lines = 200
        args.expire = 7
        args.local = True

        with patch("hermes_cli.dump.run_dump"):
            run_debug(args)


# ---------------------------------------------------------------------------
# Argparse integration
# ---------------------------------------------------------------------------

# ---------------------------------------------------------------------------
# Delete / auto-delete
# ---------------------------------------------------------------------------

class TestExtractPasteId:
    def test_paste_rs_url(self):
        from hermes_cli.debug import _extract_paste_id
        assert _extract_paste_id("https://paste.rs/abc123") == "abc123"

    def test_paste_rs_trailing_slash(self):
        from hermes_cli.debug import _extract_paste_id
        assert _extract_paste_id("https://paste.rs/abc123/") == "abc123"

    def test_http_variant(self):
        from hermes_cli.debug import _extract_paste_id
        assert _extract_paste_id("http://paste.rs/xyz") == "xyz"

    def test_non_paste_rs_returns_none(self):
        from hermes_cli.debug import _extract_paste_id
        assert _extract_paste_id("https://dpaste.com/ABCDEF") is None

    def test_empty_returns_none(self):
        from hermes_cli.debug import _extract_paste_id
        assert _extract_paste_id("") is None


class TestDeletePaste:
    def test_delete_sends_delete_request(self):
        from hermes_cli.debug import delete_paste

        mock_resp = MagicMock()
        mock_resp.status = 200
        mock_resp.__enter__ = lambda s: s
        mock_resp.__exit__ = MagicMock(return_value=False)

        with patch("hermes_cli.debug.urllib.request.urlopen",
                    return_value=mock_resp) as mock_open:
            result = delete_paste("https://paste.rs/abc123")

        assert result is True
        req = mock_open.call_args[0][0]
        assert req.method == "DELETE"
        assert "paste.rs/abc123" in req.full_url

    def test_delete_rejects_non_paste_rs(self):
        from hermes_cli.debug import delete_paste

        with pytest.raises(ValueError, match="only paste.rs"):
            delete_paste("https://dpaste.com/something")


class TestScheduleAutoDelete:
    """``_schedule_auto_delete`` used to spawn a detached Python subprocess
    per call (one per paste URL batch).  Those subprocesses slept 6 hours
    and accumulated forever under repeated use — 15+ orphaned interpreters
    were observed in production.

    The new implementation is stateless: it records pending deletions to
    ``~/.hermes/pastes/pending.json`` and lets ``_sweep_expired_pastes``
    handle the DELETE requests synchronously on the next ``hermes debug``
    invocation.
    """

    def test_does_not_spawn_subprocess(self, hermes_home):
        """Regression guard: _schedule_auto_delete must NEVER spawn subprocesses.

        We assert this structurally rather than by mocking Popen: the new
        implementation doesn't even import ``subprocess`` at module scope,
        so a mock patch wouldn't find it.
        """
        import ast
        import inspect
        from hermes_cli.debug import _schedule_auto_delete

        # Strip the docstring before scanning so the regression-rationale
        # prose inside it doesn't trigger our banned-word checks.
        source = inspect.getsource(_schedule_auto_delete)
        tree = ast.parse(source)
        func_node = tree.body[0]
        if (
            func_node.body
            and isinstance(func_node.body[0], ast.Expr)
            and isinstance(func_node.body[0].value, ast.Constant)
            and isinstance(func_node.body[0].value.value, str)
        ):
            func_node.body = func_node.body[1:]
        code_only = ast.unparse(func_node)

        assert "Popen" not in code_only, (
            "_schedule_auto_delete must not spawn subprocesses — "
            "use pending.json + _sweep_expired_pastes instead"
        )
        assert "subprocess" not in code_only, (
            "_schedule_auto_delete must not reference subprocess at all"
        )
        assert "time.sleep" not in code_only, (
            "Regression: sleeping in _schedule_auto_delete is the bug being fixed"
        )

        # And verify that calling it doesn't produce any orphaned children
        # (it should just write pending.json synchronously).
        import os as _os
        before = set(_os.listdir("/proc")) if _os.path.exists("/proc") else None
        _schedule_auto_delete(
            ["https://paste.rs/abc", "https://paste.rs/def"],
            delay_seconds=10,
        )
        if before is not None:
            after = set(_os.listdir("/proc"))
            new = after - before
            # Filter to only integer-named entries (process PIDs)
            new_pids = [p for p in new if p.isdigit()]
            # It's fine if unrelated processes appeared — we just need to make
            # sure we didn't spawn a long-sleeping one.  The old bug spawned
            # a python interpreter whose cmdline contained "time.sleep".
            for pid in new_pids:
                try:
                    with open(f"/proc/{pid}/cmdline", "rb") as f:
                        cmdline = f.read().decode("utf-8", errors="replace")
                    assert "time.sleep" not in cmdline, (
                        f"Leaked sleeper subprocess PID {pid}: {cmdline}"
                    )
                except OSError:
                    pass  # process exited already

    def test_records_pending_to_json(self, hermes_home):
        """Scheduled URLs are persisted to pending.json with expiration."""
        from hermes_cli.debug import _schedule_auto_delete, _pending_file
        import json

        _schedule_auto_delete(
            ["https://paste.rs/abc", "https://paste.rs/def"],
            delay_seconds=10,
        )

        pending_path = _pending_file()
        assert pending_path.exists()

        entries = json.loads(pending_path.read_text())
        assert len(entries) == 2
        urls = {e["url"] for e in entries}
        assert urls == {"https://paste.rs/abc", "https://paste.rs/def"}

        # expire_at is ~now + delay_seconds
        import time
        for e in entries:
            assert e["expire_at"] > time.time()
            assert e["expire_at"] <= time.time() + 15

    def test_skips_non_paste_rs_urls(self, hermes_home):
        """dpaste.com URLs auto-expire — don't track them."""
        from hermes_cli.debug import _schedule_auto_delete, _pending_file

        _schedule_auto_delete(["https://dpaste.com/something"])

        # pending.json should not be created for non-paste.rs URLs
        assert not _pending_file().exists()

    def test_merges_with_existing_pending(self, hermes_home):
        """Subsequent calls merge into existing pending.json."""
        from hermes_cli.debug import _schedule_auto_delete, _load_pending

        _schedule_auto_delete(["https://paste.rs/first"], delay_seconds=10)
        _schedule_auto_delete(["https://paste.rs/second"], delay_seconds=10)

        entries = _load_pending()
        urls = {e["url"] for e in entries}
        assert urls == {"https://paste.rs/first", "https://paste.rs/second"}

    def test_dedupes_same_url(self, hermes_home):
        """Same URL recorded twice → one entry with the later expire_at."""
        from hermes_cli.debug import _schedule_auto_delete, _load_pending

        _schedule_auto_delete(["https://paste.rs/dup"], delay_seconds=10)
        _schedule_auto_delete(["https://paste.rs/dup"], delay_seconds=100)

        entries = _load_pending()
        assert len(entries) == 1
        assert entries[0]["url"] == "https://paste.rs/dup"


class TestSweepExpiredPastes:
    """Test the opportunistic sweep that replaces the sleeping subprocess."""

    def test_sweep_empty_is_noop(self, hermes_home):
        from hermes_cli.debug import _sweep_expired_pastes

        deleted, remaining = _sweep_expired_pastes()
        assert deleted == 0
        assert remaining == 0

    def test_sweep_deletes_expired_entries(self, hermes_home):
        from hermes_cli.debug import (
            _sweep_expired_pastes,
            _save_pending,
            _load_pending,
        )
        import time

        # Seed pending.json with one expired + one future entry
        _save_pending([
            {"url": "https://paste.rs/expired", "expire_at": time.time() - 100},
            {"url": "https://paste.rs/future", "expire_at": time.time() + 3600},
        ])

        delete_calls = []

        def fake_delete(url):
            delete_calls.append(url)
            return True

        with patch("hermes_cli.debug.delete_paste", side_effect=fake_delete):
            deleted, remaining = _sweep_expired_pastes()

        assert delete_calls == ["https://paste.rs/expired"]
        assert deleted == 1
        assert remaining == 1

        entries = _load_pending()
        urls = {e["url"] for e in entries}
        assert urls == {"https://paste.rs/future"}

    def test_sweep_leaves_future_entries_alone(self, hermes_home):
        from hermes_cli.debug import _sweep_expired_pastes, _save_pending
        import time

        _save_pending([
            {"url": "https://paste.rs/future1", "expire_at": time.time() + 3600},
            {"url": "https://paste.rs/future2", "expire_at": time.time() + 7200},
        ])

        with patch("hermes_cli.debug.delete_paste") as mock_delete:
            deleted, remaining = _sweep_expired_pastes()

        mock_delete.assert_not_called()
        assert deleted == 0
        assert remaining == 2

    def test_sweep_survives_network_failure(self, hermes_home):
        """Failed DELETEs stay in pending.json until the 24h grace window."""
        from hermes_cli.debug import (
            _sweep_expired_pastes,
            _save_pending,
            _load_pending,
        )
        import time

        _save_pending([
            {"url": "https://paste.rs/flaky", "expire_at": time.time() - 100},
        ])

        with patch(
            "hermes_cli.debug.delete_paste",
            side_effect=Exception("network down"),
        ):
            deleted, remaining = _sweep_expired_pastes()

        # Failure within 24h grace → kept for retry
        assert deleted == 0
        assert remaining == 1
        assert len(_load_pending()) == 1

    def test_sweep_drops_entries_past_grace_window(self, hermes_home):
        """After 24h past expiration, give up even on network failures."""
        from hermes_cli.debug import (
            _sweep_expired_pastes,
            _save_pending,
            _load_pending,
        )
        import time

        # Expired 25 hours ago → past the 24h grace window
        very_old = time.time() - (25 * 3600)
        _save_pending([
            {"url": "https://paste.rs/ancient", "expire_at": very_old},
        ])

        with patch(
            "hermes_cli.debug.delete_paste",
            side_effect=Exception("network down"),
        ):
            deleted, remaining = _sweep_expired_pastes()

        assert deleted == 1
        assert remaining == 0
        assert _load_pending() == []


class TestRunDebugSweepsOnInvocation:
    """``run_debug`` must sweep expired pastes on every invocation."""

    def test_run_debug_calls_sweep(self, hermes_home):
        from hermes_cli.debug import run_debug

        args = MagicMock()
        args.debug_command = None  # default → prints help

        with patch("hermes_cli.debug._sweep_expired_pastes") as mock_sweep:
            run_debug(args)

        mock_sweep.assert_called_once()

    def test_run_debug_survives_sweep_failure(self, hermes_home, capsys):
        """If the sweep throws, the subcommand still runs."""
        from hermes_cli.debug import run_debug

        args = MagicMock()
        args.debug_command = None

        with patch(
            "hermes_cli.debug._sweep_expired_pastes",
            side_effect=RuntimeError("boom"),
        ):
            run_debug(args)  # must not raise

        # Default subcommand still printed help
        out = capsys.readouterr().out
        assert "Usage: hermes debug" in out


class TestRunDebugDelete:
    def test_deletes_valid_url(self, capsys):
        from hermes_cli.debug import run_debug_delete

        args = MagicMock()
        args.urls = ["https://paste.rs/abc"]

        with patch("hermes_cli.debug.delete_paste", return_value=True):
            run_debug_delete(args)

        out = capsys.readouterr().out
        assert "Deleted" in out
        assert "paste.rs/abc" in out

    def test_handles_delete_failure(self, capsys):
        from hermes_cli.debug import run_debug_delete

        args = MagicMock()
        args.urls = ["https://paste.rs/abc"]

        with patch("hermes_cli.debug.delete_paste",
                    side_effect=Exception("network error")):
            run_debug_delete(args)

        out = capsys.readouterr().out
        assert "Could not delete" in out

    def test_no_urls_shows_usage(self, capsys):
        from hermes_cli.debug import run_debug_delete

        args = MagicMock()
        args.urls = []

        run_debug_delete(args)

        out = capsys.readouterr().out
        assert "Usage" in out


class TestShareIncludesAutoDelete:
    """Verify that run_debug_share schedules auto-deletion and prints TTL."""

    def test_share_schedules_auto_delete(self, hermes_home, capsys):
        from hermes_cli.debug import run_debug_share

        args = MagicMock()
        args.lines = 50
        args.expire = 7
        args.local = False

        with patch("hermes_cli.dump.run_dump"), \
             patch("hermes_cli.debug.upload_to_pastebin",
                    return_value="https://paste.rs/test1"), \
             patch("hermes_cli.debug._schedule_auto_delete") as mock_sched:
            run_debug_share(args)

        # auto-delete was scheduled with the uploaded URLs
        mock_sched.assert_called_once()
        urls_arg = mock_sched.call_args[0][0]
        assert "https://paste.rs/test1" in urls_arg

        out = capsys.readouterr().out
        assert "auto-delete" in out

    def test_share_shows_privacy_notice(self, hermes_home, capsys):
        from hermes_cli.debug import run_debug_share

        args = MagicMock()
        args.lines = 50
        args.expire = 7
        args.local = False

        with patch("hermes_cli.dump.run_dump"), \
             patch("hermes_cli.debug.upload_to_pastebin",
                    return_value="https://paste.rs/test"), \
             patch("hermes_cli.debug._schedule_auto_delete"):
            run_debug_share(args)

        out = capsys.readouterr().out
        assert "public paste service" in out

    def test_local_no_privacy_notice(self, hermes_home, capsys):
        from hermes_cli.debug import run_debug_share

        args = MagicMock()
        args.lines = 50
        args.expire = 7
        args.local = True

        with patch("hermes_cli.dump.run_dump"):
            run_debug_share(args)

        out = capsys.readouterr().out
        assert "public paste service" not in out