feat: misfire catch-up for external cron providers
When an external scheduler (Chronos on hosted deployments) cannot deliver a fire — dead loopback hop at fire time, retry budget exhausted — the job's next_run_at stays parked in the past and nothing ever runs it: external providers have no local tick loop, so the day is silently lost even if the gateway heals minutes later (4 consecutive nightly misses in the field). fire_overdue_jobs() in cron/scheduler_provider.py, called from the gateway housekeeping loop every 5 minutes: - No-op for the built-in ticker (its tick loop already self-heals past-due jobs) and when cron.misfire_grace_minutes <= 0. - Waits out a grace window (default 10 min) so the external scheduler's own retry backoff gets first right to deliver. - Claims via the provider's claim_fire (store CAS — a concurrent late external retry is de-duplicated) and runs fire_claimed in a daemon thread, mirroring the webhook admission pattern, so housekeeping never blocks for the length of an agent run. Provider re-arm logic (Chronos NAS one-shots) runs exactly as for a normal fire. Docs: cron.md section + cron.misfire_grace_minutes reference.
This commit is contained in:
@@ -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.
|
||||
|
||||
|
||||
+28
-2
@@ -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",
|
||||
)
|
||||
|
||||
@@ -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"]]
|
||||
@@ -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.
|
||||
|
||||
Reference in New Issue
Block a user