fix(cron): tick every served profile's cron store under multiplex_profiles (#69377)

Under multiplex_profiles, the gateway starts a single InProcessCronScheduler
bound to the process-global HERMES_HOME (the default profile's home), so
only that profile's cron/jobs.json is ticked. A job registered from a
secondary-profile session lands in <profile>/cron/jobs.json, reports a valid
next_run_at — and never fires.

Changes:

1. cron/scheduler_provider.py — InProcessCronScheduler.start() now accepts
   an optional profile_homes kwarg (list of (name, Path) tuples). When set,
   _start_multiplex() iterates tick() over each profile home using
   use_cron_store(), so every served profile's cron store is ticked on
   every tick cycle. Heartbeats and interrupted-execution recovery are also
   scoped per profile via use_cron_store().

2. gateway/run.py — start_gateway() now resolves profiles_to_serve(multiplex=True)
   when multiplex_profiles is on and passes them to the cron scheduler as
   profile_homes. Only applies to InProcessCronScheduler (the built-in);
   external providers are unchanged.

3. cron/jobs.py — record_ticker_heartbeat(), get_ticker_heartbeat_age(), and
   get_ticker_success_age() now resolve paths via _current_cron_store()
   instead of module-level TICKER_HEARTBEAT_FILE / TICKER_SUCCESS_FILE
   constants. This makes heartbeats correctly scoped per profile, so
   'hermes cron status' reflects liveness for every profile independently
   under multiplex_profiles.

4. tests/cron/test_scheduler_provider.py — two new tests:
   - test_multiplex_ticker_ticks_each_profile_once: verifies tick() is called
     once per profile per tick cycle.
   - test_multiplex_heartbeat_scoped_per_profile: verifies heartbeat files
     are written to each profile's cron store.

cron/jobs.py

@@ -816,15 +816,20 @@ def record_ticker_heartbeat(success: bool = False) -> None:
     (both fresh) — a ticker stuck failing every tick would otherwise keep the
     plain heartbeat fresh and falsely report healthy (#32612, #32895).
 
+    Resolution uses ``_current_cron_store()`` so the heartbeat is correctly
+    scoped to the active profile's store — critical under multiplex_profiles
+    where each profile needs its own liveness signal (#69377).
+
     Best-effort: a write failure must never disrupt the tick loop.
     """
+    store = _current_cron_store()
     try:
-        _atomic_write_epoch(TICKER_HEARTBEAT_FILE)
+        _atomic_write_epoch(store.cron_dir / "ticker_heartbeat")
     except Exception:
         pass
     if success:
         try:
-            _atomic_write_epoch(TICKER_SUCCESS_FILE)
+            _atomic_write_epoch(store.cron_dir / "ticker_last_success")
         except Exception:
             pass
 
@@ -842,13 +847,24 @@ def get_ticker_heartbeat_age() -> Optional[float]:
 
     None = heartbeat file missing/unreadable (older build, never ran, or a
     torn read). Callers treat None as "cannot determine", not "dead".
+
+    Resolution uses ``_current_cron_store()`` so the heartbeat is correctly
+    scoped to the active profile — critical under multiplex_profiles where
+    ``hermes cron status`` must report per-profile liveness (#69377).
     """
-    return _epoch_file_age(TICKER_HEARTBEAT_FILE)
+    store = _current_cron_store()
+    return _epoch_file_age(store.cron_dir / "ticker_heartbeat")
 
 
 def get_ticker_success_age() -> Optional[float]:
-    """Seconds since the ticker last completed a tick WITHOUT raising, or None."""
-    return _epoch_file_age(TICKER_SUCCESS_FILE)
+    """Seconds since the ticker last completed a tick WITHOUT raising, or None.
+
+    Resolution uses ``_current_cron_store()`` so the heartbeat is correctly
+    scoped to the active profile — critical under multiplex_profiles where
+    ``hermes cron status`` must report per-profile liveness (#69377).
+    """
+    store = _current_cron_store()
+    return _epoch_file_age(store.cron_dir / "ticker_last_success")
 
 
 # =============================================================================

cron/scheduler_provider.py

@@ -173,13 +173,42 @@ class InProcessCronScheduler(CronScheduler):
     def name(self) -> str:
         return "builtin"
 
-    def start(self, stop_event, *, adapters=None, loop=None, interval=60, can_dispatch=None):
+    def start(
+        self,
+        stop_event,
+        *,
+        adapters=None,
+        loop=None,
+        interval=60,
+        can_dispatch=None,
+        profile_homes=None,
+    ):
         import logging
         from cron.scheduler import tick as cron_tick
         from cron.jobs import record_ticker_heartbeat
 
         logger = logging.getLogger("cron.scheduler_provider")
         logger.info("In-process cron scheduler started (interval=%ds)", interval)
+
+        # ── Multiplex profiles ────────────────────────────────────────────
+        # When profile_homes is set (multiplex_profiles on), tick EACH profile's
+        # cron store on every tick cycle so secondary-profile jobs actually fire
+        # instead of languishing in a store no ticker owns (#69377). Without this,
+        # only the process-global HERMES_HOME (the default profile) is ticked.
+        # Heartbeats and recovery are also scoped per profile so `hermes cron
+        # status` reflects liveness for every profile independently.
+        if profile_homes:
+            self._start_multiplex(
+                stop_event,
+                profile_homes=profile_homes,
+                adapters=adapters,
+                loop=loop,
+                interval=interval,
+                can_dispatch=can_dispatch,
+            )
+            return
+
+        # ── Single-profile (legacy) path ──────────────────────────────────
         recovered = self.recover_interrupted()
         if recovered:
             logger.warning(
@@ -217,3 +246,69 @@ class InProcessCronScheduler(CronScheduler):
             # "actually firing jobs" (#32612, #32895).
             record_ticker_heartbeat(success=ok)
             stop_event.wait(interval)
+
+    def _start_multiplex(
+        self,
+        stop_event,
+        *,
+        profile_homes,
+        adapters=None,
+        loop=None,
+        interval=60,
+        can_dispatch=None,
+    ):
+        """Tick every served profile's cron store when multiplex_profiles is on.
+
+        Each profile uses ``use_cron_store()`` to scope its tick, heartbeat,
+        and recovery to that profile's own ``cron/jobs.json`` — mirroring how
+        the multiplexer already scopes config/SOUL/memory per turn.
+        """
+        import logging
+        from cron.scheduler import tick as cron_tick
+        from cron.jobs import record_ticker_heartbeat, use_cron_store
+
+        logger = logging.getLogger("cron.scheduler_provider")
+        logger.info(
+            "Multiplex cron scheduler started for %d profile(s): %s",
+            len(profile_homes),
+            [p[0] if isinstance(p, tuple) else p for p in profile_homes],
+        )
+
+        # Recovery + initial heartbeat for every profile.
+        for entry in profile_homes:
+            home = entry[1] if isinstance(entry, tuple) else entry
+            with use_cron_store(home):
+                recovered = self.recover_interrupted()
+                if recovered:
+                    logger.warning(
+                        "Marked %d interrupted cron execution(s) for profile at %s",
+                        recovered,
+                        home,
+                    )
+                record_ticker_heartbeat()
+
+        while not stop_event.is_set():
+            ok = False
+            try:
+                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
+                        with use_cron_store(home):
+                            cron_tick(
+                                verbose=False,
+                                adapters=adapters,
+                                loop=loop,
+                                sync=False,
+                                can_dispatch=can_dispatch,
+                            )
+                ok = True
+            except BaseException as e:
+                logger.error("Cron tick error: %s", e, exc_info=True)
+            # Record per-profile heartbeat after each tick cycle.
+            for entry in profile_homes:
+                home = entry[1] if isinstance(entry, tuple) else entry
+                with use_cron_store(home):
+                    record_ticker_heartbeat(success=ok)
+            stop_event.wait(interval)

gateway/run.py

@@ -23973,7 +23973,35 @@ async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool =
     from cron.scheduler_provider import InProcessCronScheduler, resolve_cron_scheduler
     cron_stop = threading.Event()
     cron_provider = resolve_cron_scheduler()
-    cron_start_kwargs = {"adapters": runner.adapters, "loop": asyncio.get_running_loop()}
+    cron_start_kwargs: Dict[str, Any] = {"adapters": runner.adapters, "loop": asyncio.get_running_loop()}
+
+    # Multiplex profiles: tell the built-in ticker which profile homes to
+    # tick so secondary-profile cron jobs actually fire (#69377).
+    # Without this, only the process-global HERMES_HOME (default profile)
+    # is iterated and every secondary profile's cron store is silently
+    # ignored — jobs show as "scheduled" with a valid next_run_at but
+    # never execute because no ticker owns that store.
+    if (
+        isinstance(cron_provider, InProcessCronScheduler)
+        and getattr(runner.config, "multiplex_profiles", False)
+    ):
+        try:
+            from hermes_cli.profiles import profiles_to_serve
+
+            profile_homes = list(profiles_to_serve(multiplex=True))
+            if profile_homes:
+                cron_start_kwargs["profile_homes"] = profile_homes
+                logger.info(
+                    "Cron scheduler will tick %d profile(s) under multiplex: %s",
+                    len(profile_homes),
+                    [p[0] if isinstance(p, tuple) else p for p in profile_homes],
+                )
+        except Exception as exc:
+            logger.warning(
+                "Could not resolve profile homes for multiplex cron: %s",
+                exc,
+            )
+
     # External cron providers own their remote scheduling contract. Only the
     # in-process ticker polls local due jobs, so only it receives the local
     # external-drain dispatch gate.

tests/cron/test_scheduler_provider.py

@@ -691,3 +691,103 @@ class TestGuardJobCredentialExfil:
 
         monkeypatch.setattr(ct, "_validate_cron_base_url", _boom)
         assert _guard_job_credential_exfil({"id": "j8", "provider": "anthropic"}) is None
+
+
+# ── Multiplex profiles: cron per secondary profile (issue #69377) ─────────
+
+
+def test_multiplex_ticker_ticks_each_profile_once(tmp_path, monkeypatch):
+    """The multiplex cron scheduler calls tick() once per profile home,
+    scoped via use_cron_store, so secondary-profile jobs actually fire
+    instead of languishing in an unticked store."""
+    from cron.scheduler_provider import InProcessCronScheduler
+
+    # Set up two profile directories.
+    p1 = tmp_path / "default"
+    p2 = tmp_path / "home-ops"
+    for d in (p1, p2):
+        (d / "cron").mkdir(parents=True)
+
+    profile_homes = [("default", p1), ("home-ops", p2)]
+
+    # Count tick() calls — should be called once per profile per iteration.
+    tick_count: list[int] = []
+
+    def _tracking_tick(*args, **kwargs):
+        tick_count.append(1)
+        return 0
+
+    stop = threading.Event()
+    prov = InProcessCronScheduler()
+
+    with patch("cron.scheduler.tick", side_effect=_tracking_tick), \
+         patch("cron.jobs.record_ticker_heartbeat", lambda **kw: None):
+        t = threading.Thread(
+            target=prov.start,
+            args=(stop,),
+            kwargs={"interval": 0, "profile_homes": profile_homes},
+            daemon=True,
+        )
+        t.start()
+        # Wait for at least len(profile_homes) tick calls (one full cycle).
+        deadline = time.monotonic() + 10
+        while len(tick_count) < len(profile_homes) and time.monotonic() < deadline:
+            time.sleep(0.005)
+        # Give one more cycle to ensure it keeps ticking.
+        deadline = time.monotonic() + 3
+        while len(tick_count) < len(profile_homes) * 2 and time.monotonic() < deadline:
+            time.sleep(0.005)
+        stop.set()
+        t.join(timeout=5)
+
+    assert not t.is_alive()
+    # The ticker called tick() at least once per profile per iteration.
+    # With 2 profiles and multiple iterations, we should have seen at least 2 calls.
+    assert len(tick_count) >= len(profile_homes), \
+        f"Expected >= {len(profile_homes)} tick calls, got {len(tick_count)}"
+
+
+def test_multiplex_heartbeat_scoped_per_profile(tmp_path, monkeypatch):
+    """record_ticker_heartbeat is scoped to each profile's store under
+    multiplex, so 'hermes cron status' can report liveness per profile."""
+    from cron.scheduler_provider import InProcessCronScheduler
+    from cron.jobs import record_ticker_heartbeat as _real_heartbeat
+
+    p_default = tmp_path / "default"
+    p_sec = tmp_path / "home-ops"
+    for d in (p_default, p_sec):
+        (d / "cron").mkdir(parents=True)
+    profile_homes = [("default", p_default), ("home-ops", p_sec)]
+
+    beat_log: list[str] = []
+
+    def _track_beat(*, success=False):
+        beat_log.append(str(success))
+        # Write the real heartbeat files so we can check them after.
+        _real_heartbeat(success=success)
+
+    stop = threading.Event()
+    prov = InProcessCronScheduler()
+
+    with patch("cron.scheduler.tick", return_value=0), \
+         patch("cron.jobs.record_ticker_heartbeat", side_effect=_track_beat):
+        t = threading.Thread(
+            target=prov.start,
+            args=(stop,),
+            kwargs={"interval": 0, "profile_homes": profile_homes},
+            daemon=True,
+        )
+        t.start()
+        deadline = time.monotonic() + 10
+        # Wait for at least 2 tick iterations over all profiles (2 profiles).
+        while len(beat_log) < len(profile_homes) * 2 and time.monotonic() < deadline:
+            time.sleep(0.005)
+        stop.set()
+        t.join(timeout=5)
+
+    assert not t.is_alive()
+    # Every profile should have a heartbeat file.
+    assert (p_default / "cron" / "ticker_heartbeat").exists(), \
+        "default profile heartbeat file missing"
+    assert (p_sec / "cron" / "ticker_heartbeat").exists(), \
+        "secondary profile heartbeat file missing"