fix(bedrock): treat blank AWS_S3_* env vars as unset for batch jobs (#42528)

* test(e2e): pin bedrock batch create with blank S3 env vars

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(bedrock): treat blank S3 env vars as unset for batch jobs

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(e2e): trim blank S3 env gateway config

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(e2e): register blank_s3_env capability and clean gateway tempdir

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(e2e): move blank S3 env batch test to its own module

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: yucheng <yucheng@berri.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/llms/bedrock/common_utils.py

@@ -1593,7 +1593,7 @@ def _resolve_s3_setting(
         source.get(param_name) for source in (litellm_params, optional_params) if source is not None
     )
     explicit: Final = next((value for value in candidates if isinstance(value, str) and value), None)
-    return explicit or get_secret_str(env_var)
+    return explicit or get_secret_str(env_var) or None
 
 
 class CommonBatchFilesUtils:

tests/e2e/batches/COVERAGE.md

@@ -22,6 +22,7 @@ failures are hard test failures (see `tests/e2e/AGENTS.md`).
 | Bedrock   | yes (unified only) | yes | yes | yes (unfiltered managed list) | yes (provider-transformed) | S3 (`s3_bucket_name` + `aws_*` + `AWS_BATCH_ROLE_ARN` on model) |
 | Bedrock GovCloud (`us-gov-west-1`) | yes (unified only) | yes | no | no | yes (provider-transformed) | S3 (`s3_bucket_name` + `aws_*` on model, resolved from `AWS_GOVCLOUD_ACCESS_KEY_ID` / `AWS_GOVCLOUD_SECRET_ACCESS_KEY` / `AWS_GOVCLOUD_BATCH_S3_BUCKET` / `AWS_GOVCLOUD_BATCH_ROLE_ARN`) |
 | Bedrock split S3 identity | no | no | no | no | yes (file upload, content, delete) | S3 signed with `s3_access_key_id` / `s3_secret_access_key` (`AWS_S3_ONLY_ACCESS_KEY_ID` / `AWS_S3_ONLY_SECRET_ACCESS_KEY`, object rights on `AWS_BATCH_S3_BUCKET` only) while `aws_*` is `AWS_BEDROCK_ONLY_ACCESS_KEY_ID` / `AWS_BEDROCK_ONLY_SECRET_ACCESS_KEY`, an identity with no S3 rights on that bucket |
+| Bedrock blank S3 env | yes (unified only, on an owned gateway exporting `AWS_S3_ENCRYPTION_KEY_ID` / `AWS_S3_BUCKET_OWNER` as empty strings) | no | no | no | no | S3 (`s3_bucket_name` + `aws_*` + `AWS_BATCH_ROLE_ARN` in the gateway config); blank env vars must be treated as unset, not serialized |
 
 Bedrock cancel maps to `StopModelInvocationJob` and comes back `cancelling`; the
 lifecycle asserts it the same way it does for OpenAI (`_CANCEL_ASSERTED_PROVIDERS`).

tests/e2e/batches/bedrock_env_gateway.py

@@ -0,0 +1,145 @@
+"""An owned, source-built proxy whose process env exports AWS_S3_* vars blank.
+
+The shared fixture proxy inherits the harness env, which cannot reproduce a user
+shell that exports AWS_S3_ENCRYPTION_KEY_ID / AWS_S3_BUCKET_OWNER as empty
+strings. This gateway boots a second proxy with both vars present but blank, so
+a batch create through it proves blank means unset, not an empty string.
+"""
+
+from __future__ import annotations
+
+import os
+import shutil
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+from collections.abc import Mapping
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Final
+
+from e2e_config import unique_marker
+from e2e_http import NoBody
+from idp import stop_process_group
+from proxy_client import ProxyClient, build_proxy_client
+from pydantic import TypeAdapter
+
+STARTUP_TIMEOUT_SECONDS: Final = 240
+LOG_TAIL_BYTES: Final = 4000
+REPO_ROOT: Final = Path(__file__).resolve().parents[3]
+
+_CONFIG_YAML: Final = """model_list:
+  - model_name: bedrock-blank-s3-batch
+    litellm_params:
+      model: bedrock/us.anthropic.claude-haiku-4-5-20251001-v1:0
+      aws_access_key_id: os.environ/AWS_ACCESS_KEY_ID
+      aws_secret_access_key: os.environ/AWS_SECRET_ACCESS_KEY
+      aws_region_name: os.environ/AWS_REGION
+      s3_region_name: os.environ/AWS_REGION
+      s3_bucket_name: os.environ/AWS_BATCH_S3_BUCKET
+      s3_access_key_id: os.environ/AWS_ACCESS_KEY_ID
+      s3_secret_access_key: os.environ/AWS_SECRET_ACCESS_KEY
+      aws_batch_role_arn: os.environ/AWS_BATCH_ROLE_ARN
+
+general_settings:
+  master_key: os.environ/LITELLM_MASTER_KEY
+  database_url: os.environ/DATABASE_URL
+"""
+
+
+def available_port() -> int:
+    with socket.socket() as listener:
+        listener.bind(("127.0.0.1", 0))
+        return TypeAdapter(tuple[str, int]).validate_python(listener.getsockname())[1]
+
+
+@dataclass(slots=True)
+class BedrockEnvGateway:
+    base_url: str
+    master_key: str
+    proxy: ProxyClient
+    _environment: Mapping[str, str] = field(repr=False)
+    _command: tuple[str, ...] = field(repr=False)
+    _log_path: Path
+    _child: subprocess.Popen[bytes] | None = field(default=None, init=False, repr=False)
+
+    @classmethod
+    def start(cls) -> BedrockEnvGateway:
+        assert os.environ.get("DATABASE_URL"), "DATABASE_URL is required for the blank-S3-env gateway"
+        port: Final = available_port()
+        base_url: Final = f"http://127.0.0.1:{port}"
+        master_key: Final = f"sk-e2e-blank-s3-{unique_marker()}"
+        directory: Final = Path(tempfile.mkdtemp(prefix="litellm-e2e-blank-s3-"))
+        config: Final = directory / "blank-s3-gateway.yaml"
+        config.write_text(_CONFIG_YAML)
+        environment: Final = {
+            **{key: value for key, value in os.environ.items() if not key.startswith("REDIS_")},
+            "DATABASE_URL": os.environ["DATABASE_URL"],
+            "LITELLM_MASTER_KEY": master_key,
+            "STORE_MODEL_IN_DB": "False",
+            "PYTHONPATH": str(REPO_ROOT),
+            "AWS_S3_ENCRYPTION_KEY_ID": "",
+            "AWS_S3_BUCKET_OWNER": "",
+        }
+        gateway: Final = cls(
+            base_url=base_url,
+            master_key=master_key,
+            proxy=build_proxy_client(
+                base_url=base_url,
+                control_plane_base_url=base_url,
+                replica_urls=(base_url,),
+                master_key=master_key,
+            ),
+            _environment=environment,
+            _command=(
+                sys.executable,
+                "-m",
+                "litellm.proxy.proxy_cli",
+                "--config",
+                str(config),
+                "--port",
+                str(port),
+                "--host",
+                "127.0.0.1",
+            ),
+            _log_path=directory / "blank-s3-gateway.log",
+        )
+        with gateway._log_path.open("ab") as log:
+            gateway._child = subprocess.Popen(
+                gateway._command,
+                env=dict(gateway._environment),
+                stdout=log,
+                stderr=log,
+                start_new_session=True,
+                cwd=REPO_ROOT,
+            )
+        deadline: Final = time.monotonic() + STARTUP_TIMEOUT_SECONDS
+        while time.monotonic() < deadline:
+            assert gateway._child.poll() is None, (
+                f"blank-S3-env gateway exited early; log tail:\n{gateway.log_tail()}"
+            )
+            result = gateway.proxy.transport.probe("/health/liveliness", params=NoBody())
+            if result.status_code == 200:
+                return gateway
+            time.sleep(0.5)
+        tail: Final = gateway.log_tail()
+        gateway.stop()
+        raise AssertionError(
+            f"blank-S3-env gateway did not become ready in {STARTUP_TIMEOUT_SECONDS}s; log tail:\n{tail}"
+        )
+
+    def log_tail(self) -> str:
+        if not self._log_path.exists():
+            return "<no log written>"
+        with self._log_path.open("rb") as log:
+            log.seek(0, 2)
+            size: Final = log.tell()
+            log.seek(max(0, size - LOG_TAIL_BYTES))
+            return log.read().decode("utf-8", errors="replace")
+
+    def stop(self) -> None:
+        if self._child is not None:
+            stop_process_group(self._child)
+        shutil.rmtree(self._log_path.parent, ignore_errors=True)

tests/e2e/batches/test_bedrock_blank_s3_env_e2e.py

@@ -0,0 +1,109 @@
+"""Live e2e pin for Bedrock batch create with blank AWS_S3_* env vars.
+
+Owns its own file (not test_batches_e2e.py) so the PR changed-file e2e gate
+stays a single tiny file: this class boots its own gateway with
+AWS_S3_ENCRYPTION_KEY_ID and AWS_S3_BUCKET_OWNER exported empty, then runs the
+unified target_model_names upload + batch create lifecycle against real Bedrock.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Final
+
+import pytest
+from batch_cleanup import cleanup_batch, cleanup_file
+from batch_client import BatchClient, BatchCreateBody, BatchObject, FileObject
+from bedrock_env_gateway import BedrockEnvGateway
+from capabilities import is_managed_id
+from e2e_http import FileUploadForm, require_successful_call, unwrap
+from lifecycle import ResourceManager
+from models import KeyGenerateBody
+
+pytestmark = pytest.mark.e2e
+
+CREATED_BATCH_STATUSES = {"validating", "in_progress", "finalizing"}
+BLANK_S3_RAW_MODEL: Final = "bedrock/us.anthropic.claude-haiku-4-5-20251001-v1:0"
+
+
+def render_jsonl(model: str) -> bytes:
+    line = {
+        "custom_id": "req-1",
+        "method": "POST",
+        "url": "/v1/chat/completions",
+        "body": {
+            "model": model,
+            "messages": [{"role": "user", "content": "ping"}],
+            "max_tokens": 8,
+        },
+    }
+    return (json.dumps(line) + "\n").encode()
+
+
+def assert_file_object(file: FileObject, *, provider: str) -> None:
+    assert file.object == "file", f"file.object={file.object!r}"
+    assert file.purpose == "batch", f"file.purpose={file.purpose!r}"
+    assert file.bytes is not None, f"file.bytes={file.bytes!r}"
+    if provider != "bedrock":
+        assert file.bytes > 0, f"file.bytes={file.bytes!r}"
+    assert file.status, "file.status missing"
+    assert file.created_at is not None and file.created_at > 0, "file.created_at missing"
+
+
+def assert_batch_object(batch: BatchObject) -> None:
+    assert batch.object == "batch", f"batch.object={batch.object!r}"
+    if batch.endpoint:
+        assert batch.endpoint == "/v1/chat/completions", f"batch.endpoint={batch.endpoint!r}"
+    assert batch.completion_window == "24h", f"window={batch.completion_window!r}"
+    assert batch.input_file_id, "batch.input_file_id missing"
+    assert batch.created_at is not None and batch.created_at > 0, "batch.created_at missing"
+
+
+class TestBedrockBatchBlankS3EnvVars:
+    """Bedrock batch create with AWS_S3_* env vars exported but blank.
+
+    Regression: a blank AWS_S3_ENCRYPTION_KEY_ID or AWS_S3_BUCKET_OWNER env var
+    resolved to "" and was serialized into the create-job request, which Bedrock
+    rejects. The owned gateway exports both vars empty, so the unified lifecycle
+    only passes when blank is treated as unset.
+    """
+
+    @pytest.mark.covers(
+        "llm.batches.bedrock.blank_s3_env.nonstream.works",
+        "llm.files.bedrock.upload.nonstream.works",
+        exercised_on=["batches", "files"],
+    )
+    def test_unified_batch_create_ignores_blank_s3_env_vars(self, resources: ResourceManager) -> None:
+        gateway: Final = BedrockEnvGateway.start()
+        resources.defer(gateway.stop)
+        client: Final = BatchClient(proxy=gateway.proxy)
+
+        key: Final = client.proxy.generate_key(KeyGenerateBody(models=[], user_id="e2e-test-user"))
+        resources.defer(lambda: client.proxy.delete_key(key))
+
+        file: Final = unwrap(
+            client.upload_file(
+                content=render_jsonl(BLANK_S3_RAW_MODEL),
+                form=FileUploadForm(purpose="batch", target_model_names="bedrock-blank-s3-batch"),
+                key=key,
+            )
+        )
+        resources.defer(lambda: cleanup_file(client, file.id, key=key))
+        assert_file_object(file, provider="bedrock")
+
+        created: Final = client.create_batch(body=BatchCreateBody(input_file_id=file.id), key=key)
+        assert created.status_code < 400, (
+            f"blank AWS_S3_ENCRYPTION_KEY_ID / AWS_S3_BUCKET_OWNER must be treated as "
+            f"unset; Bedrock rejected the job: {created.body[:400]}"
+        )
+        require_successful_call(created)
+        batch: Final = BatchObject.model_validate_json(created.body)
+        resources.defer(lambda: cleanup_batch(client, batch.id, key=key))
+
+        assert is_managed_id(batch.id), (
+            f"blank-S3-env create via target_model_names must return a managed batch id, got {batch.id!r}"
+        )
+        assert batch.status in CREATED_BATCH_STATUSES, (
+            f"blank-S3-env batch has non-transitional status {batch.status!r}"
+        )
+        assert_batch_object(batch)

tests/e2e/coverage_registry/llm_nonconversational.yaml

@@ -24,6 +24,7 @@
 - {id: llm.batches.bedrock.basic.nonstream.works, module: llm, tier: P0, subject_endpoint: batches, route: bedrock_converse, capability: basic, streaming: nonstream, assertions: [works], source: "batches/capabilities.py:98", rationale: "Bedrock batches (encoded/unified only)"}
 - {id: llm.batches.bedrock.assume_role.nonstream.works, module: llm, tier: P0, subject_endpoint: batches, route: bedrock_converse, capability: assume_role, streaming: nonstream, assertions: [works], source: "test_batches_e2e.py", rationale: "Bedrock batch create under STS assume-role credentials"}
 - {id: llm.batches.bedrock.govcloud_partition.nonstream.works, module: llm, tier: P0, subject_endpoint: batches, route: bedrock_converse, capability: govcloud_partition, streaming: nonstream, assertions: [works], source: "test_batches_e2e.py", rationale: "Bedrock batch create in the us-gov-west-1 partition"}
+- {id: llm.batches.bedrock.blank_s3_env.nonstream.works, module: llm, tier: P0, subject_endpoint: batches, route: bedrock_converse, capability: blank_s3_env, streaming: nonstream, assertions: [works], source: "test_bedrock_blank_s3_env_e2e.py", rationale: "Bedrock batch create treats blank AWS_S3_ENCRYPTION_KEY_ID / AWS_S3_BUCKET_OWNER env vars as unset instead of serializing empty strings"}
 - {id: llm.batches.bedrock.cancel.nonstream.works, module: llm, tier: P0, subject_endpoint: batches, route: bedrock_converse, capability: basic, streaming: nonstream, assertions: [works], source: "test_batches_e2e.py", rationale: "Bedrock batch cancel (StopModelInvocationJob) returns the same id with a cancelling/cancelled status"}
 - {id: llm.batches.bedrock.list.nonstream.works, module: llm, tier: P0, subject_endpoint: batches, route: bedrock_converse, capability: basic, streaming: nonstream, assertions: [works], source: "test_batches_e2e.py", rationale: "A Bedrock managed batch is present in the GET /v1/batches list envelope"}
 - {id: llm.batches.hosted_vllm.basic.nonstream.works, module: llm, tier: P1, subject_endpoint: batches, route: hosted_vllm, capability: basic, streaming: nonstream, assertions: [works], source: "test_batches_e2e.py", rationale: "hosted_vllm OpenAI-compatible batch create"}

tests/e2e/coverage_registry/schema.py

@@ -65,6 +65,7 @@ LlmCapability = Literal[
     "assume_role",
     "basic",
     "batch_deployment",
+    "blank_s3_env",
     "count_tokens",
     "govcloud_partition",
     "split_s3_credentials",

tests/test_litellm/llms/bedrock/batches/test_transformation.py

@@ -19,9 +19,8 @@ from unittest.mock import MagicMock, patch
 import httpx
 import pytest
 
-
 from litellm.llms.bedrock.batches.transformation import BedrockBatchesConfig
-from litellm.types.utils import LiteLLMBatch, LlmProviders
+from litellm.types.utils import LlmProviders
 
 # AWS JobStatus -> OpenAI BatchJobStatus, exactly as encoded in transformation.py
 # (both transform_create_batch_response and transform_retrieve_batch_response).
@@ -270,6 +269,44 @@ def test_create_request_keeps_kms_key_alongside_s3_bucket_owner(config, monkeypa
     }
 
 
+def test_create_request_omits_kms_key_when_env_var_is_blank(config, monkeypatch):
+    monkeypatch.setenv("AWS_S3_ENCRYPTION_KEY_ID", "")
+    monkeypatch.delenv("AWS_S3_BUCKET_OWNER", raising=False)
+
+    bedrock_request = _signed_batch_request(config, {}, {})
+
+    assert bedrock_request["outputDataConfig"] == {
+        "s3OutputDataConfig": {"s3Uri": "s3://in-bucket/litellm-batch-outputs/litellm-batch-1/"}
+    }
+
+
+def test_create_request_omits_s3_bucket_owner_when_env_var_is_blank(config, monkeypatch):
+    monkeypatch.setenv("AWS_S3_BUCKET_OWNER", "")
+    monkeypatch.delenv("AWS_S3_ENCRYPTION_KEY_ID", raising=False)
+
+    bedrock_request = _signed_batch_request(config, {}, {})
+
+    assert bedrock_request["inputDataConfig"] == {"s3InputDataConfig": {"s3Uri": "s3://in-bucket/in.jsonl"}}
+    assert bedrock_request["outputDataConfig"] == {
+        "s3OutputDataConfig": {"s3Uri": "s3://in-bucket/litellm-batch-outputs/litellm-batch-1/"}
+    }
+
+
+def test_create_request_emits_real_values_alongside_blank_sibling_env_var(config, monkeypatch):
+    monkeypatch.setenv("AWS_S3_ENCRYPTION_KEY_ID", "kms-key-123")
+    monkeypatch.setenv("AWS_S3_BUCKET_OWNER", "")
+
+    bedrock_request = _signed_batch_request(config, {}, {})
+
+    assert bedrock_request["inputDataConfig"] == {"s3InputDataConfig": {"s3Uri": "s3://in-bucket/in.jsonl"}}
+    assert bedrock_request["outputDataConfig"] == {
+        "s3OutputDataConfig": {
+            "s3Uri": "s3://in-bucket/litellm-batch-outputs/litellm-batch-1/",
+            "s3EncryptionKeyId": "kms-key-123",
+        }
+    }
+
+
 def test_create_request_missing_input_file_id_raises(config):
     with pytest.raises(ValueError, match="input_file_id is required"):
         config.transform_create_batch_request(