common/hermes
2
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

feat: centralized logging, instrumentation, hermes logs CLI, gateway noise fix (#5430)

Browse Source 
Adds comprehensive logging infrastructure to Hermes Agent across 4 phases:

**Phase 1 — Centralized logging**
- New hermes_logging.py with idempotent setup_logging() used by CLI, gateway, and cron
- agent.log (INFO+) and errors.log (WARNING+) with RotatingFileHandler + RedactingFormatter
- config.yaml logging: section (level, max_size_mb, backup_count)
- All entry points wired (cli.py, main.py, gateway/run.py, run_agent.py)
- Fixed debug_helpers.py writing to ./logs/ instead of ~/.hermes/logs/

**Phase 2 — Event instrumentation**
- API calls: model, provider, tokens, latency, cache hit %
- Tool execution: name, duration, result size (both sequential + concurrent)
- Session lifecycle: turn start (session/model/provider/platform), compression (before/after)
- Credential pool: rotation events, exhaustion tracking

**Phase 3 — hermes logs CLI command**
- hermes logs / hermes logs -f / hermes logs errors / hermes logs gateway
- --level, --session, --since filters
- hermes logs list (file sizes + ages)

**Phase 4 — Gateway bug fix + noise reduction**
- fix: _async_flush_memories() called with wrong arg count — sessions never flushed
- Batched session expiry logs: 6 lines/cycle → 2 summary lines
- Added inbound message + response time logging

75 new tests, zero regressions on the full suite.
This commit is contained in:
Teknium
2026-04-06 00:08:20 -07:00
committed by GitHub
parent 89db3aeb2c
commit 9c96f669a1
11 changed files with 1399 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files

288
tests/hermes_cli/test_logs.py Normal file
View File

@@ -0,0 +1,288 @@
"""Tests for hermes_cli/logs.py — log viewing and filtering."""
import os
import textwrap
from datetime import datetime, timedelta
from io import StringIO
from pathlib import Path
from unittest.mock import patch
import pytest
from hermes_cli.logs import (
LOG_FILES,
_extract_level,
_matches_filters,
_parse_line_timestamp,
_parse_since,
_read_last_n_lines,
list_logs,
tail_log,
)
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def log_dir(tmp_path, monkeypatch):
"""Create a fake HERMES_HOME with a logs/ directory."""
home = Path(os.environ["HERMES_HOME"])
logs = home / "logs"
logs.mkdir(parents=True, exist_ok=True)
return logs
@pytest.fixture
def sample_agent_log(log_dir):
"""Write a realistic agent.log with mixed levels and sessions."""
lines = textwrap.dedent("""\
2026-04-05 10:00:00,000 INFO run_agent: conversation turn: session=sess_aaa model=claude provider=openrouter platform=cli history=0 msg='hello'
2026-04-05 10:00:01,000 INFO run_agent: tool terminal completed (0.50s, 200 chars)
2026-04-05 10:00:02,000 INFO run_agent: API call #1: model=claude provider=openrouter in=1000 out=200 total=1200 latency=1.5s
2026-04-05 10:00:03,000 WARNING run_agent: Tool web_search returned error (2.00s): timeout
2026-04-05 10:00:04,000 INFO run_agent: conversation turn: session=sess_bbb model=gpt-5 provider=openai platform=telegram history=5 msg='fix bug'
2026-04-05 10:00:05,000 ERROR run_agent: API call failed after 3 retries. rate limited
2026-04-05 10:00:06,000 INFO run_agent: tool read_file completed (0.01s, 500 chars)
2026-04-05 10:00:07,000 DEBUG run_agent: verbose internal detail
2026-04-05 10:00:08,000 INFO credential_pool: credential pool: marking key-1 exhausted (status=429), rotating
2026-04-05 10:00:09,000 INFO credential_pool: credential pool: rotated to key-2
""")
path = log_dir / "agent.log"
path.write_text(lines)
return path
@pytest.fixture
def sample_errors_log(log_dir):
"""Write a small errors.log."""
lines = textwrap.dedent("""\
2026-04-05 10:00:03,000 WARNING run_agent: Tool web_search returned error (2.00s): timeout
2026-04-05 10:00:05,000 ERROR run_agent: API call failed after 3 retries. rate limited
""")
path = log_dir / "errors.log"
path.write_text(lines)
return path
# ---------------------------------------------------------------------------
# _parse_since
# ---------------------------------------------------------------------------
class TestParseSince:
def test_hours(self):
cutoff = _parse_since("2h")
assert cutoff is not None
assert (datetime.now() - cutoff).total_seconds() == pytest.approx(7200, abs=5)
def test_minutes(self):
cutoff = _parse_since("30m")
assert cutoff is not None
assert (datetime.now() - cutoff).total_seconds() == pytest.approx(1800, abs=5)
def test_days(self):
cutoff = _parse_since("1d")
assert cutoff is not None
assert (datetime.now() - cutoff).total_seconds() == pytest.approx(86400, abs=5)
def test_seconds(self):
cutoff = _parse_since("60s")
assert cutoff is not None
assert (datetime.now() - cutoff).total_seconds() == pytest.approx(60, abs=5)
def test_invalid_returns_none(self):
assert _parse_since("abc") is None
assert _parse_since("") is None
assert _parse_since("10x") is None
def test_whitespace_handling(self):
cutoff = _parse_since(" 1h ")
assert cutoff is not None
# ---------------------------------------------------------------------------
# _parse_line_timestamp
# ---------------------------------------------------------------------------
class TestParseLineTimestamp:
def test_standard_format(self):
ts = _parse_line_timestamp("2026-04-05 10:00:00,123 INFO something")
assert ts is not None
assert ts.year == 2026
assert ts.hour == 10
def test_no_timestamp(self):
assert _parse_line_timestamp("just some text") is None
def test_continuation_line(self):
assert _parse_line_timestamp(" at module.function (line 42)") is None
# ---------------------------------------------------------------------------
# _extract_level
# ---------------------------------------------------------------------------
class TestExtractLevel:
def test_info(self):
assert _extract_level("2026-04-05 10:00:00 INFO run_agent: something") == "INFO"
def test_warning(self):
assert _extract_level("2026-04-05 10:00:00 WARNING run_agent: bad") == "WARNING"
def test_error(self):
assert _extract_level("2026-04-05 10:00:00 ERROR run_agent: crash") == "ERROR"
def test_debug(self):
assert _extract_level("2026-04-05 10:00:00 DEBUG run_agent: detail") == "DEBUG"
def test_no_level(self):
assert _extract_level("just a plain line") is None
# ---------------------------------------------------------------------------
# _matches_filters
# ---------------------------------------------------------------------------
class TestMatchesFilters:
def test_no_filters_always_matches(self):
assert _matches_filters("any line") is True
def test_level_filter_passes(self):
assert _matches_filters(
"2026-04-05 10:00:00 WARNING something",
min_level="WARNING",
) is True
def test_level_filter_rejects(self):
assert _matches_filters(
"2026-04-05 10:00:00 INFO something",
min_level="WARNING",
) is False
def test_session_filter_passes(self):
assert _matches_filters(
"session=sess_aaa model=claude",
session_filter="sess_aaa",
) is True
def test_session_filter_rejects(self):
assert _matches_filters(
"session=sess_aaa model=claude",
session_filter="sess_bbb",
) is False
def test_since_filter_passes(self):
# Line from the future should always pass
assert _matches_filters(
"2099-01-01 00:00:00 INFO future",
since=datetime.now(),
) is True
def test_since_filter_rejects(self):
assert _matches_filters(
"2020-01-01 00:00:00 INFO past",
since=datetime.now(),
) is False
def test_combined_filters(self):
line = "2099-01-01 00:00:00 WARNING run_agent: session=abc error"
assert _matches_filters(
line, min_level="WARNING", session_filter="abc",
since=datetime.now(),
) is True
# Fails session filter
assert _matches_filters(
line, min_level="WARNING", session_filter="xyz",
) is False
# ---------------------------------------------------------------------------
# _read_last_n_lines
# ---------------------------------------------------------------------------
class TestReadLastNLines:
def test_reads_correct_count(self, sample_agent_log):
lines = _read_last_n_lines(sample_agent_log, 3)
assert len(lines) == 3
def test_reads_all_when_fewer(self, sample_agent_log):
lines = _read_last_n_lines(sample_agent_log, 100)
assert len(lines) == 10 # sample has 10 lines
def test_empty_file(self, log_dir):
empty = log_dir / "empty.log"
empty.write_text("")
lines = _read_last_n_lines(empty, 10)
assert lines == []
def test_last_line_content(self, sample_agent_log):
lines = _read_last_n_lines(sample_agent_log, 1)
assert "rotated to key-2" in lines[0]
# ---------------------------------------------------------------------------
# tail_log
# ---------------------------------------------------------------------------
class TestTailLog:
def test_basic_tail(self, sample_agent_log, capsys):
tail_log("agent", num_lines=3)
captured = capsys.readouterr()
assert "agent.log" in captured.out
# Should have the header + 3 lines
lines = captured.out.strip().split("\n")
assert len(lines) == 4 # 1 header + 3 content
def test_level_filter(self, sample_agent_log, capsys):
tail_log("agent", num_lines=50, level="ERROR")
captured = capsys.readouterr()
assert "level>=ERROR" in captured.out
# Only the ERROR line should appear
content_lines = [l for l in captured.out.strip().split("\n") if not l.startswith("---")]
assert len(content_lines) == 1
assert "API call failed" in content_lines[0]
def test_session_filter(self, sample_agent_log, capsys):
tail_log("agent", num_lines=50, session="sess_bbb")
captured = capsys.readouterr()
content_lines = [l for l in captured.out.strip().split("\n") if not l.startswith("---")]
assert len(content_lines) == 1
assert "sess_bbb" in content_lines[0]
def test_errors_log(self, sample_errors_log, capsys):
tail_log("errors", num_lines=10)
captured = capsys.readouterr()
assert "errors.log" in captured.out
assert "WARNING" in captured.out or "ERROR" in captured.out
def test_unknown_log_exits(self):
with pytest.raises(SystemExit):
tail_log("nonexistent")
def test_missing_file_exits(self, log_dir):
with pytest.raises(SystemExit):
tail_log("agent") # agent.log doesn't exist in clean log_dir
# ---------------------------------------------------------------------------
# list_logs
# ---------------------------------------------------------------------------
class TestListLogs:
def test_lists_files(self, sample_agent_log, sample_errors_log, capsys):
list_logs()
captured = capsys.readouterr()
assert "agent.log" in captured.out
assert "errors.log" in captured.out
def test_empty_dir(self, log_dir, capsys):
list_logs()
captured = capsys.readouterr()
assert "no log files yet" in captured.out
def test_shows_sizes(self, sample_agent_log, capsys):
list_logs()
captured = capsys.readouterr()
# File is small, should show as bytes or KB
assert "B" in captured.out or "KB" in captured.out

314
tests/test_hermes_logging.py Normal file
View File

@@ -0,0 +1,314 @@
"""Tests for hermes_logging — centralized logging setup."""
import logging
import os
from logging.handlers import RotatingFileHandler
from pathlib import Path
from unittest.mock import patch
import pytest
import hermes_logging
@pytest.fixture(autouse=True)
def _reset_logging_state():
"""Reset the module-level sentinel and clean up root logger handlers
added by setup_logging() so tests don't leak state."""
hermes_logging._logging_initialized = False
root = logging.getLogger()
original_handlers = list(root.handlers)
yield
# Restore — remove any handlers added during the test.
for h in list(root.handlers):
if h not in original_handlers:
root.removeHandler(h)
h.close()
hermes_logging._logging_initialized = False
@pytest.fixture
def hermes_home(tmp_path, monkeypatch):
"""Provide an isolated HERMES_HOME for logging tests.
Uses the same tmp_path as the autouse _isolate_hermes_home from conftest,
reading it back from the env var to avoid double-mkdir conflicts.
"""
home = Path(os.environ["HERMES_HOME"])
return home
class TestSetupLogging:
"""setup_logging() creates agent.log + errors.log with RotatingFileHandler."""
def test_creates_log_directory(self, hermes_home):
log_dir = hermes_logging.setup_logging(hermes_home=hermes_home)
assert log_dir == hermes_home / "logs"
assert log_dir.is_dir()
def test_creates_agent_log_handler(self, hermes_home):
hermes_logging.setup_logging(hermes_home=hermes_home)
root = logging.getLogger()
agent_handlers = [
h for h in root.handlers
if isinstance(h, RotatingFileHandler)
and "agent.log" in getattr(h, "baseFilename", "")
]
assert len(agent_handlers) == 1
assert agent_handlers[0].level == logging.INFO
def test_creates_errors_log_handler(self, hermes_home):
hermes_logging.setup_logging(hermes_home=hermes_home)
root = logging.getLogger()
error_handlers = [
h for h in root.handlers
if isinstance(h, RotatingFileHandler)
and "errors.log" in getattr(h, "baseFilename", "")
]
assert len(error_handlers) == 1
assert error_handlers[0].level == logging.WARNING
def test_idempotent_no_duplicate_handlers(self, hermes_home):
hermes_logging.setup_logging(hermes_home=hermes_home)
hermes_logging.setup_logging(hermes_home=hermes_home) # second call — should be no-op
root = logging.getLogger()
agent_handlers = [
h for h in root.handlers
if isinstance(h, RotatingFileHandler)
and "agent.log" in getattr(h, "baseFilename", "")
]
assert len(agent_handlers) == 1
def test_force_reinitializes(self, hermes_home):
hermes_logging.setup_logging(hermes_home=hermes_home)
# Force still won't add duplicate handlers because _add_rotating_handler
# checks by resolved path.
hermes_logging.setup_logging(hermes_home=hermes_home, force=True)
root = logging.getLogger()
agent_handlers = [
h for h in root.handlers
if isinstance(h, RotatingFileHandler)
and "agent.log" in getattr(h, "baseFilename", "")
]
assert len(agent_handlers) == 1
def test_custom_log_level(self, hermes_home):
hermes_logging.setup_logging(hermes_home=hermes_home, log_level="DEBUG")
root = logging.getLogger()
agent_handlers = [
h for h in root.handlers
if isinstance(h, RotatingFileHandler)
and "agent.log" in getattr(h, "baseFilename", "")
]
assert agent_handlers[0].level == logging.DEBUG
def test_custom_max_size_and_backup(self, hermes_home):
hermes_logging.setup_logging(
hermes_home=hermes_home, max_size_mb=10, backup_count=5
)
root = logging.getLogger()
agent_handlers = [
h for h in root.handlers
if isinstance(h, RotatingFileHandler)
and "agent.log" in getattr(h, "baseFilename", "")
]
assert agent_handlers[0].maxBytes == 10 * 1024 * 1024
assert agent_handlers[0].backupCount == 5
def test_suppresses_noisy_loggers(self, hermes_home):
hermes_logging.setup_logging(hermes_home=hermes_home)
assert logging.getLogger("openai").level >= logging.WARNING
assert logging.getLogger("httpx").level >= logging.WARNING
assert logging.getLogger("httpcore").level >= logging.WARNING
def test_writes_to_agent_log(self, hermes_home):
hermes_logging.setup_logging(hermes_home=hermes_home)
test_logger = logging.getLogger("test_hermes_logging.write_test")
test_logger.info("test message for agent.log")
# Flush handlers
for h in logging.getLogger().handlers:
h.flush()
agent_log = hermes_home / "logs" / "agent.log"
assert agent_log.exists()
content = agent_log.read_text()
assert "test message for agent.log" in content
def test_warnings_appear_in_both_logs(self, hermes_home):
hermes_logging.setup_logging(hermes_home=hermes_home)
test_logger = logging.getLogger("test_hermes_logging.warning_test")
test_logger.warning("this is a warning")
for h in logging.getLogger().handlers:
h.flush()
agent_log = hermes_home / "logs" / "agent.log"
errors_log = hermes_home / "logs" / "errors.log"
assert "this is a warning" in agent_log.read_text()
assert "this is a warning" in errors_log.read_text()
def test_info_not_in_errors_log(self, hermes_home):
hermes_logging.setup_logging(hermes_home=hermes_home)
test_logger = logging.getLogger("test_hermes_logging.info_test")
test_logger.info("info only message")
for h in logging.getLogger().handlers:
h.flush()
errors_log = hermes_home / "logs" / "errors.log"
if errors_log.exists():
assert "info only message" not in errors_log.read_text()
def test_reads_config_yaml(self, hermes_home):
"""setup_logging reads logging.level from config.yaml."""
import yaml
config = {"logging": {"level": "DEBUG", "max_size_mb": 2, "backup_count": 1}}
(hermes_home / "config.yaml").write_text(yaml.dump(config))
hermes_logging.setup_logging(hermes_home=hermes_home)
root = logging.getLogger()
agent_handlers = [
h for h in root.handlers
if isinstance(h, RotatingFileHandler)
and "agent.log" in getattr(h, "baseFilename", "")
]
assert agent_handlers[0].level == logging.DEBUG
assert agent_handlers[0].maxBytes == 2 * 1024 * 1024
assert agent_handlers[0].backupCount == 1
def test_explicit_params_override_config(self, hermes_home):
"""Explicit function params take precedence over config.yaml."""
import yaml
config = {"logging": {"level": "DEBUG"}}
(hermes_home / "config.yaml").write_text(yaml.dump(config))
hermes_logging.setup_logging(hermes_home=hermes_home, log_level="WARNING")
root = logging.getLogger()
agent_handlers = [
h for h in root.handlers
if isinstance(h, RotatingFileHandler)
and "agent.log" in getattr(h, "baseFilename", "")
]
assert agent_handlers[0].level == logging.WARNING
class TestSetupVerboseLogging:
"""setup_verbose_logging() adds a DEBUG-level console handler."""
def test_adds_stream_handler(self, hermes_home):
hermes_logging.setup_logging(hermes_home=hermes_home)
hermes_logging.setup_verbose_logging()
root = logging.getLogger()
verbose_handlers = [
h for h in root.handlers
if isinstance(h, logging.StreamHandler)
and not isinstance(h, RotatingFileHandler)
and getattr(h, "_hermes_verbose", False)
]
assert len(verbose_handlers) == 1
assert verbose_handlers[0].level == logging.DEBUG
def test_idempotent(self, hermes_home):
hermes_logging.setup_logging(hermes_home=hermes_home)
hermes_logging.setup_verbose_logging()
hermes_logging.setup_verbose_logging() # second call
root = logging.getLogger()
verbose_handlers = [
h for h in root.handlers
if isinstance(h, logging.StreamHandler)
and not isinstance(h, RotatingFileHandler)
and getattr(h, "_hermes_verbose", False)
]
assert len(verbose_handlers) == 1
class TestAddRotatingHandler:
"""_add_rotating_handler() is idempotent and creates the directory."""
def test_creates_directory(self, tmp_path):
log_path = tmp_path / "subdir" / "test.log"
logger = logging.getLogger("_test_rotating")
formatter = logging.Formatter("%(message)s")
hermes_logging._add_rotating_handler(
logger, log_path,
level=logging.INFO, max_bytes=1024, backup_count=1,
formatter=formatter,
)
assert log_path.parent.is_dir()
# Clean up
for h in list(logger.handlers):
if isinstance(h, RotatingFileHandler):
logger.removeHandler(h)
h.close()
def test_no_duplicate_for_same_path(self, tmp_path):
log_path = tmp_path / "test.log"
logger = logging.getLogger("_test_rotating_dup")
formatter = logging.Formatter("%(message)s")
hermes_logging._add_rotating_handler(
logger, log_path,
level=logging.INFO, max_bytes=1024, backup_count=1,
formatter=formatter,
)
hermes_logging._add_rotating_handler(
logger, log_path,
level=logging.INFO, max_bytes=1024, backup_count=1,
formatter=formatter,
)
rotating_handlers = [
h for h in logger.handlers
if isinstance(h, RotatingFileHandler)
]
assert len(rotating_handlers) == 1
# Clean up
for h in list(logger.handlers):
if isinstance(h, RotatingFileHandler):
logger.removeHandler(h)
h.close()
class TestReadLoggingConfig:
"""_read_logging_config() reads from config.yaml."""
def test_returns_none_when_no_config(self, hermes_home):
level, max_size, backup = hermes_logging._read_logging_config()
assert level is None
assert max_size is None
assert backup is None
def test_reads_logging_section(self, hermes_home):
import yaml
config = {"logging": {"level": "DEBUG", "max_size_mb": 10, "backup_count": 5}}
(hermes_home / "config.yaml").write_text(yaml.dump(config))
level, max_size, backup = hermes_logging._read_logging_config()
assert level == "DEBUG"
assert max_size == 10
assert backup == 5
def test_handles_missing_logging_section(self, hermes_home):
import yaml
config = {"model": "test"}
(hermes_home / "config.yaml").write_text(yaml.dump(config))
level, max_size, backup = hermes_logging._read_logging_config()
assert level is None