fix(cron): reconcile external provider after a claimed direct run (#70479)

A direct run (cronjob action='run' / webhook-triggered manual fire) takes
the job's fire_claim and advances next_run_at. When it races the external
provider's scheduled fire for the same occurrence, Chronos loses the claim
and — by design — does not re-arm (the winner owns the re-arm). But the
direct-run winner never notified the provider, so the NAS one-shot for the
consumed occurrence was left stale forever and the recurring job silently
stopped firing.

Observed in production: a managed 1-minute review job stalled for 20 hours
because a GitHub-webhook direct run claimed the job 2s before the Chronos
fire arrived; every subsequent occurrence was orphaned while /api/status
stayed green.

Fix: after a *claimed* direct execution completes (success or failure —
next_run_at advances at claim time either way), call
_notify_provider_jobs_changed_safe() so the active provider re-arms the
post-run next_run_at. No-op for the built-in ticker; claim-lost direct
runs still never notify (the winning scheduler owns the re-arm).

tests/tools/test_cronjob_run_immediate.py

@@ -32,12 +32,53 @@ class TestCronjobRunExecutesImmediately:
         m_claim.assert_called_once_with("job-run-1")   # at-most-once claim taken
         m_run.assert_called_once()                       # fired via the shared body
 
+    def test_run_reconciles_external_provider_after_claimed_execution(self):
+        """A direct run must re-arm Chronos after it advances next_run_at.
+
+        Otherwise a scheduled Chronos fire that loses its claim to this direct
+        run is consumed without a successor one-shot, permanently stalling the
+        recurring job.
+        """
+        order = []
+        ran = {"id": "job-run-1", "last_status": "ok", "last_error": None}
+        with patch("tools.cronjob_tools.resolve_job_ref", return_value=dict(_JOB)), \
+             patch("tools.cronjob_tools.claim_job_for_fire", return_value=True), \
+             patch("cron.scheduler.run_one_job",
+                   side_effect=lambda *a, **kw: order.append("run") or True), \
+             patch("tools.cronjob_tools.get_job", return_value=ran), \
+             patch("tools.cronjob_tools._notify_provider_jobs_changed_safe",
+                   side_effect=lambda: order.append("notify")) as m_notify:
+            out = json.loads(cronjob(action="run", job_id="job-run-1"))
+
+        assert out["job"]["executed"] is True
+        m_notify.assert_called_once_with()
+        # Reconcile only AFTER the run persisted its final state (mark_job_run
+        # inside run_one_job), so the provider arms the post-run next_run_at.
+        assert order == ["run", "notify"]
+
+    def test_run_reconciles_external_provider_even_when_claimed_run_fails(self):
+        """A claimed direct run advances next_run_at at claim time, so the
+        provider must be reconciled even when the execution itself fails."""
+        failed = {"id": "job-run-1", "last_status": "error", "last_error": "provider 500"}
+        with patch("tools.cronjob_tools.resolve_job_ref", return_value=dict(_JOB)), \
+             patch("tools.cronjob_tools.claim_job_for_fire", return_value=True), \
+             patch("cron.scheduler.run_one_job", side_effect=RuntimeError("boom")), \
+             patch("tools.cronjob_tools.mark_job_run"), \
+             patch("tools.cronjob_tools.get_job", return_value=failed), \
+             patch("tools.cronjob_tools._notify_provider_jobs_changed_safe") as m_notify:
+            out = json.loads(cronjob(action="run", job_id="job-run-1"))
+
+        assert out["job"]["executed"] is True
+        assert out["job"]["execution_success"] is False
+        m_notify.assert_called_once_with()
+
     def test_run_skips_when_claim_lost(self):
         """If the scheduler already holds the fire claim, do NOT double-run."""
         with patch("tools.cronjob_tools.resolve_job_ref", return_value=dict(_JOB)), \
              patch("tools.cronjob_tools.claim_job_for_fire", return_value=False), \
              patch("cron.scheduler.run_one_job") as m_run, \
-             patch("tools.cronjob_tools.get_job", return_value=dict(_JOB)):
+             patch("tools.cronjob_tools.get_job", return_value=dict(_JOB)), \
+             patch("tools.cronjob_tools._notify_provider_jobs_changed_safe") as m_notify:
             out = json.loads(cronjob(action="run", job_id="job-run-1"))
 
         assert out["success"] is True
@@ -45,6 +86,7 @@ class TestCronjobRunExecutesImmediately:
         assert out["job"]["execution_success"] is False
         assert "execution_skipped" in out["job"]
         m_run.assert_not_called()  # claim lost -> never fired
+        m_notify.assert_not_called()  # the winning scheduler owns the re-arm
 
     def test_run_reports_failure_from_last_status(self):
         """A failed run is reported via the re-read job's last_status/last_error."""

tools/cronjob_tools.py

@@ -842,6 +842,12 @@ def cronjob(
             # _execute_job_now advances next_run_at and blocks a concurrent tick
             # from double-firing.
             exec_result = _execute_job_now(job)
+            # A claimed direct run advances next_run_at and may race the
+            # external one-shot for the same occurrence. If Chronos loses that
+            # claim, its consumed fire cannot re-arm itself; reconcile from the
+            # winning direct path after the run has persisted its final state.
+            if exec_result.get("claimed", False):
+                _notify_provider_jobs_changed_safe()
             # Re-read so the response reflects the post-run last_run_at/last_status.
             result = _format_job(get_job(job_id) or {"id": job_id})
             result["executed"] = exec_result.get("claimed", False)