diff --git a/cron/scheduler.py b/cron/scheduler.py index 83b7abb9b1463..6065568be37c4 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -601,6 +601,7 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]: logger.info("Running job '%s' (ID: %s)", job_name, job_id) logger.info("Prompt: %s", prompt[:100]) + agent = None try: # Inject origin context so the agent's send_message tool knows the chat. # Must be INSIDE the try block so the finally cleanup always runs. @@ -894,6 +895,11 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]: _session_db.close() except (Exception, KeyboardInterrupt) as e: logger.debug("Job '%s': failed to close SQLite session store: %s", job_id, e) + 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) def tick(verbose: bool = True, adapters=None, loop=None) -> int: diff --git a/gateway/run.py b/gateway/run.py index c8c25256b8f2b..45b50419dd9ac 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -793,10 +793,16 @@ def _flush_memories_for_session( "tools if needed, then stop.]" ) - tmp_agent.run_conversation( - user_message=flush_prompt, - conversation_history=msgs, - ) + try: + tmp_agent.run_conversation( + user_message=flush_prompt, + conversation_history=msgs, + ) + finally: + try: + tmp_agent.close() + except Exception: + pass logger.info("Pre-reset memory flush completed for session %s", old_session_id) except Exception as e: logger.debug("Pre-reset memory flush failed for session %s: %s", old_session_id, e) @@ -3448,13 +3454,19 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str): _hyg_agent._print_fn = lambda *a, **kw: None loop = asyncio.get_event_loop() - _compressed, _ = await loop.run_in_executor( - None, - lambda: _hyg_agent._compress_context( - _hyg_msgs, "", - approx_tokens=_approx_tokens, - ), - ) + try: + _compressed, _ = await loop.run_in_executor( + None, + lambda: _hyg_agent._compress_context( + _hyg_msgs, "", + approx_tokens=_approx_tokens, + ), + ) + finally: + try: + _hyg_agent.close() + except Exception: + pass # _compress_context ends the old session and creates # a new session_id. Write compressed messages into @@ -5384,10 +5396,13 @@ def run_sync(): fallback_model=self._fallback_model, ) - return agent.run_conversation( - user_message=prompt, - task_id=task_id, - ) + try: + return agent.run_conversation( + user_message=prompt, + task_id=task_id, + ) + finally: + agent.close() loop = asyncio.get_event_loop() result = await loop.run_in_executor(None, run_sync) @@ -5566,11 +5581,14 @@ def run_sync(): skip_context_files=True, persist_session=False, ) - return agent.run_conversation( - user_message=btw_prompt, - conversation_history=history_snapshot, - task_id=task_id, - ) + try: + return agent.run_conversation( + user_message=btw_prompt, + conversation_history=history_snapshot, + task_id=task_id, + ) + finally: + agent.close() loop = asyncio.get_event_loop() result = await loop.run_in_executor(None, run_sync) @@ -5901,18 +5919,24 @@ async def _handle_compress_command(self, event: MessageEvent) -> str: ) tmp_agent._print_fn = lambda *a, **kw: None - compressor = tmp_agent.context_compressor - compress_start = compressor.protect_first_n - compress_start = compressor._align_boundary_forward(msgs, compress_start) - compress_end = compressor._find_tail_cut_by_tokens(msgs, compress_start) - if compress_start >= compress_end: - return "Nothing to compress yet (the transcript is still all protected context)." - - loop = asyncio.get_event_loop() - compressed, _ = await loop.run_in_executor( - None, - lambda: tmp_agent._compress_context(msgs, "", approx_tokens=approx_tokens, focus_topic=focus_topic) - ) + try: + compressor = tmp_agent.context_compressor + compress_start = compressor.protect_first_n + compress_start = compressor._align_boundary_forward(msgs, compress_start) + compress_end = compressor._find_tail_cut_by_tokens(msgs, compress_start) + if compress_start >= compress_end: + return "Nothing to compress yet (the transcript is still all protected context)." + + loop = asyncio.get_event_loop() + compressed, _ = await loop.run_in_executor( + None, + lambda: tmp_agent._compress_context(msgs, "", approx_tokens=approx_tokens, focus_topic=focus_topic) + ) + finally: + try: + tmp_agent.close() + except Exception: + pass # _compress_context already calls end_session() on the old session # (preserving its full transcript in SQLite) and creates a new diff --git a/tests/cron/test_scheduler.py b/tests/cron/test_scheduler.py index 08b57cfa897e2..1ff6ca5b38acb 100644 --- a/tests/cron/test_scheduler.py +++ b/tests/cron/test_scheduler.py @@ -669,6 +669,7 @@ def test_run_job_passes_session_db_and_cron_platform(self, tmp_path): assert call_args[0][0].startswith("cron_test-job_") assert call_args[0][1] == "cron_complete" fake_db.close.assert_called_once() + mock_agent.close.assert_called_once() def test_run_job_empty_response_returns_empty_not_placeholder(self, tmp_path): """Empty final_response should stay empty for delivery logic (issue #2234). @@ -710,6 +711,7 @@ def test_run_job_empty_response_returns_empty_not_placeholder(self, tmp_path): assert final_response == "" # But the output log should show the placeholder assert "(No response generated)" in output + mock_agent.close.assert_called_once() def test_run_job_sets_auto_delivery_env_from_dotenv_home_channel(self, tmp_path, monkeypatch): job = { @@ -765,6 +767,39 @@ def run_conversation(self, *args, **kwargs): assert os.getenv("HERMES_CRON_AUTO_DELIVER_THREAD_ID") is None fake_db.close.assert_called_once() + def test_run_job_closes_agent_on_failure(self, tmp_path): + job = { + "id": "failing-job", + "name": "failing", + "prompt": "hello", + } + fake_db = MagicMock() + + with patch("cron.scheduler._hermes_home", tmp_path), \ + patch("cron.scheduler._resolve_origin", return_value=None), \ + patch("dotenv.load_dotenv"), \ + patch("hermes_state.SessionDB", return_value=fake_db), \ + patch( + "hermes_cli.runtime_provider.resolve_runtime_provider", + return_value={ + "api_key": "***", + "base_url": "https://example.invalid/v1", + "provider": "openrouter", + "api_mode": "chat_completions", + }, + ), \ + patch("run_agent.AIAgent") as mock_agent_cls: + mock_agent = MagicMock() + mock_agent.run_conversation.side_effect = RuntimeError("boom") + mock_agent_cls.return_value = mock_agent + + success, output, final_response, error = run_job(job) + + assert success is False + assert final_response == "" + assert "RuntimeError: boom" in error + mock_agent.close.assert_called_once() + class TestRunJobConfigLogging: """Verify that config.yaml parse failures are logged, not silently swallowed.""" diff --git a/tests/gateway/test_compress_focus.py b/tests/gateway/test_compress_focus.py index 8a1ee060f5835..202d7e87eb01c 100644 --- a/tests/gateway/test_compress_focus.py +++ b/tests/gateway/test_compress_focus.py @@ -83,6 +83,7 @@ def _estimate(messages): # Verify focus_topic was passed agent_instance._compress_context.assert_called_once() + agent_instance.close.assert_called_once() call_kwargs = agent_instance._compress_context.call_args assert call_kwargs.kwargs.get("focus_topic") == "database schema" @@ -111,6 +112,7 @@ async def test_compress_no_focus_passes_none(): result = await runner._handle_compress_command(_make_event("/compress")) agent_instance._compress_context.assert_called_once() + agent_instance.close.assert_called_once() call_kwargs = agent_instance._compress_context.call_args assert call_kwargs.kwargs.get("focus_topic") is None diff --git a/tests/gateway/test_flush_memory_stale_guard.py b/tests/gateway/test_flush_memory_stale_guard.py index 6a43817cee02a..9e8c46e87314f 100644 --- a/tests/gateway/test_flush_memory_stale_guard.py +++ b/tests/gateway/test_flush_memory_stale_guard.py @@ -100,6 +100,7 @@ def test_memory_content_injected_into_flush_prompt(self, tmp_path, monkeypatch): runner._flush_memories_for_session("session_123") tmp_agent.run_conversation.assert_called_once() + tmp_agent.close.assert_called_once() flush_prompt = tmp_agent.run_conversation.call_args.kwargs.get("user_message", "") assert "Agent knows Python" in flush_prompt @@ -124,6 +125,7 @@ def test_flush_works_without_memory_files(self, tmp_path, monkeypatch): runner._flush_memories_for_session("session_456") tmp_agent.run_conversation.assert_called_once() + tmp_agent.close.assert_called_once() flush_prompt = tmp_agent.run_conversation.call_args.kwargs.get("user_message", "") assert "Do NOT overwrite or remove entries" not in flush_prompt assert "Review the conversation above" in flush_prompt @@ -145,6 +147,7 @@ def test_empty_memory_files_no_injection(self, tmp_path, monkeypatch): runner._flush_memories_for_session("session_789") tmp_agent.run_conversation.assert_called_once() + tmp_agent.close.assert_called_once() flush_prompt = tmp_agent.run_conversation.call_args.kwargs.get("user_message", "") assert "current live state of memory" not in flush_prompt