diff --git a/cron/scheduler.py b/cron/scheduler.py index 1994a857fed..6b96fb2e1be 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -371,6 +371,21 @@ def mark_running_jobs_interrupted(reason: str) -> list: return marked +def _is_interrupted(job_id: str) -> bool: + """Non-destructive peek at whether the shutdown path has marked + ``job_id`` interrupted (see ``mark_running_jobs_interrupted``). + + Called by ``run_one_job`` BEFORE it decides what to deliver — a job + whose tool subprocess was killed mid-flight may still produce a + plausible-looking ``final_response`` from the truncated output, and + that must not go out to the user as if it were a normal result. + Unlike ``_consume_interrupted_flag`` below, this does not clear the + flag: the later, authoritative check (right before ``last_status`` is + written) still needs to see it.""" + with _running_lock: + return job_id in _interrupted_job_ids + + def _consume_interrupted_flag(job_id: str) -> bool: """Return True and clear the flag if the shutdown path already marked ``job_id`` interrupted (see ``mark_running_jobs_interrupted``). @@ -3420,6 +3435,20 @@ def run_one_job(job: dict, *, adapters=None, loop=None, verbose: bool = False) - if verbose: logger.info("Output saved to: %s", output_file) + # If the gateway shutdown killed this job's tool subprocess + # mid-flight (#60432), the agent may still have produced a + # plausible-looking final_response from the truncated output -- + # force the failure path so the delivered message is an honest + # "this run was interrupted" summary instead of that response. + # Peek-only: the flag stays set for the authoritative check + # right before mark_job_run below. + if success and _is_interrupted(job["id"]): + success = False + error = ( + "Interrupted by gateway shutdown before the run finished " + "(tool subprocess was killed mid-flight)." + ) + # Deliver the final response to the origin/target chat. # If the agent responded with [SILENT], skip delivery (but # output is already saved above). Failed jobs always deliver. diff --git a/tests/cron/test_shutdown_interrupt.py b/tests/cron/test_shutdown_interrupt.py index 2b5077fc9d2..a95567ff88b 100644 --- a/tests/cron/test_shutdown_interrupt.py +++ b/tests/cron/test_shutdown_interrupt.py @@ -113,6 +113,35 @@ class TestMarkRunningJobsInterrupted: assert marked == ["job-2"] +class TestIsInterrupted: + """Peek-only check used at the delivery gate -- must NOT clear the + flag, unlike _consume_interrupted_flag.""" + + def test_false_when_not_marked(self): + import cron.scheduler as sched + + assert sched._is_interrupted("job-1") is False + + def test_true_when_marked(self): + import cron.scheduler as sched + + sched._interrupted_job_ids.add("job-1") + + assert sched._is_interrupted("job-1") is True + + def test_does_not_clear_the_flag(self): + import cron.scheduler as sched + + sched._interrupted_job_ids.add("job-1") + + sched._is_interrupted("job-1") + + # Still set -- the later, authoritative check before mark_job_run + # must still see it. + assert "job-1" in sched._interrupted_job_ids + assert sched._is_interrupted("job-1") is True + + class TestConsumeInterruptedFlag: def test_false_when_not_marked(self): import cron.scheduler as sched @@ -165,6 +194,46 @@ class TestRunOneJobHonoursInterruptedFlag: # isn't permanently silenced. assert job["id"] not in sched._interrupted_job_ids + def test_interrupted_job_delivers_failure_summary_not_raw_response(self): + """The status-write guard alone isn't enough: delivery happens + BEFORE mark_job_run in run_one_job's own flow, so a job that kept + running post-kill and produced a plausible-looking final_response + must not have that response sent to the user just because the + eventual status write gets suppressed. Interrupted jobs must route + through the same failure-summary delivery path a real failure + would.""" + import cron.scheduler as sched + + job = self._make_job() + sched._interrupted_job_ids.add(job["id"]) + + with patch("cron.scheduler.claim_dispatch", return_value=True), \ + patch("agent.secret_scope.set_secret_scope", return_value=None), \ + patch("agent.secret_scope.build_profile_secret_scope", return_value=None), \ + patch("agent.secret_scope.reset_secret_scope"), \ + patch( + "cron.scheduler.run_job", + return_value=(True, "full output", "a plausible final response", None), + ), \ + patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \ + patch( + "cron.scheduler._summarize_cron_failure_for_delivery", + return_value="This run was interrupted.", + ) as mock_summarize, \ + patch("cron.scheduler._is_cron_silence_response", return_value=False), \ + patch("cron.scheduler._deliver_result", return_value=None) as mock_deliver, \ + patch("cron.scheduler.mark_job_run"): + result = sched.run_one_job(job) + + assert result is True + mock_summarize.assert_called_once() + # The summarizer's error argument must mention the interruption, + # not be silently None / the agent's own (possibly absent) error. + assert "interrupt" in mock_summarize.call_args.args[1].lower() + delivered_content = mock_deliver.call_args.args[1] + assert delivered_content == "This run was interrupted." + assert "plausible final response" not in delivered_content + def test_success_path_writes_normally_when_not_interrupted(self): """Control case: the guard must not swallow ordinary, un-interrupted completions -- only ones the shutdown path explicitly flagged."""