feat(monitoring): add cron operational telemetry

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",
+]

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",
 ]

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:

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)

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
 
 

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:

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

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