diff --git a/cron/jobs.py b/cron/jobs.py index ff536ea7581..a5dc49c1d45 100644 --- a/cron/jobs.py +++ b/cron/jobs.py @@ -7,6 +7,8 @@ Output is saved to ~/.hermes/cron/output/{job_id}/{timestamp}.md import contextlib import copy +from contextvars import ContextVar +from dataclasses import dataclass import json import logging import shutil @@ -62,6 +64,9 @@ except ImportError: # the default root: that re-breaks per-profile isolation. See also the dynamic # `_get_hermes_home()` / `_get_lock_paths()` resolution in cron/scheduler.py. HERMES_DIR = get_hermes_home().resolve() +# These constants remain the default-profile fallback and a compatibility +# surface for existing callers/tests. Cross-profile callers must scope paths +# with use_cron_store() instead of mutating them process-wide. CRON_DIR = HERMES_DIR / "cron" JOBS_FILE = CRON_DIR / "jobs.json" # Heartbeat file the in-process ticker touches on every loop iteration. The @@ -96,6 +101,50 @@ _JOBS_LOCK_TIMEOUT_SECONDS = 30.0 OUTPUT_DIR = CRON_DIR / "output" ONESHOT_GRACE_SECONDS = 120 + +@dataclass(frozen=True) +class _CronStorePaths: + cron_dir: Path + jobs_file: Path + output_dir: Path + + +_cron_store_override: ContextVar[Optional[_CronStorePaths]] = ContextVar( + "cron_store_override", + default=None, +) + + +def _current_cron_store() -> _CronStorePaths: + """Return paths pinned to this execution context's profile.""" + override = _cron_store_override.get() + if override is not None: + return override + return _CronStorePaths(CRON_DIR, JOBS_FILE, OUTPUT_DIR) + + +@contextlib.contextmanager +def use_cron_store(home: Union[str, Path]): + """Route cron storage to ``home`` without mutating process globals.""" + cron_dir = Path(home).expanduser().resolve() / "cron" + token = _cron_store_override.set( + _CronStorePaths( + cron_dir=cron_dir, + jobs_file=cron_dir / "jobs.json", + output_dir=cron_dir / "output", + ) + ) + try: + yield + finally: + _cron_store_override.reset(token) + + +def get_cron_output_dir() -> Path: + """Return the output directory for the active cron store context.""" + return _current_cron_store().output_dir + + # Fallback stale-recovery window for a one-shot's running-claim (#59229) when # the cron inactivity timeout is disabled (HERMES_CRON_TIMEOUT=0 → unlimited), # in which case no finite run bound exists to derive from. Also acts as the @@ -145,7 +194,7 @@ def _oneshot_run_claim_ttl_seconds() -> float: def _jobs_lock_file() -> Path: """Return the advisory lock path for the current cron directory.""" - return CRON_DIR / ".jobs.lock" + return _current_cron_store().cron_dir / ".jobs.lock" @contextlib.contextmanager @@ -265,7 +314,7 @@ def _job_output_dir(job_id: str) -> Path: raise ValueError(f"Invalid cron job id for output path: {job_id!r}") if Path(text).is_absolute() or Path(text).drive: raise ValueError(f"Invalid cron job id for output path: {job_id!r}") - return OUTPUT_DIR / text + return _current_cron_store().output_dir / text def _normalize_skill_list(skill: Optional[str] = None, skills: Optional[Any] = None) -> List[str]: @@ -372,10 +421,11 @@ def _secure_file(path: Path): def ensure_dirs(): """Ensure cron directories exist with secure permissions.""" - CRON_DIR.mkdir(parents=True, exist_ok=True) - OUTPUT_DIR.mkdir(parents=True, exist_ok=True) - _secure_dir(CRON_DIR) - _secure_dir(OUTPUT_DIR) + store = _current_cron_store() + store.cron_dir.mkdir(parents=True, exist_ok=True) + store.output_dir.mkdir(parents=True, exist_ok=True) + _secure_dir(store.cron_dir) + _secure_dir(store.output_dir) # ============================================================================= @@ -685,7 +735,7 @@ def _atomic_write_epoch(path: Path) -> None: torn/truncated file. Best-effort: failures are swallowed by callers. """ ensure_dirs() - fd, tmp_path = tempfile.mkstemp(dir=str(CRON_DIR), suffix=".tmp", prefix=".hb_") + fd, tmp_path = tempfile.mkstemp(dir=str(path.parent), suffix=".tmp", prefix=".hb_") try: with os.fdopen(fd, "w", encoding="utf-8") as f: f.write(str(time.time())) @@ -751,20 +801,21 @@ def get_ticker_success_age() -> Optional[float]: def load_jobs() -> List[Dict[str, Any]]: """Load all jobs from storage.""" + jobs_file = _current_cron_store().jobs_file ensure_dirs() - if not JOBS_FILE.exists(): + if not jobs_file.exists(): return [] _strict_retry = False # track whether we used the strict=False fallback try: - with open(JOBS_FILE, 'r', encoding='utf-8') as f: + with open(jobs_file, 'r', encoding='utf-8') as f: data = json.load(f) except json.JSONDecodeError: # Retry with strict=False to handle bare control chars in string values _strict_retry = True try: - with open(JOBS_FILE, 'r', encoding='utf-8') as f: + with open(jobs_file, 'r', encoding='utf-8') as f: data = json.loads(f.read(), strict=False) except Exception as e: logger.error("Failed to auto-repair jobs.json: %s", e) @@ -799,15 +850,16 @@ def load_jobs() -> List[Dict[str, Any]]: def _save_jobs_unlocked(jobs: List[Dict[str, Any]]): """Save all jobs to storage. Caller must hold _jobs_lock().""" + jobs_file = _current_cron_store().jobs_file ensure_dirs() - fd, tmp_path = tempfile.mkstemp(dir=str(JOBS_FILE.parent), suffix='.tmp', prefix='.jobs_') + fd, tmp_path = tempfile.mkstemp(dir=str(jobs_file.parent), suffix='.tmp', prefix='.jobs_') try: with os.fdopen(fd, 'w', encoding='utf-8') as f: json.dump({"jobs": jobs, "updated_at": _hermes_now().isoformat()}, f, indent=2) f.flush() os.fsync(f.fileno()) - atomic_replace(tmp_path, JOBS_FILE) - _secure_file(JOBS_FILE) + atomic_replace(tmp_path, jobs_file) + _secure_file(jobs_file) except BaseException: try: os.unlink(tmp_path) diff --git a/cron/scheduler.py b/cron/scheduler.py index 9e645d3728f..4606201fc84 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -2213,7 +2213,8 @@ def _build_job_prompt(job: dict, prerun_script: Optional[tuple] = None) -> str: # Inject output from referenced cron jobs as context. context_from = job.get("context_from") if context_from: - from cron.jobs import OUTPUT_DIR + from cron.jobs import get_cron_output_dir + output_dir = get_cron_output_dir() if isinstance(context_from, str): context_from = [context_from] for source_job_id in context_from: @@ -2228,7 +2229,7 @@ def _build_job_prompt(job: dict, prerun_script: Optional[tuple] = None) -> str: ) continue try: - job_output_dir = OUTPUT_DIR / source_job_id + job_output_dir = output_dir / source_job_id if not job_output_dir.exists(): continue # silent skip — no output yet output_files = sorted( diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py index 2ba21a8ed95..3e4d710eb6c 100644 --- a/hermes_cli/web_server.py +++ b/hermes_cli/web_server.py @@ -10052,9 +10052,6 @@ def _validate_dashboard_cron_context_from( ) -_CRON_PROFILE_LOCK = threading.RLock() - - def _cron_profile_dicts() -> List[Dict[str, Any]]: """Return dashboard profile records, falling back to a directory scan.""" from hermes_cli import profiles as profiles_mod @@ -10092,33 +10089,23 @@ def _annotate_cron_job(job: Dict[str, Any], profile: str, home: Path) -> Dict[st def _call_cron_for_profile(target_profile: Optional[str], func_name: str, *args, **kwargs): """Run cron.jobs helpers against the selected profile's cron directory. - cron.jobs keeps CRON_DIR/JOBS_FILE/OUTPUT_DIR as module globals resolved - from the process HERMES_HOME at import time. The dashboard is a single - process that can inspect many profiles, so temporarily retarget those - globals while holding a lock and restore them immediately after the call. + The dashboard is a single process that can inspect many profiles. Route + storage through cron.jobs' execution-context override so dashboard calls + cannot retarget a concurrent desktop ticker's load/save transaction. """ profile_name, home = _cron_profile_home(target_profile) - with _CRON_PROFILE_LOCK: - from cron import jobs as cron_jobs - from hermes_constants import ( - reset_hermes_home_override, - set_hermes_home_override, - ) + from cron import jobs as cron_jobs + from hermes_constants import ( + reset_hermes_home_override, + set_hermes_home_override, + ) - old_cron_dir = cron_jobs.CRON_DIR - old_jobs_file = cron_jobs.JOBS_FILE - old_output_dir = cron_jobs.OUTPUT_DIR - token = set_hermes_home_override(str(home)) - cron_jobs.CRON_DIR = home / "cron" - cron_jobs.JOBS_FILE = cron_jobs.CRON_DIR / "jobs.json" - cron_jobs.OUTPUT_DIR = cron_jobs.CRON_DIR / "output" - try: + token = set_hermes_home_override(str(home)) + try: + with cron_jobs.use_cron_store(home): result = getattr(cron_jobs, func_name)(*args, **kwargs) - finally: - cron_jobs.CRON_DIR = old_cron_dir - cron_jobs.JOBS_FILE = old_jobs_file - cron_jobs.OUTPUT_DIR = old_output_dir - reset_hermes_home_override(token) + finally: + reset_hermes_home_override(token) if isinstance(result, list): return [_annotate_cron_job(j, profile_name, home) for j in result] @@ -10412,31 +10399,18 @@ def _fire_cron_job_for_profile(profile: str, job_id: str) -> bool: """Run ONE due cron job end-to-end for ``profile`` via the resolved scheduler provider's ``fire_due`` (store CAS claim + ``run_one_job``). - Retargets the ``cron.jobs`` module globals to the profile's cron dir under - the shared lock — same mechanism as ``_call_cron_for_profile`` — so the - claim and the run operate on the right profile's ``jobs.json``. Runs with - no live adapters; delivery falls back to the per-platform send path (the - dashboard process has no gateway adapter handles, exactly like the desktop - cron path above). + Uses the same execution-context store routing as ``_call_cron_for_profile`` + so the claim and run operate on the right profile's ``jobs.json`` without + mutating paths observed by the desktop ticker. Runs with no live adapters; + delivery falls back to the per-platform send path. """ _profile_name, home = _cron_profile_home(profile) - with _CRON_PROFILE_LOCK: - from cron import jobs as cron_jobs - from cron.scheduler_provider import resolve_cron_scheduler + from cron import jobs as cron_jobs + from cron.scheduler_provider import resolve_cron_scheduler - old_cron_dir = cron_jobs.CRON_DIR - old_jobs_file = cron_jobs.JOBS_FILE - old_output_dir = cron_jobs.OUTPUT_DIR - cron_jobs.CRON_DIR = home / "cron" - cron_jobs.JOBS_FILE = cron_jobs.CRON_DIR / "jobs.json" - cron_jobs.OUTPUT_DIR = cron_jobs.CRON_DIR / "output" - try: - provider = resolve_cron_scheduler() - return bool(provider.fire_due(job_id, adapters=None, loop=None)) - finally: - cron_jobs.CRON_DIR = old_cron_dir - cron_jobs.JOBS_FILE = old_jobs_file - cron_jobs.OUTPUT_DIR = old_output_dir + with cron_jobs.use_cron_store(home): + provider = resolve_cron_scheduler() + return bool(provider.fire_due(job_id, adapters=None, loop=None)) @app.post("/api/cron/fire") diff --git a/tests/hermes_cli/test_web_server_cron_profiles.py b/tests/hermes_cli/test_web_server_cron_profiles.py index 185b7bac32a..6075a5881bd 100644 --- a/tests/hermes_cli/test_web_server_cron_profiles.py +++ b/tests/hermes_cli/test_web_server_cron_profiles.py @@ -1,5 +1,7 @@ """Regression tests for dashboard cron job profile routing.""" +from concurrent.futures import ThreadPoolExecutor +import json from queue import Empty, SimpleQueue import threading @@ -34,7 +36,7 @@ def _drain_queue(q): return values -def test_call_cron_for_profile_routes_storage_and_restores_globals(isolated_profiles): +def test_call_cron_for_profile_routes_storage_without_mutating_globals(isolated_profiles): from cron import jobs as cron_jobs from hermes_cli import web_server @@ -62,6 +64,91 @@ def test_call_cron_for_profile_routes_storage_and_restores_globals(isolated_prof assert cron_jobs.OUTPUT_DIR == old_output_dir +def test_profile_call_cannot_retarget_ticker_store_mid_write( + isolated_profiles, + monkeypatch, +): + """A dashboard profile call must not redirect a concurrent ticker save.""" + from cron import jobs as cron_jobs + from hermes_cli import web_server + + default_cron = isolated_profiles["default"] / "cron" + worker_cron = isolated_profiles["worker_alpha"] / "cron" + default_file = default_cron / "jobs.json" + worker_file = worker_cron / "jobs.json" + default_job = { + "id": "default-job", + "name": "default job", + "schedule": {"kind": "interval", "minutes": 60}, + "next_run_at": "2026-07-09T00:00:00+00:00", + } + worker_job = { + "id": "worker-job", + "name": "worker job", + "schedule": {"kind": "interval", "minutes": 60}, + "next_run_at": "2026-07-09T00:00:00+00:00", + } + default_file.write_text(json.dumps({"jobs": [default_job]}), encoding="utf-8") + worker_file.write_text(json.dumps({"jobs": [worker_job]}), encoding="utf-8") + + monkeypatch.setattr(cron_jobs, "CRON_DIR", default_cron) + monkeypatch.setattr(cron_jobs, "JOBS_FILE", default_file) + monkeypatch.setattr(cron_jobs, "OUTPUT_DIR", default_cron / "output") + monkeypatch.setattr( + cron_jobs, + "compute_next_run", + lambda _schedule, _last_run_at=None: "2026-07-10T00:00:00+00:00", + ) + + ticker_loaded = threading.Event() + release_ticker = threading.Event() + profile_entered = threading.Event() + ticker_done = threading.Event() + ticker_thread = threading.local() + original_load_jobs = cron_jobs.load_jobs + + def blocking_load_jobs(): + loaded = original_load_jobs() + if getattr(ticker_thread, "active", False): + ticker_loaded.set() + assert release_ticker.wait(5), "profile call did not enter in time" + return loaded + + def hold_profile_call(): + profile_entered.set() + assert ticker_done.wait(5), "ticker did not finish in time" + return True + + def run_ticker_write(): + ticker_thread.active = True + try: + return cron_jobs.advance_next_run("default-job") + finally: + ticker_done.set() + + monkeypatch.setattr(cron_jobs, "load_jobs", blocking_load_jobs) + monkeypatch.setattr(cron_jobs, "_hold_profile_call", hold_profile_call, raising=False) + + with ThreadPoolExecutor(max_workers=2) as pool: + ticker_future = pool.submit(run_ticker_write) + assert ticker_loaded.wait(5), "ticker did not load the default store" + profile_future = pool.submit( + web_server._call_cron_for_profile, + "worker_alpha", + "_hold_profile_call", + ) + assert profile_entered.wait(5), "profile call did not retarget its store" + release_ticker.set() + assert ticker_future.result(timeout=5) is True + assert profile_future.result(timeout=5) is True + + default_saved = json.loads(default_file.read_text(encoding="utf-8"))["jobs"] + worker_saved = json.loads(worker_file.read_text(encoding="utf-8"))["jobs"] + assert [job["id"] for job in worker_saved] == ["worker-job"] + assert [job["id"] for job in default_saved] == ["default-job"] + assert default_saved[0]["next_run_at"] == "2026-07-10T00:00:00+00:00" + + @pytest.mark.asyncio async def test_list_cron_jobs_all_includes_default_and_named_profiles(isolated_profiles): from hermes_cli import web_server