fix(monitoring): correct cron operational signals

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)]

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

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

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:

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

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
 

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.
 

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