ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity

refactor(cron): scheduler_provider module logger, adapter-selection helper, drop dead provider_supports_fire_cancel

Browse Source
This commit is contained in:
Teknium
2026-09-02 19:10:25 -07:00
parent fe160f1847
commit 281cc00db6
2 changed files with 40 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cron/scheduler_provider.py
+39 -110
View File
@@ -7,13 +7,17 @@ from __future__ import annotations
import contextlib
import inspect
import logging
import threading
from abc import ABC, abstractmethod
from pathlib import Path
from typing import Any
logger = logging.getLogger(__name__)
# Cap for exponential tick backoff during fd exhaustion (interval doubled per failure).
_EMFILE_BACKOFF_MAX_SECONDS = 15 * 60 # 15 minutes
_EMFILE_BACKOFF_MAX_SECONDS = 15 * 60
DEFAULT_MISFIRE_GRACE_MINUTES = 10
def _backoff_wait_seconds(interval: float, consecutive_failures: int) -> float:
@@ -203,38 +207,13 @@ def provider_supports_split_fire(provider: Any) -> bool:
return fire_due_impl is None or fire_due_impl is CronScheduler.fire_due
def provider_supports_fire_cancel(provider: Any) -> bool:
"""Return whether ``fire_claimed`` accepts a ``cancel_event`` kwarg."""
try:
parameters = inspect.signature(provider.fire_claimed).parameters.values()
except (TypeError, ValueError):
return False
return any(
parameter.kind is inspect.Parameter.VAR_KEYWORD
or (
parameter.name == "cancel_event"
and parameter.kind
in (inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY)
)
for parameter in parameters
)
DEFAULT_MISFIRE_GRACE_MINUTES = 10
def _misfire_grace_minutes() -> float:
"""``cron.misfire_grace_minutes`` from config; non-positive disables the catch-up sweep."""
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,
)
cfg_get(load_config(), "cron", "misfire_grace_minutes", default=DEFAULT_MISFIRE_GRACE_MINUTES)
)
except Exception:
return float(DEFAULT_MISFIRE_GRACE_MINUTES)
@@ -253,12 +232,8 @@ def fire_overdue_jobs(
concurrent late external retry is de-duplicated by the store CAS; waits out
``cron.misfire_grace_minutes`` so the external retry gets first right. Returns jobs dispatched.
"""
import logging
import threading
from datetime import datetime
logger = logging.getLogger("cron.scheduler_provider")
if isinstance(provider, InProcessCronScheduler):
return 0
@@ -266,7 +241,7 @@ def fire_overdue_jobs(
if grace_minutes <= 0:
return 0
from cron.jobs import _ensure_aware, _hermes_now, is_job_runnable, load_jobs
from cron.jobs import ONESHOT_GRACE_SECONDS, _ensure_aware, _hermes_now, is_job_runnable, load_jobs
if now is None:
now = _hermes_now()
@@ -288,21 +263,18 @@ def fire_overdue_jobs(
job_id = str(job.get("id") or "")
# One-shots past ONESHOT_GRACE_SECONDS "will never fire"; don't resurrect them hours late.
schedule = job.get("schedule") or {}
if str(schedule.get("kind") or "") == "once":
from cron.jobs import ONESHOT_GRACE_SECONDS
if overdue_seconds > ONESHOT_GRACE_SECONDS:
logger.warning(
"Misfire catch-up: one-shot job %s (%s) was due %s "
"(%.0f min overdue) — outside the %ss one-shot grace "
"window, not firing.",
job_id,
job.get("name") or "unnamed",
next_run_at,
overdue_seconds / 60,
ONESHOT_GRACE_SECONDS,
)
continue
if str(schedule.get("kind") or "") == "once" and overdue_seconds > ONESHOT_GRACE_SECONDS:
logger.warning(
"Misfire catch-up: one-shot job %s (%s) was due %s "
"(%.0f min overdue) — outside the %ss one-shot grace "
"window, not firing.",
job_id,
job.get("name") or "unnamed",
next_run_at,
overdue_seconds / 60,
ONESHOT_GRACE_SECONDS,
)
continue
logger.warning(
"Misfire catch-up: job %s (%s) was due %s (%.0f min overdue) and "
"no external fire arrived — firing locally.",
@@ -335,10 +307,6 @@ def fire_overdue_jobs(
def resolve_cron_scheduler() -> "CronScheduler":
"""Resolve ``cron.provider``; missing/failing/unavailable providers fall back to the built-in
with a warning — cron must never be left without a trigger."""
import logging
logger = logging.getLogger("cron.scheduler_provider")
name = ""
try:
from hermes_cli.config import cfg_get, load_config
@@ -372,10 +340,7 @@ def scheduler_for_profile_mode(
stores: fail closed to the built-in multiplex ticker until the API carries profile identity."""
if not multiplex_profiles or isinstance(provider, InProcessCronScheduler):
return provider
import logging
logging.getLogger("cron.scheduler_provider").warning(
logger.warning(
"cron.provider '%s' does not support multiplex_profiles; using built-in ticker",
provider.name,
)
@@ -403,16 +368,10 @@ class InProcessCronScheduler(CronScheduler):
default_profile=None,
profile_gate=None,
):
import logging
from cron.scheduler import CronTickYielded
from cron.scheduler import tick as cron_tick
from cron.jobs import (
clear_ticker_error,
record_ticker_error,
record_ticker_heartbeat,
)
from cron.jobs import clear_ticker_error, record_ticker_error, record_ticker_heartbeat
logger = logging.getLogger("cron.scheduler_provider")
logger.info("In-process cron scheduler started (interval=%ds)", interval)
# Multiplex: tick EACH profile's store every cycle, heartbeats/recovery scoped per profile.
@@ -432,10 +391,7 @@ class InProcessCronScheduler(CronScheduler):
recovered = self.recover_interrupted()
if recovered:
logger.warning(
"Marked %d interrupted cron execution(s) unknown after restart",
recovered,
)
logger.warning("Marked %d interrupted cron execution(s) unknown after restart", recovered)
# Heartbeat before the first sleep so `hermes cron status` sees a live ticker immediately.
record_ticker_heartbeat()
# EMFILE backoff: don't hammer the store while fds are exhausted; a clean tick resets it.
@@ -446,13 +402,7 @@ class InProcessCronScheduler(CronScheduler):
if can_dispatch is not None and not can_dispatch():
logger.debug("Cron dispatch paused while gateway drains existing work")
else:
cron_tick(
verbose=False,
adapters=adapters,
loop=loop,
sync=False,
can_dispatch=can_dispatch,
)
cron_tick(verbose=False, adapters=adapters, loop=loop, sync=False, can_dispatch=can_dispatch)
ok = True
except BaseException as e:
# BaseException, not Exception: a SystemExit must not silently kill the ticker;
@@ -488,7 +438,6 @@ class InProcessCronScheduler(CronScheduler):
"""Tick every profile's store, each scoped via ``_profile_cron_scope``. ``profile_gate(name,
home)``, when given, is consulted every cycle; a rejected profile is neither ticked nor
heartbeated."""
import logging
from cron.scheduler import tick as cron_tick
from cron.scheduler import (
CronTickYielded,
@@ -496,19 +445,25 @@ class InProcessCronScheduler(CronScheduler):
_is_fd_exhaustion,
_primary_profile_routes_for_current_home,
)
from cron.jobs import (
clear_ticker_error,
record_ticker_error,
record_ticker_heartbeat,
)
from cron.jobs import clear_ticker_error, record_ticker_error, record_ticker_heartbeat
logger = logging.getLogger("cron.scheduler_provider")
logger.info(
"Multiplex cron scheduler started for %d profile(s): %s",
len(profile_homes),
[p[0] if isinstance(p, tuple) else p for p in profile_homes],
)
def tick_adapters_for(profile_name):
# Deliver via the profile's OWN adapters; NEVER fall back to the default profile's (wrong
# bot). A credentialless satellite may ride the PRIMARY adapter only for targets an exact
# enabled route maps here; else fail closed (delivery skipped this tick).
if profile_name is None or profile_name == default_profile:
return adapters
tick_adapters = (profile_adapters or {}).get(profile_name) or {}
if not tick_adapters and adapters:
return SharedRouteAdapters(adapters, _primary_profile_routes_for_current_home())
return tick_adapters
# Recovery + heartbeat per profile; one broken store must not abort startup for the others.
for entry in _existing_profile_homes(profile_homes):
_, home = _profile_entry(entry)
@@ -517,18 +472,11 @@ class InProcessCronScheduler(CronScheduler):
recovered = self.recover_interrupted()
if recovered:
logger.warning(
"Marked %d interrupted cron execution(s) for profile at %s",
recovered,
home,
"Marked %d interrupted cron execution(s) for profile at %s", recovered, home
)
record_ticker_heartbeat()
except BaseException as e:
logger.error(
"Cron startup recovery error for profile at %s: %s",
home,
e,
exc_info=True,
)
logger.error("Cron startup recovery error for profile at %s: %s", home, e, exc_info=True)
consecutive_failures = 0
while not stop_event.is_set():
@@ -547,23 +495,9 @@ class InProcessCronScheduler(CronScheduler):
for _pname, home in cycle_homes:
try:
with _profile_cron_scope(home):
# Deliver via the profile's OWN adapters; NEVER fall back to the
# default profile's (wrong bot) — just skip delivery this tick.
if _pname is None or _pname == default_profile:
_tick_adapters = adapters
else:
_tick_adapters = (profile_adapters or {}).get(_pname) or {}
if not _tick_adapters and adapters:
# Credentialless satellite may ride the PRIMARY adapter only
# for targets an exact enabled route maps here; else fail
# closed.
_tick_adapters = SharedRouteAdapters(
adapters,
_primary_profile_routes_for_current_home(),
)
cron_tick(
verbose=False,
adapters=_tick_adapters,
adapters=tick_adapters_for(_pname),
loop=loop,
sync=False,
can_dispatch=can_dispatch,
@@ -574,12 +508,7 @@ class InProcessCronScheduler(CronScheduler):
_profile_errors[str(home)] = f"{type(e).__name__}: {e}"
except BaseException as e:
# THIS profile only; BaseException as in the single-profile loop.
logger.error(
"Cron tick error for profile at %s: %s",
home,
e,
exc_info=True,
)
logger.error("Cron tick error for profile at %s: %s", home, e, exc_info=True)
_profile_errors[str(home)] = f"{type(e).__name__}: {e}"
if _cycle_exc is None or _is_fd_exhaustion(e):
_cycle_exc = e
tests/plugins/test_chronos_cron.py
+1 -6
View File
@@ -204,16 +204,11 @@ def test_chronos_is_split_fire_capable(chronos):
fire webhook uses durable claim admission (not the legacy fire_due path).
Chronos deliberately has NO fire_due override — its re-arm logic lives in
fire_claimed, which the split path invokes."""
from cron.scheduler_provider import (
provider_supports_fire_cancel,
provider_supports_force_fire,
provider_supports_split_fire,
)
from cron.scheduler_provider import provider_supports_force_fire, provider_supports_split_fire
prov, _fake = chronos
assert provider_supports_split_fire(prov) is True
assert provider_supports_force_fire(prov) is True
assert provider_supports_fire_cancel(prov) is True
def test_fire_claimed_no_rearm_when_run_failed(chronos, monkeypatch):