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

get_due_jobs()'s one-shot stale-entry recovery (#38758) treated an
expired run_claim (#59229) as proof the claiming tick died, but a run
stalled on network I/O — or a laptop asleep mid-run — legitimately
outlives the TTL while very much alive. The recovery then deleted the
job record mid-flight: list showed the job gone, and when the run
finished mark_job_run() found nothing to update, so last_run_at /
last_status / last_delivery_error were never recorded.

Two guards, per the liveness signals available:

- Same process (the common single-gateway case): before removing a
  dispatch-limit-reached one-shot, consult the scheduler's running set
  via a lazy import; if the job is still running here it is slow, not
  stale — keep the entry.
- Cross process: run_job's monitor loop now refreshes run_claim.at
  every 60s while the run is alive (including under
  HERMES_CRON_TIMEOUT=0, which previously blocked without polling), so
  an expired claim really does mean the owner died and the TTL stays a
  dead-owner detector.

Fixes #62002
This commit is contained in:
PRATHAMESH75 2026-07-10 15:54:38 +05:30 committed by Teknium
parent 90bd5b0f9b
commit 9b72995a1d
3 changed files with 208 additions and 3 deletions

View file

@ -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",

View file

@ -20,6 +20,7 @@ import shutil
import subprocess
import sys
import threading
import time
# fcntl is Unix-only; on Windows use msvcrt for file locking
try:
@ -237,7 +238,7 @@ _LEGACY_HOME_TARGET_ENV_VARS = {
"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"):

View file

@ -20,6 +20,7 @@ from cron.jobs import (
mark_job_run,
advance_next_run,
claim_dispatch,
heartbeat_run_claim,
get_due_jobs,
save_job_output,
)
@ -1103,6 +1104,102 @@ class TestGetDueJobs:
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)