fix(cron): reconcile stale in-flight claim against executions ledger (t_8b5480b3)

cron/scheduler.py

@@ -810,6 +810,23 @@ def sweep_stale_inflight(due_jobs: Optional[list] = None) -> list:
     now = time.time()
     stale: list = []
 
+    # Latest durable execution per in-flight job id, loaded in one indexed
+    # query.  Used for the persisted-state reconciliation below (t_8b5480b3):
+    # an in-memory claim whose job's MOST RECENT execution row is terminal
+    # (completed/failed/unknown) cannot represent a live run — the durable
+    # ledger proves that run already ended — so the claim is stale by
+    # construction, regardless of its in-memory age.  This closes the
+    # restart-survival gap that the age-only sweep left open: the ledger is
+    # written by the worker that ran the job and read by ANY ticker process
+    # (including one that started AFTER the leak), so a leaked claim is
+    # recoverable without force-run/resume and without depending on which
+    # process happens to still hold it in memory.
+    try:
+        from cron.executions import latest_executions as _latest_execs
+        _latest = _latest_execs(list(_running_job_ids))
+    except Exception:
+        _latest = {}
+
     # Precompute job intervals OUTSIDE _running_lock so croniter evaluation
     # does not block try_register/release_running_job for other jobs.
     _intervals = {jid: _job_interval_minutes(j) for jid, j in by_id.items()}
@@ -827,8 +844,6 @@ def sweep_stale_inflight(due_jobs: Optional[list] = None) -> list:
             allowance = floor_seconds
             if interval_minutes:
                 allowance = max(allowance, 2.0 * interval_minutes * 60.0)
-            if age < allowance:
-                continue
             fut = _running_futures.get(job_id)
             if fut is _FUTURE_PENDING:
                 # The claim is past its allowance and the owning future still
@@ -838,13 +853,34 @@ def sweep_stale_inflight(due_jobs: Optional[list] = None) -> list:
                 pass
             elif fut is not None and not fut.done():
                 continue  # genuinely still executing
+            # Persisted-state reconciliation: if the durable executions ledger
+            # shows this job's last run reached a terminal state, the claim is
+            # provably stale even if it is still inside its in-memory age
+            # allowance (or was adopted fresh this tick).  Release it now so
+            # the job re-dispatches on the next tick without force-run/resume
+            # (t_8b5480b3 — the 2026-08-14 recurring-router wedge that
+            # survived a gateway restart because the in-memory age bound alone
+            # could not see a run the ledger had already finished).
+            if fut is None or fut is _FUTURE_PENDING or fut.done():
+                latest = _latest.get(job_id)
+                if latest is not None and latest.get("status") in (
+                    "completed", "failed", "unknown",
+                ):
+                    _running_job_ids.discard(job_id)
+                    _running_since.pop(job_id, None)
+                    _running_futures.pop(job_id, None)
+                    _forced_release_count += 1
+                    stale.append((job_id, age, allowance, fut, "ledger-terminal"))
+                    continue
+            if age < allowance:
+                continue
             _running_job_ids.discard(job_id)
             _running_since.pop(job_id, None)
             _running_futures.pop(job_id, None)
             _forced_release_count += 1
-            stale.append((job_id, age, allowance, fut))
+            stale.append((job_id, age, allowance, fut, "age"))
 
-    for job_id, age, allowance, fut in stale:
+    for job_id, age, allowance, fut, _reason in stale:
         job = by_id.get(job_id) or {}
         name = job.get("name") or job_id
         if fut is _FUTURE_PENDING:
@@ -854,16 +890,29 @@ def sweep_stale_inflight(due_jobs: Optional[list] = None) -> list:
         else:
             future_state = "finished"
         logger.warning(
-            "cron.inflight.forced_release event=forced_release job='%s' id=%s "
-            "age=%.0fs allowance=%.0fs future=%s — stale in-flight claim "
+            "cron.inflight.forced_release event=forced_release reason=%s job='%s' "
+            "id=%s age=%.0fs allowance=%.0fs future=%s — stale in-flight claim "
             "released; the job was skipping every fire with 'already running'",
+            _reason,
             name,
             job_id,
             age,
             allowance,
             future_state,
         )
         _record_forced_release(job_id, name, age, allowance)
+        # A ledger-terminal release is authoritative: the durable executions
+        # ledger ALREADY records how the last run ended (completed/failed/
+        # unknown), so we must NOT call mark_job_run here — doing so would
+        # clobber an honest completed/ok status with a synthetic failure, or
+        # double-write an already-recorded failure.  We only release the claim
+        # so the job re-dispatches on its next due tick; the ledger is the
+        # record of record for the outcome.  The age-based release below keeps
+        # the original wedge-surfacing mark_job_run behaviour (an age-release
+        # may have no ledger row at all, so surfacing last_error is the only
+        # way the wedge becomes visible).
+        if _reason == "ledger-terminal":
+            continue
         # Finite-repeat guard: a forced release is NOT a real run, so it must
         # not consume a finite one-shot's repeat budget or let mark_job_run
         # auto-delete the row (completed >= times).  The claim is released and

tests/cron/test_inflight_stale_guard.py

@@ -348,3 +348,107 @@ def test_tick_sweeps_then_dispatches_the_previously_wedged_job(self, tmp_path):
         assert n == 1, "wedged job must fire again without a gateway restart"
         assert job["id"] not in sched.get_running_job_ids()
         assert sched.get_inflight_guard_stats()["forced_releases"] == 1
+
+
+class TestLedgerTerminalReconciliation:
+    """Persisted-state recovery path (t_8b5480b3).
+
+    The age-only sweep released claims older than ``max(2 * interval, floor)``,
+    but a leaked claim could be YOUNG (inside its allowance) while the durable
+    executions ledger already proved the last run ended — e.g. the 2026-08-14
+    recurring-router wedge, which survived a gateway restart because the age
+    bound alone could not see a run the ledger had already finished.  This
+    class tests the ledger reconciliation: an in-memory claim whose job's
+    MOST RECENT execution row is terminal (completed/failed/unknown) is stale
+    by construction and is force-released regardless of in-memory age, so the
+    recurring job re-dispatches on the next tick without force-run/resume.
+
+    RED first: on the age-only sweep (main before this change) a young leaked
+    claim with a terminal ledger row is NOT released — it stays wedged.  With
+    the ledger reconciliation it IS released (and, because a terminal ledger
+    row is authoritative, WITHOUT a synthetic mark_job_run failure).
+    """
+
+    def _inject_young_claim(self, job_id: str) -> None:
+        """Claim is YOUNG (inside the 30m floor) so only the ledger-terminal
+        path, never the age path, can release it."""
+        sched._running_job_ids.add(job_id)
+        if hasattr(sched, "_running_since"):
+            sched._running_since[job_id] = time.time() - 60  # 1 minute old
+
+    def test_young_claim_with_terminal_ledger_row_is_released(self, tmp_path):
+        """RED/GREEN: a terminal ledger row proves the run ended, so a young
+        leaked claim must be force-released even though its in-memory age is
+        inside the allowance (the age-only sweep alone would leave it)."""
+        job = _job(job_id="ledger-terminal", minutes=10)
+        job_id = job["id"]
+        self._inject_young_claim(job_id)
+
+        with patch.object(sched, "_get_hermes_home", return_value=tmp_path), \
+             patch("cron.executions.latest_executions", return_value={
+                 job_id: {"status": "failed", "id": "exec-x"},
+             }), \
+             patch.object(sched, "mark_job_run") as mark:
+            released = sched.sweep_stale_inflight([job])
+
+        # RED on the age-only sweep: the young claim is NOT released.  GREEN
+        # with the ledger reconciliation: it IS released.
+        assert job_id in released, "terminal ledger row must force-release the claim"
+        assert job_id not in sched.get_running_job_ids()
+        # Ledger-terminal is authoritative: no synthetic failure written.
+        mark.assert_not_called()
+
+    def test_young_claim_without_ledger_row_is_not_released(self, tmp_path):
+        """A young claim whose job has NO execution row at all (the claim was
+        taken but create_execution never ran) is left to the age bound — the
+        ledger reconciliation must not release claims it cannot prove ended."""
+        job = _job(job_id="no-ledger-row", minutes=10)
+        job_id = job["id"]
+        self._inject_young_claim(job_id)
+
+        with patch.object(sched, "_get_hermes_home", return_value=tmp_path), \
+             patch("cron.executions.latest_executions", return_value={}), \
+             patch.object(sched, "mark_job_run"):
+            released = sched.sweep_stale_inflight([job])
+
+        assert job_id not in released
+        assert job_id in sched.get_running_job_ids()
+
+    def test_young_claim_with_running_ledger_row_is_not_released(self, tmp_path):
+        """A young claim whose job is genuinely still running per the ledger
+        ('claimed'/'running' row) is never released — reconciliation must not
+        double-dispatch a healthy long-running job."""
+        job = _job(job_id="still-running", minutes=10)
+        job_id = job["id"]
+        self._inject_young_claim(job_id)
+
+        with patch.object(sched, "_get_hermes_home", return_value=tmp_path), \
+             patch("cron.executions.latest_executions", return_value={
+                 job_id: {"status": "running", "id": "exec-y"},
+             }), \
+             patch.object(sched, "mark_job_run"):
+            released = sched.sweep_stale_inflight([job])
+
+        assert job_id not in released
+        assert job_id in sched.get_running_job_ids()
+
+    def test_old_claim_with_terminal_ledger_row_still_released_once(self, tmp_path):
+        """An old claim with a terminal ledger row is released by the ledger
+        path (one release) — it must not double-release or double-count."""
+        job = _job(job_id="old-terminal", minutes=10)
+        job_id = job["id"]
+        sched._running_job_ids.add(job_id)
+        if hasattr(sched, "_running_since"):
+            sched._running_since[job_id] = time.time() - 6 * 60 * 60  # 6h old
+
+        with patch.object(sched, "_get_hermes_home", return_value=tmp_path), \
+             patch("cron.executions.latest_executions", return_value={
+                 job_id: {"status": "completed", "id": "exec-z"},
+             }), \
+             patch.object(sched, "mark_job_run") as mark:
+            released = sched.sweep_stale_inflight([job])
+
+        assert job_id in released
+        assert job_id not in sched.get_running_job_ids()
+        assert sched.get_inflight_guard_stats()["forced_releases"] == 1
+        mark.assert_not_called()  # authoritative ledger row, no synthetic failure