fix(cron): mirror plan_retry's yield branch in will_retry so fast jobs stop holding failure notices

will_retry gated notice suppression on recurring/paused/attempt/config only.
plan_retry has a yield branch: when the schedule's own next occurrence is at or
before the pending ladder rung it schedules nothing and clears state without
consuming an attempt. For a job on a cadence at or under a rung (<=5m, and the
15m/30m rungs for faster cadences) every failure hit that branch, the attempt
counter never advanced, and will_retry kept answering True — so during a
sustained outage every failure notice was held forever. The documented escape
('once the ladder is exhausted, the next failure alerts normally') was
unreachable: the ladder could never exhaust.

will_retry now recomputes the natural next occurrence exactly as
_advance_after_run will and answers True only when the rung precedes it — i.e.
exactly when plan_retry will actually park a re-run. Notices now go out on the
first failure for cadences the ladder cannot help, and slow jobs keep their
silent bounded re-runs.

(cherry picked from commit 9d3d6006204269103188a5ee366d8eb7c9f482a2)

cron/unreachable_retry.py

@@ -69,13 +69,43 @@ def _is_recurring(job: Dict[str, Any]) -> bool:
 
 def will_retry(job: Dict[str, Any]) -> bool:
     """Predict whether ``plan_retry`` will schedule a re-run for this flagged failure —
-    used by the scheduler to suppress the interim failure notice."""
+    used by the scheduler to suppress the interim failure notice.
+
+    Must mirror ``plan_retry`` (and the ``mark_job_run`` guards around it) decision-for-
+    decision, so a notice is held exactly when a re-run really is coming:
+
+    - exhausted ladder / disabled / paused / non-recurring → no re-run;
+    - the yield branch: when the schedule's own next occurrence lands at or before the
+      pending rung, ``plan_retry`` schedules nothing (the natural fire IS the retry).
+      Without this mirror, a job on a cadence at or under a rung held EVERY failure
+      notice for as long as the outage lasted: each failure yielded without consuming an
+      attempt, so the "once the ladder is exhausted, the next failure alerts normally"
+      escape in the suppression contract was unreachable and the operator saw silence.
+      (The final run of a finite repeat is a separate terminal-state edge handled by
+      #109991.)
+
+    Called before ``mark_job_run`` records the run; the occurrence is recomputed exactly
+    as ``_advance_after_run`` will (delivery-time skew between here and the store write
+    can only err toward delivering the notice, never toward suppressing it unheard).
+    """
     if not _is_recurring(job) or job.get("state") == "paused":
         return False
     state = job.get(STATE_KEY) or {}
-    if int(state.get("attempt") or 0) >= len(RETRY_DELAYS_SECONDS):
+    attempt = int(state.get("attempt") or 0)
+    if attempt >= len(RETRY_DELAYS_SECONDS):
         return False
-    return retry_enabled()
+    if not retry_enabled():
+        return False
+    from cron.jobs import _instant_after, _parse_aware, _seconds_after, compute_next_run
+
+    natural_next = _parse_aware(
+        compute_next_run(job.get("schedule") or {}, _hermes_now().isoformat()))
+    if natural_next is None:
+        # The natural occurrence is uncomputable (e.g. croniter missing): _advance_after_run
+        # leaves the record state=error — terminal — so plan_retry never runs either.
+        return False
+    retry_dt = _seconds_after(_hermes_now(), RETRY_DELAYS_SECONDS[attempt])
+    return _instant_after(natural_next, retry_dt)
 
 
 def clear_state(job: Dict[str, Any]) -> None:

tests/cron/test_unreachable_retry.py

@@ -75,6 +75,41 @@ def test_unreachable_failure_pulls_next_run_earlier_then_ladder_exhausts(
     assert weekly["id"] not in {due["id"] for due in get_due_jobs()}
 
 
+def test_will_retry_mirrors_plan_retry_yield_and_terminal_paths(tmp_cron_home):
+    """The notice-suppression predictor must mirror what ``plan_retry`` will actually do
+    for THIS failure:
+
+    - a cadence at or under the next ladder rung makes the schedule's own fire the retry
+      (``plan_retry`` yields without consuming an attempt), so no re-run is scheduled and
+      the failure notice must go out — otherwise a fast job holds every failure notice
+      for as long as the outage lasts (the "ladder exhausted" escape is unreachable when
+      every rung yields);
+    - a slow job still gets its silent bounded re-runs (regression guard for the feature).
+    """
+    fast = create_job("fast poll", "every 2m")
+    assert mark_job_run(fast["id"], False, "ConnectError: dns", model_unreachable=True)
+    j = get_job(fast["id"])
+    assert j is not None
+    assert j.get(ur.STATE_KEY) is None, "2m cadence beats the 5m rung: plan_retry yields"
+    assert ur.will_retry(j) is False, "yielded: no re-run is scheduled, notice must go out"
+
+    slow = create_job("nightly report", "every 24h")
+    assert mark_job_run(slow["id"], False, "ConnectError: dns", model_unreachable=True)
+    js = get_job(slow["id"])
+    assert js is not None
+    assert js[ur.STATE_KEY]["attempt"] == 1
+    assert ur.will_retry(js) is True, "5m rung beats the 24h cadence: re-run is scheduled"
+
+    mid = create_job("ten minute sync", "every 10m")
+    assert mark_job_run(mid["id"], False, "ConnectError: dns", model_unreachable=True)
+    jm = get_job(mid["id"])
+    assert jm is not None
+    # 10m cadence beats the 15m and 30m rungs: the ladder can never climb past attempt 1,
+    # so the exhaustion escape is unreachable and the notice must go out immediately.
+    assert jm[ur.STATE_KEY]["attempt"] == 1
+    assert ur.will_retry(jm) is False, "10m cadence beats the 15m rung: yielded, notice goes out"
+
+
 def test_reaching_the_model_resets_ladder_and_oneshots_never_retry(tmp_cron_home):
     """Any run that reached the model clears retry state; one-shots (pre-claimed
     dispatch, at-most-times #38758) never enter the ladder."""