mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-08 13:12:08 +00:00
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
This commit is contained in:
parent
24e9ed73c2
commit
8a573bb6e7
2 changed files with 98 additions and 0 deletions
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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."""
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue