fix(cron): deliver before tearing down the agent's async clients (#58720)

Defense-in-depth alongside the interpreter-shutdown guard: run_job closed
the cron agent's async resources (agent.close + cleanup_stale_async_clients)
in its finally block BEFORE run_one_job called _deliver_result, so a live
delivery could race a torn-down async client. run_job now accepts an optional
defer_agent_teardown holder; when set it hands the live agent back instead of
closing it, and run_one_job tears it down (via the extracted _teardown_cron_agent
helper) only AFTER delivery — in a finally so a failed run never leaks. Default
path (holder=None) is unchanged, so every existing caller keeps inline teardown.

Reorder approach based on #58777 by @LavyaTandel; reworked to keep a single
delivery site in run_one_job and add regression coverage.

Co-authored-by: LavyaTandel <lavya@loom.local>

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.

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))

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)

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

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)