""" lumen_streaming.py — провайдер-агностичный стриминг (плейсхолдер, точки, троттлинг правок, скраб утечек на чанк, catch-up). Связи с bot.py — только через отложенный `import bot`; точки подмены для тестов читаются через `bot.` в момент вызова. """ from __future__ import annotations import asyncio import contextlib import json import logging import time from typing import Any import aiohttp from aiogram.enums import ParseMode from aiogram.types import Message from lumen_formatting import _md_to_html, _split_text_chunks from lumen_message_parse import _history_user_text from lumen_model_speed import ( speed_key as _model_speed_key, record_response as _record_model_latency, first_chunk_limit_sec as _model_first_chunk_limit, ) from lumen_router_config import GEMINI_MODELS from lumen_security import ( _leak_scan_window, _IDENTITY_LEAK_FALLBACK, _INJECTED_PAYLOAD_ECHO_FALLBACK, _detect_injected_payload_echo, _detect_garbled_mix, _is_garbled_echo, _detect_identity_leak, _scrub_identity_leak, ) from lumen_typing_pace import ( speed_key as _typing_speed_key, record_observed_speed as _record_typing_speed, catchup_reveal_steps as _typing_catchup_steps, blend_arrival_speed as _typing_arrival_update, display_speed_for as _typing_display_speed, ) log = logging.getLogger("bot") async def _tick_waiting_dots(placeholder: Message) -> None: """Точки в плейсхолдере до первого куска: первый кадр после паузы (быстрые модели анимации не видят), дальше кадр по тику. Правки через _tg_call с глотанием ошибок, отмена — штатная остановка.""" import bot try: await bot._dots_sleep(bot._DOTS_START_AFTER_SEC) frame_idx = 0 while True: with contextlib.suppress(Exception): await bot._tg_call(placeholder.edit_text, bot._DOTS_FRAMES[frame_idx % len(bot._DOTS_FRAMES)], parse_mode=None, call_timeout=15.0) frame_idx += 1 await bot._dots_sleep(bot._DOTS_TICK_SEC) except asyncio.CancelledError: raise async def _pieces_with_waiting_feedback(piece_agen, placeholder: Message, *, first_chunk_limit: float): """Генератор + ожидание первого куска: адаптивный предел (зависшая попытка — TimeoutError, плейсхолдер уйдёт следующей модели) и анимация точек строго до первого yield. Пустой стрим — просто конец.""" dots_task = asyncio.create_task(_tick_waiting_dots(placeholder)) try: try: first = await asyncio.wait_for(piece_agen.__anext__(), timeout=first_chunk_limit) except StopAsyncIteration: return finally: dots_task.cancel() with contextlib.suppress(asyncio.CancelledError): await dots_task yield first async for piece in piece_agen: yield piece finally: aclose = getattr(piece_agen, "aclose", None) if aclose is not None: with contextlib.suppress(Exception): await aclose() def _remaining_budget(deadline: float | None) -> float: """Сколько секунд осталось до бюджета маршрута (float('inf') — потолка нет).""" if deadline is None: return float("inf") return max(0.0, deadline - time.monotonic()) def _stream_wait(bot: Any, deadline: float | None) -> float: """Таймаут на ОДНУ операцию стрима: не больше межкускового капа и не больше остатка бюджета маршрута. Раньше ожидание шло в обход бюджета, и кусок мог ждать STREAM_CHUNK_TIMEOUT_SEC уже после его исчерпания (аудит 26.09.2026). Значение меньше 0.1с не опускаем: это всё равно мгновенный TimeoutError.""" return max(0.1, min(bot.STREAM_CHUNK_TIMEOUT_SEC, _remaining_budget(deadline))) async def _gemini_stream_pieces(model_id: str, call_contents: list, gconfig, *, deadline: float | None = None): """Куски от Gemini (тонкая обёртка над generate_content_stream). В google-genai 2.27.0 это async def: вызов надо ждать (пример из SDK: `async for chunk in await client.aio.models.generate_content_stream(...)`). Реальную сетевую жду ловлю ниже: первый кусок идёт через __anext__ под тем же _stream_wait.""" import bot stream = await bot.client.aio.models.generate_content_stream(model=model_id, contents=call_contents, config=gconfig) stream_iter = stream.__aiter__() try: while True: try: chunk = await asyncio.wait_for(stream_iter.__anext__(), timeout=_stream_wait(bot, deadline)) except StopAsyncIteration: break try: piece = getattr(chunk, "text", "") or "" except Exception: piece = "" if piece: yield piece finally: aclose = getattr(stream, "aclose", None) if aclose is not None: with contextlib.suppress(Exception): await aclose() async def _sse_pieces( session, url: str, headers: dict[str, str], payload: dict, *, err_cls, provider_label: str, deadline: float | None = None, ): """Общий разбор OpenAI-совместимого SSE для OpenRouter и Groq (оба дают `data: {"choices":[{"delta":{"content":...}}]}` и финальный `data: [DONE]`). Раньше это были две копии по 45 строк, расходившиеся деталями: у Groq не было теста на HTTP-ошибку и на ошибку внутри чанка (аудит 26.09.2026). Ошибка внутри чанка (HTTP 200, стрим уже открыт) поднимается: ранний сбой — откат на следующую модель, поздний — пометка "соединение прервалось".""" import bot post_cm = session.post(url, headers=headers, json=payload, timeout=aiohttp.ClientTimeout(total=None, connect=12.0)) try: # Заголовки без total висели вечно и держали lock чата до рестарта (аудит 29.09.2026), # поэтому вход ждём не дольше попытки маршрута, а тело стрима — по-прежнему покусочно. resp = await asyncio.wait_for(post_cm.__aenter__(), timeout=bot._attempt_timeout(bot, deadline)) except BaseException: with contextlib.suppress(Exception): await post_cm.__aexit__(None, None, None) raise try: if resp.status >= 400: body = await resp.read() raise err_cls(f"HTTP {resp.status}: {body[:300]!r}", status_code=resp.status) line_iter = resp.content.__aiter__() while True: try: raw_line = await asyncio.wait_for(line_iter.__anext__(), timeout=_stream_wait(bot, deadline)) except StopAsyncIteration: break line = raw_line.decode("utf-8", errors="ignore").strip() if not line or not line.startswith("data:"): continue data_str = line[len("data:"):].strip() if data_str == "[DONE]": break try: obj = json.loads(data_str) except Exception: continue err_obj = obj.get("error") if err_obj: err_msg = err_obj.get("message") if isinstance(err_obj, dict) else str(err_obj) err_code = err_obj.get("code") if isinstance(err_obj, dict) else None raise err_cls(err_msg or f"{provider_label} returned an error in the stream body", status_code=err_code) choices = obj.get("choices") or [] if not choices: continue delta = choices[0].get("delta") or {} piece = delta.get("content") or "" if piece: yield piece finally: with contextlib.suppress(Exception): await post_cm.__aexit__(None, None, None) async def _openrouter_stream_pieces(model_id: str, messages: list[dict], *, deadline: float | None = None): """Куски от OpenRouter: SSE через общий _sse_pieces, таймаут на строку — тот же STREAM_CHUNK_TIMEOUT_SEC, но не дольше остатка бюджета маршрута.""" import bot if not bot.OPENROUTER_API_KEY: raise bot.OpenRouterAPIError("OPENROUTER_API_KEY is not set") headers = { "Authorization": f"Bearer {bot.OPENROUTER_API_KEY}", "HTTP-Referer": bot.OPENROUTER_HTTP_REFERER, "X-OpenRouter-Title": bot.OPENROUTER_TITLE, "Content-Type": "application/json", } session = await bot._get_http_session() async for piece in _sse_pieces( session, f"{bot.OPENROUTER_BASE_URL}/chat/completions", headers, {"model": model_id, "messages": messages, "stream": True}, err_cls=bot.OpenRouterAPIError, provider_label="OpenRouter", deadline=deadline, ): yield piece async def _groq_stream_pieces(model_id: str, messages: list[dict], *, deadline: float | None = None): """Куски от Groq: тот же OpenAI-SSE через общий _sse_pieces (Groq — OpenAI-совместимый API), таймаут на строку — как у OpenRouter.""" import bot if not bot.GROQ_API_KEY: raise bot.GroqAPIError("GROQ_API_KEY is not set") headers = { "Authorization": f"Bearer {bot.GROQ_API_KEY}", "Content-Type": "application/json", } session = await bot._get_http_session() async for piece in _sse_pieces( session, f"{bot.GROQ_BASE_URL}/chat/completions", headers, {"model": model_id, "messages": messages, "stream": True}, err_cls=bot.GroqAPIError, provider_label="Groq", deadline=deadline, ): yield piece async def _run_streaming_reply( chat_id: int, user_text: str, message: Message, *, provider: str, model_id: str, piece_agen, deadline: float | None = None, ) -> tuple[str | None, Message | None]: """Стриминг: генератор кусков + плейсхолдер, чанкинг, троттлинг, обрыв при утечке/инъекции, HTML-финал, запись в историю. Возвращает (ответ, плейсхолдер): успех — (текст, None); сбой до показа — (None, плейсхолдер для следующей модели); сбой после — (показанное + пометка, None).""" import bot state = bot.get_state(chat_id) hist = state.setdefault("history", []) sent_messages: list[Message] = [] full_text = "" # Базис последнего разбиения: чанки кроме последнего стабильны, поэтому # _split_text_chunks пересчитываем только при заметном приросте, а не на # каждый кусок (иначе O(n²) в loop при посимвольном стриме). split_basis = "" chunks: list[str] = [] edit_fail_streak = 0 edits_dead = False last_edit_ts = 0.0 last_edited_plain = "" first_piece_ts: float | None = None # Замеры скорости/латентности — см. lumen_typing_pace.py (overall один раз на весь ответ, arrival — заново на каждое новое сообщение). pace_key = _typing_speed_key(provider, model_id) overall_start_ts = time.monotonic() # reveal_base_ts — момент ПЕРВОГО куска, а не старта стрима: иначе после долгой тишины формула сразу разрешала показать всё накопленное разом. last_piece_ts/arrival_ewma — замер реального темпа прихода. reveal_base_ts: float | None = None last_piece_ts: float | None = None arrival_ewma: float | None = None try: placeholder = await bot._tg_call(message.reply, "…", call_timeout=bot.TELEGRAM_REQUEST_TIMEOUT) if placeholder is None: return None, None sent_messages.append(placeholder) # Попытка к модели состоялась (запрос уже ушёл в генератор кусков); # пустой считают только ранний обрыв до показа — частичный ответ показан. bot._record_stats_event("model_attempts") # Дальше цикл читает генератор напрямую — проверки кусков/утечек/троттлинга ниже без изменений. piece_agen = _pieces_with_waiting_feedback( piece_agen, placeholder, first_chunk_limit=_model_first_chunk_limit(_model_speed_key(provider, model_id), bot.FIRST_CHUNK_TIMEOUT_SEC), ) async for piece in piece_agen: if not piece: continue # Общий бюджет маршрута — и на стрим тоже: капающий по куску раз в 29с стрим # иначе держал бы lock чата далеко за ROUTE_TOTAL_BUDGET_SEC (найдено внешним аудитом). if deadline is not None and time.monotonic() > deadline: raise bot.RouteBudgetExceededError([f"{provider}:{model_id}"]) now_piece = time.monotonic() if first_piece_ts is None: first_piece_ts = now_piece reveal_base_ts = now_piece else: arrival_ewma = _typing_arrival_update(arrival_ewma, len(piece), now_piece - (last_piece_ts or now_piece)) last_piece_ts = now_piece full_text += piece leak_kind = None _scan_text = _leak_scan_window(full_text, piece) if _detect_identity_leak(_scan_text): leak_kind = "identity" elif _detect_injected_payload_echo(_scan_text): leak_kind = "payload_echo" if leak_kind: # Проверяем ДО edit_text — утечка не успевает "мигнуть" на экране. tag = "identity-leak" if leak_kind == "identity" else "injection-echo" log.warning( '[%s] Stream %s/%s started leaking internal details/echoing an injected instruction — aborting the stream and showing a neutral reply instead (accumulated_len=%d)', tag, provider, model_id, len(full_text), ) aclose = getattr(piece_agen, "aclose", None) if aclose is not None: with contextlib.suppress(Exception): await aclose() final_answer = _IDENTITY_LEAK_FALLBACK if leak_kind == "identity" else _INJECTED_PAYLOAD_ECHO_FALLBACK await bot._tg_call(sent_messages[-1].edit_text, final_answer, parse_mode=None, call_timeout=15.0) hist.append({"role": "user", "content": _history_user_text(user_text)}) hist.append({"role": "assistant", "content": final_answer}) # Обрезка с саммари старого (см. _trim_history), а не молчаливый срез. await bot._trim_history(hist) bot._trim_combined(state) bot._record_quota_usage(provider, model_id) return final_answer, None if not chunks or len(full_text) - len(split_basis) >= 1000: chunks = _split_text_chunks(full_text, bot.TG_MAX_LEN) split_basis = full_text # Чанки кроме последнего (растущего) финализируем тем же разбиением, что у нестримленных ответов. while len(chunks) > len(sent_messages): idx = len(sent_messages) - 1 # Continuation открываем ПЕРВЫМ и финализируем одним вызовом — раньше было два edit_text при отказе (лишний HTTP-запрос). # bot.bot — инстанс aiogram Bot (имя bot занято модулем). new_msg = await bot._tg_call(bot.bot.send_message, chat_id=message.chat.id, text="…", call_timeout=bot.TELEGRAM_REQUEST_TIMEOUT) if new_msg is None: # idx ещё НЕ финализирован — не даём общему except перезаписать показанный текст чужим (там инвариант "sent[-1] == chunks[-1]" уже неверен). note = bot._t(message.chat.id, "stream_note_send_fail") with contextlib.suppress(Exception): await bot._tg_call(sent_messages[idx].edit_text, _md_to_html(chunks[idx]) + note, parse_mode=ParseMode.HTML, call_timeout=15.0) final_answer = full_text.strip() hist.append({"role": "user", "content": _history_user_text(user_text)}) hist.append({"role": "assistant", "content": final_answer}) # Обрезка с саммари (см. _trim_history): молчаливый срез оставлял # саммаризацию только нестриминговым путям (аудит 26.09.2026). await bot._trim_history(hist) bot._trim_combined(state) bot._record_quota_usage(provider, model_id) return final_answer, None if not edits_dead: edit_ok = await bot._tg_call(sent_messages[idx].edit_text, _md_to_html(chunks[idx]), parse_mode=ParseMode.HTML, call_timeout=15.0) # Серия неуспешных правок подряд — сообщение снесли посреди стрима: # дальше не правим, в финале досылаем ответ новым сообщением. edit_fail_streak = 0 if edit_ok is not None else edit_fail_streak + 1 edits_dead = edit_fail_streak >= 3 sent_messages.append(new_msg) last_edited_plain = "" # Новое сообщение — новый "лист", показ в нём начинается с нуля (темп прихода бэкенда тот же). reveal_base_ts = time.monotonic() # Троттлинг edit_text (~раз в интервал, иначе 429) + видимый срез по измеренному темпу прихода (верхняя граница — реально пришедшее). now = time.monotonic() target_full = chunks[-1] if chunks else "" typing_speed = _typing_display_speed(arrival_ewma, pace_key) reveal_len = min(len(target_full), max(0, int((now - (reveal_base_ts or now)) * typing_speed))) current_chunk_text = target_full[:reveal_len] if not edits_dead and now - last_edit_ts >= bot.STREAM_EDIT_MIN_INTERVAL_SEC and current_chunk_text != last_edited_plain: edit_ok = await bot._tg_call(sent_messages[-1].edit_text, current_chunk_text, parse_mode=None, call_timeout=15.0) edit_fail_streak = 0 if edit_ok is not None else edit_fail_streak + 1 edits_dead = edit_fail_streak >= 3 last_edited_plain = current_chunk_text last_edit_ts = now if not full_text.strip(): # Пустой стрим — неудача, но плейсхолдер отдаём вызывающему коду (см. докстринг). return None, sent_messages[-1] # Замеры — ДО довывода (пауза не должна искажать оценку), catch-up паузы в total модели — сознательно (важна видимая задержка). _record_typing_speed(pace_key, time.monotonic() - overall_start_ts, len(full_text)) _record_model_latency( _model_speed_key(provider, model_id), total_sec=time.monotonic() - overall_start_ts, ttf_sec=(first_piece_ts - overall_start_ts) if first_piece_ts is not None else None, ) # "Довывод" остатка: бэкенды одним куском в конце показывали бы "…" до самого финала. Ограничено catchup-лимитом сверху. final_chunks = _split_text_chunks(full_text, bot.TG_MAX_LEN) target_full = final_chunks[-1] already_shown_len = len(last_edited_plain) if last_edited_plain and target_full.startswith(last_edited_plain) else 0 remaining_len = len(target_full) - already_shown_len if remaining_len > 0 and not edits_dead: typing_speed = _typing_display_speed(arrival_ewma, pace_key) for step_len in _typing_catchup_steps(remaining_len, typing_speed, bot.STREAM_TYPING_TICK_SEC, bot.STREAM_TYPING_MAX_CATCHUP_TICKS): await bot._typing_sleep(bot.STREAM_TYPING_TICK_SEC) current_chunk_text = target_full[:already_shown_len + step_len] # Довывод правится тем же троттлингом, что основной цикл: тик # короче интервала правок, иначе серия быстро ловти 429. now = time.monotonic() if current_chunk_text != last_edited_plain and now - last_edit_ts >= bot.STREAM_EDIT_MIN_INTERVAL_SEC: await bot._tg_call(sent_messages[-1].edit_text, current_chunk_text, parse_mode=None, call_timeout=15.0) last_edited_plain = current_chunk_text last_edit_ts = now # Финал — с полной HTML-конвертацией (во время стрима голый текст: частичный markdown дал бы несбалансированные теги). final_text = final_chunks[-1] if edits_dead: # Правки умирали серией — сообщение снесли посреди стрима: весь ответ # досылаем новым сообщением (_send_text внутри режет на части сам), # иначе пользователь его не увидит. await bot._safe_reply(message, full_text.strip()) else: # HTML->plain fallback — внутри _edit_message_quietly (раньше дублировался здесь вручную). await bot._edit_message_quietly(sent_messages[-1], final_text, call_timeout=15.0) except Exception as exc: if not full_text.strip(): # 429 до первого куска — в /stats и ненадолго из роута. Суточный лимит # держим до утра, минутный всплеск — короткая остывка (иначе один 429 # запирал модель до полуночи — враждебное ревью 27.09.2026). try: _txt = bot._error_text(exc).strip() or exc.__class__.__name__ if bot._classify_model_error(bot._error_status(exc, _txt), _txt) == "rate_limit": # Дневную квоту опознаём предикатом провайдера, как текстовый # путь: общий account-wide ловит только OpenRouter-тексты. _daily = bot._is_gemini_daily_quota(_txt) if provider == "gemini" else bot._is_account_wide_or_rate_limit(_txt.lower()) if _daily: bot._mark_quota_exhausted(provider, model_id) else: bot._mark_rate_limited(provider, model_id) else: # 503/404 и прочие "недоступна" — остывка на тех же условиях # (прод 05.10.2026: без неё пять моделей 3.x повторялись на # каждом сообщении, по 22 секунды попытки впустую). from lumen_routes import _mark_model_unavailable _mark_model_unavailable(provider, model_id, exc) except Exception: pass # Плейсхолдер НЕ удаляем — возвращаем для переиспользования (см. докстринг). log.warning('[stream] Stream %s/%s failed before showing any content, falling back to a regular call: %s', provider, model_id, exc) bot._record_stats_event("model_failures") return None, (sent_messages[-1] if sent_messages else None) log.warning('[stream] Stream %s/%s failed after partially showing the response, finishing as-is: %s', provider, model_id, exc) if _detect_identity_leak(full_text): # Последняя страховка (основная проверка — на каждый кусок выше). log.warning('[identity-leak] Leak caught by the fallback guard (%s_stream_exception_path): %r', provider, full_text[:500]) full_text = _IDENTITY_LEAK_FALLBACK with contextlib.suppress(Exception): await bot._tg_call(sent_messages[-1].edit_text, _IDENTITY_LEAK_FALLBACK, parse_mode=None, call_timeout=15.0) elif _detect_injected_payload_echo(full_text): log.warning('[injection-echo] Injected-instruction echo caught by the fallback guard (%s_stream_exception_path): %r', provider, full_text[:500]) full_text = _INJECTED_PAYLOAD_ECHO_FALLBACK with contextlib.suppress(Exception): await bot._tg_call(sent_messages[-1].edit_text, _INJECTED_PAYLOAD_ECHO_FALLBACK, parse_mode=None, call_timeout=15.0) else: with contextlib.suppress(Exception): chunks = _split_text_chunks(full_text, bot.TG_MAX_LEN) final_text = chunks[-1] if chunks else full_text note = bot._t(message.chat.id, "stream_note_interrupted") await bot._tg_call(sent_messages[-1].edit_text, _md_to_html(final_text + note), parse_mode=ParseMode.HTML, call_timeout=15.0) # Частичный ответ уже показан и ляжет в историю ниже — запоминаем, что # его оборвало: слово «продолжи» добьёт с места обрыва (_continue_after_interrupt). state["interrupted"] = True bot.mark_state_dirty(chat_id) finally: aclose = getattr(piece_agen, "aclose", None) if aclose is not None: with contextlib.suppress(Exception): await aclose() # Пустой full_text сюда не доходит — ветки "Empty response" нет. final_answer = _scrub_identity_leak(full_text.strip(), source=f"{provider}_stream_final:{model_id}") bot._record_model_outcome(provider, model_id, bad=_detect_garbled_mix(full_text.strip()) and not _is_garbled_echo(full_text.strip(), user_text)) hist.append({"role": "user", "content": _history_user_text(user_text)}) hist.append({"role": "assistant", "content": final_answer}) # Тот же _trim_history, что в нестриминговых путях (см. комментарий выше). await bot._trim_history(hist) bot._trim_combined(state) bot._record_quota_usage(provider, model_id) return final_answer, None async def _try_gemini_streaming(chat_id: int, user_text: str, message: Message, model_id: str, *, deadline: float | None = None) -> tuple[str | None, Message | None]: """Обёртка _run_streaming_reply для Gemini (строит contents/config).""" import bot conf = GEMINI_MODELS.get(model_id, {}) if not conf.get("stream", True): return None, None contents = await bot._build_gemini_turn_contents(chat_id, user_text) call_contents, gconfig = bot._build_gemini_call_config(model_id, contents) piece_agen = _gemini_stream_pieces(model_id, call_contents, gconfig, deadline=deadline) return await _run_streaming_reply(chat_id, user_text, message, provider="gemini", model_id=model_id, piece_agen=piece_agen, deadline=deadline) async def _try_openrouter_streaming(chat_id: int, user_text: str, message: Message, model_id: str, *, deadline: float | None = None) -> tuple[str | None, Message | None]: """Обёртка _run_streaming_reply для OpenRouter (SSE через chat/completions).""" import bot messages = bot._build_openrouter_turn_messages(chat_id, user_text, model_id) # Генератор — через bot.: тесты подменяют bot._openrouter_stream_pieces фейком. piece_agen = bot._openrouter_stream_pieces(model_id, messages, deadline=deadline) return await _run_streaming_reply(chat_id, user_text, message, provider="openrouter", model_id=model_id, piece_agen=piece_agen, deadline=deadline) async def _try_groq_streaming(chat_id: int, user_text: str, message: Message, model_id: str, *, deadline: float | None = None) -> tuple[str | None, Message | None]: """Обёртка _run_streaming_reply для Groq (SSE через chat/completions). Историю режем тем же _trim_messages_for_groq, что обычный путь, — иначе стрим-проба падает с 413 ещё до цепочки.""" import bot from lumen_routes import _trim_messages_for_groq messages = _trim_messages_for_groq(bot._build_openrouter_turn_messages(chat_id, user_text, model_id)) # Генератор — через bot.: тесты подменяют bot._groq_stream_pieces фейком. piece_agen = bot._groq_stream_pieces(model_id, messages, deadline=deadline) return await _run_streaming_reply(chat_id, user_text, message, provider="groq", model_id=model_id, piece_agen=piece_agen, deadline=deadline)