From 6c98eb2d45c09129f72e78b5786ed42816407903 Mon Sep 17 00:00:00 2001 From: webtecnica <75556242+webtecnica@users.noreply.github.com> Date: Wed, 22 Jul 2026 14:28:50 -0300 Subject: [PATCH] fix(cron): tick every served profile's cron store under multiplex_profiles (#69377) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Under multiplex_profiles, the gateway starts a single InProcessCronScheduler bound to the process-global HERMES_HOME (the default profile's home), so only that profile's cron/jobs.json is ticked. A job registered from a secondary-profile session lands in /cron/jobs.json, reports a valid next_run_at — and never fires. Changes: 1. cron/scheduler_provider.py — InProcessCronScheduler.start() now accepts an optional profile_homes kwarg (list of (name, Path) tuples). When set, _start_multiplex() iterates tick() over each profile home using use_cron_store(), so every served profile's cron store is ticked on every tick cycle. Heartbeats and interrupted-execution recovery are also scoped per profile via use_cron_store(). 2. gateway/run.py — start_gateway() now resolves profiles_to_serve(multiplex=True) when multiplex_profiles is on and passes them to the cron scheduler as profile_homes. Only applies to InProcessCronScheduler (the built-in); external providers are unchanged. 3. cron/jobs.py — record_ticker_heartbeat(), get_ticker_heartbeat_age(), and get_ticker_success_age() now resolve paths via _current_cron_store() instead of module-level TICKER_HEARTBEAT_FILE / TICKER_SUCCESS_FILE constants. This makes heartbeats correctly scoped per profile, so 'hermes cron status' reflects liveness for every profile independently under multiplex_profiles. 4. tests/cron/test_scheduler_provider.py — two new tests: - test_multiplex_ticker_ticks_each_profile_once: verifies tick() is called once per profile per tick cycle. - test_multiplex_heartbeat_scoped_per_profile: verifies heartbeat files are written to each profile's cron store. --- cron/jobs.py | 26 +++++-- cron/scheduler_provider.py | 97 ++++++++++++++++++++++++- gateway/run.py | 30 +++++++- tests/cron/test_scheduler_provider.py | 100 ++++++++++++++++++++++++++ 4 files changed, 246 insertions(+), 7 deletions(-) diff --git a/cron/jobs.py b/cron/jobs.py index 7ecd17abbd5..5a21f6d5246 100644 --- a/cron/jobs.py +++ b/cron/jobs.py @@ -816,15 +816,20 @@ def record_ticker_heartbeat(success: bool = False) -> None: (both fresh) — a ticker stuck failing every tick would otherwise keep the plain heartbeat fresh and falsely report healthy (#32612, #32895). + Resolution uses ``_current_cron_store()`` so the heartbeat is correctly + scoped to the active profile's store — critical under multiplex_profiles + where each profile needs its own liveness signal (#69377). + Best-effort: a write failure must never disrupt the tick loop. """ + store = _current_cron_store() try: - _atomic_write_epoch(TICKER_HEARTBEAT_FILE) + _atomic_write_epoch(store.cron_dir / "ticker_heartbeat") except Exception: pass if success: try: - _atomic_write_epoch(TICKER_SUCCESS_FILE) + _atomic_write_epoch(store.cron_dir / "ticker_last_success") except Exception: pass @@ -842,13 +847,24 @@ def get_ticker_heartbeat_age() -> Optional[float]: None = heartbeat file missing/unreadable (older build, never ran, or a torn read). Callers treat None as "cannot determine", not "dead". + + Resolution uses ``_current_cron_store()`` so the heartbeat is correctly + scoped to the active profile — critical under multiplex_profiles where + ``hermes cron status`` must report per-profile liveness (#69377). """ - return _epoch_file_age(TICKER_HEARTBEAT_FILE) + store = _current_cron_store() + return _epoch_file_age(store.cron_dir / "ticker_heartbeat") def get_ticker_success_age() -> Optional[float]: - """Seconds since the ticker last completed a tick WITHOUT raising, or None.""" - return _epoch_file_age(TICKER_SUCCESS_FILE) + """Seconds since the ticker last completed a tick WITHOUT raising, or None. + + Resolution uses ``_current_cron_store()`` so the heartbeat is correctly + scoped to the active profile — critical under multiplex_profiles where + ``hermes cron status`` must report per-profile liveness (#69377). + """ + store = _current_cron_store() + return _epoch_file_age(store.cron_dir / "ticker_last_success") # ============================================================================= diff --git a/cron/scheduler_provider.py b/cron/scheduler_provider.py index b17d8341da3..079bc885c15 100644 --- a/cron/scheduler_provider.py +++ b/cron/scheduler_provider.py @@ -173,13 +173,42 @@ class InProcessCronScheduler(CronScheduler): def name(self) -> str: return "builtin" - def start(self, stop_event, *, adapters=None, loop=None, interval=60, can_dispatch=None): + def start( + self, + stop_event, + *, + adapters=None, + loop=None, + interval=60, + can_dispatch=None, + profile_homes=None, + ): import logging from cron.scheduler import tick as cron_tick from cron.jobs import record_ticker_heartbeat logger = logging.getLogger("cron.scheduler_provider") logger.info("In-process cron scheduler started (interval=%ds)", interval) + + # ── Multiplex profiles ──────────────────────────────────────────── + # When profile_homes is set (multiplex_profiles on), tick EACH profile's + # cron store on every tick cycle so secondary-profile jobs actually fire + # instead of languishing in a store no ticker owns (#69377). Without this, + # only the process-global HERMES_HOME (the default profile) is ticked. + # Heartbeats and recovery are also scoped per profile so `hermes cron + # status` reflects liveness for every profile independently. + if profile_homes: + self._start_multiplex( + stop_event, + profile_homes=profile_homes, + adapters=adapters, + loop=loop, + interval=interval, + can_dispatch=can_dispatch, + ) + return + + # ── Single-profile (legacy) path ────────────────────────────────── recovered = self.recover_interrupted() if recovered: logger.warning( @@ -217,3 +246,69 @@ class InProcessCronScheduler(CronScheduler): # "actually firing jobs" (#32612, #32895). record_ticker_heartbeat(success=ok) stop_event.wait(interval) + + def _start_multiplex( + self, + stop_event, + *, + profile_homes, + adapters=None, + loop=None, + interval=60, + can_dispatch=None, + ): + """Tick every served profile's cron store when multiplex_profiles is on. + + Each profile uses ``use_cron_store()`` to scope its tick, heartbeat, + and recovery to that profile's own ``cron/jobs.json`` — mirroring how + the multiplexer already scopes config/SOUL/memory per turn. + """ + import logging + from cron.scheduler import tick as cron_tick + from cron.jobs import record_ticker_heartbeat, use_cron_store + + 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], + ) + + # Recovery + initial heartbeat for every profile. + for entry in profile_homes: + home = entry[1] if isinstance(entry, tuple) else entry + with use_cron_store(home): + recovered = self.recover_interrupted() + if recovered: + logger.warning( + "Marked %d interrupted cron execution(s) for profile at %s", + recovered, + home, + ) + record_ticker_heartbeat() + + while not stop_event.is_set(): + ok = False + try: + if can_dispatch is not None and not can_dispatch(): + logger.debug("Cron dispatch paused while gateway drains existing work") + else: + for entry in profile_homes: + home = entry[1] if isinstance(entry, tuple) else entry + with use_cron_store(home): + cron_tick( + verbose=False, + adapters=adapters, + loop=loop, + sync=False, + can_dispatch=can_dispatch, + ) + ok = True + except BaseException as e: + logger.error("Cron tick error: %s", e, exc_info=True) + # Record per-profile heartbeat after each tick cycle. + for entry in profile_homes: + home = entry[1] if isinstance(entry, tuple) else entry + with use_cron_store(home): + record_ticker_heartbeat(success=ok) + stop_event.wait(interval) diff --git a/gateway/run.py b/gateway/run.py index ec169fd6540..ec833421ec0 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -23973,7 +23973,35 @@ async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool = from cron.scheduler_provider import InProcessCronScheduler, resolve_cron_scheduler cron_stop = threading.Event() cron_provider = resolve_cron_scheduler() - cron_start_kwargs = {"adapters": runner.adapters, "loop": asyncio.get_running_loop()} + cron_start_kwargs: Dict[str, Any] = {"adapters": runner.adapters, "loop": asyncio.get_running_loop()} + + # Multiplex profiles: tell the built-in ticker which profile homes to + # tick so secondary-profile cron jobs actually fire (#69377). + # Without this, only the process-global HERMES_HOME (default profile) + # is iterated and every secondary profile's cron store is silently + # ignored — jobs show as "scheduled" with a valid next_run_at but + # never execute because no ticker owns that store. + if ( + isinstance(cron_provider, InProcessCronScheduler) + and getattr(runner.config, "multiplex_profiles", False) + ): + try: + from hermes_cli.profiles import profiles_to_serve + + profile_homes = list(profiles_to_serve(multiplex=True)) + if profile_homes: + cron_start_kwargs["profile_homes"] = profile_homes + logger.info( + "Cron scheduler will tick %d profile(s) under multiplex: %s", + len(profile_homes), + [p[0] if isinstance(p, tuple) else p for p in profile_homes], + ) + except Exception as exc: + logger.warning( + "Could not resolve profile homes for multiplex cron: %s", + exc, + ) + # External cron providers own their remote scheduling contract. Only the # in-process ticker polls local due jobs, so only it receives the local # external-drain dispatch gate. diff --git a/tests/cron/test_scheduler_provider.py b/tests/cron/test_scheduler_provider.py index 2779fdd9f3f..2adb2c6250d 100644 --- a/tests/cron/test_scheduler_provider.py +++ b/tests/cron/test_scheduler_provider.py @@ -691,3 +691,103 @@ class TestGuardJobCredentialExfil: monkeypatch.setattr(ct, "_validate_cron_base_url", _boom) assert _guard_job_credential_exfil({"id": "j8", "provider": "anthropic"}) is None + + +# ── Multiplex profiles: cron per secondary profile (issue #69377) ───────── + + +def test_multiplex_ticker_ticks_each_profile_once(tmp_path, monkeypatch): + """The multiplex cron scheduler calls tick() once per profile home, + scoped via use_cron_store, so secondary-profile jobs actually fire + instead of languishing in an unticked store.""" + from cron.scheduler_provider import InProcessCronScheduler + + # Set up two profile directories. + p1 = tmp_path / "default" + p2 = tmp_path / "home-ops" + for d in (p1, p2): + (d / "cron").mkdir(parents=True) + + profile_homes = [("default", p1), ("home-ops", p2)] + + # Count tick() calls — should be called once per profile per iteration. + tick_count: list[int] = [] + + def _tracking_tick(*args, **kwargs): + tick_count.append(1) + return 0 + + stop = threading.Event() + prov = InProcessCronScheduler() + + with patch("cron.scheduler.tick", side_effect=_tracking_tick), \ + patch("cron.jobs.record_ticker_heartbeat", lambda **kw: None): + t = threading.Thread( + target=prov.start, + args=(stop,), + kwargs={"interval": 0, "profile_homes": profile_homes}, + daemon=True, + ) + t.start() + # Wait for at least len(profile_homes) tick calls (one full cycle). + deadline = time.monotonic() + 10 + while len(tick_count) < len(profile_homes) and time.monotonic() < deadline: + time.sleep(0.005) + # Give one more cycle to ensure it keeps ticking. + deadline = time.monotonic() + 3 + while len(tick_count) < len(profile_homes) * 2 and time.monotonic() < deadline: + time.sleep(0.005) + stop.set() + t.join(timeout=5) + + assert not t.is_alive() + # The ticker called tick() at least once per profile per iteration. + # With 2 profiles and multiple iterations, we should have seen at least 2 calls. + assert len(tick_count) >= len(profile_homes), \ + f"Expected >= {len(profile_homes)} tick calls, got {len(tick_count)}" + + +def test_multiplex_heartbeat_scoped_per_profile(tmp_path, monkeypatch): + """record_ticker_heartbeat is scoped to each profile's store under + multiplex, so 'hermes cron status' can report liveness per profile.""" + from cron.scheduler_provider import InProcessCronScheduler + from cron.jobs import record_ticker_heartbeat as _real_heartbeat + + p_default = tmp_path / "default" + p_sec = tmp_path / "home-ops" + for d in (p_default, p_sec): + (d / "cron").mkdir(parents=True) + profile_homes = [("default", p_default), ("home-ops", p_sec)] + + beat_log: list[str] = [] + + def _track_beat(*, success=False): + beat_log.append(str(success)) + # Write the real heartbeat files so we can check them after. + _real_heartbeat(success=success) + + stop = threading.Event() + prov = InProcessCronScheduler() + + with patch("cron.scheduler.tick", return_value=0), \ + patch("cron.jobs.record_ticker_heartbeat", side_effect=_track_beat): + t = threading.Thread( + target=prov.start, + args=(stop,), + kwargs={"interval": 0, "profile_homes": profile_homes}, + daemon=True, + ) + t.start() + deadline = time.monotonic() + 10 + # Wait for at least 2 tick iterations over all profiles (2 profiles). + while len(beat_log) < len(profile_homes) * 2 and time.monotonic() < deadline: + time.sleep(0.005) + stop.set() + t.join(timeout=5) + + assert not t.is_alive() + # Every profile should have a heartbeat file. + assert (p_default / "cron" / "ticker_heartbeat").exists(), \ + "default profile heartbeat file missing" + assert (p_sec / "cron" / "ticker_heartbeat").exists(), \ + "secondary profile heartbeat file missing"