test(responses): cover the gate that decides a background row gets written

The storage branch's three conditions sat inline in `responses_api`, so nothing
proved a foreground create or an already-terminal one stays out of the managed
table. They move into `should_store_background_response`, which the endpoint
calls and the tests exercise across both arms.

Claude-Session: https://claude.ai/code/session_01RHAjRxNhXTpKHeGMZ1nDKi

litellm/proxy/response_api_endpoints/endpoints.py

@@ -62,6 +62,20 @@ class BackgroundResponseStore(Protocol):
     ) -> None: ...
 
 
+_STORABLE_BACKGROUND_STATUSES: Final[frozenset[str]] = frozenset({"queued", "in_progress"})
+
+
+def should_store_background_response(data: Mapping[str, object], response: object) -> bool:
+    """Whether a create just produced a generation the cost poller will have to bill later.
+
+    Only a background create leaves usage unreported, and only while the provider has not
+    finished it; anything already terminal reported its usage on this very call.
+    """
+    if not data.get("background") or not isinstance(response, ResponsesAPIResponse):
+        return False
+    return response.status in _STORABLE_BACKGROUND_STATUSES
+
+
 async def store_background_response_object(
     response: ResponsesAPIResponse,
     managed_files_obj: BackgroundResponseStore,
@@ -418,12 +432,7 @@ async def responses_api(
             version=version,
         )
 
-        # Store in managed objects table if background mode is enabled
-        if (
-            data.get("background")
-            and isinstance(response, ResponsesAPIResponse)
-            and response.status in ("queued", "in_progress")
-        ):
+        if should_store_background_response(data, response):
             managed_files_obj: Final = cast(
                 BackgroundResponseStore | None,
                 proxy_logging_obj.get_proxy_hook("managed_files"),

tests/test_litellm/proxy/response_api_endpoints/test_endpoints.py

@@ -2069,3 +2069,60 @@ class TestBackgroundResponseManagedObjectId:
         store = await self._stored_kwargs(self._encrypted_id("resp_no_deployment"), model_id=None)
 
         store.assert_not_awaited()
+
+
+class TestShouldStoreBackgroundResponse:
+    """The gate `responses_api` applies before it writes a managed row.
+
+    Storing a foreground create would bill a generation whose usage the create already
+    reported, and storing one the provider has already finished leaves a row no poll can
+    retire, so both arms have to stay closed.
+    """
+
+    @staticmethod
+    def _response(status: str):
+        from litellm.types.llms.openai import ResponsesAPIResponse
+
+        return ResponsesAPIResponse(
+            id="resp_abc",
+            created_at=0,
+            model="gpt-4o",
+            object="response",
+            output=[],
+            parallel_tool_calls=False,
+            tool_choice="auto",
+            tools=[],
+            status=status,
+        )
+
+    @pytest.mark.parametrize("status", ["queued", "in_progress"])
+    def test_a_background_create_the_provider_has_not_finished_is_stored(self, status):
+        from litellm.proxy.response_api_endpoints.endpoints import (
+            should_store_background_response,
+        )
+
+        assert should_store_background_response({"background": True}, self._response(status)) is True
+
+    @pytest.mark.parametrize("status", ["completed", "failed", "cancelled", "incomplete"])
+    def test_a_background_create_already_terminal_is_not_stored(self, status):
+        from litellm.proxy.response_api_endpoints.endpoints import (
+            should_store_background_response,
+        )
+
+        assert should_store_background_response({"background": True}, self._response(status)) is False
+
+    @pytest.mark.parametrize("data", [{}, {"background": False}, {"background": None}])
+    def test_a_foreground_create_is_never_stored(self, data):
+        from litellm.proxy.response_api_endpoints.endpoints import (
+            should_store_background_response,
+        )
+
+        assert should_store_background_response(data, self._response("queued")) is False
+
+    def test_a_streaming_or_error_result_is_not_mistaken_for_a_response(self):
+        """The create path can hand back a streaming iterator, which has no status to read."""
+        from litellm.proxy.response_api_endpoints.endpoints import (
+            should_store_background_response,
+        )
+
+        assert should_store_background_response({"background": True}, object()) is False