diff --git a/cron/scheduler_provider.py b/cron/scheduler_provider.py index 28964e2301..c78549d785 100644 --- a/cron/scheduler_provider.py +++ b/cron/scheduler_provider.py @@ -299,6 +299,133 @@ def provider_supports_fire_cancel(provider: Any) -> bool: ) +DEFAULT_MISFIRE_GRACE_MINUTES = 10 + + +def _misfire_grace_minutes() -> float: + """Resolve the misfire catch-up grace window from config. + + ``cron.misfire_grace_minutes`` (number, default + ``DEFAULT_MISFIRE_GRACE_MINUTES``). A non-positive value disables the + catch-up sweep entirely. + """ + try: + from hermes_cli.config import cfg_get, load_config + + return float( + cfg_get( + load_config(), + "cron", + "misfire_grace_minutes", + default=DEFAULT_MISFIRE_GRACE_MINUTES, + ) + ) + except Exception: + return float(DEFAULT_MISFIRE_GRACE_MINUTES) + + +def fire_overdue_jobs( + provider: "CronScheduler", + *, + adapters: Any = None, + loop: Any = None, + now: Any = None, +) -> int: + """Fire jobs whose scheduled time passed without an external fire arriving. + + The misfire catch-up half of the hosted fire path. External providers + (Chronos) deliver scheduled fires over HTTP to this process's api_server + adapter; when that hop is down at fire time (gateway restart window, + api_server not bound, scheduler retry budget exhausted), the job's + ``next_run_at`` stays parked in the past and — because external providers + have no local tick loop — nothing ever runs it. The day is silently lost + even though the gateway may be healthy again minutes later. + + Called from the gateway housekeeping loop. Deliberately: + + - **No-op for the built-in provider.** Its tick loop already picks up + past-due jobs via ``get_due_jobs`` — local scheduling self-heals. + - **Routes through the provider's own two-phase fire path** — a + synchronous ``claim_fire`` (store CAS, so a late external retry + landing concurrently is de-duplicated) and then ``fire_claimed`` in + a daemon thread, mirroring the webhook admission pattern. The + housekeeping loop that calls this must never block for the length + of an agent run. Provider-specific re-arm logic (Chronos NAS + one-shots) runs exactly as for a normal fire. + - **Waits out a grace window** (``cron.misfire_grace_minutes``, default + 10, non-positive disables) so the external scheduler's own retry + backoff gets first right to deliver — catch-up is the backstop, not + a race. + - **Operates on the process-global cron store only** — same profile + scoping as the external provider's reconcile. + + Returns the number of jobs this sweep claimed and dispatched. + """ + import logging + import threading + from datetime import datetime + + logger = logging.getLogger("cron.scheduler_provider") + + if isinstance(provider, InProcessCronScheduler): + return 0 + + grace_minutes = _misfire_grace_minutes() + if grace_minutes <= 0: + return 0 + + from cron.jobs import _ensure_aware, _hermes_now, is_job_runnable, load_jobs + + if now is None: + now = _hermes_now() + + fired = 0 + for job in load_jobs(): + if not is_job_runnable(job): + continue + next_run_at = job.get("next_run_at") + if not next_run_at: + continue + try: + due_dt = _ensure_aware(datetime.fromisoformat(next_run_at)) + except (ValueError, TypeError): + continue + overdue_seconds = (now - due_dt).total_seconds() + if overdue_seconds < grace_minutes * 60: + continue + job_id = str(job.get("id") or "") + logger.warning( + "Misfire catch-up: job %s (%s) was due %s (%.0f min overdue) and " + "no external fire arrived — firing locally.", + job_id, + job.get("name") or "unnamed", + next_run_at, + overdue_seconds / 60, + ) + try: + # Two-phase, webhook-style: claim synchronously (fast store + # CAS — losing means an external retry beat us, which is + # fine), then run the job off-thread so the caller's loop is + # never blocked for the length of an agent run. + claimed = provider.claim_fire(job_id) + if claimed is None: + continue + threading.Thread( + target=provider.fire_claimed, + args=(claimed,), + kwargs={"adapters": adapters, "loop": loop}, + daemon=True, + name=f"cron-misfire-{job_id[:12]}", + ).start() + fired += 1 + except Exception as exc: + logger.warning( + "Misfire catch-up failed for job %s: %s: %s", + job_id, type(exc).__name__, exc, + ) + return fired + + def resolve_cron_scheduler() -> "CronScheduler": """Return the active cron scheduler provider. diff --git a/gateway/run.py b/gateway/run.py index a7b8e79e57..1439cc601a 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -29467,7 +29467,7 @@ def _run_planned_stop_watcher( stop_event.wait(poll_interval) -def _start_gateway_housekeeping(stop_event: threading.Event, adapters=None, loop=None, interval: int = 60): +def _start_gateway_housekeeping(stop_event: threading.Event, adapters=None, loop=None, interval: int = 60, cron_provider=None): """Background thread for gateway-only periodic chores (NOT cron). Split out of the historical ``_start_cron_ticker`` so the cron *trigger* @@ -29496,6 +29496,7 @@ def _start_gateway_housekeeping(stop_event: threading.Event, adapters=None, loop CURATOR_EVERY = 60 # ticks — poll hourly (inner gate handles the real cadence) AUTO_ARCHIVE_EVERY = 60 # ticks — poll hourly (state_meta gate owns the real cadence) MEMORY_TRIM_EVERY = 1 # shared helper cooldown bounds actual allocator work + MISFIRE_SWEEP_EVERY = 5 # ticks — every 5 minutes (grace window gates real work) # Every platform media cache prunes on the same hourly cadence — one loop # over (name, cleanup_fn), not a copy-pasted try/except per cache. @@ -29550,6 +29551,27 @@ def _start_gateway_housekeeping(stop_event: threading.Event, adapters=None, loop except Exception as e: logger.debug("Paste sweep error: %s", e) + # Misfire catch-up (external cron providers only): fire jobs whose + # scheduled time passed with no external fire delivered — the + # backstop for a dead loopback fire hop (gateway restart window, + # api_server not bound, scheduler retries exhausted). The helper + # no-ops for the built-in ticker and enforces the + # cron.misfire_grace_minutes window; the store CAS claim de-dupes + # against a late external retry arriving concurrently. + if cron_provider is not None and tick_count % MISFIRE_SWEEP_EVERY == 0: + try: + from cron.scheduler_provider import fire_overdue_jobs + + caught_up = fire_overdue_jobs( + cron_provider, adapters=adapters, loop=loop + ) + if caught_up: + logger.info( + "Misfire catch-up: fired %d overdue job(s)", caught_up + ) + except Exception as e: + logger.debug("Misfire catch-up sweep error: %s", e) + # Curator — piggy-back on the housekeeping loop so long-running # gateways get weekly skill maintenance without needing restarts. # maybe_run_curator() is internally gated by config.interval_hours @@ -30324,7 +30346,11 @@ async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool = housekeeping_thread = threading.Thread( target=_start_gateway_housekeeping, args=(cron_stop,), - kwargs={"adapters": runner.adapters, "loop": asyncio.get_running_loop()}, + kwargs={ + "adapters": runner.adapters, + "loop": asyncio.get_running_loop(), + "cron_provider": cron_provider, + }, daemon=True, name="gateway-housekeeping", ) diff --git a/tests/cron/test_misfire_catchup.py b/tests/cron/test_misfire_catchup.py new file mode 100644 index 0000000000..53ceb31887 --- /dev/null +++ b/tests/cron/test_misfire_catchup.py @@ -0,0 +1,163 @@ +"""Tests for fire_overdue_jobs — misfire catch-up for external cron providers. + +External providers (Chronos) deliver scheduled fires over HTTP; when the +loopback hop is down at fire time and the scheduler's retry budget exhausts, +the job's next_run_at stays parked in the past and nothing ever runs it +(external providers have no local tick loop). fire_overdue_jobs, called from +gateway housekeeping, claims and fires those jobs after a grace window. +""" + +import threading +import time +from datetime import timedelta + +import pytest + +from cron.jobs import _hermes_now, create_job, get_job, load_jobs, save_jobs +from cron.scheduler_provider import ( + CronScheduler, + InProcessCronScheduler, + fire_overdue_jobs, +) + + +@pytest.fixture() +def tmp_cron_dir(tmp_path, monkeypatch): + """Redirect cron storage to a temp directory.""" + monkeypatch.setattr("cron.jobs.CRON_DIR", tmp_path / "cron") + monkeypatch.setattr("cron.jobs.JOBS_FILE", tmp_path / "cron" / "jobs.json") + monkeypatch.setattr("cron.jobs.OUTPUT_DIR", tmp_path / "cron" / "output") + return tmp_path + + +class RecordingProvider(CronScheduler): + """External-provider stand-in: real base-class claim_fire (store CAS), + recorded fire_claimed instead of the blocking run_one_job.""" + + def __init__(self): + self.fired = [] + self._done = threading.Event() + + @property + def name(self): + return "recording" + + def start(self, stop_event, **kw): # pragma: no cover - unused + return None + + def fire_claimed(self, claimed_job, *, adapters=None, loop=None, + cancel_event=None): + self.fired.append(claimed_job["id"]) + self._done.set() + return True + + def wait_fired(self, timeout=5.0): + return self._done.wait(timeout) + + +def _park_in_past(job_id, minutes): + """Rewind a job's next_run_at into the past (simulates missed fires).""" + jobs = load_jobs() + for j in jobs: + if j["id"] == job_id: + j["next_run_at"] = ( + _hermes_now() - timedelta(minutes=minutes) + ).isoformat() + save_jobs(jobs) + + +class TestFireOverdueJobs: + def test_noop_for_builtin_provider(self, tmp_cron_dir): + """The in-process ticker self-heals past-due jobs — the sweep must + never double-dispatch under it.""" + job = create_job(prompt="p", schedule="every 1h") + _park_in_past(job["id"], minutes=60) + assert fire_overdue_jobs(InProcessCronScheduler()) == 0 + + def test_fires_job_past_grace(self, tmp_cron_dir): + job = create_job(prompt="p", schedule="every 1h") + _park_in_past(job["id"], minutes=30) + provider = RecordingProvider() + assert fire_overdue_jobs(provider) == 1 + assert provider.wait_fired() + assert provider.fired == [job["id"]] + + def test_respects_grace_window(self, tmp_cron_dir): + """A job only a few minutes overdue is still the external + scheduler's to retry — the backstop must not race it.""" + job = create_job(prompt="p", schedule="every 1h") + _park_in_past(job["id"], minutes=5) # < default 10 min grace + provider = RecordingProvider() + assert fire_overdue_jobs(provider) == 0 + assert provider.fired == [] + + def test_grace_zero_disables(self, tmp_cron_dir, monkeypatch): + monkeypatch.setattr( + "cron.scheduler_provider._misfire_grace_minutes", lambda: 0.0 + ) + job = create_job(prompt="p", schedule="every 1h") + _park_in_past(job["id"], minutes=600) + assert fire_overdue_jobs(RecordingProvider()) == 0 + + def test_future_job_not_fired(self, tmp_cron_dir): + create_job(prompt="p", schedule="every 1h") # next_run_at in future + provider = RecordingProvider() + assert fire_overdue_jobs(provider) == 0 + + def test_paused_job_not_fired(self, tmp_cron_dir): + from cron.jobs import pause_job + + job = create_job(prompt="p", schedule="every 1h") + _park_in_past(job["id"], minutes=30) + pause_job(job["id"]) + assert fire_overdue_jobs(RecordingProvider()) == 0 + + def test_fresh_external_claim_wins(self, tmp_cron_dir): + """A concurrent external fire holds the store claim — the sweep's + claim_fire loses the CAS and must not dispatch (at-most-once).""" + from cron.jobs import claim_job_for_fire + + job = create_job(prompt="p", schedule="every 1h") + _park_in_past(job["id"], minutes=30) + assert claim_job_for_fire(job["id"]) is True # external fire claims + + # The external claim also advanced next_run_at (recurring bump), so + # re-park it to isolate the claim-CAS as the thing that blocks us. + _park_in_past(job["id"], minutes=30) + + provider = RecordingProvider() + assert fire_overdue_jobs(provider) == 0 + assert provider.fired == [] + + def test_claim_advances_next_run_no_refire(self, tmp_cron_dir): + """After a catch-up fire, the recurring job's next_run_at moved to + the future — the next sweep pass must not fire it again.""" + job = create_job(prompt="p", schedule="every 1h") + _park_in_past(job["id"], minutes=30) + provider = RecordingProvider() + assert fire_overdue_jobs(provider) == 1 + assert provider.wait_fired() + + stamped = get_job(job["id"]) + assert stamped["next_run_at"] > _hermes_now().isoformat() + + provider2 = RecordingProvider() + assert fire_overdue_jobs(provider2) == 0 + + def test_dispatch_is_nonblocking(self, tmp_cron_dir): + """fire_claimed runs off-thread — a slow job must not stall the + sweep (housekeeping loop) for the length of an agent run.""" + + class SlowProvider(RecordingProvider): + def fire_claimed(self, claimed_job, **kw): + time.sleep(3.0) + return super().fire_claimed(claimed_job, **kw) + + job = create_job(prompt="p", schedule="every 1h") + _park_in_past(job["id"], minutes=30) + provider = SlowProvider() + start = time.monotonic() + assert fire_overdue_jobs(provider) == 1 + assert time.monotonic() - start < 1.0 # returned before the run + assert provider.wait_fired(timeout=10) + assert provider.fired == [job["id"]] diff --git a/website/docs/user-guide/features/cron.md b/website/docs/user-guide/features/cron.md index 168d7e57d3..6e12e8e5e3 100644 --- a/website/docs/user-guide/features/cron.md +++ b/website/docs/user-guide/features/cron.md @@ -678,6 +678,18 @@ These misses are stamped on the job record as `last_fire_error` (timestamp + rea The stamp always reflects **current** auto-fire health: it is overwritten by newer misses and cleared automatically by the next successful run. If you see it, the job and its schedule are fine — the gateway side of the fire path needs attention (most commonly, restart the gateway through its supervisor so it loads the full profile environment: `hermes gateway restart`). +### Misfire catch-up + +When an external scheduler provider is active (managed cron on hosted deployments), the gateway also runs a catch-up sweep: a job whose scheduled time passed with no fire delivered — and whose grace window has elapsed — is claimed and run locally, so an outage in the fire hand-off costs minutes instead of the whole day. The sweep is de-duplicated against late scheduler retries by the same store claim used for normal fires. + +```yaml +cron: + misfire_grace_minutes: 10 # wait this long for the scheduler's own retries + # before catching up locally; 0 disables catch-up +``` + +Local (built-in ticker) deployments don't need this — the ticker already picks up past-due jobs on its next tick. + ## Schedule formats The agent's final response is automatically delivered to the job's `deliver:` target — the agent no longer fires messages itself, so the user-facing content simply goes in the final response. To deliver to **additional or different** targets, list multiple `deliver:` targets on the cron job (comma-separated, e.g. `deliver: "telegram,discord"`) rather than having the agent send them.