fix(gateway): close temporary agents after one-off tasks

Add shared _cleanup_agent_resources() for temporary gateway AIAgent
instances. Apply cleanup to memory flush, background tasks, /btw,
manual /compress, and session-hygiene auto-compression. Prevents
unclosed aiohttp client session leaks.

Cherry-picked from #10899 by @LeonSGP43. Consolidates #10945 by @Lubrsy706.
Fixes #10865.

Co-authored-by: Lubrsy706 <Lubrsy706@users.noreply.github.com>

tests/gateway/test_background_command.py

@@ -220,6 +220,8 @@ class TestRunBackgroundTask:
         with patch("gateway.run._resolve_runtime_agent_kwargs", return_value={"api_key": "test-key"}), \
              patch("run_agent.AIAgent") as MockAgent:
             mock_agent_instance = MagicMock()
+            mock_agent_instance.shutdown_memory_provider = MagicMock()
+            mock_agent_instance.close = MagicMock()
             mock_agent_instance.run_conversation.return_value = mock_result
             MockAgent.return_value = mock_agent_instance
 
@@ -231,6 +233,37 @@ class TestRunBackgroundTask:
         content = call_args[1].get("content", call_args[0][1] if len(call_args[0]) > 1 else "")
         assert "Background task complete" in content
         assert "Hello from background!" in content
+        mock_agent_instance.shutdown_memory_provider.assert_called_once()
+        mock_agent_instance.close.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_agent_cleanup_runs_when_background_agent_raises(self):
+        """Temporary background agents must be cleaned up on error paths too."""
+        runner = _make_runner()
+        mock_adapter = AsyncMock()
+        mock_adapter.send = AsyncMock()
+        runner.adapters[Platform.TELEGRAM] = mock_adapter
+
+        source = SessionSource(
+            platform=Platform.TELEGRAM,
+            user_id="12345",
+            chat_id="67890",
+            user_name="testuser",
+        )
+
+        with patch("gateway.run._resolve_runtime_agent_kwargs", return_value={"api_key": "test-key"}), \
+             patch("run_agent.AIAgent") as MockAgent:
+            mock_agent_instance = MagicMock()
+            mock_agent_instance.shutdown_memory_provider = MagicMock()
+            mock_agent_instance.close = MagicMock()
+            mock_agent_instance.run_conversation.side_effect = RuntimeError("boom")
+            MockAgent.return_value = mock_agent_instance
+
+            await runner._run_background_task("say hello", source, "bg_test")
+
+        mock_adapter.send.assert_called_once()
+        mock_agent_instance.shutdown_memory_provider.assert_called_once()
+        mock_agent_instance.close.assert_called_once()
 
     @pytest.mark.asyncio
     async def test_exception_sends_error_message(self):

tests/gateway/test_compress_command.py

@@ -62,6 +62,8 @@ async def test_compress_command_reports_noop_without_success_banner():
     history = _make_history()
     runner = _make_runner(history)
     agent_instance = MagicMock()
+    agent_instance.shutdown_memory_provider = MagicMock()
+    agent_instance.close = MagicMock()
     agent_instance.context_compressor.protect_first_n = 0
     agent_instance.context_compressor._align_boundary_forward.return_value = 0
     agent_instance.context_compressor._find_tail_cut_by_tokens.return_value = 2
@@ -83,6 +85,8 @@ async def test_compress_command_reports_noop_without_success_banner():
     assert "No changes from compression" in result
     assert "Compressed:" not in result
     assert "Rough transcript estimate: ~100 tokens (unchanged)" in result
+    agent_instance.shutdown_memory_provider.assert_called_once()
+    agent_instance.close.assert_called_once()
 
 
 @pytest.mark.asyncio
@@ -95,6 +99,8 @@ async def test_compress_command_explains_when_token_estimate_rises():
     ]
     runner = _make_runner(history)
     agent_instance = MagicMock()
+    agent_instance.shutdown_memory_provider = MagicMock()
+    agent_instance.close = MagicMock()
     agent_instance.context_compressor.protect_first_n = 0
     agent_instance.context_compressor._align_boundary_forward.return_value = 0
     agent_instance.context_compressor._find_tail_cut_by_tokens.return_value = 2
@@ -119,3 +125,5 @@ async def test_compress_command_explains_when_token_estimate_rises():
     assert "Compressed: 4 → 3 messages" in result
     assert "Rough transcript estimate: ~100 → ~120 tokens" in result
     assert "denser summaries" in result
+    agent_instance.shutdown_memory_provider.assert_called_once()
+    agent_instance.close.assert_called_once()

tests/gateway/test_flush_memory_stale_guard.py

@@ -202,6 +202,22 @@ class TestFlushAgentSilenced:
             sys.stdout = old_stdout
         assert buf.getvalue() == "", "no-op print_fn spinner must not write to stdout"
 
+    def test_flush_agent_closes_resources_after_run(self, monkeypatch):
+        """Memory flush should close temporary agent resources after the turn."""
+        runner, tmp_agent, _ = _make_flush_context(monkeypatch)
+        tmp_agent.shutdown_memory_provider = MagicMock()
+        tmp_agent.close = MagicMock()
+
+        with (
+            patch("gateway.run._resolve_runtime_agent_kwargs", return_value={"api_key": "k"}),
+            patch("gateway.run._resolve_gateway_model", return_value="test-model"),
+            patch.dict("sys.modules", {"tools.memory_tool": MagicMock(get_memory_dir=lambda: Path("/nonexistent"))}),
+        ):
+            runner._flush_memories_for_session("session_cleanup")
+
+        tmp_agent.shutdown_memory_provider.assert_called_once()
+        tmp_agent.close.assert_called_once()
+
 
 class TestFlushPromptStructure:
     """Verify the flush prompt retains its core instructions."""

tests/gateway/test_session_hygiene.py

@@ -305,10 +305,15 @@ async def test_session_hygiene_messages_stay_in_originating_topic(monkeypatch, t
     monkeypatch.setitem(sys.modules, "dotenv", fake_dotenv)
 
     class FakeCompressAgent:
+        last_instance = None
+
         def __init__(self, **kwargs):
             self.model = kwargs.get("model")
             self.session_id = kwargs.get("session_id", "fake-session")
             self._print_fn = None
+            self.shutdown_memory_provider = MagicMock()
+            self.close = MagicMock()
+            type(self).last_instance = self
 
         def _compress_context(self, messages, *_args, **_kwargs):
             # Simulate real _compress_context: create a new session_id
@@ -385,3 +390,6 @@ async def test_session_hygiene_messages_stay_in_originating_topic(monkeypatch, t
     # Compression warnings are no longer sent to users — compression
     # happens silently with server-side logging only.
     assert len(adapter.sent) == 0
+    assert FakeCompressAgent.last_instance is not None
+    FakeCompressAgent.last_instance.shutdown_memory_provider.assert_called_once()
+    FakeCompressAgent.last_instance.close.assert_called_once()