Ensure working directories exist before file operations

Fix bug where files could be saved to non-existent directories. The
`get_working_directory()` function now guarantees the returned directory
exists by creating it if necessary. All toolkit __init__ methods that
resolve working directories from environment variables also ensure the
directory exists before passing it to the base class.

Fixes: Saves files to imaginary directories

Co-authored-by: lightaime <23632352+lightaime@users.noreply.github.com>

backend/app/agent/toolkit/audio_analysis_toolkit.py

@@ -39,6 +39,7 @@ class AudioAnalysisToolkit(BaseAudioAnalysisToolkit, AbstractToolkit):
             cache_dir = env(
                 "file_save_path", os.path.expanduser("~/.eigent/tmp/")
             )
+        os.makedirs(cache_dir, exist_ok=True)
         super().__init__(
             cache_dir, transcribe_model, audio_reasoning_model, timeout
         )

backend/app/agent/toolkit/excel_toolkit.py

@@ -37,4 +37,5 @@ class ExcelToolkit(BaseExcelToolkit, AbstractToolkit):
             working_directory = env(
                 "file_save_path", os.path.expanduser("~/Downloads")
             )
+        os.makedirs(working_directory, exist_ok=True)
         super().__init__(timeout=timeout, working_directory=working_directory)

backend/app/agent/toolkit/file_write_toolkit.py

@@ -47,6 +47,7 @@ class FileToolkit(BaseFileToolkit, AbstractToolkit):
             working_directory = env(
                 "file_save_path", os.path.expanduser("~/Downloads")
             )
+        os.makedirs(working_directory, exist_ok=True)
         super().__init__(
             working_directory, timeout, default_encoding, backup_enabled
         )

backend/app/agent/toolkit/pptx_toolkit.py

@@ -46,6 +46,7 @@ class PPTXToolkit(BasePPTXToolkit, AbstractToolkit):
             working_directory = env(
                 "file_save_path", os.path.expanduser("~/Downloads")
             )
+        os.makedirs(working_directory, exist_ok=True)
         super().__init__(working_directory, timeout)
 
     @listen_toolkit(

backend/app/agent/toolkit/pyautogui_toolkit.py

@@ -36,5 +36,6 @@ class PyAutoGUIToolkit(BasePyAutoGUIToolkit, AbstractToolkit):
             screenshots_dir = env(
                 "file_save_path", os.path.expanduser("~/Downloads")
             )
+        os.makedirs(screenshots_dir, exist_ok=True)
         super().__init__(timeout, screenshots_dir)
         self.api_task_id = api_task_id

backend/app/agent/toolkit/screenshot_toolkit.py

@@ -37,4 +37,5 @@ class ScreenshotToolkit(BaseScreenshotToolkit, AbstractToolkit):
             working_directory = env(
                 "file_save_path", os.path.expanduser("~/Downloads")
             )
+        os.makedirs(working_directory, exist_ok=True)
         super().__init__(working_directory, timeout)

backend/app/agent/toolkit/terminal_toolkit.py

@@ -81,6 +81,7 @@ class TerminalToolkit(BaseTerminalToolkit, AbstractToolkit):
         base_dir = env(
             "file_save_path", os.path.expanduser("~/.eigent/terminal/")
         )
+        os.makedirs(base_dir, exist_ok=True)
 
         if working_directory is None:
             working_directory = base_dir

backend/app/agent/toolkit/video_analysis_toolkit.py

@@ -44,6 +44,7 @@ class VideoAnalysisToolkit(BaseVideoAnalysisToolkit, AbstractToolkit):
             working_directory = env(
                 "file_save_path", os.path.expanduser("~/Downloads")
             )
+        os.makedirs(working_directory, exist_ok=True)
         super().__init__(
             working_directory,
             model,

backend/app/agent/toolkit/video_download_toolkit.py

@@ -37,5 +37,6 @@ class VideoDownloaderToolkit(BaseVideoDownloaderToolkit, AbstractToolkit):
             working_directory = env(
                 "file_save_path", os.path.expanduser("~/Downloads")
             )
+        os.makedirs(working_directory, exist_ok=True)
         super().__init__(working_directory, cookies_path, timeout)
         self.api_task_id = api_task_id

backend/app/utils/file_utils.py

@@ -13,15 +13,37 @@
 # ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
 """File system utilities."""
 
+import logging
+from pathlib import Path
+
 from app.component.environment import env
 from app.model.chat import Chat
 
+logger = logging.getLogger("file_utils")
+
+
+def _ensure_directory_exists(directory: str) -> str:
+    """
+    Ensure the given directory exists, creating it if necessary.
+
+    Args:
+        directory: Path string for the directory.
+
+    Returns:
+        The same directory path string.
+    """
+    path = Path(directory)
+    path.mkdir(parents=True, exist_ok=True)
+    return directory
+
 
 def get_working_directory(options: Chat, task_lock=None) -> str:
     """
     Get the correct working directory for file operations.
     First checks if there's an updated path from improve API call,
     then falls back to environment variable or default path.
+
+    The returned directory is guaranteed to exist on the filesystem.
     """
     if not task_lock:
         from app.service.task import get_task_lock_if_exists
@@ -33,6 +55,8 @@ def get_working_directory(options: Chat, task_lock=None) -> str:
         and hasattr(task_lock, "new_folder_path")
         and task_lock.new_folder_path
     ):
-        return str(task_lock.new_folder_path)
+        directory = str(task_lock.new_folder_path)
     else:
-        return env("file_save_path", options.file_save_path())
+        directory = env("file_save_path", options.file_save_path())
+
+    return _ensure_directory_exists(directory)

backend/tests/unit/utils/test_file_utils.py

@@ -0,0 +1,125 @@
+# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
+
+import os
+import tempfile
+from unittest.mock import MagicMock, patch
+
+from app.utils.file_utils import (
+    _ensure_directory_exists,
+    get_working_directory,
+)
+
+
+class TestEnsureDirectoryExists:
+    """Test _ensure_directory_exists helper function."""
+
+    def test_creates_nonexistent_directory(self):
+        """Test that a non-existent directory is created."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            new_dir = os.path.join(temp_dir, "subdir", "nested")
+            assert not os.path.exists(new_dir)
+
+            result = _ensure_directory_exists(new_dir)
+
+            assert os.path.isdir(new_dir)
+            assert result == new_dir
+
+    def test_existing_directory_no_error(self):
+        """Test that an existing directory does not raise an error."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            result = _ensure_directory_exists(temp_dir)
+
+            assert os.path.isdir(temp_dir)
+            assert result == temp_dir
+
+    def test_returns_same_path(self):
+        """Test that the function returns the same path it was given."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            path = os.path.join(temp_dir, "test_dir")
+            result = _ensure_directory_exists(path)
+            assert result == path
+
+
+class TestGetWorkingDirectory:
+    """Test get_working_directory function."""
+
+    def test_returns_task_lock_new_folder_path(self):
+        """Test that task_lock.new_folder_path is returned when available."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            new_folder = os.path.join(temp_dir, "new_folder")
+            task_lock = MagicMock()
+            task_lock.new_folder_path = new_folder
+
+            options = MagicMock()
+
+            result = get_working_directory(options, task_lock=task_lock)
+
+            assert result == new_folder
+            # Verify directory was created
+            assert os.path.isdir(new_folder)
+
+    def test_creates_directory_from_env(self, temp_dir):
+        """Test that the directory from env is created if it doesn't exist."""
+        new_dir = str(temp_dir / "env_dir" / "nested")
+
+        options = MagicMock()
+        task_lock = MagicMock(spec=[])  # No new_folder_path attribute
+
+        with patch(
+            "app.utils.file_utils.env", return_value=new_dir
+        ), patch(
+            "app.utils.file_utils.get_working_directory.__module__",
+            "app.utils.file_utils",
+        ):
+            # Patch get_task_lock_if_exists to avoid import issues
+            with patch(
+                "app.service.task.get_task_lock_if_exists",
+                return_value=task_lock,
+            ):
+                result = get_working_directory(options, task_lock=task_lock)
+
+        assert result == new_dir
+        assert os.path.isdir(new_dir)
+
+    def test_creates_directory_from_file_save_path(self, temp_dir):
+        """Test that file_save_path directory is created."""
+        new_dir = str(temp_dir / "save_path")
+
+        options = MagicMock()
+        options.file_save_path.return_value = new_dir
+
+        task_lock = MagicMock(spec=[])  # No new_folder_path attribute
+
+        with patch(
+            "app.utils.file_utils.env",
+            side_effect=lambda key, default: default,
+        ):
+            result = get_working_directory(options, task_lock=task_lock)
+
+        assert result == new_dir
+        assert os.path.isdir(new_dir)
+
+    def test_existing_directory_works(self, temp_dir):
+        """Test that an existing directory is returned without error."""
+        existing_dir = str(temp_dir)
+
+        options = MagicMock()
+        task_lock = MagicMock()
+        task_lock.new_folder_path = existing_dir
+
+        result = get_working_directory(options, task_lock=task_lock)
+
+        assert result == existing_dir
+        assert os.path.isdir(existing_dir)