diff --git a/api/extensions/storage/opendal_storage.py b/api/extensions/storage/opendal_storage.py
index 83c5c2d12f..96f5915ff0 100644
--- a/api/extensions/storage/opendal_storage.py
+++ b/api/extensions/storage/opendal_storage.py
@@ -32,7 +32,7 @@ class OpenDALStorage(BaseStorage):
         kwargs = kwargs or _get_opendal_kwargs(scheme=scheme)
 
         if scheme == "fs":
-            root = kwargs.get("root", "storage")
+            root = kwargs.setdefault("root", "storage")
             Path(root).mkdir(parents=True, exist_ok=True)
 
         retry_layer = opendal.layers.RetryLayer(max_times=3, factor=2.0, jitter=True)
diff --git a/api/tests/test_containers_integration_tests/test_opendal_fs_default_root.py b/api/tests/test_containers_integration_tests/test_opendal_fs_default_root.py
new file mode 100644
index 0000000000..34a1941c39
--- /dev/null
+++ b/api/tests/test_containers_integration_tests/test_opendal_fs_default_root.py
@@ -0,0 +1,56 @@
+from pathlib import Path
+
+from extensions.storage.opendal_storage import OpenDALStorage
+
+
+class TestOpenDALFsDefaultRoot:
+    """Test that OpenDALStorage with scheme='fs' works correctly when no root is provided."""
+
+    def test_fs_without_root_uses_default(self, tmp_path, monkeypatch):
+        """When no root is specified, the default 'storage' should be used and passed to the Operator."""
+        # Change to tmp_path so the default "storage" dir is created there
+        monkeypatch.chdir(tmp_path)
+        # Ensure no OPENDAL_FS_ROOT env var is set
+        monkeypatch.delenv("OPENDAL_FS_ROOT", raising=False)
+
+        storage = OpenDALStorage(scheme="fs")
+
+        # The default directory should have been created
+        assert (tmp_path / "storage").is_dir()
+        # The storage should be functional
+        storage.save("test_default_root.txt", b"hello")
+        assert storage.exists("test_default_root.txt")
+        assert storage.load_once("test_default_root.txt") == b"hello"
+
+        # Cleanup
+        storage.delete("test_default_root.txt")
+
+    def test_fs_with_explicit_root(self, tmp_path):
+        """When root is explicitly provided, it should be used."""
+        custom_root = str(tmp_path / "custom_storage")
+        storage = OpenDALStorage(scheme="fs", root=custom_root)
+
+        assert Path(custom_root).is_dir()
+        storage.save("test_explicit_root.txt", b"world")
+        assert storage.exists("test_explicit_root.txt")
+        assert storage.load_once("test_explicit_root.txt") == b"world"
+
+        # Cleanup
+        storage.delete("test_explicit_root.txt")
+
+    def test_fs_with_env_var_root(self, tmp_path, monkeypatch):
+        """When OPENDAL_FS_ROOT env var is set, it should be picked up via _get_opendal_kwargs."""
+        env_root = str(tmp_path / "env_storage")
+        monkeypatch.setenv("OPENDAL_FS_ROOT", env_root)
+        # Ensure .env file doesn't interfere
+        monkeypatch.chdir(tmp_path)
+
+        storage = OpenDALStorage(scheme="fs")
+
+        assert Path(env_root).is_dir()
+        storage.save("test_env_root.txt", b"env_data")
+        assert storage.exists("test_env_root.txt")
+        assert storage.load_once("test_env_root.txt") == b"env_data"
+
+        # Cleanup
+        storage.delete("test_env_root.txt")