fix(cron): renew fire claims without contending with delivery fence

cron/jobs.py

@@ -235,15 +235,18 @@ def _release_flock(lock_fd) -> None:
 
 
 @contextlib.contextmanager
-def _jobs_lock():
+def _jobs_lock(*, strict: bool = False):
     """Serialize a load_jobs→modify→save_jobs critical section: in-process RLock (parallel tick
     threads) plus a cross-process flock on ``<cron dir>/.jobs.lock`` (gateway vs. CLI writes —
     otherwise a `cron pause` could be clobbered and keep firing). Nested calls in one thread
     reuse the held lock. Without a flock backend, or on flock timeout (logged loudly), it
     degrades to in-process-only locking: a briefly torn cross-process write beats a dead
-    scheduler."""
+    scheduler. ``strict=True`` instead fails closed; nested strict calls require an actually
+    acquired outer flock, not merely an in-process lock."""
     depth = getattr(_jobs_lock_state, "depth", 0)
     if depth:
+        if strict and not getattr(_jobs_lock_state, "flock_acquired", False):
+            raise RuntimeError("Cron jobs lock is degraded; strict locking required")
         _jobs_lock_state.depth = depth + 1
         try:
             yield
@@ -258,13 +261,20 @@ def _jobs_lock():
         # stamps from unlocked loads or prior sections can never suppress a needed merge.
         # See #80703.
         _jobs_lock_state.load_stamp = None
+        _jobs_lock_state.flock_acquired = False
         lock_fd = None
         try:
             try:
                 ensure_dirs()
                 lock_fd = open(_jobs_lock_file(), "a+", encoding="utf-8")
                 lock_fd.seek(0)
-                if _acquire_flock(lock_fd, _JOBS_LOCK_TIMEOUT_SECONDS) is False:
+                acquired = _acquire_flock(lock_fd, _JOBS_LOCK_TIMEOUT_SECONDS)
+                _jobs_lock_state.flock_acquired = acquired is True
+                if strict and not _jobs_lock_state.flock_acquired:
+                    if acquired is False:
+                        raise TimeoutError("Timed out waiting for the cron jobs lock")
+                    raise RuntimeError("No cross-process lock backend for cron jobs lock")
+                if acquired is False:
                     logger.error(
                         "Timed out after %.0fs waiting for the cron "
                         "jobs lock (%s) — another process is holding "
@@ -275,17 +285,22 @@ def _jobs_lock():
                         lock_fd.close()
                     lock_fd = None
             except (OSError, IOError) as e:
+                if strict:
+                    raise
                 # A locking failure must never take down cron writes — in-process lock still held.
                 logger.warning("jobs.json cross-process lock unavailable (%s); "
                                "proceeding with in-process lock only", e)
+            yield
+        finally:
             try:
-                yield
-            finally:
-                if lock_fd is not None:
+                if lock_fd is not None and _jobs_lock_state.flock_acquired:
                     _release_flock(lock_fd)
-        finally:
-            _jobs_lock_state.depth = 0
-            _jobs_lock_state.load_stamp = None
+                elif lock_fd is not None:
+                    lock_fd.close()
+            finally:
+                _jobs_lock_state.depth = 0
+                _jobs_lock_state.load_stamp = None
+                _jobs_lock_state.flock_acquired = False
 
 
 @contextlib.contextmanager
@@ -2517,7 +2532,14 @@ def heartbeat_fire_claim(job_id: str, *, expected_owner: str) -> bool:
     def apply(jobs, _i, job):
         return _refresh_claim(jobs, job.get("fire_claim"), expected_owner)
 
-    return _under_fire_fence(job_id, lambda: _with_job(job_id, apply, False))
+    # Renewal only changes the timestamp: the jobs lock makes the owner check + write atomic
+    # against takeover/completion, just like heartbeat_run_claim. Taking the fire fence here
+    # would block behind our own delivery and mistake its lock timeout for ownership loss.
+    # Keep that fence on owner mutations and external side effects, not lease renewal.
+    # Do not renew from an unsynchronized snapshot if the jobs lock degrades. Raising lets
+    # the heartbeat loop apply its existing transient-error grace instead of reporting loss.
+    with _jobs_lock(strict=True):
+        return _with_job(job_id, apply, False)
 
 
 # Completed one-shots are retained in jobs.json (final status stays inspectable) and pruned by

tests/cron/test_heartbeat_delivery_fence.py

@@ -0,0 +1,122 @@
+"""Renewal must not contend with delivery, without weakening owner fencing."""
+
+import threading
+
+import pytest
+
+
+@pytest.mark.parametrize("claim_kind", ["fire", "run"])
+def test_claim_renewal_during_side_effect_fence(tmp_path, monkeypatch, claim_kind):
+    from cron import jobs
+
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    monkeypatch.setattr(jobs, "_JOBS_LOCK_TIMEOUT_SECONDS", 0.05)
+    job = jobs.create_job(prompt="x", schedule="in 30m")
+    claimed = jobs.claim_job_for_fire(job["id"], return_job=True)
+    owner = claimed["fire_claim"]["by"]
+    field = f"{claim_kind}_claim"
+    if claim_kind == "run":
+        jobs.update_job(job["id"], {field: dict(claimed["fire_claim"])})
+    before = jobs.get_job(job["id"])[field]
+    heartbeat = getattr(jobs, f"heartbeat_{claim_kind}_claim")
+    results = []
+    errors = []
+    done = threading.Event()
+
+    def renew():
+        try:
+            results.append(heartbeat(job["id"], expected_owner=owner))
+            # A competing dispatch and terminal write must still fail closed.
+            results.append(jobs.claim_job_for_fire(job["id"], claim_ttl_seconds=0))
+            results.append(jobs.mark_job_run(job["id"], True, expected_fire_owner=owner))
+        except BaseException as exc:
+            errors.append(exc)
+        finally:
+            done.set()
+
+    with jobs.fire_claim_fence(job["id"], expected_owner=owner) as owned:
+        assert owned
+        thread = threading.Thread(target=renew)
+        thread.start()
+        try:
+            assert done.wait(5), "renewal blocked behind delivery's fire fence"
+        finally:
+            thread.join(timeout=5)
+    assert not thread.is_alive()
+    assert not errors
+    assert results == [True, False, False]
+    refreshed = jobs.get_job(job["id"])[field]
+    assert refreshed["by"] == before["by"]
+    assert refreshed["at"] != before["at"]
+
+    # Replaced or cleared claims must never be renewed by the stale owner.
+    replacement = {"by": "replacement", "at": refreshed["at"]}
+    jobs.update_job(job["id"], {field: replacement})
+    assert heartbeat(job["id"], expected_owner=owner) is False
+    assert jobs.get_job(job["id"])[field] == replacement
+    jobs.update_job(job["id"], {field: None})
+    assert heartbeat(job["id"], expected_owner=owner) is False
+    assert jobs.get_job(job["id"])[field] is None
+
+
+@pytest.mark.parametrize("replace_owner", [False, True])
+def test_run_one_job_delivery_and_real_ownership_loss(tmp_path, monkeypatch, replace_owner):
+    from cron import executions, jobs, scheduler
+
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    monkeypatch.setattr(jobs, "_JOBS_LOCK_TIMEOUT_SECONDS", 0.05)
+    monkeypatch.setattr(scheduler, "_RUN_CLAIM_HEARTBEAT_SECONDS", 0.01)
+    job = jobs.create_job(prompt="x", schedule="every 5m", deliver="discord:123")
+    claimed = jobs.claim_job_for_fire(job["id"], return_job=True)
+    execution = executions.create_execution(job["id"], source="direct")
+    claimed["execution_id"] = execution["id"]
+    delivering = threading.Event()
+    renewed = threading.Event()
+    renewals = []
+    delivered = []
+    real_heartbeat = jobs.heartbeat_fire_claim
+
+    def observe_heartbeat(job_id, *, expected_owner):
+        result = real_heartbeat(job_id, expected_owner=expected_owner)
+        if delivering.is_set():
+            renewals.append(result)
+            renewed.set()
+        return result
+
+    def run_job(_job, *, cancel_event, **kwargs):
+        if replace_owner:
+            assert jobs.claim_job_for_fire(job["id"], claim_ttl_seconds=0)
+            assert cancel_event.wait(5), "real replacement did not cancel the stale runner"
+        return True, "saved output", "final response", None
+
+    def deliver(_job, content, **kwargs):
+        delivered.append(content)
+        delivering.set()
+        try:
+            # Delivery holds the real fire fence until the monitor has attempted renewal.
+            assert renewed.wait(5), "heartbeat never completed during delivery"
+        finally:
+            delivering.clear()
+        return None
+
+    monkeypatch.setattr(scheduler, "heartbeat_fire_claim", observe_heartbeat)
+    monkeypatch.setattr(scheduler, "run_job", run_job)
+    monkeypatch.setattr(scheduler, "_deliver_result", deliver)
+    assert scheduler.run_one_job(claimed) is True
+
+    persisted = jobs.get_job(job["id"])
+    ledger = executions.get_execution(execution["id"])
+    if replace_owner:
+        assert delivered == []
+        assert ledger["status"] == "failed"
+        assert "ownership lost" in ledger["error"].lower()
+        assert persisted["fire_claim"]["by"] != claimed["fire_claim"]["by"]
+        assert persisted.get("last_run_at") is None
+    else:
+        assert delivered == ["final response"]
+        assert renewals and all(renewals)
+        assert persisted["last_status"] == "ok"
+        assert persisted["fire_claim"] is None
+        assert ledger["status"] == "completed"
+        assert ledger["error"] is None
+        assert list((tmp_path / "cron" / "output" / job["id"]).glob("*.md"))

tests/cron/test_heartbeat_strict_lock.py

@@ -0,0 +1,68 @@
+"""Lease renewal must never use the jobs lock's degraded write mode."""
+import contextlib
+from datetime import datetime, timedelta
+from unittest.mock import Mock
+
+import pytest
+
+
+@pytest.mark.parametrize("nested", [False, True])
+@pytest.mark.parametrize("failure", ["timeout", "missing", "oserror"])
+def test_heartbeat_fails_closed_without_cross_process_lock(tmp_path, monkeypatch, failure, nested):
+    import cron.jobs as jobs
+
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    job = jobs.create_job(prompt="x", schedule="every 5m")
+    assert jobs.claim_job_for_fire(job["id"])
+    claim = jobs.get_job(job["id"])["fire_claim"]
+    now = datetime.fromisoformat(claim["at"]) + timedelta(seconds=30)
+    monkeypatch.setattr(jobs, "_hermes_now", lambda: now)
+    before = (tmp_path / "cron" / "jobs.json").read_bytes()
+
+    if failure == "missing":
+        monkeypatch.setattr(jobs, "fcntl", None)
+        monkeypatch.setattr(jobs, "msvcrt", None)
+    else:
+        acquire = (
+            Mock(return_value=False) if failure == "timeout"
+            else Mock(side_effect=OSError("lock failed"))
+        )
+        monkeypatch.setattr(jobs, "_acquire_flock", acquire)
+    load = Mock(wraps=jobs.load_jobs)
+    monkeypatch.setattr(jobs, "load_jobs", load)
+
+    error = RuntimeError if nested or failure == "missing" else (
+        TimeoutError if failure == "timeout" else OSError
+    )
+    with jobs._jobs_lock() if nested else contextlib.nullcontext():
+        with pytest.raises(error):
+            jobs.heartbeat_fire_claim(job["id"], expected_owner=claim["by"])
+    load.assert_not_called()
+    assert (tmp_path / "cron" / "jobs.json").read_bytes() == before
+    # An unsuccessful acquisition must close its descriptor, including an OSError.
+    if failure != "missing":
+        assert all(call.args[0].closed for call in acquire.call_args_list)
+    assert jobs._jobs_lock_state.depth == 0
+
+
+@pytest.mark.parametrize("strict_outer", [False, True])
+def test_heartbeat_reuses_real_outer_lock_and_preserves_owner(tmp_path, monkeypatch, strict_outer):
+    import cron.jobs as jobs
+
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    job = jobs.create_job(prompt="x", schedule="every 5m")
+    assert jobs.claim_job_for_fire(job["id"])
+    claim = jobs.get_job(job["id"])["fire_claim"]
+    now = datetime.fromisoformat(claim["at"]) + timedelta(seconds=30)
+    monkeypatch.setattr(jobs, "_hermes_now", lambda: now)
+    with jobs._jobs_lock(strict=strict_outer):
+        acquire = Mock(side_effect=AssertionError("nested lock must reuse actual flock"))
+        monkeypatch.setattr(jobs, "_acquire_flock", acquire)
+        assert jobs.heartbeat_fire_claim(job["id"], expected_owner=claim["by"])
+        refreshed = jobs.get_job(job["id"])["fire_claim"]
+        assert refreshed["at"] != claim["at"]
+        assert refreshed["by"] == claim["by"]
+        assert not jobs.heartbeat_fire_claim(job["id"], expected_owner="stale-owner")
+        assert jobs.get_job(job["id"])["fire_claim"] == refreshed
+        acquire.assert_not_called()
+    assert jobs._jobs_lock_state.depth == 0