lumen_typing_pace.py

"""
lumen_typing_pace.py — самокалибрующийся расчёт скорости "печати" при стриминге
ответа в Telegram (см. _run_streaming_reply в bot.py).

── Почему НЕ статическая таблица "N токенов/сек у модели X" ──
Проверено при разработке этой фичи: у бесплатных моделей OpenRouter реальная
скорость отдачи текста не является свойством самой модели — OpenRouter
маршрутизирует один и тот же ":free" слаг на РАЗНЫХ бэкенд-провайдеров в
зависимости от текущей загрузки (это часть их обычной механики маршрутизации),
и разные бэкенды одной и той же модели могут закончить генерацию и прислать
готовый текст ОДНИМ SSE-чанком вместо потока токен-в-токен — тогда "скорость"
в смысле частоты появления кусков вообще не определена как константа модели.
Опубликованные на сайтах провайдеров цифры throughput — это скользящая медиана
за недавнее окно, которая устаревает быстрее, чем список живых/мёртвых моделей
в _OR_MODEL_HEALTH (lumen_router_config.py), и относится к конкретному ИХ
бэкенду, а не к тому, что реально ответит на конкретный запрос этого бота.
Захардкоженная таблица "актуальных" скоростей была бы обречена на тот же
износ, только без единого способа заметить, что она устарела (в отличие от
мёртвых моделей — там хотя бы HTTP 404 в логах сигналит о проблеме).

── Что вместо этого ──
Реальная скорость появления символов измеряется по факту на каждом стриме
(см. record_observed_speed — вызывается из bot.py ОДИН раз в конце успешного
стрима, до искусственной фазы "довывода", чтобы та не искажала замер) и
усредняется экспоненциально (EMA) по ключу provider:model_id. Никакого ручного
обслуживания при добавлении/замене моделей не требуется — новая модель просто
стартует с DEFAULT_CHARS_PER_SEC и за первые несколько ответов "нащупывает"
свою реальную скорость сама, включая случаи, когда OpenRouter на лету меняет
бэкенд той же самой модели.

Единица измерения — символы в секунду, а не токены: ни google-genai SDK, ни
SSE-дельты OpenRouter не отдают надёжный подсчёт токенов на кусок, а для
визуального эффекта набора текста важны именно видимые символы. Раз в основе
не токены, а символы — сравнение с t/s дашбордов провайдеров всё равно было бы
приблизительным, что дополнительно снимает смысл держать "точную" таблицу.

Состояние (_speed_ema) — только в памяти процесса, намеренно НЕ персистентное
(в отличие от GLOBAL_QUOTA): это чисто косметическая оценка, не критичный
факт — за первые же несколько сообщений после рестарта она снова "нащупается"
сама, а тащить её через Upstash/диск ради этого было бы накоплением сложности
без реальной пользы (тот же принцип, что и у остального проекта — см. YAGNI
в других модулях).
"""

from __future__ import annotations

# ── границы скорости печати (символов/сек) ──
# Подобраны эмпирически под ощущение "похоже на живой набор текста в Telegram",
# а не взяты из чьей-то спецификации — при желании владелец может изменить эти
# три константы прямо здесь, менять их часто не нужно.
DEFAULT_CHARS_PER_SEC = 90.0
MIN_CHARS_PER_SEC = 40.0
MAX_CHARS_PER_SEC = 260.0

# Насколько сильно один новый замер сдвигает EMA. Чем меньше — тем стабильнее
# оценка (не скачет от одного нетипичного ответа), но тем медленнее подстраивается
# под реальную смену бэкенда OpenRouter под тем же слагом.
_EMA_ALPHA = 0.3

_speed_ema: dict[str, float] = {}


def speed_key(provider: str, model_id: str) -> str:
    """Единый ключ для _speed_ema — тот же принцип пары (provider, model_id),
    что уже используется в GLOBAL_QUOTA (см. _quota_entry в bot.py)."""
    return f"{provider}:{model_id}"


def get_typing_speed(key: str) -> float:
    """Текущая оценка скорости печати для этой модели — DEFAULT_CHARS_PER_SEC,
    пока не накопилось ни одного реального замера. Результат всегда в границах
    [MIN_CHARS_PER_SEC, MAX_CHARS_PER_SEC], даже если константа DEFAULT когда-нибудь
    будет отредактирована за пределы этого диапазона по ошибке."""
    return max(MIN_CHARS_PER_SEC, min(MAX_CHARS_PER_SEC, _speed_ema.get(key, DEFAULT_CHARS_PER_SEC)))


def record_observed_speed(key: str, elapsed_sec: float, chars_len: int) -> None:
    """Обновляет EMA по итогам ОДНОГО завершённого стрима — вызывать один раз в
    конце (не на каждый кусок SSE), нас интересует средняя скорость всего ответа,
    а не шум отдельных кусков. elapsed_sec должен быть временем ЕСТЕСТВЕННОГО
    получения текста (от начала стрима до его исчерпания), БЕЗ искусственной фазы
    "довывода" (см. bot.py) — иначе самим же добавленным задержкам ЕМА поверила бы
    как настоящей медленной скорости бэкенда, и оценка бы разъехалась с реальностью.

    Сырое наблюдение зажимается в [MIN_CHARS_PER_SEC, MAX_CHARS_PER_SEC] ДО
    усреднения — без этого один нетипичный ответ, пришедший от бэкенда одним
    большим куском за доли секунды (наблюдаемая "скорость" тогда — тысячи
    симв/сек), утащил бы EMA в небеса, и следующий ответ той же модели "мигал"
    бы мгновенно вместо плавного набора — то есть ровно та проблема, которую
    эта функция должна была решить."""
    if elapsed_sec <= 0 or chars_len <= 0:
        return
    observed = max(MIN_CHARS_PER_SEC, min(MAX_CHARS_PER_SEC, chars_len / elapsed_sec))
    prev = _speed_ema.get(key)
    _speed_ema[key] = observed if prev is None else (_EMA_ALPHA * observed + (1 - _EMA_ALPHA) * prev)


def catchup_reveal_steps(remaining_len: int, chars_per_sec: float, tick_interval_sec: float, max_ticks: int) -> list[int]:
    """Раскладывает "довывод" остатка уже полностью полученного, но ещё не
    полностью показанного текста на несколько шагов (см. _run_streaming_reply в
    bot.py — вызывается ПОСЛЕ того, как стрим уже исчерпан, чтобы отображение не
    "прыгало" сразу на весь текст, если бэкенд прислал его одним большим куском).

    Каждый элемент возвращённого списка — кумулятивная длина видимого текста на
    этом шаге (не дельта). Список ограничен max_ticks элементами — если по
    оценённой скорости потребовалось бы больше шагов, ПОСЛЕДНИЙ шаг форсированно
    добирает до remaining_len целиком, а не оставляет хвост невидимым навсегда.
    Это значит, что общая добавленная задержка НИКОГДА не превышает
    max_ticks * tick_interval_sec, независимо от длины ответа и того, насколько
    заниженной оказалась оценка скорости — реальная скорость ответа не должна
    страдать ради красивости.

    Чисто арифметическая функция без обращения к часам — намеренно: если бы шаг
    ориентировался на time.monotonic() внутри цикла, а вызывающий код в тестах
    подменяет фактическое ожидание между шагами на no-op (см. bot._typing_sleep и
    autouse-фикстуру в conftest.py — реальные секунды ожидания в тестах не нужны),
    время между шагами никогда не продвигалось бы, и цикл завис бы навсегда.
    Здесь такого риска нет: число шагов и их длины вычисляются заранее."""
    if remaining_len <= 0:
        return []
    chars_per_tick = max(1, int(chars_per_sec * tick_interval_sec))
    steps: list[int] = []
    revealed = 0
    while revealed < remaining_len and len(steps) < max_ticks:
        revealed = min(remaining_len, revealed + chars_per_tick)
        steps.append(revealed)
    if steps and steps[-1] < remaining_len:
        steps[-1] = remaining_len
    return steps