fix(proxy): stop billing every OpenAI pass-through /responses sub-route as a Responses call (#47171)

* test(proxy): cover OpenAI pass-through billing for each /responses sub-route

* fix(proxy): bill only the OpenAI pass-through Responses calls OpenAI bills

* fix(proxy): fail closed on unknown /responses sub-routes and label unbilled spend rows

Bill every OpenAI pass-through /responses call except input_tokens, input_items,
delete, and a non-background retrieve or cancel, so an encoded or new sub-route
is never free. Unbilled calls keep the response's model on their spend row, and
a resumed background stream takes its model from the stream's final event.

* fix(proxy): give billed background Responses reads their own spend log row

A billed retrieve or cancel of a finished background response reused the
create's resp_ id, so the spend log insert dropped the row that carried the
cost. It now takes the native path's _background_cost suffix. Adds Azure
path coverage and types the test recorder.

* fix(proxy): bill a background Responses read only once it carries usage

An in-progress poll took the _background_cost spend log id with $0, so the
finished read's row that carried the cost was dropped again. Polls with no
usage are now unbilled like any other read.

* fix(proxy): keep OpenAI's response id for callbacks on a billed background read

* fix(openai): give a streamed background create the settlement spend row id

* fix(proxy): keep a request body from giving its spend row the background settlement id

* test(proxy): give the forged settlement marker test a fixed time

* fix(passthrough): move the background settlement marker into the pass-through types module

A test that reloads spend_tracking_utils gave it a new enum class, so the
handler's marker no longer matched by identity and the settlement row lost
its _background_cost id. The types module is never reloaded.

litellm/proxy/pass_through_endpoints/llm_provider_handlers/openai_passthrough_logging_handler.py

@@ -6,7 +6,7 @@ Handles cost tracking and logging for OpenAI passthrough endpoints, specifically
 
 from collections.abc import Mapping, Sequence
 from datetime import datetime
-from typing import Final
+from typing import Final, Literal, TypeAlias
 from urllib.parse import urlparse
 
 import httpx
@@ -14,6 +14,7 @@ import httpx
 import litellm
 from litellm._logging import verbose_proxy_logger
 from litellm.litellm_core_utils.hidden_params import set_hidden_param
+from litellm.litellm_core_utils.internal_call_metadata import is_background_response
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
 from litellm.litellm_core_utils.litellm_logging import (
     get_standard_logging_object_payload,
@@ -31,6 +32,8 @@ from litellm.proxy.pass_through_endpoints.success_handler import (
 )
 from litellm.types.llms.openai import ResponsesAPIResponse
 from litellm.types.passthrough_endpoints.pass_through_endpoints import (
+    SETTLES_BACKGROUND_RESPONSE_KWARG,
+    BackgroundResponseSettlement,
     EndpointType,
     PassthroughStandardLoggingPayload,
 )
@@ -56,7 +59,7 @@ _AZURE_OPENAI_HOSTNAMES: Final = ("openai.azure.com", "cognitiveservices.azure.c
 _AZURE_OPENAI_PATH_MARKERS: Final = ("/openai/", "/v1/")
 
 
-def _hostname_matches(hostname: str, suffixes: tuple) -> bool:
+def _hostname_matches(hostname: str, suffixes: tuple[str, ...]) -> bool:
     """True if hostname equals one of `suffixes` or is a subdomain of it.
 
     Uses suffix matching (not a bare substring test) so look-alikes such as
@@ -102,6 +105,19 @@ def is_openai_compatible_url(url_route: str | None) -> bool:
 _is_openai_compatible_url: Final = is_openai_compatible_url
 
 
+ResponsesCallBilling: TypeAlias = Literal["unbilled", "billed", "background_settlement"]
+
+
+def _is_finished_background_response(response_body: Mapping[str, object]) -> bool:
+    return is_background_response(response_body) and response_body.get("usage") is not None
+
+
+def _responses_sub_route(path: str) -> tuple[str, ...] | None:
+    segments: Final = path.rstrip("/").split("/")
+    anchor: Final = max((index for index, segment in enumerate(segments) if segment == "responses"), default=None)
+    return None if anchor is None else tuple(segments[anchor + 1 :])
+
+
 def _is_remote_high_detail_image(part: object) -> bool:
     if not isinstance(part, Mapping) or part.get("type") != "image_url":
         return False
@@ -186,6 +202,30 @@ class OpenAIPassthroughLoggingHandler(BasePassthroughLoggingHandler):
             "/v1/responses" in parsed_url.path or "/responses" in parsed_url.path
         )
 
+    @staticmethod
+    def openai_responses_call_billing(
+        method: str, url_route: str, response_body: Mapping[str, object]
+    ) -> ResponsesCallBilling:
+        if not OpenAIPassthroughLoggingHandler.is_openai_responses_route(url_route):
+            return "unbilled"
+        match (method, _responses_sub_route(urlparse(url_route).path)):
+            case ("POST", ("input_tokens",)) | ("GET", (_, "input_items")) | ("DELETE", (_,)):
+                return "unbilled"
+            case ("GET", (_,)) | ("POST", (_, "cancel")):
+                return "background_settlement" if _is_finished_background_response(response_body) else "unbilled"
+            case _:
+                return "billed"
+
+    @staticmethod
+    def _label_unbilled_responses_call(
+        logging_obj: LiteLLMLoggingObj,
+        url_route: str,
+        response_body: Mapping[str, object],
+    ) -> None:
+        model: Final = response_body.get("model")
+        if OpenAIPassthroughLoggingHandler.is_openai_responses_route(url_route) and isinstance(model, str):
+            logging_obj.model_call_details["model"] = model  # rebind-ok: spend rows read the model from here
+
     @staticmethod
     def is_openai_embeddings_route(url_route: str) -> bool:
         """Check if the URL route is an OpenAI embeddings endpoint."""
@@ -348,9 +388,13 @@ class OpenAIPassthroughLoggingHandler(BasePassthroughLoggingHandler):
         is_embeddings: Final = OpenAIPassthroughLoggingHandler.is_openai_embeddings_route(url_route)
         is_image_generation: Final = OpenAIPassthroughLoggingHandler.is_openai_image_generation_route(url_route)
         is_image_editing: Final = OpenAIPassthroughLoggingHandler.is_openai_image_editing_route(url_route)
-        is_responses: Final = OpenAIPassthroughLoggingHandler.is_openai_responses_route(url_route)
+        responses_billing: Final = OpenAIPassthroughLoggingHandler.openai_responses_call_billing(
+            httpx_response.request.method, url_route, response_body
+        )
+        is_responses: Final = responses_billing != "unbilled"
 
         if not (is_chat_completions or is_embeddings or is_image_generation or is_image_editing or is_responses):
+            OpenAIPassthroughLoggingHandler._label_unbilled_responses_call(logging_obj, url_route, response_body)
             return {
                 "result": None,
                 "kwargs": kwargs,
@@ -505,6 +549,9 @@ class OpenAIPassthroughLoggingHandler(BasePassthroughLoggingHandler):
             logging_obj.model_call_details["model"] = model
             logging_obj.model_call_details["custom_llm_provider"] = custom_llm_provider
             logging_obj.model_call_details["response_cost"] = response_cost
+            logging_obj.model_call_details[SETTLES_BACKGROUND_RESPONSE_KWARG] = (  # rebind-ok: spend rows read it here
+                BackgroundResponseSettlement.SETTLES if responses_billing == "background_settlement" else None
+            )
 
             endpoint_type: Final = (
                 "chat_completions"
@@ -625,16 +672,24 @@ class OpenAIPassthroughLoggingHandler(BasePassthroughLoggingHandler):
         Handle logging for collected OpenAI streaming chunks with cost tracking.
         """
         try:
-            # Extract model from request body
-            model: Final = request_body.get("model", "gpt-4o")
-
             is_responses: Final = OpenAIPassthroughLoggingHandler.is_openai_responses_route(url_route)
+            terminal_response: Final = (
+                OpenAIResponsesAPIConfig.parse_terminal_response_from_stream_chunks(all_chunks=all_chunks)
+                if is_responses
+                else None
+            )
+            requested_model: Final[object] = request_body.get("model")
+            model: Final = (
+                requested_model
+                if isinstance(requested_model, str) and requested_model
+                else (terminal_response.model if terminal_response is not None else None) or "gpt-4o"
+            )
 
             # Build complete response from chunks using our streaming handler
             handler: Final = OpenAIPassthroughLoggingHandler()
             handler_instance: Final = handler
             complete_response: Final = (
-                OpenAIResponsesAPIConfig.parse_terminal_response_from_stream_chunks(all_chunks=all_chunks)
+                terminal_response
                 if is_responses
                 else handler._build_complete_streaming_response(
                     all_chunks=all_chunks,
@@ -667,17 +722,27 @@ class OpenAIPassthroughLoggingHandler(BasePassthroughLoggingHandler):
                 )
             )
 
+            settles_background_response: Final = (
+                BackgroundResponseSettlement.SETTLES
+                if terminal_response is not None
+                and is_background_response(terminal_response)
+                and terminal_response.usage is not None
+                else None
+            )
+
             # Preserve existing litellm_params to maintain metadata tags
             existing_litellm_params: Final = litellm_logging_obj.model_call_details.get("litellm_params", {}) or {}
+            litellm_params: Final = existing_litellm_params.copy()
 
             # Prepare kwargs for logging
             kwargs: Final = {
                 "response_cost": response_cost,
+                SETTLES_BACKGROUND_RESPONSE_KWARG: settles_background_response,
                 "model": model,
                 "custom_llm_provider": custom_llm_provider,
                 "call_type": litellm_logging_obj.call_type,
                 "messages": litellm_logging_obj.model_call_details.get("messages"),
-                "litellm_params": existing_litellm_params.copy(),
+                "litellm_params": litellm_params,
             }
 
             # Extract user information for tracking
@@ -689,9 +754,7 @@ class OpenAIPassthroughLoggingHandler(BasePassthroughLoggingHandler):
                     passthrough_logging_payload=passthrough_logging_payload,
                 )
                 if user:
-                    kwargs["litellm_params"].setdefault("proxy_server_request", {}).setdefault("body", {})["user"] = (
-                        user
-                    )
+                    litellm_params.setdefault("proxy_server_request", {}).setdefault("body", {})["user"] = user
 
             # Attach the payload to kwargs so the success handler adopts it;
             # its later rebuild runs on a copy whose Responses usage was
@@ -712,6 +775,9 @@ class OpenAIPassthroughLoggingHandler(BasePassthroughLoggingHandler):
             litellm_logging_obj.model_call_details["model"] = model
             litellm_logging_obj.model_call_details["custom_llm_provider"] = custom_llm_provider
             litellm_logging_obj.model_call_details["response_cost"] = response_cost
+            litellm_logging_obj.model_call_details[SETTLES_BACKGROUND_RESPONSE_KWARG] = (  # rebind-ok: spend row id
+                settles_background_response
+            )
 
             verbose_proxy_logger.debug(
                 f"OpenAI streaming passthrough cost tracking - Model: {model}, Cost: ${response_cost:.6f}"

litellm/proxy/spend_tracking/spend_tracking_utils.py

@@ -69,6 +69,10 @@ from litellm.proxy._types import (
 from litellm.proxy.route_llm_request import ProxyModelNotFoundError
 from litellm.proxy.spend_tracking.spend_log_error_logger import spend_log_error
 from litellm.proxy.utils import PrismaClient, hash_token
+from litellm.types.passthrough_endpoints.pass_through_endpoints import (
+    SETTLES_BACKGROUND_RESPONSE_KWARG,
+    BackgroundResponseSettlement,
+)
 from litellm.types.router import DeploymentTypedDict, LiteLLM_Params
 from litellm.types.utils import (
     BACKGROUND_RESPONSE_COST_POLL_CALL_ORIGIN,
@@ -315,10 +319,12 @@ BATCH_COST_REQUEST_ID_SUFFIX: Final = "_batch_cost"
 BACKGROUND_RESPONSE_COST_REQUEST_ID_SUFFIX: Final = "_background_cost"
 
 
-def _cost_row_request_id_suffix(call_type: str, metadata: Mapping[str, object]) -> str:
+def _cost_row_request_id_suffix(
+    call_type: str, metadata: Mapping[str, object], settles_background_response: bool
+) -> str:
     if call_type == CallTypes.aretrieve_batch.value:
         return BATCH_COST_REQUEST_ID_SUFFIX
-    if (
+    if settles_background_response or (
         call_type in NON_INFERENCE_CALL_TYPES
         and metadata.get(INTERNAL_CALL_ORIGIN_METADATA_KEY) == BACKGROUND_RESPONSE_COST_POLL_CALL_ORIGIN
     ):
@@ -327,7 +333,7 @@ def _cost_row_request_id_suffix(call_type: str, metadata: Mapping[str, object])
 
 
 def get_spend_logs_id(
-    call_type: str, response_obj: dict, kwargs: dict, metadata: Mapping[str, object] = EMPTY_MAPPING
+    call_type: str, response_obj: dict, kwargs: Mapping[str, object], metadata: Mapping[str, object] = EMPTY_MAPPING
 ) -> str | None:
     standard_logging_payload = kwargs.get("standard_logging_object")
     candidate_ids: Final = (
@@ -340,7 +346,10 @@ def get_spend_logs_id(
     )
     if resolved_id is None:
         return None
-    return f"{resolved_id}{_cost_row_request_id_suffix(call_type, metadata)}"
+    settles_background_response: Final = (
+        kwargs.get(SETTLES_BACKGROUND_RESPONSE_KWARG) is BackgroundResponseSettlement.SETTLES
+    )
+    return f"{resolved_id}{_cost_row_request_id_suffix(call_type, metadata, settles_background_response)}"
 
 
 _MISSING_ATTRIBUTE: Final = object()

litellm/types/passthrough_endpoints/pass_through_endpoints.py

@@ -25,6 +25,12 @@ LITELLM_PASS_THROUGH_DEPLOYMENT_MODEL_INFO_STATE_KEY: Final = "litellm_pass_thro
 # on a collision FastAPI dispatches the built-in handler, which does not carry this flag.
 LITELLM_PASS_THROUGH_ENDPOINT_MARKER: Final = "__litellm_pass_through_endpoint__"
 
+SETTLES_BACKGROUND_RESPONSE_KWARG: Final = "settles_background_response"
+
+
+class BackgroundResponseSettlement(str, Enum):
+    SETTLES = "settles"
+
 
 class EndpointType(str, Enum):
     VERTEX_AI = "vertex-ai"

tests/unit/proxy/pass_through_endpoints/llm_provider_handlers/test_openai_passthrough_logging_handler.py

@@ -1,19 +1,26 @@
+from dataclasses import dataclass
+import asyncio
 import json
+import logging
+from collections.abc import Mapping
 from datetime import datetime
-from typing import Any, Dict, List
+from typing import Any, Dict, Final, List
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import httpx
 import pytest
+from pydantic import BaseModel, ConfigDict
 
 
 import litellm
+from litellm.integrations.custom_logger import CustomLogger
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
 from litellm.litellm_core_utils.token_counter import high_detail_image_token_upper_bound
 from litellm.proxy.pass_through_endpoints.llm_provider_handlers.openai_passthrough_logging_handler import (
     OpenAIPassthroughLoggingHandler,
     count_relayed_prompt_tokens,
 )
+from litellm.proxy._types import SpendLogsPayload
 from litellm.proxy.pass_through_endpoints.success_handler import (
     PassThroughEndpointLogging,
 )
@@ -1837,10 +1844,7 @@ class TestOpenAIPassthroughResponsesStreamingSpendLog:
 
     def _expected_spend(self) -> float:
         rates = litellm.model_cost[self.MODEL_MAP_KEY]
-        return (
-            self.INPUT_TOKENS * rates["input_cost_per_token"]
-            + self.OUTPUT_TOKENS * rates["output_cost_per_token"]
-        )
+        return self.INPUT_TOKENS * rates["input_cost_per_token"] + self.OUTPUT_TOKENS * rates["output_cost_per_token"]
 
     def _responses_stream_chunks(self) -> List[str]:
         """Usage arrives only on the terminal `response.completed` event, nested under
@@ -1953,6 +1957,28 @@ class TestOpenAIPassthroughResponsesStreamingSpendLog:
         row_metadata = json.loads(spend_log_row["metadata"])
         assert row_metadata["model_map_information"]["model_map_key"] == self.MODEL_MAP_KEY
 
+    def test_resumed_responses_stream_spend_log_is_labeled_with_the_streamed_model(self, local_model_cost_map):
+        result: Final = OpenAIPassthroughLoggingHandler._handle_logging_openai_collected_chunks(
+            litellm_logging_obj=self._logging_obj(),
+            passthrough_success_handler_obj=None,
+            url_route=f"https://api.openai.com/v1/responses/{self.RESPONSE_ID}?stream=true",
+            request_body={},
+            endpoint_type=None,
+            start_time=self.start_time,
+            all_chunks=self._responses_stream_chunks(),
+            end_time=self.end_time,
+        )
+
+        spend_log_row: Final = get_logging_payload(
+            kwargs=result["kwargs"],
+            response_obj=result["result"],
+            start_time=self.start_time,
+            end_time=self.end_time,
+        )
+
+        assert spend_log_row["model"] == self.MODEL_MAP_KEY
+        assert spend_log_row["spend"] == pytest.approx(self._expected_spend())
+
 
 class TestOpenAIPassthroughEmbeddingsSpendLog:
     """An OpenAI-passthrough `/v1/embeddings` call must write a priced spend log row
@@ -2043,6 +2069,402 @@ class TestOpenAIPassthroughEmbeddingsSpendLog:
         assert spend_log_row["request_id"] == self.CALL_ID
 
 
+RESPONSES_URL: Final = "https://api.openai.com/v1/responses"
+STORED_RESPONSE_ID: Final = "resp_080b10ff5d740c69006ad3de732e5087"
+DISPATCH_TIME: Final = datetime(2026, 10, 17, 13, 47, 3)
+RESPONSES_USAGE: Final = {
+    "input_tokens": 11,
+    "input_tokens_details": {"cached_tokens": 0},
+    "output_tokens": 5,
+    "output_tokens_details": {"reasoning_tokens": 0},
+    "total_tokens": 16,
+}
+CREATE_REQUEST: Final = {"model": "gpt-5.4-mini", "input": "Say hi in one word"}
+COMPACT_REQUEST: Final = {"model": "gpt-5.4-mini", "input": "Remember the number 42."}
+COMPACTION: Final = {
+    "id": STORED_RESPONSE_ID,
+    "object": "response.compaction",
+    "created_at": 1792270025,
+    "output": [{"id": "cmp_082492f9", "type": "compaction", "encrypted_content": "gAAAA"}],
+    "usage": RESPONSES_USAGE,
+}
+ASSISTANT_MESSAGE: Final = {
+    "id": "msg_080b10ff5d740c69",
+    "type": "message",
+    "status": "completed",
+    "role": "assistant",
+    "content": [{"type": "output_text", "text": "Hi", "annotations": []}],
+}
+
+
+def _stored_response(status: str, background: bool, usage: Mapping[str, object] | None) -> dict[str, object]:
+    return {
+        "id": STORED_RESPONSE_ID,
+        "object": "response",
+        "created_at": 1792269939,
+        "status": status,
+        "background": background,
+        "error": None,
+        "incomplete_details": None,
+        "instructions": None,
+        "metadata": {},
+        "model": "gpt-5.4-mini-2026-03-17",
+        "output": [ASSISTANT_MESSAGE] if usage else [],
+        "parallel_tool_calls": True,
+        "temperature": 1.0,
+        "tool_choice": "auto",
+        "tools": [],
+        "top_p": 0.98,
+        "usage": usage,
+    }
+
+
+@dataclass(frozen=True, slots=True)
+class _LoggedCall:
+    spend_row: SpendLogsPayload
+    callback_response_id: str
+
+
+class _CallbackPayloadId(BaseModel):
+    model_config = ConfigDict(frozen=True)
+
+    id: str
+
+
+class _SpendRowRecorder(CustomLogger):
+    def __init__(self) -> None:
+        super().__init__()
+        self.call: Final[asyncio.Future[_LoggedCall]] = asyncio.get_running_loop().create_future()
+
+    async def async_log_success_event(
+        self, kwargs: dict[str, object], response_obj: object, start_time: datetime, end_time: datetime
+    ) -> None:
+        self.call.set_result(
+            _LoggedCall(
+                spend_row=get_logging_payload(
+                    kwargs=kwargs, response_obj=response_obj, start_time=start_time, end_time=end_time
+                ),
+                callback_response_id=_CallbackPayloadId.model_validate(kwargs.get("standard_logging_object")).id,
+            )
+        )
+
+
+class TestOpenAIPassthroughResponsesSubRouteBilling:
+    def _dispatch(
+        self,
+        method: str,
+        sub_route: str,
+        request_body: dict[str, object],
+        response_body: dict[str, object],
+        base_url: str = RESPONSES_URL,
+    ):
+        url_route: Final = f"{base_url}{sub_route}"
+        logging_obj: Final = LiteLLMLoggingObj(
+            model="gpt-5.4-mini",
+            messages=[],
+            stream=False,
+            call_type="pass_through_endpoint",
+            start_time=DISPATCH_TIME,
+            litellm_call_id="6c1f3a52-93d4-4b1e-9f0e-6f2f4f6b9a10",
+            function_id="1234",
+        )
+        return PassThroughEndpointLogging().normalize_llm_passthrough_logging_payload(
+            httpx_response=httpx.Response(200, json=response_body, request=httpx.Request(method, url_route)),
+            response_body=response_body,
+            request_body=request_body,
+            logging_obj=logging_obj,
+            url_route=url_route,
+            result="",
+            start_time=DISPATCH_TIME,
+            end_time=DISPATCH_TIME,
+            cache_hit=False,
+            custom_llm_provider="openai",
+            litellm_params={},
+            call_type="pass_through_endpoint",
+        )
+
+    @pytest.mark.parametrize(
+        ("method", "sub_route", "request_body", "response_body"),
+        [
+            pytest.param(
+                "POST", "", CREATE_REQUEST, _stored_response("completed", False, RESPONSES_USAGE), id="create"
+            ),
+            pytest.param(
+                "POST",
+                "/",
+                CREATE_REQUEST,
+                _stored_response("completed", False, RESPONSES_USAGE),
+                id="create_trailing_slash",
+            ),
+            pytest.param("POST", "/compact", COMPACT_REQUEST, COMPACTION, id="compact"),
+            pytest.param("POST", "/%2563ompact", COMPACT_REQUEST, COMPACTION, id="percent_encoded_compact"),
+            pytest.param(
+                "GET",
+                f"/{STORED_RESPONSE_ID}",
+                {},
+                _stored_response("completed", True, RESPONSES_USAGE),
+                id="background_retrieve_completed",
+            ),
+            pytest.param(
+                "POST",
+                f"/{STORED_RESPONSE_ID}/cancel",
+                {},
+                _stored_response("cancelled", True, RESPONSES_USAGE),
+                id="background_cancel",
+            ),
+        ],
+    )
+    def test_billed_responses_call_records_cost(
+        self,
+        local_model_cost_map,
+        method: str,
+        sub_route: str,
+        request_body: dict[str, object],
+        response_body: dict[str, object],
+    ):
+        dispatched: Final = self._dispatch(method, sub_route, request_body, response_body)
+
+        assert dispatched["standard_logging_response_object"] is not None
+        assert dispatched["kwargs"]["response_cost"] > 0
+
+    @pytest.mark.parametrize(
+        ("method", "sub_route", "request_body", "response_body"),
+        [
+            pytest.param(
+                "POST",
+                "/input_tokens",
+                {"model": "gpt-5.4-mini", "input": "hi"},
+                {"object": "response.input_tokens", "input_tokens": 7},
+                id="input_tokens",
+            ),
+            pytest.param(
+                "GET",
+                f"/{STORED_RESPONSE_ID}",
+                {},
+                _stored_response("completed", False, RESPONSES_USAGE),
+                id="retrieve",
+            ),
+            pytest.param(
+                "GET",
+                f"/{STORED_RESPONSE_ID}/input_items",
+                {},
+                {
+                    "object": "list",
+                    "data": [
+                        {
+                            "id": "msg_074176e3",
+                            "type": "message",
+                            "status": "completed",
+                            "role": "user",
+                            "content": [{"type": "input_text", "text": "Say hi in one word"}],
+                        }
+                    ],
+                    "first_id": "msg_074176e3",
+                    "last_id": "msg_074176e3",
+                    "has_more": False,
+                },
+                id="input_items",
+            ),
+            pytest.param(
+                "DELETE",
+                f"/{STORED_RESPONSE_ID}",
+                {},
+                {"id": STORED_RESPONSE_ID, "object": "response.deleted", "deleted": True},
+                id="delete",
+            ),
+            pytest.param(
+                "POST",
+                f"/{STORED_RESPONSE_ID}/cancel",
+                {},
+                _stored_response("cancelled", True, None),
+                id="background_cancel_before_any_output",
+            ),
+        ],
+    )
+    def test_unbilled_responses_sub_route_records_no_cost_and_no_error(
+        self,
+        local_model_cost_map,
+        caplog: pytest.LogCaptureFixture,
+        method: str,
+        sub_route: str,
+        request_body: dict[str, object],
+        response_body: dict[str, object],
+    ):
+        with caplog.at_level(logging.WARNING):
+            dispatched: Final = self._dispatch(method, sub_route, request_body, response_body)
+
+        assert not dispatched["kwargs"].get("response_cost")
+        assert [record.getMessage() for record in caplog.records if record.levelno >= logging.ERROR] == []
+
+    async def _logged_call(
+        self,
+        method: str,
+        sub_route: str,
+        request_body: dict[str, object],
+        response_body: dict[str, object],
+        litellm_call_id: str,
+    ) -> _LoggedCall:
+        url_route: Final = f"{RESPONSES_URL}{sub_route}"
+        recorder: Final = _SpendRowRecorder()
+        logging_obj: Final = LiteLLMLoggingObj(
+            model="unknown",
+            messages=[],
+            stream=False,
+            call_type="pass_through_endpoint",
+            start_time=DISPATCH_TIME,
+            litellm_call_id=litellm_call_id,
+            function_id="1234",
+            dynamic_async_success_callbacks=[recorder],
+        )
+        logging_obj.update_environment_variables(
+            model="unknown", user="unknown", optional_params={}, litellm_params={}, call_type="pass_through_endpoint"
+        )
+
+        await PassThroughEndpointLogging().pass_through_async_success_handler(
+            httpx_response=httpx.Response(200, json=response_body, request=httpx.Request(method, url_route)),
+            response_body=response_body,
+            logging_obj=logging_obj,
+            url_route=url_route,
+            result="",
+            start_time=DISPATCH_TIME,
+            end_time=DISPATCH_TIME,
+            cache_hit=False,
+            request_body=request_body,
+            passthrough_logging_payload=PassthroughStandardLoggingPayload(
+                url=url_route, request_body=request_body, request_method=method
+            ),
+            custom_llm_provider="openai",
+        )
+        return await asyncio.wait_for(recorder.call, timeout=5)
+
+    @pytest.mark.asyncio
+    async def test_unbilled_retrieve_spend_log_keeps_the_response_model(self, local_model_cost_map):
+        response_body: Final = _stored_response("completed", False, RESPONSES_USAGE)
+
+        spend_log_row: Final = (
+            await self._logged_call(
+                "GET", f"/{STORED_RESPONSE_ID}", {}, response_body, "0f6f3d2e-6a51-4c4e-9a43-6f1b0d1e9c21"
+            )
+        ).spend_row
+
+        assert spend_log_row["model"] == response_body["model"]
+        assert not spend_log_row["spend"]
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        ("method", "sub_route", "response_body"),
+        [
+            pytest.param(
+                "GET",
+                f"/{STORED_RESPONSE_ID}",
+                _stored_response("completed", True, RESPONSES_USAGE),
+                id="background_retrieve_completed",
+            ),
+            pytest.param(
+                "POST",
+                f"/{STORED_RESPONSE_ID}/cancel",
+                _stored_response("cancelled", True, RESPONSES_USAGE),
+                id="background_cancel",
+            ),
+        ],
+    )
+    async def test_billed_background_read_spend_log_survives_next_to_its_create_and_poll_rows(
+        self, local_model_cost_map, method: str, sub_route: str, response_body: dict[str, object]
+    ):
+        create: Final = await self._logged_call(
+            "POST",
+            "",
+            {**CREATE_REQUEST, "background": True},
+            _stored_response("queued", True, None),
+            "3a4f0c1e-2b7d-4c55-8e0e-0d7a51d9f6a1",
+        )
+        poll: Final = await self._logged_call(
+            "GET",
+            f"/{STORED_RESPONSE_ID}",
+            {},
+            _stored_response("in_progress", True, None),
+            "5d8c2f71-0e4b-4a9d-b3c6-7f1e2a9d0c44",
+        )
+        read: Final = await self._logged_call(
+            method, sub_route, {}, response_body, "9b2e6d40-71c3-4f8a-a5d2-4e3c8f0b7d12"
+        )
+
+        assert read.spend_row["request_id"] not in {create.spend_row["request_id"], poll.spend_row["request_id"]}
+        assert read.spend_row["request_id"].startswith(STORED_RESPONSE_ID)
+        assert read.spend_row["spend"] > 0
+        assert not poll.spend_row["spend"]
+        assert read.callback_response_id == STORED_RESPONSE_ID
+
+    @pytest.mark.asyncio
+    async def test_streamed_background_create_and_its_finished_read_share_one_spend_row(self, local_model_cost_map):
+        finished: Final = _stored_response("completed", True, RESPONSES_USAGE)
+        stream_logging_obj: Final = LiteLLMLoggingObj(
+            model="gpt-5.4-mini",
+            messages=[],
+            stream=True,
+            call_type="pass_through_endpoint",
+            start_time=DISPATCH_TIME,
+            litellm_call_id="7e3b9a14-5c2d-4f60-8d1a-2b6c0e9f4a37",
+            function_id="1234",
+        )
+        stream_logging_obj.model_call_details["custom_llm_provider"] = "openai"
+        streamed_create: Final = OpenAIPassthroughLoggingHandler._handle_logging_openai_collected_chunks(
+            litellm_logging_obj=stream_logging_obj,
+            passthrough_success_handler_obj=None,
+            url_route=RESPONSES_URL,
+            request_body={**CREATE_REQUEST, "background": True, "stream": True},
+            endpoint_type=None,
+            start_time=DISPATCH_TIME,
+            all_chunks=[
+                f"data: {json.dumps({'type': 'response.completed', 'sequence_number': 1, 'response': finished})}",
+                "data: [DONE]",
+            ],
+            end_time=DISPATCH_TIME,
+        )
+        create_row: Final = get_logging_payload(
+            kwargs=streamed_create["kwargs"],
+            response_obj=streamed_create["result"],
+            start_time=DISPATCH_TIME,
+            end_time=DISPATCH_TIME,
+        )
+
+        read: Final = await self._logged_call(
+            "GET", f"/{STORED_RESPONSE_ID}", {}, finished, "1c8d5e2a-9f47-4b3e-a6d0-5e7f2c1b8a93"
+        )
+
+        assert create_row["spend"] > 0
+        assert read.spend_row["request_id"] == create_row["request_id"]
+        assert read.callback_response_id == STORED_RESPONSE_ID
+
+    @pytest.mark.parametrize(
+        "base_url",
+        [
+            pytest.param("https://my-resource.openai.azure.com/openai/v1/responses", id="azure_openai_v1"),
+            pytest.param("https://my-resource.cognitiveservices.azure.com/openai/responses", id="azure_openai_native"),
+        ],
+    )
+    def test_azure_responses_sub_routes_bill_like_openai(self, local_model_cost_map, base_url: str):
+        retrieve: Final = self._dispatch(
+            "GET", f"/{STORED_RESPONSE_ID}", {}, _stored_response("completed", False, RESPONSES_USAGE), base_url
+        )
+        input_tokens: Final = self._dispatch(
+            "POST",
+            "/input_tokens",
+            {"model": "gpt-5.4-mini", "input": "hi"},
+            {"object": "response.input_tokens", "input_tokens": 7},
+            base_url,
+        )
+        background_retrieve: Final = self._dispatch(
+            "GET", f"/{STORED_RESPONSE_ID}", {}, _stored_response("completed", True, RESPONSES_USAGE), base_url
+        )
+        compact: Final = self._dispatch("POST", "/compact", COMPACT_REQUEST, COMPACTION, base_url)
+
+        assert not retrieve["kwargs"].get("response_cost")
+        assert not input_tokens["kwargs"].get("response_cost")
+        assert background_retrieve["kwargs"]["response_cost"] > 0
+        assert compact["kwargs"]["response_cost"] > 0
+
+
 if __name__ == "__main__":
     pytest.main([__file__])
 

tests/unit/proxy/spend_tracking/test_spend_tracking_utils.py

@@ -47,6 +47,10 @@ from litellm.proxy.spend_tracking.spend_tracking_utils import (
     spend_log_row_with_retained_metadata,
 )
 from litellm.proxy.utils import hash_token
+from litellm.types.passthrough_endpoints.pass_through_endpoints import (
+    SETTLES_BACKGROUND_RESPONSE_KWARG,
+    BackgroundResponseSettlement,
+)
 from litellm.types.router import GenericLiteLLMParams
 from litellm.types.utils import (
     BACKGROUND_RESPONSE_COST_POLL_CALL_ORIGIN,
@@ -4810,6 +4814,29 @@ def test_a_background_response_billing_event_writes_a_row_apart_from_the_read_th
     assert request_ids == ("resp_shared", "resp_shared_background_cost")
 
 
+@pytest.mark.parametrize("forged_marker", [True, BackgroundResponseSettlement.SETTLES.value])
+def test_a_request_body_cannot_move_its_spend_row_onto_a_background_settlement_id(forged_marker):
+    """A failed call logs its raw request body as kwargs, so a body field named like the settlement
+    marker must not give the failure row the id a background response's billing row owns."""
+    request_body: Final = {
+        "model": "gpt-5.4-mini",
+        "input": "hi",
+        SETTLES_BACKGROUND_RESPONSE_KWARG: forged_marker,
+        "litellm_call_id": "call-failed",
+        "litellm_params": {"metadata": {"user_api_key": "test-key"}},
+    }
+    failed_at: Final = datetime.datetime(2026, 1, 1, tzinfo=timezone.utc)
+
+    row: Final = get_logging_payload(
+        kwargs=request_body,
+        response_obj=Exception("upstream rejected the request"),
+        start_time=failed_at,
+        end_time=failed_at,
+    )
+
+    assert row["request_id"] == "call-failed"
+
+
 def test_get_spend_logs_id_prefers_the_response_id_over_the_standard_logging_id():
     """An unredacted response keeps deciding its own row key, so cache-hit ids and every
     other call type behave exactly as they did before."""