Skip to content

fix: close ephemeral agents to prevent fd leaks #12998

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
bloodcarter wants to merge 1 commit into from
Closed

fix: close ephemeral agents to prevent fd leaks #12998

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions cron/scheduler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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:
Expand Down
88 changes: 56 additions & 32 deletions gateway/run.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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
Expand Down
35 changes: 35 additions & 0 deletions tests/cron/test_scheduler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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).
Expand Down Expand Up @@ -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 = {
Expand Down Expand Up @@ -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."""
Expand Down
2 changes: 2 additions & 0 deletions tests/gateway/test_compress_focus.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"

Expand Down Expand Up @@ -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

Expand Down
3 changes: 3 additions & 0 deletions tests/gateway/test_flush_memory_stale_guard.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand All @@ -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

Expand Down