litellm/patch_audio_request_cost.py

"""Bill a declared per-request charge on requests LiteLLM prices only per unit.

Run at BUILD time (the runtime rootfs is read-only under Cloudron). Non-fatal:
if the target file or the anchor strings are missing — e.g. after an upstream
version bump — it prints a warning and exits 0, so the build and the gateway
still succeed. The build log line ``[coop-patch] applied`` is what confirms it
took effect; check it on every LiteLLM upgrade (see README, upgrade checklist).

Why: Tinfoil charges some models a flat amount PER REQUEST on top of (or instead
of) per-token pricing — ``voxtral-tts`` $0.002/request and
``whisper-large-v3-turbo`` $0.01/request, both with $0 per-token. LiteLLM can
only express per-request pricing on the search / rerank / MCP paths, via
``input_cost_per_query`` (see ``litellm/search/cost_calculator.py``, and
``litellm/types/utils.py``: "per-request pricing: rerank, search, and Bedrock
Marengo embed"). The audio paths price by character, second or token, and
``input_cost_per_request`` exists only in the proxy's cost-tracking settings —
the model cost calculator never reads it. So speech and transcription usage is
billed at $0 however the config is written.

Patch: make the model cost calculator honour a declared
``input_cost_per_request``. ``response_cost_calculator`` is the single funnel for
every call type (the proxy's audio endpoints dispatch as ``aspeech`` and
``transcription`` and reach it through ``Logging._response_cost_calculator``), so
adding the charge at its final return covers both. The declared value is resolved
from the per-deployment ``custom_pricing`` the logging layer builds out of our
``config.yaml`` ``model_info``, falling back to the shared cost map and to the
call metadata.

Scope: nothing changes unless a model declares ``input_cost_per_request``, so
this is inert for every other model on the gateway.

Upstream: no issue filed; LiteLLM has no per-request audio pricing as of 1.103.2.
"""
from __future__ import annotations

import glob
import os
import sys

MARKER = "CO-OP PATCH (audio per-request cost, 2026-10-26)"

HELPER = '''def _coop_per_request_cost(
    *,
    model: str | None = None,
    litellm_model_name: str | None = None,
    custom_pricing: Any | None = None,
    litellm_logging_obj: Any | None = None,
) -> float:
    """CO-OP PATCH (audio per-request cost, 2026-10-26)

    Return the model's declared ``input_cost_per_request``, else 0.0.

    LiteLLM cannot price a flat per-request charge on the audio paths, so
    Tinfoil's per-request models (voxtral-tts, whisper-large-v3-turbo) were
    billed at $0. Resolution order: the per-deployment custom pricing the
    logging layer builds from our config.yaml model_info, then the shared cost
    map, then the call metadata. Inert unless a model declares the field.
    """
    candidates = []
    if isinstance(custom_pricing, dict):
        candidates.append(custom_pricing)
    try:
        import litellm as _litellm  # local import: not every caller binds it globally
        cost_map = _litellm.model_cost
        for key in (litellm_model_name, model):
            entry = cost_map.get(key) if key else None
            if isinstance(entry, dict):
                candidates.append(entry)
    except Exception:
        pass
    try:
        details = getattr(litellm_logging_obj, "model_call_details", None) or {}
        info = (details.get("metadata") or {}).get("model_info")
        if isinstance(info, dict):
            candidates.append(info)
    except Exception:
        pass
    for candidate in candidates:
        try:
            value = candidate.get("input_cost_per_request")
        except Exception:
            continue
        if value:
            try:
                return float(value)
            except (TypeError, ValueError):
                continue
    return 0.0


'''

ANCHOR_DEF = "def response_cost_calculator("

ANCHOR_RETURN = """                region_name=region_name,
            )
        return response_cost
    except Exception as e:
        raise e"""

PATCHED_RETURN = """                region_name=region_name,
            )

        # CO-OP PATCH (audio per-request cost, 2026-10-26): add the declared
        # per-request charge for models LiteLLM can only price per unit
        # (Tinfoil's voxtral-tts, whisper-large-v3-turbo). Inert when the model
        # declares nothing; the early "provider reported the cost" return above
        # is unaffected.
        try:
            response_cost = response_cost + _coop_per_request_cost(
                model=model,
                litellm_model_name=litellm_model_name,
                custom_pricing=custom_pricing,
                litellm_logging_obj=litellm_logging_obj,
            )
        except Exception:
            pass

        return response_cost
    except Exception as e:
        raise e"""


def target_path() -> str:
    """The installed litellm cost calculator (or an explicit path for testing)."""
    if len(sys.argv) > 1:
        return sys.argv[1]
    env = os.environ.get("LITELLM_COST_CALCULATOR")
    if env:
        return env
    hits = []
    try:  # most reliable: ask the installed package where it lives
        import litellm  # noqa: PLC0415
        _file = getattr(litellm, "__file__", None)
        if _file:
            hits.append(os.path.join(os.path.dirname(_file), "cost_calculator.py"))
    except Exception:
        pass
    hits += glob.glob("/usr/lib/python3*/site-packages/litellm/cost_calculator.py")
    hits += glob.glob("/usr/local/lib/python3*/site-packages/litellm/cost_calculator.py")
    hits += glob.glob("/usr/local/lib/python3*/dist-packages/litellm/cost_calculator.py")
    hits += glob.glob("/opt/**/litellm/cost_calculator.py", recursive=True)
    hits = [h for h in hits if os.path.exists(h)]
    if not hits:
        raise SystemExit("[coop-patch] could not locate litellm/cost_calculator.py")
    return sorted(hits)[0]


def main() -> int:
    path = target_path()
    try:
        source = open(path).read()
    except OSError as exc:
        print(f"[coop-patch] cannot read {path}: {exc}")
        return 0

    if MARKER in source:
        print(f"[coop-patch] {path} already patched, nothing to do")
        return 0

    if ANCHOR_DEF not in source or ANCHOR_RETURN not in source:
        print(f"[coop-patch] WARNING anchors not found in {path}; audio per-request "
              f"cost will NOT be billed (check this after a LiteLLM upgrade)")
        return 0

    patched = source.replace(ANCHOR_DEF, HELPER + ANCHOR_DEF, 1)
    patched = patched.replace(ANCHOR_RETURN, PATCHED_RETURN, 1)
    compile(patched, path, "exec")  # refuse to write something that will not import

    if not os.access(path, os.W_OK):  # practice run against a read-only copy
        print(f"[coop-patch] {path} is not writable; verified anchors and syntax only")
        return 0

    with open(path, "w") as handle:
        handle.write(patched)
    print(f"[coop-patch] applied to {path}")
    return 0


if __name__ == "__main__":
    sys.exit(main())