fix(cron): isolate profile store paths by context

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)
+    store = _current_cron_store()
-    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
+    store.cron_dir.mkdir(parents=True, exist_ok=True)
-    _secure_dir(CRON_DIR)
+    store.output_dir.mkdir(parents=True, exist_ok=True)
-    _secure_dir(OUTPUT_DIR)
+    _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)
+        atomic_replace(tmp_path, jobs_file)
-        _secure_file(JOBS_FILE)
+        _secure_file(jobs_file)
     except BaseException:
         try:
             os.unlink(tmp_path)

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(

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
+    The dashboard is a single process that can inspect many profiles. Route
-    from the process HERMES_HOME at import time. The dashboard is a single
+    storage through cron.jobs' execution-context override so dashboard calls
-    process that can inspect many profiles, so temporarily retarget those
+    cannot retarget a concurrent desktop ticker's load/save transaction.
-    globals while holding a lock and restore them immediately after the call.
     """
     profile_name, home = _cron_profile_home(target_profile)
-    with _CRON_PROFILE_LOCK:
+    from cron import jobs as cron_jobs
-        from cron import jobs as cron_jobs
+    from hermes_constants import (
-        from hermes_constants import (
+        reset_hermes_home_override,
-            reset_hermes_home_override,
+        set_hermes_home_override,
-            set_hermes_home_override,
+    )
-        )
 
-        old_cron_dir = cron_jobs.CRON_DIR
+    token = set_hermes_home_override(str(home))
-        old_jobs_file = cron_jobs.JOBS_FILE
+    try:
-        old_output_dir = cron_jobs.OUTPUT_DIR
+        with cron_jobs.use_cron_store(home):
-        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:
             result = getattr(cron_jobs, func_name)(*args, **kwargs)
-        finally:
+    finally:
-            cron_jobs.CRON_DIR = old_cron_dir
+        reset_hermes_home_override(token)
-            cron_jobs.JOBS_FILE = old_jobs_file
-            cron_jobs.OUTPUT_DIR = old_output_dir
-            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
+    Uses the same execution-context store routing as ``_call_cron_for_profile``
-    the shared lock — same mechanism as ``_call_cron_for_profile`` — so the
+    so the claim and run operate on the right profile's ``jobs.json`` without
-    claim and the run operate on the right profile's ``jobs.json``. Runs with
+    mutating paths observed by the desktop ticker. Runs with no live adapters;
-    no live adapters; delivery falls back to the per-platform send path (the
+    delivery falls back to the per-platform send path.
-    dashboard process has no gateway adapter handles, exactly like the desktop
-    cron path above).
     """
     _profile_name, home = _cron_profile_home(profile)
-    with _CRON_PROFILE_LOCK:
+    from cron import jobs as cron_jobs
-        from cron import jobs as cron_jobs
+    from cron.scheduler_provider import resolve_cron_scheduler
-        from cron.scheduler_provider import resolve_cron_scheduler
 
-        old_cron_dir = cron_jobs.CRON_DIR
+    with cron_jobs.use_cron_store(home):
-        old_jobs_file = cron_jobs.JOBS_FILE
+        provider = resolve_cron_scheduler()
-        old_output_dir = cron_jobs.OUTPUT_DIR
+        return bool(provider.fire_due(job_id, adapters=None, loop=None))
-        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
 
 
 @app.post("/api/cron/fire")

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