diff --git a/cron/scheduler.py b/cron/scheduler.py index 4c764bd13a44..3c3cf975bd26 100644 --- a/cron/scheduler.py +++ b/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 diff --git a/tests/cron/test_run_one_job.py b/tests/cron/test_run_one_job.py index 7da6b1c14f41..cdfd612c5d03 100644 --- a/tests/cron/test_run_one_job.py +++ b/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.""" diff --git a/tests/tools/test_cronjob_run_immediate.py b/tests/tools/test_cronjob_run_immediate.py index 9efa60e82cb1..64b6ff7c603b 100644 --- a/tests/tools/test_cronjob_run_immediate.py +++ b/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), \ diff --git a/tools/cronjob_tools.py b/tools/cronjob_tools.py index 02ac58f9c608..5851bc155721 100644 --- a/tools/cronjob_tools.py +++ b/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: