fix: подтема без end, классификация error_code, 503 провайдера

lecturelog/application/error_classifier.py

@@ -1,5 +1,8 @@
 from __future__ import annotations
 
+import json
+import re
+
 import httpx
 
 from lecturelog.domain.enums import ErrorCode
@@ -12,6 +15,15 @@
 # лимит независимо от конкретного провайдера/клиента.
 _RATE_LIMIT_TOKENS = ("429", "RESOURCE_EXHAUSTED", "503", "UNAVAILABLE")
 
+# Статус ответа провайдера в тексте ошибки openai SDK: «Error code: 401 - {...}».
+# Берём первое вхождение: дальше в теле лежит metadata.previous_errors — история
+# ретраев самого OpenRouter, где встречаются чужие 429/503. По подстрокам такой
+# 401 (отозванный ключ) раньше превращался в rate_limit. Решающим статус считаем
+# только для 4xx, кроме 429: это отказ по самому запросу или ключу, а не
+# перегрузка. 5xx Gemini бывает и 500/504 с UNAVAILABLE в теле — их оставляем
+# подстрокам.
+_PROVIDER_STATUS_RE = re.compile(r"ERROR CODE: (\d{3})")
+
 # Подстроки-сигналы протухших/непринятых YouTube-cookies (текст yt-dlp).
 _COOKIES_INVALID_TOKENS = ("SIGN IN TO CONFIRM", "CONFIRM YOU'RE NOT A BOT")
 
@@ -37,15 +49,22 @@ def classify_error(exc: BaseException) -> ErrorCode:
         }:
             return ErrorCode.BAD_INPUT
         return ErrorCode.INTERNAL
-    # 2) Типовые сигналы битого/нераспознанного входа.
+    # 2) Обрыв JSON в ответе LLM — подкласс ValueError, но вход пользователя
+    # тут ни при чём.
+    if isinstance(exc, json.JSONDecodeError):
+        return ErrorCode.INTERNAL
+    # 3) Типовые сигналы битого/нераспознанного входа.
     if isinstance(exc, (FileNotFoundError, ValueError, InvalidSlidesDocument)):
         return ErrorCode.BAD_INPUT
-    # 3) Текстовый сигнал лимита (LlmClient оборачивает last_error в RuntimeError).
+    # 4) Текстовый сигнал лимита (LlmClient оборачивает last_error в RuntimeError).
     message = str(exc).upper()
     # Сигнал протухших cookies от yt-dlp.
     if any(token in message for token in _COOKIES_INVALID_TOKENS):
         return ErrorCode.COOKIES_INVALID
+    status = _PROVIDER_STATUS_RE.search(message)
+    if status is not None and 400 <= int(status.group(1)) < 500 and status.group(1) != "429":
+        return ErrorCode.INTERNAL
     if any(token in message for token in _RATE_LIMIT_TOKENS):
         return ErrorCode.RATE_LIMIT
-    # 4) Дефолт.
+    # 5) Дефолт.
     return ErrorCode.INTERNAL

lecturelog/infrastructure/llm/llm_client.py

@@ -14,6 +14,7 @@
 import inspect
 import json
 import logging
+import time
 from collections.abc import Awaitable, Callable
 from typing import Any
 
@@ -36,6 +37,10 @@
 # следующая попытка должна уйти на другую модель, но выводить эту из ротации
 # надолго незачем — отказ транзиентный.
 _UPSTREAM_ERROR_COOLDOWN_S = 60.0
+# Сколько один вызов готов пережидать 5xx апстрима. Перегрузка Gemini на проде
+# длилась минутами, а пять попыток с бэк-оффом укладывались в ~30 с. Пока бюджет
+# не исчерпан, попытки на 5xx не расходуют retries.
+_UPSTREAM_BUDGET_S = 300.0
 
 
 async def _emit_usage(on_usage: UsageCallback | None, payload: dict) -> None:
@@ -123,10 +128,14 @@ def __init__(
         cooldown: ModelCooldown,
         *,
         max_tokens: int = _DEFAULT_MAX_TOKENS,
+        upstream_budget_s: float = _UPSTREAM_BUDGET_S,
+        clock: Callable[[], float] | None = None,
     ) -> None:
         self._client = async_openai_client
         self._cooldown = cooldown
         self._max_tokens = max_tokens
+        self._upstream_budget_s = upstream_budget_s
+        self._clock = clock or time.monotonic
 
     async def call(
         self,
@@ -149,9 +158,24 @@ async def call(
             extra_body["reasoning"] = {"effort": effort, "exclude": True}
 
         last_error: Exception | None = None
-        last_reason = "429/RESOURCE_EXHAUSTED"
-        for attempt in range(retries):
+        # Причина попадает в текст итоговой ошибки, а по нему error_classifier
+        # выбирает error_code задачи — поэтому она всегда от последней ошибки.
+        last_reason = "ни одной попытки"
+        tried: list[str] = []
+        upstream_deadline: float | None = None
+        upstream_retry = False
+        attempt = 0
+        while True:
+            if upstream_retry:
+                if self._clock() >= upstream_deadline:
+                    break
+            elif attempt >= retries:
+                break
+            upstream_retry = False
+            attempt += 1
             model = await self._cooldown.acquire(models)
+            if model not in tried:
+                tried.append(model)
             kwargs: dict[str, Any] = {
                 "model": model,
                 "messages": messages,
@@ -177,6 +201,7 @@ async def call(
                 resp = await self._client.chat.completions.create(**kwargs)
             except openai.RateLimitError as error:
                 last_error = error
+                last_reason = "429/RESOURCE_EXHAUSTED"
                 raw = _extract_rate_limit_raw(error)
                 ttl, _kind = parse_cooldown_ttl(
                     raw, seconds_to_midnight=self._cooldown.seconds_to_midnight()
@@ -189,6 +214,7 @@ async def call(
                 if not _is_google_byok_auth_error(error):
                     raise
                 last_error = error
+                last_reason = "401: BYOK-ключ отклонён"
                 logger.warning(
                     "Google BYOK credential rejected for %s; trying next model",
                     model,
@@ -200,30 +226,41 @@ async def call(
                 # с запросом: пока есть другие модели, задача падать не должна.
                 last_error = error
                 last_reason = f"HTTP {getattr(error, 'status_code', '5xx')} апстрима"
+                if upstream_deadline is None:
+                    upstream_deadline = self._clock() + self._upstream_budget_s
+                upstream_retry = True
+                await self._cooldown.mark_rate_limited(model, _UPSTREAM_ERROR_COOLDOWN_S)
+                # Апстрим сам просит подождать. Если остывают все модели, ждём ту,
+                # что освободится раньше, а не бьём в горячую; иначе — короткий
+                # нарастающий бэк-офф перед следующей моделью. Пауза не выходит
+                # за бюджет.
+                wait = await self._cooldown.seconds_until_free(models)
+                if wait <= 0:
+                    wait = _NETWORK_BACKOFF_S * attempt
+                wait = max(0.0, min(wait, upstream_deadline - self._clock()))
                 logger.warning(
-                    "Апстрим недоступен (%s), попытка %d/%d: %s",
+                    "Апстрим недоступен (%s), попытка %d, ждём %.0f с (бюджет до %.0f с): %s",
                     model,
-                    attempt + 1,
-                    retries,
+                    attempt,
+                    wait,
+                    upstream_deadline - self._clock(),
                     error,
                 )
-                await self._cooldown.mark_rate_limited(model, _UPSTREAM_ERROR_COOLDOWN_S)
-                # Апстрим сам просит подождать, поэтому ретрай без паузы сжёг бы
-                # все попытки за доли секунды.
-                await asyncio.sleep(_NETWORK_BACKOFF_S * (attempt + 1))
+                await asyncio.sleep(wait)
                 continue
             except (openai.APITimeoutError, openai.APIConnectionError) as error:
                 # Сетевой флап (не 429): модель не виновата, cooldown не трогаем —
                 # ждём с нарастающим бэк-оффом и пробуем снова.
                 last_error = error
+                last_reason = "сетевые ошибки"
                 logger.warning(
                     "Сетевая ошибка OpenRouter (%s), попытка %d/%d: %s",
                     model,
-                    attempt + 1,
+                    attempt,
                     retries,
                     error,
                 )
-                await asyncio.sleep(_NETWORK_BACKOFF_S * (attempt + 1))
+                await asyncio.sleep(_NETWORK_BACKOFF_S * attempt)
                 continue
 
             # Провайдер может отвалиться до начала генерации: тогда OpenRouter
@@ -275,5 +312,6 @@ async def call(
             return text
 
         raise RuntimeError(
-            f"OpenRouter не дал ответ за {retries} попыток ({last_reason}): {last_error}"
+            f"OpenRouter не дал ответ за {attempt} попыток ({last_reason}), "
+            f"модели: {', '.join(tried)}: {last_error}"
         )

lecturelog/infrastructure/llm/model_cooldown.py

@@ -55,6 +55,12 @@ async def acquire(self, models: list[str]) -> str:
             logger.warning("все модели остывают, беру %s (раньше всех освободится)", soonest)
             return soonest
 
+    async def seconds_until_free(self, models: list[str]) -> float:
+        """Сколько ждать, пока освободится хотя бы одна модель списка (0 — уже есть)."""
+        async with self._lock:
+            now = self._time()
+            return max(0.0, min(self._blocked_until[m] for m in models) - now)
+
     async def mark_rate_limited(self, model: str, ttl: float) -> None:
         async with self._lock:
             self._blocked_until[model] = self._time() + ttl

lecturelog/infrastructure/structurize/gemini_structurizer.py

@@ -44,14 +44,16 @@
 
 
 def _repair_section_timeline(topics_sections: list[list[dict[str, Any]]], srt_content: str) -> None:
-    """Чинит секции, у которых сплиттер выдал end <= start.
+    """Чинит секции, у которых сплиттер выдал end <= start или не выдал end вовсе.
 
     Битая граница обходится дорого: ffmpeg падает на нарезке («-to value smaller
     than -ss») уже после того, как потрачены транскрипция и все LLM-стадии, а
     выравнивание слайдов отбрасывает всю колоду разом (fail-closed по
     «invalid section timeline»). Чиним по соседям — конец битой секции берём из
     начала следующей, а для последней секции из конца транскрипта, — потому что
     содержание секции при этом уже отрендерено и терять его незачем.
+
+    Тот же путь чинит и темы: им достаточно передать список тем одним «разделом».
     """
 
     flat = [section for sections in topics_sections for section in sections]
@@ -61,8 +63,9 @@ def _repair_section_timeline(topics_sections: list[list[dict[str, Any]]], srt_co
     transcript_end = max((block.end_s for block in parse_srt_blocks(srt_content)), default=0.0)
     for index, section in enumerate(flat):
         start = parse_srt_time(str(section["start"]))
-        end = parse_srt_time(str(section["end"]))
-        if end > start:
+        raw_end = section.get("end")
+        end_missing = raw_end in (None, "")
+        if not end_missing and parse_srt_time(str(raw_end)) > start:
             continue
 
         repaired = next(
@@ -76,12 +79,19 @@ def _repair_section_timeline(topics_sections: list[list[dict[str, Any]]], srt_co
             ),
             start + 1.0,
         )
-        logger.warning(
-            "секция %r имеет end <= start (%s → %s); конец исправлен по соседям",
-            section.get("title"),
-            section["start"],
-            section["end"],
-        )
+        if end_missing:
+            logger.warning(
+                "секция %r пришла без end (start %s); конец достроен по соседям",
+                section.get("title"),
+                section["start"],
+            )
+        else:
+            logger.warning(
+                "секция %r имеет end <= start (%s → %s); конец исправлен по соседям",
+                section.get("title"),
+                section["start"],
+                raw_end,
+            )
         section["end"] = format_srt_seconds(repaired)
 
 
@@ -388,6 +398,8 @@ async def structurize(
         topics_data = _parse_json(split_raw)
         if not isinstance(topics_data, list):
             raise ValueError("Ответ split_topics должен быть JSON-массивом")
+        # Конец темы нужен уже для нарезки фрагмента под subsplit.
+        _repair_section_timeline([topics_data], srt_content)
         await _emit_progress(on_progress, 10)
 
         # ── Этап 2: Subsplit (10% → 30%) ───────────────────────────

tests/fixtures/provider_errors/openrouter_401_with_previous_429.txt

@@ -0,0 +1 @@
+Error code: 401 - {'error': {'message': 'Provider returned error', 'code': 401, 'metadata': {'raw': '{\n  "error": {\n    "code": 401,\n    "message": "The bound service account is deleted or disabled. The service account bound to the API key must be active.",\n    "status": "UNAUTHENTICATED",\n    "details": [\n      {\n        "@type": "type.googleapis.com/google.rpc.ErrorInfo",\n        "reason": "ACCOUNT_STATE_INVALID",\n        "metadata": {\n          "method": "google.ai.generativelanguage.v1alpha.GenerativeService.StreamGenerateContent",\n          "service": "generativelanguage.googleapis.com"\n        }\n      }\n    ]\n  }\n}\n', 'provider_name': 'Google AI Studio', 'is_byok': True, 'provider_error_code': '401', 'previous_errors': [{'code': 429, 'message': 'Provider returned error', 'provider_name': 'Google AI Studio', 'raw': '{\n  "error": {\n    "code": 429,\n    "message": "You exceeded your current quota, please check your plan and billing details. For more information on this error, head to: https://ai.google.dev/gemini-api/docs/rate-limits. To monitor your current usage, head to: https://ai.dev/rate-limit. \\n* Quota exceeded for metric: generativelanguage.googleapis.com/generate_content_free_tier_requests, limit: 20, model: gemini-3.6-flash\\nPlease retry in 31.989211488s.",\n    "status": "RESOURCE_EXHAUSTED",\n    "details": [\n      {\n        "@type": "type.googleapis.com/google.rpc.Help",\n        "links": [\n          {\n            "description": "Learn more about Gemini API quotas",\n            "url": "https://ai.google.dev/gemini-api/docs/rate-limits"\n          }\n        ]\n      },\n      {\n        "@type": "type.googleapis.com/google.rpc.QuotaFailure",\n        "violations": [\n          {\n            "quotaMetric": "generativelanguage.googleapis.com/generate_content_free_tier_requests",\n            "quotaId": "GenerateRequestsPerDayPerProjectPerModel-FreeTier",\n            "quotaDimensions": {\n              "location": "global",\n              "model": "gemini-3.6-flash"\n            },\n            "quotaValue": "20"\n          }\n        ]\n      },\n      {\n        "@type": "type.googleapis.com/google.rpc.RetryInfo",\n        "retryDelay": "31s"\n      }\n    ]\n  }\n}\n'}, {'code': 429, 'message': 'Provider returned error', 'provider_name': 'Google AI Studio', 'raw': '{\n  "error": {\n    "code": 429,\n    "message": "You exceeded your current quota, please check your plan and billing details. For more information on this error, head to: https://ai.google.dev/gemini-api/docs/rate-limits. To monitor your current usage, head to: https://ai.dev/rate-limit. \\n* Quota exceeded for metric: generativelanguage.googleapis.com/generate_content_free_tier_requests, limit: 20, model: gemini-3.6-flash\\nPlease retry in 31.687644476s.",\n    "status": "RESOURCE_EXHAUSTED",\n    "details": [\n      {\n        "@type": "type.googleapis.com/google.rpc.Help",\n        "links": [\n          {\n            "description": "Learn more about Gemini API quotas",\n            "url": "https://ai.google.dev/gemini-api/docs/rate-limits"\n          }\n        ]\n      },\n      {\n        "@type": "type.googleapis.com/google.rpc.QuotaFailure",\n        "violations": [\n          {\n            "quotaMetric": "generativelanguage.googleapis.com/generate_content_free_tier_requests",\n            "quotaId": "GenerateRequestsPerDayPerProjectPerModel-FreeTier",\n            "quotaDimensions": {\n              "location": "global",\n              "model": "gemini-3.6-flash"\n            },\n            "quotaValue": "20"\n          }\n        ]\n      },\n      {\n        "@type": "type.googleapis.com/google.rpc.RetryInfo",\n        "retryDelay": "31s"\n      }\n    ]\n  }\n}\n'}, {'code': 429, 'message': 'Provider returned error', 'provider_name': 'Google AI Studio', 'raw': '{\n  "error": {\n    "code": 429,\n    "message": "You exceeded your current quota, please check your plan and billing details. For more information on this error, head to: https://ai.google.dev/gemini-api/docs/rate-limits. To monitor your current usage, head to: https://ai.dev/rate-limit. \\n* Quota exceeded for metric: generativelanguage.googleapis.com/generate_content_free_tier_requests, limit: 20, model: gemini-3.6-flash\\nPlease retry in 31.381486872s.",\n    "status": "RESOURCE_EXHAUSTED",\n    "details": [\n      {\n        "@type": "type.googleapis.com/google.rpc.Help",\n        "links": [\n          {\n            "description": "Learn more about Gemini API quotas",\n            "url": "https://ai.google.dev/gemini-api/docs/rate-limits"\n          }\n        ]\n      },\n      {\n        "@type": "type.googleapis.com/google.rpc.QuotaFailure",\n        "violations": [\n          {\n            "quotaMetric": "generativelanguage.googleapis.com/generate_content_free_tier_requests",\n            "quotaId": "GenerateRequestsPerDayPerProjectPerModel-FreeTier",\n            "quotaDimensions": {\n              "model": "gemini-3.6-flash",\n              "location": "global"\n            },\n            "quotaValue": "20"\n          }\n        ]\n      },\n      {\n        "@type": "type.googleapis.com/google.rpc.RetryInfo",\n        "retryDelay": "31s"\n      }\n    ]\n  }\n}\n'}]}}, 'user_REDACTED': 'user_REDACTED'}