From 34e692c903c9d75b52065b4093c7d80b7eb2e00b Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Wed, 19 Aug 2026 20:16:22 -0700 Subject: [PATCH] fix(batches): decode model-encoded output file id so completed batches book spend Adds e2e coverage for batches terminal state and cost write-back, failure paths, per-backend file content downloads, and two-gateway routing (LIT-5730). --- litellm/batches/batch_utils.py | 40 +- tests/e2e/batches/COVERAGE.md | 87 +++- tests/e2e/batches/batch_client.py | 29 +- tests/e2e/batches/capabilities.py | 10 + tests/e2e/batches/test_batches_e2e.py | 471 +++++++++++++++++- .../llm_nonconversational.yaml | 10 + .../test_litellm/batches/test_batch_utils.py | 37 ++ 7 files changed, 646 insertions(+), 38 deletions(-) diff --git a/litellm/batches/batch_utils.py b/litellm/batches/batch_utils.py index 0cf22d82ca6..6eb13d2cba7 100644 --- a/litellm/batches/batch_utils.py +++ b/litellm/batches/batch_utils.py @@ -296,6 +296,32 @@ def calculate_vertex_ai_batch_cost_and_usage( ) +def _provider_output_file_id(output_file_id: str) -> str: + """ + Resolve the file id the provider actually knows: unified ids yield their embedded + llm_output_file_id, model-encoded ids decode to the raw provider id, raw ids pass through. + """ + from litellm.proxy.openai_files_endpoints.common_utils import ( + _is_base64_encoded_unified_file_id, + get_original_file_id, + ) + + unified_file_id: Final = _is_base64_encoded_unified_file_id(output_file_id) + if not unified_file_id: + return get_original_file_id(output_file_id) + try: + extracted: Final = unified_file_id.split("llm_output_file_id,")[1].split(";")[0] + except (IndexError, AttributeError) as e: + verbose_logger.error( + "Failed to extract LLM output file ID from unified file ID: %s, error: %s", + output_file_id, + e, + ) + return output_file_id + verbose_logger.debug("Extracted LLM output file ID from unified file ID: %s", extracted) + return extracted + + async def _fetch_batch_output_file_content( batch: Batch, custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm", "anthropic"] = "openai", @@ -311,23 +337,11 @@ async def _fetch_batch_output_file_content( Required for Azure and other providers that need authentication """ from litellm.files.main import afile_content - from litellm.proxy.openai_files_endpoints.common_utils import ( - _is_base64_encoded_unified_file_id, - ) if batch.output_file_id is None: raise ValueError("Output file id is None cannot retrieve file content") - file_id = batch.output_file_id - is_base64_unified_file_id: Final = _is_base64_encoded_unified_file_id(file_id) - if is_base64_unified_file_id: - try: - file_id = is_base64_unified_file_id.split("llm_output_file_id,")[1].split(";")[0] - verbose_logger.debug("Extracted LLM output file ID from unified file ID: %s", file_id) - except (IndexError, AttributeError) as e: - verbose_logger.error( - "Failed to extract LLM output file ID from unified file ID: %s, error: %s", batch.output_file_id, e - ) + file_id: Final = _provider_output_file_id(batch.output_file_id) # Build kwargs for afile_content with credentials from litellm_params file_content_kwargs: Final = { diff --git a/tests/e2e/batches/COVERAGE.md b/tests/e2e/batches/COVERAGE.md index ca48204962a..f02d4eb4fe4 100644 --- a/tests/e2e/batches/COVERAGE.md +++ b/tests/e2e/batches/COVERAGE.md @@ -1,9 +1,11 @@ # Batches Test Coverage Matrix Live e2e coverage of the Batches API over a real proxy, real provider keys, and -real cost. Synchronous tier only: a batch's completion window is 24h, so these -tests never wait for `completed`. They assert the proxy accepts, routes, retrieves, -cancels, and lists a batch; everything created is deleted on teardown. +real cost. Mostly synchronous tier: a batch's completion window is 24h, so the +lifecycle matrix never waits for `completed`. It asserts the proxy accepts, routes, +retrieves, cancels, and lists a batch; everything created is deleted on teardown. +The exception is `TestBatchTerminalState`, which covers the completed state and +cost write-back via a cross-run marker baton (design below). ## Provider x operation @@ -12,19 +14,26 @@ row per supported (provider, scenario) pair, so there are no skipped cells in th parametrized run. The batches suite never skips: missing provider creds or upstream failures are hard test failures (see `tests/e2e/CLAUDE.md`). -| Provider | create | retrieve | cancel | list | file backing | -|-----------|--------|----------|--------|------|--------------| -| OpenAI | yes | yes | yes | yes | OpenAI Files | -| Azure | yes | yes | yes | yes | Azure Files | -| Vertex AI | yes | yes | yes | yes | GCS (`gcs_bucket_name` / `GCS_BUCKET_NAME` on model) | -| Bedrock | yes (unified only) | yes | no (limited upstream) | no | S3 (`s3_bucket_name` + `aws_*` + `AWS_BATCH_ROLE_ARN` on model) | +| Provider | create | retrieve | cancel | list | content download | file backing | +|-----------|--------|----------|--------|------|------------------|--------------| +| OpenAI | yes | yes | yes | yes | yes (lifecycle + terminal output) | OpenAI Files | +| Azure | yes | yes | yes | yes | yes (byte-verbatim) | Azure Files | +| Vertex AI | yes | yes | yes | yes | yes (provider-transformed) | GCS (`gcs_bucket_name` / `GCS_BUCKET_NAME` on model) | +| Bedrock | yes (unified only) | yes | no (limited upstream) | no | yes (provider-transformed) | S3 (`s3_bucket_name` + `aws_*` + `AWS_BATCH_ROLE_ARN` on model) | Bedrock cancel is unreliable upstream and list is unsupported, so both are gated off -(`can_cancel=False`, `can_list=False`) when that provider is enabled in the matrix. +(`can_cancel=False`, `can_list=False`) when that provider is enabled in the matrix; +flipping those gates is tracked in LIT-4774 and deliberately not part of this suite. Bedrock file upload requires a model on the request (`encoded` / `unified` scenarios only); `model_param` and `provider_fallback` are omitted because `POST /bedrock/v1/files` has no model-less passthrough path. +`GET /v1/files/{id}/content` is exercised for the unified upload path per backend in +`test_unified_file_content_downloads`. Azure stores the JSONL verbatim, so its download +is asserted byte-equal to the upload. Vertex (GCS) and Bedrock (S3) transform lines at +upload time, so those assert a 200 with non-empty parseable JSON lines instead. Gemini +(non-Vertex) raises `NotImplementedError` for file content and has no cell here. + ## Routing scenarios (per `litellm/proxy/batches_endpoints/endpoints.py`) Each create-capable provider runs all four. The test asserts the returned file id @@ -71,11 +80,59 @@ File delete asserts `object=="file"` and `deleted==True`. | `batch_client.py` | typed file upload/download + batch create/retrieve/cancel/list/delete over the shared ProxyClient; runtime batch model registration via /model/new; denial helpers | | `capabilities.py` | the provider x scenario matrix + per-provider /model/new params + id-shape classifiers + per-provider raw-id assertion | | `conftest.py` | session-scoped batch deployment registration and teardown | -| `test_batches_e2e.py` | parametrized lifecycle with per-endpoint output assertions, file upload/delete outputs, key-model-access denial | +| `test_batches_e2e.py` | parametrized lifecycle with per-endpoint output assertions, file upload/delete outputs, key-model-access denial, per-backend content download, failure paths, second-hop routing, terminal state + cost | + +## Failure paths + +`TestBatchFailurePaths` pins the customer-facing error contracts. A malformed input +file is a 400 at upload naming the bad content. A JSONL line whose url contradicts +the batch endpoint passes create (providers validate asynchronously) and drives the +batch to `failed` with structured `errors.data` (code/line/message), a null +`output_file_id`, and a $0 spend row keyed `{batch_id}_batch_cost` (LIT-4852: a +failed batch books $0 instead of crashing cost tracking). Cancelling that failed +batch is a 409 naming the terminal status. A file id encoded for one deployment wins +over a conflicting `model` param on create: the batch routes and re-encodes by the +file's embedded model (foreign-id precedence). + +## Second hop (two chained gateways) + +`TestBatchSecondHop` registers a `litellm_proxy/` deployment pointing at +the proxy's own base URL with a freshly minted virtual key, so unified upload and +create traverse gateway -> gateway -> OpenAI (LIT-5347, PR #36240). The pin: +`target_model_names` is rewritten to the inner deployment on the second hop and the +nested managed ids round-trip retrieve. This self-chaining only needs the proxy to +reach its own `PROXY_BASE_URL`, which holds both locally and on the e2e stage. + +## Terminal state + cost write-back (cross-run marker baton) + +The 24h completion window rules out submit-and-wait inside one run, so +`TestBatchTerminalState` amortizes across runs. Each run submits a 1-line marker +batch (stable metadata key/value plus a per-run field) and deliberately never +cancels or deletes it or its input file: the marker is the baton the next run picks +up (OpenAI files expire on their own after ~30 days). Polling is list-only, up to 5 +minutes, because retrieving a non-terminal batch books a $0 spend row whose +request_id then blocks the later real-cost row (`skip_duplicates`); the single +retrieve happens only once a completed marker exists. The assertion target is the +newest completed marker from ANY run: run-scoped deployment names mean the list +re-encodes prior-run batches under new encoded ids, so their spend keys are fresh +and a prior-run marker is billable by this run. On the 6h stage cadence the full +assertions are therefore deterministic from run 2 onward. On a cold start (no +completed marker within the poll budget) the test passes on the submission +assertions alone: a documented vacuous pass, not a skip. Markers aged past the 24h +window (25h-73h band, within the newest 100-item list page) must be terminal. + +The cost assertion is the LIT-5730 headline: retrieving a completed model-encoded +batch must write a positive spend row with call_type `aretrieve_batch` and token +usage. Before the fix in `litellm/batches/batch_utils.py`, the retrieve endpoint +re-encoded the response's `output_file_id` in place before the queued logging +worker ran, the worker sent that encoded id to OpenAI, got a 404, and the spend row +never landed. ## Out of scope (intentionally) -Driving a batch to `completed`, cost tracking on completion, and the DB write-back -are not covered here; the 24h window makes them unfit for a synchronous gate. That -logic belongs in a DI-stubbed proxy integration test under `tests/test_litellm/proxy/` -where the provider client is injected to return `completed` deterministically. +Unified (managed) batch cost is owned by the hourly `CheckBatchCost` poller, and a +terminal DB status short-circuits retrieve for those ids, so the terminal-state cell +uses the encoded path; poller timing does not fit an e2e gate and belongs in a +DI-stubbed proxy integration test under `tests/test_litellm/proxy/`. Bedrock +cancel/list stay gated pending LIT-4774. Gemini (non-Vertex) file content raises +`NotImplementedError` upstream and is not a coverage cell. diff --git a/tests/e2e/batches/batch_client.py b/tests/e2e/batches/batch_client.py index 5cc5d1dae3b..21a56f3398f 100644 --- a/tests/e2e/batches/batch_client.py +++ b/tests/e2e/batches/batch_client.py @@ -44,6 +44,17 @@ class FileList(BaseModel): data: list[FileObject] = [] +class BatchErrorItem(BaseModel): + code: str | None = None + line: int | None = None + message: str | None = None + + +class BatchErrorList(BaseModel): + object: str | None = None + data: list[BatchErrorItem] = [] + + class BatchObject(BaseModel): id: str object: str | None = None @@ -51,6 +62,9 @@ class BatchObject(BaseModel): endpoint: str | None = None input_file_id: str | None = None output_file_id: str | None = None + error_file_id: str | None = None + errors: BatchErrorList | None = None + metadata: dict[str, str] | None = None completion_window: str | None = None created_at: int | None = None model: str | None = None @@ -72,12 +86,18 @@ class BatchCreateBody(BaseModel): endpoint: str = "/v1/chat/completions" completion_window: str = "24h" model: str | None = None + metadata: dict[str, str] | None = None class ModelQuery(BaseModel): model: str | None = None +class BatchListQuery(BaseModel): + model: str | None = None + limit: int | None = None + + def is_model_access_denied(resp: StreamingResponse) -> bool: """True if the proxy rejected the call because the key may not access the model.""" return resp.status_code == 403 and "key_model_access_denied" in resp.body @@ -168,12 +188,17 @@ class BatchClient: ) def list_batches( - self, *, key: str, provider: str | None = None + self, + *, + key: str, + provider: str | None = None, + model: str | None = None, + limit: int | None = None, ) -> Result[BatchList]: return self.proxy.transport.get( _batches_path(provider), headers=self.proxy.transport.bearer(key), - params=NoBody(), + params=BatchListQuery(model=model, limit=limit), response_type=BatchList, ) diff --git a/tests/e2e/batches/capabilities.py b/tests/e2e/batches/capabilities.py index 3988fb5e7e1..ce1f68184a7 100644 --- a/tests/e2e/batches/capabilities.py +++ b/tests/e2e/batches/capabilities.py @@ -210,6 +210,16 @@ def is_model_encoded_id(id_str: str) -> bool: return False +def decoded_model_from_id(id_str: str) -> str | None: + """Deployment name embedded in a model-encoded file/batch id, or None.""" + for prefix in ("file-", "batch_"): + if id_str.startswith(prefix): + decoded = _b64_decode(id_str[len(prefix) :]) + if decoded.startswith("litellm:") and ";model," in decoded: + return decoded.split(";model,", 1)[1].split(";")[0] + return None + + def matches_id_shape(shape: IdShape, id_str: str) -> bool: if shape == "managed": return is_managed_id(id_str) diff --git a/tests/e2e/batches/test_batches_e2e.py b/tests/e2e/batches/test_batches_e2e.py index 1376bdbed38..75b3a6cd758 100644 --- a/tests/e2e/batches/test_batches_e2e.py +++ b/tests/e2e/batches/test_batches_e2e.py @@ -1,11 +1,12 @@ """Live e2e for the Batches API across every provider LiteLLM supports. -Synchronous tier only: a batch's completion window is 24h, so these never wait for -"completed". Each case uploads a tiny JSONL, creates the batch through one of the -four routing scenarios, asserts it was accepted (non-terminal status) and routed to -the right provider, then retrieves / cancels / lists where the provider supports it. -Everything created is deleted on teardown. Completion + cost tracking are out of -scope here (see COVERAGE.md). +Mostly synchronous tier: a batch's completion window is 24h, so the lifecycle +matrix never waits for "completed". Each case uploads a tiny JSONL, creates the +batch through one of the four routing scenarios, asserts it was accepted +(non-terminal status) and routed to the right provider, then retrieves / cancels / +lists where the provider supports it. Everything created is deleted on teardown. +The exception is TestBatchTerminalState, which carries completed-state + cost +write-back coverage via a cross-run marker baton (design in COVERAGE.md). Routing signal: for provider_fallback the raw batch id discriminates the provider; for the encoded/unified/model_param scenarios the proxy re-encodes the id, so the @@ -22,8 +23,9 @@ from datetime import datetime, timedelta, timezone from typing import Callable import pytest +from pydantic import BaseModel -from e2e_config import unique_marker +from e2e_config import PROXY_BASE_URL, unique_marker from batch_client import ( UPLOAD_FILENAME, @@ -40,9 +42,12 @@ from capabilities import ( CAPABILITIES, FILE_ID_SHAPE, OPENAI_BATCH_MODEL, + PROVIDERS, Capability, + Provider, batch_model_name, coverage_cells_for_lifecycle, + decoded_model_from_id, is_managed_id, matches_id_shape, raw_id_matches_provider, @@ -475,9 +480,22 @@ def test_rate_limited_batch_create_leaves_no_unattributed_spend_row( OPENAI_FILE_CONTENT_BACKEND = "gpt-4o-mini" +FILE_CONTENT_CELLS = { + "azure": "llm.files.azure_openai.content.nonstream.works", + "vertex_ai": "llm.files.vertex.content.nonstream.works", + "bedrock": "llm.files.bedrock.content.nonstream.works", +} +BYTE_FIDELITY_CONTENT_PROVIDERS = frozenset({"azure"}) + class TestBatchFileContent: - """GET /v1/files/{id}/content returns the uploaded batch JSONL bytes.""" + """GET /v1/files/{id}/content returns the uploaded batch JSONL bytes. + + Azure stores the upload verbatim, so its download is asserted byte-equal. + Vertex (GCS) and Bedrock (S3) transform each JSONL line into the provider's + request format at upload time, so their downloads assert 200 plus non-empty + parseable JSON lines instead of byte equality. + """ @pytest.mark.covers( "llm.files.openai.content.nonstream.works", @@ -521,6 +539,62 @@ class TestBatchFileContent: "downloaded file content must match the uploaded JSONL bytes" ) + @pytest.mark.parametrize( + "provider", + [ + pytest.param( + p, + id=p.name, + marks=pytest.mark.covers( + FILE_CONTENT_CELLS[p.name], exercised_on=["files"] + ), + ) + for p in PROVIDERS + if p.name in FILE_CONTENT_CELLS + ], + ) + def test_unified_file_content_downloads( + self, + provider: Provider, + client: BatchClient, + resources: ResourceManager, + batch_deployments: None, + ) -> None: + key = resources.key() + payload = render_jsonl(provider.raw_model) + file = unwrap( + client.upload_file( + content=payload, + form=FileUploadForm(purpose="batch", target_model_names=provider.model), + key=key, + ) + ) + resources.defer(quietly(lambda: client.delete_file(file.id, key=key))) + assert_file_object(file, provider=provider.name) + assert is_managed_id(file.id), ( + f"{provider.name}: unified upload must return a managed file id, got {file.id!r}" + ) + + downloaded = client.proxy.transport.download( + f"/v1/files/{file.id}/content", + headers=client.proxy.transport.bearer(key), + ) + assert downloaded.status_code == 200, ( + f"{provider.name}: file content must be 200, " + f"got {downloaded.status_code}: {downloaded.body[:300]}" + ) + body = downloaded.body.strip() + assert body, f"{provider.name}: file content download returned an empty body" + if provider.name in BYTE_FIDELITY_CONTENT_PROVIDERS: + assert body == payload.decode().strip(), ( + f"{provider.name}: downloaded content must match the uploaded JSONL bytes" + ) + else: + for line in body.splitlines(): + assert json.loads(line), ( + f"{provider.name}: content line is not JSON: {line[:200]}" + ) + class TestOpenAIFiles: """GET /v1/files (list) and GET /v1/files/{id} (retrieve) over the OpenAI route. @@ -866,3 +940,384 @@ class TestHostedVllmBatch: f"hosted_vllm batch has non-transitional status {batch.status!r}" ) assert_batch_object(batch) + + +BATCH_TERMINAL_STATUSES = frozenset({"completed", "failed", "expired", "cancelled"}) +FAILED_BATCH_POLL_SECONDS = 120.0 +FAILED_BATCH_POLL_INTERVAL_SECONDS = 5.0 + +AZURE_BATCH_RAW_MODEL = next(p.raw_model for p in PROVIDERS if p.name == "azure") + + +def _mismatched_endpoint_jsonl(model: str) -> bytes: + line = { + "custom_id": "req-1", + "method": "POST", + "url": "/v1/embeddings", + "body": {"model": model, "input": "ping"}, + } + return (json.dumps(line) + "\n").encode() + + +def _poll_until_terminal(client: BatchClient, batch_id: str, key: str) -> BatchObject: + deadline = time.monotonic() + FAILED_BATCH_POLL_SECONDS + fetched = retrieve_batch(client, batch_id, key=key, provider=None) + while fetched.status not in BATCH_TERMINAL_STATUSES and time.monotonic() < deadline: + time.sleep(FAILED_BATCH_POLL_INTERVAL_SECONDS) + fetched = retrieve_batch(client, batch_id, key=key, provider=None) + return fetched + + +class TestBatchFailurePaths: + """Customer-facing failure contracts for /v1/batches. + + A malformed input file is rejected at upload with a 400 naming the bad + content. A JSONL line whose url contradicts the batch endpoint is accepted + at create (providers validate asynchronously) and drives the batch to + "failed" with structured per-line errors, a null output_file_id, and a + zero-cost spend row (LIT-4852: a failed batch must book $0, not crash cost + tracking). Cancelling that already-failed batch returns a 409 naming the + terminal status. A file id encoded for one deployment wins over a + conflicting model param on create: the batch routes (and re-encodes) by the + file's embedded model, pinning that precedence. + """ + + @pytest.mark.covers( + "llm.batches.openai.malformed_jsonl.nonstream.works", + exercised_on=["files"], + ) + def test_malformed_jsonl_upload_rejected( + self, client: BatchClient, resources: ResourceManager, batch_deployments: None + ) -> None: + result = client.upload_file( + content=b"this is not json\n", + form=FileUploadForm(purpose="batch"), + model=OPENAI_BATCH_MODEL, + key=resources.key(), + ) + match result: + case UnknownApiError(status_code=400, body=body): + assert "json" in body.lower(), ( + f"400 must name the malformed JSONL so users can fix the file, got: {body[:300]}" + ) + case _: + pytest.fail(f"malformed JSONL upload must be rejected with a 400, got: {result}") + + @pytest.mark.covers( + "llm.batches.openai.jsonl_endpoint_mismatch.nonstream.works", + "llm.batches.openai.cancel_terminal.nonstream.works", + exercised_on=["batches", "files"], + ) + def test_endpoint_mismatch_fails_batch_and_cancel_conflicts( + self, client: BatchClient, resources: ResourceManager, batch_deployments: None + ) -> None: + key = resources.key() + file = unwrap( + client.upload_file( + content=_mismatched_endpoint_jsonl("gpt-4o-mini"), + form=FileUploadForm(purpose="batch"), + model=OPENAI_BATCH_MODEL, + key=key, + ) + ) + resources.defer(quietly(lambda: client.delete_file(file.id, key=key))) + + created = client.create_batch(body=BatchCreateBody(input_file_id=file.id), key=key) + require_successful_call(created) + batch = BatchObject.model_validate_json(created.body) + + fetched = _poll_until_terminal(client, batch.id, key) + assert fetched.status == "failed", ( + f"endpoint-mismatched batch must fail, got {fetched.status!r}" + ) + assert fetched.output_file_id is None, ( + f"failed batch must have no output file, got {fetched.output_file_id!r}" + ) + assert fetched.errors is not None and fetched.errors.data, ( + "failed batch must surface structured errors so users can fix the JSONL" + ) + first_error = fetched.errors.data[0] + assert first_error.message, "batch error item has no message" + assert first_error.code, "batch error item has no code" + + rows = client.proxy.poll_logs_for_request_id(f"{fetched.id}_batch_cost") + assert rows, ( + f"failed batch {fetched.id} wrote no spend row; retrieve must book $0 (LIT-4852)" + ) + assert all((row.spend or 0) == 0 for row in rows), ( + f"failed batch must cost $0, got {[(r.request_id, r.spend) for r in rows]}" + ) + assert rows[0].call_type == "aretrieve_batch", ( + f"batch cost row call_type={rows[0].call_type!r}" + ) + + conflict = client.cancel_batch(batch.id, key=key) + match conflict: + case UnknownApiError(status_code=409, body=body): + assert "failed" in body.lower(), ( + f"409 must name the terminal status blocking the cancel, got: {body[:300]}" + ) + case _: + pytest.fail(f"cancel of a failed batch must return a 409 conflict, got: {conflict}") + + @pytest.mark.covers( + "llm.batches.openai.foreign_file_id.nonstream.works", + exercised_on=["batches", "files"], + ) + def test_foreign_encoded_file_id_routes_by_file_model( + self, client: BatchClient, resources: ResourceManager, batch_deployments: None + ) -> None: + key = resources.key() + file = unwrap( + client.upload_file( + content=render_jsonl(AZURE_BATCH_RAW_MODEL), + form=FileUploadForm(purpose="batch"), + model=AZURE_BATCH_MODEL, + key=key, + ) + ) + resources.defer(quietly(lambda: client.delete_file(file.id, key=key))) + assert decoded_model_from_id(file.id) == AZURE_BATCH_MODEL, ( + f"upload did not encode the azure deployment into the file id: {file.id!r}" + ) + + created = client.create_batch( + body=BatchCreateBody(input_file_id=file.id, model=OPENAI_BATCH_MODEL), key=key + ) + require_successful_call(created) + batch = BatchObject.model_validate_json(created.body) + resources.defer(quietly(lambda: client.cancel_batch(batch.id, key=key))) + + assert decoded_model_from_id(batch.id) == AZURE_BATCH_MODEL, ( + "create with a foreign encoded file id must route by the file's embedded model, " + f"but the batch id encodes {decoded_model_from_id(batch.id)!r} " + f"(model param was {OPENAI_BATCH_MODEL!r})" + ) + fetched = retrieve_batch(client, batch.id, key=key, provider=None) + assert fetched.id == batch.id + assert fetched.status, "retrieved foreign-file batch has no status" + + +class TestBatchSecondHop: + """Two-proxy batch routing: a litellm_proxy deployment chained to the gateway + itself (LIT-5347, PR #36240). + + The hop deployment's litellm_params point litellm_proxy/ at this + gateway's own base URL with a freshly minted virtual key, so the unified + upload and batch create traverse gateway -> gateway -> OpenAI. The regression + this pins: target_model_names must be rewritten to the inner deployment on + the second hop and the nested managed ids must round-trip retrieve. + """ + + @pytest.mark.covers( + "llm.batches.openai.second_hop.nonstream.works", + exercised_on=["batches", "files"], + ) + def test_unified_create_and_retrieve_via_chained_gateway( + self, client: BatchClient, resources: ResourceManager, batch_deployments: None + ) -> None: + key = resources.key() + hop_name = batch_model_name("openai-batch-hop") + model_id = client.create_model( + hop_name, + LiteLLMParamsBody( + model=f"litellm_proxy/{OPENAI_BATCH_MODEL}", + api_base=PROXY_BASE_URL, + api_key=key, + ), + ) + resources.defer(lambda: client.delete_model(model_id)) + + file = unwrap( + client.upload_file( + content=render_jsonl("gpt-4o-mini"), + form=FileUploadForm(purpose="batch", target_model_names=hop_name), + key=key, + ) + ) + resources.defer(quietly(lambda: client.delete_file(file.id, key=key))) + assert is_managed_id(file.id), ( + f"second-hop unified upload must return a managed file id, got {file.id!r}" + ) + + created = client.create_batch(body=BatchCreateBody(input_file_id=file.id), key=key) + require_successful_call(created) + batch = BatchObject.model_validate_json(created.body) + resources.defer(quietly(lambda: client.cancel_batch(batch.id, key=key))) + + assert is_managed_id(batch.id), ( + f"second-hop create must return a managed batch id, got {batch.id!r}" + ) + assert batch.status in CREATED_BATCH_STATUSES, ( + f"second-hop batch has non-transitional status {batch.status!r}" + ) + assert_batch_object(batch) + + fetched = retrieve_batch(client, batch.id, key=key, provider=None) + assert fetched.id == batch.id + assert fetched.status, "second-hop retrieve returned no status" + + +class BatchOutputBody(BaseModel): + choices: list[object] = [] + + +class BatchOutputResponse(BaseModel): + status_code: int | None = None + body: BatchOutputBody | None = None + + +class BatchOutputLine(BaseModel): + response: BatchOutputResponse + + +TERMINAL_MARKER_KEY = "litellm_e2e_suite" +TERMINAL_MARKER_VALUE = "batches-terminal-baton" +TERMINAL_POLL_SECONDS = 300.0 +TERMINAL_POLL_INTERVAL_SECONDS = 10.0 +TERMINAL_LIST_LIMIT = 100 +TERMINAL_BAND_MIN_AGE_SECONDS = 25 * 3600 +TERMINAL_BAND_MAX_AGE_SECONDS = 73 * 3600 + + +def _marker_batches(client: BatchClient, key: str) -> list[BatchObject]: + listed = unwrap( + client.list_batches(key=key, model=OPENAI_BATCH_MODEL, limit=TERMINAL_LIST_LIMIT) + ) + return [ + b + for b in listed.data + if (b.metadata or {}).get(TERMINAL_MARKER_KEY) == TERMINAL_MARKER_VALUE + ] + + +def _await_completed_marker( + client: BatchClient, key: str +) -> tuple[BatchObject | None, list[BatchObject]]: + deadline = time.monotonic() + TERMINAL_POLL_SECONDS + while True: + markers = _marker_batches(client, key) + completed = max( + (b for b in markers if b.status == "completed"), + key=lambda b: b.created_at or 0, + default=None, + ) + if completed is not None or time.monotonic() >= deadline: + return completed, markers + time.sleep(TERMINAL_POLL_INTERVAL_SECONDS) + + +def _assert_aged_markers_terminal(markers: list[BatchObject]) -> None: + now = time.time() + stuck = [ + b + for b in markers + if b.created_at is not None + and TERMINAL_BAND_MIN_AGE_SECONDS <= now - b.created_at <= TERMINAL_BAND_MAX_AGE_SECONDS + and b.status not in BATCH_TERMINAL_STATUSES + ] + assert not stuck, ( + "marker batches past their 24h completion window must be terminal; stuck: " + f"{[(b.id, b.status, b.created_at) for b in stuck]}" + ) + + +class TestBatchTerminalState: + """Terminal state + cost write-back via a cross-run marker baton. + + Each run submits a 1-line marker batch (stable metadata key/value plus a + per-run field) and never cancels or deletes it: the marker is the baton the + next run picks up. Polling is list-only for up to 5 minutes because a + retrieve of a non-terminal batch books a $0 spend row whose request_id then + blocks the real-cost row (skip_duplicates); the single retrieve happens only + once a completed marker exists. The assertion target is the newest completed + marker from ANY run, so on the 6h stage cadence the full assertions are + deterministic from run 2 onward. On a cold start (no marker has ever + completed within the poll budget) the test passes on the submission + assertions alone: that is a documented vacuous pass, not a skip, and this + run's marker becomes the next run's target. Markers aged past OpenAI's 24h + completion window (25h-73h band, within the newest list page) must be + terminal. The cost assertion is the LIT-5730 headline: retrieving a + completed model-encoded batch must write a positive spend row keyed + {batch_id}_batch_cost; before the fix the logging worker fetched the + re-encoded output_file_id, 404d, and the row never landed. + """ + + @pytest.mark.covers( + "llm.batches.openai.terminal_state.nonstream.works", + "llm.batches.openai.terminal_state.nonstream.cost_logged", + exercised_on=["batches", "files"], + ) + def test_completed_batch_downloads_output_and_books_cost( + self, client: BatchClient, resources: ResourceManager, batch_deployments: None + ) -> None: + key = resources.key() + file = unwrap( + client.upload_file( + content=render_jsonl("gpt-4o-mini"), + form=FileUploadForm(purpose="batch"), + model=OPENAI_BATCH_MODEL, + key=key, + ) + ) + created = client.create_batch( + body=BatchCreateBody( + input_file_id=file.id, + metadata={ + TERMINAL_MARKER_KEY: TERMINAL_MARKER_VALUE, + "run": unique_marker(), + }, + ), + key=key, + ) + require_successful_call(created) + submitted = BatchObject.model_validate_json(created.body) + assert submitted.status in CREATED_BATCH_STATUSES, ( + f"marker batch has non-transitional status {submitted.status!r}" + ) + assert (submitted.metadata or {}).get(TERMINAL_MARKER_KEY) == TERMINAL_MARKER_VALUE, ( + f"create dropped the marker metadata: {submitted.metadata!r}" + ) + + completed, markers = _await_completed_marker(client, key) + _assert_aged_markers_terminal(markers) + if completed is None: + return + + fetched = retrieve_batch(client, completed.id, key=key, provider=None) + assert fetched.status == "completed", ( + f"listed-completed marker retrieved as {fetched.status!r}" + ) + assert fetched.output_file_id, "completed batch has no output_file_id" + + downloaded = client.proxy.transport.download( + f"/v1/files/{fetched.output_file_id}/content", + headers=client.proxy.transport.bearer(key), + ) + assert downloaded.status_code == 200, ( + f"output content must be 200, got {downloaded.status_code}: {downloaded.body[:300]}" + ) + first_line = BatchOutputLine.model_validate_json(downloaded.body.strip().splitlines()[0]) + assert first_line.response.status_code == 200, ( + f"batch output line reports failure: {downloaded.body[:400]}" + ) + assert first_line.response.body is not None and first_line.response.body.choices, ( + "batch output line has no choices" + ) + + rows = client.proxy.poll_logs_for_request_id( + f"{fetched.id}_batch_cost", + predicate=lambda found: any((row.spend or 0) > 0 for row in found), + ) + priced = [row for row in rows if (row.spend or 0) > 0] + assert priced, ( + f"completed batch {fetched.id} wrote no positive-cost spend row under " + f"request_id {fetched.id}_batch_cost; cost write-back is broken (LIT-5730)" + ) + cost_row = priced[0] + assert cost_row.call_type == "aretrieve_batch", ( + f"batch cost row call_type={cost_row.call_type!r}" + ) + assert (cost_row.total_tokens or 0) > 0, ( + f"batch cost row has no token usage: {cost_row.total_tokens!r}" + ) diff --git a/tests/e2e/coverage_registry/llm_nonconversational.yaml b/tests/e2e/coverage_registry/llm_nonconversational.yaml index bb7169509eb..3de462683d2 100644 --- a/tests/e2e/coverage_registry/llm_nonconversational.yaml +++ b/tests/e2e/coverage_registry/llm_nonconversational.yaml @@ -24,6 +24,13 @@ - {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"} - {id: llm.batches.openai.key_model_access_denied.nonstream.works, module: llm, tier: P0, subject_endpoint: batches, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "test_batches_e2e.py", rationale: "Key model restriction 403 on upload/create"} - {id: llm.batches.openai.input_validation.nonstream.works, module: llm, tier: P1, subject_endpoint: batches, route: openai, capability: input_validation, streaming: nonstream, assertions: [works], source: "vendor strategy §9.18 / LIT-4778", rationale: "Missing input_file_id and invalid batch id rejected"} +- {id: llm.batches.openai.terminal_state.nonstream.works, module: llm, tier: P0, subject_endpoint: batches, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "test_batches_e2e.py / LIT-5730", rationale: "A batch actually reaches completed and its output file downloads through GET /v1/files/{id}/content with per-line provider responses"} +- {id: llm.batches.openai.terminal_state.nonstream.cost_logged, module: llm, tier: P0, subject_endpoint: batches, route: openai, capability: basic, streaming: nonstream, assertions: [cost_logged], source: "test_batches_e2e.py / LIT-5730", fail_before_fix: proven, rationale: "Retrieving a completed model-encoded batch writes a positive spend row keyed {batch_id}_batch_cost (pins LIT-4852/LIT-5666; before the fix the logging worker 404d fetching the re-encoded output_file_id and the row was never written)"} +- {id: llm.batches.openai.malformed_jsonl.nonstream.works, module: llm, tier: P1, subject_endpoint: batches, route: openai, capability: input_validation, streaming: nonstream, assertions: [works], source: "test_batches_e2e.py / LIT-5730", rationale: "Uploading a non-JSON batch file is rejected with a 400 naming the bad line"} +- {id: llm.batches.openai.jsonl_endpoint_mismatch.nonstream.works, module: llm, tier: P1, subject_endpoint: batches, route: openai, capability: input_validation, streaming: nonstream, assertions: [works], source: "test_batches_e2e.py / LIT-5730", rationale: "JSONL line url that contradicts the batch endpoint drives the batch to failed with structured errors, retrieve stays clean, and the terminal retrieve books a zero-cost spend row (LIT-4852)"} +- {id: llm.batches.openai.cancel_terminal.nonstream.works, module: llm, tier: P1, subject_endpoint: batches, route: openai, capability: input_validation, streaming: nonstream, assertions: [works], source: "test_batches_e2e.py / LIT-5730", rationale: "Cancelling an already-terminal batch returns a 409 conflict naming the terminal status"} +- {id: llm.batches.openai.foreign_file_id.nonstream.works, module: llm, tier: P1, subject_endpoint: batches, route: openai, capability: input_validation, streaming: nonstream, assertions: [works], source: "test_batches_e2e.py / LIT-5730", rationale: "Create with one deployment's encoded file id and a conflicting model param routes by the file's embedded model; the returned batch id pins that precedence"} +- {id: llm.batches.openai.second_hop.nonstream.works, module: llm, tier: P0, subject_endpoint: batches, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "test_batches_e2e.py / LIT-5347", rationale: "A litellm_proxy deployment chained to the gateway itself preserves target_model_names through nested unified ids; upload, create, and retrieve work over the two-hop chain (PR #36240)"} - {id: llm.files.openai.upload.nonstream.works, module: llm, tier: P0, subject_endpoint: files, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "openai_files_endpoints/files_endpoints.py:46", rationale: "File upload returns OpenAIFileObject"} - {id: llm.files.openai.input_validation.nonstream.works, module: llm, tier: P1, subject_endpoint: files, route: openai, capability: input_validation, streaming: nonstream, assertions: [works], source: "vendor strategy §9.16 / LIT-4778", rationale: "File upload without purpose rejected"} - {id: llm.files.openai.retrieve.nonstream.works, module: llm, tier: P0, subject_endpoint: files, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "files_endpoints.py", rationale: "File retrieve by id"} @@ -36,6 +43,9 @@ - {id: llm.files.hosted_vllm.upload.nonstream.works, module: llm, tier: P1, subject_endpoint: files, route: hosted_vllm, capability: basic, streaming: nonstream, assertions: [works], source: "test_batches_e2e.py", rationale: "hosted_vllm OpenAI-compatible file upload"} - {id: llm.rerank.cohere.basic.nonstream.works, module: llm, tier: P1, subject_endpoint: rerank, route: cohere, capability: basic, streaming: nonstream, assertions: [works], source: "test_rerank_e2e.py:29", rationale: "Cohere rerank, top_n + relevance_score"} - {id: llm.files.openai.content.nonstream.works, module: llm, tier: P0, subject_endpoint: files, route: openai, capability: basic, streaming: nonstream, assertions: [works], source: "test_batches_e2e.py", rationale: "GET /v1/files/{id}/content returns uploaded batch JSONL bytes"} +- {id: llm.files.azure_openai.content.nonstream.works, module: llm, tier: P0, subject_endpoint: files, route: azure_openai, capability: basic, streaming: nonstream, assertions: [works], source: "test_batches_e2e.py / LIT-5730", rationale: "GET /v1/files/{id}/content on an Azure unified file returns the uploaded JSONL bytes verbatim"} +- {id: llm.files.vertex.content.nonstream.works, module: llm, tier: P0, subject_endpoint: files, route: vertex, capability: basic, streaming: nonstream, assertions: [works], source: "test_batches_e2e.py / LIT-5730", rationale: "GET /v1/files/{id}/content on a Vertex unified file streams the GCS object back (provider-transformed JSONL, so asserts non-empty JSON lines rather than byte equality)"} +- {id: llm.files.bedrock.content.nonstream.works, module: llm, tier: P0, subject_endpoint: files, route: bedrock_converse, capability: basic, streaming: nonstream, assertions: [works], source: "test_batches_e2e.py / LIT-5730", rationale: "GET /v1/files/{id}/content on a Bedrock unified file streams the S3 object back (provider-transformed JSONL, so asserts non-empty JSON lines rather than byte equality)"} - {id: llm.realtime.bedrock_converse.basic.stream.works, module: llm, tier: P0, subject_endpoint: realtime, route: bedrock_converse, capability: basic, streaming: stream, assertions: [works], source: "test_realtime_bedrock_e2e.py", rationale: "Nova Sonic realtime session emits response.done (LIT-2239)"} - {id: llm.google_native.gemini.basic.nonstream.cost_logged, module: llm, tier: P0, subject_endpoint: google_native, route: gemini, capability: basic, streaming: nonstream, assertions: [cost_logged], source: "LIT-4076 / proxy/google_endpoints/endpoints.py", fail_before_fix: proven, rationale: "google-native generateContent must stamp x-litellm-response-cost so SDK traffic reconciles against spend"} - {id: llm.google_native.gemini.basic.stream.works, module: llm, tier: P0, subject_endpoint: google_native, route: gemini, capability: basic, streaming: stream, assertions: [works], source: "PR #28213 / proxy/proxy_server.py async_data_generator", fail_before_fix: proven, rationale: "streamGenerateContent must relay single-prefixed SSE frames with no [DONE] sentinel; doubled data: prefixes and the OpenAI terminator both break the Vertex Java SDK"} diff --git a/tests/test_litellm/batches/test_batch_utils.py b/tests/test_litellm/batches/test_batch_utils.py index ebe093c591c..08cdf945b80 100644 --- a/tests/test_litellm/batches/test_batch_utils.py +++ b/tests/test_litellm/batches/test_batch_utils.py @@ -800,6 +800,43 @@ async def test_output_file_content_vertex_unified_file_id_extracts_gcs_uri(monke assert captured["custom_llm_provider"] == "vertex_ai" +@pytest.mark.asyncio +async def test_output_file_content_model_encoded_file_id_decoded_to_provider_id(monkeypatch): + import litellm.files.main as files_main + from litellm.proxy.openai_files_endpoints.common_utils import encode_file_id_with_model + + captured: dict = {} + + async def fake_afile_content(**kw): + captured.update(kw) + return type("R", (), {"content": b'{"a": 1}'})() + + monkeypatch.setattr(files_main, "afile_content", fake_afile_content) + encoded_id = encode_file_id_with_model("file-Y3FHrMpi7uCkDpY6fgWGeR", "my-batch-model") + + await bu._fetch_batch_output_file_content(_batch(encoded_id), custom_llm_provider="openai") + + assert captured["file_id"] == "file-Y3FHrMpi7uCkDpY6fgWGeR" + assert captured["custom_llm_provider"] == "openai" + + +@pytest.mark.asyncio +async def test_output_file_content_raw_openai_file_id_passes_through(monkeypatch): + import litellm.files.main as files_main + + captured: dict = {} + + async def fake_afile_content(**kw): + captured.update(kw) + return type("R", (), {"content": b'{"a": 1}'})() + + monkeypatch.setattr(files_main, "afile_content", fake_afile_content) + + await bu._fetch_batch_output_file_content(_batch("file-abc123"), custom_llm_provider="openai") + + assert captured["file_id"] == "file-abc123" + + def _vertex_predictions_row(custom_id, prompt_tokens, completion_tokens): return { "request": {