Desktop and gateway cron scheduler concurrency conflicts

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #59524 fix(cron): durable run-claim prevents one-shot double-execution across concurrent schedulers

  merged to main

  files: cron/jobs.py, scripts/release.py, tests/cron/test_jobs.py

  Show the diff

  • cron/jobs.py

    @@ -87,6 +87,16 @@
     OUTPUT_DIR = CRON_DIR / "output"
     ONESHOT_GRACE_SECONDS = 120
     
    +# How long a one-shot's running-claim (#59229) is honored before it is
    +# considered stale and the job may be re-dispatched. The claim's real job is
    +# to be cleared by mark_job_run() the moment the run completes (success or
    +# failure); this TTL is only a safety valve for a claiming tick that DIED
    +# mid-run (gateway kill, OOM, hard-timeout) so a one-shot is never wedged
    +# forever. It must exceed the longest legitimate run: the default cron
    +# inactivity timeout is 600s and a job that keeps producing output can run
    +# past that, so 30 min gives generous headroom over any healthy run.
    +ONESHOT_RUN_CLAIM_TTL_SECONDS = 1800
    +
     
     def _jobs_lock_file() -> Path:
         """Return the advisory lock path for the current cron directory."""
    @@ -1306,6 +1316,11 @@ def mark_job_run(job_id: str, success: bool, error: Optional[str] = None,
                     # Clear any external-fire claim so a re-armed recurring job can
                     # be claimed again on its next fire (Phase 4C CAS).
                     job["fire_claim"] = None
    +                # Clear the one-shot running-claim (#59229): the run is over, so
    +                # a re-armed recurring job or a re-dispatched one-shot recovery
    +                # is claimable again. No-op if the job never carried a claim.
    +                if job.get("run_claim") is not None:
    +                    job["run_claim"] = None
                     
                     # Increment completed count.  Finite one-shot jobs are
                     # pre-claimed by claim_dispatch() BEFORE the side effect runs
    @@ -1559,6 +1574,24 @@ def _get_due_jobs_locked() -> List[Dict[str, Any]]:
             if not job.get("enabled", True):
                 continue
     
    +        # Cross-process running-claim guard (#59229): if another scheduler
    +        # process already claimed this one-shot and its run is still in flight
    +        # (claim younger than the TTL), skip it — do NOT re-dispatch. The
    +        # claim is stamped just before we return the job as due (below) and
    +        # cleared by mark_job_run() on completion. A claim older than the TTL
    +        # is treated as stale (the claiming tick died mid-run) and allowed
    +        # through so the job is recovered rather than wedged forever.
    +        existing_claim = job.get("run_claim")
    +        if existing_claim and job.get("schedule", {}).get("kind") == "once":
    +            try:
    +                claimed_at = _ensure_aware(
    +                    datetime.fromisoformat(existing_claim["at"])
    +                )
    +                if (now - claimed_at).total_seconds() < ONESHOT_RUN_CLAIM_TTL_SECONDS:
    +                    continue  # a fresh claim is held by an in-flight run
    +            except (KeyError, ValueError, TypeError):
    +                pass  # malformed claim → fall through and (re)claim
    +
             next_run = job.get("next_run_at")
             if not next_run:
                 schedule = job.get("schedule", {})
    @@ -1703,6 +1736,30 @@ def _get_due_jobs_locked() -> List[Dict[str, Any]]:
                                     break
                             continue
     
    +            # Durably claim a one-shot for the DURATION of its run before
    +            # returning it as due, so a second scheduler process (gateway +
    +            # desktop both run in-process 60s tickers on one HERMES_HOME)
    +            # cannot re-dispatch it while the first run is still in flight
    +            # (#59229). A plain one-shot's due-state is not resolved until
    +            # mark_job_run() completes it minutes later, so advancing
    +            # next_run_at by a fixed window is not enough — a job that outlives
    +            # one tick (e.g. a 2.5-min research prompt) would simply re-fire on
    +            # the next tick after the window. Instead we stamp a run_claim under
    +            # the same lock get_due_jobs already holds; the other process reads
    +            # a fresh claim on its next tick and skips (handled at the top of
    +            # this loop). mark_job_run() clears the claim on completion. The TTL
    +            # is only a safety valve: a claiming tick that DIES mid-run leaves a
    +            # stale claim that expires after ONESHOT_RUN_CLAIM_TTL_SECONDS, so
    +            # the job is re-dispatched rather than wedged forever.
    +            if kind == "once":
    +                claim = {"at": now.isoformat(), "by": _machine_id()}
    +                job["run_claim"] = claim
    +                for rj in raw_jobs:
    +                    if rj["id"] == job["id"]:
    +                        rj["run_claim"] = claim
    +                        needs_save = True
    +                        break
    +
                 due.append(job)
     
         if needs_save:

  • scripts/release.py

    @@ -45,7 +45,7 @@
     
     # Auto-extracted from noreply emails + manual overrides
     AUTHOR_MAP = {
    -    "[redacted:email]": "isheng-eqi",  # PR #59428 salvage (cron: reject past one-shot timestamps in update_job fallback + resume_job; #59395)
    +    "[redacted:email]": "isheng-eqi",  # PR #59428 salvage (cron: reject past one-shot timestamps in update_job fallback + resume_job; #59395). Also PR #59446 salvage (cron: advance one-shot next_run_at before dispatch so concurrent gateway+desktop schedulers can't double-execute; #59229).
         "[redacted:email]": "derek2000139",  # PR #57838 salvage (desktop/windows: pre-write update marker before quit dwell so the renderer's waitForUpdateToFinish gate parks instead of respawning a backend that re-locks venv .pyd files mid-update)
         "[email protected]": "AndreasHiltner",  # PR #56854 salvage (gateway: route multiplex profile responses through the profile's own adapter — 53-site _adapter_for_source sweep)
         "[email protected]": "allenliang2022",  # PR #56932 test coverage folded into #56909 salvage (408 → retryable timeout)

  • tests/cron/test_jobs.py

    @@ -927,7 +927,13 @@ def test_broken_recent_one_shot_without_next_run_is_recovered(self, tmp_cron_dir
             due = get_due_jobs()
     
             assert [job["id"] for job in due] == ["oneshot-recover"]
    -        assert get_job("oneshot-recover")["next_run_at"] == run_at
    +        # Recovery restores next_run_at to the original run time; the
    +        # cross-process double-exec guard (#59229) is a separate run_claim
    +        # stamped under the lock, not a next_run_at mutation.
    +        recovered = get_job("oneshot-recover")
    +        assert recovered["next_run_at"] == run_at
    +        assert recovered.get("run_claim") is not None
    +        assert recovered["run_claim"]["at"] == now.isoformat()
     
         def test_broken_stale_one_shot_without_next_run_is_not_recovered(self, tmp_cron_dir, monkeypatch):
             now = datetime(2026, 3, 18, 4, 30, 0, tzinfo=timezone.utc)
    @@ -958,6 +964,80 @@ def test_broken_stale_one_shot_without_next_run_is_not_recovered(self, tmp_cron_
             assert get_due_jobs() == []
             assert get_job("oneshot-stale")["next_run_at"] is None
     
    +    def test_one_shot_not_redispatched_while_running(self, tmp_cron_dir, monkeypatch):
    +        """#59229: two concurrent schedulers must not double-execute a one-shot.
    +
    +        Reproduces the reported failure with a job whose run OUTLIVES the tick
    +        interval (a ~2.5-min research prompt). Process A's tick returns it as
    +        due and stamps a run_claim; while A is still running, every later tick
    +        (process B, or A's own next tick) must see the fresh claim and skip —
    +        not just for one tick window but for the whole run.
    +        """
    +        from cron.jobs import _hermes_now
    +        t0 = _hermes_now()
    +        run_at = (t0 - timedelta(seconds=5)).isoformat()
    +        save_jobs([{
    +            "id": "long-oneshot", "name": "R", "prompt": "2.5min research",
    +            "schedule": {"kind": "once", "run_at": run_at},
    +            "next_run_at": run_at, "enabled": True, "state": "scheduled",
    +        }])
    +
    +        # Process A tick: picks it up + claims it.
    +        dueA = get_due_jobs()
    +        assert [j["id"] for j in dueA] == ["long-oneshot"]
    +        assert get_job("long-oneshot").get("run_claim") is not None
    +
    +        # Process B (and A's own subsequent ticks) while A is still running:
    +        # 28s later (the exact gap in the report) AND 61s later (past any
    +        # fixed +60s window) — both must skip.
    +        for gap in (28, 61, 130):
    +            monkeypatch.setattr("cron.jobs._hermes_now",
    +                                lambda t0=t0, g=gap: t0 + timedelta(seconds=g))
    +            assert get_due_jobs() == [], f"double-dispatched at +{gap}s"
    +
    +    def test_one_shot_run_claim_expires_after_ttl(self, tmp_cron_dir, monkeypatch):
    +        """A claiming tick that DIED mid-run must not wedge the one-shot forever:
    +        once the run_claim is older than the TTL it is re-dispatched (recovered)."""
    +        from cron.jobs import _hermes_now, ONESHOT_RUN_CLAIM_TTL_SECONDS
    +        t0 = _hermes_now()
    +        run_at = (t0 - timedelta(seconds=5)).isoformat()
    +        save_jobs([{
    +            "id": "wedged", "name": "R", "prompt": "x",
    +            "schedule": {"kind": "once", "run_at": run_at},
    +            "next_run_at": run_at, "enabled": True, "state": "scheduled",
    +        }])
    +        assert [j["id"] for j in get_due_jobs()] == ["wedged"]  # A claims, then dies
    +
    +        # Just inside the TTL: still claimed → skipped.
    +        monkeypatch.setattr("cron.jobs._hermes_now",
    +                            lambda: t0 + timedelta(seconds=ONESHOT_RUN_CLAIM_TTL_SECONDS - 10))
    +        assert get_due_jobs() == []
    +
    +        # Just past the TTL: stale claim → re-dispatched (recovered), re-claimed.
    +        monkeypatch.setattr("cron.jobs._hermes_now",
    +                            lambda: t0 + timedelta(seconds=ONESHOT_RUN_CLAIM_TTL_SECONDS + 10))
    +        recovered = get_due_jobs()
    +        assert [j["id"] for j in recovered] == ["wedged"]
    +
    +    def test_mark_job_run_clears_one_shot_run_claim(self, tmp_cron_dir, monkeypatch):
    +        """mark_job_run() clears the run_claim on completion so a re-dispatched
    +        one-shot (e.g. a stale-recovered retry) is claimable again."""
    +        from cron.jobs import _hermes_now
    +        t0 = _hermes_now()
    +        run_at = (t0 - timedelta(seconds=5)).isoformat()
    +        # Give it repeat headroom so mark_job_run keeps the job around.
    +        save_jobs([{
    +            "id": "claimclear", "name": "R", "prompt": "x",
    +            "schedule": {"kind": "once", "run_at": run_at},
    +            "next_run_at": run_at, "enabled": True, "state": "scheduled",
    +            "repeat": {"times": 2, "completed": 0},
    +        }])
    +        assert [j["id"] for j in get_due_jobs()] == ["claimclear"]
    +        assert get_job("claimclear").get("run_claim") is not None
    +        mark_job_run("claimclear", True)
    +        assert get_job("claimclear")["run_claim"] is None
    +
    +
         def test_broken_cron_without_next_run_is_recovered(self, tmp_cron_dir, monkeypatch):
             now = datetime(2026, 3, 18, 10, 0, 0, tzinfo=timezone.utc)
             monkeypatch.setattr("cron.jobs._hermes_now", lambda: now)

• #122943 fix(desktop): stand down cron ticker when a live gateway owns cron on same HERMES_HOME

  merged to main

  files: hermes_cli/web_server.py, tests/hermes_cli/test_desktop_cron_ticker_gateway_standdown.py