fix(cron): never stale-remove a one-shot whose run is still alive (#62002)

cron/jobs.py

@@ -192,6 +192,26 @@ def _oneshot_run_claim_ttl_seconds() -> float:
     )
 
 
+def _job_running_in_this_process(job_id: str) -> bool:
+    """Return True when the scheduler in THIS process is still running ``job_id``.
+
+    Direct liveness signal for stale-entry recovery (#62002): the run_claim
+    TTL alone cannot distinguish "the claiming tick died" from "the run is
+    alive but slow" — a run stalled on network I/O (or a laptop that slept
+    mid-run) legitimately outlives the TTL. The in-process ticker and the run
+    share this process, so the scheduler's running set settles the common
+    single-gateway case without any claim-age guesswork.
+
+    Imported lazily: the scheduler imports this module at load, so a
+    module-level import here would be circular.
+    """
+    try:
+        from cron.scheduler import get_running_job_ids
+        return job_id in get_running_job_ids()
+    except Exception:
+        return False
+
+
 def _jobs_lock_file() -> Path:
     """Return the advisory lock path for the current cron directory."""
     return _current_cron_store().cron_dir / ".jobs.lock"
@@ -1603,6 +1623,32 @@ def claim_dispatch(job_id: str) -> bool:
         return True
 
 
+def heartbeat_run_claim(job_id: str) -> bool:
+    """Refresh a one-shot's ``run_claim`` timestamp while its run is alive.
+
+    Called periodically from the scheduler's run monitor (#62002) so a
+    legitimately long run keeps its claim fresh: an expired claim then really
+    does mean "the claiming process died", and neither another process's tick
+    nor this process's own next tick will re-dispatch or stale-remove the job
+    while the run is in flight. mark_job_run() clears the claim on completion.
+
+    Returns True if a claim was refreshed; False when the job or its claim is
+    gone (nothing to refresh — e.g. a manual run that never stamped one).
+    """
+    with _jobs_lock():
+        jobs = load_jobs()
+        for job in jobs:
+            if job.get("id") != job_id:
+                continue
+            claim = job.get("run_claim")
+            if not isinstance(claim, dict):
+                return False
+            claim["at"] = _hermes_now().isoformat()
+            save_jobs(jobs)
+            return True
+    return False
+
+
 def advance_next_run(job_id: str) -> bool:
     """Preemptively advance next_run_at for a recurring job before execution.
 
@@ -1986,6 +2032,25 @@ def _get_due_jobs_locked() -> List[Dict[str, Any]]:
                         times = repeat.get("times")
                         completed = repeat.get("completed", 0)
                         if times is not None and times > 0 and completed >= times:
+                            # A live run must never have its job record deleted
+                            # underneath it (#62002): a run that outlives the
+                            # run_claim TTL (stream stall, laptop asleep
+                            # mid-run) satisfies the same completed >= times +
+                            # expired-claim condition as a dead tick, but
+                            # mark_job_run() still needs the record to land
+                            # last_run_at / last_status / last_delivery_error.
+                            # If this process is still running the job, it is
+                            # slow, not stale — keep the entry and skip.
+                            if _job_running_in_this_process(job.get("id", "")):
+                                logger.info(
+                                    "Job '%s': dispatch limit reached (%d/%d) "
+                                    "but its run is still in flight in this "
+                                    "process — keeping entry",
+                                    job.get("name", job.get("id", "?")),
+                                    completed,
+                                    times,
+                                )
+                                continue
                             logger.info(
                                 "Job '%s': one-shot dispatch limit reached (%d/%d) "
                                 "— removing stale due entry",

cron/scheduler.py

@@ -20,6 +20,7 @@
 import subprocess
 import sys
 import threading
+import time
 
 # fcntl is Unix-only; on Windows use msvcrt for file locking
 try:
@@ -237,7 +238,7 @@ def _resolve_cron_enabled_toolsets(job: dict, cfg: dict) -> list[str] | None:
     "QQBOT_HOME_CHANNEL": "QQ_HOME_CHANNEL",
 }
 
-from cron.jobs import get_due_jobs, mark_job_run, save_job_output, advance_next_run, claim_dispatch
+from cron.jobs import get_due_jobs, mark_job_run, save_job_output, advance_next_run, claim_dispatch, heartbeat_run_claim
 
 # Sentinel: when a cron agent has nothing new to report, it can start its
 # response with this marker to suppress delivery.  Output is still saved
@@ -3097,6 +3098,35 @@ def run_job(
             _cron_timeout = 600.0
         _cron_inactivity_limit = _cron_timeout if _cron_timeout > 0 else None
         _POLL_INTERVAL = 5.0
+        # Keep the one-shot run_claim fresh while the run is alive (#62002):
+        # the claim TTL is a dead-owner detector, but without a heartbeat a
+        # run that legitimately outlives it (stream stall, laptop asleep
+        # mid-run) is indistinguishable from a dead tick — another process
+        # re-dispatches it and get_due_jobs stale-removes the job record out
+        # from under the live run. Refreshing the claim from this monitor
+        # keeps "expired claim" meaning "owner died".
+        _job_schedule = job.get("schedule")
+        _is_oneshot = (
+            isinstance(_job_schedule, dict) and _job_schedule.get("kind") == "once"
+        )
+        _CLAIM_HEARTBEAT_SECONDS = 60.0
+        _last_claim_heartbeat = time.monotonic()
+
+        def _heartbeat_run_claim_if_due():
+            nonlocal _last_claim_heartbeat
+            if not _is_oneshot:
+                return
+            _mono = time.monotonic()
+            if _mono - _last_claim_heartbeat < _CLAIM_HEARTBEAT_SECONDS:
+                return
+            _last_claim_heartbeat = _mono
+            try:
+                heartbeat_run_claim(job_id)
+            except Exception:
+                logger.debug(
+                    "Job '%s': run_claim heartbeat failed", job_name, exc_info=True
+                )
+
         _cron_pool = concurrent.futures.ThreadPoolExecutor(max_workers=1)
         # Preserve scheduler-scoped ContextVar state (for example skill-declared
         # env passthrough registrations) when the cron run hops into the worker
@@ -3106,8 +3136,20 @@ def run_job(
         _inactivity_timeout = False
         try:
             if _cron_inactivity_limit is None:
-                # Unlimited — just wait for the result.
-                result = _cron_future.result()
+                # Unlimited — no inactivity watchdog, but a one-shot still
+                # needs its run_claim heartbeat, so poll instead of blocking.
+                if _is_oneshot:
+                    result = None
+                    while True:
+                        done, _ = concurrent.futures.wait(
+                            {_cron_future}, timeout=_POLL_INTERVAL,
+                        )
+                        if done:
+                            result = _cron_future.result()
+                            break
+                        _heartbeat_run_claim_if_due()
+                else:
+                    result = _cron_future.result()
             else:
                 result = None
                 while True:
@@ -3117,6 +3159,7 @@ def run_job(
                     if done:
                         result = _cron_future.result()
                         break
+                    _heartbeat_run_claim_if_due()
                     # Agent still running — check inactivity.
                     _idle_secs = 0.0
                     if hasattr(agent, "get_activity_summary"):

tests/cron/test_jobs.py

@@ -20,6 +20,7 @@
     mark_job_run,
     advance_next_run,
     claim_dispatch,
+    heartbeat_run_claim,
     get_due_jobs,
     save_job_output,
 )
@@ -1103,6 +1104,102 @@ def test_mark_job_run_clears_one_shot_run_claim(self, tmp_cron_dir, monkeypatch)
         mark_job_run("claimclear", True)
         assert get_job("claimclear")["run_claim"] is None
 
+    def test_stale_maxed_oneshot_kept_while_running_in_this_process(
+        self, tmp_cron_dir, monkeypatch
+    ):
+        """#62002: a live run must never have its job record deleted underneath it.
+
+        A one-shot whose run outlives the run_claim TTL (stream stall, laptop
+        asleep mid-run) satisfies the same completed >= times + expired-claim
+        condition as a dead tick. When the scheduler in this process still has
+        the job in its running set, the stale-entry recovery must keep the
+        record so the in-flight run's mark_job_run() can land its outcome —
+        and remove it only once the run is actually gone.
+        """
+        import cron.scheduler as scheduler_mod
+        from cron.jobs import _hermes_now, _oneshot_run_claim_ttl_seconds
+        monkeypatch.delenv("HERMES_CRON_TIMEOUT", raising=False)
+        ttl = _oneshot_run_claim_ttl_seconds()
+        t0 = _hermes_now()
+        run_at = (t0 - timedelta(seconds=ttl + 300)).isoformat()
+        # Mid-run store shape: claim_dispatch committed completed=1 and the
+        # run_claim was stamped at fire time; next_run_at is only resolved by
+        # mark_job_run, so it still points at the (past) fire time.
+        save_jobs([{
+            "id": "inflight", "name": "flight check", "prompt": "x",
+            "schedule": {"kind": "once", "run_at": run_at},
+            "next_run_at": run_at, "enabled": True, "state": "scheduled",
+            "repeat": {"times": 1, "completed": 1},
+            "run_claim": {"at": run_at, "by": "this-machine"},
+        }])
+
+        # Run still alive in this process → keep the record, dispatch nothing.
+        monkeypatch.setattr(
+            scheduler_mod, "get_running_job_ids", lambda: frozenset({"inflight"})
+        )
+        assert get_due_jobs() == []
+        assert get_job("inflight") is not None  # still visible to list/run
+
+        # The claiming tick really died (running set empty) → recovered as before.
+        monkeypatch.setattr(
+            scheduler_mod, "get_running_job_ids", lambda: frozenset()
+        )
+        assert get_due_jobs() == []
+        assert get_job("inflight") is None  # stale entry cleaned up
+
+    def test_run_claim_heartbeat_keeps_long_run_claimed_past_ttl(
+        self, tmp_cron_dir, monkeypatch
+    ):
+        """#62002 cross-process leg: a heartbeat-refreshed claim never expires
+        while the run is alive, so no other tick re-dispatches or stale-removes
+        the job even when the run outlives the original TTL horizon."""
+        monkeypatch.delenv("HERMES_CRON_TIMEOUT", raising=False)
+        from cron.jobs import _hermes_now, _oneshot_run_claim_ttl_seconds
+        ttl = _oneshot_run_claim_ttl_seconds()
+        t0 = _hermes_now()
+        run_at = (t0 - timedelta(seconds=5)).isoformat()
+        save_jobs([{
+            "id": "slowrun", "name": "R", "prompt": "x",
+            "schedule": {"kind": "once", "run_at": run_at},
+            "next_run_at": run_at, "enabled": True, "state": "scheduled",
+            "repeat": {"times": 1, "completed": 0},
+        }])
+
+        # Tick claims + dispatches the job.
+        assert [j["id"] for j in get_due_jobs()] == ["slowrun"]
+        assert claim_dispatch("slowrun") is True
+
+        # Mid-run heartbeat before the TTL horizon refreshes the claim.
+        monkeypatch.setattr("cron.jobs._hermes_now",
+                            lambda: t0 + timedelta(seconds=ttl - 60))
+        assert heartbeat_run_claim("slowrun") is True
+
+        # Past the ORIGINAL claim's TTL horizon: without the heartbeat this
+        # tick would stale-remove the maxed one-shot; with it the claim is
+        # fresh, so the job is skipped and the record survives.
+        monkeypatch.setattr("cron.jobs._hermes_now",
+                            lambda: t0 + timedelta(seconds=ttl + 10))
+        assert get_due_jobs() == []
+        assert get_job("slowrun") is not None
+
+        # Run completes → outcome lands on a record that still exists
+        # (times=1 reached, so mark_job_run retires the job normally).
+        mark_job_run("slowrun", True)
+        assert get_job("slowrun") is None
+
+    def test_heartbeat_run_claim_noop_without_claim(self, tmp_cron_dir):
+        """heartbeat_run_claim is a safe no-op when there is nothing to refresh
+        (manual run that never stamped a claim, or the job is gone)."""
+        future = (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat()
+        save_jobs([{
+            "id": "noclaim", "name": "R", "prompt": "x",
+            "schedule": {"kind": "once", "run_at": future},
+            "next_run_at": future, "enabled": True, "state": "scheduled",
+        }])
+        assert heartbeat_run_claim("noclaim") is False
+        assert heartbeat_run_claim("missing-job") is False
+        assert get_job("noclaim").get("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)