test(batches): pin the all-or-nothing retry and warm the burst model first

An unreadable error file emits no lines and the next retrieve emits all of them.
The burst cell drives one speech model after a warm-up request so a cold worker
cannot fail it, and waits for the batched key spend.

tests/integration/spend/test_batch_parent_id_body_field.py

@@ -237,17 +237,18 @@ def test_any_shape_of_batch_parent_id_on_a_speech_request_is_still_tracked(
 
 def test_a_concurrent_burst_mixing_forged_and_plain_requests_bills_every_request(gateway: Gateway) -> None:
     with gateway.scenario() as scenario:
-        models: Final = tuple(
-            _scripted_model(gateway, scenario, "openai/omni-moderation-latest", _moderation_reply) for _ in range(20)
-        )
-        key: Final = scenario.key(models=list(models))
+        model: Final = _scripted_model(gateway, scenario, "openai/tts-1", _speech_reply)
+        key: Final = scenario.key(models=[model])
+        warm_up: Final = _speech(gateway, model, key, {})
+        assert warm_up.status_code == 200, warm_up.text
+        eventually(lambda: _spend_rows(key), lambda values: len(values) == 1, seconds=70)
         extras: Final = tuple({"batch_parent_id": FORGED_PARENT} if index % 2 else {} for index in range(20))
 
         with ThreadPoolExecutor(max_workers=10) as pool:
-            responses: Final = tuple(
-                pool.map(lambda pair: _moderation(gateway, pair[0], key, pair[1]), zip(models, extras))
-            )
+            responses: Final = tuple(pool.map(lambda extra: _speech(gateway, model, key, extra), extras))
 
         assert [response.status_code for response in responses] == [200] * 20, [r.text for r in responses]
-        rows: Final = eventually(lambda: _spend_rows(key), lambda values: len(values) == 20, seconds=90)
-        assert {row["call_type"] for row in rows} == {"amoderation"}, rows
+        rows: Final = eventually(lambda: _spend_rows(key), lambda values: len(values) == 21, seconds=90)
+        assert {row["call_type"] for row in rows} == {"aspeech"}, rows
+        expected_spend: Final = 21 * float(str(rows[0]["spend"]))
+        eventually(lambda: _key_spend(key), lambda value: value == pytest.approx(expected_spend), seconds=90)

tests/unit/batches/test_batch_line_item_logging.py

@@ -1241,6 +1241,40 @@ async def test_supplied_result_files_skip_the_output_and_error_fetch(recorder):
     assert file_mock.await_args.kwargs["file_id"] == "input-file-1"
 
 
+def _error_file_unreadable(file_id: str, **_kwargs):
+    if file_id == "error-file-1":
+        raise ValueError("error file gone")
+    return SimpleNamespace(content=_FILE_BYTES[file_id])
+
+
+@pytest.mark.asyncio
+async def test_an_unreadable_error_file_defers_every_line_to_the_next_retrieve(recorder):
+    claim_cache: Final = DualCache()
+    batch: Final = _batch()
+    parent: Final = _parent_logging()
+    emitted: list[int] = []  # mutable-ok: collects one count per retrieve
+    for side_effect in (_error_file_unreadable, _file_content):
+        with patch(
+            "litellm.files.main.afile_content", AsyncMock(side_effect=side_effect)
+        ):  # test-quality-ok: afile_content is the provider boundary; no injection seam for managed file fetch
+            emitted.append(
+                await log_batch_line_items(
+                    batch=batch,
+                    custom_llm_provider="openai",
+                    parent=parent,
+                    model_name="gpt-4o",
+                    litellm_params=None,
+                    model_info=None,
+                    result_files=BatchResultFiles(output=OUTPUT_JSONL, error=None),
+                    claim_cache=claim_cache,
+                )
+            )
+
+    assert emitted == [0, 2]
+    assert len(recorder.success_events) == 1
+    assert len(recorder.failure_events) == 1
+
+
 @pytest.mark.asyncio
 async def test_empty_result_files_fall_back_to_fetching_everything(recorder):
     file_mock: Final = AsyncMock(side_effect=_file_content)