diff --git a/litellm/batches/batch_utils.py b/litellm/batches/batch_utils.py
index 97be5f77d79..eaac3bf0e9f 100644
--- a/litellm/batches/batch_utils.py
+++ b/litellm/batches/batch_utils.py
@@ -25,6 +25,26 @@ class BatchCostUsageResult:
     failed_requests: int
 
 
+_TERMINAL_BATCH_STATUSES: Final = frozenset({"completed", "failed", "cancelled", "expired"})
+
+
+def batch_cost_is_final(batch: Batch) -> bool:
+    """Whether this retrieve of the batch is the one to account its cost from.
+
+    A batch still in flight has nothing to price, and a "completed" batch can report
+    no output_file_id for a moment before the output populates; pricing either records
+    $0 under the batch's single spend row and pins it there. Final means a completed
+    batch whose output file has arrived or whose counts prove no line succeeded, or
+    any other terminal status (failed, cancelled, expired).
+    """
+    if batch.status not in _TERMINAL_BATCH_STATUSES:
+        return False
+    if batch.status != "completed" or batch.output_file_id is not None:
+        return True
+    request_counts: Final = batch.request_counts
+    return request_counts is not None and request_counts.total > 0 and request_counts.completed == 0
+
+
 async def calculate_batch_cost_and_usage(
     file_content_dictionary: list[dict],
     custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm", "anthropic"],
diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py
index c31c4323157..09ddd1b9720 100644
--- a/litellm/litellm_core_utils/litellm_logging.py
+++ b/litellm/litellm_core_utils/litellm_logging.py
@@ -36,7 +36,7 @@ from litellm._logging import (
     verbose_logger,
 )
 from litellm._uuid import uuid
-from litellm.batches.batch_utils import _handle_completed_batch
+from litellm.batches.batch_utils import _handle_completed_batch, batch_cost_is_final
 from litellm.caching.caching import DualCache, InMemoryCache
 from litellm.caching.caching_handler import LLMCachingHandler
 from litellm.constants import (
@@ -2899,13 +2899,6 @@ class Logging(LiteLLMLoggingBaseClass):
             ):  # polling job will query these frequently, don't spam db logs
                 return
 
-            from litellm.proxy.openai_files_endpoints.common_utils import (
-                _is_base64_encoded_unified_file_id,
-            )
-
-            # check if file id is a unified file id
-            is_base64_unified_file_id: Final = _is_base64_encoded_unified_file_id(result.id)
-
             batch_cost: Final = kwargs.get("batch_cost", None)
             batch_usage = kwargs.get("batch_usage", None)
             batch_models = kwargs.get("batch_models", None)
@@ -2913,9 +2906,7 @@ class Logging(LiteLLMLoggingBaseClass):
             batch_failed_requests: Final = kwargs.get("batch_failed_requests", None)
             has_explicit_batch_data: Final = all(x is not None for x in (batch_cost, batch_usage, batch_models))
 
-            should_compute_batch_data: Final = (
-                not is_base64_unified_file_id or not has_explicit_batch_data and result.status == "completed"
-            )
+            should_compute_batch_data: Final = not has_explicit_batch_data and batch_cost_is_final(result)
             if has_explicit_batch_data:
                 result._hidden_params["response_cost"] = batch_cost
                 result._hidden_params["batch_models"] = batch_models
diff --git a/litellm/proxy/db/db_spend_update_writer.py b/litellm/proxy/db/db_spend_update_writer.py
index e6880d521f1..ff48b00dc70 100644
--- a/litellm/proxy/db/db_spend_update_writer.py
+++ b/litellm/proxy/db/db_spend_update_writer.py
@@ -82,6 +82,7 @@ else:
 
 
 RESPONSES_SESSION_CALL_TYPES: Final = frozenset({CallTypes.responses.value, CallTypes.aresponses.value})
+IMMEDIATE_FLUSH_CALL_TYPES: Final = RESPONSES_SESSION_CALL_TYPES | frozenset({CallTypes.aretrieve_batch.value})
 
 
 class _SpendBatch(Protocol):
@@ -939,7 +940,7 @@ class DBSpendUpdateWriter:
             from litellm.proxy.utils import enqueue_spend_logs, request_spend_log_flush
 
             await enqueue_spend_logs(prisma_client, (payload,))
-            if payload.get("call_type") in RESPONSES_SESSION_CALL_TYPES:
+            if payload.get("call_type") in IMMEDIATE_FLUSH_CALL_TYPES:
                 request_spend_log_flush()
         else:
             verbose_proxy_logger.debug("prisma_client is None. Skipping writing spend logs to db.")
diff --git a/litellm/proxy/hooks/proxy_track_cost_callback.py b/litellm/proxy/hooks/proxy_track_cost_callback.py
index 7254b05db2e..95e61fdd98c 100644
--- a/litellm/proxy/hooks/proxy_track_cost_callback.py
+++ b/litellm/proxy/hooks/proxy_track_cost_callback.py
@@ -6,6 +6,7 @@ from typing import TYPE_CHECKING, Any, Final, cast
 
 import litellm
 from litellm._logging import verbose_proxy_logger
+from litellm.batches.batch_utils import batch_cost_is_final
 from litellm.constants import BACKGROUND_INTERACTION_COST_POLLING_ENABLED
 from litellm.integrations.custom_logger import CustomLogger
 from litellm.litellm_core_utils.core_helpers import (
@@ -33,17 +34,21 @@ from litellm.proxy.spend_tracking.spend_log_error_logger import (
 from litellm.proxy.spend_tracking.spend_tracking_utils import (
     _sanitize_error_information_for_spend_logs,
     get_request_model_access_groups,
+    get_spend_logs_id,
 )
 from litellm.proxy.utils import ProxyUpdateSpend
 from litellm.types.utils import (
     CallTypes,
+    LiteLLMBatch,
     StandardLoggingPayload,
     StandardLoggingPayloadErrorInformation,
 )
 from litellm.utils import get_end_user_id_for_cost_tracking
 
 if TYPE_CHECKING:
-    from litellm.proxy.utils import ProxyLogging
+    from prisma.types import LiteLLM_SpendLogsWhereUniqueInput
+
+    from litellm.proxy.utils import PrismaClient, ProxyLogging
 
 _UNATTRIBUTED_TRACKABLE_CALL_TYPES: Final[frozenset[str]] = frozenset(
     {
@@ -224,6 +229,7 @@ class _ProxyDBLogger(CustomLogger):
     ):
         from litellm.proxy.proxy_server import (
             increment_spend_counters,
+            prisma_client,
             proxy_logging_obj,
             update_cache,
         )
@@ -248,6 +254,18 @@ class _ProxyDBLogger(CustomLogger):
                 )
                 _write_spend_metadata_to_kwargs(kwargs=kwargs, metadata=metadata)
             budget_reservation: Final = _get_budget_reservation_from_metadata(metadata=metadata)
+            if (
+                isinstance(completion_response, LiteLLMBatch)
+                and kwargs.get("call_type") == CallTypes.aretrieve_batch.value
+            ):
+                batch_spend_log_id: Final = get_spend_logs_id(
+                    CallTypes.aretrieve_batch.value, completion_response.model_dump(), kwargs
+                )
+                if not await _batch_cost_is_trackable_now(
+                    batch=completion_response, spend_log_id=batch_spend_log_id, prisma_client=prisma_client
+                ):
+                    await _release_budget_reservation(budget_reservation=budget_reservation)
+                    return
             user_id: Final = cast(str | None, metadata.get("user_api_key_user_id", None))
             team_id: Final = cast(str | None, metadata.get("user_api_key_team_id", None))
             org_id: Final = cast(str | None, metadata.get("user_api_key_org_id", None))
@@ -491,6 +509,42 @@ def _write_spend_metadata_to_kwargs(kwargs: dict, metadata: dict) -> None:
                     bucket[key] = value
 
 
+async def _batch_cost_is_trackable_now(
+    batch: LiteLLMBatch, spend_log_id: str | None, prisma_client: "PrismaClient | None"
+) -> bool:
+    """A batch is billed exactly once, from the first retrieve that sees it final.
+
+    Every retrieve of one batch shares a single spend row (its id plus the batch cost
+    suffix), so a poll that lands before the output exists would write that row at $0
+    and pin it there, and every retrieve after the first would add the cost to the
+    key, team, and user counters again.
+    """
+    if not batch_cost_is_final(batch):
+        verbose_proxy_logger.debug("Cost tracking deferred for batch %s still in status %s", batch.id, batch.status)
+        return False
+    if prisma_client is None or spend_log_id is None:
+        return True
+    if not await _spend_log_already_recorded(prisma_client=prisma_client, request_id=spend_log_id):
+        return True
+    verbose_proxy_logger.debug(
+        "Cost tracking skipped for batch %s: spend row %s already recorded", batch.id, spend_log_id
+    )
+    return False
+
+
+async def _spend_log_already_recorded(prisma_client: "PrismaClient", request_id: str) -> bool:
+    from litellm.proxy.utils import spend_log_is_queued
+
+    if await spend_log_is_queued(prisma_client, request_id):
+        return True
+    spend_log_row: Final[LiteLLM_SpendLogsWhereUniqueInput] = {"request_id": request_id}
+    try:
+        return await prisma_client.db.litellm_spendlogs.find_unique(where=spend_log_row) is not None
+    except Exception as e:  # noqa: BLE001  # prisma raises its own hierarchy; an unreadable DB must not drop the batch's only spend row
+        verbose_proxy_logger.warning("Could not check for an existing spend row %s, tracking anyway: %s", request_id, e)
+        return False
+
+
 def _is_unbilled_interaction_response(completion_response: object) -> bool:
     from litellm.interactions.background_cost_polling import missing_usage_is_expected
     from litellm.types.interactions import InteractionsAPIResponse
diff --git a/litellm/proxy/openai_files_endpoints/common_utils.py b/litellm/proxy/openai_files_endpoints/common_utils.py
index 15eeddbc489..b1f282a0978 100644
--- a/litellm/proxy/openai_files_endpoints/common_utils.py
+++ b/litellm/proxy/openai_files_endpoints/common_utils.py
@@ -15,6 +15,7 @@ from typing import (
     runtime_checkable,
 )
 
+from litellm.batches.batch_utils import batch_cost_is_final
 from litellm.proxy._types import ProxyException
 from litellm.repositories.table_repositories import (
     ManagedFileRepository,
@@ -1357,12 +1358,7 @@ def _completed_batch_safe_to_retire(response: "LiteLLMBatch") -> bool:
     enumerated the batch and none succeeded. A zero or unknown total means counts
     are unreported, so stay eligible and let the next poller pass revisit it. (#37713)
     """
-    if response.output_file_id is not None:
-        return True
-    request_counts = response.request_counts
-    if request_counts is None:
-        return False
-    return request_counts.total > 0 and request_counts.completed == 0
+    return batch_cost_is_final(response)
 
 
 async def update_batch_in_database(
diff --git a/litellm/proxy/utils.py b/litellm/proxy/utils.py
index accf7b720fb..f64b51bc6c6 100644
--- a/litellm/proxy/utils.py
+++ b/litellm/proxy/utils.py
@@ -6251,7 +6251,9 @@ def request_spend_log_flush() -> None:
 
     The Responses API hands the client an id it can chain from straight away, and that
     lookup reads the DB, so the row cannot sit in this worker's queue for a poll interval.
-    Repeated requests coalesce into the monitor's next pass, so the batching holds.
+    A batch's cost row is what every other worker checks before charging the same batch
+    again, so it cannot wait either. Repeated requests coalesce into the monitor's next
+    pass, so the batching holds.
     """
     PrismaClient.spend_log_flush_requested.set()
 
@@ -6266,6 +6268,12 @@ async def _wait_for_spend_log_flush_request(interval: float) -> bool:
     return True
 
 
+async def spend_log_is_queued(prisma_client: PrismaClient, request_id: str) -> bool:
+    """Whether a spend log with ``request_id`` is still waiting for the next flush."""
+    async with prisma_client._spend_log_transactions_lock:
+        return any(row.get("request_id") == request_id for row in prisma_client.spend_log_transactions)
+
+
 async def dequeue_spend_logs(prisma_client: PrismaClient, limit: int) -> list[dict[str, object]]:
     """Take up to ``limit`` of the oldest queued spend logs off the queue.
 
diff --git a/tests/test_litellm/batches/test_batch_utils.py b/tests/test_litellm/batches/test_batch_utils.py
index c86c7c4df03..8d4f68164b4 100644
--- a/tests/test_litellm/batches/test_batch_utils.py
+++ b/tests/test_litellm/batches/test_batch_utils.py
@@ -21,11 +21,12 @@ from types import MappingProxyType
 import httpx
 import pytest
 import respx
+from openai.types.batch import BatchRequestCounts
 
 
 import litellm
 import litellm.batches.batch_utils as bu
-from litellm.types.utils import Usage
+from litellm.types.utils import LiteLLMBatch, Usage
 
 # --------------------------------------------------------------------------- #
 # Builders for batch OUTPUT file rows.
@@ -1718,3 +1719,55 @@ def test_unparsable_bedrock_batch_usage_warns(caplog):
     assert usage.total_tokens == 0
     assert "does not understand" in caplog.text
     assert "inputTextTokenCount" in caplog.text
+
+
+# --------------------------------------------------------------------------- #
+# batch_cost_is_final
+# --------------------------------------------------------------------------- #
+
+def _retrieved_batch(
+    status: str, output_file_id: str | None = None, counts: BatchRequestCounts | None = None
+) -> LiteLLMBatch:
+    return LiteLLMBatch(
+        id="batch_abc",
+        completion_window="24h",
+        created_at=1,
+        endpoint="/v1/chat/completions",
+        input_file_id="file-in",
+        object="batch",
+        status=status,
+        output_file_id=output_file_id,
+        request_counts=counts,
+    )
+
+
+class TestBatchCostIsFinal:
+    """Every retrieve of one batch writes the same spend row, so the first retrieve
+    that prices it decides the row for good. A poll before the output exists must
+    therefore not count as final: pricing it recorded $0 and pinned it (LIT-7048)."""
+
+    @pytest.mark.parametrize("status", ["validating", "in_progress", "finalizing", "cancelling"])
+    def test_in_flight_batch_is_not_final(self, status):
+        assert bu.batch_cost_is_final(_retrieved_batch(status)) is False
+
+    def test_completed_with_output_is_final(self):
+        assert bu.batch_cost_is_final(_retrieved_batch("completed", output_file_id="file-out")) is True
+
+    def test_completed_without_output_and_unknown_counts_is_not_final(self):
+        assert bu.batch_cost_is_final(_retrieved_batch("completed")) is False
+
+    def test_completed_without_output_and_zero_counts_is_not_final(self):
+        counts = BatchRequestCounts(total=0, completed=0, failed=0)
+        assert bu.batch_cost_is_final(_retrieved_batch("completed", counts=counts)) is False
+
+    def test_completed_without_output_but_successful_lines_is_not_final(self):
+        counts = BatchRequestCounts(total=2, completed=2, failed=0)
+        assert bu.batch_cost_is_final(_retrieved_batch("completed", counts=counts)) is False
+
+    def test_completed_without_output_and_every_line_failed_is_final(self):
+        counts = BatchRequestCounts(total=2, completed=0, failed=2)
+        assert bu.batch_cost_is_final(_retrieved_batch("completed", counts=counts)) is True
+
+    @pytest.mark.parametrize("status", ["failed", "expired", "cancelled"])
+    def test_other_terminal_statuses_are_final(self, status):
+        assert bu.batch_cost_is_final(_retrieved_batch(status)) is True
diff --git a/tests/test_litellm/litellm_core_utils/test_litellm_logging.py b/tests/test_litellm/litellm_core_utils/test_litellm_logging.py
index 16a99713a06..4583429bd11 100644
--- a/tests/test_litellm/litellm_core_utils/test_litellm_logging.py
+++ b/tests/test_litellm/litellm_core_utils/test_litellm_logging.py
@@ -632,6 +632,88 @@ class TestRetrieveBatchCostPassesModelIdentity:
         assert captured["model_info"]["input_cost_per_token"] == 0.0
 
 
+class TestRetrieveBatchPricesOnlyFinalBatches:
+    """Regression (LIT-7048): retrieving a provider-id batch priced it on every poll.
+
+    Every retrieve of one batch logs under the same spend row, so pricing a poll
+    that landed before the output existed wrote that row at $0 and pinned it there.
+    Only a final batch gets priced; an in-flight poll carries no cost at all.
+    """
+
+    @staticmethod
+    def _logging_obj() -> LitellmLogging:
+        obj = LitellmLogging(
+            model="gpt-5.6-luna",
+            messages=[{"role": "user", "content": "Hey"}],
+            stream=False,
+            call_type="aretrieve_batch",
+            start_time=time.time(),
+            litellm_call_id="batch-call-2",
+            function_id="f",
+        )
+        obj.custom_llm_provider = "openai"
+        return obj
+
+    @staticmethod
+    def _batch(status: str, output_file_id: str | None):
+        from litellm.types.utils import LiteLLMBatch
+
+        return LiteLLMBatch(
+            id="batch_6a9c99e185588190877d391f8b9d7f8a",
+            completion_window="24h",
+            created_at=1,
+            endpoint="/v1/chat/completions",
+            input_file_id="file-in",
+            object="batch",
+            status=status,
+            output_file_id=output_file_id,
+        )
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        ("status", "output_file_id"),
+        [("validating", None), ("in_progress", None), ("finalizing", None), ("completed", None)],
+    )
+    async def test_non_final_batch_is_not_priced(self, monkeypatch, status, output_file_id) -> None:
+        from litellm.litellm_core_utils import litellm_logging as logging_module
+
+        handle_completed_batch = AsyncMock()
+        monkeypatch.setattr(logging_module, "_handle_completed_batch", handle_completed_batch)
+        batch = self._batch(status, output_file_id)
+
+        with contextlib.suppress(Exception):
+            await self._logging_obj()._async_success_handler_body(result=batch, start_time=None, end_time=None)
+
+        handle_completed_batch.assert_not_awaited()
+        assert "response_cost" not in batch._hidden_params
+
+    @pytest.mark.asyncio
+    async def test_completed_batch_with_output_is_priced(self, monkeypatch) -> None:
+        from litellm.batches.batch_utils import BatchCostUsageResult
+        from litellm.litellm_core_utils import litellm_logging as logging_module
+        from litellm.types.utils import Usage
+
+        handle_completed_batch = AsyncMock(
+            return_value=BatchCostUsageResult(
+                cost=8e-06,
+                usage=Usage(prompt_tokens=26, completion_tokens=9, total_tokens=35),
+                models=["gpt-5.6-luna"],
+                successful_requests=2,
+                failed_requests=0,
+            )
+        )
+        monkeypatch.setattr(logging_module, "_handle_completed_batch", handle_completed_batch)
+        batch = self._batch("completed", "file-out")
+
+        with contextlib.suppress(Exception):
+            await self._logging_obj()._async_success_handler_body(result=batch, start_time=None, end_time=None)
+
+        handle_completed_batch.assert_awaited_once()
+        assert batch._hidden_params["response_cost"] == 8e-06
+        assert batch.usage is not None
+        assert batch.usage.total_tokens == 35
+
+
 class TestAnthropicPassthroughCustomPricing:
     """Verify the Anthropic pass-through handler forwards custom pricing."""
 
diff --git a/tests/test_litellm/proxy/db/test_db_spend_update_writer.py b/tests/test_litellm/proxy/db/test_db_spend_update_writer.py
index 11ef911de3e..e1b2d151c6d 100644
--- a/tests/test_litellm/proxy/db/test_db_spend_update_writer.py
+++ b/tests/test_litellm/proxy/db/test_db_spend_update_writer.py
@@ -2934,12 +2934,16 @@ async def test_commit_spend_updates_retries_deadlock_on_every_entity_path(monkey
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     "call_type, expects_flush",
-    [("aresponses", True), ("responses", True), ("acompletion", False)],
+    [("aresponses", True), ("responses", True), ("aretrieve_batch", True), ("acompletion", False)],
 )
-async def test_insert_spend_log_asks_for_an_immediate_flush_on_responses_calls(call_type: str, expects_flush: bool):
+async def test_insert_spend_log_asks_for_an_immediate_flush_on_rows_other_workers_read_back(
+    call_type: str, expects_flush: bool
+):
     """
     A `previous_response_id` chained straight off the previous turn reads the DB, so a
-    Responses row cannot sit in this worker's queue until the monitor's next poll.
+    Responses row cannot sit in this worker's queue until the monitor's next poll. A
+    batch's cost row is what another worker checks before charging the same batch again
+    (LIT-7048), so it cannot wait either.
     """
     from litellm.proxy.utils import PrismaClient
 
diff --git a/tests/test_litellm/proxy/hooks/test_proxy_track_cost_callback.py b/tests/test_litellm/proxy/hooks/test_proxy_track_cost_callback.py
index 8043a1aca3f..b7037e8d621 100644
--- a/tests/test_litellm/proxy/hooks/test_proxy_track_cost_callback.py
+++ b/tests/test_litellm/proxy/hooks/test_proxy_track_cost_callback.py
@@ -1,4 +1,3 @@
-
 from datetime import datetime
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -7,6 +6,7 @@ import pytest
 from litellm.litellm_core_utils.internal_call_metadata import MODEL_ACCESS_GROUP_METADATA_KEY
 from litellm.proxy._types import UserAPIKeyAuth
 from litellm.proxy.hooks.proxy_track_cost_callback import (
+    _batch_cost_is_trackable_now,
     _get_budget_reservation_from_metadata,
     _ProxyDBLogger,
     _should_track_cost_callback,
@@ -70,9 +70,7 @@ async def test_async_post_call_failure_hook():
 
         # Check that metadata was properly updated
         assert "litellm_params" in call_args["kwargs"]
-        assert call_args["kwargs"]["litellm_params"]["proxy_server_request"] == {
-            "request_id": "test_request_id"
-        }
+        assert call_args["kwargs"]["litellm_params"]["proxy_server_request"] == {"request_id": "test_request_id"}
         metadata = call_args["kwargs"]["litellm_params"]["metadata"]
         assert metadata["user_api_key"] == "test_api_key"
         assert metadata["status"] == "failure"
@@ -336,9 +334,7 @@ async def test_should_continue_failure_tracking_when_budget_release_fails():
         )
         assert mock_invalidate_budget_reservation_counters.await_count == 1
         assert (
-            mock_invalidate_budget_reservation_counters.await_args.kwargs[
-                "budget_reservation"
-            ]
+            mock_invalidate_budget_reservation_counters.await_args.kwargs["budget_reservation"]
             is user_api_key_dict.budget_reservation
         )
         assert user_api_key_dict.budget_reservation["finalized"] is True
@@ -433,36 +429,21 @@ def test_get_budget_reservation_from_metadata_handles_dict_auth_object():
         "entries": [{"counter_key": "spend:key:test_api_key"}],
     }
 
+    assert _get_budget_reservation_from_metadata(metadata={"user_api_key_auth": dict(UserAPIKeyAuth())}) is None
     assert (
         _get_budget_reservation_from_metadata(
-            metadata={"user_api_key_auth": dict(UserAPIKeyAuth())}
-        )
-        is None
-    )
-    assert (
-        _get_budget_reservation_from_metadata(
-            metadata={
-                "user_api_key_auth": UserAPIKeyAuth(
-                    budget_reservation=budget_reservation
-                )
-            }
+            metadata={"user_api_key_auth": UserAPIKeyAuth(budget_reservation=budget_reservation)}
         )
         == budget_reservation
     )
     assert (
         _get_budget_reservation_from_metadata(
-            metadata={
-                "user_api_key_auth": dict(
-                    UserAPIKeyAuth(budget_reservation=budget_reservation)
-                )
-            }
+            metadata={"user_api_key_auth": dict(UserAPIKeyAuth(budget_reservation=budget_reservation))}
         )
         == budget_reservation
     )
     assert (
-        _get_budget_reservation_from_metadata(
-            metadata={"user_api_key_budget_reservation": budget_reservation}
-        )
+        _get_budget_reservation_from_metadata(metadata={"user_api_key_budget_reservation": budget_reservation})
         is budget_reservation
     )
 
@@ -470,9 +451,7 @@ def test_get_budget_reservation_from_metadata_handles_dict_auth_object():
 @pytest.mark.asyncio
 async def test_update_database_and_spend_counters_releases_reservation_when_db_update_fails():
     proxy_logging_obj = MagicMock()
-    proxy_logging_obj.db_spend_update_writer.update_database = AsyncMock(
-        side_effect=Exception("db unavailable")
-    )
+    proxy_logging_obj.db_spend_update_writer.update_database = AsyncMock(side_effect=Exception("db unavailable"))
     increment_spend_counters = AsyncMock()
     budget_reservation = {"reserved_cost": 0.5, "entries": []}
 
@@ -508,9 +487,7 @@ async def test_update_database_and_spend_counters_releases_reservation_when_db_u
 async def test_update_database_and_spend_counters_preserves_db_exception_when_release_fails():
     proxy_logging_obj = MagicMock()
     db_exception = RuntimeError("db unavailable")
-    proxy_logging_obj.db_spend_update_writer.update_database = AsyncMock(
-        side_effect=db_exception
-    )
+    proxy_logging_obj.db_spend_update_writer.update_database = AsyncMock(side_effect=db_exception)
     increment_spend_counters = AsyncMock()
     budget_reservation = {"reserved_cost": 0.5, "entries": []}
 
@@ -554,12 +531,8 @@ async def test_update_database_and_spend_counters_preserves_db_exception_when_re
             budget_reservation=budget_reservation,
         )
         assert mock_log_exception.call_count == 2
-        mock_log_exception.assert_any_call(
-            "Failed to release budget reservation after database update failed"
-        )
-        mock_log_exception.assert_any_call(
-            "Failed to invalidate budget reservation counters after release failed"
-        )
+        mock_log_exception.assert_any_call("Failed to release budget reservation after database update failed")
+        mock_log_exception.assert_any_call("Failed to invalidate budget reservation counters after release failed")
 
     increment_spend_counters.assert_not_awaited()
 
@@ -778,6 +751,169 @@ async def test_track_cost_callback_defers_in_progress_background_interaction():
         mock_proxy_logging.failed_tracking_alert.assert_not_called()
 
 
+def _batch_retrieve_kwargs(call_type: str, reservation: dict | None = None) -> dict:
+    metadata = {
+        "user_api_key": "hashed_key",
+        "user_api_key_user_id": "user-1",
+        "user_api_key_team_id": "team-1",
+        **({"user_api_key_budget_reservation": reservation} if reservation is not None else {}),
+    }
+    return {
+        "call_type": call_type,
+        "model": "gpt-5.6-luna",
+        "litellm_call_id": "test-call-id",
+        "litellm_params": {"metadata": metadata},
+        "standard_logging_object": {"response_cost": 0.0, "request_tags": None},
+        "stream": False,
+    }
+
+
+def _retrieved_batch(status: str, output_file_id: str | None):
+    from litellm.types.utils import LiteLLMBatch
+
+    return LiteLLMBatch(
+        id="batch_abc",
+        completion_window="24h",
+        created_at=1,
+        endpoint="/v1/chat/completions",
+        input_file_id="file-in",
+        object="batch",
+        status=status,
+        output_file_id=output_file_id,
+    )
+
+
+def _prisma_client_with(queued_request_ids: tuple[str, ...], stored_row: object) -> MagicMock:
+    import asyncio
+
+    prisma_client = MagicMock()
+    prisma_client._spend_log_transactions_lock = asyncio.Lock()
+    prisma_client.spend_log_transactions = [{"request_id": request_id} for request_id in queued_request_ids]
+    prisma_client.db.litellm_spendlogs.find_unique = AsyncMock(return_value=stored_row)
+    return prisma_client
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("status", "output_file_id", "spend_log_id", "prisma_client", "trackable"),
+    [
+        ("in_progress", None, "batch_abc_batch_cost", None, False),
+        ("in_progress", None, "batch_abc_batch_cost", _prisma_client_with((), None), False),
+        ("completed", None, "batch_abc_batch_cost", _prisma_client_with((), None), False),
+        ("completed", "file-out", "batch_abc_batch_cost", None, True),
+        ("completed", "file-out", None, _prisma_client_with((), None), True),
+        ("completed", "file-out", "batch_abc_batch_cost", _prisma_client_with(("batch_abc_batch_cost",), None), False),
+        (
+            "completed",
+            "file-out",
+            "batch_abc_batch_cost",
+            _prisma_client_with((), {"request_id": "batch_abc_batch_cost"}),
+            False,
+        ),
+        ("completed", "file-out", "batch_abc_batch_cost", _prisma_client_with((), None), True),
+        ("failed", None, "batch_abc_batch_cost", _prisma_client_with((), None), True),
+    ],
+    ids=[
+        "in_progress_without_db",
+        "in_progress_never_consults_db",
+        "completed_without_output_yet",
+        "final_without_db",
+        "final_without_spend_log_id",
+        "final_row_queued_for_flush",
+        "final_row_already_stored",
+        "final_first_sighting",
+        "failed_first_sighting",
+    ],
+)
+async def test_batch_cost_is_trackable_now(status, output_file_id, spend_log_id, prisma_client, trackable):
+    """
+    A batch is billed from the first retrieve that sees it final and never again:
+    a poll before that wrote the shared spend row at $0 and pinned it there, and
+    every completed retrieve after the first charged the key again (LIT-7048).
+    """
+    assert (
+        await _batch_cost_is_trackable_now(
+            batch=_retrieved_batch(status, output_file_id), spend_log_id=spend_log_id, prisma_client=prisma_client
+        )
+        is trackable
+    )
+
+
+@pytest.mark.asyncio
+async def test_batch_cost_is_trackable_now_when_the_spend_row_lookup_fails():
+    """An unreadable spend log table must not drop the batch's only spend row."""
+    prisma_client = _prisma_client_with((), None)
+    prisma_client.db.litellm_spendlogs.find_unique = AsyncMock(side_effect=RuntimeError("db unreachable"))
+
+    assert (
+        await _batch_cost_is_trackable_now(
+            batch=_retrieved_batch("completed", "file-out"),
+            spend_log_id="batch_abc_batch_cost",
+            prisma_client=prisma_client,
+        )
+        is True
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("call_type", "status", "output_file_id", "stored_row", "charged"),
+    [
+        ("aretrieve_batch", "in_progress", None, None, False),
+        ("aretrieve_batch", "completed", "file-out", {"request_id": "batch_abc_batch_cost"}, False),
+        ("aretrieve_batch", "completed", "file-out", None, True),
+        ("acreate_batch", "validating", None, None, True),
+    ],
+    ids=["retrieve_before_final", "retrieve_already_recorded", "retrieve_first_final", "create_before_final"],
+)
+async def test_track_cost_callback_charges_a_batch_once_and_only_when_final(  # test-quality-ok: whether the spend writer runs and whether the poll's reservation is handed back is the whole observable contract of the gate
+    call_type, status, output_file_id, stored_row, charged
+):
+    """
+    Only retrieves are gated, since creating a batch is its own billable request.
+    A retrieve that writes nothing hands its budget reservation back instead.
+    """
+    logger = _ProxyDBLogger()
+    budget_reservation = None if charged else {"reserved_cost": 0.5, "entries": []}
+    kwargs = _batch_retrieve_kwargs(call_type, reservation=budget_reservation)
+
+    with (
+        patch(  # test-quality-ok: prisma_client is a proxy_server global the callback reads lazily, no seam
+            "litellm.proxy.proxy_server.prisma_client", _prisma_client_with((), stored_row)
+        ),
+        patch(  # test-quality-ok: increment_spend_counters is a proxy_server global the callback reads lazily, no seam
+            "litellm.proxy.proxy_server.increment_spend_counters", new_callable=AsyncMock
+        ),
+        patch(  # test-quality-ok: update_cache is a proxy_server global the callback reads lazily, no seam
+            "litellm.proxy.proxy_server.update_cache", new_callable=AsyncMock
+        ),
+        patch(  # test-quality-ok: callback imports proxy_logging_obj off proxy_server in its body, no seam
+            "litellm.proxy.proxy_server.proxy_logging_obj"
+        ) as mock_proxy_logging,
+        patch(  # test-quality-ok: the release is imported inside the callback's helper, no seam
+            "litellm.proxy.spend_tracking.budget_reservation.release_budget_reservation", new_callable=AsyncMock
+        ) as mock_release_budget_reservation,
+    ):
+        mock_proxy_logging.failed_tracking_alert = AsyncMock()
+        mock_proxy_logging.db_spend_update_writer.update_database = AsyncMock()
+        mock_proxy_logging.slack_alerting_instance.customer_spend_alert = AsyncMock()
+
+        await logger._PROXY_track_cost_callback(
+            kwargs=kwargs,
+            completion_response=_retrieved_batch(status, output_file_id),
+            start_time=datetime.now(),
+            end_time=datetime.now(),
+        )
+
+        mock_proxy_logging.failed_tracking_alert.assert_not_called()
+        if charged:
+            mock_proxy_logging.db_spend_update_writer.update_database.assert_awaited_once()
+            mock_release_budget_reservation.assert_not_awaited()
+        else:
+            mock_proxy_logging.db_spend_update_writer.update_database.assert_not_called()
+            mock_release_budget_reservation.assert_awaited_once_with(budget_reservation=budget_reservation)
+
+
 def _in_progress_interaction_kwargs(reservation: dict) -> dict:
     return {
         "call_type": "acreate_interaction",
@@ -1101,10 +1237,7 @@ async def test_async_post_call_failure_hook_propagates_trace_id_from_logging_obj
 
         # standard_logging_object should have been propagated from logging obj
         assert call_kwargs.get("standard_logging_object") is not None
-        assert (
-            call_kwargs["standard_logging_object"]["trace_id"]
-            == "trace-id-from-logging-obj"
-        )
+        assert call_kwargs["standard_logging_object"]["trace_id"] == "trace-id-from-logging-obj"
         # litellm_trace_id should also be propagated as a fallback
         assert call_kwargs.get("litellm_trace_id") == "trace-id-from-logging-obj"
 
@@ -1691,9 +1824,7 @@ async def test_async_post_call_failure_hook_records_recovered_partial_spend():
         "metadata": {},
         "proxy_server_request": {"request_id": "rid"},
         "response_cost": 3.5e-05,
-        "combined_usage_object": Usage(
-            prompt_tokens=30, completion_tokens=1, total_tokens=31
-        ),
+        "combined_usage_object": Usage(prompt_tokens=30, completion_tokens=1, total_tokens=31),
     }
 
     with patch(
@@ -1772,15 +1903,10 @@ async def test_track_cost_callback_enriches_user_id_for_mcp_style_metadata():
         assert mock_increment.call_args.kwargs["team_id"] == "team-123"
         assert mock_increment.call_args.kwargs["org_id"] == "org-456"
 
-        update_kwargs = (
-            mock_proxy_logging.db_spend_update_writer.update_database.await_args.kwargs
-        )
+        update_kwargs = mock_proxy_logging.db_spend_update_writer.update_database.await_args.kwargs
         assert update_kwargs["user_id"] == "mcp-user@example.com"
         assert update_kwargs["team_id"] == "team-123"
-        assert (
-            kwargs["litellm_params"]["metadata"]["user_api_key_user_id"]
-            == "mcp-user@example.com"
-        )
+        assert kwargs["litellm_params"]["metadata"]["user_api_key_user_id"] == "mcp-user@example.com"
 
 
 @pytest.mark.parametrize(
@@ -1828,9 +1954,7 @@ def test_should_track_cost_callback_pass_through_without_owner(call_type, expect
     ],
 )
 @pytest.mark.asyncio
-async def test_track_cost_callback_logs_unauthenticated_pass_through_request(
-    call_type, expect_spend_log
-):
+async def test_track_cost_callback_logs_unauthenticated_pass_through_request(call_type, expect_spend_log):
     """Regression for LIT-3782: a pass-through request with auth=false reaches the
     cost callback with no key/user/team/end-user. Before the fix the spend-log
     write was skipped and the request never appeared in request/usage logs. It
@@ -1876,9 +2000,7 @@ async def test_track_cost_callback_logs_unauthenticated_pass_through_request(
             end_time=datetime.now(),
         )
 
-        assert mock_proxy_logging.db_spend_update_writer.update_database.await_count == (
-            1 if expect_spend_log else 0
-        )
+        assert mock_proxy_logging.db_spend_update_writer.update_database.await_count == (1 if expect_spend_log else 0)
 
 
 class _FakeDeploymentLookup:
diff --git a/tests/test_litellm/proxy/utils/prisma_and_spend/test_spend_functions.py b/tests/test_litellm/proxy/utils/prisma_and_spend/test_spend_functions.py
index a1eb88a7834..fc97d760226 100644
--- a/tests/test_litellm/proxy/utils/prisma_and_spend/test_spend_functions.py
+++ b/tests/test_litellm/proxy/utils/prisma_and_spend/test_spend_functions.py
@@ -6,6 +6,7 @@ Symbols pinned here:
   - ``update_spend_logs_job``
   - ``_monitor_spend_logs_queue``
   - ``_raise_failed_update_spend_exception``
+  - ``spend_log_is_queued``
 """
 
 from __future__ import annotations
@@ -22,6 +23,7 @@ from litellm.proxy.utils import (
     _monitor_spend_logs_queue,
     _raise_failed_update_spend_exception,
     drain_spend_logs_queue,
+    spend_log_is_queued,
     update_daily_tag_spend,
     update_spend,
     update_spend_logs_job,
@@ -629,3 +631,16 @@ def test_raise_failed_update_spend_exception_raises_original_error() -> None:
 
     with pytest.raises(ValueError, match="specific"):
         asyncio.run(_runner())
+
+
+@pytest.mark.asyncio
+async def test_spend_log_is_queued_matches_only_rows_awaiting_flush(
+    mock_prisma_client: Any, make_spend_log_row: Any
+) -> None:
+    mock_prisma_client.spend_log_transactions = [make_spend_log_row(request_id="batch_abc_batch_cost")]
+
+    assert await spend_log_is_queued(mock_prisma_client, "batch_abc_batch_cost") is True
+    assert await spend_log_is_queued(mock_prisma_client, "batch_abc") is False
+
+    mock_prisma_client.spend_log_transactions = []
+    assert await spend_log_is_queued(mock_prisma_client, "batch_abc_batch_cost") is False