fix(cron): widen deleted-profile protection to all cron mkdir sites

Replace #96637's inline active_profile_homes() closure with #96508's
module-level _existing_profile_homes() filter (testable in isolation).
Widen _ensure_cron_dir from 3 to 12 mkdir sites across cron/ so every
directory creation fails closed for deleted named profiles, not just
the 3 originally protected. Add _is_named_profile_path() that checks
'profiles' in path parts (works for subdirs like cron/output/<job> and
scripts/ that the original parent.name heuristic couldn't reach).

Co-authored-by: misterdas <das7514@gmail.com>

cron/incidents.py

@@ -53,8 +53,10 @@ _lock = threading.RLock()
 
 
 def _connect() -> sqlite3.Connection:
+    from cron.jobs import _ensure_cron_dir
+
     path = _db_path()
-    path.parent.mkdir(parents=True, exist_ok=True)
+    _ensure_cron_dir(path.parent)
     return sqlite3.connect(path, timeout=5)
 
 

cron/jobs.py

@@ -726,13 +726,30 @@ def _preserve_file_ownership(path: Path, before: Optional[os.stat_result]) -> No
         )
 
 
+def _is_named_profile_path(path: Path) -> bool:
+    """Return True if *path* is inside a named profile home.
+
+    Named profiles live under ``<hermes_home>/profiles/<name>/``.  The
+    default profile lives at ``<hermes_home>`` directly (no ``profiles``
+    parent), as do custom ``HERMES_HOME`` paths outside ``~/.hermes``.
+    """
+    try:
+        return "profiles" in path.resolve().parts
+    except (OSError, RuntimeError):
+        return False
+
+
 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.
+    """Create a cron directory without resurrecting a deleted profile home.
+
+    Named profiles are created by the profile lifecycle, not cron.  A stale
+    multiplex scheduler may still hold a path to a deleted profile after the
+    user removes it; ``parents=False`` makes that race fail closed
+    (FileNotFoundError) instead of silently restoring the directory tree.
+    Default and custom Hermes homes keep ``parents=True`` so first-run
+    directory creation still works.
+    """
+    if _is_named_profile_path(cron_dir):
         cron_dir.mkdir(exist_ok=True)
         return
     cron_dir.mkdir(parents=True, exist_ok=True)
@@ -1094,7 +1111,7 @@ def _record_persisted_error_recovery(job: Dict[str, Any], previous_next_run: str
     del _persisted_error_recoveries_recent[:-_PERSISTED_ERROR_RECOVERY_HISTORY]
     try:
         path = _current_cron_store().cron_dir / "persisted_error_recoveries.jsonl"
-        path.parent.mkdir(parents=True, exist_ok=True)
+        _ensure_cron_dir(path.parent)
         with open(path, "a", encoding="utf-8") as fh:
             fh.write(json.dumps(entry) + "\n")
     except Exception as exc:  # never let telemetry break a tick
@@ -3981,7 +3998,7 @@ def save_job_output(job_id: str, output: str):
     """Save job output to file."""
     ensure_dirs()
     job_output_dir = _job_output_dir(job_id)
-    job_output_dir.mkdir(parents=True, exist_ok=True)
+    _ensure_cron_dir(job_output_dir)
     _secure_dir(job_output_dir)
 
     timestamp = _hermes_now().strftime("%Y-%m-%d_%H-%M-%S")

cron/monitor.py

@@ -102,7 +102,8 @@ def _read_last_output(job_id: str) -> str:
 def _write_last_output(job_id: str, output: str) -> None:
     try:
         path = _snapshot_path(job_id)
-        path.parent.mkdir(parents=True, exist_ok=True)
+        from cron.jobs import _ensure_cron_dir
+        _ensure_cron_dir(path.parent)
         path.write_text(output, encoding="utf-8")
     except Exception as exc:
         logger.warning("Monitor: failed to persist last output for %r: %s", job_id, exc)

cron/notepad.py

@@ -39,7 +39,9 @@ _lock = threading.RLock()
 
 
 def _connect() -> sqlite3.Connection:
-    NOTEPAD_FILE.parent.mkdir(parents=True, exist_ok=True)
+    from cron.jobs import _ensure_cron_dir
+
+    _ensure_cron_dir(NOTEPAD_FILE.parent)
     return sqlite3.connect(NOTEPAD_FILE, timeout=5)
 
 

cron/scheduler.py

@@ -972,7 +972,7 @@ def _record_forced_release(job_id: str, name: str, age_seconds: float, allowance
         del _forced_releases[:-_FORCED_RELEASE_HISTORY]
     try:
         path = _get_hermes_home() / "cron" / "inflight_forced_releases.jsonl"
-        path.parent.mkdir(parents=True, exist_ok=True)
+        _ensure_cron_dir(path.parent)
         with open(path, "a", encoding="utf-8") as fh:
             fh.write(json.dumps(entry) + "\n")
     except Exception as e:  # never let telemetry break a tick
@@ -1513,7 +1513,7 @@ def _write_usage_audit(record: dict) -> None:
     """
     try:
         path = _usage_audit_path()
-        path.parent.mkdir(parents=True, exist_ok=True)
+        _ensure_cron_dir(path.parent)
         line = json.dumps(record, ensure_ascii=False)
         with open(path, "a", encoding="utf-8") as f:
             f.write(line + "\n")
@@ -4311,7 +4311,7 @@ def _run_job_script(
         LLM can report the problem to the user.
     """
     scripts_dir = _get_hermes_home() / "scripts"
-    scripts_dir.mkdir(parents=True, exist_ok=True)
+    _ensure_cron_dir(scripts_dir)
     scripts_dir_resolved = scripts_dir.resolve()
 
     # Same ingestion contract as cron.lifecycle_guard._expand_candidate_path:

cron/scheduler_provider.py

@@ -65,6 +65,31 @@ def _note_tick_failure(exc: BaseException, consecutive_failures: int) -> int:
     return 0
 
 
+def _existing_profile_homes(profile_homes: list) -> list:
+    """Drop profile homes whose directory no longer exists on disk.
+
+    The multiplex ticker's ``profile_homes`` is a snapshot taken at startup
+    (``web_server.py`` calls ``profiles_to_serve(multiplex=True)`` once, and
+    the gateway multiplex path does the same). If a profile is deleted while
+    the ticker runs — via ``hermes profile delete``, the desktop's DELETE
+    ``/api/profiles/<name>`` route, or any other path that removes the home
+    directory — that stale entry stays in the list.
+
+    Ticking or heartbeating a deleted home recreates its ``cron/`` workspace
+    (``record_ticker_heartbeat`` -> ``ensure_dirs`` -> ``mkdir(parents=True)``)
+    on every 60s cycle, so the "deleted" profile silently comes back on disk
+    and in ``hermes profile list`` (#47368). Filtering on directory existence
+    leaves a deleted profile's home untouched, which is the correct invariant:
+    a home that does not exist cannot hold jobs to fire.
+    """
+    live = []
+    for entry in profile_homes:
+        home = entry[1] if isinstance(entry, tuple) else entry
+        if Path(home).is_dir():
+            live.append(entry)
+    return live
+
+
 class CronScheduler(ABC):
     """Axis-B trigger provider. Decides WHEN a due cron job fires.
 
@@ -657,28 +682,12 @@ class InProcessCronScheduler(CronScheduler):
             [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, home in active_profile_homes():
+        # A profile may have been deleted since this snapshot was taken;
+        # never recreate a deleted home's cron workspace via the heartbeat
+        # below (#47368).
+        for entry in _existing_profile_homes(profile_homes):
+            home = entry[1] if isinstance(entry, tuple) else entry
             home_token = set_hermes_home_override(str(home))
             try:
                 with use_cron_store(home):
@@ -701,7 +710,8 @@ class InProcessCronScheduler(CronScheduler):
                 if can_dispatch is not None and not can_dispatch():
                     logger.debug("Cron dispatch paused while gateway drains existing work")
                 else:
-                    for _entry, home in active_profile_homes():
+                    for entry in _existing_profile_homes(profile_homes):
+                        home = entry[1] if isinstance(entry, tuple) else entry
                         home_token = set_hermes_home_override(str(home))
                         try:
                             with use_cron_store(home):
@@ -723,7 +733,8 @@ class InProcessCronScheduler(CronScheduler):
             else:
                 _tick_error = None
             # Record per-profile heartbeat after each tick cycle.
-            for _entry, home in active_profile_homes():
+            for entry in _existing_profile_homes(profile_homes):
+                home = entry[1] if isinstance(entry, tuple) else entry
                 home_token = set_hermes_home_override(str(home))
                 try:
                     with use_cron_store(home):

cron/suggestions.py

@@ -70,7 +70,9 @@ def _secure_file(path: Path) -> None:
 
 
 def _ensure_dir() -> None:
-    CRON_DIR.mkdir(parents=True, exist_ok=True)
+    from cron.jobs import _ensure_cron_dir
+
+    _ensure_cron_dir(CRON_DIR)
 
 
 def _load_raw() -> Dict[str, Any]:

tests/cron/test_jobs.py

@@ -1628,3 +1628,73 @@ class TestCompletedOneshotRetentionSweep:
         assert updated["enabled"] is True
         assert updated["state"] == "scheduled"
         assert updated["next_run_at"] is not None
+
+
+class TestEnsureCronDirWidened:
+    """Tests for the widened _ensure_cron_dir covering all cron mkdir sites."""
+
+    def test_ensure_cron_dir_named_profile_subdir_fails_closed(self, tmp_path):
+        """A subdir under a deleted named profile's cron/ must not recreate it."""
+        import cron.jobs as jobs
+
+        profiles_dir = tmp_path / "profiles"
+        profiles_dir.mkdir()
+        deleted_home = profiles_dir / "deleted"
+        # cron_dir doesn't exist because the profile was deleted
+        output_dir = deleted_home / "cron" / "output" / "job_123"
+
+        import pytest
+        with pytest.raises(FileNotFoundError):
+            jobs._ensure_cron_dir(output_dir)
+        assert not deleted_home.exists()
+
+    def test_ensure_cron_dir_default_home_creates_subdir(self, tmp_path):
+        """A subdir under a default home's cron/ should be created normally."""
+        import cron.jobs as jobs
+
+        default_home = tmp_path / "default_home"
+        default_home.mkdir()
+        output_dir = default_home / "cron" / "output" / "job_123"
+
+        jobs._ensure_cron_dir(output_dir)
+        assert output_dir.is_dir()
+
+    def test_ensure_cron_dir_named_profile_cron_dir_fails_closed(self, tmp_path):
+        """The cron dir of a deleted named profile must not be recreated."""
+        import cron.jobs as jobs
+
+        profiles_dir = tmp_path / "profiles"
+        profiles_dir.mkdir()
+        deleted_home = profiles_dir / "deleted"
+        cron_dir = deleted_home / "cron"
+
+        import pytest
+        with pytest.raises(FileNotFoundError):
+            jobs._ensure_cron_dir(cron_dir)
+        assert not deleted_home.exists()
+
+    def test_ensure_cron_dir_existing_named_profile_cron_dir_works(self, tmp_path):
+        """An existing named profile's cron dir should be created normally."""
+        import cron.jobs as jobs
+
+        profiles_dir = tmp_path / "profiles"
+        active_home = profiles_dir / "active"
+        active_home.mkdir(parents=True)
+        cron_dir = active_home / "cron"
+
+        jobs._ensure_cron_dir(cron_dir)
+        assert cron_dir.is_dir()
+
+    def test_ensure_cron_dir_scripts_dir_under_named_profile_fails_closed(self, tmp_path):
+        """A scripts dir under a deleted named profile must not be recreated."""
+        import cron.jobs as jobs
+
+        profiles_dir = tmp_path / "profiles"
+        profiles_dir.mkdir()
+        deleted_home = profiles_dir / "deleted"
+        scripts_dir = deleted_home / "scripts"
+
+        import pytest
+        with pytest.raises(FileNotFoundError):
+            jobs._ensure_cron_dir(scripts_dir)
+        assert not deleted_home.exists()

tests/cron/test_scheduler_provider.py

@@ -674,3 +674,20 @@ def test_multiplex_ticker_skips_deleted_profile_from_startup_snapshot(tmp_path):
     assert not thread.is_alive()
     assert ticked_homes == [default_home.resolve()]
     assert not deleted_home.exists()
+
+
+def test_existing_profile_homes_filters_deleted(tmp_path):
+    """The existence filter keeps live homes and drops deleted ones, whether
+    entries are (name, path) tuples or bare paths."""
+    from cron.scheduler_provider import _existing_profile_homes
+
+    live = tmp_path / "live"
+    deleted = tmp_path / "deleted"
+    live.mkdir(parents=True)
+    # deleted intentionally not created
+
+    as_tuples = _existing_profile_homes([("live", live), ("deleted", deleted)])
+    assert [p[0] for p in as_tuples] == ["live"]
+
+    as_paths = _existing_profile_homes([live, deleted])
+    assert [p for p in as_paths] == [live]