diff --git a/agent/monitoring/cron_health.py b/agent/monitoring/cron_health.py new file mode 100644 index 00000000000..03e8b8dd934 --- /dev/null +++ b/agent/monitoring/cron_health.py @@ -0,0 +1,181 @@ +"""Content-free cron service-health and execution telemetry projection.""" + +from __future__ import annotations + +import hashlib +import logging +from dataclasses import dataclass +from datetime import datetime +from typing import Any, Optional + +from agent.monitoring.events import CronExecutionEvent +from agent.monitoring.gateway_health import GatewayHealthSnapshot, GatewayMetric +from cron.jobs import ( + _compute_grace_seconds, + get_ticker_heartbeat_age, + get_ticker_success_age, + load_jobs, +) +from cron.scheduler import get_running_job_ids +from hermes_time import now as _hermes_now + +logger = logging.getLogger(__name__) +_KNOWN_STATUSES = {"claimed", "running", "completed", "failed", "unknown"} +_KNOWN_SOURCES = {"builtin", "direct", "external"} +_KNOWN_DELIVERY_OUTCOMES = {"delivered", "failed", "suppressed", "not_configured"} + + +@dataclass(frozen=True, slots=True) +class CronHealthSnapshot: + metrics: list[GatewayMetric] + events: list[CronExecutionEvent] + + +def _now() -> datetime: + return _hermes_now() + + +def _job_key(raw: Any) -> str: + value = str(raw or "unknown").encode("utf-8", errors="replace") + return f"sha256:{hashlib.sha256(value).hexdigest()[:24]}" + + +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")): + return "auth_failed" + if "rate limit" in text or "429" in text or "quota" in text: + return "rate_limited" + if "timeout" in text or "timed out" in text: + return "timeout" + if any(value in text for value in ("network", "connection", "dns", "socket", "unreachable")): + return "network_error" + if "dispatch" in text or "executor" in text: + return "dispatch_failed" + if "interrupt" in text or "owner exited" in text or "restarted" in text: + return "interrupted" + if "empty response" in text: + return "empty_response" + if any(value in text for value in ("config", "missing", "invalid")): + return "invalid_config" + return "unknown" + + +def _parse_time(raw: Any) -> Optional[datetime]: + try: + return datetime.fromisoformat(str(raw)) if raw else None + except (TypeError, ValueError): + return None + + +def _duration_ms(record: dict[str, Any]) -> Optional[int]: + start = _parse_time(record.get("started_at")) or _parse_time(record.get("claimed_at")) + finish = _parse_time(record.get("finished_at")) + if start is None or finish is None: + return None + try: + duration = int((finish - start).total_seconds() * 1000) + except (TypeError, ValueError): + return None + return max(0, duration) + + +def project_execution_event( + record: dict[str, Any], *, delivery_outcome: Optional[str] = None +) -> CronExecutionEvent: + status = str(record.get("status") or "unknown").lower() + source = str(record.get("source") or "unknown").lower() + outcome = str(delivery_outcome).lower() if delivery_outcome is not None else None + return CronExecutionEvent( + status=status if status in _KNOWN_STATUSES else "unknown", + job_key=_job_key(record.get("job_id")), + source=source if source in _KNOWN_SOURCES else "unknown", + duration_ms=_duration_ms(record), + delivery_outcome=( + outcome if outcome in _KNOWN_DELIVERY_OUTCOMES else None + ), + error_class=( + classify_cron_error(record.get("error")) + if status in {"failed", "unknown"} + else None + ), + ) + + +def emit_execution_state( + record: Optional[dict[str, Any]], *, delivery_outcome: Optional[str] = None +) -> None: + """Best-effort lifecycle emit; terminal states synchronously cross the queue barrier.""" + if not record: + return + try: + from agent.monitoring import emitter + + event = project_execution_event(record, delivery_outcome=delivery_outcome) + target = emitter.get_emitter() + target.emit(event) + if event.status in {"completed", "failed", "unknown"}: + target.flush(timeout=1.0) + except Exception: + logger.debug("cron execution telemetry emit failed", exc_info=True) + + +def _is_overdue(job: dict[str, Any], now: datetime) -> bool: + if not job.get("enabled", True): + return False + next_run = _parse_time(job.get("next_run_at")) + schedule = job.get("schedule") + if next_run is None or not isinstance(schedule, dict): + return False + try: + if next_run.tzinfo is None and now.tzinfo is not None: + next_run = next_run.replace(tzinfo=now.tzinfo) + lateness = (now - next_run).total_seconds() + return lateness > _compute_grace_seconds(schedule) + except (TypeError, ValueError): + return False + + +def build_cron_health_snapshot() -> CronHealthSnapshot: + metrics: list[GatewayMetric] = [] + for name, reader in ( + ("hermes.cron.scheduler.heartbeat_age_seconds", get_ticker_heartbeat_age), + ("hermes.cron.scheduler.last_success_age_seconds", get_ticker_success_age), + ): + try: + value = reader() + if value is not None: + metrics.append(GatewayMetric(name, max(0.0, float(value)), {})) + except Exception: + logger.debug("cron freshness metric unavailable", exc_info=True) + + try: + jobs = load_jobs() + enabled = [job for job in jobs if job.get("enabled", True)] + metrics.append(GatewayMetric("hermes.cron.jobs.enabled", len(enabled), {})) + metrics.append( + GatewayMetric( + "hermes.cron.jobs.overdue", + sum(1 for job in enabled if _is_overdue(job, _now())), + {}, + ) + ) + except Exception: + logger.debug("cron job metrics unavailable", exc_info=True) + + try: + metrics.append( + GatewayMetric("hermes.cron.jobs.running", len(get_running_job_ids()), {}) + ) + except Exception: + logger.debug("cron running-job metric unavailable", exc_info=True) + return CronHealthSnapshot(metrics=metrics, events=[]) + + +__all__ = [ + "CronHealthSnapshot", + "build_cron_health_snapshot", + "classify_cron_error", + "emit_execution_state", + "project_execution_event", +] diff --git a/agent/monitoring/events.py b/agent/monitoring/events.py index 0d5b964556e..a17f4a49425 100644 --- a/agent/monitoring/events.py +++ b/agent/monitoring/events.py @@ -63,7 +63,24 @@ class GatewayDiagnosticEvent: return {"event": "gateway_diagnostic", **asdict(self)} +@dataclass(slots=True) +class CronExecutionEvent: + """Content-free durable cron execution lifecycle projection.""" + + status: str + job_key: str + source: str = "unknown" + duration_ms: Optional[int] = None + delivery_outcome: Optional[str] = None + error_class: Optional[str] = None + ts_ns: int = field(default_factory=_now_ns) + + def to_dict(self) -> Dict[str, Any]: + return {"event": "cron_execution", **asdict(self)} + + __all__ = [ "GatewayHealthEvent", "GatewayDiagnosticEvent", + "CronExecutionEvent", ] diff --git a/agent/monitoring/gateway_health_export.py b/agent/monitoring/gateway_health_export.py index e10043daba6..a0c851682e8 100644 --- a/agent/monitoring/gateway_health_export.py +++ b/agent/monitoring/gateway_health_export.py @@ -278,7 +278,7 @@ def _supervision_mode() -> str: return "manual" -def _read_runtime_snapshot(config: Dict[str, Any]): +def _read_gateway_snapshot(config: Dict[str, Any]): from agent.monitoring.gateway_health import build_gateway_health_snapshot try: from gateway.status import read_runtime_status @@ -295,6 +295,23 @@ def _read_runtime_snapshot(config: Dict[str, Any]): ) +def _read_cron_snapshot(): + from agent.monitoring.cron_health import build_cron_health_snapshot + + return build_cron_health_snapshot() + + +def _read_runtime_snapshot(config: Dict[str, Any]): + gateway_snapshot = _read_gateway_snapshot(config) + try: + cron_snapshot = _read_cron_snapshot() + except Exception: + logger.debug("cron health snapshot unavailable", exc_info=True) + return gateway_snapshot + gateway_snapshot.metrics.extend(cron_snapshot.metrics) + return gateway_snapshot + + def _emit_snapshot_events(config: Dict[str, Any]) -> None: gh = _gateway_health_config(config) if not gh.get("diagnostic_events_enabled", True): @@ -337,6 +354,11 @@ def _start_metric_provider(config: Dict[str, Any], sdk: Dict[str, Any]) -> Any: "hermes.gateway.restart_requested", "hermes.platform.up", "hermes.platform.degraded", + "hermes.cron.scheduler.heartbeat_age_seconds", + "hermes.cron.scheduler.last_success_age_seconds", + "hermes.cron.jobs.enabled", + "hermes.cron.jobs.running", + "hermes.cron.jobs.overdue", ] def callback(name: str): @@ -467,7 +489,7 @@ def _attach_log_handler(config: Dict[str, Any]) -> Any: def _gateway_health_event(ev: Dict[str, Any]) -> bool: - return ev.get("event") == "gateway_health" + return ev.get("event") in {"gateway_health", "cron_execution"} def start_gateway_health_export(config: Dict[str, Any]) -> GatewayHealthExportRuntime: diff --git a/agent/monitoring/otlp_exporter.py b/agent/monitoring/otlp_exporter.py index db584a3b098..cde95673ec5 100644 --- a/agent/monitoring/otlp_exporter.py +++ b/agent/monitoring/otlp_exporter.py @@ -149,6 +149,8 @@ def _span_attrs(ev: Dict[str, Any]) -> Dict[str, Any]: "gateway_diagnostic": ("name", "subsystem", "error_class", "error_code", "platform", "old_state", "new_state", "version", "severity"), + "cron_execution": ("status", "job_key", "source", "duration_ms", + "delivery_outcome", "error_class"), } for col in keep_by_kind.get(kind, ()): # type: ignore[arg-type] v = ev.get(col) diff --git a/cron/executions.py b/cron/executions.py index 6ffe1eef1fc..d46eaf0d829 100644 --- a/cron/executions.py +++ b/cron/executions.py @@ -65,6 +65,18 @@ def _record(row: Optional[sqlite3.Row]) -> Optional[Dict[str, Any]]: return dict(row) if row is not None else None +def _emit_execution_state( + record: Optional[Dict[str, Any]], *, delivery_outcome: Optional[str] = None +) -> None: + """Project durable state to monitoring without affecting ledger behavior.""" + try: + from agent.monitoring.cron_health import emit_execution_state + + emit_execution_state(record, delivery_outcome=delivery_outcome) + except Exception: + pass + + def _process_start_time(pid: int) -> Optional[int]: try: from gateway.status import get_process_start_time @@ -115,7 +127,9 @@ def create_execution(job_id: str, *, source: str) -> Dict[str, Any]: row = conn.execute( "SELECT * FROM executions WHERE id=?", (execution_id,) ).fetchone() - return _record(row) # type: ignore[return-value] + record = _record(row) + _emit_execution_state(record) + return record # type: ignore[return-value] def mark_execution_running(execution_id: str) -> Optional[Dict[str, Any]]: @@ -129,13 +143,16 @@ def mark_execution_running(execution_id: str) -> Optional[Dict[str, Any]]: ) if cur.rowcount != 1: return None - return _record(conn.execute( + record = _record(conn.execute( "SELECT * FROM executions WHERE id=?", (execution_id,) ).fetchone()) + _emit_execution_state(record) + return record def finish_execution( execution_id: str, *, success: bool, error: Optional[str] = None, + delivery_outcome: Optional[str] = None, ) -> Optional[Dict[str, Any]]: """Write a terminal result once; terminal attempts cannot be rewritten.""" now = _hermes_now().isoformat() @@ -150,15 +167,18 @@ def finish_execution( if cur.rowcount != 1: return None _prune_unlocked(conn) - return _record(conn.execute( + record = _record(conn.execute( "SELECT * FROM executions WHERE id=?", (execution_id,) ).fetchone()) + _emit_execution_state(record, delivery_outcome=delivery_outcome) + return record def recover_interrupted_executions() -> int: """Mark provably abandoned attempts unknown without scheduling retries.""" now = _hermes_now().isoformat() changed = 0 + recovered: List[Dict[str, Any]] = [] with _lock, _connect() as conn: rows = conn.execute( """SELECT id, process_id, pid, process_started_at FROM executions @@ -178,8 +198,16 @@ def recover_interrupted_executions() -> int: row["id"]), ) changed += cur.rowcount + if cur.rowcount: + record = _record(conn.execute( + "SELECT * FROM executions WHERE id=?", (row["id"],) + ).fetchone()) + if record is not None: + recovered.append(record) if changed: _prune_unlocked(conn) + for record in recovered: + _emit_execution_state(record) return changed diff --git a/cron/scheduler.py b/cron/scheduler.py index c309c59c4e5..830c18aabab 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -3977,7 +3977,18 @@ def run_one_job(job: dict, *, adapters=None, loop=None, verbose: bool = False) - if not _consume_interrupted_flag(job["id"]): mark_job_run(job["id"], success, error, delivery_error=delivery_error) - finish_execution(execution_id, success=success, error=error) + if delivery_error: + delivery_outcome = "failed" + elif should_deliver and job.get("deliver", "local") != "local": + delivery_outcome = "delivered" + else: + delivery_outcome = "suppressed" + finish_execution( + execution_id, + success=success, + error=error, + delivery_outcome=delivery_outcome, + ) return True except Exception as e: diff --git a/docs/observability/monitoring.md b/docs/observability/monitoring.md index bb5088929d4..027ac9db8c7 100644 --- a/docs/observability/monitoring.md +++ b/docs/observability/monitoring.md @@ -4,11 +4,12 @@ Service health monitoring plus structured operational diagnostics for the Hermes gateway daemon, exported over OTLP/HTTP to an operator-configured endpoint (OpenTelemetry Collector, DataDog, or any OTLP receiver). -This plane is content-free by construction. It exports gateway lifecycle -state, platform connector health, and content-free warning/error diagnostics. -It never exports prompts, messages, tool arguments or results, session -history, usage analytics, audit logs, or execution traces. Run/model/tool -trajectory capture is a separate plane served by the NeMo Relay integration +This plane is content-free by construction. It exports gateway and cron +lifecycle state, platform connector health, and content-free warning/error +diagnostics. It never exports prompts, messages, tool arguments or results, +job names, destinations, schedules, raw errors, session history, usage +analytics, audit logs, or detailed execution traces. Run/model/tool trajectory +capture is a separate plane served by the NeMo Relay integration (`plugins/observability/nemo_relay/`) and its Hermes-owned subscribers. ## What gets exported @@ -18,6 +19,8 @@ trajectory 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 | Signals carry `service.name`, version, supervision mode, and a stable one-way hash of the install id so an operator can distinguish instances without @@ -94,8 +97,11 @@ python scripts/observability/gateway_health_export_probe.py \ ## Boundaries and roadmap -The `hermes monitoring` CLI intentionally exposes `status` only. Shared -client usage metrics and enterprise trace telemetry are being designed on +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 +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 boundaries; this monitoring plane stays narrow so an operator can enable it without touching any content-bearing signal. The telemetry surface may be diff --git a/tests/monitoring/test_cron_health_export.py b/tests/monitoring/test_cron_health_export.py new file mode 100644 index 00000000000..8dfa4d41647 --- /dev/null +++ b/tests/monitoring/test_cron_health_export.py @@ -0,0 +1,153 @@ +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + + +def _metric(snapshot, name): + return next(metric for metric in snapshot.metrics if metric.name == name) + + +def test_cron_snapshot_projects_freshness_counts_and_overdue_without_content(monkeypatch): + from agent.monitoring import cron_health + + now = datetime(2026, 7, 24, 12, 0, tzinfo=timezone.utc) + secret = "Quarterly payroll for alice@example.com" + monkeypatch.setattr(cron_health, "_now", lambda: now) + monkeypatch.setattr(cron_health, "get_ticker_heartbeat_age", lambda: 4.5) + monkeypatch.setattr(cron_health, "get_ticker_success_age", lambda: 9.0) + monkeypatch.setattr(cron_health, "get_running_job_ids", lambda: frozenset({"job-private-1"})) + monkeypatch.setattr( + cron_health, + "load_jobs", + lambda: [ + { + "id": "job-private-1", + "name": secret, + "prompt": secret, + "enabled": True, + "schedule": {"kind": "interval", "minutes": 10}, + "next_run_at": (now - timedelta(minutes=6)).isoformat(), + }, + { + "id": "job-private-2", + "name": "disabled private job", + "enabled": False, + "schedule": {"kind": "interval", "minutes": 10}, + "next_run_at": (now - timedelta(days=1)).isoformat(), + }, + ], + ) + + snapshot = cron_health.build_cron_health_snapshot() + + assert _metric(snapshot, "hermes.cron.scheduler.heartbeat_age_seconds").value == 4.5 + assert _metric(snapshot, "hermes.cron.scheduler.last_success_age_seconds").value == 9.0 + assert _metric(snapshot, "hermes.cron.jobs.enabled").value == 1 + assert _metric(snapshot, "hermes.cron.jobs.running").value == 1 + assert _metric(snapshot, "hermes.cron.jobs.overdue").value == 1 + assert secret not in str(snapshot) + assert "job-private-1" not in str(snapshot) + + +def test_cron_snapshot_omits_unknown_freshness_instead_of_inventing_values(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: []) + + names = {metric.name for metric in cron_health.build_cron_health_snapshot().metrics} + + assert "hermes.cron.scheduler.heartbeat_age_seconds" not in names + assert "hermes.cron.scheduler.last_success_age_seconds" not in names + + +def test_execution_projection_is_opaque_bounded_and_content_free(): + from agent.monitoring.cron_health import project_execution_event + + event = project_execution_event( + { + "id": "execution-private-id", + "job_id": "Payroll for alice@example.com and token top-secret-token", + "source": "builtin", + "status": "failed", + "claimed_at": "2026-07-24T12:00:00+00:00", + "started_at": "2026-07-24T12:00:01+00:00", + "finished_at": "2026-07-24T12:00:03.250000+00:00", + "error": "Bearer top-secret-token rejected for alice@example.com", + }, + delivery_outcome="failed", + ).to_dict() + + assert event["event"] == "cron_execution" + assert event["status"] == "failed" + assert event["job_key"].startswith("sha256:") + assert len(event["job_key"]) == len("sha256:") + 24 + assert event["duration_ms"] == 2250 + assert event["delivery_outcome"] == "failed" + assert event["error_class"] == "auth_failed" + assert "job_id" not in event + assert "error" not in event + assert "alice@example.com" not in str(event) + assert "top-secret-token" not in str(event) + + +def test_execution_projection_omits_duration_and_delivery_when_not_known(): + from agent.monitoring.cron_health import project_execution_event + + event = project_execution_event( + { + "job_id": "private", + "source": "external-value-must-not-leak", + "status": "claimed", + "claimed_at": "2026-07-24T12:00:00+00:00", + } + ).to_dict() + + assert event["status"] == "claimed" + assert event["source"] == "unknown" + assert event["duration_ms"] is None + assert event["delivery_outcome"] is None + + +def test_terminal_execution_emission_flushes_and_failures_are_fail_open(monkeypatch): + from agent.monitoring import cron_health, emitter + + calls = [] + + class FakeEmitter: + def emit(self, event): + calls.append(("emit", event.to_dict()["status"])) + + def flush(self, timeout): + calls.append(("flush", timeout)) + raise RuntimeError("collector unavailable") + + monkeypatch.setattr(emitter, "get_emitter", lambda: FakeEmitter()) + + cron_health.emit_execution_state( + {"job_id": "private", "source": "builtin", "status": "completed"} + ) + + assert calls == [("emit", "completed"), ("flush", 1.0)] + + +def test_gateway_export_includes_cron_metrics_and_only_accepted_event_planes(monkeypatch): + from agent.monitoring import gateway_health_export + + gateway_snapshot = type("Snapshot", (), {"metrics": []})() + cron_snapshot = type( + "Snapshot", + (), + {"metrics": [type("Metric", (), {"name": "hermes.cron.jobs.enabled", "value": 2, "attributes": {}})()]}, + )() + monkeypatch.setattr(gateway_health_export, "_read_gateway_snapshot", lambda config: gateway_snapshot) + monkeypatch.setattr(gateway_health_export, "_read_cron_snapshot", lambda: cron_snapshot) + + snapshot = gateway_health_export._read_runtime_snapshot({}) + + assert [metric.name for metric in snapshot.metrics] == ["hermes.cron.jobs.enabled"] + 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