From 04cc8f855fc19add1bcc73a094b49b57250babbe Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Sun, 6 Sep 2026 01:03:53 -0700
Subject: [PATCH 1/4] fix(proxy): resolve config include directives for
 bucket-hosted configs

A config loaded from a GCS or S3 bucket skipped include processing entirely,
so every model, guardrail, and setting behind an `include` was silently
dropped. Both bucket types shared the same branch in `get_config`, which
never called `_process_includes`, and that helper only ever read from disk.

The merge now lives in one async helper that takes the loader as a
dependency, so disk and bucket configs share the same semantics: list values
extend, everything else overrides, nested includes are followed, and the
`include` key is stripped. Bucket entries resolve as object keys relative to
the config object's prefix, with a leading `/` meaning the bucket root, and
an include that cannot be read now raises instead of being skipped.
---
 litellm/proxy/common_utils/config_includes.py |  68 +++++++
 .../proxy/common_utils/load_config_utils.py   |  66 ++++++-
 litellm/proxy/proxy_server.py                 |  49 ++---
 .../common_utils/test_load_config_utils.py    | 180 +++++++++++++++++-
 .../proxy/proxy_server/test_proxy_config.py   |  44 ++++-
 5 files changed, 370 insertions(+), 37 deletions(-)
 create mode 100644 litellm/proxy/common_utils/config_includes.py

diff --git a/litellm/proxy/common_utils/config_includes.py b/litellm/proxy/common_utils/config_includes.py
new file mode 100644
index 00000000000..d6939db2e95
--- /dev/null
+++ b/litellm/proxy/common_utils/config_includes.py
@@ -0,0 +1,68 @@
+from collections.abc import Awaitable, Mapping
+from types import MappingProxyType
+from typing import Final, Protocol
+
+INCLUDE_KEY: Final = "include"
+
+
+class ConfigLoader(Protocol):
+    def __call__(self, include_entry: str, /) -> Awaitable[Mapping[str, object]]: ...
+
+
+def _merged_value(base_value: object, included_value: object) -> object:
+    if isinstance(included_value, list) and isinstance(base_value, list):
+        return [*base_value, *included_value]  # mutable-ok: a merged config value stays the plain list the proxy loads
+    return included_value
+
+
+def _merged_entry(base: Mapping[str, object], included: Mapping[str, object], key: str) -> object:
+    if key not in included:
+        return base[key]
+    return _merged_value(base.get(key), included[key])
+
+
+def _merged(base: Mapping[str, object], included: Mapping[str, object]) -> Mapping[str, object]:
+    return MappingProxyType({key: _merged_entry(base, included, key) for key in (*base, *included)})
+
+
+def _without_include(config: Mapping[str, object]) -> Mapping[str, object]:
+    return MappingProxyType({key: value for key, value in config.items() if key != INCLUDE_KEY})
+
+
+def include_entries(config: Mapping[str, object]) -> tuple[str, ...]:
+    if INCLUDE_KEY not in config:
+        return ()
+
+    entries: Final = config[INCLUDE_KEY]
+    if not isinstance(entries, list):
+        raise ValueError("'include' must be a list of file paths")
+
+    paths: Final = tuple(entry for entry in entries if isinstance(entry, str))
+    if len(paths) != len(entries):
+        raise ValueError("'include' must be a list of file paths")
+
+    return paths
+
+
+async def _resolve(config: Mapping[str, object], pending: tuple[str, ...], load: ConfigLoader) -> Mapping[str, object]:
+    if not pending:
+        return _without_include(config)
+
+    included: Final = await load(pending[0])
+    return await _resolve(
+        _merged(config, _without_include(included)),
+        (*pending[1:], *include_entries(included)),
+        load,
+    )
+
+
+async def resolve_includes(config: Mapping[str, object], load: ConfigLoader) -> dict[str, object]:
+    """
+    Merge every config named by the `include` directive into the config that declares it.
+
+    List values are extended and every other value is overridden, an included config may declare
+    further includes, and `load` decides where an entry is read from, so the same merge applies to
+    configs on disk and to configs hosted in a bucket.
+    """
+    merged: Final = await _resolve(config, include_entries(config), load)
+    return dict(merged)  # mutable-ok: the proxy mutates the config it loads
diff --git a/litellm/proxy/common_utils/load_config_utils.py b/litellm/proxy/common_utils/load_config_utils.py
index 62649ad6ca1..927deb68826 100644
--- a/litellm/proxy/common_utils/load_config_utils.py
+++ b/litellm/proxy/common_utils/load_config_utils.py
@@ -1,9 +1,19 @@
 import os
-from typing import Final
+import posixpath
+from collections.abc import Awaitable, Mapping
+from typing import Final, Protocol
 
 import yaml
+from pydantic import TypeAdapter, ValidationError
 
 from litellm._logging import verbose_proxy_logger
+from litellm.proxy.common_utils.config_includes import resolve_includes
+
+_BUCKET_CONFIG_ADAPTER: Final = TypeAdapter(dict[str, object])
+
+
+class BucketObjectFetcher(Protocol):
+    def __call__(self, object_key: str, /) -> Awaitable[Mapping[str, object] | None]: ...
 
 
 def get_file_contents_from_s3(bucket_name, object_key):
@@ -62,6 +72,60 @@ async def get_config_file_contents_from_gcs(bucket_name, object_key):
         return None
 
 
+def resolve_include_object_key(config_object_key: str, include_entry: str) -> str:
+    """
+    Resolve one `include` entry to the object key it names, relative to the config object's prefix.
+
+    A leading "/" means the bucket root, mirroring how an absolute path on disk ignores the
+    directory the including config sits in.
+    """
+    if include_entry.startswith("/"):
+        return posixpath.normpath(include_entry).lstrip("/")
+    return posixpath.normpath(posixpath.join(posixpath.dirname(config_object_key), include_entry))
+
+
+async def resolve_bucket_includes(
+    *,
+    config: Mapping[str, object],
+    object_key: str,
+    fetch: BucketObjectFetcher,
+) -> dict[str, object]:
+    async def load(include_entry: str) -> Mapping[str, object]:
+        include_key: Final = resolve_include_object_key(object_key, include_entry)
+        included: Final = await fetch(include_key)
+        if included is None:
+            raise FileNotFoundError(f"Included config could not be read from bucket: {include_key}")
+        return included
+
+    return await resolve_includes(config=config, load=load)
+
+
+async def get_config_from_bucket(
+    *,
+    bucket_type: str | None,
+    bucket_name: str,
+    object_key: str,
+) -> dict[str, object] | None:
+    async def fetch(key: str) -> Mapping[str, object] | None:
+        raw: Final = (
+            await get_config_file_contents_from_gcs(bucket_name=bucket_name, object_key=key)
+            if bucket_type == "gcs"
+            else get_file_contents_from_s3(bucket_name=bucket_name, object_key=key)
+        )
+        if raw is None:
+            return None
+        try:
+            return _BUCKET_CONFIG_ADAPTER.validate_python(raw)
+        except ValidationError as e:
+            raise ValueError(f"Config object in bucket is not a YAML mapping: {key}") from e
+
+    config: Final = await fetch(object_key)
+    if config is None:
+        return None
+
+    return await resolve_bucket_includes(config=config, object_key=object_key, fetch=fetch)
+
+
 def download_python_file_from_s3(
     bucket_name: str,
     object_key: str,
diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py
index ba5714fe950..a49df166dca 100644
--- a/litellm/proxy/proxy_server.py
+++ b/litellm/proxy/proxy_server.py
@@ -341,6 +341,7 @@ from litellm.proxy.common_utils.auth_cache_invalidation_pubsub import (
     AuthCacheInvalidationSubscriber,
 )
 from litellm.proxy.common_utils.callback_utils import initialize_callbacks_on_proxy
+from litellm.proxy.common_utils.config_includes import resolve_includes
 from litellm.proxy.common_utils.config_sync_pubsub import ConfigSyncSubscriber
 from litellm.proxy.common_utils.debug_utils import init_verbose_loggers
 from litellm.proxy.common_utils.debug_utils import router as debugging_endpoints_router
@@ -359,10 +360,7 @@ from litellm.proxy.common_utils.http_parsing_utils import (
     check_file_size_under_limit,
     get_form_data,
 )
-from litellm.proxy.common_utils.load_config_utils import (
-    get_config_file_contents_from_gcs,
-    get_file_contents_from_s3,
-)
+from litellm.proxy.common_utils.load_config_utils import get_config_from_bucket
 from litellm.proxy.common_utils.model_deprecation import collect_model_deprecations
 from litellm.proxy.common_utils.model_listing_utils import (
     TeamModelNameTranslator,
@@ -4538,12 +4536,12 @@ class ProxyConfig:
         if config is None:
             raise Exception("Config cannot be None or Empty.")
         # Process includes
-        config = self._process_includes(config=config, base_dir=os.path.dirname(os.path.abspath(file_path or "")))
+        config = await self._process_includes(config=config, base_dir=os.path.dirname(os.path.abspath(file_path or "")))
 
         # verbose_proxy_logger.debug(f"loaded config={json.dumps(config, indent=4)}")
         return config
 
-    def _process_includes(self, config: dict, base_dir: str) -> dict:
+    async def _process_includes(self, config: dict, base_dir: str) -> dict:
         """
         Process includes by appending their contents to the main config
 
@@ -4558,29 +4556,14 @@ class ProxyConfig:
             callbacks: ["prometheus"]
         ```
         """
-        if "include" not in config:
-            return config
 
-        if not isinstance(config["include"], list):
-            raise ValueError("'include' must be a list of file paths")
-
-        # Load and append all included files
-        for include_file in config["include"]:
-            file_path = os.path.join(base_dir, include_file)
+        async def load_included(include_file: str) -> Mapping[str, object]:
+            file_path: Final = os.path.join(base_dir, include_file)
             if not os.path.exists(file_path):
                 raise FileNotFoundError(f"Included file not found: {file_path}")
+            return self._load_yaml_file(file_path)
 
-            included_config = self._load_yaml_file(file_path)
-            # Simply update/extend the main config with included config
-            for key, value in included_config.items():
-                if isinstance(value, list) and key in config:
-                    config[key].extend(value)
-                else:
-                    config[key] = value
-
-        # Remove the include directive
-        del config["include"]
-        return config
+        return await resolve_includes(config=config, load=load_included)
 
     async def save_config(self, new_config: dict, include_env_vars: bool = False):
         global prisma_client, general_settings, user_config_file_path, store_model_in_db
@@ -4936,15 +4919,19 @@ class ProxyConfig:
         global prisma_client, store_model_in_db
         # Load existing config
 
-        if os.environ.get("LITELLM_CONFIG_BUCKET_NAME") is not None:
-            bucket_name: Final = os.environ.get("LITELLM_CONFIG_BUCKET_NAME")
+        bucket_name: Final = os.environ.get("LITELLM_CONFIG_BUCKET_NAME")
+        if bucket_name is not None:
             object_key: Final = os.environ.get("LITELLM_CONFIG_BUCKET_OBJECT_KEY")
             bucket_type: Final = os.environ.get("LITELLM_CONFIG_BUCKET_TYPE")
             verbose_proxy_logger.debug("bucket_name: %s, object_key: %s", bucket_name, object_key)
-            if bucket_type == "gcs":
-                config = await get_config_file_contents_from_gcs(bucket_name=bucket_name, object_key=object_key)
-            else:
-                config = get_file_contents_from_s3(bucket_name=bucket_name, object_key=object_key)
+            if object_key is None:
+                raise Exception("LITELLM_CONFIG_BUCKET_OBJECT_KEY must be set to load the config from a bucket.")
+
+            config = await get_config_from_bucket(
+                bucket_type=bucket_type,
+                bucket_name=bucket_name,
+                object_key=object_key,
+            )
 
             if config is None:
                 raise Exception("Unable to load config from given source.")
diff --git a/tests/test_litellm/proxy/common_utils/test_load_config_utils.py b/tests/test_litellm/proxy/common_utils/test_load_config_utils.py
index 524c260e94a..c5d84460d39 100644
--- a/tests/test_litellm/proxy/common_utils/test_load_config_utils.py
+++ b/tests/test_litellm/proxy/common_utils/test_load_config_utils.py
@@ -1,9 +1,14 @@
+import re
 from unittest.mock import MagicMock, mock_open, patch
 
 import pytest
 import yaml
 
-from litellm.proxy.common_utils.load_config_utils import get_file_contents_from_s3
+from litellm.proxy.common_utils.load_config_utils import (
+    get_config_from_bucket,
+    get_file_contents_from_s3,
+    resolve_bucket_includes,
+)
 
 
 class TestGetFileContentsFromS3:
@@ -83,3 +88,176 @@ class TestGetFileContentsFromS3:
 
         # Verify yaml.safe_load was called with the decoded content
         mock_yaml_load.assert_called_once_with(yaml_content)
+
+
+class TestBucketConfigIncludes:
+    """`include:` directives in a bucket-hosted config.yaml (LIT-6982).
+
+    They used to be dropped silently: the proxy booted with the root config applied and everything
+    the included objects declared missing, with nothing logged.
+    """
+
+    @staticmethod
+    def _bucket(objects):
+        async def fetch(object_key):
+            return objects.get(object_key)
+
+        return fetch
+
+    @pytest.mark.asyncio
+    async def test_include_resolves_against_the_config_objects_prefix(self):
+        merged = await resolve_bucket_includes(
+            config={"include": ["model_config.yaml"], "general_settings": {"master_key": "sk-1234"}},
+            object_key="configs/prod/config.yaml",
+            fetch=self._bucket(
+                {"configs/prod/model_config.yaml": {"model_list": [{"model_name": "gpt-4o-mini"}]}}
+            ),
+        )
+
+        assert merged == {
+            "general_settings": {"master_key": "sk-1234"},
+            "model_list": [{"model_name": "gpt-4o-mini"}],
+        }
+
+    @pytest.mark.asyncio
+    async def test_include_with_a_leading_slash_reads_from_the_bucket_root(self):
+        merged = await resolve_bucket_includes(
+            config={"include": ["/shared/models.yaml"]},
+            object_key="configs/prod/config.yaml",
+            fetch=self._bucket({"shared/models.yaml": {"model_list": [{"model_name": "shared"}]}}),
+        )
+
+        assert merged == {"model_list": [{"model_name": "shared"}]}
+
+    @pytest.mark.asyncio
+    async def test_include_walks_out_of_the_prefix_with_dot_dot(self):
+        merged = await resolve_bucket_includes(
+            config={"include": ["../shared/models.yaml"]},
+            object_key="configs/prod/config.yaml",
+            fetch=self._bucket({"configs/shared/models.yaml": {"model_list": [{"model_name": "shared"}]}}),
+        )
+
+        assert merged == {"model_list": [{"model_name": "shared"}]}
+
+    @pytest.mark.asyncio
+    async def test_included_configs_may_declare_further_includes(self):
+        merged = await resolve_bucket_includes(
+            config={"include": ["models.yaml"]},
+            object_key="configs/config.yaml",
+            fetch=self._bucket(
+                {
+                    "configs/models.yaml": {
+                        "include": ["extra/more_models.yaml"],
+                        "model_list": [{"model_name": "first"}],
+                    },
+                    "configs/extra/more_models.yaml": {"model_list": [{"model_name": "second"}]},
+                }
+            ),
+        )
+
+        assert merged == {"model_list": [{"model_name": "first"}, {"model_name": "second"}]}
+
+    @pytest.mark.asyncio
+    async def test_list_values_are_extended_and_other_values_are_overridden(self):
+        merged = await resolve_bucket_includes(
+            config={
+                "include": ["models.yaml"],
+                "model_list": [{"model_name": "from-root"}],
+                "litellm_settings": {"drop_params": True},
+            },
+            object_key="config.yaml",
+            fetch=self._bucket(
+                {
+                    "models.yaml": {
+                        "model_list": [{"model_name": "from-include"}],
+                        "litellm_settings": {"num_retries": 3},
+                    }
+                }
+            ),
+        )
+
+        assert merged == {
+            "model_list": [{"model_name": "from-root"}, {"model_name": "from-include"}],
+            "litellm_settings": {"num_retries": 3},
+        }
+
+    @pytest.mark.asyncio
+    async def test_a_missing_included_object_fails_loudly_with_its_key(self):
+        with pytest.raises(FileNotFoundError, match=re.escape("configs/prod/model_config.yaml")):
+            await resolve_bucket_includes(
+                config={"include": ["model_config.yaml"]},
+                object_key="configs/prod/config.yaml",
+                fetch=self._bucket({}),
+            )
+
+    @pytest.mark.asyncio
+    async def test_a_non_list_include_fails_loudly(self):
+        with pytest.raises(ValueError, match="'include' must be a list of file paths"):
+            await resolve_bucket_includes(
+                config={"include": "model_config.yaml"},
+                object_key="config.yaml",
+                fetch=self._bucket({}),
+            )
+
+    @pytest.mark.asyncio
+    async def test_get_config_from_bucket_merges_includes_over_s3(self, monkeypatch):
+        objects = {
+            "lit6982/config.yaml": {
+                "include": ["model_config.yaml"],
+                "general_settings": {"master_key": "sk-1234"},
+            },
+            "lit6982/model_config.yaml": {"model_list": [{"model_name": "included-model"}]},
+        }
+        monkeypatch.setattr(
+            "litellm.proxy.common_utils.load_config_utils.get_file_contents_from_s3",
+            lambda bucket_name, object_key: objects.get(object_key),
+        )
+
+        config = await get_config_from_bucket(
+            bucket_type="s3", bucket_name="litellm-configs", object_key="lit6982/config.yaml"
+        )
+
+        assert config == {
+            "general_settings": {"master_key": "sk-1234"},
+            "model_list": [{"model_name": "included-model"}],
+        }
+
+    @pytest.mark.asyncio
+    async def test_get_config_from_bucket_merges_includes_over_gcs(self, monkeypatch):
+        objects = {
+            "lit6982/config.yaml": {
+                "include": ["model_config.yaml"],
+                "general_settings": {"master_key": "sk-1234"},
+            },
+            "lit6982/model_config.yaml": {"model_list": [{"model_name": "included-model"}]},
+        }
+
+        async def fake_gcs(bucket_name, object_key):
+            return objects.get(object_key)
+
+        monkeypatch.setattr(
+            "litellm.proxy.common_utils.load_config_utils.get_config_file_contents_from_gcs", fake_gcs
+        )
+
+        config = await get_config_from_bucket(
+            bucket_type="gcs", bucket_name="litellm-configs", object_key="lit6982/config.yaml"
+        )
+
+        assert config == {
+            "general_settings": {"master_key": "sk-1234"},
+            "model_list": [{"model_name": "included-model"}],
+        }
+
+    @pytest.mark.asyncio
+    async def test_get_config_from_bucket_returns_none_when_the_root_object_is_missing(self, monkeypatch):
+        monkeypatch.setattr(
+            "litellm.proxy.common_utils.load_config_utils.get_file_contents_from_s3",
+            lambda bucket_name, object_key: None,
+        )
+
+        assert (
+            await get_config_from_bucket(
+                bucket_type="s3", bucket_name="litellm-configs", object_key="missing.yaml"
+            )
+            is None
+        )
diff --git a/tests/test_litellm/proxy/proxy_server/test_proxy_config.py b/tests/test_litellm/proxy/proxy_server/test_proxy_config.py
index 2babfe432f3..9b9c4af9e42 100644
--- a/tests/test_litellm/proxy/proxy_server/test_proxy_config.py
+++ b/tests/test_litellm/proxy/proxy_server/test_proxy_config.py
@@ -710,22 +710,32 @@ async def test_ProxyConfig__get_config_from_file_missing_path_raises():
 # ---------------------------------------------------------------------------
 
 
-def test_ProxyConfig__process_includes_merges_files(tmp_path):
+@pytest.mark.asyncio
+async def test_ProxyConfig__process_includes_merges_files(tmp_path):
     inc = tmp_path / "models.yaml"
     inc.write_text("model_list:\n  - model_name: gpt-4\n")
     pc = ProxyConfig()
     cfg = {"include": ["models.yaml"], "model_list": [], "litellm_settings": {}}
-    result = pc._process_includes(cfg, base_dir=str(tmp_path))
+    result = await pc._process_includes(cfg, base_dir=str(tmp_path))
     assert result == {
         "model_list": [{"model_name": "gpt-4"}],
         "litellm_settings": {},
     }
 
 
-def test_ProxyConfig__process_includes_missing_file_raises(tmp_path):
+@pytest.mark.asyncio
+async def test_ProxyConfig__process_includes_missing_file_raises(tmp_path):
     pc = ProxyConfig()
     with pytest.raises(FileNotFoundError):
-        pc._process_includes({"include": ["nope.yaml"]}, base_dir=str(tmp_path))
+        await pc._process_includes({"include": ["nope.yaml"]}, base_dir=str(tmp_path))
+
+
+@pytest.mark.asyncio
+async def test_ProxyConfig__process_includes_follows_nested_includes(tmp_path):
+    (tmp_path / "models.yaml").write_text("include:\n  - more_models.yaml\nmodel_list:\n  - model_name: first\n")
+    (tmp_path / "more_models.yaml").write_text("model_list:\n  - model_name: second\n")
+    result = await ProxyConfig()._process_includes({"include": ["models.yaml"]}, base_dir=str(tmp_path))
+    assert result == {"model_list": [{"model_name": "first"}, {"model_name": "second"}]}
 
 
 # ---------------------------------------------------------------------------
@@ -1042,6 +1052,32 @@ async def test_ProxyConfig_get_config_loads_from_file(tmp_path, monkeypatch):
     }
 
 
+@pytest.mark.asyncio
+async def test_ProxyConfig_get_config_from_a_bucket_merges_includes(monkeypatch):
+    """A bucket-hosted config.yaml used to drop its `include:` entries silently (LIT-6982)."""
+    objects = {
+        "lit6982/config.yaml": {
+            "include": ["model_config.yaml"],
+            "general_settings": {"master_key": "sk-1234"},
+        },
+        "lit6982/model_config.yaml": {"model_list": [{"model_name": "included-model"}]},
+    }
+    monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", None)
+    monkeypatch.setattr("litellm.proxy.proxy_server.store_model_in_db", False)
+    monkeypatch.setattr(
+        "litellm.proxy.common_utils.load_config_utils.get_file_contents_from_s3",
+        lambda bucket_name, object_key: objects.get(object_key),
+    )
+    monkeypatch.setenv("LITELLM_CONFIG_BUCKET_NAME", "litellm-configs")
+    monkeypatch.setenv("LITELLM_CONFIG_BUCKET_OBJECT_KEY", "lit6982/config.yaml")
+    monkeypatch.setenv("LITELLM_CONFIG_BUCKET_TYPE", "s3")
+
+    cfg = await ProxyConfig().get_config()
+
+    assert cfg["model_list"] == [{"model_name": "included-model"}]
+    assert "include" not in cfg
+
+
 @pytest.mark.asyncio
 async def test_ProxyConfig_get_config_missing_file_raises(monkeypatch):
     monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", None)

From 0a763bf00dcc335c0e9735936cc9d8679154332e Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Sun, 6 Sep 2026 02:05:18 -0700
Subject: [PATCH 2/4] fix(proxy): read a bucket config's include tree off the
 event loop

Reading a config from a bucket ran a blocking boto3 GET straight from the
event loop for every object in the include tree, and on GCS it built a new
bucket client per object, each one starting a flush task that never ends.

S3 reads now go through a worker thread, and one bucket client serves the
whole include tree.
---
 litellm/integrations/gcs_bucket/gcs_bucket.py |  3 +-
 litellm/proxy/common_utils/config_includes.py | 33 +++++--
 .../proxy/common_utils/load_config_utils.py   | 65 ++++++++++---
 litellm/proxy/proxy_server.py                 | 18 ++--
 .../gcs_bucket/test_gcs_bucket_base.py        | 17 ++++
 .../common_utils/test_load_config_utils.py    | 91 ++++++++++++++++++-
 .../proxy/proxy_server/test_proxy_config.py   | 62 ++++++++++++-
 7 files changed, 251 insertions(+), 38 deletions(-)

diff --git a/litellm/integrations/gcs_bucket/gcs_bucket.py b/litellm/integrations/gcs_bucket/gcs_bucket.py
index 31ceb338dcd..e338f490496 100644
--- a/litellm/integrations/gcs_bucket/gcs_bucket.py
+++ b/litellm/integrations/gcs_bucket/gcs_bucket.py
@@ -29,8 +29,6 @@ class GCSBucketLogger(GCSBucketBase, AdditionalLoggingUtils):
     def __init__(self, bucket_name: str | None = None) -> None:
         from litellm.proxy.proxy_server import premium_user
 
-        super().__init__(bucket_name=bucket_name)
-
         self.batch_size = int(os.getenv("GCS_BATCH_SIZE", GCS_DEFAULT_BATCH_SIZE))
         self.flush_interval = int(os.getenv("GCS_FLUSH_INTERVAL", GCS_DEFAULT_FLUSH_INTERVAL_SECONDS))
         self.use_batched_logging = (
@@ -38,6 +36,7 @@ class GCSBucketLogger(GCSBucketBase, AdditionalLoggingUtils):
         )
         self.flush_lock = asyncio.Lock()
         super().__init__(
+            bucket_name=bucket_name,
             flush_lock=self.flush_lock,
             batch_size=self.batch_size,
             flush_interval=self.flush_interval,
diff --git a/litellm/proxy/common_utils/config_includes.py b/litellm/proxy/common_utils/config_includes.py
index d6939db2e95..cf39bd6435e 100644
--- a/litellm/proxy/common_utils/config_includes.py
+++ b/litellm/proxy/common_utils/config_includes.py
@@ -6,7 +6,7 @@ INCLUDE_KEY: Final = "include"
 
 
 class ConfigLoader(Protocol):
-    def __call__(self, include_entry: str, /) -> Awaitable[Mapping[str, object]]: ...
+    def __call__(self, include_entry: str, declared_in: str, /) -> Awaitable[tuple[str, Mapping[str, object]]]: ...
 
 
 def _merged_value(base_value: object, included_value: object) -> object:
@@ -44,25 +44,40 @@ def include_entries(config: Mapping[str, object]) -> tuple[str, ...]:
     return paths
 
 
-async def _resolve(config: Mapping[str, object], pending: tuple[str, ...], load: ConfigLoader) -> Mapping[str, object]:
+def _pending_from(config: Mapping[str, object], location: str) -> tuple[tuple[str, str], ...]:
+    return tuple((entry, location) for entry in include_entries(config))
+
+
+async def _resolve(
+    config: Mapping[str, object],
+    pending: tuple[tuple[str, str], ...],
+    loaded: frozenset[str],
+    load: ConfigLoader,
+) -> Mapping[str, object]:
     if not pending:
         return _without_include(config)
 
-    included: Final = await load(pending[0])
+    entry, declared_in = pending[0]
+    location, included = await load(entry, declared_in)
+    if location in loaded:
+        return await _resolve(config, pending[1:], loaded, load)
+
     return await _resolve(
         _merged(config, _without_include(included)),
-        (*pending[1:], *include_entries(included)),
+        (*pending[1:], *_pending_from(included, location)),
+        loaded | frozenset((location,)),
         load,
     )
 
 
-async def resolve_includes(config: Mapping[str, object], load: ConfigLoader) -> dict[str, object]:
+async def resolve_includes(config: Mapping[str, object], location: str, load: ConfigLoader) -> dict[str, object]:
     """
     Merge every config named by the `include` directive into the config that declares it.
 
-    List values are extended and every other value is overridden, an included config may declare
-    further includes, and `load` decides where an entry is read from, so the same merge applies to
-    configs on disk and to configs hosted in a bucket.
+    List values are extended and every other value is overridden, each entry is resolved relative to
+    the config that declares it, a config already pulled in is not merged a second time, and `load`
+    decides where an entry is read from, so the same merge applies to configs on disk and to configs
+    hosted in a bucket.
     """
-    merged: Final = await _resolve(config, include_entries(config), load)
+    merged: Final = await _resolve(config, _pending_from(config, location), frozenset((location,)), load)
     return dict(merged)  # mutable-ok: the proxy mutates the config it loads
diff --git a/litellm/proxy/common_utils/load_config_utils.py b/litellm/proxy/common_utils/load_config_utils.py
index 927deb68826..e72272b1aca 100644
--- a/litellm/proxy/common_utils/load_config_utils.py
+++ b/litellm/proxy/common_utils/load_config_utils.py
@@ -1,7 +1,8 @@
+import asyncio
 import os
 import posixpath
 from collections.abc import Awaitable, Mapping
-from typing import Final, Protocol
+from typing import TYPE_CHECKING, Final, Protocol
 
 import yaml
 from pydantic import TypeAdapter, ValidationError
@@ -9,6 +10,9 @@ from pydantic import TypeAdapter, ValidationError
 from litellm._logging import verbose_proxy_logger
 from litellm.proxy.common_utils.config_includes import resolve_includes
 
+if TYPE_CHECKING:
+    from litellm.integrations.gcs_bucket.gcs_bucket_base import GCSBucketBase
+
 _BUCKET_CONFIG_ADAPTER: Final = TypeAdapter(dict[str, object])
 
 
@@ -16,6 +20,10 @@ class BucketObjectFetcher(Protocol):
     def __call__(self, object_key: str, /) -> Awaitable[Mapping[str, object] | None]: ...
 
 
+class BucketObjectReader(Protocol):
+    def __call__(self, object_key: str, /) -> Awaitable[object | None]: ...
+
+
 def get_file_contents_from_s3(bucket_name, object_key):
     try:
         # v0 rely on boto3 for authentication - allowing boto3 to handle IAM credentials etc
@@ -51,14 +59,22 @@ def get_file_contents_from_s3(bucket_name, object_key):
         return None
 
 
-async def get_config_file_contents_from_gcs(bucket_name, object_key):
+def gcs_config_bucket(bucket_name: str) -> "GCSBucketBase | None":
     try:
         from litellm.integrations.gcs_bucket.gcs_bucket import GCSBucketLogger
 
-        gcs_bucket: Final = GCSBucketLogger(
-            bucket_name=bucket_name,
-        )
-        file_contents = await gcs_bucket.download_gcs_object(object_key)
+        return GCSBucketLogger(bucket_name=bucket_name)
+    except Exception as e:
+        verbose_proxy_logger.error("Error creating the GCS client for bucket %s: %s", bucket_name, e)
+        return None
+
+
+async def get_config_file_contents_from_gcs(bucket_name, object_key, gcs_bucket=None):
+    try:
+        bucket: Final = gcs_config_bucket(bucket_name) if gcs_bucket is None else gcs_bucket
+        if bucket is None:
+            return None
+        file_contents = await bucket.download_gcs_object(object_key)
         if file_contents is None:
             raise Exception(f"File contents are None for {object_key}")
         # file_contentis is a bytes object, so we need to convert it to yaml
@@ -90,14 +106,35 @@ async def resolve_bucket_includes(
     object_key: str,
     fetch: BucketObjectFetcher,
 ) -> dict[str, object]:
-    async def load(include_entry: str) -> Mapping[str, object]:
-        include_key: Final = resolve_include_object_key(object_key, include_entry)
+    async def load(include_entry: str, declared_in: str) -> tuple[str, Mapping[str, object]]:
+        include_key: Final = resolve_include_object_key(declared_in, include_entry)
         included: Final = await fetch(include_key)
         if included is None:
             raise FileNotFoundError(f"Included config could not be read from bucket: {include_key}")
-        return included
+        return include_key, included
 
-    return await resolve_includes(config=config, load=load)
+    return await resolve_includes(config=config, location=object_key, load=load)
+
+
+def bucket_object_reader(bucket_type: str | None, bucket_name: str) -> BucketObjectReader:
+    """
+    Build one reader for a whole config, so an `include` tree costs one bucket client rather than one per object.
+    """
+    if bucket_type != "gcs":
+
+        async def read_from_s3(object_key: str) -> object | None:
+            return await asyncio.to_thread(get_file_contents_from_s3, bucket_name, object_key)
+
+        return read_from_s3
+
+    gcs_bucket: Final = gcs_config_bucket(bucket_name)
+
+    async def read_from_gcs(object_key: str) -> object | None:
+        if gcs_bucket is None:
+            return None
+        return await get_config_file_contents_from_gcs(bucket_name, object_key, gcs_bucket)
+
+    return read_from_gcs
 
 
 async def get_config_from_bucket(
@@ -106,12 +143,10 @@ async def get_config_from_bucket(
     bucket_name: str,
     object_key: str,
 ) -> dict[str, object] | None:
+    read: Final = bucket_object_reader(bucket_type, bucket_name)
+
     async def fetch(key: str) -> Mapping[str, object] | None:
-        raw: Final = (
-            await get_config_file_contents_from_gcs(bucket_name=bucket_name, object_key=key)
-            if bucket_type == "gcs"
-            else get_file_contents_from_s3(bucket_name=bucket_name, object_key=key)
-        )
+        raw: Final = await read(key)
         if raw is None:
             return None
         try:
diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py
index a49df166dca..7ac5e697b61 100644
--- a/litellm/proxy/proxy_server.py
+++ b/litellm/proxy/proxy_server.py
@@ -4536,12 +4536,12 @@ class ProxyConfig:
         if config is None:
             raise Exception("Config cannot be None or Empty.")
         # Process includes
-        config = await self._process_includes(config=config, base_dir=os.path.dirname(os.path.abspath(file_path or "")))
+        config = await self._process_includes(config=config, config_file_path=os.path.abspath(file_path or ""))
 
         # verbose_proxy_logger.debug(f"loaded config={json.dumps(config, indent=4)}")
         return config
 
-    async def _process_includes(self, config: dict, base_dir: str) -> dict:
+    async def _process_includes(self, config: dict, config_file_path: str) -> dict:
         """
         Process includes by appending their contents to the main config
 
@@ -4557,13 +4557,19 @@ class ProxyConfig:
         ```
         """
 
-        async def load_included(include_file: str) -> Mapping[str, object]:
-            file_path: Final = os.path.join(base_dir, include_file)
+        included_config_adapter: Final = TypeAdapter(dict[str, object])
+
+        async def load_included(include_file: str, declared_in: str) -> tuple[str, Mapping[str, object]]:
+            file_path: Final = os.path.abspath(os.path.join(os.path.dirname(declared_in), include_file))
             if not os.path.exists(file_path):
                 raise FileNotFoundError(f"Included file not found: {file_path}")
-            return self._load_yaml_file(file_path)
+            try:
+                included: Final = included_config_adapter.validate_python(self._load_yaml_file(file_path))
+            except ValidationError as e:
+                raise ValueError(f"Included config file is not a YAML mapping: {file_path}") from e
+            return file_path, included
 
-        return await resolve_includes(config=config, load=load_included)
+        return await resolve_includes(config=config, location=config_file_path, load=load_included)
 
     async def save_config(self, new_config: dict, include_env_vars: bool = False):
         global prisma_client, general_settings, user_config_file_path, store_model_in_db
diff --git a/tests/test_litellm/integrations/gcs_bucket/test_gcs_bucket_base.py b/tests/test_litellm/integrations/gcs_bucket/test_gcs_bucket_base.py
index 8d662311da1..a458752bed0 100644
--- a/tests/test_litellm/integrations/gcs_bucket/test_gcs_bucket_base.py
+++ b/tests/test_litellm/integrations/gcs_bucket/test_gcs_bucket_base.py
@@ -128,3 +128,20 @@ class TestGCSBucketBase:
         assert object_name.endswith("-target_uploadType_media")
         assert ".." not in object_name
         assert "?" not in object_name
+
+
+class TestGCSBucketLoggerBucketName:
+    @pytest.mark.asyncio
+    async def test_the_bucket_name_it_is_constructed_with_survives(self, monkeypatch):
+        """Reading config.yaml out of a GCS bucket asks for that bucket, not the logging one (LIT-6982)."""
+        monkeypatch.setenv("GCS_BUCKET_NAME", "logging-bucket")
+        monkeypatch.setattr("litellm.proxy.proxy_server.premium_user", True)
+
+        assert GCSBucketLogger(bucket_name="config-bucket").BUCKET_NAME == "config-bucket"
+
+    @pytest.mark.asyncio
+    async def test_no_bucket_name_still_falls_back_to_the_environment(self, monkeypatch):
+        monkeypatch.setenv("GCS_BUCKET_NAME", "logging-bucket")
+        monkeypatch.setattr("litellm.proxy.proxy_server.premium_user", True)
+
+        assert GCSBucketLogger().BUCKET_NAME == "logging-bucket"
diff --git a/tests/test_litellm/proxy/common_utils/test_load_config_utils.py b/tests/test_litellm/proxy/common_utils/test_load_config_utils.py
index c5d84460d39..b654320569c 100644
--- a/tests/test_litellm/proxy/common_utils/test_load_config_utils.py
+++ b/tests/test_litellm/proxy/common_utils/test_load_config_utils.py
@@ -1,4 +1,6 @@
+import asyncio
 import re
+import threading
 from unittest.mock import MagicMock, mock_open, patch
 
 import pytest
@@ -157,6 +159,60 @@ class TestBucketConfigIncludes:
 
         assert merged == {"model_list": [{"model_name": "first"}, {"model_name": "second"}]}
 
+    @pytest.mark.asyncio
+    async def test_a_nested_include_resolves_against_the_object_that_declares_it(self):
+        """A nested `include` names a neighbour of the object declaring it, not of the root config."""
+        merged = await resolve_bucket_includes(
+            config={"include": ["shared/models.yaml"]},
+            object_key="configs/config.yaml",
+            fetch=self._bucket(
+                {
+                    "configs/shared/models.yaml": {
+                        "include": ["more_models.yaml"],
+                        "model_list": [{"model_name": "first"}],
+                    },
+                    "configs/shared/more_models.yaml": {"model_list": [{"model_name": "second"}]},
+                    "configs/more_models.yaml": {"model_list": [{"model_name": "wrong-prefix"}]},
+                }
+            ),
+        )
+
+        assert merged == {"model_list": [{"model_name": "first"}, {"model_name": "second"}]}
+
+    @pytest.mark.asyncio
+    async def test_an_object_pulled_in_twice_is_merged_once(self):
+        merged = await resolve_bucket_includes(
+            config={"include": ["a.yaml", "b.yaml"]},
+            object_key="configs/config.yaml",
+            fetch=self._bucket(
+                {
+                    "configs/a.yaml": {"include": ["shared.yaml"]},
+                    "configs/b.yaml": {"include": ["./shared.yaml"]},
+                    "configs/shared.yaml": {"model_list": [{"model_name": "shared"}]},
+                }
+            ),
+        )
+
+        assert merged == {"model_list": [{"model_name": "shared"}]}
+
+    @pytest.mark.asyncio
+    async def test_a_cycle_between_included_objects_terminates(self):
+        merged = await asyncio.wait_for(
+            resolve_bucket_includes(
+                config={"include": ["a.yaml"]},
+                object_key="configs/config.yaml",
+                fetch=self._bucket(
+                    {
+                        "configs/a.yaml": {"include": ["b.yaml"], "model_list": [{"model_name": "from-a"}]},
+                        "configs/b.yaml": {"include": ["a.yaml"], "model_list": [{"model_name": "from-b"}]},
+                    }
+                ),
+            ),
+            timeout=10,
+        )
+
+        assert merged == {"model_list": [{"model_name": "from-a"}, {"model_name": "from-b"}]}
+
     @pytest.mark.asyncio
     async def test_list_values_are_extended_and_other_values_are_overridden(self):
         merged = await resolve_bucket_includes(
@@ -222,6 +278,23 @@ class TestBucketConfigIncludes:
             "model_list": [{"model_name": "included-model"}],
         }
 
+    @pytest.mark.asyncio
+    async def test_the_blocking_s3_read_runs_off_the_event_loop_thread(self, monkeypatch):
+        loop_thread = threading.current_thread()
+        read_threads = []
+
+        def record_thread(bucket_name, object_key):
+            read_threads.append(threading.current_thread())
+            return {"model_list": [{"model_name": "a-model"}]}
+
+        monkeypatch.setattr(
+            "litellm.proxy.common_utils.load_config_utils.get_file_contents_from_s3", record_thread
+        )
+
+        await get_config_from_bucket(bucket_type="s3", bucket_name="litellm-configs", object_key="config.yaml")
+
+        assert read_threads and loop_thread not in read_threads
+
     @pytest.mark.asyncio
     async def test_get_config_from_bucket_merges_includes_over_gcs(self, monkeypatch):
         objects = {
@@ -232,11 +305,20 @@ class TestBucketConfigIncludes:
             "lit6982/model_config.yaml": {"model_list": [{"model_name": "included-model"}]},
         }
 
-        async def fake_gcs(bucket_name, object_key):
-            return objects.get(object_key)
+        buckets = []
+
+        class FakeGCSBucket:
+            def __init__(self):
+                self.requested = []
+                buckets.append(self)
+
+            async def download_gcs_object(self, object_key):
+                self.requested.append(object_key)
+                return yaml.dump(objects[object_key]).encode("utf-8")
 
         monkeypatch.setattr(
-            "litellm.proxy.common_utils.load_config_utils.get_config_file_contents_from_gcs", fake_gcs
+            "litellm.proxy.common_utils.load_config_utils.gcs_config_bucket",
+            lambda bucket_name: FakeGCSBucket(),
         )
 
         config = await get_config_from_bucket(
@@ -247,6 +329,9 @@ class TestBucketConfigIncludes:
             "general_settings": {"master_key": "sk-1234"},
             "model_list": [{"model_name": "included-model"}],
         }
+        assert [bucket.requested for bucket in buckets] == [
+            ["lit6982/config.yaml", "lit6982/model_config.yaml"]
+        ]
 
     @pytest.mark.asyncio
     async def test_get_config_from_bucket_returns_none_when_the_root_object_is_missing(self, monkeypatch):
diff --git a/tests/test_litellm/proxy/proxy_server/test_proxy_config.py b/tests/test_litellm/proxy/proxy_server/test_proxy_config.py
index 9b9c4af9e42..3d9138ad098 100644
--- a/tests/test_litellm/proxy/proxy_server/test_proxy_config.py
+++ b/tests/test_litellm/proxy/proxy_server/test_proxy_config.py
@@ -8,6 +8,7 @@ Pins covered:
 
 from __future__ import annotations
 
+import asyncio
 import json
 import os
 import re
@@ -716,7 +717,7 @@ async def test_ProxyConfig__process_includes_merges_files(tmp_path):
     inc.write_text("model_list:\n  - model_name: gpt-4\n")
     pc = ProxyConfig()
     cfg = {"include": ["models.yaml"], "model_list": [], "litellm_settings": {}}
-    result = await pc._process_includes(cfg, base_dir=str(tmp_path))
+    result = await pc._process_includes(cfg, config_file_path=str(tmp_path / "config.yaml"))
     assert result == {
         "model_list": [{"model_name": "gpt-4"}],
         "litellm_settings": {},
@@ -727,17 +728,72 @@ async def test_ProxyConfig__process_includes_merges_files(tmp_path):
 async def test_ProxyConfig__process_includes_missing_file_raises(tmp_path):
     pc = ProxyConfig()
     with pytest.raises(FileNotFoundError):
-        await pc._process_includes({"include": ["nope.yaml"]}, base_dir=str(tmp_path))
+        await pc._process_includes({"include": ["nope.yaml"]}, config_file_path=str(tmp_path / "config.yaml"))
 
 
 @pytest.mark.asyncio
 async def test_ProxyConfig__process_includes_follows_nested_includes(tmp_path):
     (tmp_path / "models.yaml").write_text("include:\n  - more_models.yaml\nmodel_list:\n  - model_name: first\n")
     (tmp_path / "more_models.yaml").write_text("model_list:\n  - model_name: second\n")
-    result = await ProxyConfig()._process_includes({"include": ["models.yaml"]}, base_dir=str(tmp_path))
+    result = await ProxyConfig()._process_includes(
+        {"include": ["models.yaml"]}, config_file_path=str(tmp_path / "config.yaml")
+    )
     assert result == {"model_list": [{"model_name": "first"}, {"model_name": "second"}]}
 
 
+@pytest.mark.asyncio
+async def test_ProxyConfig__process_includes_resolves_a_nested_include_next_to_its_own_file(tmp_path):
+    """A nested `include` names a sibling of the file that declares it, not of the root config."""
+    (tmp_path / "shared").mkdir()
+    (tmp_path / "shared" / "models.yaml").write_text(
+        "include:\n  - more_models.yaml\nmodel_list:\n  - model_name: first\n"
+    )
+    (tmp_path / "shared" / "more_models.yaml").write_text("model_list:\n  - model_name: second\n")
+    (tmp_path / "more_models.yaml").write_text("model_list:\n  - model_name: wrong-directory\n")
+
+    result = await ProxyConfig()._process_includes(
+        {"include": ["shared/models.yaml"]}, config_file_path=str(tmp_path / "config.yaml")
+    )
+
+    assert result == {"model_list": [{"model_name": "first"}, {"model_name": "second"}]}
+
+
+@pytest.mark.asyncio
+async def test_ProxyConfig__process_includes_merges_a_shared_file_once(tmp_path):
+    (tmp_path / "shared.yaml").write_text("model_list:\n  - model_name: shared\n")
+    (tmp_path / "a.yaml").write_text("include:\n  - shared.yaml\n")
+    (tmp_path / "b.yaml").write_text("include:\n  - ./shared.yaml\n")
+
+    result = await ProxyConfig()._process_includes(
+        {"include": ["a.yaml", "b.yaml"]}, config_file_path=str(tmp_path / "config.yaml")
+    )
+
+    assert result == {"model_list": [{"model_name": "shared"}]}
+
+
+@pytest.mark.asyncio
+async def test_ProxyConfig__process_includes_names_the_file_when_it_is_not_a_mapping(tmp_path):
+    (tmp_path / "models.yaml").write_text("- model_name: gpt-4\n")
+
+    with pytest.raises(ValueError, match=re.escape(str(tmp_path / "models.yaml"))):
+        await ProxyConfig()._process_includes(
+            {"include": ["models.yaml"]}, config_file_path=str(tmp_path / "config.yaml")
+        )
+
+
+@pytest.mark.asyncio
+async def test_ProxyConfig__process_includes_terminates_on_a_cycle(tmp_path):
+    (tmp_path / "a.yaml").write_text("include:\n  - b.yaml\nmodel_list:\n  - model_name: from-a\n")
+    (tmp_path / "b.yaml").write_text("include:\n  - a.yaml\nmodel_list:\n  - model_name: from-b\n")
+
+    result = await asyncio.wait_for(
+        ProxyConfig()._process_includes({"include": ["a.yaml"]}, config_file_path=str(tmp_path / "config.yaml")),
+        timeout=10,
+    )
+
+    assert result == {"model_list": [{"model_name": "from-a"}, {"model_name": "from-b"}]}
+
+
 # ---------------------------------------------------------------------------
 # ProxyConfig.save_config
 # ---------------------------------------------------------------------------

From 12204e523079f07aeefdb2257e2b2c938771aae4 Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Sun, 6 Sep 2026 02:47:47 -0700
Subject: [PATCH 3/4] fix(proxy): resolve disk includes next to the config that
 declares them

Keep reading an include left beside the root config, with a warning naming where it
was found, so a nested include written against the old rule still boots.

Also build one S3 client per config load rather than one per included object, treat an
empty included object as an empty config instead of failing the boot, and point the
error a dropped bucket include raises at the bucket error logged with it.
---
 litellm/proxy/common_utils/config_includes.py | 28 +++++++
 .../proxy/common_utils/load_config_utils.py   | 77 ++++++++++-------
 litellm/proxy/proxy_server.py                 |  4 +-
 .../common_utils/test_load_config_utils.py    | 83 ++++++++++++++-----
 .../proxy/proxy_server/test_proxy_config.py   | 21 ++++-
 5 files changed, 160 insertions(+), 53 deletions(-)

diff --git a/litellm/proxy/common_utils/config_includes.py b/litellm/proxy/common_utils/config_includes.py
index cf39bd6435e..23e6c615a09 100644
--- a/litellm/proxy/common_utils/config_includes.py
+++ b/litellm/proxy/common_utils/config_includes.py
@@ -1,10 +1,38 @@
+import os
 from collections.abc import Awaitable, Mapping
 from types import MappingProxyType
 from typing import Final, Protocol
 
+from litellm._logging import verbose_proxy_logger
+
 INCLUDE_KEY: Final = "include"
 
 
+def resolve_include_file_path(include_file: str, declared_in: str, root_config_path: str) -> str:
+    """
+    Resolve one `include` entry to the file it names, next to the config that declares it.
+
+    A config written before nested entries resolved this way can name a file sitting next to the root
+    config instead, so that file is still read, with a warning naming where it was found.
+    """
+    declared_relative: Final = os.path.abspath(os.path.join(os.path.dirname(declared_in), include_file))
+    if os.path.exists(declared_relative):
+        return declared_relative
+
+    root_relative: Final = os.path.abspath(os.path.join(os.path.dirname(root_config_path), include_file))
+    if root_relative == declared_relative or not os.path.exists(root_relative):
+        return declared_relative
+
+    verbose_proxy_logger.warning(
+        "Config include '%s' declared in %s was not found next to it, so %s was read instead. "
+        "Move the included file next to the config that declares it.",
+        include_file,
+        declared_in,
+        root_relative,
+    )
+    return root_relative
+
+
 class ConfigLoader(Protocol):
     def __call__(self, include_entry: str, declared_in: str, /) -> Awaitable[tuple[str, Mapping[str, object]]]: ...
 
diff --git a/litellm/proxy/common_utils/load_config_utils.py b/litellm/proxy/common_utils/load_config_utils.py
index e72272b1aca..d62286a50f6 100644
--- a/litellm/proxy/common_utils/load_config_utils.py
+++ b/litellm/proxy/common_utils/load_config_utils.py
@@ -2,6 +2,7 @@ import asyncio
 import os
 import posixpath
 from collections.abc import Awaitable, Mapping
+from types import MappingProxyType
 from typing import TYPE_CHECKING, Final, Protocol
 
 import yaml
@@ -24,7 +25,19 @@ class BucketObjectReader(Protocol):
     def __call__(self, object_key: str, /) -> Awaitable[object | None]: ...
 
 
-def get_file_contents_from_s3(bucket_name, object_key):
+class SyncBucketObjectReader(Protocol):
+    def __call__(self, object_key: str, /) -> object | None: ...
+
+
+def _parsed_config(file_contents: str) -> object:
+    parsed: Final = yaml.safe_load(file_contents)
+    return MappingProxyType({}) if parsed is None else parsed
+
+
+def s3_object_reader(bucket_name: str) -> SyncBucketObjectReader:
+    """
+    Build one reader for a whole config, so an `include` tree costs one S3 client rather than one per object.
+    """
     try:
         # v0 rely on boto3 for authentication - allowing boto3 to handle IAM credentials etc
         import boto3
@@ -39,24 +52,28 @@ def get_file_contents_from_s3(bucket_name, object_key):
             aws_secret_access_key=credentials.secret_key,
             aws_session_token=credentials.token,  # Optional, if using temporary credentials
         )
-        verbose_proxy_logger.debug("Retrieving %s from S3 bucket: %s", object_key, bucket_name)
-        response: Final = s3_client.get_object(Bucket=bucket_name, Key=object_key)
-        verbose_proxy_logger.debug("Response: %s", response)
-
-        # Read the file contents and directly parse YAML
-        file_contents: Final = response["Body"].read().decode("utf-8")
-        verbose_proxy_logger.debug("File contents retrieved from S3")
-
-        # Parse YAML directly from string
-        config: Final = yaml.safe_load(file_contents)
-        return config
-
     except ImportError as e:
         # this is most likely if a user is not using the litellm docker container
         verbose_proxy_logger.error("ImportError: %s", e)
+        return lambda object_key: None
     except Exception as e:
-        verbose_proxy_logger.error("Error retrieving file contents: %s", e)
-        return None
+        verbose_proxy_logger.error("Error creating the S3 client for bucket %s: %s", bucket_name, e)
+        return lambda object_key: None
+
+    def read(object_key: str) -> object | None:
+        try:
+            verbose_proxy_logger.debug("Retrieving %s from S3 bucket: %s", object_key, bucket_name)
+            response: Final = s3_client.get_object(Bucket=bucket_name, Key=object_key)
+            return _parsed_config(response["Body"].read().decode("utf-8"))
+        except Exception as e:  # noqa: BLE001  # any boto3 error must read as a missing object
+            verbose_proxy_logger.error("Error retrieving %s from S3 bucket %s: %s", object_key, bucket_name, e)
+            return None
+
+    return read
+
+
+def get_file_contents_from_s3(bucket_name: str, object_key: str) -> object | None:
+    return s3_object_reader(bucket_name)(object_key)
 
 
 def gcs_config_bucket(bucket_name: str) -> "GCSBucketBase | None":
@@ -64,27 +81,27 @@ def gcs_config_bucket(bucket_name: str) -> "GCSBucketBase | None":
         from litellm.integrations.gcs_bucket.gcs_bucket import GCSBucketLogger
 
         return GCSBucketLogger(bucket_name=bucket_name)
-    except Exception as e:
+    except Exception as e:  # noqa: BLE001  # an unbuildable client must read as an unreadable bucket
         verbose_proxy_logger.error("Error creating the GCS client for bucket %s: %s", bucket_name, e)
         return None
 
 
-async def get_config_file_contents_from_gcs(bucket_name, object_key, gcs_bucket=None):
+async def get_config_file_contents_from_gcs(
+    bucket_name: str,
+    object_key: str,
+    gcs_bucket: "GCSBucketBase | None" = None,
+) -> object | None:
     try:
         bucket: Final = gcs_config_bucket(bucket_name) if gcs_bucket is None else gcs_bucket
         if bucket is None:
             return None
-        file_contents = await bucket.download_gcs_object(object_key)
+        file_contents: Final = await bucket.download_gcs_object(object_key)
         if file_contents is None:
             raise Exception(f"File contents are None for {object_key}")
-        # file_contentis is a bytes object, so we need to convert it to yaml
-        file_contents = file_contents.decode("utf-8")
-        # convert to yaml
-        config: Final = yaml.safe_load(file_contents)
-        return config
+        return _parsed_config(file_contents.decode("utf-8"))
 
     except Exception as e:
-        verbose_proxy_logger.error("Error retrieving file contents: %s", e)
+        verbose_proxy_logger.error("Error retrieving %s from GCS bucket %s: %s", object_key, bucket_name, e)
         return None
 
 
@@ -110,20 +127,24 @@ async def resolve_bucket_includes(
         include_key: Final = resolve_include_object_key(declared_in, include_entry)
         included: Final = await fetch(include_key)
         if included is None:
-            raise FileNotFoundError(f"Included config could not be read from bucket: {include_key}")
+            raise FileNotFoundError(
+                f"Included config could not be read from bucket: {include_key}. "
+                "The underlying bucket error is logged above."
+            )
         return include_key, included
 
     return await resolve_includes(config=config, location=object_key, load=load)
 
 
-def bucket_object_reader(bucket_type: str | None, bucket_name: str) -> BucketObjectReader:
+async def bucket_object_reader(bucket_type: str | None, bucket_name: str) -> BucketObjectReader:
     """
     Build one reader for a whole config, so an `include` tree costs one bucket client rather than one per object.
     """
     if bucket_type != "gcs":
+        read_object: Final = await asyncio.to_thread(s3_object_reader, bucket_name)
 
         async def read_from_s3(object_key: str) -> object | None:
-            return await asyncio.to_thread(get_file_contents_from_s3, bucket_name, object_key)
+            return await asyncio.to_thread(read_object, object_key)
 
         return read_from_s3
 
@@ -143,7 +164,7 @@ async def get_config_from_bucket(
     bucket_name: str,
     object_key: str,
 ) -> dict[str, object] | None:
-    read: Final = bucket_object_reader(bucket_type, bucket_name)
+    read: Final = await bucket_object_reader(bucket_type, bucket_name)
 
     async def fetch(key: str) -> Mapping[str, object] | None:
         raw: Final = await read(key)
diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py
index 7ac5e697b61..7fada09febf 100644
--- a/litellm/proxy/proxy_server.py
+++ b/litellm/proxy/proxy_server.py
@@ -341,7 +341,7 @@ from litellm.proxy.common_utils.auth_cache_invalidation_pubsub import (
     AuthCacheInvalidationSubscriber,
 )
 from litellm.proxy.common_utils.callback_utils import initialize_callbacks_on_proxy
-from litellm.proxy.common_utils.config_includes import resolve_includes
+from litellm.proxy.common_utils.config_includes import resolve_include_file_path, resolve_includes
 from litellm.proxy.common_utils.config_sync_pubsub import ConfigSyncSubscriber
 from litellm.proxy.common_utils.debug_utils import init_verbose_loggers
 from litellm.proxy.common_utils.debug_utils import router as debugging_endpoints_router
@@ -4560,7 +4560,7 @@ class ProxyConfig:
         included_config_adapter: Final = TypeAdapter(dict[str, object])
 
         async def load_included(include_file: str, declared_in: str) -> tuple[str, Mapping[str, object]]:
-            file_path: Final = os.path.abspath(os.path.join(os.path.dirname(declared_in), include_file))
+            file_path: Final = resolve_include_file_path(include_file, declared_in, config_file_path)
             if not os.path.exists(file_path):
                 raise FileNotFoundError(f"Included file not found: {file_path}")
             try:
diff --git a/tests/test_litellm/proxy/common_utils/test_load_config_utils.py b/tests/test_litellm/proxy/common_utils/test_load_config_utils.py
index b654320569c..75aa5a3f0f8 100644
--- a/tests/test_litellm/proxy/common_utils/test_load_config_utils.py
+++ b/tests/test_litellm/proxy/common_utils/test_load_config_utils.py
@@ -93,11 +93,6 @@ class TestGetFileContentsFromS3:
 
 
 class TestBucketConfigIncludes:
-    """`include:` directives in a bucket-hosted config.yaml (LIT-6982).
-
-    They used to be dropped silently: the proxy booted with the root config applied and everything
-    the included objects declared missing, with nothing logged.
-    """
 
     @staticmethod
     def _bucket(objects):
@@ -161,7 +156,6 @@ class TestBucketConfigIncludes:
 
     @pytest.mark.asyncio
     async def test_a_nested_include_resolves_against_the_object_that_declares_it(self):
-        """A nested `include` names a neighbour of the object declaring it, not of the root config."""
         merged = await resolve_bucket_includes(
             config={"include": ["shared/models.yaml"]},
             object_key="configs/config.yaml",
@@ -265,8 +259,8 @@ class TestBucketConfigIncludes:
             "lit6982/model_config.yaml": {"model_list": [{"model_name": "included-model"}]},
         }
         monkeypatch.setattr(
-            "litellm.proxy.common_utils.load_config_utils.get_file_contents_from_s3",
-            lambda bucket_name, object_key: objects.get(object_key),
+            "litellm.proxy.common_utils.load_config_utils.s3_object_reader",
+            lambda bucket_name: objects.get,
         )
 
         config = await get_config_from_bucket(
@@ -279,21 +273,72 @@ class TestBucketConfigIncludes:
         }
 
     @pytest.mark.asyncio
-    async def test_the_blocking_s3_read_runs_off_the_event_loop_thread(self, monkeypatch):
+    async def test_the_blocking_s3_work_runs_off_the_event_loop_thread(self, monkeypatch):
         loop_thread = threading.current_thread()
-        read_threads = []
+        threads = []
 
-        def record_thread(bucket_name, object_key):
-            read_threads.append(threading.current_thread())
-            return {"model_list": [{"model_name": "a-model"}]}
+        def build_reader(bucket_name):
+            threads.append(threading.current_thread())
 
-        monkeypatch.setattr(
-            "litellm.proxy.common_utils.load_config_utils.get_file_contents_from_s3", record_thread
-        )
+            def read(object_key):
+                threads.append(threading.current_thread())
+                return {"model_list": [{"model_name": "a-model"}]}
+
+            return read
+
+        monkeypatch.setattr("litellm.proxy.common_utils.load_config_utils.s3_object_reader", build_reader)
 
         await get_config_from_bucket(bucket_type="s3", bucket_name="litellm-configs", object_key="config.yaml")
 
-        assert read_threads and loop_thread not in read_threads
+        assert len(threads) == 2 and loop_thread not in threads
+
+    @pytest.mark.asyncio
+    async def test_one_s3_client_serves_the_whole_include_tree(self, monkeypatch):
+        objects = {
+            "lit6982/config.yaml": {"include": ["model_config.yaml"]},
+            "lit6982/model_config.yaml": {"model_list": [{"model_name": "included-model"}]},
+        }
+        readers = []
+
+        def build_reader(bucket_name):
+            requested = []
+            readers.append(requested)
+
+            def read(object_key):
+                requested.append(object_key)
+                return objects.get(object_key)
+
+            return read
+
+        monkeypatch.setattr("litellm.proxy.common_utils.load_config_utils.s3_object_reader", build_reader)
+
+        await get_config_from_bucket(
+            bucket_type="s3", bucket_name="litellm-configs", object_key="lit6982/config.yaml"
+        )
+
+        assert readers == [["lit6982/config.yaml", "lit6982/model_config.yaml"]]
+
+    @pytest.mark.asyncio
+    async def test_an_empty_included_object_merges_as_an_empty_config(self, monkeypatch):
+        objects = {
+            "lit6982/config.yaml": "include:\n  - empty.yaml\nmodel_list:\n  - model_name: only-model\n",
+            "lit6982/empty.yaml": "",
+        }
+
+        class FakeGCSBucket:
+            async def download_gcs_object(self, object_key):
+                return objects[object_key].encode("utf-8")
+
+        monkeypatch.setattr(
+            "litellm.proxy.common_utils.load_config_utils.gcs_config_bucket",
+            lambda bucket_name: FakeGCSBucket(),
+        )
+
+        config = await get_config_from_bucket(
+            bucket_type="gcs", bucket_name="litellm-configs", object_key="lit6982/config.yaml"
+        )
+
+        assert config == {"model_list": [{"model_name": "only-model"}]}
 
     @pytest.mark.asyncio
     async def test_get_config_from_bucket_merges_includes_over_gcs(self, monkeypatch):
@@ -336,8 +381,8 @@ class TestBucketConfigIncludes:
     @pytest.mark.asyncio
     async def test_get_config_from_bucket_returns_none_when_the_root_object_is_missing(self, monkeypatch):
         monkeypatch.setattr(
-            "litellm.proxy.common_utils.load_config_utils.get_file_contents_from_s3",
-            lambda bucket_name, object_key: None,
+            "litellm.proxy.common_utils.load_config_utils.s3_object_reader",
+            lambda bucket_name: (lambda object_key: None),
         )
 
         assert (
diff --git a/tests/test_litellm/proxy/proxy_server/test_proxy_config.py b/tests/test_litellm/proxy/proxy_server/test_proxy_config.py
index 3d9138ad098..e90f101a555 100644
--- a/tests/test_litellm/proxy/proxy_server/test_proxy_config.py
+++ b/tests/test_litellm/proxy/proxy_server/test_proxy_config.py
@@ -743,7 +743,6 @@ async def test_ProxyConfig__process_includes_follows_nested_includes(tmp_path):
 
 @pytest.mark.asyncio
 async def test_ProxyConfig__process_includes_resolves_a_nested_include_next_to_its_own_file(tmp_path):
-    """A nested `include` names a sibling of the file that declares it, not of the root config."""
     (tmp_path / "shared").mkdir()
     (tmp_path / "shared" / "models.yaml").write_text(
         "include:\n  - more_models.yaml\nmodel_list:\n  - model_name: first\n"
@@ -758,6 +757,21 @@ async def test_ProxyConfig__process_includes_resolves_a_nested_include_next_to_i
     assert result == {"model_list": [{"model_name": "first"}, {"model_name": "second"}]}
 
 
+@pytest.mark.asyncio
+async def test_ProxyConfig__process_includes_still_reads_a_nested_include_left_beside_the_root_config(tmp_path):
+    (tmp_path / "shared").mkdir()
+    (tmp_path / "shared" / "models.yaml").write_text(
+        "include:\n  - more_models.yaml\nmodel_list:\n  - model_name: first\n"
+    )
+    (tmp_path / "more_models.yaml").write_text("model_list:\n  - model_name: second\n")
+
+    result = await ProxyConfig()._process_includes(
+        {"include": ["shared/models.yaml"]}, config_file_path=str(tmp_path / "config.yaml")
+    )
+
+    assert result == {"model_list": [{"model_name": "first"}, {"model_name": "second"}]}
+
+
 @pytest.mark.asyncio
 async def test_ProxyConfig__process_includes_merges_a_shared_file_once(tmp_path):
     (tmp_path / "shared.yaml").write_text("model_list:\n  - model_name: shared\n")
@@ -1110,7 +1124,6 @@ async def test_ProxyConfig_get_config_loads_from_file(tmp_path, monkeypatch):
 
 @pytest.mark.asyncio
 async def test_ProxyConfig_get_config_from_a_bucket_merges_includes(monkeypatch):
-    """A bucket-hosted config.yaml used to drop its `include:` entries silently (LIT-6982)."""
     objects = {
         "lit6982/config.yaml": {
             "include": ["model_config.yaml"],
@@ -1121,8 +1134,8 @@ async def test_ProxyConfig_get_config_from_a_bucket_merges_includes(monkeypatch)
     monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", None)
     monkeypatch.setattr("litellm.proxy.proxy_server.store_model_in_db", False)
     monkeypatch.setattr(
-        "litellm.proxy.common_utils.load_config_utils.get_file_contents_from_s3",
-        lambda bucket_name, object_key: objects.get(object_key),
+        "litellm.proxy.common_utils.load_config_utils.s3_object_reader",
+        lambda bucket_name: objects.get,
     )
     monkeypatch.setenv("LITELLM_CONFIG_BUCKET_NAME", "litellm-configs")
     monkeypatch.setenv("LITELLM_CONFIG_BUCKET_OBJECT_KEY", "lit6982/config.yaml")

From eca59aa90b9ea6b696040857bbc3b495e717ca84 Mon Sep 17 00:00:00 2001
From: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Date: Sun, 6 Sep 2026 03:35:07 -0700
Subject: [PATCH 4/4] fix(proxy): make an ambiguous config include loud, not
 silent

An include entry that matches both a file next to the config that declares it and
one next to the root config now warns naming both, so a config that resolves to a
different file than it used to says so instead of quietly serving other models.

Also from reviewing that change:

- an empty root object in a bucket fails the boot again instead of coming up empty
- a YAML syntax error in a bucket object logs its own line naming the object
- an include already loaded is skipped before it is read rather than after
- reading a config out of GCS builds the plain bucket client, so it needs no
  enterprise license and starts no flush loop that nothing ever cancels
---
 litellm/proxy/common_utils/config_includes.py | 59 ++++++++-----
 .../proxy/common_utils/load_config_utils.py   | 46 ++++++----
 litellm/proxy/proxy_server.py                 | 11 +--
 .../common_utils/test_load_config_utils.py    | 83 +++++++++++++++++++
 .../proxy/proxy_server/test_proxy_config.py   | 24 ++++++
 5 files changed, 183 insertions(+), 40 deletions(-)

diff --git a/litellm/proxy/common_utils/config_includes.py b/litellm/proxy/common_utils/config_includes.py
index 23e6c615a09..c1bb5ae952f 100644
--- a/litellm/proxy/common_utils/config_includes.py
+++ b/litellm/proxy/common_utils/config_includes.py
@@ -13,28 +13,41 @@ def resolve_include_file_path(include_file: str, declared_in: str, root_config_p
     Resolve one `include` entry to the file it names, next to the config that declares it.
 
     A config written before nested entries resolved this way can name a file sitting next to the root
-    config instead, so that file is still read, with a warning naming where it was found.
+    config instead, so that file is still read, with a warning naming where it was found. When both
+    files exist the one next to the declaring config wins and the other is named in a warning.
     """
     declared_relative: Final = os.path.abspath(os.path.join(os.path.dirname(declared_in), include_file))
-    if os.path.exists(declared_relative):
-        return declared_relative
-
     root_relative: Final = os.path.abspath(os.path.join(os.path.dirname(root_config_path), include_file))
     if root_relative == declared_relative or not os.path.exists(root_relative):
         return declared_relative
 
+    if not os.path.exists(declared_relative):
+        verbose_proxy_logger.warning(
+            "Config include '%s' declared in %s was not found next to it, so %s was read instead. "
+            "Move the included file next to the config that declares it.",
+            include_file,
+            declared_in,
+            root_relative,
+        )
+        return root_relative
+
     verbose_proxy_logger.warning(
-        "Config include '%s' declared in %s was not found next to it, so %s was read instead. "
-        "Move the included file next to the config that declares it.",
+        "Config include '%s' declared in %s matches two files. %s sits next to that config and was read, "
+        "so %s was skipped. Rename one of the two to say which one you meant.",
         include_file,
         declared_in,
+        declared_relative,
         root_relative,
     )
-    return root_relative
+    return declared_relative
 
 
-class ConfigLoader(Protocol):
-    def __call__(self, include_entry: str, declared_in: str, /) -> Awaitable[tuple[str, Mapping[str, object]]]: ...
+class IncludeResolver(Protocol):
+    def __call__(self, include_entry: str, declared_in: str, /) -> str: ...
+
+
+class ConfigReader(Protocol):
+    def __call__(self, location: str, /) -> Awaitable[Mapping[str, object]]: ...
 
 
 def _merged_value(base_value: object, included_value: object) -> object:
@@ -80,32 +93,40 @@ async def _resolve(
     config: Mapping[str, object],
     pending: tuple[tuple[str, str], ...],
     loaded: frozenset[str],
-    load: ConfigLoader,
+    resolve: IncludeResolver,
+    read: ConfigReader,
 ) -> Mapping[str, object]:
     if not pending:
         return _without_include(config)
 
     entry, declared_in = pending[0]
-    location, included = await load(entry, declared_in)
+    location: Final = resolve(entry, declared_in)
     if location in loaded:
-        return await _resolve(config, pending[1:], loaded, load)
+        return await _resolve(config, pending[1:], loaded, resolve, read)
 
+    included: Final = await read(location)
     return await _resolve(
         _merged(config, _without_include(included)),
         (*pending[1:], *_pending_from(included, location)),
         loaded | frozenset((location,)),
-        load,
+        resolve,
+        read,
     )
 
 
-async def resolve_includes(config: Mapping[str, object], location: str, load: ConfigLoader) -> dict[str, object]:
+async def resolve_includes(
+    config: Mapping[str, object],
+    location: str,
+    resolve: IncludeResolver,
+    read: ConfigReader,
+) -> dict[str, object]:
     """
     Merge every config named by the `include` directive into the config that declares it.
 
-    List values are extended and every other value is overridden, each entry is resolved relative to
-    the config that declares it, a config already pulled in is not merged a second time, and `load`
-    decides where an entry is read from, so the same merge applies to configs on disk and to configs
-    hosted in a bucket.
+    List values are extended and every other value is overridden, `resolve` turns each entry into the
+    location it names relative to the config that declares it, a config already pulled in is neither
+    read nor merged a second time, and `read` decides where a location is read from, so the same merge
+    applies to configs on disk and to configs hosted in a bucket.
     """
-    merged: Final = await _resolve(config, _pending_from(config, location), frozenset((location,)), load)
+    merged: Final = await _resolve(config, _pending_from(config, location), frozenset((location,)), resolve, read)
     return dict(merged)  # mutable-ok: the proxy mutates the config it loads
diff --git a/litellm/proxy/common_utils/load_config_utils.py b/litellm/proxy/common_utils/load_config_utils.py
index d62286a50f6..4a082eb307b 100644
--- a/litellm/proxy/common_utils/load_config_utils.py
+++ b/litellm/proxy/common_utils/load_config_utils.py
@@ -29,8 +29,12 @@ class SyncBucketObjectReader(Protocol):
     def __call__(self, object_key: str, /) -> object | None: ...
 
 
-def _parsed_config(file_contents: str) -> object:
-    parsed: Final = yaml.safe_load(file_contents)
+def _parsed_config(object_key: str, file_contents: str) -> object | None:
+    try:
+        parsed: Final = yaml.safe_load(file_contents)
+    except yaml.YAMLError as e:
+        verbose_proxy_logger.error("Config object %s is not valid YAML: %s", object_key, e)
+        return None
     return MappingProxyType({}) if parsed is None else parsed
 
 
@@ -64,11 +68,13 @@ def s3_object_reader(bucket_name: str) -> SyncBucketObjectReader:
         try:
             verbose_proxy_logger.debug("Retrieving %s from S3 bucket: %s", object_key, bucket_name)
             response: Final = s3_client.get_object(Bucket=bucket_name, Key=object_key)
-            return _parsed_config(response["Body"].read().decode("utf-8"))
+            file_contents: Final = response["Body"].read().decode("utf-8")
         except Exception as e:  # noqa: BLE001  # any boto3 error must read as a missing object
             verbose_proxy_logger.error("Error retrieving %s from S3 bucket %s: %s", object_key, bucket_name, e)
             return None
 
+        return _parsed_config(object_key, file_contents)
+
     return read
 
 
@@ -77,10 +83,16 @@ def get_file_contents_from_s3(bucket_name: str, object_key: str) -> object | Non
 
 
 def gcs_config_bucket(bucket_name: str) -> "GCSBucketBase | None":
-    try:
-        from litellm.integrations.gcs_bucket.gcs_bucket import GCSBucketLogger
+    """
+    Build a plain GCS client for reading config objects.
 
-        return GCSBucketLogger(bucket_name=bucket_name)
+    Reading a config out of a bucket is not GCS logging, so it neither needs the enterprise license
+    that gate covers nor the batching task the logger starts and never stops.
+    """
+    try:
+        from litellm.integrations.gcs_bucket.gcs_bucket_base import GCSBucketBase
+
+        return GCSBucketBase(bucket_name=bucket_name)
     except Exception as e:  # noqa: BLE001  # an unbuildable client must read as an unreadable bucket
         verbose_proxy_logger.error("Error creating the GCS client for bucket %s: %s", bucket_name, e)
         return None
@@ -98,12 +110,14 @@ async def get_config_file_contents_from_gcs(
         file_contents: Final = await bucket.download_gcs_object(object_key)
         if file_contents is None:
             raise Exception(f"File contents are None for {object_key}")
-        return _parsed_config(file_contents.decode("utf-8"))
+        decoded: Final = file_contents.decode("utf-8")
 
     except Exception as e:
         verbose_proxy_logger.error("Error retrieving %s from GCS bucket %s: %s", object_key, bucket_name, e)
         return None
 
+    return _parsed_config(object_key, decoded)
+
 
 def resolve_include_object_key(config_object_key: str, include_entry: str) -> str:
     """
@@ -123,17 +137,19 @@ async def resolve_bucket_includes(
     object_key: str,
     fetch: BucketObjectFetcher,
 ) -> dict[str, object]:
-    async def load(include_entry: str, declared_in: str) -> tuple[str, Mapping[str, object]]:
-        include_key: Final = resolve_include_object_key(declared_in, include_entry)
+    async def read(include_key: str) -> Mapping[str, object]:
         included: Final = await fetch(include_key)
         if included is None:
             raise FileNotFoundError(
                 f"Included config could not be read from bucket: {include_key}. "
                 "The underlying bucket error is logged above."
             )
-        return include_key, included
+        return included
 
-    return await resolve_includes(config=config, location=object_key, load=load)
+    def resolve(include_entry: str, declared_in: str) -> str:
+        return resolve_include_object_key(declared_in, include_entry)
+
+    return await resolve_includes(config=config, location=object_key, resolve=resolve, read=read)
 
 
 async def bucket_object_reader(bucket_type: str | None, bucket_name: str) -> BucketObjectReader:
@@ -176,7 +192,7 @@ async def get_config_from_bucket(
             raise ValueError(f"Config object in bucket is not a YAML mapping: {key}") from e
 
     config: Final = await fetch(object_key)
-    if config is None:
+    if not config:
         return None
 
     return await resolve_bucket_includes(config=config, object_key=object_key, fetch=fetch)
@@ -256,11 +272,9 @@ async def download_python_file_from_gcs(
         bool: True if successful, False otherwise
     """
     try:
-        from litellm.integrations.gcs_bucket.gcs_bucket import GCSBucketLogger
+        from litellm.integrations.gcs_bucket.gcs_bucket_base import GCSBucketBase
 
-        gcs_bucket: Final = GCSBucketLogger(
-            bucket_name=bucket_name,
-        )
+        gcs_bucket: Final = GCSBucketBase(bucket_name=bucket_name)
         file_contents = await gcs_bucket.download_gcs_object(object_key)
         if file_contents is None:
             raise Exception(f"File contents are None for {object_key}")
diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py
index 7fada09febf..ae9cd71a956 100644
--- a/litellm/proxy/proxy_server.py
+++ b/litellm/proxy/proxy_server.py
@@ -4559,17 +4559,18 @@ class ProxyConfig:
 
         included_config_adapter: Final = TypeAdapter(dict[str, object])
 
-        async def load_included(include_file: str, declared_in: str) -> tuple[str, Mapping[str, object]]:
-            file_path: Final = resolve_include_file_path(include_file, declared_in, config_file_path)
+        def resolve(include_file: str, declared_in: str) -> str:
+            return resolve_include_file_path(include_file, declared_in, config_file_path)
+
+        async def read_included(file_path: str) -> Mapping[str, object]:
             if not os.path.exists(file_path):
                 raise FileNotFoundError(f"Included file not found: {file_path}")
             try:
-                included: Final = included_config_adapter.validate_python(self._load_yaml_file(file_path))
+                return included_config_adapter.validate_python(self._load_yaml_file(file_path))
             except ValidationError as e:
                 raise ValueError(f"Included config file is not a YAML mapping: {file_path}") from e
-            return file_path, included
 
-        return await resolve_includes(config=config, location=config_file_path, load=load_included)
+        return await resolve_includes(config=config, location=config_file_path, resolve=resolve, read=read_included)
 
     async def save_config(self, new_config: dict, include_env_vars: bool = False):
         global prisma_client, general_settings, user_config_file_path, store_model_in_db
diff --git a/tests/test_litellm/proxy/common_utils/test_load_config_utils.py b/tests/test_litellm/proxy/common_utils/test_load_config_utils.py
index 75aa5a3f0f8..1042dbe9653 100644
--- a/tests/test_litellm/proxy/common_utils/test_load_config_utils.py
+++ b/tests/test_litellm/proxy/common_utils/test_load_config_utils.py
@@ -1,4 +1,5 @@
 import asyncio
+import logging
 import re
 import threading
 from unittest.mock import MagicMock, mock_open, patch
@@ -7,6 +8,7 @@ import pytest
 import yaml
 
 from litellm.proxy.common_utils.load_config_utils import (
+    gcs_config_bucket,
     get_config_from_bucket,
     get_file_contents_from_s3,
     resolve_bucket_includes,
@@ -391,3 +393,84 @@ class TestBucketConfigIncludes:
             )
             is None
         )
+
+    @pytest.mark.asyncio
+    async def test_an_object_pulled_in_twice_is_read_once(self):
+        objects = {
+            "configs/a.yaml": {"include": ["shared.yaml"]},
+            "configs/b.yaml": {"include": ["./shared.yaml"]},
+            "configs/shared.yaml": {"model_list": [{"model_name": "shared"}]},
+        }
+        requested = []
+
+        async def fetch(object_key):
+            requested.append(object_key)
+            return objects.get(object_key)
+
+        await resolve_bucket_includes(
+            config={"include": ["a.yaml", "b.yaml"]},
+            object_key="configs/config.yaml",
+            fetch=fetch,
+        )
+
+        assert requested == ["configs/a.yaml", "configs/b.yaml", "configs/shared.yaml"]
+
+    @pytest.mark.asyncio
+    async def test_an_empty_root_object_does_not_boot_an_empty_proxy(self, monkeypatch):
+        class FakeGCSBucket:
+            async def download_gcs_object(self, object_key):
+                return b""
+
+        monkeypatch.setattr(
+            "litellm.proxy.common_utils.load_config_utils.gcs_config_bucket",
+            lambda bucket_name: FakeGCSBucket(),
+        )
+
+        config = await get_config_from_bucket(
+            bucket_type="gcs", bucket_name="litellm-configs", object_key="lit6982/config.yaml"
+        )
+
+        assert config is None
+
+    @pytest.mark.asyncio
+    async def test_an_object_that_is_not_valid_yaml_is_reported_as_a_yaml_error(self, monkeypatch, caplog):
+        class FakeGCSBucket:
+            async def download_gcs_object(self, object_key):
+                return b"model_list: [\n"
+
+        monkeypatch.setattr(
+            "litellm.proxy.common_utils.load_config_utils.gcs_config_bucket",
+            lambda bucket_name: FakeGCSBucket(),
+        )
+
+        with caplog.at_level(logging.ERROR, logger="LiteLLM Proxy"):
+            config = await get_config_from_bucket(
+                bucket_type="gcs", bucket_name="litellm-configs", object_key="lit6982/config.yaml"
+            )
+
+        assert config is None
+        assert [
+            record
+            for record in caplog.records
+            if "not valid YAML" in record.getMessage() and "lit6982/config.yaml" in record.getMessage()
+        ]
+
+
+class TestGCSConfigBucketClient:
+    @pytest.mark.asyncio
+    async def test_reading_a_config_from_gcs_does_not_need_an_enterprise_license(self, monkeypatch):
+        monkeypatch.setattr("litellm.proxy.proxy_server.premium_user", False)
+
+        bucket = gcs_config_bucket("litellm-configs")
+
+        assert bucket is not None
+        assert bucket.BUCKET_NAME == "litellm-configs"
+
+    @pytest.mark.asyncio
+    async def test_reading_a_config_from_gcs_starts_no_background_task(self, monkeypatch):
+        monkeypatch.setattr("litellm.proxy.proxy_server.premium_user", True)
+        running_before = asyncio.all_tasks()
+
+        gcs_config_bucket("litellm-configs")
+
+        assert asyncio.all_tasks() - running_before == set()
diff --git a/tests/test_litellm/proxy/proxy_server/test_proxy_config.py b/tests/test_litellm/proxy/proxy_server/test_proxy_config.py
index e90f101a555..fe64bcc2c10 100644
--- a/tests/test_litellm/proxy/proxy_server/test_proxy_config.py
+++ b/tests/test_litellm/proxy/proxy_server/test_proxy_config.py
@@ -10,6 +10,7 @@ from __future__ import annotations
 
 import asyncio
 import json
+import logging
 import os
 import re
 from types import SimpleNamespace
@@ -772,6 +773,29 @@ async def test_ProxyConfig__process_includes_still_reads_a_nested_include_left_b
     assert result == {"model_list": [{"model_name": "first"}, {"model_name": "second"}]}
 
 
+@pytest.mark.asyncio
+async def test_ProxyConfig__process_includes_names_both_files_when_a_nested_include_matches_two(tmp_path, caplog):
+    (tmp_path / "shared").mkdir()
+    (tmp_path / "shared" / "models.yaml").write_text(
+        "include:\n  - more_models.yaml\nmodel_list:\n  - model_name: first\n"
+    )
+    (tmp_path / "shared" / "more_models.yaml").write_text("model_list:\n  - model_name: next-to-the-declaring-file\n")
+    (tmp_path / "more_models.yaml").write_text("model_list:\n  - model_name: next-to-the-root-config\n")
+
+    with caplog.at_level(logging.WARNING, logger="LiteLLM Proxy"):
+        result = await ProxyConfig()._process_includes(
+            {"include": ["shared/models.yaml"]}, config_file_path=str(tmp_path / "config.yaml")
+        )
+
+    assert result == {"model_list": [{"model_name": "first"}, {"model_name": "next-to-the-declaring-file"}]}
+    assert [
+        record
+        for record in caplog.records
+        if str(tmp_path / "shared" / "more_models.yaml") in record.getMessage()
+        and str(tmp_path / "more_models.yaml") in record.getMessage()
+    ]
+
+
 @pytest.mark.asyncio
 async def test_ProxyConfig__process_includes_merges_a_shared_file_once(tmp_path):
     (tmp_path / "shared.yaml").write_text("model_list:\n  - model_name: shared\n")