# NousResearch/hermes-agent · Cron stale-entry removal deleting live one-shot jobs ## Open now - Nothing open on GitHub. A bug in get_due_jobs causes stale-entry cleanup to incorrectly delete one-shot cron jobs whose runs are still active, losing run outcome and delivery records. The related items are fixes adding liveness checks before any stale-entry removal. ## TL;DR - 3 member(s): 0 open issue(s), 1 closed issue(s), 0 open PR(s), 1 merged PR(s), 1 PR(s) closed without merging. - A fix was merged to main in PR #62014. Verify it addresses your case. - **Recommended:** read PR #62014 (and any competing PRs below) before filing. Confirm the merged change covers your symptom. ## Maintainer resolution verdict - Internal-tracker verdict over this cluster's issue members: `resolved` · deciding PR(s): `pr-62014`. This is distinct from the fix status above. Weigh it yourself. ## How this cluster was grouped - Link evidence (`STRUCTURAL_SPANNING`): every member is connected by structural evidence: shared changed files, or a closing/superseded reference. - Pairwise link density `0.67`: the share of member pairs that carry a link, counting a structural link in full and a relevance-only link at 0.36. `1.00` means every pair is linked. ## Closed / duplicate issues - #62002: cron: get_due_jobs stale-entry recovery deletes a live one-shot mid-run when the run outlives its run_claim TTL (run outcome + delivery record lost) ## Prior & competing PRs - **#62014: fix(cron): never stale-remove a one-shot whose run is still alive (#62002)**. merged to main · fix tier `verified` - maintainer direction: category `merged` - files: `cron/jobs.py`, `cron/scheduler.py`, `tests/cron/test_jobs.py` - fix hunks: - `cron/jobs.py` ```diff @@ -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" ``` ```diff @@ -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. ``` ```diff @@ -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` ```diff @@ -20,6 +20,7 @@ import subprocess import sys import threading +import time # fcntl is Unix-only; on Windows use msvcrt for file locking try: ``` ```diff @@ -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 ``` ```diff @@ -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 ``` ```diff @@ -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: ``` ```diff @@ -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` ```diff @@ -20,6 +20,7 @@ mark_job_run, advance_next_run, claim_dispatch, + heartbeat_run_claim, get_due_jobs, save_job_output, ) ``` ```diff @@ -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) ``` - **#62013: fix(cron): guard stale-entry removal with liveness check**. closed without merge - maintainer direction: category `superseded_by_pr`; implication `superseded`; refs `#62002`, `#62014` ## Fix status - Merged to main in PR #62014 (cited above). No release data is tracked. Verify downstream. - Fix last verified (merged) `2026-07-10`.