diff --git a/agent/monitoring/cron_health.py b/agent/monitoring/cron_health.py index 03e8b8dd934..2ed0f1cb9b2 100644 --- a/agent/monitoring/cron_health.py +++ b/agent/monitoring/cron_health.py @@ -4,6 +4,7 @@ from __future__ import annotations import hashlib import logging +import re from dataclasses import dataclass from datetime import datetime from typing import Any, Optional @@ -12,6 +13,7 @@ from agent.monitoring.events import CronExecutionEvent from agent.monitoring.gateway_health import GatewayHealthSnapshot, GatewayMetric from cron.jobs import ( _compute_grace_seconds, + get_catch_up_occurrence_count, get_ticker_heartbeat_age, get_ticker_success_age, load_jobs, @@ -42,7 +44,12 @@ def _job_key(raw: Any) -> str: def classify_cron_error(raw: Any) -> str: text = str(raw or "").lower() - if any(value in text for value in ("auth", "token", "unauthorized", "forbidden", "401", "403")): + if ( + re.search(r"\b(?:authentication|authenticated|authenticate|authorization|authorized|authorize|unauthorized|forbidden)\b", text) + or re.search(r"\bbearer\b", text) + or re.search(r"\b(?:access|api|refresh) token\b", text) + or re.search(r"\b(?:401|403)\b", text) + ): return "auth_failed" if "rate limit" in text or "429" in text or "quota" in text: return "rate_limited" @@ -85,6 +92,8 @@ def project_execution_event( ) -> CronExecutionEvent: status = str(record.get("status") or "unknown").lower() source = str(record.get("source") or "unknown").lower() + if source not in _KNOWN_SOURCES and source != "unknown": + source = "external" outcome = str(delivery_outcome).lower() if delivery_outcome is not None else None return CronExecutionEvent( status=status if status in _KNOWN_STATUSES else "unknown", @@ -149,6 +158,17 @@ def build_cron_health_snapshot() -> CronHealthSnapshot: except Exception: logger.debug("cron freshness metric unavailable", exc_info=True) + try: + metrics.append( + GatewayMetric( + "hermes.cron.scheduler.catch_up_occurrences", + get_catch_up_occurrence_count(), + {}, + ) + ) + except Exception: + logger.debug("cron catch-up metric unavailable", exc_info=True) + try: jobs = load_jobs() enabled = [job for job in jobs if job.get("enabled", True)] diff --git a/agent/monitoring/gateway_health_export.py b/agent/monitoring/gateway_health_export.py index a0c851682e8..222ac599523 100644 --- a/agent/monitoring/gateway_health_export.py +++ b/agent/monitoring/gateway_health_export.py @@ -356,6 +356,7 @@ def _start_metric_provider(config: Dict[str, Any], sdk: Dict[str, Any]) -> Any: "hermes.platform.degraded", "hermes.cron.scheduler.heartbeat_age_seconds", "hermes.cron.scheduler.last_success_age_seconds", + "hermes.cron.scheduler.catch_up_occurrences", "hermes.cron.jobs.enabled", "hermes.cron.jobs.running", "hermes.cron.jobs.overdue", diff --git a/cron/jobs.py b/cron/jobs.py index 5a21f6d5246..00e0d5c40c3 100644 --- a/cron/jobs.py +++ b/cron/jobs.py @@ -806,6 +806,24 @@ def _atomic_write_epoch(path: Path) -> None: raise +def _atomic_write_counter(path: Path, value: int) -> None: + """Atomically persist a non-negative integer counter.""" + ensure_dirs() + fd, tmp_path = tempfile.mkstemp(dir=str(path.parent), suffix=".tmp", prefix=".count_") + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + f.write(str(max(0, value))) + f.flush() + os.fsync(f.fileno()) + atomic_replace(tmp_path, path) + except BaseException: + try: + os.unlink(tmp_path) + except OSError: + pass + raise + + def record_ticker_heartbeat(success: bool = False) -> None: """Record a ticker liveness signal, and optionally a successful-tick signal. @@ -867,6 +885,28 @@ def get_ticker_success_age() -> Optional[float]: return _epoch_file_age(store.cron_dir / "ticker_last_success") +def record_catch_up_occurrence() -> None: + """Increment the profile-local stale-schedule catch-up counter, best effort.""" + path = _current_cron_store().cron_dir / "catch_up_occurrences" + try: + try: + value = int(path.read_text(encoding="utf-8").strip()) + except (OSError, ValueError): + value = 0 + _atomic_write_counter(path, max(0, value) + 1) + except Exception: + pass + + +def get_catch_up_occurrence_count() -> int: + """Return the profile-local stale-schedule catch-up count.""" + path = _current_cron_store().cron_dir / "catch_up_occurrences" + try: + return max(0, int(path.read_text(encoding="utf-8").strip())) + except (OSError, ValueError): + return 0 + + # ============================================================================= # Job CRUD Operations # ============================================================================= @@ -2088,6 +2128,7 @@ def _get_due_jobs_locked() -> List[Dict[str, Any]]: rj["next_run_at"] = new_next needs_save = True break + record_catch_up_occurrence() # Fall through to due.append(job) — execute once now # One-shot dispatch-limit guard (issue #38758): a finite one-shot diff --git a/cron/scheduler.py b/cron/scheduler.py index 830c18aabab..e1cb061af38 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -3945,6 +3945,7 @@ def run_one_job(job: dict, *, adapters=None, loop=None, verbose: bool = False) - # responses: do not deliver a blank message, and let the # empty-response guard below mark the run as a soft failure. should_deliver = bool(deliver_content.strip()) + unresolved_origin = False # Cron silence suppression — see _is_cron_silence_response. Replaces the # old `SILENT_MARKER in ...upper()` substring check, which both leaked # bracketless near-markers ("SILENT" / "NO_REPLY") and wrongly swallowed @@ -3956,6 +3957,10 @@ def run_one_job(job: dict, *, adapters=None, loop=None, verbose: bool = False) - should_deliver = False if should_deliver: + unresolved_origin = ( + _normalize_deliver_value(job.get("deliver", "local")) == "origin" + and not _resolve_delivery_targets(job) + ) try: delivery_error = _deliver_result(job, deliver_content, adapters=adapters, loop=loop) except Exception as de: @@ -3979,6 +3984,8 @@ def run_one_job(job: dict, *, adapters=None, loop=None, verbose: bool = False) - mark_job_run(job["id"], success, error, delivery_error=delivery_error) if delivery_error: delivery_outcome = "failed" + elif should_deliver and unresolved_origin: + delivery_outcome = "not_configured" elif should_deliver and job.get("deliver", "local") != "local": delivery_outcome = "delivered" else: diff --git a/docs/observability/monitoring.md b/docs/observability/monitoring.md index 027ac9db8c7..cdec47c496e 100644 --- a/docs/observability/monitoring.md +++ b/docs/observability/monitoring.md @@ -19,8 +19,8 @@ capture is a separate plane served by the NeMo Relay integration | Gateway gauges | `/v1/metrics` | `hermes.gateway.up/state/busy/drainable/active_agents/restart_requested`, `hermes.platform.up/degraded` with bounded `error_code` attributes | | Health/lifecycle events | `/v1/traces` | `gateway.lifecycle` state transitions (`starting -> running -> draining -> stopped`, `startup_failed`, exit), `gateway.health_snapshot`, platform state changes | | Diagnostics | `/v1/logs` | Warning/error gateway events with a constant body and bounded subsystem, severity, error class, and error code attributes; rendered log messages are never exported | -| Cron scheduler gauges | `/v1/metrics` | Ticker heartbeat and last-success age (omitted when unavailable), enabled/running job counts, and overdue count derived from persisted `next_run_at` plus the scheduler's existing grace rule | -| Cron execution lifecycle | `/v1/traces` | Durable `claimed/running/completed/failed/unknown` states, bounded source and error class, opaque hashed job key, elapsed duration when timestamps exist, and delivery outcome when the scheduler knows it; terminal states flush through a bounded fail-open barrier | +| Cron scheduler gauges | `/v1/metrics` | Ticker heartbeat and last-success age (omitted when unavailable), a monotonic catch-up-occurrence count from the scheduler's stale-window branch, enabled/running job counts, and overdue count derived from persisted `next_run_at` plus the scheduler's existing grace rule | +| Cron execution lifecycle | `/v1/traces` | Durable `claimed/running/completed/failed/unknown` states, bounded source and error class, opaque hashed job key, elapsed duration when timestamps exist, and delivery outcome when the scheduler knows it; terminal states make a fail-open flush attempt that can delay completion by up to one second | Signals carry `service.name`, version, supervision mode, and a stable one-way hash of the install id so an operator can distinguish instances without @@ -99,7 +99,8 @@ python scripts/observability/gateway_health_export_probe.py \ The `hermes monitoring` CLI intentionally exposes `status` only. This first release covers only Hermes Agent-owned service-health and operational-diagnostic -signals. Team Gateway/shared-connector signals are explicitly out of scope, as +signals, including Hermes Agent-owned Relay transport health. Team Gateway's +authoritative shared connector/platform state is explicitly out of scope, as are product analytics, audit/quality reporting, and detailed execution traces. Shared client usage metrics and enterprise trace telemetry are being designed on the NeMo Relay integration with their own consent, policy, and export diff --git a/tests/cron/test_execution_ledger.py b/tests/cron/test_execution_ledger.py index 7d032b06a85..28c9b3a330d 100644 --- a/tests/cron/test_execution_ledger.py +++ b/tests/cron/test_execution_ledger.py @@ -269,6 +269,37 @@ def test_run_one_job_records_running_then_terminal(monkeypatch): assert events[-1][2]["success"] is True +def test_run_one_job_records_unresolved_origin_as_not_configured(monkeypatch): + import cron.scheduler as scheduler + + finished = [] + monkeypatch.setattr(scheduler, "mark_execution_running", lambda _execution_id: None) + monkeypatch.setattr( + scheduler, + "finish_execution", + lambda execution_id, **kwargs: finished.append((execution_id, kwargs)), + ) + monkeypatch.setattr(scheduler, "claim_dispatch", lambda _job_id: True) + monkeypatch.setattr( + scheduler, + "run_job", + lambda job, *, defer_agent_teardown=None: (True, "output", "response", None), + ) + monkeypatch.setattr(scheduler, "save_job_output", lambda *_args: None) + monkeypatch.setattr(scheduler, "_resolve_delivery_targets", lambda _job: []) + monkeypatch.setattr(scheduler, "_deliver_result", lambda *_args, **_kwargs: None) + monkeypatch.setattr(scheduler, "mark_job_run", lambda *_args, **_kwargs: None) + + job = { + "id": "job-unresolved-origin", + "execution_id": "exec-unresolved-origin", + "deliver": "origin", + } + assert scheduler.run_one_job(job) is True + + assert finished[-1][1]["delivery_outcome"] == "not_configured" + + def test_provider_start_recovers_interrupted_records_before_tick(monkeypatch): import cron.scheduler_provider as provider diff --git a/tests/cron/test_jobs.py b/tests/cron/test_jobs.py index 09fc7d2faab..704363a57ea 100644 --- a/tests/cron/test_jobs.py +++ b/tests/cron/test_jobs.py @@ -836,6 +836,24 @@ class TestGetDueJobs: next_dt = _ensure_aware(datetime.fromisoformat(updated["next_run_at"])) assert next_dt > _hermes_now() + def test_stale_past_due_records_one_catch_up_occurrence(self, tmp_cron_dir, monkeypatch): + import cron.jobs as jobs_module + + recorded = [] + monkeypatch.setattr( + jobs_module, + "record_catch_up_occurrence", + lambda: recorded.append("catch-up"), + raising=False, + ) + create_job(prompt="Stale", schedule="every 1h") + jobs = load_jobs() + jobs[0]["next_run_at"] = (datetime.now() - timedelta(minutes=35)).isoformat() + save_jobs(jobs) + + assert len(get_due_jobs()) == 1 + assert recorded == ["catch-up"] + def test_idless_job_does_not_crash_or_block_sibling_jobs(self, tmp_cron_dir): """A job missing its 'id' key must not crash the tick or freeze siblings. diff --git a/tests/monitoring/test_cron_health_export.py b/tests/monitoring/test_cron_health_export.py index 8dfa4d41647..a44fa5ef2fc 100644 --- a/tests/monitoring/test_cron_health_export.py +++ b/tests/monitoring/test_cron_health_export.py @@ -2,6 +2,8 @@ from __future__ import annotations from datetime import datetime, timedelta, timezone +import pytest + def _metric(snapshot, name): return next(metric for metric in snapshot.metrics if metric.name == name) @@ -106,11 +108,52 @@ def test_execution_projection_omits_duration_and_delivery_when_not_known(): ).to_dict() assert event["status"] == "claimed" - assert event["source"] == "unknown" + assert event["source"] == "external" assert event["duration_ms"] is None assert event["delivery_outcome"] is None +def test_external_provider_source_is_normalized_to_external(): + from agent.monitoring.cron_health import project_execution_event + + event = project_execution_event( + {"job_id": "private", "source": "Chronos", "status": "claimed"} + ) + + assert event.source == "external" + + +@pytest.mark.parametrize("message", ["oauth refresh failed", "tokenizer crashed", "HTTP 4015"]) +def test_error_classification_avoids_auth_substring_false_positives(message): + from agent.monitoring.cron_health import classify_cron_error + + assert classify_cron_error(message) == "unknown" + + +@pytest.mark.parametrize( + "message", + ["authentication failed", "not authorized", "access token expired", "HTTP 401"], +) +def test_error_classification_recognizes_auth_terms_and_status_tokens(message): + from agent.monitoring.cron_health import classify_cron_error + + assert classify_cron_error(message) == "auth_failed" + + +def test_cron_snapshot_exports_catch_up_occurrence_counter(monkeypatch): + from agent.monitoring import cron_health + + monkeypatch.setattr(cron_health, "get_ticker_heartbeat_age", lambda: None) + monkeypatch.setattr(cron_health, "get_ticker_success_age", lambda: None) + monkeypatch.setattr(cron_health, "get_running_job_ids", lambda: frozenset()) + monkeypatch.setattr(cron_health, "load_jobs", lambda: []) + monkeypatch.setattr(cron_health, "get_catch_up_occurrence_count", lambda: 3) + + snapshot = cron_health.build_cron_health_snapshot() + + assert _metric(snapshot, "hermes.cron.scheduler.catch_up_occurrences").value == 3 + + def test_terminal_execution_emission_flushes_and_failures_are_fail_open(monkeypatch): from agent.monitoring import cron_health, emitter @@ -151,3 +194,14 @@ def test_gateway_export_includes_cron_metrics_and_only_accepted_event_planes(mon assert gateway_health_export._gateway_health_event({"event": "cron_execution"}) is True assert gateway_health_export._gateway_health_event({"event": "gateway_health"}) is True assert gateway_health_export._gateway_health_event({"event": "run"}) is False + + +def test_monitoring_docs_distinguish_relay_health_scope_and_terminal_flush(): + from pathlib import Path + + text = Path("docs/observability/monitoring.md").read_text(encoding="utf-8") + + assert "Hermes Agent-owned Relay transport health" in text + assert "authoritative shared connector/platform state" in text + assert "up to one second" in text + assert "terminal" in text