fix(cron): durable run-claim for one-shots instead of a fixed +60s advance

The +60s next_run_at advance only delayed a duplicate one-shot dispatch by
one tick — a job that outlives the 60s tick interval (the reported 2.5-min
research prompt) still re-fired on the next tick after the window expired,
so the concurrent gateway+desktop double-delivery persisted.

Replace it with a durable run_claim (at+by, mirroring fire_claim) stamped
on the one-shot under the same jobs lock get_due_jobs holds, and checked at
the top of the due-scan: a fresh claim held by an in-flight run makes every
other scheduler process skip the job for its ENTIRE run, not one tick.
mark_job_run() clears the claim on completion; a ONESHOT_RUN_CLAIM_TTL
(30 min) safety valve re-dispatches a claim left by a tick that died mid-run
so a one-shot is never wedged.

E2E: long-running one-shot no longer double-fires at +28/+61/+120/+179s;
completion clears the claim + disables the job; crash recovery re-arms past
the TTL. +3 regression tests.

cron/jobs.py

@@ -87,6 +87,16 @@ _jobs_lock_state = threading.local()
 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,20 +1736,27 @@ def _get_due_jobs_locked() -> List[Dict[str, Any]]:
                                 break
                         continue
 
-            # Before returning a one-shot job as due, advance its
-            # next_run_at past the next scheduler tick so other processes
-            # (gateway + desktop running concurrent schedulers) don't
-            # double-execute it (#59229).  The advance is persisted
-            # immediately under the same lock that get_due_jobs holds.
-            # mark_job_run re-anchors next_run_at on completion (success
-            # or failure), so a tick death between here and execution
-            # only delays the job by a single tick window — not lost.
+            # 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":
-                claimed_next = (now + timedelta(seconds=60)).isoformat()
-                job["next_run_at"] = claimed_next
+                claim = {"at": now.isoformat(), "by": _machine_id()}
+                job["run_claim"] = claim
                 for rj in raw_jobs:
                     if rj["id"] == job["id"]:
-                        rj["next_run_at"] = claimed_next
+                        rj["run_claim"] = claim
                         needs_save = True
                         break
 

tests/cron/test_jobs.py

@@ -927,9 +927,13 @@ class TestGetDueJobs:
         due = get_due_jobs()
 
         assert [job["id"] for job in due] == ["oneshot-recover"]
-        # next_run_at is now advanced past the tick window to prevent
-        # cross-process double-execution (#59229).
-        assert get_job("oneshot-recover")["next_run_at"] == (now + timedelta(seconds=60)).isoformat()
+        # 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)
@@ -960,6 +964,80 @@ class TestGetDueJobs:
         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)