"""Durable cron failure incidents: signature dedup, lifecycle, ack, CLI.""" from __future__ import annotations import argparse import sys from datetime import timedelta from pathlib import Path from unittest.mock import MagicMock, patch sys.path.insert(0, str(Path(__file__).parent.parent.parent)) import cron.incidents as incidents import cron.jobs as cron_jobs import cron.scheduler as sched from hermes_time import now as _hermes_now def _point_db(monkeypatch, tmp_path): """Point the incident store at a throwaway executions.db (same file shape the scheduler uses). ``cron.executions.EXECUTIONS_FILE`` stays None so the incident store falls back to its own override.""" monkeypatch.setattr(incidents, "EXECUTIONS_FILE", tmp_path / "cron" / "executions.db") return incidents def _job(**overrides): job = { "id": "incident-gating-test", "name": "incident gating test", "prompt": "hello", "enabled": True, "state": "scheduled", "schedule": {"kind": "interval", "minutes": 5, "display": "every 5m"}, "deliver": "local", "model": None, "provider": None, "provider_snapshot": "openrouter", "base_url": None, } job.update(overrides) return job def _tick_failing(job, tmp_path, deliveries, error="boom unrelated"): """Run one run_one_job tick whose agent raises ``error`` (the failure path that composes the per-run failure ping). Mirrors the preflight alert-once harness so the incident gating is exercised through the real scheduler.""" fake_db = MagicMock() def fake_deliver(jb, content, adapters=None, loop=None, **kwargs): deliveries.append(content) return None with cron_jobs.use_cron_store(tmp_path), \ patch("cron.scheduler._hermes_home", tmp_path), \ patch("cron.scheduler_delivery._resolve_origin", return_value=None), \ patch("hermes_cli.env_loader.load_hermes_dotenv"), \ patch("hermes_cli.env_loader.reset_secret_source_cache"), \ patch("hermes_state_registry.acquire", return_value=fake_db), \ patch("tools.mcp_tool_discovery.discover_mcp_tools", return_value=[]), \ patch("hermes_cli.runtime_provider.resolve_runtime_provider", return_value={ "api_key": "test-key", "base_url": "https://example.invalid/v1", "provider": "openrouter", "api_mode": "chat_completions", }), \ patch.object(sched, "_deliver_result", side_effect=fake_deliver), \ patch("run_agent.AIAgent") as mock_agent_cls: mock_agent = MagicMock() mock_agent.run_conversation.side_effect = RuntimeError(error) mock_agent_cls.return_value = mock_agent sched.run_one_job(dict(job)) return mock_agent_cls.called # ── Store + dedup ────────────────────────────────────────────────────────── def test_new_failure_creates_incident_and_is_new(monkeypatch, tmp_path): inc = _point_db(monkeypatch, tmp_path) inc_id, is_new = inc.upsert_incident("job-1", "Provider timeout: read timed out") assert is_new is True assert inc_id.startswith("job-1_") row = inc.get_incident(inc_id) assert row is not None assert row["job_id"] == "job-1" assert row["state"] == "detected" assert row["failure_type"] == "timeout" assert row["first_seen_at"] == row["last_seen_at"] assert inc.count_incidents() == 1 def test_incidents_list_newest_instant_first_across_dst_fall_back(monkeypatch, tmp_path): """01:10-05:00 is 20 minutes after 01:50-04:00 but sorts first as text.""" from datetime import datetime from zoneinfo import ZoneInfo inc = _point_db(monkeypatch, tmp_path) new_york = ZoneInfo("America/New_York") monkeypatch.setattr( inc, "_hermes_now", lambda: datetime(2026, 11, 1, 1, 50, tzinfo=new_york, fold=0) ) earlier, _ = inc.upsert_incident("job-1", "first failure") monkeypatch.setattr( inc, "_hermes_now", lambda: datetime(2026, 11, 1, 1, 10, tzinfo=new_york, fold=1) ) later, _ = inc.upsert_incident("job-2", "second failure") assert [row["id"] for row in inc.list_incidents()] == [later, earlier] def test_same_signature_dedups_same_incident(monkeypatch, tmp_path): inc = _point_db(monkeypatch, tmp_path) id1, new1 = inc.upsert_incident("job-1", "Provider timeout: read timed out") id2, new2 = inc.upsert_incident("job-1", "PROVIDER TIMEOUT: read timed out ") assert id1 == id2, "normalized (case/whitespace) signature must dedup" assert new1 is True assert new2 is False assert inc.count_incidents() == 1 # Refresh updates last_seen but never resets an open state. assert inc.get_incident(id1)["state"] == "detected" def test_error_change_mints_new_incident(monkeypatch, tmp_path): inc = _point_db(monkeypatch, tmp_path) id1, _ = inc.upsert_incident("job-1", "provider timeout") id2, new2 = inc.upsert_incident("job-1", "provider rate limit 429") assert id1 != id2 assert new2 is True assert inc.count_incidents() == 2 # ── Redaction / classification ───────────────────────────────────────────── def test_redaction_applied_to_incident_error(monkeypatch, tmp_path): # agent.redact snapshots _REDACT_ENABLED from HERMES_REDACT_SECRETS at # module-import time. When another collected test module imports the # gateway/scheduler chain (e.g. test_codex_execution_paths.py), that # import happens at COLLECTION time — before the conftest env scrub — # so a developer shell exporting HERMES_REDACT_SECRETS=false freezes # redaction off and this test fails only in full-directory runs. # Pin the flag explicitly, matching the repo-wide pattern. monkeypatch.setattr("agent.redact._REDACT_ENABLED", True, raising=False) inc = _point_db(monkeypatch, tmp_path) secret = "ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghij" inc_id, _ = inc.upsert_incident("job-1", f"failed: {secret} boom") row = inc.get_incident(inc_id) assert secret not in row["error"] assert "boom" in row["error"] def test_error_truncated_to_bounded_length(monkeypatch, tmp_path): inc = _point_db(monkeypatch, tmp_path) long_error = "x" * 2000 inc_id, _ = inc.upsert_incident("job-1", long_error) assert len(inc.get_incident(inc_id)["error"]) <= 500 def test_failure_type_classification(monkeypatch, tmp_path): inc = _point_db(monkeypatch, tmp_path) cases = [ ("delivery failed for telegram chat", "delivery"), ("Provider read timed out after 60s", "timeout"), ("authentication failed: invalid API key", "auth"), ("HTTP 429: rate limit exceeded", "rate_limit"), ("configuration validation blocked the run", "config"), ("script exited with code 1", "script"), ("agent crashed mid-conversation", "agent"), ("something completely unexpected happened", "unknown"), ] for error, expected in cases: assert inc._classify_failure_type(error) == expected, (error, expected) # ── Lifecycle / ack ──────────────────────────────────────────────────────── def test_lifecycle_transitions(monkeypatch, tmp_path): inc = _point_db(monkeypatch, tmp_path) inc_id, _ = inc.upsert_incident("job-1", "boom") assert inc.get_incident(inc_id)["state"] == "detected" assert inc.set_incident_state(inc_id, "alerted") is True assert inc.set_incident_state(inc_id, "closed") is True row = inc.get_incident(inc_id) assert row["state"] == "closed" assert row["acked_at"] and row["closed_at"] # Closed is terminal for that signature: no re-open, no re-transition. assert inc.set_incident_state(inc_id, "alerted") is False assert inc.set_incident_state(inc_id, "closed") is False assert inc.ack_incident(inc_id) is False assert inc.get_incident(inc_id)["state"] == "closed" # Invalid states are rejected, not raised. assert inc.set_incident_state(inc_id, "bogus") is False assert inc.list_incidents(state="bogus") == [] assert inc.count_incidents(state="bogus") == 0 def test_acked_signature_stays_closed_on_refresh(monkeypatch, tmp_path): """Ack is per-signature: upserting the same error after ack must NOT resurrect the incident — a changed error is what mints a new one.""" inc = _point_db(monkeypatch, tmp_path) inc_id, _ = inc.upsert_incident("job-1", "same failure text") inc.ack_incident(inc_id) same_id, is_new = inc.upsert_incident("job-1", "SAME FAILURE TEXT") assert same_id == inc_id assert is_new is False assert inc.get_incident(inc_id)["state"] == "closed" # ── Missing DB / lazy schema ─────────────────────────────────────────────── def test_missing_db_no_crash(monkeypatch, tmp_path): inc = _point_db(monkeypatch, tmp_path) inc_id, is_new = inc.upsert_incident("job-1", "boom") assert is_new is True assert (tmp_path / "cron" / "executions.db").is_file() assert inc.list_incidents() == [inc.get_incident(inc_id)] assert inc.count_incidents() == 1 assert inc.get_incident("nope") is None # ── Scheduler gating ─────────────────────────────────────────────────────── def test_repeat_failure_alerts_once_then_reminds_after_cooldown(monkeypatch, tmp_path): """Same job + same error: the first failing run delivers, the repeat is withheld (but still recorded as a run), one reminder goes out once ``cron.failure_repeat_alert_hours`` has elapsed, and a green run re-arms the signature so the same error alerts again.""" inc = _point_db(monkeypatch, tmp_path) deliveries = [] # A real (non-local) lane: the ping leaves the process, so the incident is marked alerted. job = _job(deliver="telegram:123") (tmp_path / "config.yaml").write_text("cron:\n preflight: false\n", encoding="utf-8") monkeypatch.setenv("HERMES_HOME", str(tmp_path)) with cron_jobs.use_cron_store(tmp_path): cron_jobs.save_jobs([job]) _tick_failing(job, tmp_path, deliveries, error="repeat boom") _tick_failing(job, tmp_path, deliveries, error="repeat boom") assert len(deliveries) == 1, "an alerted signature must not re-ping on every run" rows = inc.list_incidents() assert len(rows) == 1 and rows[0]["state"] == "alerted" and rows[0]["alerted_at"] stored = [j for j in cron_jobs.load_jobs() if j["id"] == job["id"]][0] assert stored["last_status"] == "error", "the withheld run is still recorded" # Cooldown elapsed: exactly one reminder, then silent again. stale = (_hermes_now() - timedelta(hours=7)).isoformat() with inc._transaction() as conn: conn.execute("UPDATE cron_incidents SET alerted_at=?", (stale,)) _tick_failing(job, tmp_path, deliveries, error="repeat boom") _tick_failing(job, tmp_path, deliveries, error="repeat boom") assert len(deliveries) == 2, "one reminder after the cooldown, not one per run" # Recovery re-arms: the same error after a green run alerts immediately. sched._resolve_incidents_for_recovered_job(job) _tick_failing(job, tmp_path, deliveries, error="repeat boom") assert len(deliveries) == 3 assert inc.count_incidents() == 1 def test_ack_suppresses_alert_until_signature_changes(monkeypatch, tmp_path): inc = _point_db(monkeypatch, tmp_path) deliveries = [] job = _job() with cron_jobs.use_cron_store(tmp_path): cron_jobs.save_jobs([job]) # First failure: alert delivered, incident minted. _tick_failing(job, tmp_path, deliveries, error="boom signature A") assert len(deliveries) == 1 rows = inc.list_incidents() assert len(rows) == 1 and rows[0]["state"] == "detected" inc_id = rows[0]["id"] # Acknowledge it. assert inc.ack_incident(inc_id) is True # Same signature: alert suppressed, incident stays closed. _tick_failing(job, tmp_path, deliveries, error="boom signature A") assert len(deliveries) == 1, "acked signature must not re-ping" assert inc.get_incident(inc_id)["state"] == "closed" # Changed signature: new incident, alert again. _tick_failing(job, tmp_path, deliveries, error="boom signature B") assert len(deliveries) == 2, "changed signature must re-alert" assert inc.count_incidents() == 2 def test_ack_holds_when_only_a_measured_duration_differs(monkeypatch, tmp_path): """The inactivity watchdog polls every few seconds, so the same stall reports a slightly different idle time each run. That is the same failure: the ack must keep it silent.""" inc = _point_db(monkeypatch, tmp_path) deliveries = [] job = _job() stall = "Cron job 'x' idle for {}s (limit 600s) — last activity: waiting for tool" with cron_jobs.use_cron_store(tmp_path): cron_jobs.save_jobs([job]) _tick_failing(job, tmp_path, deliveries, error=stall.format(601)) assert len(deliveries) == 1 (first,) = inc.list_incidents() assert inc.ack_incident(first["id"]) is True _tick_failing(job, tmp_path, deliveries, error=stall.format(604)) assert len(deliveries) == 1, "acked stall must not re-ping on a new idle reading" assert [row["id"] for row in inc.list_incidents()] == [first["id"]] def test_mark_incident_alerted_sets_state_never_resurrects(monkeypatch, tmp_path): """The post-delivery 'alerted' transition records that a ping went out, and is a no-op on a closed (acked) incident — it can never resurrect one.""" inc = _point_db(monkeypatch, tmp_path) inc_id, _ = inc.upsert_incident("job-1", "boom") sched._mark_incident_alerted(inc_id) assert inc.get_incident(inc_id)["state"] == "alerted" inc.ack_incident(inc_id) sched._mark_incident_alerted(inc_id) assert inc.get_incident(inc_id)["state"] == "closed" # Best-effort: bad/missing ids never raise. sched._mark_incident_alerted(None) sched._mark_incident_alerted("nonexistent") def test_best_effort_incident_store_failure_returns_false(monkeypatch, tmp_path): """An incident-store error must never break the cron delivery path.""" _point_db(monkeypatch, tmp_path) with patch("cron.incidents.upsert_incident", side_effect=RuntimeError("db locked")): assert sched._upsert_incident_for_failure(_job(), "boom") == (False, None) # ── CLI ──────────────────────────────────────────────────────────────────── def test_cli_list_and_ack(monkeypatch, tmp_path, capsys): from hermes_cli.cron import cron_incidents inc = _point_db(monkeypatch, tmp_path) inc_id, _ = inc.upsert_incident("job-1", "provider timeout boom") # List. list_args = argparse.Namespace( incident_action="list", state=None, incident_id=None ) assert cron_incidents(list_args) == 0 out = capsys.readouterr().out assert inc_id in out assert "job-1" in out # State filter. filter_args = argparse.Namespace( incident_action="list", state="closed", incident_id=None ) assert cron_incidents(filter_args) == 0 out = capsys.readouterr().out assert "No cron failure incidents recorded." in out # Ack. ack_args = argparse.Namespace( incident_action="ack", state=None, incident_id=inc_id ) assert cron_incidents(ack_args) == 0 assert inc.get_incident(inc_id)["state"] == "closed" out = capsys.readouterr().out assert "acknowledged" in out.lower() # Ack again: already closed, still a clean exit. assert cron_incidents(ack_args) == 0 out = capsys.readouterr().out assert "already closed" in out.lower() # Ack with a missing id is a usage error. missing_args = argparse.Namespace( incident_action="ack", state=None, incident_id=None ) assert cron_incidents(missing_args) == 1 def test_alerted_gate_honours_cooldown_opt_out_and_legacy_rows(monkeypatch, tmp_path): """Unit level: ``alerted`` withholds only inside the cooldown; ``0`` restores per-run alerts; a pre-migration row (state alerted, no ``alerted_at``) delivers rather than swallowing the alert; ``closed`` still wins regardless of the cooldown.""" inc = _point_db(monkeypatch, tmp_path) monkeypatch.setattr(sched, "_failure_repeat_alert_hours", lambda: 6.0) job = _job() withheld, inc_id = sched._upsert_incident_for_failure(job, "repeat boom") assert withheld is False and inc_id is not None sched._mark_incident_alerted(inc_id) assert inc.get_incident(inc_id)["state"] == "alerted" assert sched._upsert_incident_for_failure(job, "repeat boom") == (True, inc_id) monkeypatch.setattr(sched, "_failure_repeat_alert_hours", lambda: 0) assert sched._upsert_incident_for_failure(job, "repeat boom") == (False, inc_id) monkeypatch.setattr(sched, "_failure_repeat_alert_hours", lambda: 6.0) with inc._transaction() as conn: conn.execute("UPDATE cron_incidents SET alerted_at=NULL WHERE id=?", (inc_id,)) assert sched._upsert_incident_for_failure(job, "repeat boom") == (False, inc_id) # Re-alerting restarts the window (the reminder stamps alerted_at again). sched._mark_incident_alerted(inc_id) assert inc.get_incident(inc_id)["alerted_at"] assert sched._upsert_incident_for_failure(job, "repeat boom") == (True, inc_id) assert inc.ack_incident(inc_id) is True monkeypatch.setattr(sched, "_failure_repeat_alert_hours", lambda: 0) assert sched._upsert_incident_for_failure(job, "repeat boom") == (True, inc_id)