ci(realtime): clear the repo gates surfaced by the upstream merge

Four repository gates flagged the merged tree; each fix is the narrowest
one that keeps the PR behavior and the budgets untouched.

- type-discipline LIT013 (upstream's new dead-suppression rule) flagged
  eleven `*-ok` reason comments in the Live/realtime modules that no
  longer suppress any rule: removed those comments only, keeping every
  suppression the checker still applies. budget_ratchet_check confirms no
  budget file moved.
- type-discipline LIT014 (upstream's new one-for/one-if comprehension
  cap) flagged the double-for generator that flattened the Redis cluster
  acquire arguments; replaced it with an explicit loop appending the same
  (limit, ttl, slot) triple per key, identical wire payload.
- basedpyright reportGeneralTypeIssues sat at 102/101: the cached-rate
  branch of calculate_image_response_cost_from_usage redeclared the
  `model_info` parameter as a `Final` local; the local is now
  `catalog_model_info`, and the four rate lookups read the same
  get_model_info entry as before.
- upstream's pre_call_hook now always forwards skip_guardrails; the PR's
  PolicyHook test double accepts that keyword and still pins
  internal_realtime_observer in its assertions.

Validation on this exact tree: 550 passed in the realtime/Live suites,
474 passed across the three cost-calculator suites, 715 passed for the
v3 limiter plus the two auth files that exercise it, and the
test-quality, ruff-strict, budget-ratchet, type-discipline, and
basedpyright gates all pass with --base upstream/main.
This commit is contained in:
jibanez-staticduo 2026-09-25 17:49:45 +02:00
parent 042f26b567
commit 68b573ea8d
No known key found for this signature in database
6 changed files with 23 additions and 25 deletions

View file

@ -1860,17 +1860,17 @@ def calculate_image_response_cost_from_usage(
) )
if cached_details is None: if cached_details is None:
return prompt_cost + completion_cost return prompt_cost + completion_cost
model_info: Final = get_model_info(model=model, custom_llm_provider=custom_llm_provider) catalog_model_info: Final = get_model_info(model=model, custom_llm_provider=custom_llm_provider)
cached_text: Final = _get_token_detail_value(cached_details, "text_tokens") or 0 cached_text: Final = _get_token_detail_value(cached_details, "text_tokens") or 0
cached_image: Final = _get_token_detail_value(cached_details, "image_tokens") or 0 cached_image: Final = _get_token_detail_value(cached_details, "image_tokens") or 0
input_text_tokens: Final = _get_token_detail_value(input_tokens_details, "text_tokens") or 0 input_text_tokens: Final = _get_token_detail_value(input_tokens_details, "text_tokens") or 0
input_image_tokens: Final = _get_token_detail_value(input_tokens_details, "image_tokens") or 0 input_image_tokens: Final = _get_token_detail_value(input_tokens_details, "image_tokens") or 0
if not (0 <= cached_text <= input_text_tokens and 0 <= cached_image <= input_image_tokens): if not (0 <= cached_text <= input_text_tokens and 0 <= cached_image <= input_image_tokens):
raise ValueError("Image cached token counts exceed their input modality counts") raise ValueError("Image cached token counts exceed their input modality counts")
text_rate: Final = model_info.get("input_cost_per_token") or 0.0 text_rate: Final = catalog_model_info.get("input_cost_per_token") or 0.0
image_rate: Final = model_info.get("input_cost_per_image_token") image_rate: Final = catalog_model_info.get("input_cost_per_image_token")
cache_text_rate: Final = model_info.get("cache_read_input_token_cost") cache_text_rate: Final = catalog_model_info.get("cache_read_input_token_cost")
cache_image_rate: Final = model_info.get("cache_read_input_image_token_cost") cache_image_rate: Final = catalog_model_info.get("cache_read_input_image_token_cost")
text_savings: Final = cached_text * (text_rate - cache_text_rate) if cache_text_rate is not None else 0.0 text_savings: Final = cached_text * (text_rate - cache_text_rate) if cache_text_rate is not None else 0.0
image_savings: Final = ( image_savings: Final = (
cached_image * ((image_rate if image_rate is not None else text_rate) - cache_image_rate) cached_image * ((image_rate if image_rate is not None else text_rate) - cache_image_rate)

View file

@ -53,10 +53,10 @@ def _validate_session_id(session_id: str) -> None:
or any(char in ("/", "\\") or category(char) in ("Cc", "Cs") for char in candidate) or any(char in ("/", "\\") or category(char) in ("Cc", "Cs") for char in candidate)
): ):
raise ValueError("Invalid Live session ID") raise ValueError("Invalid Live session ID")
decoded: str = unquote(candidate, errors="strict") # rebind-ok: validate successive decoding layers iteratively decoded: str = unquote(candidate, errors="strict")
if decoded == candidate: if decoded == candidate:
return return
candidate = decoded # rebind-ok: each percent-decoding pass reduces the input length candidate = decoded
def _validate_path(path: str) -> None: def _validate_path(path: str) -> None:

View file

@ -245,7 +245,7 @@ class ChatGPTRealtimeHTTPConfig(OpenAIRealtimeHTTPConfig):
self, response: Response, model: str, model_id: str | None, headers: Mapping[str, object] | None self, response: Response, model: str, model_id: str | None, headers: Mapping[str, object] | None
) -> Response: ) -> Response:
response.extensions["chatgpt_realtime"] = ( response.extensions["chatgpt_realtime"] = (
MappingProxyType( # rebind-ok: HTTPX response extensions carry provider routing metadata MappingProxyType(
{ {
"model": model, "model": model,
"model_id": model_id, "model_id": model_id,

View file

@ -1733,15 +1733,11 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
if self.parallel_acquire_script is None: if self.parallel_acquire_script is None:
raise RuntimeError("Redis cluster parallel acquire script is unavailable") raise RuntimeError("Redis cluster parallel acquire script is unavailable")
attempted.extend(keys) attempted.extend(keys)
acquire_args: list[object] = [] # mutable-ok: Redis EVAL args are flattened per slot below
for key in keys:
acquire_args.extend((by_key[key]["limit"], PARALLEL_REQUEST_SLOT_TTL_SECONDS, slot_id))
(raw,) = ( (raw,) = (
await self.parallel_acquire_script( await self.parallel_acquire_script(keys=keys, args=tuple(acquire_args)),
keys=keys,
args=tuple(
arg
for key in keys
for arg in (by_key[key]["limit"], PARALLEL_REQUEST_SLOT_TTL_SECONDS, slot_id)
),
),
) )
if int(raw[0]) == 1: if int(raw[0]) == 1:
await self._rollback_cluster_parallel_slots(tuple(attempted), slot_id, parent_otel_span) await self._rollback_cluster_parallel_slots(tuple(attempted), slot_id, parent_otel_span)

View file

@ -85,18 +85,18 @@ def _json_value(value: object) -> JsonValue:
source, parent, key, depth = pending.pop() source, parent, key, depth = pending.pop()
if depth > 256: if depth > 256:
raise ValueError("Live JSON nesting exceeds the supported depth") raise ValueError("Live JSON nesting exceeds the supported depth")
converted: JsonValue # rebind-ok: each visited input produces a new JSON value converted: JsonValue
if isinstance(source, Mapping): if isinstance(source, Mapping):
entries: Mapping[str, object] = _MAPPING.validate_python( entries: Mapping[str, object] = _MAPPING.validate_python(
source source
) # rebind-ok: entries belong to the current node )
converted = {name: None for name in entries} # mutable-ok: JSON wire objects require dicts converted = {name: None for name in entries} # mutable-ok: JSON wire objects require dicts
pending.extend((item, converted, name, depth + 1) for name, item in entries.items()) pending.extend((item, converted, name, depth + 1) for name, item in entries.items())
elif isinstance(source, (tuple, list)): elif isinstance(source, (tuple, list)):
items: tuple[object, ...] = TypeAdapter(tuple[object, ...]).validate_python( items: tuple[object, ...] = TypeAdapter(tuple[object, ...]).validate_python(
source source
) # rebind-ok: items belong to the current node )
array: list[JsonValue] = [None] * len(items) # mutable-ok: JSON output; # rebind-ok: per-node buffer array: list[JsonValue] = [None] * len(items)
pending.extend((item, array, index, depth + 1) for index, item in enumerate(items)) pending.extend((item, array, index, depth + 1) for index, item in enumerate(items))
converted = array converted = array
else: else:
@ -583,9 +583,9 @@ def _managed_constraints(auth: UserAPIKeyAuth) -> bool:
) )
if isinstance(value, (Mapping, list, tuple)) if isinstance(value, (Mapping, list, tuple))
] ]
visited: Final[set[int]] = set() # mutable-ok: cycle guard for hook-provided metadata visited: Final[set[int]] = set()
while pending: while pending:
current: object = pending.pop() # rebind-ok: advance the explicit metadata traversal stack current: object = pending.pop()
if id(current) in visited: if id(current) in visited:
continue continue
visited.add(id(current)) visited.add(id(current))
@ -594,7 +594,7 @@ def _managed_constraints(auth: UserAPIKeyAuth) -> bool:
if isinstance(current, Mapping): if isinstance(current, Mapping):
entries: Mapping[str, object] = _MAPPING.validate_python( entries: Mapping[str, object] = _MAPPING.validate_python(
current current
) # rebind-ok: entries belong to the current metadata node )
for key, item in entries.items(): for key, item in entries.items():
if key in ( if key in (
"rpm_limit", "rpm_limit",
@ -1411,7 +1411,7 @@ async def _wait_started(
while True: while True:
event: Mapping[str, JsonValue] = _OBJECT.validate_json( event: Mapping[str, JsonValue] = _OBJECT.validate_json(
await connection.recv() await connection.recv()
) # rebind-ok: each received event has a new value )
if event.get("type") == "session.started": if event.get("type") == "session.started":
return event return event
if startup is not None: if startup is not None:

View file

@ -301,7 +301,9 @@ async def test_codex_processing_merges_model_guardrails(monkeypatch, route_type,
from litellm.proxy.realtime_endpoints.call_sessions import process_codex_request from litellm.proxy.realtime_endpoints.call_sessions import process_codex_request
class PolicyHook: class PolicyHook:
async def pre_call_hook(self, user_api_key_dict, data, call_type, *, internal_realtime_observer=False): async def pre_call_hook(
self, user_api_key_dict, data, call_type, *, skip_guardrails=False, internal_realtime_observer=False
):
assert internal_realtime_observer is observer assert internal_realtime_observer is observer
if "model-policy" in data.get("metadata", {}).get("guardrails", []): if "model-policy" in data.get("metadata", {}).get("guardrails", []):
raise HTTPException(403, "Model policy rejected request") raise HTTPException(403, "Model policy rejected request")