Cron stale-entry removal deleting live one-shot jobs

Fixes on record

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

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

  merged to main

  files: cron/jobs.py, cron/scheduler.py, tests/cron/test_jobs.py

  Show the diff

  • 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)