fix(cron): keep deleted profiles from returning

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)
 
 

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)
 

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

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):

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

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()