From 8a573bb6e7d595efb6f2b94d7645c45a65d49469 Mon Sep 17 00:00:00 2001 From: joaomarcos Date: Tue, 7 Jul 2026 23:22:57 -0300 Subject: [PATCH] fix(cron): stop interrupted jobs from delivering their pre-kill output Follow-up to the previous commit on #60432. The status-write guard (_consume_interrupted_flag, checked right before mark_job_run) closes the false-success bookkeeping gap, but run_one_job delivers its result BEFORE that check: delivery happens right after run_job() returns, mark_job_run happens at the very end. A job whose tool subprocess was killed mid-flight can still produce a plausible-looking final_response from the truncated output, and that response would reach the user via _deliver_result before the interrupted flag was ever consulted -- correct status in jobs.json, wrong message already sent. Adds _is_interrupted(), a non-destructive peek at the same _interrupted_job_ids set (_consume_interrupted_flag stays as the consuming, authoritative check right before the status write -- this needed a peek instead since the flag has to still be visible there). Checked right after save_job_output, before the deliver_content decision: if the run looked successful but was flagged interrupted, force success=False with an explicit interruption message. This routes delivery through the existing _summarize_cron_failure_for_delivery path (the same one a real failure already uses) instead of the raw final_response, so the user gets an honest "this run was interrupted" instead of a truncated/misleading result. Testing: 4 new tests in tests/cron/test_shutdown_interrupt.py -- _is_interrupted peek semantics (false/true/does-not-clear, as opposed to the consuming _consume_interrupted_flag), and the delivery-gate test itself, which mocks run_job to return a normal-looking success with a "plausible final response" while the job is pre-marked interrupted, and asserts _deliver_result receives the failure summary ("This run was interrupted.") instead, with the summarizer's error argument confirmed to mention the interruption. Fail-then-pass: reverted cron/scheduler.py only, the 4 new tests fail (3 on the missing _is_interrupted attribute, 1 -- the delivery-gate test -- on _summarize_cron_failure_for_delivery never being called, i.e. the raw response would have gone out); restored, all 16 tests in the file pass. Regression: tests/cron/ (683 tests) + test_cron_active_work_drain.py + test_gateway_shutdown.py + test_shutdown_cache_cleanup.py -- 11 pre-existing failures (Unix file-permission-bit and path-tilde assertions that don't apply on this Windows dev box), matching the same set already established as pre-existing in the prior commit's regression check. Zero new failures. Continues #60432 --- cron/scheduler.py | 29 +++++++++++ tests/cron/test_shutdown_interrupt.py | 69 +++++++++++++++++++++++++++ 2 files changed, 98 insertions(+) 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."""