diff --git a/cron/scheduler.py b/cron/scheduler.py index c168f3fef9d..b4e6001d19a 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -428,6 +428,13 @@ def _interpreter_shutting_down(exc: Optional[BaseException] = None) -> bool: if sys.is_finalizing(): return True if exc is not None: + # Match the SHORT prefix deliberately: CPython emits two shutdown + # variants — "cannot schedule new futures after interpreter shutdown" + # (asyncio.run_coroutine_threadsafe / a torn-down default executor) and + # "cannot schedule new futures after shutdown" (a plain + # ThreadPoolExecutor). Both are documented in #58720. The common prefix + # catches both; the sibling agent/tool_executor._is_interpreter_shutdown_submit_error + # matches only the fuller "...after interpreter shutdown" form. return "cannot schedule new futures" in str(exc).lower() return False @@ -2376,10 +2383,22 @@ def _guard_job_credential_exfil(job: dict) -> None: raise RuntimeError(f"Cron job '{job_id}' blocked for safety: {err}") -def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]: +def run_job( + job: dict, *, defer_agent_teardown: Optional[list] = None +) -> tuple[bool, str, str, Optional[str]]: """ Execute a single cron job. - + + ``defer_agent_teardown``: when a caller passes a list, ``run_job`` skips + the agent's async-resource teardown (``agent.close()`` + + ``cleanup_stale_async_clients()``) in its ``finally`` block and instead + appends the live agent to that list. The caller is then responsible for + calling ``_teardown_cron_agent(agent)`` AFTER it has delivered the result. + This closes the ordering window in #58720 where delivery ran against a + torn-down async client (defense-in-depth alongside the interpreter-shutdown + guard). When ``None`` (the default) teardown happens inline as before, so + every existing caller is unchanged. + Returns: Tuple of (success, full_output_doc, final_response, error_message) """ @@ -3194,20 +3213,40 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]: # main OpenAI/httpx client held by this ephemeral cron agent. Without # this, a gateway that ticks cron every N minutes leaks fds per job # until it hits EMFILE (#10200 / "too many open files"). - try: + # + # When the caller opted to defer teardown (passed a list), hand the live + # agent back instead of closing it here — delivery must run against a + # live async client, and the caller tears down afterwards (#58720). + if defer_agent_teardown is not None: if agent is not None: - agent.close() - except (Exception, KeyboardInterrupt) as e: - logger.debug("Job '%s': failed to close agent resources: %s", job_id, e) - # Each cron run spins up a short-lived worker thread whose event loop - # dies as soon as the ``ThreadPoolExecutor`` shuts down. Any async - # httpx clients cached under that loop are now unusable — reap them - # so their transports don't accumulate in the process-global cache. - try: - from agent.auxiliary_client import cleanup_stale_async_clients - cleanup_stale_async_clients() - except Exception as e: - logger.debug("Job '%s': failed to reap stale auxiliary clients: %s", job_id, e) + defer_agent_teardown.append(agent) + else: + _teardown_cron_agent(agent, job_id) + + +def _teardown_cron_agent(agent, job_id: str) -> None: + """Release an ephemeral cron agent's async resources. + + Split out of ``run_job``'s ``finally`` so a caller that defers teardown + (to deliver first — #58720) can invoke the identical cleanup AFTER delivery. + Closes the agent (subprocesses, sandboxes, browser daemons, OpenAI/httpx + client) and reaps stale async clients whose loop has since closed. Idempotent + and independently guarded, matching the original inline behavior. + """ + try: + if agent is not None: + agent.close() + except (Exception, KeyboardInterrupt) as e: + logger.debug("Job '%s': failed to close agent resources: %s", job_id, e) + # Each cron run spins up a short-lived worker thread whose event loop + # dies as soon as the ``ThreadPoolExecutor`` shuts down. Any async + # httpx clients cached under that loop are now unusable — reap them + # so their transports don't accumulate in the process-global cache. + try: + from agent.auxiliary_client import cleanup_stale_async_clients + cleanup_stale_async_clients() + except Exception as e: + 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: @@ -3256,40 +3295,72 @@ def run_one_job(job: dict, *, adapters=None, loop=None, verbose: bool = False) - _scope_token = set_secret_scope( build_profile_secret_scope(_get_hermes_home()) ) + # Defer the cron agent's async-resource teardown until AFTER delivery. + # run_job normally closes the agent (and reaps stale async clients) in + # its finally block; doing that before _deliver_result runs means the + # live send races a torn-down async client (#58720). Passing a holder + # list makes run_job hand the agent back instead, and we tear it down + # below once delivery is done. Defense-in-depth alongside the + # interpreter-shutdown guard in _deliver_result. + _deferred_agents: list = [] try: - success, output, final_response, error = run_job(job) + success, output, final_response, error = run_job( + job, defer_agent_teardown=_deferred_agents + ) + except BaseException: + # run_job's finally still hands back the agent when it raises; tear + # it down here so a failed run never leaks its async resources + # (#10200), then re-raise into the outer handler. BaseException + # (not just Exception) so a KeyboardInterrupt/SystemExit mid-run + # still triggers teardown before propagating. + for _deferred_agent in _deferred_agents: + _teardown_cron_agent(_deferred_agent, job["id"]) + raise finally: reset_secret_scope(_scope_token) - output_file = save_job_output(job["id"], output) - if verbose: - logger.info("Output saved to: %s", output_file) - - # 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. - deliver_content = final_response if success else _summarize_cron_failure_for_delivery(job, error) - # Treat whitespace-only final responses the same as empty - # responses: do not deliver a blank message, and let the - # empty-response guard below mark the run as a soft failure. - should_deliver = bool(deliver_content.strip()) - # Cron silence suppression — see _is_cron_silence_response. Replaces the - # old `SILENT_MARKER in ...upper()` substring check, which both leaked - # bracketless near-markers ("SILENT" / "NO_REPLY") and wrongly swallowed - # a real report that merely quoted "[SILENT]" mid-sentence (#51438, - # #46917). Keeps the intentional bracketed-prefix / trailing-line - # tolerance the cron contract relies on. - if should_deliver and success and _is_cron_silence_response(deliver_content): - logger.info("Job '%s': agent returned %s — skipping delivery", job["id"], SILENT_MARKER) - should_deliver = False - + # Everything from here through delivery runs with the agent still live + # (deferred teardown). Wrap it ALL in a try/finally so that if any step + # between run_job returning and delivery — save_job_output, the [SILENT] + # / empty-response computation, or _deliver_result itself — raises, the + # deferred agent is still torn down. Otherwise the outer `except` would + # swallow the error and leak the agent's subprocesses/clients (#10200). delivery_error = None - if should_deliver: - try: - delivery_error = _deliver_result(job, deliver_content, adapters=adapters, loop=loop) - except Exception as de: - delivery_error = str(de) - logger.error("Delivery failed for job %s: %s", job["id"], de) + try: + output_file = save_job_output(job["id"], output) + if verbose: + logger.info("Output saved to: %s", output_file) + + # 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. + deliver_content = final_response if success else _summarize_cron_failure_for_delivery(job, error) + # Treat whitespace-only final responses the same as empty + # responses: do not deliver a blank message, and let the + # empty-response guard below mark the run as a soft failure. + should_deliver = bool(deliver_content.strip()) + # Cron silence suppression — see _is_cron_silence_response. Replaces the + # old `SILENT_MARKER in ...upper()` substring check, which both leaked + # bracketless near-markers ("SILENT" / "NO_REPLY") and wrongly swallowed + # a real report that merely quoted "[SILENT]" mid-sentence (#51438, + # #46917). Keeps the intentional bracketed-prefix / trailing-line + # tolerance the cron contract relies on. + if should_deliver and success and _is_cron_silence_response(deliver_content): + logger.info("Job '%s': agent returned %s — skipping delivery", job["id"], SILENT_MARKER) + should_deliver = False + + if should_deliver: + try: + delivery_error = _deliver_result(job, deliver_content, adapters=adapters, loop=loop) + except Exception as de: + delivery_error = str(de) + logger.error("Delivery failed for job %s: %s", job["id"], de) + finally: + # Tear down the deferred agent(s) now that save + delivery have run + # (or raised). Must happen on every path so cron agents never leak + # their subprocesses/clients (#10200). + for _deferred_agent in _deferred_agents: + _teardown_cron_agent(_deferred_agent, job["id"]) # Treat empty final_response as a soft failure so last_status # is not "ok" — the agent ran but produced nothing useful. diff --git a/tests/cron/test_cron_workdir.py b/tests/cron/test_cron_workdir.py index d8efdfb4855..253bde4769b 100644 --- a/tests/cron/test_cron_workdir.py +++ b/tests/cron/test_cron_workdir.py @@ -220,7 +220,7 @@ class TestTickWorkdirPartition: calls: list[tuple[str, str]] = [] order_lock = threading.Lock() - def fake_run_job(job): + def fake_run_job(job, *, defer_agent_teardown=None): # Return a minimal tuple matching run_job's signature. with order_lock: calls.append((job["id"], threading.current_thread().name)) diff --git a/tests/cron/test_parallel_pool.py b/tests/cron/test_parallel_pool.py index 01e65adc4fb..4c4d3f4887e 100644 --- a/tests/cron/test_parallel_pool.py +++ b/tests/cron/test_parallel_pool.py @@ -84,7 +84,7 @@ class TestRunningJobGuard: dispatched = [] monkeypatch.setattr(sched, "get_due_jobs", lambda: [job]) monkeypatch.setattr(sched, "advance_next_run", lambda *_a, **_kw: None) - monkeypatch.setattr(sched, "run_job", lambda j: dispatched.append(j["id"]) or (True, "out", "resp", None)) + monkeypatch.setattr(sched, "run_job", lambda j, **_kw: dispatched.append(j["id"]) or (True, "out", "resp", None)) monkeypatch.setattr(sched, "save_job_output", lambda *_a, **_kw: None) monkeypatch.setattr(sched, "mark_job_run", lambda *_a, **_kw: None) monkeypatch.setattr(sched, "_deliver_result", lambda *_a, **_kw: None) @@ -117,7 +117,7 @@ class TestSyncMode: monkeypatch.setattr(sched, "get_due_jobs", lambda: jobs) monkeypatch.setattr(sched, "advance_next_run", lambda *_a, **_kw: None) - monkeypatch.setattr(sched, "run_job", lambda j: (True, "out", "resp", None)) + monkeypatch.setattr(sched, "run_job", lambda j, **_kw: (True, "out", "resp", None)) monkeypatch.setattr(sched, "save_job_output", lambda *_a, **_kw: "/tmp/out") monkeypatch.setattr(sched, "mark_job_run", lambda *_a, **_kw: None) monkeypatch.setattr(sched, "_deliver_result", lambda *_a, **_kw: None) @@ -147,7 +147,7 @@ class TestSyncMode: barrier = threading.Barrier(2, timeout=5) - def slow_run(j): + def slow_run(j, *, defer_agent_teardown=None): barrier.wait() # blocks until test thread also waits return True, "out", "resp", None @@ -201,7 +201,7 @@ class TestSequentialPool: barrier = threading.Barrier(2, timeout=5) - def slow_run(j): + def slow_run(j, *, defer_agent_teardown=None): barrier.wait() return True, "out", "resp", None @@ -249,7 +249,7 @@ class TestSequentialPool: dispatched = [] monkeypatch.setattr(sched, "get_due_jobs", lambda: [job]) monkeypatch.setattr(sched, "advance_next_run", lambda *_a, **_kw: None) - monkeypatch.setattr(sched, "run_job", lambda j: dispatched.append(j["id"]) or (True, "out", "resp", None)) + monkeypatch.setattr(sched, "run_job", lambda j, **_kw: dispatched.append(j["id"]) or (True, "out", "resp", None)) monkeypatch.setattr(sched, "save_job_output", lambda *_a, **_kw: None) monkeypatch.setattr(sched, "mark_job_run", lambda *_a, **_kw: None) monkeypatch.setattr(sched, "_deliver_result", lambda *_a, **_kw: None) diff --git a/tests/cron/test_run_one_job.py b/tests/cron/test_run_one_job.py index c8528109d9a..decb6c4e35f 100644 --- a/tests/cron/test_run_one_job.py +++ b/tests/cron/test_run_one_job.py @@ -18,7 +18,7 @@ def _patch_pipeline(monkeypatch, *, success=True, output="out", final="final res """Patch the job pipeline primitives and record the call order.""" calls = [] - def fake_run_job(job): + def fake_run_job(job, *, defer_agent_teardown=None): calls.append(("run_job", job["id"])) fr = final if silent_marker_in is None else silent_marker_in return (success, output, fr, error) @@ -103,7 +103,7 @@ def test_run_one_job_failed_job_delivers_error(monkeypatch): def test_run_one_job_exception_marks_failure(monkeypatch): """If run_job raises, the helper marks the run failed and returns False rather than propagating.""" - def boom(job): + def boom(job, *, defer_agent_teardown=None): raise RuntimeError("kaboom") monkeypatch.setattr(s, "run_job", boom) @@ -136,7 +136,7 @@ def test_run_one_job_installs_secret_scope_under_multiplex(monkeypatch, tmp_path scope_during_run = {} - def fake_run_job(job): + def fake_run_job(job, *, defer_agent_teardown=None): # This is where resolve_runtime_provider() would read a secret. Prove a # scope is installed and the profile's secret resolves without raising. scope_during_run["scope"] = ss.current_secret_scope() @@ -160,3 +160,117 @@ def test_run_one_job_installs_secret_scope_under_multiplex(monkeypatch, tmp_path assert scope_during_run["base_url"] == "https://openrouter.ai/api/v1" # And it was torn down after run_one_job returned (no leak). assert ss.current_secret_scope() is None + + +def test_run_one_job_delivers_before_agent_teardown(monkeypatch): + """Regression for #58720: the cron agent's async-resource teardown + (agent.close + cleanup_stale_async_clients) MUST run AFTER delivery, not + before. run_job defers teardown by appending the live agent to the holder + list; run_one_job tears it down only after _deliver_result has run. If the + order flips, delivery races a torn-down async client and dies with + 'cannot schedule new futures after interpreter shutdown'. + """ + order = [] + + class FakeAgent: + def close(self): + order.append("agent.close") + + def fake_run_job(job, *, defer_agent_teardown=None): + order.append("run_job") + # Mimic run_job's deferral contract: hand the live agent back so the + # caller tears it down after delivery instead of in run_job's finally. + assert defer_agent_teardown is not None, "run_one_job must defer teardown" + defer_agent_teardown.append(FakeAgent()) + return (True, "out", "final response", None) + + def fake_deliver(job, content, adapters=None, loop=None): + order.append("deliver") + return None + + monkeypatch.setattr(s, "run_job", fake_run_job) + monkeypatch.setattr(s, "save_job_output", lambda jid, out: f"/tmp/{jid}.txt") + monkeypatch.setattr(s, "_deliver_result", fake_deliver) + monkeypatch.setattr(s, "mark_job_run", lambda *a, **k: None) + # cleanup_stale_async_clients is imported lazily inside _teardown_cron_agent; + # stub it so the teardown records its own marker without touching real caches. + import agent.auxiliary_client as aux + monkeypatch.setattr(aux, "cleanup_stale_async_clients", + lambda: order.append("cleanup_stale")) + + ok = s.run_one_job({"id": "j8", "name": "t"}) + + assert ok is True + # Delivery must strictly precede agent teardown + stale-client reap. + assert order == ["run_job", "deliver", "agent.close", "cleanup_stale"], order + + +def test_run_one_job_tears_down_deferred_agent_when_delivery_raises(monkeypatch): + """Even if _deliver_result raises, the deferred agent is still torn down + (no fd/client leak — #10200). Teardown lives in a finally around delivery. + """ + order = [] + + class FakeAgent: + def close(self): + order.append("agent.close") + + def fake_run_job(job, *, defer_agent_teardown=None): + defer_agent_teardown.append(FakeAgent()) + return (True, "out", "final response", None) + + def boom_deliver(job, content, adapters=None, loop=None): + order.append("deliver-raise") + raise RuntimeError("send blew up") + + monkeypatch.setattr(s, "run_job", fake_run_job) + monkeypatch.setattr(s, "save_job_output", lambda jid, out: f"/tmp/{jid}.txt") + monkeypatch.setattr(s, "_deliver_result", boom_deliver) + monkeypatch.setattr(s, "mark_job_run", lambda *a, **k: None) + import agent.auxiliary_client as aux + monkeypatch.setattr(aux, "cleanup_stale_async_clients", + lambda: order.append("cleanup_stale")) + + ok = s.run_one_job({"id": "j9", "name": "t"}) + + assert ok is True # delivery error is recorded, not propagated + assert order == ["deliver-raise", "agent.close", "cleanup_stale"], order + + +def test_run_one_job_tears_down_deferred_agent_when_save_raises(monkeypatch): + """#58720 W1: if save_job_output (or the [SILENT]/empty computation) raises + AFTER run_job hands the agent back but BEFORE delivery, the deferred agent + must still be torn down. The outer `except` would otherwise swallow the + error and leak the agent (#10200). Teardown lives in a finally spanning + save→deliver. + """ + order = [] + + class FakeAgent: + def close(self): + order.append("agent.close") + + def fake_run_job(job, *, defer_agent_teardown=None): + defer_agent_teardown.append(FakeAgent()) + return (True, "out", "final response", None) + + def boom_save(jid, out): + order.append("save-raise") + raise RuntimeError("disk full") + + monkeypatch.setattr(s, "run_job", fake_run_job) + monkeypatch.setattr(s, "save_job_output", boom_save) + monkeypatch.setattr(s, "_deliver_result", + lambda *a, **k: order.append("deliver")) + monkeypatch.setattr(s, "mark_job_run", lambda *a, **k: None) + import agent.auxiliary_client as aux + monkeypatch.setattr(aux, "cleanup_stale_async_clients", + lambda: order.append("cleanup_stale")) + + ok = s.run_one_job({"id": "j10", "name": "t"}) + + # save raised → outer handler marks failure and returns False, but the + # deferred agent was still torn down (no delivery, no leak). + assert ok is False + assert "deliver" not in order + assert order == ["save-raise", "agent.close", "cleanup_stale"], order diff --git a/tests/cron/test_scheduler.py b/tests/cron/test_scheduler.py index 104147cf471..775840ecec8 100644 --- a/tests/cron/test_scheduler.py +++ b/tests/cron/test_scheduler.py @@ -2589,7 +2589,7 @@ class TestOneShotDispatchClaim: order = [] with patch("cron.scheduler.get_due_jobs", return_value=[self._oneshot()]), \ patch("cron.scheduler.claim_dispatch", side_effect=lambda _id: order.append("claim") or True), \ - patch("cron.scheduler.run_job", side_effect=lambda _j: order.append("run") or (True, "# out", "ok", None)), \ + patch("cron.scheduler.run_job", side_effect=lambda _j, **_kw: order.append("run") or (True, "# out", "ok", None)), \ patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \ patch("cron.scheduler._deliver_result"), \ patch("cron.scheduler.mark_job_run"): @@ -3010,7 +3010,7 @@ class TestParallelTick: barrier = threading.Barrier(2, timeout=5) call_order = [] - def mock_run_job(job): + def mock_run_job(job, *, defer_agent_teardown=None): """Each job hits a barrier — both must be active simultaneously.""" call_order.append(("start", job["id"])) barrier.wait() # blocks until both threads reach here @@ -3044,7 +3044,7 @@ class TestParallelTick: from gateway.session_context import get_session_env seen = {} - def mock_run_job(job): + def mock_run_job(job, *, defer_agent_teardown=None): origin = job.get("origin", {}) # run_job sets ContextVars — verify each job sees its own from gateway.session_context import set_session_vars, clear_session_vars @@ -3084,7 +3084,7 @@ class TestParallelTick: monkeypatch.setenv("HERMES_CRON_MAX_PARALLEL", "1") call_times = [] - def mock_run_job(job): + def mock_run_job(job, *, defer_agent_teardown=None): import time call_times.append(("start", job["id"], time.monotonic())) time.sleep(0.05)