fix(vertex_ai): support fine-tuned Gemini endpoints in managed batches

Managed batches mangled any Vertex model that is not a plain publisher
model: a fine-tuned Gemini endpoint id was filed under
publishers/google/models/gemini/<id> at upload, then the batch create
parse dropped the id and targeted the nonexistent publisher model
'publishers/google/models/gemini', which Vertex rejects.

Fine-tuned endpoints are now stored under endpoints/<id> in the GCS
object path, and batch create resolves the endpoint to its deployed
tuned model resource (projects/../models/<id>) via GET endpoints/<id>,
which is the only form the v1 batch API accepts for tuned models. The
cost poller's bare-model parse round-trips the endpoint id so unmanaged
batch spend still maps to the configured deployment.

custom_endpoint deployments have no Vertex batch surface, so batch file
uploads and batch creation against them now return a clear 400 instead
of creating a doomed job.

Resolves LIT-6899

litellm/batches/main.py

@@ -301,6 +301,21 @@ def create_batch(
                 litellm_params=litellm_params,
             )
         elif custom_llm_provider == "vertex_ai":
+            if optional_params.get("custom_endpoint"):
+                raise litellm.exceptions.BadRequestError(
+                    message=(
+                        "Vertex AI batch prediction is not supported for `custom_endpoint` deployments. "
+                        "The OpenAI-compatible custom endpoint path has no batch surface in LiteLLM; "
+                        "use a publisher model or fine-tuned Gemini endpoint deployment instead."
+                    ),
+                    model=model or "n/a",
+                    llm_provider=custom_llm_provider,
+                    response=httpx.Response(
+                        status_code=400,
+                        content="custom_endpoint deployments do not support vertex_ai batches",
+                        request=httpx.Request(method="create_batch", url="https://github.com/BerriAI/litellm"),
+                    ),
+                )
             api_base = optional_params.api_base or ""
             vertex_ai_project: Final = (
                 optional_params.vertex_project or litellm.vertex_project or get_secret_str("VERTEXAI_PROJECT")

litellm/llms/vertex_ai/batches/handler.py

@@ -12,6 +12,7 @@ from litellm.litellm_core_utils.url_utils import (
     safe_get,
 )
 from litellm.llms.custom_httpx.http_handler import (
+    HTTPHandler,
     _get_httpx_client,
     get_async_httpx_client,
 )
@@ -55,6 +56,20 @@ class _FetchedResponseView(TypedDict):
     response: ReadOnly[httpx.Response]
 
 
+class _VertexEndpointDeployedModel(TypedDict, total=False):
+    model: ReadOnly[str]
+
+
+class _VertexEndpointResponse(TypedDict, total=False):
+    deployedModels: ReadOnly[list[_VertexEndpointDeployedModel]]
+
+
+class _VertexEndpointPayloadView(TypedDict):
+    """Holds one decoded GET endpoints/<id> response so the payload reads back typed."""
+
+    payload: ReadOnly[_VertexEndpointResponse]
+
+
 def _vertex_batch_payload(response: _VertexBatchJsonSource) -> VertexBatchPredictionResponse:
     return response.json()
 
@@ -116,11 +131,19 @@ class VertexAIBatchPrediction(VertexLLM):
             "Authorization": f"Bearer {access_token}",
         }
 
-        vertex_batch_request: Final[VertexAIBatchPredictionJob] = (
+        transformed_batch_request: Final[VertexAIBatchPredictionJob] = (
             VertexAIBatchTransformation.transform_openai_batch_request_to_vertex_ai_batch_request(
-                request=create_batch_data
+                request=create_batch_data,
+                vertex_project=vertex_project or project_id,
+                vertex_location=vertex_location or "us-central1",
             )
         )
+        vertex_batch_request: Final = self._resolve_fine_tuned_endpoint_model(
+            vertex_batch_request=transformed_batch_request,
+            headers=headers,
+            sync_handler=sync_handler,
+            vertex_location=vertex_location or "us-central1",
+        )
 
         if _is_async is True:
             return self._async_create_batch(
@@ -142,6 +165,43 @@ class VertexAIBatchPrediction(VertexLLM):
         )
         return vertex_batch_response
 
+    def _resolve_fine_tuned_endpoint_model(
+        self,
+        vertex_batch_request: VertexAIBatchPredictionJob,
+        headers: dict[str, str],
+        sync_handler: HTTPHandler,
+        vertex_location: str,
+    ) -> VertexAIBatchPredictionJob:
+        """
+        A fine-tuned Gemini deployment is configured by its endpoint id, but the v1 batch API only
+        accepts Model resources, so swap the endpoint resource for its deployed tuned model
+        (`projects/../locations/../models/<id>`) read from GET endpoints/<id>.
+        """
+        model: Final = vertex_batch_request.get("model", "")
+        if "/endpoints/" not in model:
+            return vertex_batch_request
+
+        endpoint_url: Final = f"{get_vertex_base_url(vertex_location)}/v1/{model}"
+        response: Final = sync_handler.get(url=endpoint_url, headers=headers)
+        if response.status_code != 200:
+            raise VertexAIError(
+                status_code=response.status_code,
+                message=f"Failed to resolve fine-tuned Vertex endpoint '{model}': {response.text}",
+            )
+
+        payload_view: Final[_VertexEndpointPayloadView] = {"payload": response.json()}
+        deployed_models: Final = payload_view["payload"].get("deployedModels") or []
+        deployed_model: Final = deployed_models[0].get("model", "") if deployed_models else ""
+        if not deployed_model:
+            raise VertexAIError(
+                status_code=400,
+                message=(
+                    f"Vertex endpoint '{model}' has no deployed model, so there is no tuned model "
+                    "resource to run batch predictions against"
+                ),
+            )
+        return {**vertex_batch_request, "model": deployed_model}
+
     async def _async_create_batch(
         self,
         vertex_batch_request: VertexAIBatchPredictionJob,

litellm/llms/vertex_ai/batches/transformation.py

@@ -22,6 +22,8 @@ class VertexAIBatchTransformation:
     def transform_openai_batch_request_to_vertex_ai_batch_request(
         cls,
         request: CreateBatchRequest,
+        vertex_project: str | None = None,
+        vertex_location: str | None = None,
     ) -> VertexAIBatchPredictionJob:
         """
         Transforms OpenAI Batch requests to Vertex AI Batch requests
@@ -31,7 +33,11 @@ class VertexAIBatchTransformation:
         if input_file_id is None:
             raise ValueError("input_file_id is required, but not provided")
         input_config: InputConfig = InputConfig(gcsSource=GcsSource(uris=[input_file_id]), instancesFormat="jsonl")
-        model: Final[str] = cls._get_model_from_gcs_file(input_file_id)
+        model: Final[str] = cls._get_batch_job_model(
+            input_file_id=input_file_id,
+            vertex_project=vertex_project,
+            vertex_location=vertex_location,
+        )
         output_config: Final[OutputConfig] = OutputConfig(
             predictionsFormat="jsonl",
             gcsDestination=GcsDestination(outputUriPrefix=cls._get_gcs_uri_prefix_from_file(input_file_id)),
@@ -188,6 +194,33 @@ class VertexAIBatchTransformation:
         path_parts: Final = input_file_id.rsplit("/", 1)
         return path_parts[0]
 
+    @classmethod
+    def _get_batch_job_model(
+        cls,
+        input_file_id: str,
+        vertex_project: str | None,
+        vertex_location: str | None,
+    ) -> str:
+        """
+        Returns the `model` for the batchPredictionJobs request: the publisher model path as-is, or
+        the full `projects/../locations/../endpoints/<id>` resource name for a fine-tuned endpoint.
+
+        The v1 batch API only accepts Model resources, so the handler resolves an endpoint resource
+        to its deployed tuned model (`projects/../locations/../models/<id>`) before sending the job.
+        """
+        parsed_model: Final = cls._get_model_from_gcs_file(input_file_id)
+        if not parsed_model.startswith("endpoints/"):
+            return parsed_model
+        if not vertex_project:
+            raise VertexAIError(
+                status_code=400,
+                message=(
+                    f"Vertex AI batch jobs against a fine-tuned endpoint ('{parsed_model}') require "
+                    "`vertex_project` to build the endpoint resource name"
+                ),
+            )
+        return f"projects/{vertex_project}/locations/{vertex_location or 'us-central1'}/{parsed_model}"
+
     @classmethod
     def _get_model_from_gcs_file(cls, gcs_file_uri: str) -> str:
         """
@@ -202,6 +235,9 @@ class VertexAIBatchTransformation:
         gcs_file_uri format: gs://litellm-testing-bucket/litellm-vertex-files/publishers/google/models/gemini-1.5-flash-001/e9412502-2c91-42a6-8e61-f5c294cc0fc8
         returns: "publishers/google/models/gemini-1.5-flash-001"
 
+        Fine-tuned Gemini endpoints are stored as `endpoints/<numeric id>` in the uri and returned
+        in that form.
+
         Raises a 400 `VertexAIError` when the uri carries no parseable model path.
         """
         model: Final = cls._parse_model_from_gcs_file(gcs_file_uri)
@@ -210,11 +246,13 @@ class VertexAIBatchTransformation:
                 status_code=400,
                 message=(
                     "Vertex AI batch creation requires the model to be part of `input_file_id`, but "
-                    f"'{gcs_file_uri}' contains no 'publishers/<publisher>/models/<model>' path segment. "
+                    f"'{gcs_file_uri}' contains no 'publishers/<publisher>/models/<model>' or "
+                    "'endpoints/<numeric endpoint id>' path segment. "
                     "Either upload the input file through LiteLLM (POST /v1/files with "
                     "custom_llm_provider=vertex_ai), which encodes the model into the returned file id, or "
                     "pass a uri of the form "
-                    "gs://<bucket>/<prefix>/publishers/<publisher>/models/<model>/<file>"
+                    "gs://<bucket>/<prefix>/publishers/<publisher>/models/<model>/<file> "
+                    "(or gs://<bucket>/<prefix>/endpoints/<numeric endpoint id>/<file> for fine-tuned models)"
                 ),
             )
         return model
@@ -222,10 +260,16 @@ class VertexAIBatchTransformation:
     @classmethod
     def _parse_model_from_gcs_file(cls, gcs_file_uri: str) -> str | None:
         """
-        Returns the `publishers/<publisher>/models/<model>` path from a gcs uri, or None if the uri
-        does not contain one.
+        Returns the `publishers/<publisher>/models/<model>` or `endpoints/<numeric id>` path from a
+        gcs uri, or None if the uri does not contain one.
         """
-        _, separator, model_path = unquote(gcs_file_uri).partition("publishers/")
+        unquoted_uri: Final = unquote(gcs_file_uri)
+        _, endpoint_separator, endpoint_path = unquoted_uri.partition("endpoints/")
+        endpoint_id: Final = endpoint_path.split("/")[0] if endpoint_separator else ""
+        if endpoint_id.isdigit():
+            return f"endpoints/{endpoint_id}"
+
+        _, separator, model_path = unquoted_uri.partition("publishers/")
         if not separator:
             return None
 

litellm/llms/vertex_ai/common_utils.py

@@ -310,6 +310,19 @@ def get_vertex_base_model_name(model: str) -> str:
     return model
 
 
+def get_vertex_ai_fine_tuned_endpoint_id(model: str) -> str | None:
+    """
+    Fine-tuned Gemini deployments are addressed by a numeric endpoint id,
+    configured as `vertex_ai/<id>` or `vertex_ai/gemini/<id>`.
+
+    Returns the endpoint id, or None when `model` is a regular publisher model.
+    Mirrors the online chat path in `_get_vertex_url`, which sends numeric
+    models to `endpoints/{id}` instead of `publishers/google/models/{model}`.
+    """
+    candidate: Final = model.split("/")[-1] if "gemini/" in model else model
+    return candidate if candidate.isdigit() else None
+
+
 def validate_vertex_location(vertex_location: str | None) -> str:
     """
     Validate a Vertex AI location before interpolating it into a request host or

litellm/llms/vertex_ai/files/transformation.py

@@ -39,6 +39,7 @@ from litellm.llms.base_llm.files.transformation import (
 )
 from litellm.llms.vertex_ai.common_utils import (
     _convert_vertex_datetime_to_openai_datetime,
+    get_vertex_ai_fine_tuned_endpoint_id,
 )
 from litellm.llms.vertex_ai.gemini.transformation import _transform_request_body
 from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import (
@@ -712,11 +713,20 @@ class VertexAIFilesConfig(VertexBase, BaseFilesConfig):
         Gets a unique GCS object name for the VertexAI batch prediction job
 
         named as: litellm-vertex-{model}-{uuid}
+
+        Fine-tuned Gemini deployments (numeric endpoint ids) are stored under
+        `endpoints/<id>` so the batch transformation can round-trip them into a
+        `projects/../locations/../endpoints/<id>` batch job model instead of a
+        nonexistent publisher model.
         """
-        _model = openai_jsonl_content[0].get("body", {}).get("model", "")
-        if "publishers/google/models" not in _model:
-            _model = f"publishers/google/models/{_model}"
-        safe_model_path: Final = sanitize_cloud_object_path(_model, fallback="model")
+        raw_model: Final = openai_jsonl_content[0].get("body", {}).get("model", "")
+        endpoint_id: Final = get_vertex_ai_fine_tuned_endpoint_id(raw_model)
+        model_path: Final = (
+            f"endpoints/{endpoint_id}"
+            if endpoint_id is not None
+            else (raw_model if "publishers/google/models" in raw_model else f"publishers/google/models/{raw_model}")
+        )
+        safe_model_path: Final = sanitize_cloud_object_path(model_path, fallback="model")
         object_name: Final = f"{VERTEX_AI_MANAGED_GCS_PREFIX}{safe_model_path}/{uuid.uuid4()}"
         return object_name
 
@@ -761,6 +771,16 @@ class VertexAIFilesConfig(VertexBase, BaseFilesConfig):
         """
         Get the complete url for the request
         """
+        if data.get("purpose") == "batch" and litellm_params.get("custom_endpoint"):
+            raise VertexAIError(
+                status_code=400,
+                message=(
+                    "Vertex AI batch prediction is not supported for `custom_endpoint` deployments. "
+                    "The OpenAI-compatible custom endpoint path has no batch surface in LiteLLM; "
+                    "remove this deployment from the batch request (e.g. `target_model_names`) or "
+                    "use a publisher model / fine-tuned Gemini endpoint instead."
+                ),
+            )
         bucket_name = self._get_configured_bucket_name(litellm_params)
         bucket_name, object_prefix = split_configured_cloud_bucket_name(bucket_name)
         file_data: Final = data.get("file")

tests/proxy_unit_tests/test_check_batch_cost.py

@@ -1760,6 +1760,35 @@ class TestUnmanagedVertexRouting:
         )
         router.get_model_ids.assert_called_once_with(model_name="gemini-2.5-flash")
 
+    def test_flag_on_routes_fine_tuned_endpoint_to_vertex_deployment(self):
+        """A fine-tuned Gemini batch stores `endpoints/<id>` in the gs:// path; the bare model
+        (the endpoint id) must round-trip to the deployment configured as
+        `vertex_ai/gemini/<id>` (LIT-6899)."""
+        endpoint_id = "7768560373388541952"
+        router = MagicMock()
+        router.resolve_model_name_from_model_id.return_value = None
+        router.get_model_list.return_value = [
+            {
+                "model_name": "gemini-2.5-flash-dts-usc1",
+                "litellm_params": {
+                    "model": f"vertex_ai/gemini/{endpoint_id}",
+                    "custom_llm_provider": "vertex_ai",
+                },
+                "model_info": {"id": "deploy-ft"},
+            },
+        ]
+        instance = self._instance(track_unmanaged=True, router=router)
+        job = self._job(
+            file_object=_unmanaged_vertex_file_object(
+                input_file_id=f"gs://bucket/litellm-vertex-files/endpoints/{endpoint_id}/abc.jsonl"
+            )
+        )
+
+        with patch(_IS_B64, return_value=False):
+            result = instance._resolve_job_routing(job, MagicMock())
+
+        assert result == ("deploy-ft", "8823717160934178816")
+
     def test_flag_on_skips_non_vertex_deployment_sharing_model_group(self):
         """Flag on, but the only deployment for the model group is a non-vertex_ai
         provider: must not be selected, even though the model group name matches."""

tests/test_litellm/batches/test_main.py

@@ -158,6 +158,16 @@ def test_create__vertex_ai_dispatch(seams):
     _assert_only(seams.vertex.create_batch, seams, "create_batch")
 
 
+def test_create__vertex_ai_custom_endpoint_raises_badrequest(seams):
+    """custom_endpoint deployments have no Vertex batch surface; creating a job would target a
+    nonexistent publisher model, so the SDK must 400 before dispatching (LIT-6899)."""
+    with pytest.raises(litellm.exceptions.BadRequestError, match="custom_endpoint"):
+        bm.create_batch(**CREATE_KW, custom_llm_provider="vertex_ai", custom_endpoint=True)
+
+    for m in _all_seam_methods(seams, "create_batch"):
+        m.assert_not_called()
+
+
 def test_create__provider_config_routes_to_base_http_handler(seams):
     """model + a provider batches config (bedrock-style) routes to the generic
     base_llm_http_handler, NOT the per-provider instance."""

tests/test_litellm/llms/vertex_ai/batches/test_handler.py

@@ -178,6 +178,125 @@ def test_create_batch_async_returns_coroutine_and_uses_async_client():
     sync_client.post.assert_not_called()
 
 
+def test_create_batch_sync_does_not_resolve_publisher_models():
+    """Publisher-model jobs must not incur the endpoint-resolution GET."""
+    h = _make_handler()
+    client = MagicMock()
+    client.post.return_value = _http_response()
+
+    with patch(f"{HMOD}._get_httpx_client", return_value=client):
+        h.create_batch(
+            _is_async=False,
+            create_batch_data=CREATE_DATA,
+            api_base=None,
+            vertex_credentials=None,
+            vertex_project=PROJECT,
+            vertex_location=LOCATION,
+            timeout=600.0,
+            max_retries=None,
+        )
+
+    client.get.assert_not_called()
+
+
+ENDPOINT_ID = "7768560373388541952"
+ENDPOINT_CREATE_DATA = {
+    "input_file_id": f"gs://bucket/litellm-vertex-files/endpoints/{ENDPOINT_ID}/file-uuid"
+}
+TUNED_MODEL_RESOURCE = f"projects/{PROJECT}/locations/{LOCATION}/models/1234509876"
+
+
+def _endpoint_get_response(deployed_models: list | None = None) -> MagicMock:
+    resp = MagicMock()
+    resp.status_code = 200
+    resp.json.return_value = {
+        "name": f"projects/{PROJECT}/locations/{LOCATION}/endpoints/{ENDPOINT_ID}",
+        "deployedModels": (
+            deployed_models if deployed_models is not None else [{"model": TUNED_MODEL_RESOURCE}]
+        ),
+    }
+    return resp
+
+
+def test_create_batch_sync_resolves_fine_tuned_endpoint_to_tuned_model():
+    """A fine-tuned Gemini file id must produce a batch job against the endpoint's deployed
+    tuned model resource; the v1 batch API rejects endpoint resources in `model` (LIT-6899)."""
+    h = _make_handler()
+    client = MagicMock()
+    client.get.return_value = _endpoint_get_response()
+    client.post.return_value = _http_response()
+
+    with patch(f"{HMOD}._get_httpx_client", return_value=client):
+        out = h.create_batch(
+            _is_async=False,
+            create_batch_data=ENDPOINT_CREATE_DATA,
+            api_base=None,
+            vertex_credentials=None,
+            vertex_project=PROJECT,
+            vertex_location=LOCATION,
+            timeout=600.0,
+            max_retries=None,
+        )
+
+    assert isinstance(out, LiteLLMBatch)
+    get_kwargs = client.get.call_args.kwargs
+    assert get_kwargs["url"] == (
+        f"https://{LOCATION}-aiplatform.googleapis.com/v1/projects/{PROJECT}"
+        f"/locations/{LOCATION}/endpoints/{ENDPOINT_ID}"
+    )
+    assert get_kwargs["headers"]["Authorization"] == f"Bearer {TOKEN}"
+    sent = json.loads(client.post.call_args.kwargs["data"])
+    assert sent["model"] == TUNED_MODEL_RESOURCE
+
+
+def test_create_batch_sync_endpoint_resolution_error_raises():
+    h = _make_handler()
+    client = MagicMock()
+    resolve_response = MagicMock()
+    resolve_response.status_code = 404
+    resolve_response.text = "endpoint not found"
+    client.get.return_value = resolve_response
+
+    with patch(f"{HMOD}._get_httpx_client", return_value=client):
+        with pytest.raises(VertexAIError) as exc_info:
+            h.create_batch(
+                _is_async=False,
+                create_batch_data=ENDPOINT_CREATE_DATA,
+                api_base=None,
+                vertex_credentials=None,
+                vertex_project=PROJECT,
+                vertex_location=LOCATION,
+                timeout=600.0,
+                max_retries=None,
+            )
+
+    assert exc_info.value.status_code == 404
+    client.post.assert_not_called()
+
+
+def test_create_batch_sync_endpoint_without_deployed_model_raises_400():
+    h = _make_handler()
+    client = MagicMock()
+    client.get.return_value = _endpoint_get_response(deployed_models=[])
+
+    with patch(f"{HMOD}._get_httpx_client", return_value=client):
+        with pytest.raises(VertexAIError) as exc_info:
+            h.create_batch(
+                _is_async=False,
+                create_batch_data=ENDPOINT_CREATE_DATA,
+                api_base=None,
+                vertex_credentials=None,
+                vertex_project=PROJECT,
+                vertex_location=LOCATION,
+                timeout=600.0,
+                max_retries=None,
+            )
+
+    assert exc_info.value.status_code == 400
+    assert "no deployed model" in str(exc_info.value)
+    client.post.assert_not_called()
+
+
 def test_create_batch_sync_httpstatuserror_propagates():
     """``HTTPHandler.post`` raises for non-2xx via ``raise_for_status``; the
     sync create path must surface that error, not swallow it."""

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

@@ -34,6 +34,12 @@ INPUT_FILE = (
     "models/gemini-1.5-flash-001/e9412502-2c91-42a6-8e61-f5c294cc0fc8"
 )
 
+ENDPOINT_ID = "7768560373388541952"
+ENDPOINT_INPUT_FILE = (
+    f"gs://litellm-testing-bucket/litellm-vertex-files/endpoints/{ENDPOINT_ID}/"
+    "e9412502-2c91-42a6-8e61-f5c294cc0fc8"
+)
+
 
 # =========================================================================== #
 # transform_openai_batch_request_to_vertex_ai_batch_request
@@ -67,6 +73,41 @@ def test_transform_openai_request_missing_input_file_id_raises():
         T.transform_openai_batch_request_to_vertex_ai_batch_request({})
 
 
+def test_transform_openai_request_fine_tuned_endpoint_builds_endpoint_resource():
+    """A fine-tuned Gemini file id (endpoints/<numeric id>) must target the endpoint resource,
+    not a nonexistent publisher model (LIT-6899)."""
+    job = T.transform_openai_batch_request_to_vertex_ai_batch_request(
+        {"input_file_id": ENDPOINT_INPUT_FILE},
+        vertex_project="my-project",
+        vertex_location="us-central1",
+    )
+    assert job["model"] == f"projects/my-project/locations/us-central1/endpoints/{ENDPOINT_ID}"
+
+
+def test_transform_openai_request_fine_tuned_endpoint_defaults_location():
+    job = T.transform_openai_batch_request_to_vertex_ai_batch_request(
+        {"input_file_id": ENDPOINT_INPUT_FILE},
+        vertex_project="my-project",
+    )
+    assert job["model"] == f"projects/my-project/locations/us-central1/endpoints/{ENDPOINT_ID}"
+
+
+def test_transform_openai_request_fine_tuned_endpoint_without_project_raises_400():
+    with pytest.raises(VertexAIError) as exc_info:
+        T.transform_openai_batch_request_to_vertex_ai_batch_request({"input_file_id": ENDPOINT_INPUT_FILE})
+    assert exc_info.value.status_code == 400
+    assert "vertex_project" in str(exc_info.value)
+
+
+def test_transform_openai_request_publisher_model_ignores_project_and_location():
+    job = T.transform_openai_batch_request_to_vertex_ai_batch_request(
+        {"input_file_id": INPUT_FILE},
+        vertex_project="my-project",
+        vertex_location="europe-west4",
+    )
+    assert job["model"] == "publishers/google/models/gemini-1.5-flash-001"
+
+
 @pytest.mark.parametrize(
     "input_file_id",
     [
@@ -321,6 +362,21 @@ def test_get_model_from_gcs_file_no_publishers_raises_400():
     assert exc_info.value.status_code == 400
 
 
+def test_get_model_from_gcs_file_fine_tuned_endpoint():
+    """The whole endpoint id must survive parsing; the old 3-segment publishers/ parse dropped it."""
+    assert T._get_model_from_gcs_file(ENDPOINT_INPUT_FILE) == f"endpoints/{ENDPOINT_ID}"
+
+
+def test_get_model_from_gcs_file_non_numeric_endpoints_segment_raises_400():
+    with pytest.raises(VertexAIError) as exc_info:
+        T._get_model_from_gcs_file("gs://bucket/endpoints/not-a-number/file-uuid")
+    assert exc_info.value.status_code == 400
+
+
+def test_get_bare_model_name_from_gcs_file_fine_tuned_endpoint():
+    assert T.get_bare_model_name_from_gcs_file(ENDPOINT_INPUT_FILE) == ENDPOINT_ID
+
+
 # =========================================================================== #
 # is_unmanaged_gcs_batch_input_file_id
 # =========================================================================== #
@@ -334,6 +390,8 @@ def test_get_model_from_gcs_file_no_publishers_raises_400():
         ("file-abc123", False),
         ("gs://bucket/no-model-here.jsonl", False),
         ("gs://bucket/publishers/google/gemini-1.5-flash-001/file-uuid", False),
+        (ENDPOINT_INPUT_FILE, True),
+        ("gs://bucket/endpoints/not-a-number/file-uuid", False),
     ],
 )
 def test_is_unmanaged_gcs_batch_input_file_id(input_file_id, expected):

tests/test_litellm/llms/vertex_ai/files/test_vertex_ai_files_transformation.py

@@ -159,6 +159,61 @@ class TestCreateFileUrl:
         assert "?" not in object_name
 
 
+class TestBatchObjectNaming:
+    def test_should_store_publisher_model_under_publishers_path(self, config):
+        object_name = config._get_gcs_object_name_from_batch_jsonl([{"body": {"model": "gemini-2.5-flash"}}])
+        assert object_name.startswith("litellm-vertex-files/publishers/google/models/gemini-2.5-flash/")
+
+    def test_should_store_fine_tuned_endpoint_under_endpoints_path(self, config):
+        """A numeric endpoint id must not be filed under publishers/google/models/gemini/<id>,
+        which the batch transformation later mangles into a nonexistent publisher model (LIT-6899)."""
+        object_name = config._get_gcs_object_name_from_batch_jsonl(
+            [{"body": {"model": "gemini/7768560373388541952"}}]
+        )
+        assert object_name.startswith("litellm-vertex-files/endpoints/7768560373388541952/")
+        assert "publishers" not in object_name
+
+    def test_should_store_bare_numeric_endpoint_under_endpoints_path(self, config):
+        object_name = config._get_gcs_object_name_from_batch_jsonl([{"body": {"model": "7768560373388541952"}}])
+        assert object_name.startswith("litellm-vertex-files/endpoints/7768560373388541952/")
+
+
+class TestCustomEndpointBatchUpload:
+    def test_should_reject_batch_upload_for_custom_endpoint_deployment(self, config):
+        """custom_endpoint deployments have no Vertex batch surface; the upload must 400 instead
+        of staging a file that can only produce a doomed batch job (LIT-6899)."""
+        from litellm.llms.vertex_ai.common_utils import VertexAIError
+
+        with pytest.raises(VertexAIError) as exc_info:
+            config.get_complete_file_url(
+                api_base=None,
+                api_key=None,
+                model="",
+                optional_params={},
+                litellm_params={"gcs_bucket_name": "my-bucket", "custom_endpoint": True},
+                data={
+                    "file": ("batch.jsonl", b'{"body": {"model": "openai/gemma-2-2b-it"}}', "application/jsonl"),
+                    "purpose": "batch",
+                },
+            )
+        assert exc_info.value.status_code == 400
+        assert "custom_endpoint" in str(exc_info.value)
+
+    def test_should_allow_non_batch_upload_for_custom_endpoint_deployment(self, config):
+        url = config.get_complete_file_url(
+            api_base=None,
+            api_key=None,
+            model="",
+            optional_params={},
+            litellm_params={"gcs_bucket_name": "my-bucket", "custom_endpoint": True},
+            data={
+                "file": ("notes.txt", b"hello", "text/plain"),
+                "purpose": "assistants",
+            },
+        )
+        assert "/b/my-bucket/" in url
+
+
 class TestTransformRetrieveFile:
     def test_should_build_correct_gcs_metadata_url(self, config):
         file_id = "gs://my-bucket/litellm-vertex-files/path/to/file.jsonl"