fix: report immediate cron run status for depleted repeat jobs

cron/scheduler.py

@@ -3082,7 +3082,8 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
             logger.debug("Job '%s': failed to reap stale auxiliary clients: %s", job_id, e)
 
 
-def run_one_job(job: dict, *, adapters=None, loop=None, verbose: bool = False) -> bool:
+def run_one_job(job: dict, *, adapters=None, loop=None, verbose: bool = False,
+                return_result: bool = False):
     """Run ONE due job end-to-end: execute → save output → deliver → mark.
 
     This is the shared firing body extracted from ``tick``'s per-job closure so
@@ -3094,8 +3095,12 @@ def run_one_job(job: dict, *, adapters=None, loop=None, verbose: bool = False) -
     under the file lock before dispatch; an external provider claims via the
     store CAS). This function only fires the given job once.
 
-    Returns True if the job was processed (even if the job itself failed —
-    failure is recorded via ``mark_job_run``), False only if processing raised.
+    By default this returns the historical bool: True if the job was processed
+    (even if the job itself failed — failure is recorded via ``mark_job_run``),
+    False only if processing raised. With ``return_result=True``, return a
+    structured snapshot of the run status so immediate-run callers can report
+    status even when a finite repeat job is removed by ``mark_job_run`` before
+    they can re-read it from the job store.
     """
     try:
         # Pre-run dispatch claim (issue #38758): atomically commit a finite
@@ -3110,7 +3115,15 @@ def run_one_job(job: dict, *, adapters=None, loop=None, verbose: bool = False) -
                 "Job '%s': one-shot dispatch limit reached — skipping",
                 job.get("name", job["id"]),
             )
-            return True  # not an error — already handled/removed
+            result = {
+                "processed": True,
+                "success": True,
+                "status": "ok",
+                "error": None,
+                "delivery_error": None,
+                "skipped": True,
+            }
+            return result if return_result else True  # not an error — already handled/removed
 
         success, output, final_response, error = run_job(job)
 
@@ -3152,11 +3165,27 @@ def run_one_job(job: dict, *, adapters=None, loop=None, verbose: bool = False) -
             error = "Agent completed but produced empty response (model error, timeout, or misconfiguration)"
 
         mark_job_run(job["id"], success, error, delivery_error=delivery_error)
+        if return_result:
+            return {
+                "processed": True,
+                "success": bool(success),
+                "status": "ok" if success else "error",
+                "error": error,
+                "delivery_error": delivery_error,
+            }
         return True
 
     except Exception as e:
         logger.error("Error processing job %s: %s", job['id'], e)
         mark_job_run(job["id"], False, str(e))
+        if return_result:
+            return {
+                "processed": False,
+                "success": False,
+                "status": "error",
+                "error": str(e),
+                "delivery_error": None,
+            }
         return False
 
 
@@ -3260,7 +3289,7 @@ def _process_job(job: dict) -> bool:
             module-level ``run_one_job`` so ``tick`` and external providers
             (Chronos ``fire_due``) use the identical execute→save→deliver→mark
             body."""
-            return run_one_job(job, adapters=adapters, loop=loop, verbose=verbose)
+            return bool(run_one_job(job, adapters=adapters, loop=loop, verbose=verbose))
 
         # Partition due jobs: those with a per-job workdir mutate
         # os.environ["TERMINAL_CWD"] inside run_job, which is process-global, so

tests/cron/test_run_one_job.py

@@ -66,6 +66,40 @@ def test_run_one_job_success_sequence(monkeypatch):
     assert calls[-1] == ("mark", "j2", True)
 
 
+def test_run_one_job_return_result_success_snapshot(monkeypatch):
+    """return_result=True exposes the success snapshot without changing the pipeline."""
+    calls = _patch_pipeline(monkeypatch)
+
+    result = s.run_one_job({"id": "j2r", "name": "t"}, return_result=True)
+
+    assert result == {
+        "processed": True,
+        "success": True,
+        "status": "ok",
+        "error": None,
+        "delivery_error": None,
+    }
+    assert [c[0] for c in calls] == ["run_job", "save", "deliver", "mark"]
+    assert calls[-1] == ("mark", "j2r", True)
+
+
+def test_run_one_job_return_result_failed_snapshot(monkeypatch):
+    """return_result=True distinguishes a processed failed job from an exception."""
+    calls = _patch_pipeline(monkeypatch, success=False, final="", error="boom")
+
+    result = s.run_one_job({"id": "j5r", "name": "t"}, return_result=True)
+
+    assert result == {
+        "processed": True,
+        "success": False,
+        "status": "error",
+        "error": "boom",
+        "delivery_error": None,
+    }
+    assert "deliver" in [c[0] for c in calls]
+    assert [c for c in calls if c[0] == "mark"][0] == ("mark", "j5r", False)
+
+
 def test_run_one_job_silent_skips_delivery(monkeypatch):
     """A [SILENT] final response saves output + marks the run but does NOT
     deliver."""

tests/tools/test_cronjob_run_immediate.py

@@ -59,6 +59,43 @@ def test_run_reports_failure_from_last_status(self):
         assert out["job"]["execution_success"] is False
         assert out["job"]["execution_error"] == "provider 500"
 
+    def test_run_repeat1_depleted_success_reports_success(self):
+        """A repeat=1 job removed after a successful run still reports success."""
+        run_result = {"processed": True, "success": True, "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", return_value=run_result), \
+             patch("tools.cronjob_tools.get_job", return_value=None):
+            out = json.loads(cronjob(action="run", job_id="job-run-1"))
+
+        assert out["job"]["executed"] is True
+        assert out["job"]["execution_success"] is True
+
+    def test_run_repeat1_depleted_failure_reports_failure(self):
+        """A repeat=1 job removed after a failed run still reports failure."""
+        run_result = {"processed": True, "success": False, "error": "Script exited with code 7"}
+        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", return_value=run_result), \
+             patch("tools.cronjob_tools.get_job", return_value=None):
+            out = json.loads(cronjob(action="run", job_id="job-run-1"))
+
+        assert out["job"]["executed"] is True
+        assert out["job"]["execution_success"] is False
+        assert out["job"]["execution_error"] == "Script exited with code 7"
+
+    def test_run_repeat1_depleted_empty_no_agent_reports_silent_success(self):
+        """A removed repeat=1 no_agent empty-stdout run keeps silent success semantics."""
+        run_result = {"processed": True, "success": True, "error": None, "status": "ok"}
+        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", return_value=run_result), \
+             patch("tools.cronjob_tools.get_job", return_value=None):
+            out = json.loads(cronjob(action="run", job_id="job-run-1"))
+
+        assert out["job"]["executed"] is True
+        assert out["job"]["execution_success"] is True
+
     def test_execute_job_now_bails_without_claim(self):
         """_execute_job_now never calls run_one_job when the claim is lost."""
         with patch("tools.cronjob_tools.claim_job_for_fire", return_value=False), \

tools/cronjob_tools.py

@@ -627,14 +627,25 @@ def _execute_job_now(job: Dict[str, Any]) -> Dict[str, Any]:
                     "error": "Job is already being fired by the scheduler; not run again."}
 
         # run_one_job records last_run_at/last_status via mark_job_run (which
-        # also clears the fire claim) and returns True iff it processed the job.
-        processed = run_one_job(job)
+        # also clears the fire claim).  Ask for a structured status snapshot so
+        # repeat-limited one-shot jobs that are removed by mark_job_run can still
+        # report the real run outcome instead of relying only on a post-run
+        # get_job() re-read.
+        run_result = run_one_job(job, return_result=True)
         refreshed = get_job(job_id) or {}
-        ok = refreshed.get("last_status") == "ok"
+        if isinstance(run_result, dict):
+            processed = bool(run_result.get("processed", False))
+            ok = bool(run_result.get("success", False))
+            error = run_result.get("error")
+        else:
+            # Back-compat fallback for tests or older scheduler implementations.
+            processed = bool(run_result)
+            ok = refreshed.get("last_status") == "ok"
+            error = refreshed.get("last_error")
         return {
             "claimed": True,
             "success": bool(processed and ok),
-            "error": refreshed.get("last_error"),
+            "error": error,
         }
 
     except Exception as e: