diff --git a/litellm-proxy-extras/litellm_proxy_extras/migrations/20260915000000_add_background_interaction_settlement/migration.sql b/litellm-proxy-extras/litellm_proxy_extras/migrations/20260915000000_add_background_interaction_settlement/migration.sql new file mode 100644 index 00000000000..94d5e98f2a7 --- /dev/null +++ b/litellm-proxy-extras/litellm_proxy_extras/migrations/20260915000000_add_background_interaction_settlement/migration.sql @@ -0,0 +1,16 @@ +-- CreateTable +CREATE TABLE IF NOT EXISTS "LiteLLM_BackgroundInteractionSettlement" ( + "interaction_id" TEXT NOT NULL, + "custom_llm_provider" TEXT NOT NULL, + "create_context" JSONB NOT NULL, + "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "claimed_at" TIMESTAMP(3), + "claimed_by" TEXT, + "settled_at" TIMESTAMP(3), + "outcome" TEXT, + + CONSTRAINT "LiteLLM_BackgroundInteractionSettlement_pkey" PRIMARY KEY ("interaction_id") +); + +-- CreateIndex +CREATE INDEX IF NOT EXISTS "idx_background_interaction_settlement_claimed_at" ON "LiteLLM_BackgroundInteractionSettlement"("claimed_at"); diff --git a/litellm-proxy-extras/litellm_proxy_extras/schema.prisma b/litellm-proxy-extras/litellm_proxy_extras/schema.prisma index aba89526cf6..e75a9b71c7c 100644 --- a/litellm-proxy-extras/litellm_proxy_extras/schema.prisma +++ b/litellm-proxy-extras/litellm_proxy_extras/schema.prisma @@ -1916,6 +1916,22 @@ model LiteLLM_WorkflowMessage { @@index([run_id]) } +// Pending billing settlements for background interactions, keyed by the +// interaction id so any replica can settle one that another replica created. +// `claimed_at` is the exactly-once gate: the first conditional update wins. +model LiteLLM_BackgroundInteractionSettlement { + interaction_id String @id + custom_llm_provider String + create_context Json + created_at DateTime @default(now()) + claimed_at DateTime? + claimed_by String? + settled_at DateTime? + outcome String? + + @@index([claimed_at], map: "idx_background_interaction_settlement_claimed_at") +} + model LiteLLM_Lens { id String @id version Int @default(0) diff --git a/litellm/interactions/background_cost_polling.py b/litellm/interactions/background_cost_polling.py index b48c7c03573..ca93436490d 100644 --- a/litellm/interactions/background_cost_polling.py +++ b/litellm/interactions/background_cost_polling.py @@ -23,16 +23,30 @@ caller retrieve the completed output themselves and then delete it before the poll task settles, leaving the work unbilled and the budget reservation refunded at the poll timeout. ``adelete`` therefore settles any pending poll for the interaction before dispatching the delete: it fetches the current -state with the create's credentials, bills it if it is terminal with usage, -and releases the reservation otherwise. A settlement gate on the create's -logging object makes the poll task and the delete path mutually exclusive, so -the interaction is billed exactly once no matter who settles first. +state, bills it if it is terminal with usage, and releases the reservation +otherwise. + +The poll task lives in the process that served the create, so a delete +served by another replica, or by the same replica after a restart, finds no +task to settle. A ``BackgroundSettlementStore`` makes the pending settlement +durable across processes: the create registers the request context that +billing needs (never provider credentials), the settlement is claimed +exactly once through the store, and a delete on any replica rebuilds the +billing context from the store when the poll task is not local. Rows left +unclaimed by a process that died are resumed at startup. The default store is +in-memory, which keeps the SDK and single-process behavior unchanged; the +proxy installs a database-backed one. """ import asyncio -from collections.abc import Awaitable, Callable, Iterator, Mapping -from dataclasses import dataclass -from typing import TYPE_CHECKING, Final, TypeAlias +from collections.abc import Awaitable, Callable, Iterable, Iterator, Mapping, Sequence +from dataclasses import dataclass, field +from datetime import datetime, timezone +from types import MappingProxyType +from typing import TYPE_CHECKING, Final, Literal, Protocol, TypeAlias + +from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter, ValidationError +from pydantic_core import PydanticSerializationError, to_jsonable_python from litellm._logging import verbose_logger from litellm.constants import ( @@ -43,6 +57,7 @@ from litellm.constants import ( ) from litellm.litellm_core_utils.core_helpers import get_litellm_metadata_from_kwargs from litellm.types.interactions import InteractionsAPIResponse +from litellm.types.utils import CustomPricingLiteLLMParams if TYPE_CHECKING: from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj @@ -55,6 +70,98 @@ _POLLABLE_STATUSES: Final = frozenset({"in_progress", "queued"}) _STATUSES_THAT_PRODUCED_OUTPUT: Final = frozenset({"completed", "requires_action"}) +SettlementOutcome: TypeAlias = Literal["billed", "released", "unsettled"] + + +class BackgroundInteractionCreateContext(BaseModel): + """ + The part of a create's logging state that billing its settled result needs, + in a shape any replica can store and rebuild a logging object from. Provider + credentials are deliberately absent: the replica that settles fetches the + interaction with its own, exactly as it would serve the delete itself. + """ + + model_config = ConfigDict(frozen=True) + + model: str | None + call_type: str + litellm_call_id: str + function_id: str + litellm_trace_id: str + start_time: datetime + custom_llm_provider: str + metadata: Mapping[str, JsonValue] + custom_pricing: Mapping[str, JsonValue] + + +@dataclass(frozen=True, slots=True) +class PendingBackgroundInteraction: + interaction_id: str + custom_llm_provider: str + create_context: BackgroundInteractionCreateContext + created_at: datetime + + +class BackgroundSettlementStore(Protocol): + async def register(self, pending: PendingBackgroundInteraction) -> None: ... + + async def pending(self, interaction_id: str) -> PendingBackgroundInteraction | None: ... + + async def is_claimed(self, interaction_id: str) -> bool: ... + + async def claim(self, interaction_id: str) -> bool: ... + + async def record_outcome(self, interaction_id: str, outcome: SettlementOutcome) -> None: ... + + async def unclaimed(self) -> Sequence[PendingBackgroundInteraction]: ... + + +@dataclass(frozen=True, slots=True) +class InMemoryBackgroundSettlementStore: + """ + Per-process store: a registered interaction maps to its pending row until + it is claimed, after which it maps to ``None``. Claiming an interaction the + store never saw succeeds once, which is what a poll built without a + registration relies on. + """ + + _rows: dict[str, PendingBackgroundInteraction | None] = field( # mutable-ok: the registry every settler shares + default_factory=dict + ) + + async def register(self, pending: PendingBackgroundInteraction) -> None: + self._rows[pending.interaction_id] = pending + + async def pending(self, interaction_id: str) -> PendingBackgroundInteraction | None: + return self._rows.get(interaction_id) + + async def is_claimed(self, interaction_id: str) -> bool: + return interaction_id in self._rows and self._rows[interaction_id] is None + + async def claim(self, interaction_id: str) -> bool: + if await self.is_claimed(interaction_id): + return False + self._rows[interaction_id] = None + return True + + async def record_outcome(self, interaction_id: str, outcome: SettlementOutcome) -> None: + return None + + async def unclaimed(self) -> Sequence[PendingBackgroundInteraction]: + return tuple(row for row in self._rows.values() if row is not None) + + +@dataclass(slots=True) +class _StoreSlot: + store: BackgroundSettlementStore + + +_STORE: Final = _StoreSlot(store=InMemoryBackgroundSettlementStore()) + + +def configure_background_settlement_store(store: BackgroundSettlementStore) -> None: + _STORE.store = store + @dataclass(frozen=True, slots=True) class BackgroundInteractionPollContext: @@ -66,12 +173,14 @@ class BackgroundInteractionPollContext: initial_interval_seconds: float = BACKGROUND_INTERACTION_COST_POLL_INITIAL_INTERVAL_SECONDS max_interval_seconds: float = BACKGROUND_INTERACTION_COST_POLL_MAX_INTERVAL_SECONDS timeout_seconds: float = BACKGROUND_INTERACTION_COST_POLL_TIMEOUT_SECONDS + store: BackgroundSettlementStore = field(default_factory=InMemoryBackgroundSettlementStore) + resumed: bool = False FetchInteraction: TypeAlias = Callable[[BackgroundInteractionPollContext], Awaitable[InteractionsAPIResponse]] -async def _fetch_interaction(context: BackgroundInteractionPollContext) -> InteractionsAPIResponse: +async def fetch_background_interaction(context: BackgroundInteractionPollContext) -> InteractionsAPIResponse: from litellm.interactions import aget return await aget( @@ -84,46 +193,159 @@ async def _fetch_interaction(context: BackgroundInteractionPollContext) -> Inter def _poll_intervals(initial: float, maximum: float, timeout: float) -> Iterator[float]: - elapsed = 0.0 - interval = initial + elapsed = 0.0 # rebind-ok: the schedule accumulates the time it has already yielded + interval = initial # rebind-ok: the schedule doubles the interval up to the cap while interval > 0 and elapsed + interval <= timeout: yield interval elapsed += interval interval = min(interval * 2, maximum) -_SETTLED_KEY = "background_interaction_settled" +_CUSTOM_PRICING_KEYS: Final = frozenset(CustomPricingLiteLLMParams.model_fields.keys()) + +_CARRIED_METADATA_KEYS: Final = frozenset( + { + "model_info", + "model_group", + "deployment", + "tags", + "spend_logs_metadata", + "requester_metadata", + "requester_ip_address", + "user_agent", + "agent_id", + "session_id", + "endpoint", + "team_alias", + "team_id", + "applied_guardrails", + "prompt_management_metadata", + } +) + +_CARRIED_METADATA_PREFIX: Final = "user_api_" + +_UNCARRIED_METADATA_KEY: Final = "user_api_key_auth" + +_JSON_VALUE: Final = TypeAdapter(JsonValue) +_STRING: Final = TypeAdapter(str) +_OBJECT_MAPPING: Final = TypeAdapter(Mapping[str, object]) -def _is_settled(logging_obj: "LiteLLMLoggingObj") -> bool: - return logging_obj.model_call_details.get(_SETTLED_KEY) is True - - -def _claim_settlement(logging_obj: "LiteLLMLoggingObj") -> bool: - """ - Exactly-once gate between the poll task and the delete-time settlement: - both run on the same event loop and neither awaits between reading and - setting the flag, so whichever claims first owns billing or release. - """ - if _is_settled(logging_obj): +def _carries(key: str) -> bool: + if key == _UNCARRIED_METADATA_KEY: return False - logging_obj.model_call_details[_SETTLED_KEY] = True # rebind-ok: both settlers must see the same settlement flag - return True + return key in _CARRIED_METADATA_KEYS or key.startswith(_CARRIED_METADATA_PREFIX) + + +def _json_value(value: object) -> tuple[JsonValue, ...]: + try: + return (_JSON_VALUE.validate_python(to_jsonable_python(value)),) + except (PydanticSerializationError, ValidationError): + verbose_logger.debug("Dropping a background interaction metadata value that has no JSON form: %r", type(value)) + return () + + +def _json_values(items: Iterable[tuple[str, object]]) -> Mapping[str, JsonValue]: + parsed: Final = ((key, _json_value(value)) for key, value in items) + return MappingProxyType({key: values[0] for key, values in parsed if values}) + + +def _as_datetime(start_time: datetime | float) -> datetime: + return start_time if isinstance(start_time, datetime) else datetime.fromtimestamp(start_time, tz=timezone.utc) + + +def _create_context(logging_obj: "LiteLLMLoggingObj", custom_llm_provider: str) -> BackgroundInteractionCreateContext: + metadata: Final = _OBJECT_MAPPING.validate_python( + get_litellm_metadata_from_kwargs(kwargs=logging_obj.model_call_details) + ) + litellm_params: Final = _OBJECT_MAPPING.validate_python(logging_obj.litellm_params) + model: Final = logging_obj.model_call_details.get("model") + return BackgroundInteractionCreateContext( + model=model if isinstance(model, str) else logging_obj.model, + call_type=_STRING.validate_python(logging_obj.call_type), + litellm_call_id=logging_obj.litellm_call_id, + function_id=logging_obj.function_id, + litellm_trace_id=logging_obj.litellm_trace_id, + start_time=_as_datetime(logging_obj.start_time), + custom_llm_provider=custom_llm_provider, + metadata=_json_values((key, value) for key, value in metadata.items() if _carries(key)), + custom_pricing=_json_values( + (key, value) for key, value in litellm_params.items() if key in _CUSTOM_PRICING_KEYS and value is not None + ), + ) + + +def _rebuild_logging_obj(create_context: BackgroundInteractionCreateContext) -> "LiteLLMLoggingObj": + from litellm.litellm_core_utils.litellm_logging import Logging + + logging_obj: Final = Logging( + model=create_context.model, # pyright: ignore[reportArgumentType] # function_setup builds the live object with the same None for an agent-only create + messages=None, + stream=False, + call_type=create_context.call_type, + start_time=create_context.start_time, + litellm_call_id=create_context.litellm_call_id, + function_id=create_context.function_id, + litellm_trace_id=create_context.litellm_trace_id, + ) + litellm_params: Final = { + "metadata": dict(create_context.metadata), + **create_context.custom_pricing, + } + logging_obj.update_environment_variables( + litellm_params=litellm_params, + optional_params={}, + model=create_context.model, + custom_llm_provider=create_context.custom_llm_provider, + ) + return logging_obj + + +async def _settled_elsewhere(context: BackgroundInteractionPollContext) -> bool: + try: + return await context.store.is_claimed(context.interaction_id) + except Exception as e: # noqa: BLE001 # an unreadable store must not stop the poll; the claim below decides + verbose_logger.debug( + "Could not read the settlement state of background interaction %s: %s", context.interaction_id, e + ) + return False + + +async def _claim(context: BackgroundInteractionPollContext) -> bool | None: + """ + Exactly-once gate between every settler of one interaction, on every + replica: whoever claims first owns billing or release. ``None`` means the + store could not answer, so nothing is owned and the caller retries later. + """ + try: + return await context.store.claim(context.interaction_id) + except Exception: # noqa: BLE001 # an unanswerable claim is retried on the next poll rather than billed twice + verbose_logger.exception("Could not claim the settlement of background interaction %s", context.interaction_id) + return None + + +async def _record(context: BackgroundInteractionPollContext, outcome: SettlementOutcome) -> SettlementOutcome: + try: + await context.store.record_outcome(context.interaction_id, outcome) + except Exception: # noqa: BLE001 # the outcome is an audit trail; the claim already made the settlement exclusive + verbose_logger.exception("Could not record the settlement of background interaction %s", context.interaction_id) + return outcome async def poll_and_log_background_interaction_cost( context: BackgroundInteractionPollContext, - fetch_interaction: FetchInteraction = _fetch_interaction, -) -> None: - last_seen_status: str | None = None + fetch_interaction: FetchInteraction = fetch_background_interaction, +) -> SettlementOutcome | None: + last_response: InteractionsAPIResponse | None = None # rebind-ok: the give-up path settles from the last poll for interval in _poll_intervals( initial=context.initial_interval_seconds, maximum=context.max_interval_seconds, timeout=context.timeout_seconds, ): await asyncio.sleep(interval) - if _is_settled(context.logging_obj): - return + if await _settled_elsewhere(context): + return None try: response = await fetch_interaction(context) except Exception as e: # noqa: BLE001 # any fetch error must not kill the billing poll loop @@ -133,26 +355,26 @@ async def poll_and_log_background_interaction_cost( e, ) continue - last_seen_status = response.status + last_response = response if response.status not in _TERMINAL_STATUSES: continue - if not _claim_settlement(context.logging_obj): - return - if response.usage is not None: - await _bill_settled_interaction(logging_obj=context.logging_obj, response=response) - else: - await _release_open_budget_reservation(logging_obj=context.logging_obj) - return - if not _claim_settlement(context.logging_obj): - return - if last_seen_status is not None and last_seen_status not in _POLLABLE_STATUSES: + if (claimed := await _claim(context)) is None: + continue + if not claimed: + return None + return await _record(context, await _settle_terminal(logging_obj=context.logging_obj, response=response)) + if not await _claim(context): + return None + if last_response is not None and last_response.status in _TERMINAL_STATUSES: + return await _record(context, await _settle_terminal(logging_obj=context.logging_obj, response=last_response)) + if last_response is not None and last_response.status not in _POLLABLE_STATUSES: verbose_logger.error( "Gave up cost polling for background interaction %s after %ss: its last status %r is in neither " "the pollable nor the terminal set, so this proxy never learned how to settle it and its usage " "will not be tracked", context.interaction_id, context.timeout_seconds, - last_seen_status, + last_response.status, ) else: verbose_logger.warning( @@ -161,6 +383,15 @@ async def poll_and_log_background_interaction_cost( context.timeout_seconds, ) await _release_open_budget_reservation(logging_obj=context.logging_obj) + return await _record(context, "unsettled") + + +async def _settle_terminal(logging_obj: "LiteLLMLoggingObj", response: InteractionsAPIResponse) -> SettlementOutcome: + if response.status in _TERMINAL_STATUSES and response.usage is not None: + await _bill_settled_interaction(logging_obj=logging_obj, response=response) + return "billed" + await _release_open_budget_reservation(logging_obj=logging_obj) + return "released" async def _release_open_budget_reservation(logging_obj: "LiteLLMLoggingObj") -> None: @@ -173,8 +404,8 @@ async def _release_open_budget_reservation(logging_obj: "LiteLLMLoggingObj") -> settlement must release the reservation here or the spend counters stay pinned at the estimated cost. """ - metadata = get_litellm_metadata_from_kwargs(kwargs=logging_obj.model_call_details) - budget_reservation = metadata.get("user_api_key_budget_reservation") + metadata: Final = get_litellm_metadata_from_kwargs(kwargs=logging_obj.model_call_details) + budget_reservation: Final = metadata.get("user_api_key_budget_reservation") if not isinstance(budget_reservation, dict): return @@ -234,24 +465,88 @@ def missing_usage_is_expected(response: InteractionsAPIResponse) -> bool: @dataclass(frozen=True, slots=True) class _ActiveBackgroundPoll: - task: "asyncio.Task[None]" + task: "asyncio.Task[SettlementOutcome | None]" context: BackgroundInteractionPollContext -_ACTIVE_POLLS: dict[str, _ActiveBackgroundPoll] = {} # mutable-ok: asyncio needs strong refs to running poll tasks +_ACTIVE_POLLS: Final[dict[str, _ActiveBackgroundPoll]] = {} # mutable-ok: asyncio needs strong refs to poll tasks -def _discard_poll(interaction_id: str, task: "asyncio.Task[None]") -> None: - entry = _ACTIVE_POLLS.get(interaction_id) +def _discard_poll(interaction_id: str, task: "asyncio.Task[SettlementOutcome | None]") -> None: + entry: Final = _ACTIVE_POLLS.get(interaction_id) if entry is not None and entry.task is task: del _ACTIVE_POLLS[interaction_id] -def maybe_schedule_background_interaction_cost_polling( +def _track_poll( + context: BackgroundInteractionPollContext, fetch_interaction: FetchInteraction +) -> "asyncio.Task[SettlementOutcome | None]": + task: Final = asyncio.create_task(poll_and_log_background_interaction_cost(context, fetch_interaction)) + _ACTIVE_POLLS[context.interaction_id] = _ActiveBackgroundPoll(task=task, context=context) + task.add_done_callback( + lambda finished, interaction_id=context.interaction_id: _discard_poll(interaction_id, finished) + ) + return task + + +@dataclass(frozen=True, slots=True) +class _UnverifiedRegistrationStore: + """ + Store of a create whose registration raised, so whether its row landed is + unknown until the durable store answers. The settlement claim asks it + first, and only an interaction it reports as never stored settles through + the local gate, which no other process can reach. + """ + + durable: BackgroundSettlementStore + local: InMemoryBackgroundSettlementStore = field(default_factory=InMemoryBackgroundSettlementStore) + + async def register(self, pending: PendingBackgroundInteraction) -> None: + await self.durable.register(pending) + + async def pending(self, interaction_id: str) -> PendingBackgroundInteraction | None: + return await self.durable.pending(interaction_id) + + async def is_claimed(self, interaction_id: str) -> bool: + return await self.local.is_claimed(interaction_id) or await self.durable.is_claimed(interaction_id) + + async def claim(self, interaction_id: str) -> bool: + if await self.durable.claim(interaction_id): + return True + if await self.durable.is_claimed(interaction_id): + return False + return await self.local.claim(interaction_id) + + async def record_outcome(self, interaction_id: str, outcome: SettlementOutcome) -> None: + if await self.local.is_claimed(interaction_id): + return + await self.durable.record_outcome(interaction_id, outcome) + + async def unclaimed(self) -> Sequence[PendingBackgroundInteraction]: + return await self.durable.unclaimed() + + +async def _registered_store( + store: BackgroundSettlementStore, pending: PendingBackgroundInteraction +) -> BackgroundSettlementStore: + try: + await store.register(pending) + except Exception: # noqa: BLE001 # a store outage must not fail the create; the claim learns if the row landed + verbose_logger.exception( + "Could not durably register background interaction %s; its settlement claim decides whether the row landed", + pending.interaction_id, + ) + return _UnverifiedRegistrationStore(durable=store) + return store + + +async def maybe_schedule_background_interaction_cost_polling( response: object, create_kwargs: Mapping[str, object], custom_llm_provider: str, -) -> "asyncio.Task[None] | None": + store: BackgroundSettlementStore | None = None, + fetch_interaction: FetchInteraction = fetch_background_interaction, +) -> "asyncio.Task[SettlementOutcome | None] | None": from litellm.litellm_core_utils.litellm_logging import Logging if not BACKGROUND_INTERACTION_COST_POLLING_ENABLED: @@ -260,52 +555,141 @@ def maybe_schedule_background_interaction_cost_polling( return None if not is_pollable_background_interaction(response): return None - logging_obj = create_kwargs.get("litellm_logging_obj") + logging_obj: Final = create_kwargs.get("litellm_logging_obj") if not isinstance(logging_obj, Logging): return None - try: - asyncio.get_running_loop() - except RuntimeError: - return None - api_key = create_kwargs.get("api_key") - api_base = create_kwargs.get("api_base") - context = BackgroundInteractionPollContext( + api_key: Final = create_kwargs.get("api_key") + api_base: Final = create_kwargs.get("api_base") + pending: Final = PendingBackgroundInteraction( + interaction_id=response.id, + custom_llm_provider=custom_llm_provider, + create_context=_create_context(logging_obj, custom_llm_provider), + created_at=datetime.now(timezone.utc), + ) + context: Final = BackgroundInteractionPollContext( interaction_id=response.id, custom_llm_provider=custom_llm_provider, logging_obj=logging_obj, api_key=api_key if isinstance(api_key, str) else None, api_base=api_base if isinstance(api_base, str) else None, + store=await _registered_store(store or _STORE.store, pending), ) - task = asyncio.create_task(poll_and_log_background_interaction_cost(context)) - _ACTIVE_POLLS[context.interaction_id] = _ActiveBackgroundPoll(task=task, context=context) - task.add_done_callback( - lambda finished, interaction_id=context.interaction_id: _discard_poll(interaction_id, finished) - ) - return task + return _track_poll(context, fetch_interaction) + + +async def _pending(store: BackgroundSettlementStore, interaction_id: str) -> PendingBackgroundInteraction | None: + try: + return await store.pending(interaction_id) + except Exception: # noqa: BLE001 # an unreadable store leaves the interaction to its poll or the counter TTL + verbose_logger.exception("Could not look up background interaction %s before its delete", interaction_id) + return None + + +async def _fetch_before_delete( + context: BackgroundInteractionPollContext, fetch_interaction: FetchInteraction +) -> InteractionsAPIResponse | None: + try: + return await fetch_interaction(context) + except Exception as e: # noqa: BLE001 # the caller decides what an unfetchable pre-delete state means + verbose_logger.debug( + "Could not fetch background interaction %s before its delete: %s", context.interaction_id, e + ) + return None + + +async def _settle_before_delete( + context: BackgroundInteractionPollContext, response: InteractionsAPIResponse | None +) -> SettlementOutcome | None: + if not await _claim(context): + return None + if response is None: + await _release_open_budget_reservation(logging_obj=context.logging_obj) + return await _record(context, "released") + return await _record(context, await _settle_terminal(logging_obj=context.logging_obj, response=response)) async def maybe_settle_background_interaction_before_delete( interaction_id: str, - fetch_interaction: FetchInteraction = _fetch_interaction, -) -> None: - entry = _ACTIVE_POLLS.get(interaction_id) - if entry is None: - return - context = entry.context + delete_kwargs: Mapping[str, object], + fetch_interaction: FetchInteraction = fetch_background_interaction, + store: BackgroundSettlementStore | None = None, +) -> SettlementOutcome | None: + entry: Final = _ACTIVE_POLLS.get(interaction_id) + if entry is not None and not entry.context.resumed: + return await _settle_before_delete(entry.context, await _fetch_before_delete(entry.context, fetch_interaction)) + settlement_store: Final = store or _STORE.store + pending: Final = await _pending(settlement_store, interaction_id) + if pending is None: + return None + api_key: Final = delete_kwargs.get("api_key") + api_base: Final = delete_kwargs.get("api_base") + context: Final = BackgroundInteractionPollContext( + interaction_id=interaction_id, + custom_llm_provider=pending.custom_llm_provider, + logging_obj=_rebuild_logging_obj(pending.create_context), + api_key=api_key if isinstance(api_key, str) else None, + api_base=api_base if isinstance(api_base, str) else None, + store=settlement_store, + ) try: - response = await fetch_interaction(context) - except Exception as e: # noqa: BLE001 # unfetchable pre-delete state settles by releasing the reservation + response: Final = await fetch_interaction(context) + except Exception: verbose_logger.debug( - "Could not fetch background interaction %s before delete, releasing its reservation: %s", + "Failing the delete of background interaction %s: this process could not fetch it with the delete's " + "credentials, so the poll that created it keeps the bill", interaction_id, - e, ) - if _claim_settlement(context.logging_obj): - await _release_open_budget_reservation(logging_obj=context.logging_obj) - return - if not _claim_settlement(context.logging_obj): - return - if response.status in _TERMINAL_STATUSES and response.usage is not None: - await _bill_settled_interaction(logging_obj=context.logging_obj, response=response) - return - await _release_open_budget_reservation(logging_obj=context.logging_obj) + raise + return await _settle_before_delete(context, response) + + +async def _unclaimed(store: BackgroundSettlementStore) -> Sequence[PendingBackgroundInteraction]: + try: + return await store.unclaimed() + except Exception: # noqa: BLE001 # an unreadable store at startup leaves its rows for the next boot + verbose_logger.exception("Could not list the unsettled background interactions") + return () + + +@dataclass(frozen=True, slots=True) +class PollSchedule: + initial_interval_seconds: float = BACKGROUND_INTERACTION_COST_POLL_INITIAL_INTERVAL_SECONDS + max_interval_seconds: float = BACKGROUND_INTERACTION_COST_POLL_MAX_INTERVAL_SECONDS + timeout_seconds: float = BACKGROUND_INTERACTION_COST_POLL_TIMEOUT_SECONDS + + +DEFAULT_POLL_SCHEDULE: Final = PollSchedule() + + +def _resumed_context( + row: PendingBackgroundInteraction, store: BackgroundSettlementStore, schedule: PollSchedule +) -> BackgroundInteractionPollContext: + age_seconds: Final = (datetime.now(timezone.utc) - row.created_at).total_seconds() + return BackgroundInteractionPollContext( + interaction_id=row.interaction_id, + custom_llm_provider=row.custom_llm_provider, + logging_obj=_rebuild_logging_obj(row.create_context), + initial_interval_seconds=schedule.initial_interval_seconds, + max_interval_seconds=schedule.max_interval_seconds, + timeout_seconds=max(schedule.timeout_seconds - age_seconds, schedule.initial_interval_seconds), + store=store, + resumed=True, + ) + + +async def resume_unsettled_background_interactions( + store: BackgroundSettlementStore, + fetch_interaction: FetchInteraction = fetch_background_interaction, + schedule: PollSchedule = DEFAULT_POLL_SCHEDULE, +) -> tuple["asyncio.Task[SettlementOutcome | None]", ...]: + """ + Pick up every settlement no process has claimed, which is what a replica + that died mid-poll leaves behind. Each resumed poll keeps the remaining + share of the original timeout and gets at least one fetch, so a completed + interaction is still billed however late the resume comes. + """ + return tuple( + _track_poll(_resumed_context(row, store, schedule), fetch_interaction) + for row in await _unclaimed(store) + if row.interaction_id not in _ACTIVE_POLLS + ) diff --git a/litellm/interactions/main.py b/litellm/interactions/main.py index 8a33e9b39c5..74c1799190d 100644 --- a/litellm/interactions/main.py +++ b/litellm/interactions/main.py @@ -175,7 +175,7 @@ async def acreate( else: response = init_response - maybe_schedule_background_interaction_cost_polling( + await maybe_schedule_background_interaction_cost_polling( response=response, create_kwargs=kwargs, custom_llm_provider=custom_llm_provider, @@ -464,7 +464,7 @@ async def adelete( extra_headers: dict[str, Any] | None = None, timeout: float | httpx.Timeout | None = None, custom_llm_provider: str | None = None, - **kwargs, + **kwargs: object, ) -> DeleteInteractionResult: """Async: Delete an interaction by its ID.""" local_vars: Final = locals() @@ -472,7 +472,7 @@ async def adelete( loop: Final = asyncio.get_event_loop() kwargs["adelete_interaction"] = True - await maybe_settle_background_interaction_before_delete(interaction_id=interaction_id) + await maybe_settle_background_interaction_before_delete(interaction_id=interaction_id, delete_kwargs=kwargs) func: Final = partial( delete, diff --git a/litellm/llms/gemini/interactions/transformation.py b/litellm/llms/gemini/interactions/transformation.py index ab2c1440fb7..0e898147d90 100644 --- a/litellm/llms/gemini/interactions/transformation.py +++ b/litellm/llms/gemini/interactions/transformation.py @@ -313,6 +313,12 @@ class GoogleAIStudioInteractionsConfig(BaseInteractionsAPIConfig): raw_response: httpx.Response, logging_obj: LiteLLMLoggingObj, ) -> InteractionsAPIResponse: + if not 200 <= raw_response.status_code < 300: + raise GeminiError( + message=raw_response.text, + status_code=raw_response.status_code, + headers=dict(raw_response.headers), + ) try: raw_json: Final = _interaction_body(raw_response) except Exception: diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index e406e6faec4..caec0ed524a 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -777,6 +777,9 @@ from litellm.proxy.shutdown.scheduled_jobs import ( pause_scheduled_jobs, stop_in_flight_scheduler_jobs, ) +from litellm.proxy.spend_tracking.background_interaction_settlement import ( + install_background_interaction_settlement, +) from litellm.proxy.spend_tracking.budget_reservation import ( get_budget_window_start, release_unbound_budget_reservation, @@ -1394,6 +1397,7 @@ async def proxy_startup_event(app: FastAPI) -> AsyncGenerator[ProxyLifespanState await asyncio.sleep(5) asyncio.create_task(_run_agent_grant_id_migration()) + await install_background_interaction_settlement(prisma_client) ## A coordination_redis block saved from the admin UI lives in the database, ## which is only reachable once the prisma client exists. Apply it here, before diff --git a/litellm/proxy/schema.prisma b/litellm/proxy/schema.prisma index aba89526cf6..e75a9b71c7c 100644 --- a/litellm/proxy/schema.prisma +++ b/litellm/proxy/schema.prisma @@ -1916,6 +1916,22 @@ model LiteLLM_WorkflowMessage { @@index([run_id]) } +// Pending billing settlements for background interactions, keyed by the +// interaction id so any replica can settle one that another replica created. +// `claimed_at` is the exactly-once gate: the first conditional update wins. +model LiteLLM_BackgroundInteractionSettlement { + interaction_id String @id + custom_llm_provider String + create_context Json + created_at DateTime @default(now()) + claimed_at DateTime? + claimed_by String? + settled_at DateTime? + outcome String? + + @@index([claimed_at], map: "idx_background_interaction_settlement_claimed_at") +} + model LiteLLM_Lens { id String @id version Int @default(0) diff --git a/litellm/proxy/spend_tracking/background_interaction_settlement.py b/litellm/proxy/spend_tracking/background_interaction_settlement.py new file mode 100644 index 00000000000..716dc28565e --- /dev/null +++ b/litellm/proxy/spend_tracking/background_interaction_settlement.py @@ -0,0 +1,211 @@ +import asyncio +import os +import socket +from collections.abc import Awaitable, Mapping, Sequence +from dataclasses import dataclass +from datetime import datetime, timezone +from itertools import chain +from types import MappingProxyType +from typing import TYPE_CHECKING, Final, Protocol, TypeVar + +from pydantic import ValidationError +from typing_extensions import ReadOnly, TypedDict + +from litellm._logging import verbose_proxy_logger +from litellm.constants import BACKGROUND_INTERACTION_COST_POLLING_ENABLED +from litellm.interactions.background_cost_polling import ( + DEFAULT_POLL_SCHEDULE, + BackgroundInteractionCreateContext, + FetchInteraction, + PendingBackgroundInteraction, + PollSchedule, + SettlementOutcome, + configure_background_settlement_store, + fetch_background_interaction, + resume_unsettled_background_interactions, +) +from litellm.repositories.table_repositories import BackgroundInteractionSettlementRepository + +if TYPE_CHECKING: + from litellm.proxy.utils import PrismaClient + + +class _SettlementRow(Protocol): + @property + def interaction_id(self) -> str: ... + @property + def custom_llm_provider(self) -> str: ... + @property + def create_context(self) -> object: ... + @property + def created_at(self) -> datetime: ... + @property + def claimed_at(self) -> datetime | None: ... + + +class _NewSettlementRow(TypedDict): + interaction_id: ReadOnly[str] + custom_llm_provider: ReadOnly[str] + create_context: ReadOnly[object] + created_at: ReadOnly[datetime] + + +class _RowKey(TypedDict): + interaction_id: ReadOnly[str] + + +class _UnclaimedRowKey(TypedDict): + interaction_id: ReadOnly[str] + claimed_at: ReadOnly[None] + + +class _UnclaimedRows(TypedDict): + claimed_at: ReadOnly[None] + + +class _Claim(TypedDict): + claimed_at: ReadOnly[datetime] + claimed_by: ReadOnly[str] + + +class _Outcome(TypedDict): + settled_at: ReadOnly[datetime] + outcome: ReadOnly[SettlementOutcome] + create_context: ReadOnly[object] + + +class _SettlementTableActions(Protocol): + def create(self, *, data: _NewSettlementRow) -> Awaitable[_SettlementRow]: ... + + def find_unique(self, *, where: _RowKey) -> Awaitable[_SettlementRow | None]: ... + + def find_many(self, *, where: _UnclaimedRows) -> Awaitable[Sequence[_SettlementRow]]: ... + + def update_many(self, *, data: _Claim | _Outcome, where: _RowKey | _UnclaimedRowKey) -> Awaitable[int]: ... + + +def _settlement_table(prisma_client: "PrismaClient") -> _SettlementTableActions: + return BackgroundInteractionSettlementRepository(prisma_client).table + + +_CLEARED_CREATE_CONTEXT: Final[Mapping[str, object]] = MappingProxyType({}) +_T = TypeVar("_T") + + +async def _read_from_a_table_that_may_not_exist(query: Awaitable[_T], when_missing: _T) -> _T: + from prisma.errors import TableNotFoundError # noqa: PLC0415 # local import: prisma may be ungenerated at load + + try: + return await query + except TableNotFoundError: + return when_missing + + +def _json(data: Mapping[str, object]) -> object: + from prisma import Json # noqa: PLC0415 # local import: prisma may be ungenerated at module load in some tools + + return Json.keys(**data) + + +def _pending_rows(rows: Sequence[_SettlementRow]) -> tuple[PendingBackgroundInteraction, ...]: + return tuple(chain.from_iterable(_pending_row(row) for row in rows)) + + +def _pending_row(row: _SettlementRow) -> tuple[PendingBackgroundInteraction, ...]: + try: + create_context: Final = BackgroundInteractionCreateContext.model_validate(row.create_context) + except ValidationError: + verbose_proxy_logger.exception( + "Background interaction %s has a settlement row this version cannot read; leaving it unsettled", + row.interaction_id, + ) + return () + return ( + PendingBackgroundInteraction( + interaction_id=row.interaction_id, + custom_llm_provider=row.custom_llm_provider, + create_context=create_context, + created_at=row.created_at, + ), + ) + + +@dataclass(frozen=True, slots=True) +class PrismaBackgroundSettlementStore: + table: _SettlementTableActions + claimed_by: str + + async def register(self, pending: PendingBackgroundInteraction) -> None: + await self.table.create( + data=_NewSettlementRow( + interaction_id=pending.interaction_id, + custom_llm_provider=pending.custom_llm_provider, + create_context=_json(pending.create_context.model_dump(mode="json")), + created_at=pending.created_at, + ) + ) + + async def pending(self, interaction_id: str) -> PendingBackgroundInteraction | None: + row: Final = await self._row(interaction_id) + if row is None or row.claimed_at is not None: + return None + return next(iter(_pending_row(row)), None) + + async def is_claimed(self, interaction_id: str) -> bool: + row: Final = await self._row(interaction_id) + return row is not None and row.claimed_at is not None + + async def claim(self, interaction_id: str) -> bool: + claimed_rows: Final = await _read_from_a_table_that_may_not_exist( + self.table.update_many( + data=_Claim(claimed_at=datetime.now(timezone.utc), claimed_by=self.claimed_by), + where=_UnclaimedRowKey(interaction_id=interaction_id, claimed_at=None), + ), + when_missing=0, + ) + return claimed_rows == 1 + + async def _row(self, interaction_id: str) -> _SettlementRow | None: + return await _read_from_a_table_that_may_not_exist( + self.table.find_unique(where=_RowKey(interaction_id=interaction_id)), when_missing=None + ) + + async def record_outcome(self, interaction_id: str, outcome: SettlementOutcome) -> None: + await self.table.update_many( + data=_Outcome( + settled_at=datetime.now(timezone.utc), outcome=outcome, create_context=_json(_CLEARED_CREATE_CONTEXT) + ), + where=_RowKey(interaction_id=interaction_id), + ) + + async def unclaimed(self) -> Sequence[PendingBackgroundInteraction]: + return _pending_rows(await self.table.find_many(where=_UnclaimedRows(claimed_at=None))) + + +async def configure_background_interaction_settlement( + table: _SettlementTableActions, + claimed_by: str, + fetch_interaction: FetchInteraction = fetch_background_interaction, + schedule: PollSchedule = DEFAULT_POLL_SCHEDULE, +) -> tuple["asyncio.Task[SettlementOutcome | None]", ...]: + if not BACKGROUND_INTERACTION_COST_POLLING_ENABLED: + return () + store: Final = PrismaBackgroundSettlementStore(table=table, claimed_by=claimed_by) + configure_background_settlement_store(store) + resumed: Final = await resume_unsettled_background_interactions(store, fetch_interaction, schedule) + if resumed: + verbose_proxy_logger.info("Resumed cost polling for %s unsettled background interactions", len(resumed)) + return resumed + + +async def install_background_interaction_settlement(prisma_client: "PrismaClient") -> None: + try: + await configure_background_interaction_settlement( + table=BackgroundInteractionSettlementRepository(prisma_client).table, + claimed_by=f"{socket.gethostname()}:{os.getpid()}", + ) + except Exception as e: # noqa: BLE001 # a boot step must survive any DB error; billing then settles in-process as before + verbose_proxy_logger.warning( + "Durable background interaction settlement is off on this replica, so billing settles in-process only: %s", + e, + ) diff --git a/litellm/repositories/table_repositories.py b/litellm/repositories/table_repositories.py index 4e511a2ec93..0f85818cba3 100644 --- a/litellm/repositories/table_repositories.py +++ b/litellm/repositories/table_repositories.py @@ -267,5 +267,11 @@ class AdaptiveRouterSessionRepository(PrismaTableRepository["prisma_models.LiteL table_name = "litellm_adaptiveroutersession" +class BackgroundInteractionSettlementRepository( + PrismaTableRepository["prisma_models.LiteLLM_BackgroundInteractionSettlement"] +): + table_name = "litellm_backgroundinteractionsettlement" + + class RetiredAgentRepository(PrismaTableRepository["prisma_models.LiteLLM_RetiredAgent"]): table_name = "litellm_retiredagent" diff --git a/schema.prisma b/schema.prisma index aba89526cf6..e75a9b71c7c 100644 --- a/schema.prisma +++ b/schema.prisma @@ -1916,6 +1916,22 @@ model LiteLLM_WorkflowMessage { @@index([run_id]) } +// Pending billing settlements for background interactions, keyed by the +// interaction id so any replica can settle one that another replica created. +// `claimed_at` is the exactly-once gate: the first conditional update wins. +model LiteLLM_BackgroundInteractionSettlement { + interaction_id String @id + custom_llm_provider String + create_context Json + created_at DateTime @default(now()) + claimed_at DateTime? + claimed_by String? + settled_at DateTime? + outcome String? + + @@index([claimed_at], map: "idx_background_interaction_settlement_claimed_at") +} + model LiteLLM_Lens { id String @id version Int @default(0) diff --git a/tests/integration/_support/process.py b/tests/integration/_support/process.py index 891874bdfa6..978ac2ec092 100644 --- a/tests/integration/_support/process.py +++ b/tests/integration/_support/process.py @@ -5,7 +5,7 @@ import subprocess import sys import time import uuid -from collections.abc import Iterator, Mapping +from collections.abc import Generator, Iterator, Mapping from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path @@ -219,3 +219,65 @@ def owned_proxy_process( yield OwnedProxy(Gateway(client, gateway.key, gateway.upstream_url), process, launch.log) finally: _stop(process) + + +_UPSTREAM_READY_SECONDS: Final = 60 + + +class UpstreamSlot: + """A scripted upstream a test module owns on a fixed port, so a cell can take it down and bring it back.""" + + __slots__ = ("directory", "port", "process", "root") + + def __init__(self, directory: Path, port: int, root: Path) -> None: + self.directory = directory + self.port = port + self.root = root + self.process: subprocess.Popen[bytes] | None = None + + @property + def url(self) -> str: + return f"http://127.0.0.1:{self.port}" + + def start(self) -> None: + assert self.process is None, "Owned upstream is already running" + output: Final = Path(os.environ.get("INTEGRATION_RESULTS_DIR") or self.directory) + log_path: Final = output / f"owned-upstream-{self.port}-{uuid.uuid4().hex}.log" + with log_path.open("w") as log: + process: Final = subprocess.Popen( + [sys.executable, "-m", "integration._support.upstream", "--port", str(self.port)], + cwd=self.root, + env=dict(os.environ), + stdout=log, + stderr=subprocess.STDOUT, + start_new_session=True, + ) + self.process = process + deadline: Final = time.monotonic() + _UPSTREAM_READY_SECONDS + while process.poll() is None: + try: + if httpx.get(f"{self.url}/health", timeout=2, trust_env=False).status_code == 200: + return + except httpx.TransportError: + pass + assert time.monotonic() < deadline, f"Owned upstream readiness deadline exceeded: {log_path}" + time.sleep(0.1) + raise AssertionError(f"Owned upstream exited before readiness: {log_path}") + + def stop(self) -> None: + process: Final = self.process + assert process is not None, "Owned upstream is not running" + self.process = None + _stop(process) + + +@contextmanager +def owned_upstream(directory: Path) -> Generator[UpstreamSlot]: + root: Final = Path(os.environ.get("INTEGRATION_PROXY_ROOT") or Path(__file__).resolve().parents[3]) + slot: Final = UpstreamSlot(directory, _free_port(), root) + slot.start() + try: + yield slot + finally: + if slot.process is not None: + slot.stop() diff --git a/tests/integration/_support/upstream.py b/tests/integration/_support/upstream.py index a06c6099f0e..40423ae4d3a 100644 --- a/tests/integration/_support/upstream.py +++ b/tests/integration/_support/upstream.py @@ -29,7 +29,7 @@ from integration.cost_calculation.cost_tracking_case import ( StoredResponse, TextResponse, ) -from pydantic import BaseModel, JsonValue, TypeAdapter, ValidationError +from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter, ValidationError from starlette.applications import Starlette from starlette.requests import Request from starlette.responses import JSONResponse, Response, StreamingResponse @@ -64,6 +64,19 @@ class Observation: path: str authorization: str body: dict[str, JsonValue] + method: str = "POST" + api_key: str = "" + + +class InteractionState(BaseModel): + """What the scripted Interactions API answers for one interaction id until a DELETE drops it.""" + + model_config = ConfigDict(extra="forbid") + + status: str + usage: dict[str, JsonValue] | None = None + get_status: int = 200 + delay_seconds: float = 0 class _ScenarioRegistration(BaseModel): @@ -137,6 +150,7 @@ class Provider: observations: SimpleQueue[Observation] = field(default_factory=SimpleQueue) scripts: dict[str, deque[int]] = field(default_factory=dict) scenario_store: ScenarioStore = field(default_factory=ScenarioStore) + interactions: dict[str, InteractionState] = field(default_factory=dict) async def chat(self, request: Request) -> Response: body: Final = JSON_OBJECT.validate_json(await request.body()) @@ -218,7 +232,14 @@ class Provider: return JSONResponse( { "requests": [ - {"path": value.path, "authorization": value.authorization, "body": value.body} for value in values + { + "path": value.path, + "authorization": value.authorization, + "body": value.body, + "method": value.method, + "api_key": value.api_key, + } + for value in values ] } ) @@ -264,7 +285,14 @@ class Provider: if raw_body: body: Final = JSON_OBJECT.validate_json(raw_body) if isinstance(body, dict): - self.observations.put(Observation(request.url.path, request.headers.get("authorization", ""), body)) + self.observations.put( + Observation( + request.url.path, + request.headers.get("authorization", ""), + body, + api_key=request.headers.get("x-goog-api-key", ""), + ) + ) if isinstance(response, RoutedResponse): route_key: Final = f"{request.method} /{'/'.join(segments[1:])}" route: Final = next( @@ -280,6 +308,58 @@ class Provider: return self._response(route, scenario_id) return self._response(response, scenario_id) + async def interaction_state(self, request: Request) -> Response: + interaction_id: Final = cast(str, request.path_params["interaction_id"]) + if request.method == "DELETE": + self.interactions.pop(interaction_id, None) + return JSONResponse({"interaction_id": interaction_id, "registered": False}) + try: + state: Final = InteractionState.model_validate_json(await request.body()) + except ValidationError as exc: + return JSONResponse({"error": str(exc)}, status_code=400) + self.interactions[interaction_id] = state + return JSONResponse({"interaction_id": interaction_id, "registered": True}) + + def _observe_interaction(self, request: Request) -> None: + self.observations.put( + Observation( + request.url.path, + request.headers.get("authorization", ""), + {}, + method=request.method, + api_key=request.headers.get("x-goog-api-key", ""), + ) + ) + + async def interaction(self, request: Request) -> Response: + self._observe_interaction(request) + interaction_id: Final = cast(str, request.path_params["interaction_id"]) + state: Final = self.interactions.get(interaction_id) + if state is None: + return JSONResponse(_interaction_not_found(interaction_id), status_code=404) + if state.delay_seconds: + await asyncio.sleep(state.delay_seconds) + if request.method == "DELETE": + if self.interactions.pop(interaction_id, None) is None: + return JSONResponse(_interaction_not_found(interaction_id), status_code=404) + return JSONResponse({}) + if state.get_status != 200: + return JSONResponse( + {"error": {"code": state.get_status, "message": "Scripted interaction fetch failure"}}, + status_code=state.get_status, + ) + return JSONResponse(_interaction_body(interaction_id, state)) + + async def cancel_interaction(self, request: Request) -> Response: + self._observe_interaction(request) + interaction_id: Final = cast(str, request.path_params["interaction_id"]) + state: Final = self.interactions.get(interaction_id) + if state is None: + return JSONResponse(_interaction_not_found(interaction_id), status_code=404) + cancelled: Final = InteractionState(status="cancelled", usage=state.usage, get_status=state.get_status) + self.interactions[interaction_id] = cancelled + return JSONResponse(_interaction_body(interaction_id, cancelled)) + async def realtime(self, websocket: WebSocket) -> None: scenario_id: Final = websocket.headers.get("authorization", "").removeprefix("Bearer ") response: Final = self.scenario_store.get(scenario_id) @@ -392,6 +472,19 @@ class Provider: Route("/v1/embeddings", embeddings, methods=["POST"]), Route("/v1/moderations", moderations, methods=["POST"]), Route("/vector_stores/{vector_store_id}/search", self.vector_store_search, methods=["POST"]), + Route("/__interactions/{interaction_id}", self.interaction_state, methods=["PUT", "DELETE"]), + Route("/v1beta/interactions/{interaction_id}:cancel", self.cancel_interaction, methods=["POST"]), + Route("/v1beta/interactions/{interaction_id}", self.interaction, methods=["GET", "DELETE"]), + Route( + "/{prefix:path}/v1beta/interactions/{interaction_id}:cancel", + self.cancel_interaction, + methods=["POST"], + ), + Route( + "/{prefix:path}/v1beta/interactions/{interaction_id}", + self.interaction, + methods=["GET", "DELETE"], + ), Route("/{path:path}", self.scripted, methods=["POST"]), Route("/{path:path}", self.scripted, methods=["GET"]), WebSocketRoute("/v1/realtime", self.realtime), @@ -402,6 +495,21 @@ class Provider: CONTROL_URL: Final = os.environ.get("INTEGRATION_UPSTREAM_URL", "http://127.0.0.1:8190").rstrip("/") +def _interaction_not_found(interaction_id: str) -> dict[str, JsonValue]: + return {"error": {"code": 404, "message": f"Interaction {interaction_id} not found", "status": "NOT_FOUND"}} + + +def _interaction_body(interaction_id: str, state: InteractionState) -> dict[str, JsonValue]: + return { + "id": interaction_id, + "object": "interaction", + "model": "gemini-3.8-flash", + "status": state.status, + "steps": [], + "usage": state.usage, + } + + @dataclass(frozen=True, slots=True) class ScenarioHandle: scenario_id: str @@ -411,9 +519,9 @@ class ScenarioHandle: return f"{self.control_url}/{self.scenario_id}" -def register_scenario(scenario_id: str, response: StoredResponse) -> ScenarioHandle: +def register_scenario(scenario_id: str, response: StoredResponse, *, control_url: str = CONTROL_URL) -> ScenarioHandle: http_response: Final = httpx.post( - f"{CONTROL_URL}/__scenarios", + f"{control_url}/__scenarios", json={"scenario_id": scenario_id, "response": response.model_dump(mode="json")}, trust_env=False, timeout=15, @@ -421,19 +529,35 @@ def register_scenario(scenario_id: str, response: StoredResponse) -> ScenarioHan http_response.raise_for_status() return ScenarioHandle( scenario_id=scenario_id, - control_url=CONTROL_URL, + control_url=control_url, ) def delete_scenario(handle: ScenarioHandle) -> None: response: Final = httpx.delete( - f"{CONTROL_URL}/__scenarios/{handle.scenario_id}", + f"{handle.control_url}/__scenarios/{handle.scenario_id}", trust_env=False, timeout=15, ) response.raise_for_status() +def set_interaction_state(control_url: str, interaction_id: str, state: InteractionState) -> None: + response: Final = httpx.put( + f"{control_url}/__interactions/{interaction_id}", + content=state.model_dump_json(), + headers={"content-type": "application/json"}, + trust_env=False, + timeout=15, + ) + response.raise_for_status() + + +def clear_interaction_state(control_url: str, interaction_id: str) -> None: + response: Final = httpx.delete(f"{control_url}/__interactions/{interaction_id}", trust_env=False, timeout=15) + response.raise_for_status() + + def main() -> None: parser: Final = argparse.ArgumentParser() parser.add_argument("--port", type=int, default=8190) diff --git a/tests/integration/spend/test_background_interaction_settlement.py b/tests/integration/spend/test_background_interaction_settlement.py new file mode 100644 index 00000000000..11444396697 --- /dev/null +++ b/tests/integration/spend/test_background_interaction_settlement.py @@ -0,0 +1,791 @@ +import math +import socket +import time +import uuid +from collections.abc import Iterator, Mapping, Sequence +from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass +from pathlib import Path +from typing import Final + +import httpx +import psutil +import pytest +import yaml +from integration._support.client import ( + JSON_OBJECT, + Gateway, + eventually, + gateway_from_environment, + object_value, + string_value, +) +from integration._support.database import read_rows, write_rows +from integration._support.process import ( + UpstreamSlot, + group_members, + owned_proxy, + owned_proxy_process, + owned_upstream, +) +from integration._support.upstream import ( + InteractionState, + clear_interaction_state, + register_scenario, + set_interaction_state, +) +from integration.cost_calculation.cost_tracking_case import JsonResponse, RoutedResponse +from pydantic import JsonValue + +from litellm.proxy.spend_tracking.budget_reservation import DEFAULT_MAX_OUTPUT_TOKENS_FALLBACK + +pytestmark: Final = pytest.mark.timeout(900) + +_MODEL: Final = "gemini/gemini-3.8-flash" +_INPUT_TOKENS: Final = 300 +_OUTPUT_TOKENS: Final = 41 +_USAGE: Final[dict[str, JsonValue]] = { + "total_input_tokens": _INPUT_TOKENS, + "total_output_tokens": _OUTPUT_TOKENS, + "total_tool_use_tokens": 0, + "total_reasoning_tokens": 0, +} +_CUSTOM_INPUT_RATE: Final = 2e-06 +_CUSTOM_OUTPUT_RATE: Final = 4e-05 +_ENV_KEY: Final = "integration-gemini-env-key" +_DEPLOYMENT_KEY: Final = "integration-gemini-deployment-key" +_CREATOR_POLL: Final = {"BACKGROUND_INTERACTION_COST_POLL_INITIAL_INTERVAL_SECONDS": "300"} +_SETTLER_POLL: Final = { + "BACKGROUND_INTERACTION_COST_POLL_INITIAL_INTERVAL_SECONDS": "1", + "BACKGROUND_INTERACTION_COST_POLL_MAX_INTERVAL_SECONDS": "1", + "BACKGROUND_INTERACTION_COST_POLL_TIMEOUT_SECONDS": "8", +} +_RESUMER_POLL: Final = { + "BACKGROUND_INTERACTION_COST_POLL_INITIAL_INTERVAL_SECONDS": "1", + "BACKGROUND_INTERACTION_COST_POLL_MAX_INTERVAL_SECONDS": "1", + "BACKGROUND_INTERACTION_COST_POLL_TIMEOUT_SECONDS": "120", +} +_SPEND_QUERY: Final = ( + "SELECT request_id, spend, call_type, status, model, prompt_tokens, completion_tokens " + 'FROM "LiteLLM_SpendLogs" WHERE request_id = %s' +) +_SETTLEMENT_QUERY: Final = ( + "SELECT interaction_id, claimed_by, outcome, claimed_at IS NOT NULL AS claimed, " + 'settled_at IS NOT NULL AS settled, create_context FROM "LiteLLM_BackgroundInteractionSettlement" ' + "WHERE interaction_id = %s" +) +_SETTLEMENT_TABLE_PRESENT_QUERY: Final = "SELECT to_regclass(%s) IS NOT NULL AS present" +_SETTLEMENT_TABLE: Final = '"LiteLLM_BackgroundInteractionSettlement"' +_SETTLEMENT_BY_CALL_QUERY: Final = ( + 'SELECT interaction_id FROM "LiteLLM_BackgroundInteractionSettlement" WHERE create_context->>%s = %s' +) +_OUTAGE_RENAME: Final = ( + 'ALTER TABLE IF EXISTS "LiteLLM_BackgroundInteractionSettlement" ' + 'RENAME TO "LiteLLM_BackgroundInteractionSettlement_outage"' +) +_OUTAGE_RESTORE: Final = ( + 'ALTER TABLE IF EXISTS "LiteLLM_BackgroundInteractionSettlement_outage" ' + 'RENAME TO "LiteLLM_BackgroundInteractionSettlement"' +) + + +@dataclass(frozen=True, slots=True) +class Deployments: + """Config deployments every replica boots with, so no worker ever misses a model added at run time.""" + + in_progress: str + completed_at_once: str + failing_create: str + custom_priced: str + + +@dataclass(frozen=True, slots=True) +class Rig: + gateway: Gateway + upstream: UpstreamSlot + config: Path + models: Deployments + creator: Gateway + settler: Gateway + settler_pid: int + directory: Path + + def environment(self, **poll: str) -> dict[str, str]: + return {"GEMINI_API_BASE": self.upstream.url, "GEMINI_API_KEY": _ENV_KEY, **poll} + + +@pytest.fixture(scope="module") +def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Rig]: + directory: Final = tmp_path_factory.mktemp("settlement") + with gateway_from_environment() as gateway, owned_upstream(directory) as upstream: + models: Final = _register_deployments(upstream.url) + config: Final = _write_config(directory, upstream.url, models) + environment: Final = {"GEMINI_API_BASE": upstream.url, "GEMINI_API_KEY": _ENV_KEY} + with ( + owned_proxy(gateway, directory, {**environment, **_CREATOR_POLL}, config=config, workers=1) as creator, + owned_proxy_process( + gateway, directory, {**environment, **_SETTLER_POLL}, config=config, workers=2 + ) as settler, + ): + yield Rig(gateway, upstream, config, models, creator, settler.gateway, settler.process.pid, directory) + + +def _register_deployments(upstream_url: str) -> Deployments: + suffix: Final = uuid.uuid4().hex[:8] + models: Final = Deployments( + in_progress=f"settle-in-progress-{suffix}", + completed_at_once=f"settle-completed-at-once-{suffix}", + failing_create=f"settle-failing-create-{suffix}", + custom_priced=f"settle-custom-priced-{suffix}", + ) + _register_scenarios(upstream_url, models) + return models + + +def _register_scenarios(upstream_url: str, models: Deployments) -> None: + scripted: Final = { + models.in_progress: _interaction("in_progress", None), + models.completed_at_once: _interaction("completed", _USAGE), + models.failing_create: JsonResponse( + content_type="application/json", body={"error": {"message": "boom"}}, status=500 + ), + models.custom_priced: _interaction("in_progress", None), + } + for name, response in scripted.items(): + register_scenario( + name, + RoutedResponse(content_type="application/x-routed", routes={"POST /v1beta/interactions": response}), + control_url=upstream_url, + ) + + +def _write_config(directory: Path, upstream_url: str, models: Deployments) -> Path: + base: Final = JSON_OBJECT.validate_python(yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())) + custom_pricing: Final = {"input_cost_per_token": _CUSTOM_INPUT_RATE, "output_cost_per_token": _CUSTOM_OUTPUT_RATE} + model_list: Final = [ + { + "model_name": name, + "litellm_params": { + "model": _MODEL, + "api_base": f"{upstream_url}/{name}", + "api_key": _DEPLOYMENT_KEY, + **(custom_pricing if name == models.custom_priced else {}), + }, + } + for name in (models.in_progress, models.completed_at_once, models.failing_create, models.custom_priced) + ] + path: Final = directory / "settlement_config.yaml" + path.write_text(yaml.safe_dump({**base, "model_list": model_list})) + return path + + +def _interaction(status: str, usage: dict[str, JsonValue] | None, http_status: int = 200) -> JsonResponse: + return JsonResponse( + content_type="application/json", + body={ + "id": "$UNIQUE_ID", + "object": "interaction", + "model": "gemini-3.8-flash", + "status": status, + "steps": [], + "usage": usage, + }, + status=http_status, + ) + + +def _completed() -> InteractionState: + return InteractionState(status="completed", usage=_USAGE) + + +def _create( + replica: Gateway, + model: str, + key: str, + *, + path: str = "/v1beta/interactions", + background: bool = True, + text: str | None = None, +) -> str: + response: Final = replica.request( + "POST", + path, + {"model": model, "input": text or f"settle {uuid.uuid4().hex}", "background": background}, + key=key, + ) + assert response.status_code == 200, response.text + return string_value(JSON_OBJECT.validate_json(response.content)["id"]) + + +def _state(rig: Rig, interaction_id: str, state: InteractionState) -> None: + set_interaction_state(rig.upstream.url, interaction_id, state) + + +def _delete(replica: Gateway, interaction_id: str, key: str, *, path: str = "/v1beta/interactions") -> httpx.Response: + return replica.request("DELETE", f"{path}/{interaction_id}", key=key) + + +def _delete_ok(replica: Gateway, interaction_id: str, key: str) -> None: + deleted: Final = _delete(replica, interaction_id, key) + assert deleted.status_code == 200, deleted.text + + +def _delete_concurrently(replica: Gateway, interaction_ids: Sequence[str], key: str) -> tuple[int, ...]: + def status(interaction_id: str) -> int: + return _delete(replica, interaction_id, key).status_code + + with ThreadPoolExecutor(max_workers=8) as pool: + return tuple(pool.map(status, interaction_ids)) + + +def _assert_unclaimed(interaction_id: str) -> None: + row: Final = _settlement(interaction_id) + assert row is not None and row["claimed"] is False and row["outcome"] is None, row + + +def _spend_rows(request_id: str) -> list[dict[str, JsonValue]]: + return read_rows(_SPEND_QUERY, (request_id,)) + + +def _settlement(interaction_id: str) -> dict[str, JsonValue] | None: + rows: Final = read_rows(_SETTLEMENT_QUERY, (interaction_id,)) + return rows[0] if rows else None + + +def _settlement_table_present() -> bool: + return read_rows(_SETTLEMENT_TABLE_PRESENT_QUERY, (_SETTLEMENT_TABLE,))[0]["present"] is True + + +def _settlement_if_stored(interaction_id: str) -> dict[str, JsonValue] | None: + return _settlement(interaction_id) if _settlement_table_present() else None + + +def _settlements_by_call_if_stored(call_id: str) -> list[dict[str, JsonValue]]: + return read_rows(_SETTLEMENT_BY_CALL_QUERY, ("litellm_call_id", call_id)) if _settlement_table_present() else [] + + +def _await_spend_row(interaction_id: str, seconds: float = 30) -> dict[str, JsonValue]: + return eventually(lambda: _spend_rows(interaction_id), lambda rows: len(rows) == 1, seconds=seconds)[0] + + +def _await_outcome(interaction_id: str, outcome: str, seconds: float = 30) -> dict[str, JsonValue]: + row: Final = eventually( + lambda: _settlement(interaction_id), + lambda value: value is not None and value["outcome"] == outcome, + seconds=seconds, + ) + assert row is not None + return row + + +def _model_info(replica: Gateway, model: str) -> Mapping[str, JsonValue]: + entries: Final = replica.get("/model/info")["data"] + assert isinstance(entries, list), entries + return object_value( + next(object_value(entry)["model_info"] for entry in entries if object_value(entry)["model_name"] == model) + ) + + +def _rates(replica: Gateway, model: str) -> tuple[float, float]: + info: Final = _model_info(replica, model) + input_rate: Final = info["input_cost_per_token"] + output_rate: Final = info["output_cost_per_token"] + assert isinstance(input_rate, float) and isinstance(output_rate, float), info + return input_rate, output_rate + + +def _reservation_pin(replica: Gateway, model: str) -> float: + """What one background create estimates before its usage is known: the output tokens the estimator assumes, + at the deployment's output rate, with the prompt's few input tokens left as slack. A key budget below that + is filled by the first create's reservation, so the next create is refused until a settlement releases it.""" + info: Final = _model_info(replica, model) + max_output: Final = info["max_output_tokens"] + output_rate: Final = info["output_cost_per_token"] + assert isinstance(max_output, int) and isinstance(output_rate, float), info + return min(max_output, DEFAULT_MAX_OUTPUT_TOKENS_FALLBACK) * output_rate + + +def _assert_billed(row: Mapping[str, JsonValue], rates: tuple[float, float]) -> float: + expected: Final = _INPUT_TOKENS * rates[0] + _OUTPUT_TOKENS * rates[1] + spend: Final = row["spend"] + assert isinstance(spend, float) and math.isclose(spend, expected, rel_tol=1e-9), (row, expected) + assert row["call_type"] == "acreate_interaction", row + assert row["status"] == "success", row + assert row["prompt_tokens"] == _INPUT_TOKENS and row["completion_tokens"] == _OUTPUT_TOKENS, row + return spend + + +def _key_spend(replica: Gateway, key: str) -> float: + spend: Final = object_value(replica.get("/key/info", {"key": key})["info"])["spend"] + assert isinstance(spend, float | int), spend + return float(spend) + + +def _await_key_spend(replica: Gateway, key: str, expected: float) -> None: + eventually(lambda: _key_spend(replica, key), lambda spend: math.isclose(spend, expected, rel_tol=1e-9), seconds=30) + + +def _drain(rig: Rig) -> list[JsonValue]: + observed: Final = httpx.get(f"{rig.upstream.url}/__observations", trust_env=False, timeout=15) + observed.raise_for_status() + requests: Final = JSON_OBJECT.validate_json(observed.content)["requests"] + assert isinstance(requests, list), requests + return requests + + +def _calls(rig: Rig, interaction_id: str) -> tuple[tuple[str, str], ...]: + suffix: Final = f"/v1beta/interactions/{interaction_id}" + return tuple( + (string_value(object_value(entry)["method"]), string_value(object_value(entry)["api_key"])) + for entry in _drain(rig) + if string_value(object_value(entry)["path"]).endswith(suffix) + ) + + +def _claimer_pid(row: Mapping[str, JsonValue]) -> int: + claimed_by: Final = string_value(row["claimed_by"]) + host, _, pid = claimed_by.rpartition(":") + assert host == socket.gethostname(), claimed_by + return int(pid) + + +def _booted_after(pid: int, moment: float) -> bool: + try: + return psutil.Process(pid).create_time() > moment + except psutil.NoSuchProcess: + return False + + +def _worker_pids(root_pid: int) -> frozenset[int]: + return frozenset( + process.pid for process in group_members(root_pid) if process.pid != root_pid and _is_spawned_worker(process) + ) + + +def _is_spawned_worker(process: psutil.Process) -> bool: + try: + return process.name().lower().startswith("python") and "resource_tracker" not in " ".join(process.cmdline()) + except psutil.Error: + return False + + +def _readiness(replica: Gateway) -> int: + try: + return replica.request("GET", "/health/readiness").status_code + except httpx.TransportError: + return 0 + + +def test_creator_poll_bills_a_completed_background_interaction_once(rig: Rig) -> None: + with rig.settler.scenario() as scenario: + model: Final = rig.models.in_progress + key: Final = scenario.key() + created: Final = _create(rig.settler, model, key) + _state(rig, created, _completed()) + spend: Final = _assert_billed(_await_spend_row(created), _rates(rig.settler, model)) + _await_key_spend(rig.settler, key, spend) + assert len(_spend_rows(created)) == 1 + + +def test_creator_poll_records_its_settlement_durably(rig: Rig) -> None: + with rig.settler.scenario() as scenario: + model: Final = rig.models.in_progress + created: Final = _create(rig.settler, model, scenario.key()) + _state(rig, created, _completed()) + _await_spend_row(created) + row: Final = _await_outcome(created, "billed") + assert row["claimed"] is True and row["settled"] is True, row + assert row["create_context"] == {}, row + _claimer_pid(row) + + +@pytest.mark.parametrize("path", ["/v1beta/interactions", "/interactions"]) +def test_delete_on_another_replica_bills_the_creators_interaction_once(rig: Rig, path: str) -> None: + with rig.creator.scenario() as scenario: + model: Final = rig.models.in_progress + key: Final = scenario.key() + created: Final = _create(rig.creator, model, key, path=path) + _state(rig, created, _completed()) + _drain(rig) + deleted: Final = _delete(rig.settler, created, key, path=path) + assert deleted.status_code == 200, deleted.text + spend: Final = _assert_billed(_await_spend_row(created), _rates(rig.creator, model)) + row: Final = _await_outcome(created, "billed") + assert _claimer_pid(row) in _worker_pids(rig.settler_pid), row + assert _calls(rig, created) == (("GET", _ENV_KEY), ("DELETE", _ENV_KEY)) + _await_key_spend(rig.creator, key, spend) + assert len(_spend_rows(created)) == 1 + + +def test_delete_of_a_failed_interaction_releases_without_a_spend_row(rig: Rig) -> None: + with rig.creator.scenario() as scenario: + model: Final = rig.models.in_progress + key: Final = scenario.key() + created: Final = _create(rig.creator, model, key) + _state(rig, created, InteractionState(status="failed", usage=None)) + deleted: Final = _delete(rig.settler, created, key) + assert deleted.status_code == 200, deleted.text + _await_outcome(created, "released") + assert _spend_rows(created) == [] + assert _key_spend(rig.creator, key) == 0 + + +def test_delete_of_a_requires_action_interaction_bills_its_usage(rig: Rig) -> None: + with rig.creator.scenario() as scenario: + model: Final = rig.models.in_progress + key: Final = scenario.key() + created: Final = _create(rig.creator, model, key) + _state(rig, created, InteractionState(status="requires_action", usage=_USAGE)) + deleted: Final = _delete(rig.settler, created, key) + assert deleted.status_code == 200, deleted.text + _assert_billed(_await_spend_row(created), _rates(rig.creator, model)) + _await_outcome(created, "billed") + + +def test_a_replica_booting_later_resumes_and_bills_unclaimed_interactions(rig: Rig) -> None: + with rig.creator.scenario() as scenario: + model: Final = rig.models.in_progress + key: Final = scenario.key() + created_after: Final = time.time() + created: Final = tuple(_create(rig.creator, model, key) for _ in range(3)) + for item in created: + _state(rig, item, _completed()) + rates: Final = _rates(rig.creator, model) + with owned_proxy_process( + rig.gateway, rig.directory, rig.environment(**_RESUMER_POLL), config=rig.config, workers=2 + ) as resumer: + pids: Final = _worker_pids(resumer.process.pid) + assert len(pids) == 2, pids + for item in created: + _assert_billed(_await_spend_row(item, seconds=90), rates) + claimer: Final = _claimer_pid(_await_outcome(item, "billed")) + assert claimer in pids or _booted_after(claimer, created_after), (claimer, pids) + for item in created: + assert len(_spend_rows(item)) == 1 + + +def test_deletes_on_the_creating_proxy_bill_each_interaction_once(rig: Rig) -> None: + with rig.settler.scenario() as scenario: + model: Final = rig.models.in_progress + key: Final = scenario.key() + created: Final = tuple(_create(rig.settler, model, key) for _ in range(8)) + for item in created: + _state(rig, item, _completed()) + assert _delete_concurrently(rig.settler, created, key) == (200,) * 8 + rates: Final = _rates(rig.settler, model) + for item in created: + _assert_billed(_await_spend_row(item), rates) + _await_outcome(item, "billed") + _await_key_spend(rig.settler, key, 8 * (_INPUT_TOKENS * rates[0] + _OUTPUT_TOKENS * rates[1])) + for item in created: + assert len(_spend_rows(item)) == 1 + + +def test_custom_deployment_pricing_bills_at_the_deployment_rate_on_another_replica(rig: Rig) -> None: + with rig.creator.scenario() as scenario: + model: Final = rig.models.custom_priced + key: Final = scenario.key() + created: Final = _create(rig.creator, model, key) + _state(rig, created, _completed()) + deleted: Final = _delete(rig.settler, created, key) + assert deleted.status_code == 200, deleted.text + _assert_billed(_await_spend_row(created), (_CUSTOM_INPUT_RATE, _CUSTOM_OUTPUT_RATE)) + _await_outcome(created, "billed") + + +def test_cancel_then_delete_on_another_replica_releases_without_a_spend_row(rig: Rig) -> None: + with rig.creator.scenario() as scenario: + model: Final = rig.models.in_progress + key: Final = scenario.key() + created: Final = _create(rig.creator, model, key) + _state(rig, created, InteractionState(status="in_progress")) + cancelled: Final = rig.settler.request("POST", f"/v1beta/interactions/{created}/cancel", {}, key=key) + assert cancelled.status_code == 200, cancelled.text + before_delete: Final = _settlement(created) + assert before_delete is not None and before_delete["claimed"] is False, before_delete + deleted: Final = _delete(rig.settler, created, key) + assert deleted.status_code == 200, deleted.text + _await_outcome(created, "released") + assert _spend_rows(created) == [] + + +def test_delete_fails_closed_when_the_settling_replica_cannot_fetch(rig: Rig) -> None: + with rig.creator.scenario() as scenario: + model: Final = rig.models.in_progress + key: Final = scenario.key() + created: Final = _create(rig.creator, model, key) + _state(rig, created, InteractionState(status="completed", usage=_USAGE, get_status=500)) + _drain(rig) + refused: Final = _delete(rig.settler, created, key) + assert refused.status_code >= 500, refused.text + assert "Scripted interaction fetch failure" in refused.text, refused.text + assert _calls(rig, created) == (("GET", _ENV_KEY),) + _assert_unclaimed(created) + assert _spend_rows(created) == [] + _state(rig, created, _completed()) + deleted: Final = _delete(rig.settler, created, key) + assert deleted.status_code == 200, deleted.text + _assert_billed(_await_spend_row(created), _rates(rig.creator, model)) + _await_outcome(created, "billed") + + +def test_delete_of_an_interaction_the_vendor_purged_sends_no_delete_and_keeps_the_row(rig: Rig) -> None: + with rig.creator.scenario() as scenario: + model: Final = rig.models.in_progress + key: Final = scenario.key() + created: Final = _create(rig.creator, model, key) + clear_interaction_state(rig.upstream.url, created) + _drain(rig) + deleted: Final = _delete(rig.settler, created, key) + assert deleted.status_code == 404, deleted.text + assert _calls(rig, created) == (("GET", _ENV_KEY),) + row: Final = _settlement(created) + assert row is not None and row["claimed"] is False, row + assert _spend_rows(created) == [] + + +def test_reading_an_interaction_never_bills_it(rig: Rig) -> None: + with rig.creator.scenario() as scenario: + model: Final = rig.models.in_progress + key: Final = scenario.key() + created: Final = _create(rig.creator, model, key) + _state(rig, created, InteractionState(status="in_progress")) + read_ids: Final = tuple(str(uuid.uuid4()) for _ in range(2)) + first: Final = rig.settler.request( + "GET", f"/v1beta/interactions/{created}", key=key, headers={"x-litellm-call-id": read_ids[0]} + ) + assert first.status_code == 200 and JSON_OBJECT.validate_json(first.content)["status"] == "in_progress", ( + first.text + ) + _state(rig, created, _completed()) + second: Final = rig.settler.request( + "GET", f"/v1beta/interactions/{created}", key=key, headers={"x-litellm-call-id": read_ids[1]} + ) + assert second.status_code == 200 and JSON_OBJECT.validate_json(second.content)["usage"] == _USAGE, second.text + assert _key_spend(rig.creator, key) == 0 + assert _spend_rows(created) == [] + deleted: Final = _delete(rig.settler, created, key) + assert deleted.status_code == 200, deleted.text + spend: Final = _assert_billed(_await_spend_row(created), _rates(rig.creator, model)) + _await_key_spend(rig.creator, key, spend) + for read_id in read_ids: + assert all(row["spend"] == 0 for row in _spend_rows(read_id)), _spend_rows(read_id) + + +@pytest.mark.parametrize( + "interaction_id", + [f"missing-{uuid.uuid4().hex}", "x" * 5000, "a.b:c", "%2F..%2Fup"], + ids=["unknown", "five-kilobytes", "punctuation", "encoded-traversal"], +) +def test_delete_of_an_odd_or_unknown_id_is_refused_and_the_proxy_keeps_serving(rig: Rig, interaction_id: str) -> None: + with rig.settler.scenario() as scenario: + key: Final = scenario.key() + deleted: Final = rig.settler.request("DELETE", f"/v1beta/interactions/{interaction_id}", key=key) + assert 400 <= deleted.status_code < 500, deleted.text + assert _readiness(rig.settler) == 200 + assert _key_spend(rig.settler, key) == 0 + + +def test_a_missing_settlement_table_leaves_in_process_billing_intact(rig: Rig) -> None: + write_rows(_OUTAGE_RENAME, ()) + try: + with rig.settler.scenario() as scenario: + model: Final = rig.models.in_progress + key: Final = scenario.key() + created: Final = _create(rig.settler, model, key) + _state(rig, created, _completed()) + spend: Final = _assert_billed(_await_spend_row(created), _rates(rig.settler, model)) + _await_key_spend(rig.settler, key, spend) + deleted: Final = _delete(rig.creator, created, key) + assert deleted.status_code == 200, deleted.text + assert len(_spend_rows(created)) == 1 + finally: + write_rows(_OUTAGE_RESTORE, ()) + + +def test_a_failed_create_registers_nothing(rig: Rig) -> None: + with rig.creator.scenario() as scenario: + model: Final = rig.models.failing_create + key: Final = scenario.key() + call_id: Final = str(uuid.uuid4()) + response: Final = rig.creator.request( + "POST", + "/v1beta/interactions", + {"model": model, "input": f"settle {uuid.uuid4().hex}", "background": True}, + key=key, + headers={"x-litellm-call-id": call_id}, + ) + assert response.status_code >= 500, response.text + assert _key_spend(rig.creator, key) == 0 + assert all(row["spend"] == 0 for row in _spend_rows(call_id)), _spend_rows(call_id) + assert _settlements_by_call_if_stored(call_id) == [] + + +def test_polling_disabled_replica_registers_nothing_and_never_bills(rig: Rig) -> None: + disabled: Final = rig.environment(BACKGROUND_INTERACTION_COST_POLLING_ENABLED="false") + with ( + owned_proxy(rig.gateway, rig.directory, disabled, config=rig.config, workers=1) as quiet, + quiet.scenario() as scenario, + ): + model: Final = rig.models.in_progress + key: Final = scenario.key() + created: Final = _create(quiet, model, key) + _state(rig, created, _completed()) + deleted: Final = _delete(rig.settler, created, key) + assert deleted.status_code == 200, deleted.text + assert _settlement_if_stored(created) is None + assert _key_spend(quiet, key) == 0 + assert _spend_rows(created) == [] + + +@pytest.mark.parametrize("background", [False, True], ids=["synchronous", "background"]) +def test_a_create_that_completes_at_once_is_billed_by_the_create_alone(rig: Rig, background: bool) -> None: + with rig.creator.scenario() as scenario: + model: Final = rig.models.completed_at_once + key: Final = scenario.key() + created: Final = _create(rig.creator, model, key, background=background) + spend: Final = _assert_billed(_await_spend_row(created), _rates(rig.creator, model)) + _await_key_spend(rig.creator, key, spend) + assert _settlement_if_stored(created) is None + _state(rig, created, _completed()) + deleted: Final = _delete(rig.settler, created, key) + assert deleted.status_code == 200, deleted.text + _await_key_spend(rig.creator, key, spend) + assert len(_spend_rows(created)) == 1 + + +def test_identical_creates_settle_as_separate_interactions(rig: Rig) -> None: + with rig.creator.scenario() as scenario: + model: Final = rig.models.in_progress + key: Final = scenario.key() + text: Final = f"settle {uuid.uuid4().hex}" + created: Final = tuple(_create(rig.creator, model, key, text=text) for _ in range(3)) + assert len({item for item in created}) == 3, created + for item in created: + _state(rig, item, _completed()) + _delete_ok(rig.settler, item, key) + rates: Final = _rates(rig.creator, model) + for item in created: + _assert_billed(_await_spend_row(item), rates) + _await_outcome(item, "billed") + _await_key_spend(rig.creator, key, 3 * (_INPUT_TOKENS * rates[0] + _OUTPUT_TOKENS * rates[1])) + + +def test_settlement_on_another_replica_releases_the_creators_budget_reservation(rig: Rig) -> None: + with rig.creator.scenario() as scenario: + model: Final = rig.models.in_progress + key: Final = scenario.key(max_budget=0.5 * _reservation_pin(rig.creator, model)) + janitor: Final = scenario.key() + first: Final = _create(rig.creator, model, key) + pinned: Final = rig.creator.request( + "POST", "/v1beta/interactions", {"model": model, "input": "settle pinned", "background": True}, key=key + ) + assert pinned.status_code == 422 and pinned.json()["error"]["type"] == "budget_exceeded", pinned.text + _state(rig, first, _completed()) + still_pinned: Final = _delete(rig.settler, first, key) + assert still_pinned.status_code == 422 and still_pinned.json()["error"]["type"] == "budget_exceeded", ( + still_pinned.text + ) + deleted: Final = _delete(rig.settler, first, janitor) + assert deleted.status_code == 200, deleted.text + spend: Final = _assert_billed(_await_spend_row(first), _rates(rig.creator, model)) + _await_key_spend(rig.creator, key, spend) + released: Final = eventually( + lambda: ( + rig.creator.request( + "POST", + "/v1beta/interactions", + {"model": model, "input": "settle released", "background": True}, + key=key, + ).status_code + ), + lambda status: status == 200, + seconds=20, + return_last_on_timeout=True, + ) + assert released == 200 + + +def test_a_poll_that_never_sees_a_terminal_status_records_unsettled_and_releases(rig: Rig) -> None: + with rig.settler.scenario() as scenario: + model: Final = rig.models.in_progress + key: Final = scenario.key() + created: Final = _create(rig.settler, model, key) + _state(rig, created, InteractionState(status="in_progress")) + row: Final = _await_outcome(created, "unsettled", seconds=40) + assert row["create_context"] == {}, row + assert _spend_rows(created) == [] + assert _key_spend(rig.settler, key) == 0 + deleted: Final = _delete(rig.creator, created, key) + assert deleted.status_code == 200, deleted.text + assert _spend_rows(created) == [] + + +def test_an_upstream_outage_fails_deletes_closed_and_every_interaction_bills_once_after_recovery(rig: Rig) -> None: + with rig.creator.scenario() as scenario: + model: Final = rig.models.in_progress + key: Final = scenario.key() + created: Final = tuple(_create(rig.creator, model, key) for _ in range(16)) + rates: Final = _rates(rig.creator, model) + rig.upstream.stop() + try: + refused: Final = _delete_concurrently(rig.settler, created, key) + assert all(status >= 500 for status in refused), refused + for item in created: + _assert_unclaimed(item) + assert _readiness(rig.creator) == 200 and _readiness(rig.settler) == 200 + finally: + rig.upstream.start() + _register_scenarios(rig.upstream.url, rig.models) + for item in created: + _state(rig, item, _completed()) + assert _delete_concurrently(rig.settler, created, key) == (200,) * 16 + for item in created: + _assert_billed(_await_spend_row(item), rates) + _await_outcome(item, "billed") + _await_key_spend(rig.creator, key, 16 * (_INPUT_TOKENS * rates[0] + _OUTPUT_TOKENS * rates[1])) + for item in created: + assert len(_spend_rows(item)) == 1 + + +def test_killed_workers_leave_their_polls_to_the_respawned_workers(rig: Rig) -> None: + with ( + owned_proxy_process( + rig.gateway, rig.directory, rig.environment(**_RESUMER_POLL), config=rig.config, workers=2 + ) as resumer, + resumer.gateway.scenario() as scenario, + ): + model: Final = rig.models.in_progress + key: Final = scenario.key() + created: Final = tuple(_create(resumer.gateway, model, key) for _ in range(16)) + for item in created: + _state(rig, item, InteractionState(status="in_progress")) + rates: Final = _rates(resumer.gateway, model) + killed: Final = _worker_pids(resumer.process.pid) + assert len(killed) == 2, killed + victims: Final = tuple(psutil.Process(pid) for pid in killed) + for victim in victims: + victim.kill() + psutil.wait_procs(victims, timeout=15) + for item in created: + _state(rig, item, _completed()) + for item in created: + _assert_billed(_await_spend_row(item, seconds=150), rates) + assert _claimer_pid(_await_outcome(item, "billed")) not in killed + assert eventually(lambda: _readiness(resumer.gateway), lambda status: status == 200, seconds=60) == 200 + for item in created: + assert len(_spend_rows(item)) == 1 + + +def test_concurrent_deletes_on_a_slow_upstream_settle_exactly_once(rig: Rig) -> None: + with rig.creator.scenario() as scenario: + model: Final = rig.models.in_progress + key: Final = scenario.key() + created: Final = _create(rig.creator, model, key) + _state(rig, created, InteractionState(status="completed", usage=_USAGE, delay_seconds=1.5)) + statuses: Final = _delete_concurrently(rig.settler, (created, created), key) + assert sorted(statuses) == [200, 404], statuses + spend: Final = _assert_billed(_await_spend_row(created), _rates(rig.creator, model)) + _await_outcome(created, "billed") + _await_key_spend(rig.creator, key, spend) + assert len(_spend_rows(created)) == 1 diff --git a/tests/unit/interactions/test_background_cost_polling.py b/tests/unit/interactions/test_background_cost_polling.py index 97f09de1b52..9dc71a4fa1b 100644 --- a/tests/unit/interactions/test_background_cost_polling.py +++ b/tests/unit/interactions/test_background_cost_polling.py @@ -1,18 +1,25 @@ import asyncio import time +from datetime import datetime, timezone from itertools import islice from typing import Optional import pytest from litellm.interactions.background_cost_polling import ( - _SETTLED_KEY, + _create_context, _poll_intervals, + _rebuild_logging_obj, BackgroundInteractionPollContext, + InMemoryBackgroundSettlementStore, maybe_schedule_background_interaction_cost_polling, maybe_settle_background_interaction_before_delete, + PendingBackgroundInteraction, poll_and_log_background_interaction_cost, + PollSchedule, + resume_unsettled_background_interactions, ) +from litellm.litellm_core_utils.core_helpers import get_litellm_metadata_from_kwargs from litellm.litellm_core_utils.litellm_logging import Logging as LitellmLogging from litellm.types.interactions import InteractionsAPIResponse @@ -63,7 +70,11 @@ async def _raise_on_billing(result: InteractionsAPIResponse) -> None: raise RuntimeError("cost calculation failed for a settled background interaction") -def _context(logging_obj: LitellmLogging, timeout_seconds: float = 1.0) -> BackgroundInteractionPollContext: +def _context( + logging_obj: LitellmLogging, + timeout_seconds: float = 1.0, + store: Optional[InMemoryBackgroundSettlementStore] = None, +) -> BackgroundInteractionPollContext: return BackgroundInteractionPollContext( interaction_id="interactions/bg-abc", custom_llm_provider="gemini", @@ -71,6 +82,7 @@ def _context(logging_obj: LitellmLogging, timeout_seconds: float = 1.0) -> Backg initial_interval_seconds=0.001, max_interval_seconds=0.002, timeout_seconds=timeout_seconds, + store=store if store is not None else InMemoryBackgroundSettlementStore(), ) @@ -246,10 +258,11 @@ async def test_poller_retries_after_fetch_error_and_still_bills(): @pytest.mark.asyncio async def test_schedule_creates_poll_task_for_in_progress_create(): logging_obj = _logging_obj() - task = maybe_schedule_background_interaction_cost_polling( + task = await maybe_schedule_background_interaction_cost_polling( response=_response("in_progress", with_usage=False), create_kwargs={"litellm_logging_obj": logging_obj}, custom_llm_provider="gemini", + store=InMemoryBackgroundSettlementStore(), ) assert isinstance(task, asyncio.Task) @@ -258,6 +271,37 @@ async def test_schedule_creates_poll_task_for_in_progress_create(): await task +@pytest.mark.asyncio +async def test_schedule_registers_an_agent_only_create_that_names_no_model(): + logging_obj = LitellmLogging( + model=None, + messages=None, + stream=False, + call_type="acreate_interaction", + start_time=time.time(), + litellm_call_id="bg-agent-call-id", + function_id="bg-agent-fn-id", + ) + logging_obj.update_environment_variables(litellm_params={}, optional_params={}, custom_llm_provider="gemini") + store = InMemoryBackgroundSettlementStore() + + task = await maybe_schedule_background_interaction_cost_polling( + response=_response("in_progress", with_usage=False), + create_kwargs={"litellm_logging_obj": logging_obj}, + custom_llm_provider="gemini", + store=store, + ) + + assert isinstance(task, asyncio.Task) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + pending = await store.pending("interactions/bg-abc") + assert pending is not None + assert pending.create_context.model is None + assert _rebuild_logging_obj(pending.create_context).model is None + + @pytest.mark.asyncio @pytest.mark.parametrize( "response,create_kwargs", @@ -271,21 +315,22 @@ async def test_schedule_skips_non_pollable_results(response, create_kwargs): if create_kwargs.get("litellm_logging_obj") == "placeholder": create_kwargs = {"litellm_logging_obj": _logging_obj()} - task = maybe_schedule_background_interaction_cost_polling( + task = await maybe_schedule_background_interaction_cost_polling( response=response, create_kwargs=create_kwargs, custom_llm_provider="gemini", + store=InMemoryBackgroundSettlementStore(), ) assert task is None -def _register_poll(logging_obj: LitellmLogging, poll_fetch=None) -> asyncio.Task: +def _register_poll(logging_obj: LitellmLogging, poll_fetch=None, store=None) -> asyncio.Task: import litellm.interactions.background_cost_polling as bg if poll_fetch is None: poll_fetch, _ = _fetch_sequence(_response("in_progress", with_usage=False)) - context = _context(logging_obj) + context = _context(logging_obj, store=store) task = asyncio.create_task(poll_and_log_background_interaction_cost(context, fetch_interaction=poll_fetch)) bg._ACTIVE_POLLS[context.interaction_id] = bg._ActiveBackgroundPoll(task=task, context=context) task.add_done_callback(lambda finished: bg._discard_poll(context.interaction_id, finished)) @@ -300,6 +345,7 @@ async def test_delete_settlement_bills_an_interaction_paused_for_a_tool_result() await maybe_settle_background_interaction_before_delete( interaction_id="interactions/bg-abc", + delete_kwargs={}, fetch_interaction=fetch, ) @@ -316,6 +362,7 @@ async def test_delete_settlement_bills_pending_background_interaction(): await maybe_settle_background_interaction_before_delete( interaction_id="interactions/bg-abc", + delete_kwargs={}, fetch_interaction=fetch, ) @@ -334,6 +381,7 @@ async def test_delete_settlement_releases_reservation_when_still_in_progress(): await maybe_settle_background_interaction_before_delete( interaction_id="interactions/bg-abc", + delete_kwargs={}, fetch_interaction=fetch, ) @@ -351,6 +399,7 @@ async def test_delete_settlement_releases_reservation_when_prefetch_fails(): await maybe_settle_background_interaction_before_delete( interaction_id="interactions/bg-abc", + delete_kwargs={}, fetch_interaction=fetch, ) @@ -370,6 +419,7 @@ async def test_delete_settlement_releases_reservation_when_billing_raises(): with pytest.raises(RuntimeError): await maybe_settle_background_interaction_before_delete( interaction_id="interactions/bg-abc", + delete_kwargs={}, fetch_interaction=fetch, ) @@ -383,6 +433,7 @@ async def test_delete_settlement_ignores_interactions_without_pending_poll(): await maybe_settle_background_interaction_before_delete( interaction_id="interactions/never-polled", + delete_kwargs={}, fetch_interaction=fetch, ) @@ -400,6 +451,7 @@ async def test_delete_settlement_noop_after_poll_task_finished(): settle_fetch, settle_calls = _fetch_sequence(_response("completed", with_usage=True)) await maybe_settle_background_interaction_before_delete( interaction_id="interactions/bg-abc", + delete_kwargs={}, fetch_interaction=settle_fetch, ) @@ -409,12 +461,14 @@ async def test_delete_settlement_noop_after_poll_task_finished(): @pytest.mark.asyncio async def test_delete_settlement_does_not_rebill_when_gate_already_claimed(): logging_obj = _logging_obj() - logging_obj.model_call_details[_SETTLED_KEY] = True - task = _register_poll(logging_obj) + store = InMemoryBackgroundSettlementStore() + assert await store.claim("interactions/bg-abc") + task = _register_poll(logging_obj, store=store) fetch, calls = _fetch_sequence(_response("completed", with_usage=True)) await maybe_settle_background_interaction_before_delete( interaction_id="interactions/bg-abc", + delete_kwargs={}, fetch_interaction=fetch, ) @@ -426,10 +480,11 @@ async def test_delete_settlement_does_not_rebill_when_gate_already_claimed(): @pytest.mark.asyncio async def test_poller_exits_without_billing_once_settled_elsewhere(): logging_obj = _logging_obj() - logging_obj.model_call_details[_SETTLED_KEY] = True + store = InMemoryBackgroundSettlementStore() + assert await store.claim("interactions/bg-abc") fetch, calls = _fetch_sequence(_response("completed", with_usage=True)) - await poll_and_log_background_interaction_cost(_context(logging_obj), fetch_interaction=fetch) + await poll_and_log_background_interaction_cost(_context(logging_obj, store=store), fetch_interaction=fetch) assert calls == [] assert logging_obj.model_call_details.get("response_cost") is None @@ -441,10 +496,11 @@ async def test_schedule_respects_kill_switch(monkeypatch): monkeypatch.setattr(module, "BACKGROUND_INTERACTION_COST_POLLING_ENABLED", False) - task = maybe_schedule_background_interaction_cost_polling( + task = await maybe_schedule_background_interaction_cost_polling( response=_response("in_progress", with_usage=False), create_kwargs={"litellm_logging_obj": _logging_obj()}, custom_llm_provider="gemini", + store=InMemoryBackgroundSettlementStore(), ) assert task is None @@ -480,10 +536,11 @@ async def test_schedule_creates_poll_task_for_queued_create(): without a poll task it is never charged at all. """ logging_obj = _logging_obj() - task = maybe_schedule_background_interaction_cost_polling( + task = await maybe_schedule_background_interaction_cost_polling( response=_response("queued", with_usage=False), create_kwargs={"litellm_logging_obj": logging_obj}, custom_llm_provider="gemini", + store=InMemoryBackgroundSettlementStore(), ) assert isinstance(task, asyncio.Task) @@ -543,3 +600,492 @@ async def test_giving_up_on_an_unrecognized_status_says_which_status_it_was(monk assert len(errors) == 1 assert "halted_for_review" in errors[0] + + +KEY_HASH = "0123456789abcdef" * 4 + +FAST_SCHEDULE = PollSchedule(initial_interval_seconds=0.001, max_interval_seconds=0.002, timeout_seconds=1.0) + + +def _capturing_fetch(response: InteractionsAPIResponse): + captured = [] + + async def fetch(context): + captured.append(context) + return response + + return fetch, captured + + +def _create_metadata(**extra) -> dict: + return { + "user_api_key": KEY_HASH, + "user_api_key_team_id": "team-1", + "user_api_key_auth": object(), + **extra, + } + + +async def _create_on_a_replica_that_then_dies(logging_obj: LitellmLogging, store) -> None: + import litellm.interactions.background_cost_polling as bg + + poll_fetch, _ = _fetch_sequence(_response("in_progress", with_usage=False)) + task = await maybe_schedule_background_interaction_cost_polling( + response=_response("in_progress", with_usage=False), + create_kwargs={"litellm_logging_obj": logging_obj}, + custom_llm_provider="gemini", + store=store, + fetch_interaction=poll_fetch, + ) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + await asyncio.sleep(0) + assert "interactions/bg-abc" not in bg._ACTIVE_POLLS + + +@pytest.mark.asyncio +async def test_delete_on_another_replica_bills_the_create_from_the_store(): + """ + The regression: the replica that served the create owns the poll task, so + a delete served by any other replica used to find nothing to settle and + the work went unbilled. The store carries the create's attribution, never + its auth object, to whichever replica settles. + """ + store = InMemoryBackgroundSettlementStore() + logging_obj = _logging_obj(litellm_params={"metadata": _create_metadata()}) + await _create_on_a_replica_that_then_dies(logging_obj, store) + fetch, captured = _capturing_fetch(_response("completed", with_usage=True)) + + outcome = await maybe_settle_background_interaction_before_delete( + interaction_id="interactions/bg-abc", + delete_kwargs={}, + fetch_interaction=fetch, + store=store, + ) + + assert outcome == "billed" + settled = captured[0].logging_obj + assert settled is not logging_obj + assert settled.model_call_details["response_cost"] > 0 + payload_metadata = settled.model_call_details["standard_logging_object"]["metadata"] + assert payload_metadata["user_api_key_hash"] == KEY_HASH + assert payload_metadata["user_api_key_team_id"] == "team-1" + assert "user_api_key_auth" not in get_litellm_metadata_from_kwargs(kwargs=settled.model_call_details) + + +@pytest.mark.asyncio +async def test_delete_on_another_replica_releases_the_create_reservation(): + store = InMemoryBackgroundSettlementStore() + logging_obj = _logging_obj( + litellm_params={"metadata": _create_metadata(user_api_key_budget_reservation=_reservation())} + ) + await _create_on_a_replica_that_then_dies(logging_obj, store) + fetch, captured = _capturing_fetch(_response("in_progress", with_usage=False)) + + outcome = await maybe_settle_background_interaction_before_delete( + interaction_id="interactions/bg-abc", + delete_kwargs={}, + fetch_interaction=fetch, + store=store, + ) + + assert outcome == "released" + settled_metadata = get_litellm_metadata_from_kwargs(kwargs=captured[0].logging_obj.model_call_details) + assert settled_metadata["user_api_key_budget_reservation"]["finalized"] is True + + +@pytest.mark.asyncio +async def test_delete_on_another_replica_fails_when_it_cannot_fetch_and_leaves_the_bill_to_the_creating_poll(): + """ + The settling replica fetches with the delete's credentials, never the + create's, so a fetch it cannot make (a key only the deployment carries) + says nothing about the interaction. Deleting anyway would strand the bill + behind a deleted interaction, so the delete fails with the fetch's error + and the poll on the creating replica still owns the bill. + """ + store = InMemoryBackgroundSettlementStore() + logging_obj = _logging_obj(litellm_params={"metadata": _create_metadata()}) + await _create_on_a_replica_that_then_dies(logging_obj, store) + fetch, _ = _fetch_sequence(RuntimeError("Google API key is required")) + + with pytest.raises(RuntimeError, match="Google API key is required"): + await maybe_settle_background_interaction_before_delete( + interaction_id="interactions/bg-abc", delete_kwargs={}, fetch_interaction=fetch, store=store + ) + + assert await store.is_claimed("interactions/bg-abc") is False + poll_fetch, _ = _fetch_sequence(_response("completed", with_usage=True)) + await asyncio.wait_for(_register_poll(logging_obj, poll_fetch=poll_fetch, store=store), timeout=5) + assert logging_obj.model_call_details["response_cost"] > 0 + + +@pytest.mark.asyncio +async def test_delete_settles_once_however_many_replicas_try(): + store = InMemoryBackgroundSettlementStore() + await _create_on_a_replica_that_then_dies(_logging_obj(litellm_params={"metadata": _create_metadata()}), store) + first_fetch, first_calls = _capturing_fetch(_response("completed", with_usage=True)) + second_fetch, second_calls = _capturing_fetch(_response("completed", with_usage=True)) + + first = await maybe_settle_background_interaction_before_delete( + interaction_id="interactions/bg-abc", delete_kwargs={}, fetch_interaction=first_fetch, store=store + ) + second = await maybe_settle_background_interaction_before_delete( + interaction_id="interactions/bg-abc", delete_kwargs={}, fetch_interaction=second_fetch, store=store + ) + + assert (first, second) == ("billed", None) + assert len(first_calls) == 1 + assert second_calls == [] + + +def test_create_context_carries_no_request_headers(): + logging_obj = _logging_obj( + litellm_params={ + "metadata": _create_metadata( + requester_custom_headers={"x-api-key": "sk-customer-secret"}, + proxy_server_request={"headers": {"x-api-key": "sk-customer-secret"}}, + ) + } + ) + + carried = _create_context(logging_obj, "gemini").metadata + + assert carried["user_api_key_team_id"] == "team-1" + assert "requester_custom_headers" not in carried + assert "proxy_server_request" not in carried + + +@pytest.mark.asyncio +async def test_restart_resumes_only_the_rows_no_replica_claimed(): + store = InMemoryBackgroundSettlementStore() + create_context = _create_context(_logging_obj(litellm_params={"metadata": _create_metadata()}), "gemini") + for interaction_id in ("interactions/bg-orphaned", "interactions/bg-settled"): + await store.register( + PendingBackgroundInteraction( + interaction_id=interaction_id, + custom_llm_provider="gemini", + create_context=create_context, + created_at=datetime.now(timezone.utc), + ) + ) + assert await store.claim("interactions/bg-settled") + fetch, captured = _capturing_fetch(_response("completed", with_usage=True)) + + resumed = await resume_unsettled_background_interactions(store, fetch, schedule=FAST_SCHEDULE) + + assert len(resumed) == 1 + assert await asyncio.wait_for(resumed[0], timeout=5) == "billed" + assert [context.interaction_id for context in captured] == ["interactions/bg-orphaned"] + assert captured[0].logging_obj.model_call_details["response_cost"] > 0 + assert await store.is_claimed("interactions/bg-orphaned") + + +class _ClaimAnswersOnlyAfterTheLastFetch: + def __init__(self): + self.store = InMemoryBackgroundSettlementStore() + self.fetches = 0 + self.fetches_at_last_claim = -1 + + async def fetch(self, context): + self.fetches += 1 + return _response("completed", with_usage=True) + + async def register(self, pending): + await self.store.register(pending) + + async def pending(self, interaction_id): + return await self.store.pending(interaction_id) + + async def is_claimed(self, interaction_id): + return await self.store.is_claimed(interaction_id) + + async def claim(self, interaction_id): + if self.fetches != self.fetches_at_last_claim: + self.fetches_at_last_claim = self.fetches + raise RuntimeError("database unavailable") + return await self.store.claim(interaction_id) + + async def record_outcome(self, interaction_id, outcome): + return None + + async def unclaimed(self): + return await self.store.unclaimed() + + +@pytest.mark.asyncio +async def test_poller_bills_the_completed_response_it_saw_when_the_claim_only_answers_at_the_deadline(): + logging_obj = _logging_obj() + store = _ClaimAnswersOnlyAfterTheLastFetch() + + outcome = await poll_and_log_background_interaction_cost( + _context(logging_obj, timeout_seconds=0.01, store=store), + fetch_interaction=store.fetch, + ) + + assert store.fetches >= 2 + assert outcome == "billed" + assert logging_obj.model_call_details["response_cost"] > 0 + + +class _DownStore: + async def register(self, pending): + raise RuntimeError("database unavailable") + + async def pending(self, interaction_id): + raise RuntimeError("database unavailable") + + async def is_claimed(self, interaction_id): + raise RuntimeError("database unavailable") + + async def claim(self, interaction_id): + raise RuntimeError("database unavailable") + + async def record_outcome(self, interaction_id, outcome): + raise RuntimeError("database unavailable") + + async def unclaimed(self): + raise RuntimeError("database unavailable") + + +class _RegistersThenRaises: + def __init__(self): + self.store = InMemoryBackgroundSettlementStore() + + async def register(self, pending): + await self.store.register(pending) + raise RuntimeError("connection reset after the row was committed") + + async def pending(self, interaction_id): + return await self.store.pending(interaction_id) + + async def is_claimed(self, interaction_id): + return await self.store.is_claimed(interaction_id) + + async def claim(self, interaction_id): + return await self.store.claim(interaction_id) + + async def record_outcome(self, interaction_id, outcome): + return None + + async def unclaimed(self): + return await self.store.unclaimed() + + +@pytest.mark.asyncio +async def test_create_whose_registration_raised_after_landing_still_claims_the_stored_row(): + """ + A registration that raises after its row committed used to move the poll + to a private in-memory gate, so the creating worker billed while the + stored row stayed unclaimed for another replica's delete or the next boot + to bill again. The row that landed is the gate every settler shares. + """ + store = _RegistersThenRaises() + logging_obj = _logging_obj(litellm_params={"metadata": _create_metadata()}) + poll_fetch, _ = _fetch_sequence(_response("in_progress", with_usage=False)) + task = await maybe_schedule_background_interaction_cost_polling( + response=_response("in_progress", with_usage=False), + create_kwargs={"litellm_logging_obj": logging_obj}, + custom_llm_provider="gemini", + store=store, + fetch_interaction=poll_fetch, + ) + fetch, _ = _capturing_fetch(_response("completed", with_usage=True)) + + outcome = await maybe_settle_background_interaction_before_delete( + interaction_id="interactions/bg-abc", delete_kwargs={}, fetch_interaction=fetch, store=store + ) + + assert outcome == "billed" + assert await store.is_claimed("interactions/bg-abc") + assert await store.unclaimed() == () + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + +@pytest.mark.asyncio +async def test_delete_on_a_worker_that_resumed_the_poll_fails_when_it_cannot_fetch(): + """ + After a restart every worker resumes the unclaimed rows, so none of them + is the creator whose delete may release and delete on a failed fetch. A + resumed worker's delete fails like any other replica's, and its own poll + still bills the interaction once it completes. + """ + store = InMemoryBackgroundSettlementStore() + await store.register( + PendingBackgroundInteraction( + interaction_id="interactions/bg-abc", + custom_llm_provider="gemini", + create_context=_create_context(_logging_obj(litellm_params={"metadata": _create_metadata()}), "gemini"), + created_at=datetime.now(timezone.utc), + ) + ) + responses = [_response("in_progress", with_usage=False)] + + async def poll_fetch(context): + return responses[-1] + + (resumed,) = await resume_unsettled_background_interactions(store, poll_fetch, schedule=FAST_SCHEDULE) + fetch, _ = _fetch_sequence(RuntimeError("Google API key is required")) + + with pytest.raises(RuntimeError, match="Google API key is required"): + await maybe_settle_background_interaction_before_delete( + interaction_id="interactions/bg-abc", delete_kwargs={}, fetch_interaction=fetch, store=store + ) + + assert await store.is_claimed("interactions/bg-abc") is False + responses.append(_response("completed", with_usage=True)) + assert await asyncio.wait_for(resumed, timeout=5) == "billed" + + +class _LandsThenGoesDown: + """Register commits the row and loses its acknowledgement; every read fails until the store recovers.""" + + def __init__(self): + self.store = InMemoryBackgroundSettlementStore() + self.down = True + + async def register(self, pending): + await self.store.register(pending) + raise RuntimeError("connection reset after the row was committed") + + async def pending(self, interaction_id): + self._answer() + return await self.store.pending(interaction_id) + + async def is_claimed(self, interaction_id): + self._answer() + return await self.store.is_claimed(interaction_id) + + async def claim(self, interaction_id): + self._answer() + return await self.store.claim(interaction_id) + + async def record_outcome(self, interaction_id, outcome): + return None + + async def unclaimed(self): + self._answer() + return await self.store.unclaimed() + + def _answer(self): + if self.down: + raise RuntimeError("database unavailable") + + +class _TableLessStore: + """A replica whose database never got the settlement table: writes fail and reads see no rows.""" + + async def register(self, pending): + raise RuntimeError("the settlement table does not exist") + + async def pending(self, interaction_id): + return None + + async def is_claimed(self, interaction_id): + return False + + async def claim(self, interaction_id): + return False + + async def record_outcome(self, interaction_id, outcome): + raise RuntimeError("the settlement table does not exist") + + async def unclaimed(self): + raise RuntimeError("the settlement table does not exist") + + +@pytest.mark.asyncio +async def test_create_whose_registration_and_read_back_both_failed_bills_once_through_the_landed_row(): + """ + A registration that raised and could not be read back used to give the + creator a private in-memory gate, so it billed while the stored row stayed + unclaimed for the next boot to resume and bill again. With the durable + state unknown, the claim waits for the store and settles through the row. + """ + store = _LandsThenGoesDown() + logging_obj = _logging_obj(litellm_params={"metadata": _create_metadata()}) + poll_fetch, _ = _fetch_sequence(_response("in_progress", with_usage=False)) + task = await maybe_schedule_background_interaction_cost_polling( + response=_response("in_progress", with_usage=False), + create_kwargs={"litellm_logging_obj": logging_obj}, + custom_llm_provider="gemini", + store=store, + fetch_interaction=poll_fetch, + ) + fetch, calls = _fetch_sequence(_response("completed", with_usage=True), _response("completed", with_usage=True)) + + while_down = await maybe_settle_background_interaction_before_delete( + interaction_id="interactions/bg-abc", delete_kwargs={}, fetch_interaction=fetch, store=store + ) + store.down = False + recovered = await maybe_settle_background_interaction_before_delete( + interaction_id="interactions/bg-abc", delete_kwargs={}, fetch_interaction=fetch, store=store + ) + + assert (while_down, recovered) == (None, "billed") + assert len(calls) == 2 + assert await store.is_claimed("interactions/bg-abc") + assert await store.unclaimed() == () + assert await resume_unsettled_background_interactions(store, poll_fetch, schedule=FAST_SCHEDULE) == () + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + +@pytest.mark.asyncio +async def test_create_whose_store_never_answers_is_not_billed_through_a_private_gate(): + logging_obj = _logging_obj() + poll_fetch, _ = _fetch_sequence(_response("in_progress", with_usage=False)) + task = await maybe_schedule_background_interaction_cost_polling( + response=_response("in_progress", with_usage=False), + create_kwargs={"litellm_logging_obj": logging_obj}, + custom_llm_provider="gemini", + store=_DownStore(), + fetch_interaction=poll_fetch, + ) + fetch, calls = _fetch_sequence(_response("completed", with_usage=True)) + + outcome = await maybe_settle_background_interaction_before_delete( + interaction_id="interactions/bg-abc", + delete_kwargs={}, + fetch_interaction=fetch, + store=_DownStore(), + ) + + assert outcome is None + assert len(calls) == 1 + assert "response_cost" not in logging_obj.model_call_details + assert not task.done() + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + +@pytest.mark.asyncio +async def test_create_on_a_replica_without_the_settlement_table_still_settles_in_process(): + logging_obj = _logging_obj() + poll_fetch, _ = _fetch_sequence(_response("in_progress", with_usage=False)) + task = await maybe_schedule_background_interaction_cost_polling( + response=_response("in_progress", with_usage=False), + create_kwargs={"litellm_logging_obj": logging_obj}, + custom_llm_provider="gemini", + store=_TableLessStore(), + fetch_interaction=poll_fetch, + ) + fetch, calls = _fetch_sequence(_response("completed", with_usage=True), _response("completed", with_usage=True)) + + outcome = await maybe_settle_background_interaction_before_delete( + interaction_id="interactions/bg-abc", delete_kwargs={}, fetch_interaction=fetch, store=_TableLessStore() + ) + again = await maybe_settle_background_interaction_before_delete( + interaction_id="interactions/bg-abc", delete_kwargs={}, fetch_interaction=fetch, store=_TableLessStore() + ) + + assert (outcome, again) == ("billed", None) + assert len(calls) == 2 + assert logging_obj.model_call_details["response_cost"] > 0 + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task diff --git a/tests/unit/interactions/test_gemini_interactions_transformation.py b/tests/unit/interactions/test_gemini_interactions_transformation.py index 2809c12ae47..5e394e9f218 100644 --- a/tests/unit/interactions/test_gemini_interactions_transformation.py +++ b/tests/unit/interactions/test_gemini_interactions_transformation.py @@ -10,12 +10,13 @@ Covers: from unittest.mock import MagicMock, patch +import httpx import pytest - from litellm.interactions.litellm_responses_transformation.streaming_iterator import ( LiteLLMResponsesInteractionsStreamingIterator, ) +from litellm.llms.gemini.common_utils import GeminiError from litellm.llms.gemini.interactions.transformation import ( GoogleAIStudioInteractionsConfig, ) @@ -464,6 +465,32 @@ class TestInteractionOperationUrls: ) +class TestGetInteractionResponse: + @pytest.mark.parametrize("status_code", [404, 500]) + def test_non_2xx_raises_even_when_the_error_body_is_json( + self, config: GoogleAIStudioInteractionsConfig, status_code: int + ) -> None: + raw_response = httpx.Response( + status_code, + json={"error": {"code": status_code, "message": "boom", "status": "INTERNAL"}}, + request=httpx.Request("GET", "https://generativelanguage.googleapis.com/v1beta/interactions/x"), + ) + with pytest.raises(GeminiError) as raised: + config.transform_get_interaction_response(raw_response=raw_response, logging_obj=MagicMock()) + assert raised.value.status_code == status_code + assert "boom" in str(raised.value) + + def test_2xx_parses_the_interaction(self, config: GoogleAIStudioInteractionsConfig) -> None: + raw_response = httpx.Response( + 200, + json={"id": "interaction-1", "object": "interaction", "status": "completed", "steps": []}, + request=httpx.Request("GET", "https://generativelanguage.googleapis.com/v1beta/interactions/x"), + ) + response = config.transform_get_interaction_response(raw_response=raw_response, logging_obj=MagicMock()) + assert response.id == "interaction-1" + assert response.status == "completed" + + class TestTransformRequestSchemaCoalescing: """Test new-schema request coalescing (Api-Revision: 2026-05-20).""" diff --git a/tests/unit/proxy/spend_tracking/test_background_interaction_settlement.py b/tests/unit/proxy/spend_tracking/test_background_interaction_settlement.py new file mode 100644 index 00000000000..21384279bbb --- /dev/null +++ b/tests/unit/proxy/spend_tracking/test_background_interaction_settlement.py @@ -0,0 +1,298 @@ +import asyncio +import time +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Optional + +import pytest + +import litellm.interactions.background_cost_polling as bg +from litellm.interactions.background_cost_polling import ( + _create_context, + configure_background_settlement_store, + maybe_settle_background_interaction_before_delete, + PendingBackgroundInteraction, + PollSchedule, +) +from litellm.litellm_core_utils.litellm_logging import Logging as LitellmLogging +from litellm.proxy.spend_tracking.background_interaction_settlement import ( + configure_background_interaction_settlement, + install_background_interaction_settlement, + PrismaBackgroundSettlementStore, +) +from litellm.types.interactions import InteractionsAPIResponse + +USAGE_BLOCK = { + "total_tokens": 175, + "total_input_tokens": 100, + "input_tokens_by_modality": [{"modality": "text", "tokens": 100}], + "total_cached_tokens": 0, + "total_output_tokens": 50, + "output_tokens_by_modality": [{"modality": "text", "tokens": 50}], + "total_tool_use_tokens": 0, + "total_thought_tokens": 25, +} + +FAST_SCHEDULE = PollSchedule(initial_interval_seconds=0.001, max_interval_seconds=0.002, timeout_seconds=1.0) + + +@dataclass +class _Row: + interaction_id: str + custom_llm_provider: str + create_context: object + created_at: datetime + claimed_at: Optional[datetime] = None + claimed_by: Optional[str] = None + settled_at: Optional[datetime] = None + outcome: Optional[str] = None + + +class _FakeSettlementTable: + """Just enough of prisma's per-model actions: Json is stored as the data it wraps and read back parsed.""" + + def __init__(self, rows: tuple[_Row, ...] = ()): + self.rows = {row.interaction_id: row for row in rows} + + async def create(self, *, data): + row = _Row( + interaction_id=data["interaction_id"], + custom_llm_provider=data["custom_llm_provider"], + create_context=data["create_context"].data, + created_at=data["created_at"], + ) + self.rows[row.interaction_id] = row + return row + + async def find_unique(self, *, where): + return self.rows.get(where["interaction_id"]) + + async def find_many(self, *, where): + return self._matching(where) + + async def update_many(self, *, data, where): + matched = self._matching(where) + for row in matched: + for column, value in data.items(): + setattr(row, column, getattr(value, "data", value) if column == "create_context" else value) + return len(matched) + + def _matching(self, where) -> list: + return [row for row in self.rows.values() if all(getattr(row, column) == value for column, value in where.items())] + + +def _logging_obj(metadata: Optional[dict] = None) -> LitellmLogging: + logging_obj = LitellmLogging( + model="gemini-2.5-flash", + messages=[], + stream=False, + call_type="acreate_interaction", + start_time=time.time(), + litellm_call_id="bg-settlement-call-id", + function_id="bg-settlement-fn-id", + ) + logging_obj.update_environment_variables( + litellm_params={"metadata": metadata or {"user_api_key": "0123456789abcdef" * 4}}, + optional_params={}, + model="gemini-2.5-flash", + custom_llm_provider="gemini", + input="hi", + ) + return logging_obj + + +def _pending(interaction_id: str) -> PendingBackgroundInteraction: + return PendingBackgroundInteraction( + interaction_id=interaction_id, + custom_llm_provider="gemini", + create_context=_create_context(_logging_obj(), "gemini"), + created_at=datetime.now(timezone.utc), + ) + + +def _stored_row(interaction_id: str, claimed: bool = False, create_context: Optional[object] = None) -> _Row: + return _Row( + interaction_id=interaction_id, + custom_llm_provider="gemini", + create_context=( + create_context + if create_context is not None + else _create_context(_logging_obj(), "gemini").model_dump(mode="json") + ), + created_at=datetime.now(timezone.utc), + claimed_at=datetime.now(timezone.utc) if claimed else None, + claimed_by="replica-a:1" if claimed else None, + ) + + +def _completed(interaction_id: str) -> InteractionsAPIResponse: + return InteractionsAPIResponse( + id=interaction_id, model="gemini-2.5-flash", status="completed", steps=[], usage=dict(USAGE_BLOCK) + ) + + +def _capturing_fetch(): + captured = [] + + async def fetch(context): + captured.append(context) + return _completed(context.interaction_id) + + return fetch, captured + + +@pytest.mark.asyncio +async def test_registered_row_reads_back_as_the_same_pending_interaction(): + table = _FakeSettlementTable() + store = PrismaBackgroundSettlementStore(table=table, claimed_by="replica-a:1") + pending = _pending("interactions/bg-1") + + await store.register(pending) + + assert await store.pending("interactions/bg-1") == pending + assert await store.unclaimed() == (pending,) + + +@pytest.mark.asyncio +async def test_claim_is_won_by_exactly_one_settler(): + table = _FakeSettlementTable() + replica_a = PrismaBackgroundSettlementStore(table=table, claimed_by="replica-a:1") + replica_b = PrismaBackgroundSettlementStore(table=table, claimed_by="replica-b:1") + await replica_a.register(_pending("interactions/bg-1")) + + assert await replica_b.claim("interactions/bg-1") is True + assert await replica_a.claim("interactions/bg-1") is False + assert await replica_a.is_claimed("interactions/bg-1") is True + assert await replica_a.pending("interactions/bg-1") is None + assert table.rows["interactions/bg-1"].claimed_by == "replica-b:1" + + +class _MissingSettlementTable: + """Prisma's per-model actions against a database whose migration for this table was held back.""" + + async def create(self, *, data): + raise self._missing() + + async def find_unique(self, *, where): + raise self._missing() + + async def find_many(self, *, where): + raise self._missing() + + async def update_many(self, *, data, where): + raise self._missing() + + def _missing(self): + from prisma.errors import TableNotFoundError + + return TableNotFoundError( + { + "user_facing_error": { + "error_code": "P2021", + "meta": {"table": "public.LiteLLM_BackgroundInteractionSettlement"}, + "message": "The table does not exist in the current database.", + } + } + ) + + +@pytest.mark.asyncio +async def test_a_missing_table_holds_no_rows_and_takes_no_registration(): + from prisma.errors import TableNotFoundError + + store = PrismaBackgroundSettlementStore(table=_MissingSettlementTable(), claimed_by="replica-a:1") + + with pytest.raises(TableNotFoundError): + await store.register(_pending("interactions/bg-1")) + assert await store.pending("interactions/bg-1") is None + assert await store.is_claimed("interactions/bg-1") is False + assert await store.claim("interactions/bg-1") is False + with pytest.raises(TableNotFoundError): + await store.unclaimed() + + +@pytest.mark.asyncio +async def test_unclaimed_skips_claimed_and_unreadable_rows(): + table = _FakeSettlementTable( + rows=( + _stored_row("interactions/bg-orphaned"), + _stored_row("interactions/bg-settled", claimed=True), + _stored_row("interactions/bg-from-the-future", create_context={"schema": "unknown"}), + ) + ) + store = PrismaBackgroundSettlementStore(table=table, claimed_by="replica-b:1") + + unclaimed = await store.unclaimed() + + assert [row.interaction_id for row in unclaimed] == ["interactions/bg-orphaned"] + + +@pytest.mark.asyncio +async def test_record_outcome_keeps_the_audit_trail_and_drops_the_stored_request_context(): + table = _FakeSettlementTable() + store = PrismaBackgroundSettlementStore(table=table, claimed_by="replica-a:1") + await store.register(_pending("interactions/bg-1")) + assert await store.claim("interactions/bg-1") + assert table.rows["interactions/bg-1"].create_context + + await store.record_outcome("interactions/bg-1", "billed") + + row = table.rows["interactions/bg-1"] + assert row.outcome == "billed" + assert row.settled_at is not None + assert row.claimed_at <= row.settled_at + assert row.create_context == {} + + +@pytest.mark.asyncio +async def test_configure_installs_the_store_and_resumes_the_orphaned_rows(): + table = _FakeSettlementTable( + rows=(_stored_row("interactions/bg-orphaned"), _stored_row("interactions/bg-settled", claimed=True)) + ) + fetch, captured = _capturing_fetch() + previous_store = bg._STORE.store + try: + resumed = await configure_background_interaction_settlement( + table=table, claimed_by="replica-b:1", fetch_interaction=fetch, schedule=FAST_SCHEDULE + ) + + assert len(resumed) == 1 + assert await asyncio.wait_for(resumed[0], timeout=5) == "billed" + assert [context.interaction_id for context in captured] == ["interactions/bg-orphaned"] + assert table.rows["interactions/bg-orphaned"].claimed_by == "replica-b:1" + assert table.rows["interactions/bg-orphaned"].outcome == "billed" + + await table.create( + data={ + "interaction_id": "interactions/bg-created-elsewhere", + "custom_llm_provider": "gemini", + "create_context": _JsonLike(_create_context(_logging_obj(), "gemini").model_dump(mode="json")), + "created_at": datetime.now(timezone.utc), + } + ) + outcome = await maybe_settle_background_interaction_before_delete( + interaction_id="interactions/bg-created-elsewhere", delete_kwargs={}, fetch_interaction=fetch + ) + + assert outcome == "billed" + assert table.rows["interactions/bg-created-elsewhere"].claimed_by == "replica-b:1" + finally: + configure_background_settlement_store(previous_store) + + +class _PrismaClientWithoutSettlementTable: + pass + + +@pytest.mark.asyncio +async def test_install_keeps_booting_when_the_settlement_table_is_unreachable(): + previous_store = bg._STORE.store + + await install_background_interaction_settlement(_PrismaClientWithoutSettlementTable()) + + assert bg._STORE.store is previous_store + + +@dataclass(frozen=True) +class _JsonLike: + data: object