diff --git a/cron/executions.py b/cron/executions.py index 40a9780700a8d..efadab464a2f4 100644 --- a/cron/executions.py +++ b/cron/executions.py @@ -28,8 +28,10 @@ def _connect() -> sqlite3.Connection: + from cron.jobs import _ensure_cron_dir + path = EXECUTIONS_FILE or (get_hermes_home().resolve() / "cron" / "executions.db") - path.parent.mkdir(parents=True, exist_ok=True) + _ensure_cron_dir(path.parent) return sqlite3.connect(path, timeout=5) diff --git a/cron/jobs.py b/cron/jobs.py index 32d54a22c7cac..c74af5afbb320 100644 --- a/cron/jobs.py +++ b/cron/jobs.py @@ -695,11 +695,23 @@ def _preserve_file_ownership(path: Path, before: Optional[os.stat_result]) -> No ) +def _ensure_cron_dir(cron_dir: Path) -> None: + """Create a cron directory without resurrecting a deleted profile home.""" + profile_home = cron_dir.parent + if profile_home.parent.name == "profiles": + # Named profiles are created by the profile lifecycle, not cron. A + # stale multiplex scheduler may still hold this path after deletion; + # parents=False makes that race fail closed instead of restoring it. + cron_dir.mkdir(exist_ok=True) + return + cron_dir.mkdir(parents=True, exist_ok=True) + + def ensure_dirs(): """Ensure cron directories exist with secure permissions.""" store = _current_cron_store() - store.cron_dir.mkdir(parents=True, exist_ok=True) - store.output_dir.mkdir(parents=True, exist_ok=True) + _ensure_cron_dir(store.cron_dir) + store.output_dir.mkdir(exist_ok=True) _secure_dir(store.cron_dir) _secure_dir(store.output_dir) diff --git a/cron/scheduler.py b/cron/scheduler.py index a96b4175aefaf..86d2f8e493558 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -640,6 +640,7 @@ def _resolve_job_reasoning_config(job: dict, cfg: dict, model: str) -> dict | No } from cron.jobs import ( + _ensure_cron_dir, advance_next_runs, claim_dispatch, claim_job_for_fire, @@ -7687,7 +7688,7 @@ def tick( Number of jobs executed (0 if another tick is already running) """ lock_dir, lock_file = _get_lock_paths() - lock_dir.mkdir(parents=True, exist_ok=True) + _ensure_cron_dir(lock_dir) # Cross-platform file locking: fcntl on Unix, msvcrt on Windows. # Only genuine lock contention (another ticker holds the lock) skips the diff --git a/cron/scheduler_provider.py b/cron/scheduler_provider.py index 8b5281fd67927..dd91f8bbd5bb6 100644 --- a/cron/scheduler_provider.py +++ b/cron/scheduler_provider.py @@ -22,6 +22,7 @@ import inspect import threading from abc import ABC, abstractmethod +from pathlib import Path from typing import Any # Cap for the exponential tick backoff applied while consecutive ticks fail @@ -656,9 +657,28 @@ def _start_multiplex( [p[0] if isinstance(p, tuple) else p for p in profile_homes], ) + removed_profiles = set() + + def active_profile_homes(): + for entry in profile_homes: + if isinstance(entry, tuple): + name, raw_home = entry + home = Path(raw_home) + if name != "default" and not home.is_dir(): + if name not in removed_profiles: + logger.info( + "Skipping removed profile %r in multiplex cron scheduler", + name, + ) + removed_profiles.add(name) + continue + removed_profiles.discard(name) + else: + home = Path(entry) + yield entry, home + # Recovery + initial heartbeat for every profile. - for entry in profile_homes: - home = entry[1] if isinstance(entry, tuple) else entry + for _entry, home in active_profile_homes(): home_token = set_hermes_home_override(str(home)) try: with use_cron_store(home): @@ -681,8 +701,7 @@ def _start_multiplex( 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 + for _entry, home in active_profile_homes(): home_token = set_hermes_home_override(str(home)) try: with use_cron_store(home): @@ -704,8 +723,7 @@ def _start_multiplex( else: _tick_error = None # Record per-profile heartbeat after each tick cycle. - for entry in profile_homes: - home = entry[1] if isinstance(entry, tuple) else entry + for _entry, home in active_profile_homes(): home_token = set_hermes_home_override(str(home)) try: with use_cron_store(home): diff --git a/tests/cron/test_jobs.py b/tests/cron/test_jobs.py index df45211ed42bd..fdc9adb77070c 100644 --- a/tests/cron/test_jobs.py +++ b/tests/cron/test_jobs.py @@ -1128,6 +1128,29 @@ def test_use_cron_store_override_still_wins(self, tmp_path, monkeypatch): store = jobs._current_cron_store() assert store.jobs_file == (tmp_path / "override-home").resolve() / "cron" / "jobs.json" + def test_heartbeat_does_not_recreate_deleted_named_profile(self, tmp_path): + import cron.jobs as jobs + + profiles_dir = tmp_path / "profiles" + profiles_dir.mkdir() + deleted_home = profiles_dir / "deleted" + + with jobs.use_cron_store(deleted_home): + jobs.record_ticker_heartbeat() + + assert not deleted_home.exists() + + def test_heartbeat_initializes_existing_named_profile(self, tmp_path): + import cron.jobs as jobs + + profile_home = tmp_path / "profiles" / "active" + profile_home.mkdir(parents=True) + + with jobs.use_cron_store(profile_home): + jobs.record_ticker_heartbeat() + + assert (profile_home / "cron" / "ticker_heartbeat").is_file() + def test_public_io_after_late_env_repoint_leaves_old_file_untouched( self, tmp_path, monkeypatch diff --git a/tests/cron/test_scheduler_provider.py b/tests/cron/test_scheduler_provider.py index 552fbdf5a11dd..8a59f4497873f 100644 --- a/tests/cron/test_scheduler_provider.py +++ b/tests/cron/test_scheduler_provider.py @@ -640,3 +640,37 @@ def _tracking_tick(*args, **kwargs): f"Expected >= {len(profile_homes)} tick calls, got {len(tick_count)}" +def test_multiplex_ticker_skips_deleted_profile_from_startup_snapshot(tmp_path): + """A stale profile_homes entry must not recreate a deleted profile.""" + import cron.jobs as jobs + from cron.scheduler_provider import InProcessCronScheduler + + default_home = tmp_path / "default" + (default_home / "cron").mkdir(parents=True) + profiles_dir = tmp_path / "profiles" + profiles_dir.mkdir() + deleted_home = profiles_dir / "deleted" + profile_homes = [("default", default_home), ("deleted", deleted_home)] + + ticked_homes = [] + stop = threading.Event() + + def _tracking_tick(*args, **kwargs): + ticked_homes.append(jobs._current_cron_store().cron_dir.parent) + stop.set() + return 0 + + provider = InProcessCronScheduler() + with patch("cron.scheduler.tick", side_effect=_tracking_tick): + thread = threading.Thread( + target=provider.start, + args=(stop,), + kwargs={"interval": 0, "profile_homes": profile_homes}, + daemon=True, + ) + thread.start() + thread.join(timeout=5) + + assert not thread.is_alive() + assert ticked_homes == [default_home.resolve()] + assert not deleted_home.exists()