fix(proxy): flush the spend-log queue only for Responses turns the bridge reads back (#47614)

Every /v1/responses spend log asked its worker to flush the queue right away so a
chained previous_response_id could find the row. Only the chat-completions bridge
reads that row. Native Responses providers hold the conversation upstream, so their
rows now wait for the batch write like chat rows do

Co-authored-by: devin-ai-integration[bot] <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Kerry Lu 2026-10-20 12:21:42 -07:00 • committed by GitHub
parent e276bf34f7
commit 423362bab2
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 91 additions and 20 deletions

View file

@ -95,6 +95,7 @@ if TYPE_CHECKING:
from litellm.proxy.db.autorouter_session_rollup import AutoRouterTurnTransaction
from litellm.proxy.db.baseline_accounting import DailyBaselineAttribution
from litellm.proxy.utils import PrismaClient, ProxyLogging
from litellm.router import Router
else:
PrismaClient = Any
ProxyLogging = Any
@ -122,6 +123,34 @@ def _parsed_request_tags(request_tags: object) -> Iterable[object] | TypeError |
return TypeError(f"'{type(parsed).__name__}' object is not iterable")
def _turn_chains_through_the_db(payload: Mapping[str, object], llm_router: "Router | None") -> bool:
"""A Responses turn the chat-completions bridge replays from its spend log on the next `previous_response_id`.
Native Responses providers hold the conversation themselves, so their rows can wait for the batch write.
A row with no router deployment behind it is treated as bridged, which keeps today's immediate flush.
"""
if payload.get("call_type") not in RESPONSES_SESSION_CALL_TYPES:
return False
model_id: Final = payload.get("model_id")
deployment: Final = (
llm_router.get_deployment(model_id=model_id) if llm_router is not None and isinstance(model_id, str) else None
)
if deployment is None:
return True
from litellm.responses.main import provider_stores_the_conversation
litellm_params: Final = deployment.litellm_params
return not provider_stores_the_conversation(
litellm_params.model,
litellm_params.custom_llm_provider,
{
"use_chat_completions_api": litellm_params.use_chat_completions_api,
"model_info": deployment.model_info.model_dump(),
"api_base": litellm_params.api_base,
},
)
def _org_member_transaction_key(org_id: str, user_id: str) -> str:
return f"organization_id::{quote(org_id, safe='')}::user_id::{quote(user_id, safe='')}"
@ -565,12 +594,13 @@ class DBSpendUpdateWriter:
get_request_model_access_groups,
)
llm_router: Final = get_llm_router()
payload: Final = get_logging_payload(
kwargs=kwargs,
response_obj=completion_response,
start_time=start_time,
end_time=end_time,
llm_router=get_llm_router(),
llm_router=llm_router,
)
payload["spend"] = response_cost or 0.0
if isinstance(payload["startTime"], datetime):
@ -585,7 +615,10 @@ class DBSpendUpdateWriter:
payload["team_id"] = team_id
if not await self._record_spend_log(
payload=payload, prisma_client=prisma_client, disable_spend_logs=disable_spend_logs
payload=payload,
prisma_client=prisma_client,
disable_spend_logs=disable_spend_logs,
llm_router=llm_router,
):
return False
@ -649,14 +682,18 @@ class DBSpendUpdateWriter:
return True
async def _record_spend_log(
self, payload: SpendLogsPayload, prisma_client: "PrismaClient | None", disable_spend_logs: bool
self,
payload: SpendLogsPayload,
prisma_client: "PrismaClient | None",
disable_spend_logs: bool,
llm_router: "Router | None" = None,
) -> bool:
if prisma_client is not None and _is_batch_cost_row(payload):
return await self._claim_batch_cost_spend_log(
payload=payload, prisma_client=prisma_client, disable_spend_logs=disable_spend_logs
)
if disable_spend_logs is False:
await self._insert_spend_log_to_db(payload=payload, prisma_client=prisma_client)
await self._insert_spend_log_to_db(payload=payload, prisma_client=prisma_client, llm_router=llm_router)
return True
async def _claim_batch_cost_spend_log(
@ -1534,6 +1571,7 @@ class DBSpendUpdateWriter:
payload: dict | SpendLogsPayload,
prisma_client: PrismaClient | None = None,
spend_logs_url: str | None = os.getenv("SPEND_LOGS_URL"),
llm_router: "Router | None" = None,
) -> PrismaClient | None:
verbose_proxy_logger.debug(
"Writing spend log to db - request_id: {}, spend: {}".format(
@ -1544,7 +1582,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 _turn_chains_through_the_db(_SPEND_METADATA_ADAPTER.validate_python(payload), llm_router):
request_spend_log_flush(prisma_client)
else:
verbose_proxy_logger.debug("prisma_client is None. Skipping writing spend logs to db.")

View file

@ -306,7 +306,7 @@ async def aresponses_api_with_mcp(
joins_request_trace
and not stream
and previous_response_id is None
and not _provider_stores_the_first_round(model, custom_llm_provider, kwargs)
and not provider_stores_the_conversation(model, custom_llm_provider, kwargs)
)
listing_trace_id: Final = (
@ -463,7 +463,7 @@ async def aresponses_api_with_mcp(
if tool_results:
chains_on_first_round: Final = not LiteLLM_Proxy_MCP_Handler.is_persistence_disabled(
call_params
) and _provider_stores_the_first_round(model, custom_llm_provider, kwargs)
) and provider_stores_the_conversation(model, custom_llm_provider, kwargs)
follow_up_input: Final = LiteLLM_Proxy_MCP_Handler.create_follow_up_input(
response=response,
@ -687,8 +687,12 @@ def _will_bridge_to_chat_completions(
)
def _provider_stores_the_first_round(model: str, custom_llm_provider: str | None, kwargs: Mapping[str, object]) -> bool:
"""A bridged first MCP round runs as an internal call, so no spend log holds it for the follow-up to replay."""
def provider_stores_the_conversation(model: str, custom_llm_provider: str | None, kwargs: Mapping[str, object]) -> bool:
"""Whether the provider, not a LiteLLM spend log, holds this turn for a `previous_response_id` follow-up.
A bridged turn is replayed from its spend log, so the bridge needs that row in the DB, and a bridged
first MCP round runs as an internal call that writes no spend log at all.
"""
api_base: Final = _api_base_kwarg(kwargs)
return not _will_bridge_to_chat_completions(
model,

View file

@ -4130,28 +4130,57 @@ async def test_commit_spend_updates_raises_after_exhausting_lock_timeout_retries
assert mock_prisma_client.db.tx.call_count == 3
def _responses_router() -> litellm.Router:
return litellm.Router(
model_list=[
{
"model_name": "native",
"litellm_params": {"model": "openai/gpt-5.1", "api_key": "sk-test"},
"model_info": {"id": "dep-native"},
},
{
"model_name": "bridged",
"litellm_params": {"model": "anthropic/claude-sonnet-4-5", "api_key": "sk-test"},
"model_info": {"id": "dep-bridged"},
},
{
"model_name": "forced-bridge",
"litellm_params": {"model": "openai/gpt-5.1", "api_key": "sk-test", "use_chat_completions_api": True},
"model_info": {"id": "dep-forced-bridge"},
},
]
)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"call_type, expects_flush",
[("aresponses", True), ("responses", True), ("acompletion", False)],
"call_type, model_id, expects_flush",
[
("aresponses", "dep-bridged", True),
("responses", "dep-bridged", True),
("aresponses", "dep-forced-bridge", True),
("aresponses", None, True),
("aresponses", "dep-native", False),
("responses", "dep-native", False),
("acompletion", "dep-bridged", False),
],
)
async def test_insert_spend_log_asks_for_an_immediate_flush_on_rows_other_workers_read_back(
call_type: str, expects_flush: bool
async def test_insert_spend_log_asks_for_an_immediate_flush_only_on_rows_the_bridge_reads_back(
call_type: str, model_id: str | None, 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.
A `previous_response_id` chained straight off a bridged turn reads that turn's row from the
DB, so it cannot sit in this worker's queue until the monitor's next poll. A native Responses
provider holds the conversation itself, so its row waits for the batch write like a chat row.
"""
db_writer = DBSpendUpdateWriter()
prisma = _tool_usage_prisma()
prisma.spend_log_flush_requested = asyncio.Event()
payload = {"request_id": "req-1", "call_type": call_type, "model_id": model_id}
await db_writer._insert_spend_log_to_db(
payload={"request_id": "req-1", "call_type": call_type},
prisma_client=prisma,
)
await db_writer._insert_spend_log_to_db(payload=payload, prisma_client=prisma, llm_router=_responses_router())
assert prisma.spend_log_transactions == [{"request_id": "req-1", "call_type": call_type}]
assert prisma.spend_log_transactions == [payload]
assert prisma.spend_log_flush_requested.is_set() is expects_flush