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

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 = {
-    "ishengeqi@163.com": "isheng-eqi",  # PR #59428 salvage (cron: reject past one-shot timestamps in update_job fallback + resume_job; #59395)
+    "ishengeqi@163.com": "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).
     "derek2000139@qq.com": "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)
     "AndreasHiltner@users.noreply.github.com": "AndreasHiltner",  # PR #56854 salvage (gateway: route multiplex profile responses through the profile's own adapter — 53-site _adapter_for_source sweep)
     "allenliang2022@users.noreply.github.com": "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)