From dabae386ef19010d2b0c9e4c4d7ffb984e9115db Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Fri, 10 Jul 2026 19:12:35 -0700 Subject: [PATCH] fix(cron): bind claim heartbeats to dispatch owner (#62155) --- cron/jobs.py | 14 ++++++--- cron/scheduler.py | 8 +++-- tests/cron/test_jobs.py | 37 +++++++++++++++++++++-- tests/cron/test_scheduler.py | 58 ++++++++++++++++++++++++++++++++++++ 4 files changed, 108 insertions(+), 9 deletions(-) diff --git a/cron/jobs.py b/cron/jobs.py index 1ea3361fdef..90c318742e6 100644 --- a/cron/jobs.py +++ b/cron/jobs.py @@ -1623,7 +1623,7 @@ def claim_dispatch(job_id: str) -> bool: return True -def heartbeat_run_claim(job_id: str) -> bool: +def heartbeat_run_claim(job_id: str, *, expected_owner: 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 @@ -1632,16 +1632,22 @@ def heartbeat_run_claim(job_id: str) -> bool: 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). + ``expected_owner`` is the stable owner copied from the dispatched job. The + compare-and-refresh prevents a stale runner that resumes after a long sleep + from extending a claim another scheduler process has since taken over. + + Returns True if this owner's one-shot claim was refreshed; False when the + job, claim, or ownership no longer matches. """ with _jobs_lock(): jobs = load_jobs() for job in jobs: if job.get("id") != job_id: continue + if job.get("schedule", {}).get("kind") != "once": + return False claim = job.get("run_claim") - if not isinstance(claim, dict): + if not isinstance(claim, dict) or claim.get("by") != expected_owner: return False claim["at"] = _hermes_now().isoformat() save_jobs(jobs) diff --git a/cron/scheduler.py b/cron/scheduler.py index d75398d1945..7cf688f3734 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -3109,19 +3109,23 @@ def run_job( _is_oneshot = ( isinstance(_job_schedule, dict) and _job_schedule.get("kind") == "once" ) + _run_claim = job.get("run_claim") + _run_claim_owner = ( + str(_run_claim.get("by") or "") if isinstance(_run_claim, dict) else "" + ) _CLAIM_HEARTBEAT_SECONDS = 60.0 _last_claim_heartbeat = time.monotonic() def _heartbeat_run_claim_if_due(): nonlocal _last_claim_heartbeat - if not _is_oneshot: + if not _is_oneshot or not _run_claim_owner: return _mono = time.monotonic() if _mono - _last_claim_heartbeat < _CLAIM_HEARTBEAT_SECONDS: return _last_claim_heartbeat = _mono try: - heartbeat_run_claim(job_id) + heartbeat_run_claim(job_id, expected_owner=_run_claim_owner) except Exception: logger.debug( "Job '%s': run_claim heartbeat failed", job_name, exc_info=True diff --git a/tests/cron/test_jobs.py b/tests/cron/test_jobs.py index 18a59125706..d7d16e55134 100644 --- a/tests/cron/test_jobs.py +++ b/tests/cron/test_jobs.py @@ -1172,7 +1172,8 @@ class TestGetDueJobs: # 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 + owner = get_job("slowrun")["run_claim"]["by"] + assert heartbeat_run_claim("slowrun", expected_owner=owner) 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 @@ -1196,10 +1197,40 @@ class TestGetDueJobs: "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 heartbeat_run_claim("noclaim", expected_owner="owner") is False + assert heartbeat_run_claim("missing-job", expected_owner="owner") is False assert get_job("noclaim").get("run_claim") is None + def test_heartbeat_run_claim_rejects_replaced_owner(self, tmp_cron_dir): + """A resumed stale runner must not keep a newer owner's claim alive.""" + future = (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat() + original_at = datetime.now(timezone.utc).isoformat() + save_jobs([{ + "id": "reclaimed", "name": "R", "prompt": "x", + "schedule": {"kind": "once", "run_at": future}, + "next_run_at": future, "enabled": True, "state": "scheduled", + "run_claim": {"at": original_at, "by": "new-owner"}, + }]) + + assert heartbeat_run_claim("reclaimed", expected_owner="old-owner") is False + assert get_job("reclaimed")["run_claim"] == { + "at": original_at, + "by": "new-owner", + } + + def test_heartbeat_run_claim_rejects_non_oneshot(self, tmp_cron_dir): + """Heartbeat ownership applies only to one-shot dispatch claims.""" + original_at = datetime.now(timezone.utc).isoformat() + save_jobs([{ + "id": "recurring", "name": "R", "prompt": "x", + "schedule": {"kind": "interval", "seconds": 60}, + "enabled": True, + "run_claim": {"at": original_at, "by": "owner"}, + }]) + + assert heartbeat_run_claim("recurring", expected_owner="owner") is False + assert get_job("recurring")["run_claim"]["at"] == original_at + 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) diff --git a/tests/cron/test_scheduler.py b/tests/cron/test_scheduler.py index 8cb81cb6177..3d262155ba3 100644 --- a/tests/cron/test_scheduler.py +++ b/tests/cron/test_scheduler.py @@ -1,6 +1,7 @@ """Tests for cron/scheduler.py — origin resolution, delivery routing, and error logging.""" import contextlib +import itertools import json import logging import os @@ -1623,6 +1624,63 @@ class TestRunJobSessionPersistence: assert os.getenv("HERMES_CRON_AUTO_DELIVER_THREAD_ID") is None fake_db.close.assert_called_once() + @pytest.mark.parametrize("timeout_value", ["600", "0"]) + def test_run_job_heartbeats_oneshot_claim_in_both_wait_modes( + self, tmp_path, monkeypatch, timeout_value + ): + """Timed and unlimited one-shot monitors both refresh their owned claim.""" + job = { + "id": "heartbeat-job", + "name": "heartbeat", + "prompt": "hello", + "schedule": {"kind": "once", "run_at": "2026-07-10T12:00:00Z"}, + "run_claim": {"at": "2026-07-10T12:00:00Z", "by": "owner-token"}, + } + fake_db = MagicMock() + + class FakeAgent: + def __init__(self, *args, **kwargs): + pass + + def run_conversation(self, *args, **kwargs): + return {"final_response": "ok"} + + class FakeFuture: + def result(self): + return {"final_response": "ok"} + + fake_future = FakeFuture() + fake_pool = MagicMock() + fake_pool.submit.return_value = fake_future + wait_results = [(set(), set()), ({fake_future}, set())] + monotonic_ticks = itertools.count(step=61.0) + monkeypatch.setenv("HERMES_CRON_TIMEOUT", timeout_value) + + with patch("cron.scheduler._hermes_home", tmp_path), \ + patch("hermes_state.SessionDB", return_value=fake_db), \ + patch( + "hermes_cli.runtime_provider.resolve_runtime_provider", + return_value={ + "api_key": "***", + "base_url": "https://example.invalid/v1", + "provider": "openrouter", + "api_mode": "chat_completions", + }, + ), \ + patch("run_agent.AIAgent", FakeAgent), \ + patch("cron.scheduler.concurrent.futures.ThreadPoolExecutor", return_value=fake_pool), \ + patch("cron.scheduler.concurrent.futures.wait", side_effect=wait_results), \ + patch("cron.scheduler.time.monotonic", side_effect=monotonic_ticks.__next__), \ + patch("cron.scheduler.heartbeat_run_claim", return_value=True) as heartbeat: + success, _output, final_response, error = run_job(job) + + assert success is True + assert error is None + assert final_response == "ok" + heartbeat.assert_called_once_with( + "heartbeat-job", expected_owner="owner-token" + ) + def test_run_job_resets_secret_source_cache_before_reload(self, tmp_path, monkeypatch): """Each run must clear the secret-source cache before re-reading the env, so a long-running gateway re-resolves Bitwarden/BSM-backed secrets