Cron agent timeout, shutdown, and session lifecycle bugs

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #9012 fix(run_agent): refresh activity during streaming responses

  merged to main

  files: run_agent.py, tests/run_agent/test_streaming.py

  Show the diff

  • run_agent.py

    @@ -4278,6 +4278,7 @@ def _run_codex_stream(self, api_kwargs: dict, client: Any = None, on_first_delta
                 try:
                     with active_client.responses.stream(**api_kwargs) as stream:
                         for event in stream:
    +                        self._touch_activity("receiving stream response")
                             if self._interrupt_requested:
                                 break
                             event_type = getattr(event, "type", "")
    @@ -4402,6 +4403,7 @@ def _run_codex_create_stream_fallback(self, api_kwargs: dict, client: Any = None
             collected_text_deltas: list = []
             try:
                 for event in stream_or_response:
    +                self._touch_activity("receiving stream response")
                     event_type = getattr(event, "type", None)
                     if not event_type and isinstance(event, dict):
                         event_type = event.get("type")
    @@ -5034,12 +5036,9 @@ def _call_chat_completions():
                 role = "assistant"
                 reasoning_parts: list = []
                 usage_obj = None
    -            _first_chunk_seen = False
                 for chunk in stream:
                     last_chunk_time["t"] = time.time()
    -                if not _first_chunk_seen:
    -                    _first_chunk_seen = True
    -                    self._touch_activity("receiving stream response")
    +                self._touch_activity("receiving stream response")
     
                     if self._interrupt_requested:
                         break
    @@ -5215,6 +5214,7 @@ def _call_anthropic():
                         # actively arriving (the chat_completions path
                         # already does this at the top of its chunk loop).
                         last_chunk_time["t"] = time.time()
    +                    self._touch_activity("receiving stream response")
     
                         if self._interrupt_requested:
                             break

  • tests/run_agent/test_streaming.py

    @@ -291,6 +291,38 @@ def test_on_first_delta_fires_once(self, mock_close, mock_create):
     
             assert len(first_delta_calls) == 1
     
    +    @patch("run_agent.AIAgent._create_request_openai_client")
    +    @patch("run_agent.AIAgent._close_request_openai_client")
    +    def test_chat_stream_refreshes_activity_on_every_chunk(self, mock_close, mock_create):
    +        """Each streamed chat chunk should refresh the activity timestamp."""
    +        from run_agent import AIAgent
    +
    +        chunks = [
    +            _make_stream_chunk(content="a"),
    +            _make_stream_chunk(content="b"),
    +            _make_stream_chunk(finish_reason="stop"),
    +        ]
    +
    +        mock_client = MagicMock()
    +        mock_client.chat.completions.create.return_value = iter(chunks)
    +        mock_create.return_value = mock_client
    +
    +        agent = AIAgent(
    +            model="test/model",
    +            quiet_mode=True,
    +            skip_context_files=True,
    +            skip_memory=True,
    +        )
    +        agent.api_mode = "chat_completions"
    +        agent._interrupt_requested = False
    +
    +        touch_calls = []
    +        agent._touch_activity = lambda desc: touch_calls.append(desc)
    +
    +        agent._interruptible_streaming_api_call({})
    +
    +        assert touch_calls.count("receiving stream response") == len(chunks)
    +
         @patch("run_agent.AIAgent._create_request_openai_client")
         @patch("run_agent.AIAgent._close_request_openai_client")
         def test_tool_only_does_not_fire_callback(self, mock_close, mock_create):
    @@ -693,6 +725,55 @@ def test_codex_text_delta_fires_callback(self):
             response = agent._run_codex_stream({}, client=mock_client)
             assert "Hello from Codex!" in deltas
     
    +    def test_codex_stream_refreshes_activity_on_every_event(self):
    +        from run_agent import AIAgent
    +
    +        agent = AIAgent(
    +            model="test/model",
    +            quiet_mode=True,
    +            skip_context_files=True,
    +            skip_memory=True,
    +        )
    +        agent.api_mode = "codex_responses"
    +        agent._interrupt_requested = False
    +
    +        touch_calls = []
    +        agent._touch_activity = lambda desc: touch_calls.append(desc)
    +
    +        mock_event_text_1 = SimpleNamespace(
    +            type="response.output_text.delta",
    +            delta="Hello",
    +        )
    +        mock_event_text_2 = SimpleNamespace(
    +            type="response.output_text.delta",
    +            delta=" world",
    +        )
    +        mock_event_done = SimpleNamespace(
    +            type="response.completed",
    +            delta="",
    +        )
    +
    +        mock_stream = MagicMock()
    +        mock_stream.__enter__ = MagicMock(return_value=mock_stream)
    +        mock_stream.__exit__ = MagicMock(return_value=False)
    +        mock_stream.__iter__ = MagicMock(
    +            return_value=iter([mock_event_text_1, mock_event_text_2, mock_event_done])
    +        )
    +        mock_stream.get_final_response.return_value = SimpleNamespace(
    +            output=[SimpleNamespace(
    +                type="message",
    +                content=[SimpleNamespace(type="output_text", text="Hello world")],
    +            )],
    +            status="completed",
    +        )
    +
    +        mock_client = MagicMock()
    +        mock_client.responses.stream.return_value = mock_stream
    +
    +        agent._run_codex_stream({}, client=mock_client)
    +
    +        assert touch_calls.count("receiving stream response") == 3
    +
         def test_codex_remote_protocol_error_falls_back_to_create_stream(self):
             from run_agent import AIAgent
             import httpx
    @@ -724,3 +805,102 @@ def test_codex_remote_protocol_error_falls_back_to_create_stream(self):
     
             assert response is fallback_response
             mock_fallback.assert_called_once_with({}, client=mock_client)
    +
    +    def test_codex_create_stream_fallback_refreshes_activity_on_every_event(self):
    +        from run_agent import AIAgent
    +
    +        agent = AIAgent(
    +            model="test/model",
    +            quiet_mode=True,
    +            skip_context_files=True,
    +            skip_memory=True,
    +        )
    +        agent.api_mode = "codex_responses"
    +
    +        touch_calls = []
    +        agent._touch_activity = lambda desc: touch_calls.append(desc)
    +
    +        events = [
    +            SimpleNamespace(type="response.output_text.delta", delta="Hello"),
    +            SimpleNamespace(type="response.output_item.done", item=SimpleNamespace(type="message")),
    +            SimpleNamespace(
    +                type="response.completed",
    +                response=SimpleNamespace(
    +                    output=[SimpleNamespace(
    +                        type="message",
    +                        content=[SimpleNamespace(type="output_text", text="Hello")],
    +                    )]
    +                ),
    +            ),
    +        ]
    +
    +        class _FakeCreateStream:
    +            def __iter__(self_inner):
    +                return iter(events)
    +
    +            def close(self_inner):
    +                return None
    +
    +        mock_stream = _FakeCreateStream()
    +
    +        mock_client = MagicMock()
    +        mock_client.responses.create.return_value = mock_stream
    +
    +        agent._run_codex_create_stream_fallback(
    +            {"model": "test/model", "instructions": "hi", "input": []},
    +            client=mock_client,
    +        )
    +
    +        assert touch_calls.count("receiving stream response") == len(events)
    +
    +
    +class TestAnthropicStreamCallbacks:
    +    """Verify Anthropic streaming refreshes activity on every event."""
    +
    +    def test_anthropic_stream_refreshes_activity_on_every_event(self):
    +        from run_agent import AIAgent
    +
    +        agent = AIAgent(
    +            model="test/model",
    +            quiet_mode=True,
    +            skip_context_files=True,
    +            skip_memory=True,
    +        )
    +        agent.api_mode = "anthropic_messages"
    +        agent._interrupt_requested = False
    +
    +        touch_calls = []
    +        agent._touch_activity = lambda desc: touch_calls.append(desc)
    +
    +        events = [
    +            SimpleNamespace(
    +                type="content_block_delta",
    +                delta=SimpleNamespace(type="text_delta", text="Hello"),
    +            ),
    +            SimpleNamespace(
    +                type="content_block_delta",
    +                delta=SimpleNamespace(type="thinking_delta", thinking="thinking"),
    +            ),
    +            SimpleNamespace(
    +                type="content_block_start",
    +                content_block=SimpleNamespace(type="tool_use", name="terminal"),
    +            ),
    +        ]
    +
    +        final_message = SimpleNamespace(
    +            content=[],
    +            stop_reason="end_turn",
    +        )
    +
    +        mock_stream = MagicMock()
    +        mock_stream.__enter__ = MagicMock(return_value=mock_stream)
    +        mock_stream.__exit__ = MagicMock(return_value=False)
    +        mock_stream.__iter__ = MagicMock(return_value=iter(events))
    +        mock_stream.get_final_message.return_value = final_message
    +
    +        agent._anthropic_client = MagicMock()
    +        agent._anthropic_client.messages.stream.return_value = mock_stream
    +
    +        agent._interruptible_streaming_api_call({})
    +
    +        assert touch_calls.count("receiving stream response") == len(events)

• #59093 fix(cron): guard cron delivery against interpreter shutdown + deliver before agent teardown (#58720)

  merged to main

  files: cron/scheduler.py, tests/cron/test_cron_workdir.py, tests/cron/test_parallel_pool.py, tests/cron/test_run_one_job.py, tests/cron/test_scheduler.py, tests/cron/test_scheduler_shutdown_guard.py

  Show the diff

  • cron/scheduler.py

    @@ -407,6 +407,38 @@ def _shutdown_parallel_pool() -> None:
     atexit.register(_shutdown_parallel_pool)
     
     
    +def _interpreter_shutting_down(exc: Optional[BaseException] = None) -> bool:
    +    """True when the Python interpreter is finalizing.
    +
    +    A cron tick can fire while the gateway is tearing down — SIGTERM from
    +    ``hermes update`` / ``hermes gateway stop`` / systemd restart, or an
    +    OOM-kill. Once finalization starts, ``concurrent.futures`` refuses new
    +    work with ``RuntimeError: cannot schedule new futures after interpreter
    +    shutdown`` and asyncio's default executor is gone, so *any* attempt to
    +    schedule delivery (live-adapter, ``asyncio.run``, or a fresh pool) is
    +    doomed and only pollutes ``errors.log`` with a traceback. Callers use
    +    this to skip gracefully with a warning instead of crashing (#58720,
    +    #55924).
    +
    +    ``exc`` lets a caller also treat an already-raised scheduling error as a
    +    shutdown signal: the ``concurrent.futures`` module-global flag can be set
    +    a hair before ``sys.is_finalizing()`` flips, so matching the error text is
    +    a safe fallback for that race.
    +    """
    +    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
    +
    +
     # Backward-compatible module override used by tests and emergency monkeypatches.
     _hermes_home: Path | None = None
     
    @@ -1758,16 +1790,37 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
                     )
     
             if not delivered:
    +            # If the interpreter is finalizing (gateway SIGTERM / restart /
    +            # OOM), scheduling any new delivery is futile — asyncio.run and a
    +            # fresh ThreadPoolExecutor both raise "cannot schedule new futures
    +            # after interpreter shutdown". Skip gracefully with a warning
    +            # rather than emitting an ERROR traceback on every restart-race
    +            # (#58720, #55924).
    +            if _interpreter_shutting_down():
    +                msg = f"delivery to {platform_name}:{chat_id} skipped — interpreter is shutting down"
    +                logger.warning("Job '%s': %s", job["id"], msg)
    +                target_errors.append(msg)
    +                delivery_errors.extend(target_errors)
    +                continue
                 # Standalone path: run the async send in a fresh event loop (safe from any thread)
                 coro = _send_to_platform(platform, pconfig, chat_id, cleaned_delivery_content, thread_id=thread_id, media_files=media_files)
                 try:
                     result = asyncio.run(coro)
    -            except RuntimeError:
    +            except RuntimeError as run_err:
                     # asyncio.run() checks for a running loop before awaiting the coroutine;
                     # when it raises, the original coro was never started — close it to
                     # prevent "coroutine was never awaited" RuntimeWarning, then retry in a
                     # fresh thread that has no running loop.
                     coro.close()
    +                # If the RuntimeError is the interpreter-finalization signal,
    +                # the fresh-thread fallback would fail identically — skip
    +                # gracefully instead of logging a shutdown-race traceback.
    +                if _interpreter_shutting_down(run_err):
    +                    msg = f"delivery to {platform_name}:{chat_id} skipped — interpreter is shutting down"
    +                    logger.warning("Job '%s': %s", job["id"], msg)
    +                    target_errors.append(msg)
    +                    delivery_errors.extend(target_errors)
    +                    continue
                     # The thread-pool fallback can itself raise (SMTP ConnectionError,
                     # future.result timeout, etc.). An exception raised inside this
                     # `except RuntimeError` block is NOT caught by the sibling
    @@ -1784,6 +1837,14 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
                         finally:
                             pool.shutdown(wait=False)
                     except Exception as e:
    +                    # A shutdown-race here is expected during teardown; downgrade
    +                    # to a warning so it doesn't read as a genuine failure.
    +                    if _interpreter_shutting_down(e):
    +                        msg = f"delivery to {platform_name}:{chat_id} skipped — interpreter is shutting down"
    +                        logger.warning("Job '%s': %s", job["id"], msg)
    +                        target_errors.append(msg)
    +                        delivery_errors.extend(target_errors)
    +                        continue
                         msg = f"delivery to {platform_name}:{chat_id} failed: {e}"
                         logger.error("Job '%s': %s", job["id"], msg, exc_info=True)
                         target_errors.extend([msg])
    @@ -2322,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)
         """
    @@ -3140,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:
    @@ -3202,40 +3295,72 @@ def run_one_job(job: dict, *, adapters=None, loop=None, verbose: bool = False) -
             _scope_token = [redacted:secret-assignment]
                 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.
    @@ -3375,6 +3500,17 @@ def _submit_with_guard(job: dict, pool: concurrent.futures.ThreadPoolExecutor):
                 membership is released in the worker's finally block.
                 """
                 job_id = job["id"]
    +            # A tick can race gateway teardown: once the interpreter is
    +            # finalizing, ``pool.submit`` raises "cannot schedule new futures
    +            # after interpreter shutdown" and crashes the tick. Skip cleanly —
    +            # the job stays due and will fire on the next healthy tick
    +            # (#58720, #55924).
    +            if _interpreter_shutting_down():
    +                logger.warning(
    +                    "Job '%s' not dispatched — interpreter is shutting down",
    +                    job.get("name", job_id),
    +                )
    +                return None
                 with _running_lock:
                     if job_id in _running_job_ids:
                         logger.info("Job '%s' already running — skipping", job.get("name", job_id))
    @@ -3389,7 +3525,20 @@ def _run_and_release(j=job, ctx=_ctx):
                         with _running_lock:
                             _running_job_ids.discard(j["id"])
     
    -            return pool.submit(_run_and_release)
    +            try:
    +                return pool.submit(_run_and_release)
    +            except RuntimeError as submit_err:
    +                # Interpreter began finalizing between the guard above and the
    +                # submit — release the in-flight claim we just took and skip.
    +                if _interpreter_shutting_down(submit_err):
    +                    with _running_lock:
    +                        _running_job_ids.discard(job_id)
    +                    logger.warning(
    +                        "Job '%s' not dispatched — interpreter is shutting down",
    +                        job.get("name", job_id),
    +                    )
    +                    return None
    +                raise
     
             # Sequential pass for env-mutating (workdir) jobs.
             # Queued to a persistent single-thread pool so they run one at a time

  • tests/cron/test_cron_workdir.py

    @@ -220,7 +220,7 @@ def test_workdir_jobs_run_sequentially(self, tmp_path, monkeypatch):
             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 @@ def test_running_set_prevents_double_dispatch(self, tmp_path, monkeypatch):
             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 @@ def test_sync_true_blocks_and_returns_correct_count(self, tmp_path, monkeypatch)
     
             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 @@ def test_sync_false_returns_immediately(self, tmp_path, monkeypatch):
     
             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 @@ def test_sequential_job_does_not_block_ticker(self, tmp_path, monkeypatch):
     
             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 @@ def test_sequential_running_guard_prevents_double_dispatch(self, tmp_path, monke
             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 fake_run_job(job):
         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 @@ def test_claim_runs_before_run_job(self):
             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 @@ def test_parallel_jobs_run_concurrently(self):
             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 @@ def test_parallel_jobs_isolated_contextvars(self):
             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 @@ def test_max_parallel_env_var(self, monkeypatch):
             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)

  • tests/cron/test_scheduler_shutdown_guard.py

    @@ -0,0 +1,137 @@
    +"""Regression coverage for #58720 / #55924 — cron scheduling races
    +interpreter finalization.
    +
    +When the gateway tears down (SIGTERM from ``hermes update`` /
    +``hermes gateway stop`` / systemd restart, or an OOM-kill), a cron tick can
    +still fire. Once the Python interpreter is finalizing, ``concurrent.futures``
    +refuses new work with ``RuntimeError: cannot schedule new futures after
    +interpreter shutdown`` and asyncio's default executor is gone. The cron
    +delivery + dispatch paths used to hit that unguarded, crashing the tick and
    +spraying a traceback into ``errors.log`` on every restart-race.
    +
    +The fix adds ``_interpreter_shutting_down()`` and guards the scheduling
    +sites so they skip gracefully with a warning instead of raising.
    +"""
    +
    +from __future__ import annotations
    +
    +import sys
    +from unittest.mock import AsyncMock, MagicMock, patch
    +
    +import pytest
    +
    +
    +class TestInterpreterShuttingDownHelper:
    +    def test_true_when_finalizing(self):
    +        from cron.scheduler import _interpreter_shutting_down
    +
    +        with patch("sys.is_finalizing", return_value=True):
    +            assert _interpreter_shutting_down() is True
    +
    +    def test_false_when_not_finalizing_and_no_exc(self):
    +        from cron.scheduler import _interpreter_shutting_down
    +
    +        with patch("sys.is_finalizing", return_value=False):
    +            assert _interpreter_shutting_down() is False
    +
    +    def test_matches_shutdown_error_text_as_fallback(self):
    +        """The concurrent.futures module-global flag can be set a hair before
    +        ``sys.is_finalizing()`` flips — matching the error text catches that
    +        race so a shutdown RuntimeError isn't misread as a real failure."""
    +        from cron.scheduler import _interpreter_shutting_down
    +
    +        exc = RuntimeError("cannot schedule new futures after interpreter shutdown")
    +        with patch("sys.is_finalizing", return_value=False):
    +            assert _interpreter_shutting_down(exc) is True
    +
    +    def test_unrelated_error_is_not_shutdown(self):
    +        from cron.scheduler import _interpreter_shutting_down
    +
    +        exc = RuntimeError("some other problem")
    +        with patch("sys.is_finalizing", return_value=False):
    +            assert _interpreter_shutting_down(exc) is False
    +
    +
    +class TestStandaloneDeliverySkipsDuringShutdown:
    +    def _telegram_cfg(self):
    +        from gateway.config import Platform
    +
    +        pconfig = MagicMock()
    +        pconfig.enabled = True
    +        mock_cfg = MagicMock()
    +        mock_cfg.platforms = {Platform.TELEGRAM: pconfig}
    +        return mock_cfg
    +
    +    def test_standalone_path_skips_without_scheduling(self):
    +        """With the interpreter finalizing, the standalone delivery path must
    +        skip BEFORE attempting to schedule the send — no ``_send_to_platform``
    +        call, a graceful warning-level skip, and an error string returned
    +        (not a raised exception)."""
    +        from cron.scheduler import _deliver_result
    +
    +        job = {
    +            "id": "gov-job",
    +            "name": "model-governor",
    +            "deliver": "origin",
    +            "origin": {"platform": "telegram", "chat_id": "123"},
    +        }
    +        send_mock = AsyncMock(return_value={"success": True})
    +        with patch("gateway.config.load_gateway_config", return_value=self._telegram_cfg()), \
    +             patch("tools.send_message_tool._send_to_platform", new=send_mock), \
    +             patch("sys.is_finalizing", return_value=True):
    +            result = _deliver_result(job, "daily report body")
    +
    +        send_mock.assert_not_called()
    +        assert result is not None
    +        assert "shutting down" in result
    +
    +    def test_normal_delivery_still_works_when_not_finalizing(self):
    +        """Guard must not regress the happy path: a normal (non-finalizing)
    +        run still delivers via the standalone send."""
    +        from cron.scheduler import _deliver_result
    +
    +        job = {
    +            "id": "gov-job",
    +            "name": "model-governor",
    +            "deliver": "origin",
    +            "origin": {"platform": "telegram", "chat_id": "123"},
    +        }
    +        send_mock = AsyncMock(return_value={"success": True})
    +        with patch("gateway.config.load_gateway_config", return_value=self._telegram_cfg()), \
    +             patch("tools.send_message_tool._send_to_platform", new=send_mock), \
    +             patch("sys.is_finalizing", return_value=False):
    +            result = _deliver_result(job, "daily report body")
    +
    +        send_mock.assert_called_once()
    +        assert result is None
    +
    +
    +class TestSourceGuardrail:
    +    @pytest.fixture
    +    def source(self) -> str:
    +        from pathlib import Path
    +
    +        return (
    +            Path(__file__).resolve().parents[2] / "cron" / "scheduler.py"
    +        ).read_text(encoding="utf-8")
    +
    +    def test_helper_defined(self, source):
    +        assert "def _interpreter_shutting_down(" in source
    +        assert "#58720" in source
    +
    +    def test_helper_guards_dispatch_submit(self, source):
    +        """The tick dispatch (``_submit_with_guard``) must consult the guard so
    +        a tick that races teardown skips instead of crashing."""
    +        idx_submit = source.find("def _submit_with_guard(")
    +        assert idx_submit >= 0
    +        tail = source[idx_submit:idx_submit + 1600]
    +        assert "_interpreter_shutting_down(" in tail
    +
    +    def test_helper_guards_standalone_delivery(self, source):
    +        """The standalone delivery path must consult the guard before
    +        scheduling ``asyncio.run`` / a fresh pool."""
    +        idx = source.find("Standalone path: run the async send")
    +        assert idx >= 0
    +        # The guard appears shortly before the standalone send comment.
    +        window = source[max(0, idx - 600):idx]
    +        assert "_interpreter_shutting_down()" in window

• #104456 SessionDB teardown no longer overlaps a live handle for the same state.db: counted per-path barrier + cron workers own their teardown (#102827, salvage of #103118 by @JoaoMarcos44)

  merged to main

  files: agent/trace_upload.py, cron/scheduler.py, cron/scheduler_detached_worker.py, gateway/platforms/api_server.py, gateway/run.py, hermes_cli/web_server_lifecycle.py, hermes_cli/web_server_sessions.py, hermes_state_registry.py, tests/cron/test_cleanup_timeout.py, tests/cron/test_scheduler.py, tests/hermes_state/test_shared_session_db_registry.py