RomanNum3ral
/
hermes-agent
mirror of https://github.com/NousResearch/hermes-agent.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix: follow-up for salvaged PR #18255 

- Fix _usage_audit_path() to use _get_hermes_home() instead of hardcoded
  Path.home() / '.hermes' (profile-safe resolution, sweeper finding)
- Rewrite skip_background_review tests to exercise finalize_turn() directly
  instead of duplicating the guard expression (sweeper finding)
- Fix response_silent audit field to use _is_cron_silence_response()
  instead of the buggy SILENT_MARKER substring check it was meant to
  replace (simplify-code review finding)
- Remove dead 'model' in locals() guard — model is always in scope
  before the try block (simplify-code review finding)
- Extract _stub_agent_for_finalize() helper to eliminate ~40 lines of
  copy-pasted agent stubbing in tests (simplify-code review finding)
- Clean up 'Phase 0.5' instrumentation comments
This commit is contained in:
kshitij 2026-08-08 00:03:13 +05:30
parent 15927c1d24
commit 7307f88993
4 changed files with 99 additions and 88 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
contributors/emails/0xarkstar@users.noreply.github.com Normal file
View File

@ -0,0 +1 @@
0xarkstar

14
cron/scheduler.py
View File

@ -4125,8 +4125,7 @@ def run_job(
# env passthrough registrations) when the cron run hops into the worker
# thread used for inactivity timeout monitoring.
_cron_context = contextvars.copy_context()
# Tag this fire and time the
# run_conversation call for the usage_audit.jsonl entry.
# Tag this fire and time the run_conversation call for the usage_audit.jsonl entry.
_audit_fire_id = uuid.uuid4().hex
_audit_t_start = time.monotonic()
_cron_future = _cron_pool.submit(_cron_context.run, agent.run_conversation, prompt)
@ -4285,10 +4284,7 @@ def run_job(
# Emit one JSONL line per fire for usage audit.
_audit_duration_ms = int((time.monotonic() - _audit_t_start) * 1000)
_audit_response_silent = (
not final_response.strip()
or SILENT_MARKER in (final_response or "").upper()
)
_audit_response_silent = _is_cron_silence_response(final_response or "")
_write_usage_audit({
"ts": _utcnow_iso_ms(),
"job_id": job_id,
@ -4308,8 +4304,8 @@ def run_job(
error_msg = f"{type(e).__name__}: {str(e)}"
logger.exception("Job '%s' failed: %s", job_name, error_msg)
# Best-effort audit write on failure path. _audit_fire_id
# may be unset if the exception fired before submit() — guard with
# locals() lookup so the audit write itself never raises.
# may be unset if the exception fired before submit() — guard
# with a None check so the audit write itself never raises.
if "_audit_fire_id" in locals():
_audit_duration_ms = int((time.monotonic() - _audit_t_start) * 1000)
_write_usage_audit({
@ -4321,7 +4317,7 @@ def run_job(
"total_tokens": None,
"response_silent": False,
"deliver_target": job.get("deliver"),
"model": (model or None) if "model" in locals() else None,
"model": model or None,
"duration_ms": _audit_duration_ms,
"error": error_msg,
})

132
tests/agent/test_skip_background_review.py
View File

@ -4,22 +4,17 @@ Verifies that AIAgent can be instructed to skip the end-of-turn
_spawn_background_review fork (~30K tokens / event), which is essential
on cron sessions that have no human-in-the-loop value from skill/memory
review forks.
Plan reference: ralplan-hermes-token-leaks.md §3.9 (Phase 8).
"""
from __future__ import annotations
from unittest.mock import patch
from unittest.mock import MagicMock
from run_agent import AIAgent
from agent.turn_finalizer import finalize_turn
def _make_agent(skip_background_review: bool = False) -> AIAgent:
"""Construct a minimally-configured AIAgent for unit testing.
Mirrors the kwargs in tests/hermes_cli/test_timeouts.py — provider /
base_url stub plus skip_memory + skip_context_files to keep init fast.
"""
"""Construct a minimally-configured AIAgent for unit testing."""
return AIAgent(
model="openai/gpt-4o-mini",
provider="openrouter",
@ -33,6 +28,57 @@ def _make_agent(skip_background_review: bool = False) -> AIAgent:
)
def _stub_agent_for_finalize(agent: AIAgent) -> None:
"""Stub the heavy finalizer dependencies to isolate the review gate."""
agent._spawn_background_review = MagicMock()
agent._save_trajectory = MagicMock()
agent._cleanup_task_resources = MagicMock()
agent._persist_session = MagicMock()
agent._session_messages = []
agent._file_mutation_verifier_enabled = lambda: False
agent.clear_interrupt = MagicMock()
agent._stream_callback = None
agent._sync_external_memory_for_turn = MagicMock()
agent._skill_nudge_interval = 10
agent._iters_since_skill = 20 # exceeds nudge interval → _should_review_skills = True
agent.valid_tool_names = {"skill_manage"}
agent.iteration_budget = MagicMock()
agent.iteration_budget.remaining = 100
agent.iteration_budget.used = 5
agent.iteration_budget.max_total = 100
agent.max_iterations = 50
agent._emit_status = MagicMock()
agent._safe_print = MagicMock()
agent._apply_persist_user_message_override = MagicMock()
agent.context_compressor = None
agent._turn_preflight_display_snapshot = None
agent._turn_received_provider_response = False
agent.model = "test-model"
agent.session_id = "test-session"
agent.quiet_mode = True
agent._turn_failed_file_mutations = {}
agent._db_flush_scan_prefix = None
def _run_finalize(agent: AIAgent) -> None:
"""Call finalize_turn with conditions that would trigger background review."""
finalize_turn(
agent,
final_response="ok",
api_call_count=1,
interrupted=False,
failed=False,
messages=[{"role": "assistant", "content": "ok"}],
conversation_history=[],
effective_task_id="test",
turn_id="test-turn",
user_message="test",
original_user_message="test",
_should_review_memory=True,
_turn_exit_reason="text_response(1)",
)
def test_default_skip_background_review_is_false() -> None:
"""Without an explicit override, AIAgent does NOT skip background review."""
agent = _make_agent()
@ -45,78 +91,38 @@ def test_skip_background_review_flag_persists() -> None:
assert agent.skip_background_review is True
def test_review_path_short_circuits_when_flag_set() -> None:
"""The end-of-turn review block is gated on `not self.skip_background_review`.
def test_finalize_turn_skips_review_when_flag_set() -> None:
"""finalize_turn must NOT call _spawn_background_review when skip_background_review=True.
We don't drive a full conversation — instead we exercise the boolean
guard expression directly to confirm the gate works as wired.
Exercises the actual finalizer call path (not a duplicated guard expression)
so it catches divergence between the production guard and the test.
"""
agent = _make_agent(skip_background_review=True)
# Simulate the conditions that would have fired the review:
final_response = "ok"
interrupted = False
_should_review_memory = True
_should_review_skills = True
with patch.object(agent, "_spawn_background_review") as mock_spawn:
# This is the exact guard from run_agent.py end-of-turn block.
if (
final_response
and not interrupted
and not getattr(agent, "skip_background_review", False)
and (_should_review_memory or _should_review_skills)
):
agent._spawn_background_review(
messages_snapshot=[],
review_memory=_should_review_memory,
review_skills=_should_review_skills,
)
mock_spawn.assert_not_called()
_stub_agent_for_finalize(agent)
_run_finalize(agent)
agent._spawn_background_review.assert_not_called()
def test_review_path_fires_when_flag_unset() -> None:
"""Counterpart: with the flag off, the review path is reachable."""
def test_finalize_turn_fires_review_when_flag_unset() -> None:
"""Counterpart: with the flag off, finalize_turn DOES call _spawn_background_review."""
agent = _make_agent(skip_background_review=False)
final_response = "ok"
interrupted = False
_should_review_memory = False
_should_review_skills = True
with patch.object(agent, "_spawn_background_review") as mock_spawn:
if (
final_response
and not interrupted
and not getattr(agent, "skip_background_review", False)
and (_should_review_memory or _should_review_skills)
):
agent._spawn_background_review(
messages_snapshot=[],
review_memory=_should_review_memory,
review_skills=_should_review_skills,
)
mock_spawn.assert_called_once()
_stub_agent_for_finalize(agent)
_run_finalize(agent)
agent._spawn_background_review.assert_called_once()
def test_cron_construction_sets_skip_background_review() -> None:
"""The cron scheduler MUST construct AIAgent with skip_background_review=True.
Verified via source-text inspection — the cron scheduler is heavy to
boot in tests (loads gateway config, profile, telemetry), so we
assert that the source declares the flag rather than running the
scheduler. This still catches accidental removal of the flag.
boot in tests, so we assert that the source declares the flag rather
than running the scheduler. This catches accidental removal.
"""
import pathlib
scheduler_src = pathlib.Path(__file__).resolve().parents[2] / "cron" / "scheduler.py"
text = scheduler_src.read_text(encoding="utf-8")
# The flag must appear inside the cron AIAgent(...) construction block.
# We look for it next to the existing skip_memory=True line.
assert "skip_background_review=True" in text, (
"cron/scheduler.py must construct AIAgent with skip_background_review=True "
"(see ralplan-hermes-token-leaks.md §3.9 / Phase 8)."
"cron/scheduler.py must construct AIAgent with skip_background_review=True."
)

40
tests/cron/test_usage_audit_logger.py
View File

@ -1,4 +1,4 @@
"""Tests for the Phase 0.5 cron usage_audit.jsonl logger.
"""Tests for the cron usage_audit.jsonl logger.
Covers:
- successful write produces a single valid JSONL line with full schema
@ -6,6 +6,7 @@ Covers:
- writer exception is swallowed (json.dumps raises) — call must return cleanly
- file path is created if parent dir is missing
- timestamp format is RFC3339 UTC with millisecond precision and 'Z' suffix
- path resolves through _get_hermes_home() (profile-safe)
"""
from __future__ import annotations
@ -21,11 +22,11 @@ from cron import scheduler
@pytest.fixture
def tmp_home(tmp_path, monkeypatch):
"""Redirect Path.home() so the audit logger writes under tmp_path."""
fake_home = tmp_path / "home"
fake_home.mkdir()
monkeypatch.setattr(Path, "home", lambda: fake_home)
def tmp_hermes_home(tmp_path, monkeypatch):
"""Redirect _get_hermes_home() so the audit logger writes under tmp_path."""
fake_home = tmp_path / "home" / ".hermes"
fake_home.mkdir(parents=True)
monkeypatch.setattr(scheduler, "_get_hermes_home", lambda: fake_home)
return fake_home
@ -34,9 +35,16 @@ def _read_jsonl(path: Path) -> list[dict]:
class TestUsageAuditPath:
def test_resolves_under_user_home(self, tmp_home):
def test_resolves_through_get_hermes_home(self, tmp_hermes_home):
p = scheduler._usage_audit_path()
assert p == tmp_home / ".hermes" / "cron" / "usage_audit.jsonl"
assert p == tmp_hermes_home / "cron" / "usage_audit.jsonl"
def test_does_not_use_path_home(self, tmp_hermes_home):
"""Audit path must NOT hardcode Path.home() — it bypasses profile-aware resolution."""
with patch.object(Path, "home") as mock_home:
p = scheduler._usage_audit_path()
mock_home.assert_not_called()
assert p == tmp_hermes_home / "cron" / "usage_audit.jsonl"
class TestUtcnowIsoMs:
@ -47,7 +55,7 @@ class TestUtcnowIsoMs:
class TestWriteUsageAudit:
def test_successful_write_produces_valid_jsonl(self, tmp_home):
def test_successful_write_produces_valid_jsonl(self, tmp_hermes_home):
record = {
"ts": "2026-05-01T04:23:11.123Z",
"job_id": "bluenode-dispatch-recommend-sweep",
@ -69,7 +77,7 @@ class TestWriteUsageAudit:
assert len(lines) == 1
assert lines[0] == record
def test_missing_token_info_writes_line_with_null_fields(self, tmp_home):
def test_missing_token_info_writes_line_with_null_fields(self, tmp_hermes_home):
record = {
"ts": "2026-05-01T04:23:11.123Z",
"job_id": "j",
@ -91,7 +99,7 @@ class TestWriteUsageAudit:
assert lines[0]["total_tokens"] is None
assert lines[0]["error"] == "boom"
def test_writer_exception_swallowed(self, tmp_home, caplog):
def test_writer_exception_swallowed(self, tmp_hermes_home, caplog):
# Force json.dumps to raise — writer must NOT propagate.
with patch("cron.scheduler.json.dumps", side_effect=RuntimeError("kaboom")):
scheduler._write_usage_audit({"job_id": "x"})
@ -101,9 +109,9 @@ class TestWriteUsageAudit:
# Warning logged with our marker.
assert any("usage_audit write failed" in rec.message for rec in caplog.records)
def test_parent_dir_created_if_missing(self, tmp_home):
# Ensure the .hermes/cron path does not exist yet.
target = tmp_home / ".hermes" / "cron"
def test_parent_dir_created_if_missing(self, tmp_hermes_home):
# Ensure the cron path does not exist yet.
target = tmp_hermes_home / "cron"
assert not target.exists()
scheduler._write_usage_audit({"k": "v"})
@ -111,14 +119,14 @@ class TestWriteUsageAudit:
assert target.exists() and target.is_dir()
assert (target / "usage_audit.jsonl").exists()
def test_appends_multiple_records(self, tmp_home):
def test_appends_multiple_records(self, tmp_hermes_home):
scheduler._write_usage_audit({"i": 1})
scheduler._write_usage_audit({"i": 2})
scheduler._write_usage_audit({"i": 3})
lines = _read_jsonl(scheduler._usage_audit_path())
assert [r["i"] for r in lines] == [1, 2, 3]
def test_unicode_preserved_not_escaped(self, tmp_home):
def test_unicode_preserved_not_escaped(self, tmp_hermes_home):
# ensure_ascii=False so non-ASCII model names / job names round-trip cleanly.
scheduler._write_usage_audit({"job_id": "한글", "model": "gemma"})
text = scheduler._usage_audit_path().read_text(encoding="utf-8")