fix(cron): yield desktop ticker to running gateway

cron/scheduler.py

@@ -599,6 +599,21 @@ def _get_lock_paths() -> tuple[Path, Path]:
     return lock_dir, lock_dir / ".tick.lock"
 
 
+def _gateway_scheduler_owner_active() -> bool:
+    """Return True when a gateway process owns the scheduler for this profile."""
+    try:
+        from gateway.status import is_gateway_runtime_lock_active
+
+        return is_gateway_runtime_lock_active()
+    except Exception:
+        # Fail open: preserve legacy cron behavior if the owner probe fails.
+        logger.debug(
+            "gateway scheduler-owner check failed; assuming no owner",
+            exc_info=True,
+        )
+        return False
+
+
 def _resolve_origin(job: dict) -> Optional[dict]:
     """Extract origin info from a job, preserving any extra routing metadata.
 
@@ -3892,7 +3907,8 @@ def tick(
     sync: bool = True,
     *,
     can_dispatch=None,
-):
+    defer_to_gateway_owner: bool = False,
+) -> int:
     """
     Check and run all due jobs.
 
@@ -3905,10 +3921,21 @@ def tick(
         loop: Optional asyncio event loop (from gateway) for live adapter sends
         can_dispatch: Optional synchronous gate; false leaves due jobs untouched
             for the next allowed tick
+        defer_to_gateway_owner: When True, skip this tick if a gateway runtime
+            lock already owns scheduler execution for this profile.
 
     Returns:
-        Number of jobs executed (0 if another tick is already running)
+        Number of jobs executed (0 if another tick is already running, or if a
+        gateway owner is handling this profile)
     """
+    if defer_to_gateway_owner and _gateway_scheduler_owner_active():
+        log = logger.info if verbose else logger.debug
+        log(
+            "Cron tick skipped — a gateway owns the scheduler for this profile; "
+            "deferring execution to preserve job provenance"
+        )
+        return 0
+
     lock_dir, lock_file = _get_lock_paths()
     lock_dir.mkdir(parents=True, exist_ok=True)
 

cron/scheduler_provider.py

@@ -165,15 +165,25 @@ class InProcessCronScheduler(CronScheduler):
     ``start()`` blocks in the tick loop until ``stop_event`` is set, identical
     to the pre-refactor ``_start_cron_ticker`` core loop. The caller runs it in
     a daemon thread. ``can_dispatch`` is an optional synchronous gate supplied
-    by GatewayRunner during external drain; skipped ticks leave due jobs intact
-    for the next allowed tick.
+    by GatewayRunner during external drain; ``defer_to_gateway_owner`` lets the
+    desktop fallback yield to a live same-profile gateway. Skipped ticks leave
+    due jobs intact for the next allowed tick.
     """
 
     @property
     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,
+        defer_to_gateway_owner=False,
+    ):
         import logging
         from cron.scheduler import tick as cron_tick
         from cron.jobs import record_ticker_heartbeat
@@ -201,6 +211,7 @@ def start(self, stop_event, *, adapters=None, loop=None, interval=60, can_dispat
                         loop=loop,
                         sync=False,
                         can_dispatch=can_dispatch,
+                        defer_to_gateway_owner=defer_to_gateway_owner,
                     )
                 ok = True
             except BaseException as e:

hermes_cli/web_server.py

@@ -151,16 +151,19 @@ def _start_desktop_cron_ticker(stop_event: "threading.Event", interval: int = 60
     scheduler provider here (no live adapters; delivery falls back to the
     per-platform send path).
 
-    Cross-process safe: the built-in provider's ``cron.scheduler.tick`` takes
-    the ``cron/.tick.lock`` file lock, so this never double-fires alongside a
-    real gateway on the same HERMES_HOME — whichever process grabs the lock
-    first wins the tick.
+    The shared ``cron/.tick.lock`` only guarantees at-most-once execution; it
+    does not guarantee that the correct process ancestry runs the job. The
+    desktop therefore asks the built-in provider to defer each tick while a
+    same-profile gateway owns scheduler execution.
     """
-    from cron.scheduler_provider import resolve_cron_scheduler
+    from cron.scheduler_provider import InProcessCronScheduler, resolve_cron_scheduler
 
     provider = resolve_cron_scheduler()
     _log.info("Desktop cron scheduler started (provider=%s, interval=%ds)", provider.name, interval)
-    provider.start(stop_event, interval=interval)
+    start_kwargs = {"interval": interval}
+    if isinstance(provider, InProcessCronScheduler):
+        start_kwargs["defer_to_gateway_owner"] = True
+    provider.start(stop_event, **start_kwargs)
 
 
 def _warm_gateway_module() -> None:
@@ -198,9 +201,9 @@ async def _lifespan(app: "FastAPI"):
     # the server socket is already open and accepting probes.
     asyncio.get_event_loop().run_in_executor(None, _warm_gateway_module)
 
-    # Desktop-spawned backends (HERMES_DESKTOP=1) fire cron jobs themselves,
-    # since the app has no gateway running the scheduler. Server `hermes
-    # dashboard` is unaffected — it relies on its own gateway.
+    # Desktop-spawned backends (HERMES_DESKTOP=1) keep a minimal ticker alive
+    # so jobs still run when no same-profile gateway exists. The ticker opts
+    # into scheduler-level gateway-owner deferral on each tick.
     cron_stop: "threading.Event | None" = None
     cron_thread: "threading.Thread | None" = None
     if os.getenv("HERMES_DESKTOP") == "1":

tests/cron/test_scheduler_ownership.py

@@ -0,0 +1,84 @@
+import logging
+
+
+def test_gateway_scheduler_owner_helper_delegates_to_runtime_lock(monkeypatch):
+    import cron.scheduler as sched
+
+    monkeypatch.setattr(
+        "gateway.status.is_gateway_runtime_lock_active",
+        lambda: True,
+    )
+
+    assert sched._gateway_scheduler_owner_active() is True
+
+
+def test_gateway_scheduler_owner_helper_fails_open(monkeypatch, caplog):
+    import cron.scheduler as sched
+
+    def _boom():
+        raise RuntimeError("boom")
+
+    monkeypatch.setattr(
+        "gateway.status.is_gateway_runtime_lock_active",
+        _boom,
+    )
+
+    with caplog.at_level(logging.DEBUG):
+        assert sched._gateway_scheduler_owner_active() is False
+
+    assert "gateway scheduler-owner check failed" in caplog.text
+
+
+def test_tick_defers_before_lock_or_job_lookup(monkeypatch):
+    import cron.scheduler as sched
+
+    monkeypatch.setattr(sched, "_gateway_scheduler_owner_active", lambda: True)
+    monkeypatch.setattr(
+        sched,
+        "get_due_jobs",
+        lambda: (_ for _ in ()).throw(AssertionError("should not inspect due jobs")),
+    )
+    monkeypatch.setattr(
+        "builtins.open",
+        lambda *_a, **_k: (_ for _ in ()).throw(
+            AssertionError("should not acquire the cron tick lock")
+        ),
+    )
+
+    assert sched.tick(verbose=False, defer_to_gateway_owner=True) == 0
+
+
+def test_tick_runs_normally_when_no_gateway_owner(tmp_path, monkeypatch):
+    import cron.scheduler as sched
+
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    monkeypatch.setattr(sched, "_gateway_scheduler_owner_active", lambda: False)
+
+    seen = {"due_jobs": 0}
+
+    def _get_due_jobs():
+        seen["due_jobs"] += 1
+        return []
+
+    monkeypatch.setattr(sched, "get_due_jobs", _get_due_jobs)
+
+    assert sched.tick(verbose=False, defer_to_gateway_owner=True) == 0
+    assert seen["due_jobs"] == 1
+
+
+def test_tick_ignores_gateway_owner_without_opt_in(tmp_path, monkeypatch):
+    import cron.scheduler as sched
+
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    monkeypatch.setattr(sched, "_gateway_scheduler_owner_active", lambda: True)
+
+    seen = {"due_jobs": 0}
+
+    def _get_due_jobs():
+        seen["due_jobs"] += 1
+        return []
+
+    monkeypatch.setattr(sched, "get_due_jobs", _get_due_jobs)
+
+    assert sched.tick(verbose=False) == 0
+    assert seen["due_jobs"] == 1

tests/cron/test_scheduler_provider.py

@@ -69,6 +69,7 @@ def fake_tick(*args, **kwargs):
     # Contract: the ticker invokes tick with sync=False (fire-and-forget from
     # the background thread, never the synchronous CLI path).
     assert calls[0].get("sync") is False
+    assert calls[0].get("defer_to_gateway_owner") is False
 
 
 def test_desktop_ticker_calls_tick_then_stops():
@@ -99,6 +100,29 @@ def fake_tick(*args, **kwargs):
     assert not t.is_alive(), "desktop ticker did not exit after stop_event was set"
     assert len(calls) >= 1, "desktop ticker never called tick()"
     assert calls[0].get("sync") is False
+    assert calls[0].get("defer_to_gateway_owner") is True
+
+
+def test_desktop_does_not_pass_builtin_owner_option_to_external_provider():
+    """Desktop keeps built-in tick policy out of external provider contracts."""
+    from hermes_cli.web_server import _start_desktop_cron_ticker
+
+    calls = []
+    stop = threading.Event()
+
+    class ExternalProvider:
+        name = "external"
+
+        def start(self, stop_event, **kwargs):
+            calls.append((stop_event, kwargs))
+
+    with patch(
+        "cron.scheduler_provider.resolve_cron_scheduler",
+        return_value=ExternalProvider(),
+    ):
+        _start_desktop_cron_ticker(stop, interval=17)
+
+    assert calls == [(stop, {"interval": 17})]
 
 
 # ── Phase 1: CronScheduler ABC + InProcessCronScheduler ──────────────────────
@@ -170,6 +194,62 @@ def test_inprocess_provider_ticks_and_stops():
     assert not t.is_alive(), "provider did not exit after stop_event was set"
     assert len(calls) >= 1, "provider never called tick()"
     assert calls[0].get("sync") is False
+    assert calls[0].get("defer_to_gateway_owner") is False
+
+
+def test_inprocess_provider_forwards_gateway_owner_deferral():
+    """The desktop opt-in reaches tick through the provider boundary."""
+    from cron.scheduler_provider import InProcessCronScheduler
+
+    calls = []
+    stop = threading.Event()
+
+    def fake_tick(*args, **kwargs):
+        calls.append(kwargs)
+        stop.set()
+        return 0
+
+    with patch("cron.scheduler.tick", side_effect=fake_tick):
+        InProcessCronScheduler().start(
+            stop,
+            interval=0,
+            defer_to_gateway_owner=True,
+        )
+
+    assert calls[0].get("defer_to_gateway_owner") is True
+
+
+def test_inprocess_provider_runs_when_no_gateway_owner(tmp_path, monkeypatch):
+    """Opting into deferral still executes a tick when no gateway owns it."""
+    import cron.jobs as jobs
+    import cron.scheduler as sched
+    from cron.scheduler_provider import InProcessCronScheduler
+
+    stop = threading.Event()
+    due_jobs_checked = threading.Event()
+
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    monkeypatch.setattr(sched, "_gateway_scheduler_owner_active", lambda: False)
+    monkeypatch.setattr(
+        sched,
+        "get_due_jobs",
+        lambda: due_jobs_checked.set() or [],
+    )
+    monkeypatch.setattr(jobs, "record_ticker_heartbeat", lambda **_kwargs: None)
+    monkeypatch.setattr(InProcessCronScheduler, "recover_interrupted", lambda _self: 0)
+
+    thread = threading.Thread(
+        target=InProcessCronScheduler().start,
+        args=(stop,),
+        kwargs={"interval": 0, "defer_to_gateway_owner": True},
+        daemon=True,
+    )
+    thread.start()
+    assert due_jobs_checked.wait(5), "provider did not execute a tick"
+    stop.set()
+    thread.join(timeout=5)
+
+    assert not thread.is_alive()
 
 
 def test_inprocess_provider_skips_dispatch_while_draining():