fix(tg): связь с Telegram не встаёт колом, ответ оператора не теряется

Замер прода за сутки 12.09.2026: 576 строк `network error` в логе `tradein-tgbot`
и 7 полных исчерпаний бюджета ретраев, после которых падала итерация poll loop.
Три причины, все подтверждены на коде и в рантайме.

## Ответ оператора мог пропасть навсегда

`process_update` заканчивался безусловным `finally: save_offset(update_id)`.
Замысел верный — «ядовитый» апдейт не должен блокировать поток, — но он не
отличал неисправимый апдейт от транзиентного сетевого отказа. Оператор отвечает
клиенту в топике, `copy_message` падает по сети, `TelegramNetworkError` улетает
в общий `except Exception`, offset сдвигается. Telegram этот апдейт больше не
отдаст, `record_message` не выполнился, оператор уверен, что ответил. Следа нет
нигде, кроме строчки в логе.

Теперь `process_update` возвращает `bool`. На `TelegramNetworkError` делается
`rollback()`, offset НЕ сохраняется, возвращается `False`, и `run_poll_loop`
прерывает разбор пачки — offset у Telegram единая «высшая отметка», подтверждение
любого следующего апдейта неявно подтвердило бы и этот. Остаток пачки Telegram
отдаст заново.

Переигрывания ограничены сверху `_MAX_NETWORK_REPLAYS = 3`: без потолка «вечно
недоставляемый» апдейт заклинил бы очередь навсегда, а это хуже потери одного
сообщения. На потолке offset всё-таки двигается, но с `logger.error` и с
`chat_id`/`message_id`, по которым человек найдёт ответ в топике и перешлёт
руками. Текст переписки в лог по-прежнему не идёт.

Дубли: `TelegramNetworkError` означает исчерпанный бюджет ретраев, при этом
запрос мог дойти до Telegram, а ответ потеряться. Переигрывание тогда доставит
сообщение второй раз. Это осознанный at-least-once компромисс — дубль видят и
клиент, и оператор, а тихая потеря не видна никому. Полная идемпотентность по
паре (update_id, target_chat_id) потребовала бы новой персистентной таблицы ради
редкого случая; вместо неё число дублей жёстко ограничено сверху.

Ветка `except TelegramApiError` с разбором `error_code == 403` («бот заблокирован»)
не тронута — там повтор действительно ничего не изменит.

## Таймаут задавался скаляром, поэтому connect ждал сорок секунд

`httpx.AsyncClient(timeout=effective_timeout)` разворачивается в
connect=read=write=pool. Для `getUpdates` бюджет ответа 40 секунд (30 держит
Telegram плюс запас), и те же 40 секунд уходили на установку соединения — при
живом connect в 0.036 секунды. Худший цикл: четыре попытки по 40 секунд плюс
backoff, около трёх минут, в течение которых бот не видит ответов оператора.
В логе это ровно те разрывы: 06:40:10, 06:42:22, 06:43:35.

Теперь `httpx.Timeout(connect=5, read=<бюджет вызывающего>, write=10, pool=5)`,
значения в именованных константах. Запас `+10s` у `get_updates` относится к read,
докстринг поправлен.

## Клиент создавался заново на каждую попытку

`httpx.AsyncClient` стоял ВНУТРИ цикла ретраев — keep-alive не было вовсе: полный
TCP+TLS-хендшейк на каждый запрос и на каждый повтор, и заново кидался кубик
«встанет ли коннект». Для long-polling это была основная статья сетевых отказов.
Плюс три HTTP-ручки создавали `TelegramClient` на каждый входящий запрос.

Теперь один ленивый переиспользуемый `AsyncClient` на экземпляр, с `aclose()` и
`async with`. Общий клиент приложения живёт в новом `app/services/tgbot/shared.py`,
создаётся и закрывается в lifespan; воркер бота держит свой на время поллинга.
`keepalive_expiry` задан явно: дефолт httpx — 5 секунд, и с ним пул не давал бы
ничего там, где нужнее всего. Poll loop переиспользует соединение и так, а вот
веб-поддержка шлёт раз в минуты и за 5 секунд теряла бы его каждый раз. Плата за
длинный keep-alive — шанс взять из пула закрытое той стороной соединение; httpx
отдаёт это как `RemoteProtocolError`, который ретраится с #3457.

## Уведомления оператору шли с воркерным бюджетом внутри poll loop

Обе отправки в топик («бот заблокирован», «веб-чат не поддерживает медиа») звались
без своего бюджета, то есть с дефолтом в 5 ретраев и backoff до 30 секунд. Одна
такая отправка стопорила весь цикл на минуты, а её отказ решал судьбу апдейта.
Вынесены в `_notify_topic` с узким бюджетом и собственным `except`: провал
вторичного действия больше не отменяет основную ветку.

## Тесты

`tests/services/tgbot/test_shared.py` — новый, на жизненный цикл общего клиента.
В `test_bridge.py` — сетевой отказ оставляет offset нетронутым и апдейт
переигрывается, потолок разблокирует поток, отказ уведомления не отменяет основную
ветку, прежнее поведение на 403 не изменилось. В `test_client.py` — раздельные
таймауты доезжают до httpx per-request, два вызова используют один `AsyncClient`,
`aclose()` его закрывает.

Прогон по затронутым файлам: 127 passed. Ruff check и format чистые.

Прокси намеренно не добавлялся: замер был на восьми запросах, это не статистика,
и решение инфраструктурное. Если обрывы останутся — мерить сотней попыток отдельно.

tradein-mvp/backend/app/api/v1/glitchtip.py

@@ -53,7 +53,8 @@ from fastapi import APIRouter, Header, HTTPException, Query, Request
 from pydantic import BaseModel, ConfigDict, ValidationError
 
 from app.core.config import settings
-from app.services.tgbot.client import TelegramClient, TelegramError
+from app.services.tgbot.client import TelegramError
+from app.services.tgbot.shared import get_telegram_client
 
 logger = logging.getLogger(__name__)
 
@@ -213,7 +214,10 @@ async def glitchtip_webhook(
     received_at = datetime.now(UTC)
     text = _build_message(raw_body, received_at)
 
-    client = TelegramClient(settings.telegram_bot_token)
+    # Общий клиент приложения (#tg-connection-resilience): на каждый запрос
+    # свой создавать нельзя — это ноль keep-alive и полный TCP+TLS-хендшейк
+    # до api.telegram.org перед каждой отправкой. Живёт в lifespan.
+    client = get_telegram_client()
     try:
         await client.send_message(
             chat_id=settings.telegram_alerts_chat_id,

tradein-mvp/backend/app/api/v1/support.py

@@ -73,7 +73,8 @@ from app.core.db import get_db
 from app.core.ratelimit import SlidingWindowLimiter, _client_ip
 from app.services.tgbot import web_support_storage as storage
 from app.services.tgbot.bridge import SERVICE_UNAVAILABLE_TEXT
-from app.services.tgbot.client import TelegramClient, TelegramError
+from app.services.tgbot.client import TelegramError
+from app.services.tgbot.shared import get_telegram_client
 
 logger = logging.getLogger(__name__)
 
@@ -219,7 +220,10 @@ async def send_support_message(
             headers={"Retry-After": str(int(retry_after) + 1)},
         )
 
-    client = TelegramClient(settings.telegram_bot_token)
+    # Общий клиент приложения (#tg-connection-resilience): на каждый запрос
+    # свой создавать нельзя — это ноль keep-alive и полный TCP+TLS-хендшейк
+    # до api.telegram.org перед каждой отправкой. Живёт в lifespan.
+    client = get_telegram_client()
     try:
         mirrored = await client.send_message(
             chat_id=settings.telegram_support_chat_id,
@@ -411,7 +415,10 @@ async def send_anon_support_message(
             )
 
     display_id = _anon_display_id(token)
-    client = TelegramClient(settings.telegram_bot_token)
+    # Общий клиент приложения (#tg-connection-resilience): на каждый запрос
+    # свой создавать нельзя — это ноль keep-alive и полный TCP+TLS-хендшейк
+    # до api.telegram.org перед каждой отправкой. Живёт в lifespan.
+    client = get_telegram_client()
     try:
         mirrored = await client.send_message(
             chat_id=settings.telegram_support_chat_id,

tradein-mvp/backend/app/main.py

@@ -50,6 +50,7 @@ from app.core.rbac import rbac_guard
 from app.core.request_audit import RequestAuditMiddleware
 from app.observability import metrics as app_metrics
 from app.observability.sentry_scrub import scrub_pii_event
+from app.services.tgbot.shared import close_telegram_client, init_telegram_client
 
 logger = logging.getLogger(__name__)
 
@@ -216,7 +217,19 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     # in the tradein-scraper container (`python -m app.scheduler_main`, kit scheduler).
     # Prod backend has always run with SCHEDULER_ENABLE=false (see docker-compose.prod.yml);
     # this API process never actually launched scheduler_loop() in production.
-    yield
+
+    # Общий на приложение Telegram-клиент (#tg-connection-resilience): ручки
+    # support/glitchtip раньше создавали его на КАЖДЫЙ запрос, то есть каждое
+    # зеркалирование сообщения начиналось с полного TCP+TLS-хендшейка до
+    # api.telegram.org. Один пул keep-alive на процесс, закрываем на shutdown.
+    # Без токена не создаём: ручки в этом случае и так отвечают 503.
+    if settings.telegram_bot_token:
+        init_telegram_client()
+
+    try:
+        yield
+    finally:
+        await close_telegram_client()
 
 
 app = FastAPI(

tradein-mvp/backend/app/services/tgbot/bridge.py

@@ -29,9 +29,13 @@
      Telegram 403 (клиент заблокировал бота) → is_blocked=true + уведомление в
      топике (только для Telegram-ветки — у веб-клиента нет "заблокировал бота").
   C) Дедуп: update_id <= сохранённого offset — skip. Offset сохраняется И
-     коммитится в той же транзакции, что и запись сообщения (см. `process_update`
-     `finally`), после КАЖДОГО апдейта — рестарт воркера не переигрывает уже
-     обработанные апдейты и не подвисает вечно на «ядовитом» апдейте.
+     коммитится в той же транзакции, что и запись сообщения (см. `process_update`),
+     после КАЖДОГО апдейта — рестарт воркера не переигрывает уже обработанные
+     апдейты и не подвисает вечно на «ядовитом» апдейте. Исключение —
+     ТРАНЗИЕНТНЫЙ сетевой отказ (`TelegramNetworkError`, т.е. исчерпанный бюджет
+     ретраев клиента): такой апдейт СОЗНАТЕЛЬНО остаётся неподтверждённым, чтобы
+     Telegram отдал его снова, иначе ответ оператора пропадал бы навсегда
+     (#tg-connection-resilience). Потолок переигрываний — `_MAX_NETWORK_REPLAYS`.
   D) TELEGRAM_BOT_TOKEN пуст → бот выключен — проверяется в `app.tgbot_main`
      (entrypoint), не здесь.
   E) /start клиенту → короткое приветствие МЕРЫ, без зеркалирования в топик
@@ -76,7 +80,7 @@ from app.core.config import settings
 from app.core.ratelimit import SlidingWindowLimiter
 from app.core.shutdown import shutdown_requested
 from app.services.tgbot import web_support_storage
-from app.services.tgbot.client import TelegramApiError, TelegramClient
+from app.services.tgbot.client import TelegramApiError, TelegramClient, TelegramNetworkError
 
 logger = logging.getLogger(__name__)
 
@@ -131,6 +135,44 @@ FLOOD_LIMITED_TEXT = (
 # получает это уведомление в топике вместо тихого игнора (иначе уверен, что ответил).
 _WEB_UNSUPPORTED_MEDIA_REPLY_TEXT = "Веб-чат поддерживает только текст, сообщение не доставлено."
 
+# Уведомления оператору в топике отправляются ИЗНУТРИ poll loop, который
+# однопоточный: пока висит одна отправка, не обрабатывается НИ ОДИН следующий
+# апдейт. Поэтому им нужен интерактивный бюджет, а не воркерный дефолт клиента
+# (5 ретраев, backoff до 30с, полный retry_after на 429 — минуты стопа на
+# ВТОРИЧНОМ действии). Числа — те же, что у интерактивных отправок веб-чата
+# поддержки (`_INTERACTIVE_SEND_*` в app/api/v1/support.py, подобраны замером
+# прода #tgsupport-retry); сознательно ДУБЛИРУЕМ, а не импортируем из слоя API —
+# воркер не должен зависеть от роутера.
+_NOTIFY_SEND_TIMEOUT_S = 5.0
+_NOTIFY_SEND_MAX_RETRIES = 3
+_NOTIFY_SEND_MAX_BACKOFF_S = 1.0
+
+# Потолок переигрываний ОДНОГО update_id на транзиентных сетевых отказах
+# (#tg-connection-resilience).
+#
+# Зачем потолок: без него «вечно недоставляемый» апдейт заклинил бы очередь
+# НАВСЕГДА — ровно то, от чего защищал прежний безусловный `finally: save_offset`.
+# Потерять одно сообщение плохо, потерять весь поток — хуже, поэтому на потолке
+# offset всё-таки сдвигается, но ГРОМКО (`logger.error`), а не молча.
+#
+# Про дубли: `TelegramNetworkError` означает исчерпанный бюджет ретраев клиента,
+# при этом запрос МОГ дойти до Telegram (потерялся ответ) — переигрывание тогда
+# доставит клиенту то же сообщение второй раз. Это осознанный at-least-once
+# компромисс: дубль и клиент, и оператор ВИДЯТ и могут поправить, а тихая потеря
+# ответа не оставляет следа нигде, кроме строчки в логе. Полноценная
+# идемпотентность по (update_id, target_chat_id) потребовала бы нового
+# персистентного состояния (колонка/таблица + миграция) ради редкого случая;
+# вместо этого число возможных дублей жёстко ограничено сверху — не больше
+# (_MAX_NETWORK_REPLAYS - 1) повторов на апдейт.
+_MAX_NETWORK_REPLAYS = 3
+
+# update_id -> сколько раз мы уже отказались подтверждать этот апдейт.
+# In-memory осознанно: воркер long-polling однопоточный, запись живёт ровно до
+# подтверждения апдейта (`pop` в `process_update`), так что словарь не растёт.
+# Рестарт воркера обнуляет счётчик — это допустимо (новый процесс = новая сеть),
+# потолок всё равно действует в пределах каждой жизни процесса.
+_network_replay_attempts: dict[int, int] = {}
+
 
 # ── Storage abstraction (testable без реальной БД) ──────────────────────────
 class BridgeStorage(Protocol):
@@ -400,6 +442,43 @@ def _format_topic_header(
     return f"Новое обращение от {display_name}{username_part} (chat_id={chat_id})"
 
 
+async def _notify_topic(
+    client: TelegramClient,
+    *,
+    text: str,
+    reply_to_message_id: int | None,
+    context: str,
+) -> None:
+    """Служебное уведомление оператору в support-топик (вторичное действие).
+
+    Два свойства, которых не было у прямых `client.send_message` вызовов:
+    1) узкий интерактивный бюджет (`_NOTIFY_SEND_*`) — иначе одна такая отправка
+       стопорит весь однопоточный poll loop на минуты;
+    2) собственный `except` — провал УВЕДОМЛЕНИЯ не отменяет основную ветку
+       обработки (клиент уже помечен заблокированным / медиа-реплай уже отклонён)
+       и не решает судьбу апдейта.
+    `context` — только технические идентификаторы (chat_id/thread_id), НЕ текст
+    переписки: логи моста принципиально не содержат ПДн.
+    """
+    try:
+        await client.send_message(
+            chat_id=settings.telegram_support_chat_id,
+            text=text,
+            message_thread_id=settings.telegram_support_topic_id or None,
+            reply_to_message_id=reply_to_message_id,
+            timeout=_NOTIFY_SEND_TIMEOUT_S,
+            max_retries=_NOTIFY_SEND_MAX_RETRIES,
+            max_backoff=_NOTIFY_SEND_MAX_BACKOFF_S,
+        )
+    except Exception:
+        logger.warning(
+            "tgbot bridge: не удалось отправить уведомление оператору в топик (%s) — "
+            "основная ветка обработки не отменяется",
+            context,
+            exc_info=True,
+        )
+
+
 # ── Update routing ────────────────────────────────────────────────────────────
 async def _handle_private_message(
     message: dict[str, Any], client: TelegramClient, storage: BridgeStorage
@@ -553,14 +632,14 @@ async def _handle_group_reply(
             if exc.error_code == 403:
                 # Клиент заблокировал бота — фиксируем и уведомляем оператора в топике.
                 storage.mark_blocked(target_chat_id)
-                await client.send_message(
-                    chat_id=settings.telegram_support_chat_id,
+                await _notify_topic(
+                    client,
                     text=(
                         f"Не удалось доставить сообщение клиенту (chat_id={target_chat_id}) — "
                         "бот заблокирован."
                     ),
-                    message_thread_id=settings.telegram_support_topic_id or None,
                     reply_to_message_id=message_id,
+                    context=f"403 на доставке клиенту chat_id={target_chat_id}",
                 )
                 return
             raise
@@ -592,11 +671,11 @@ async def _handle_group_reply(
                 web_thread_id,
                 kind,
             )
-            await client.send_message(
-                chat_id=settings.telegram_support_chat_id,
+            await _notify_topic(
+                client,
                 text=_WEB_UNSUPPORTED_MEDIA_REPLY_TEXT,
-                message_thread_id=settings.telegram_support_topic_id or None,
                 reply_to_message_id=message_id if isinstance(message_id, int) else None,
+                context=f"медиа-реплай на веб-зеркало thread_id={web_thread_id}",
             )
             return
 
@@ -636,25 +715,41 @@ async def _handle_group_reply(
 
 async def process_update(
     update: dict[str, Any], client: TelegramClient, storage: BridgeStorage
-) -> None:
+) -> bool:
     """Маршрутизирует один Telegram update. Дедуп (C) + атомарный offset-commit.
 
-    Дедуп: update_id <= сохранённого offset — skip без side-effects. Offset
-    сохраняется и коммитится ПОСЛЕ обработки (в т.ч. если обработка упала —
-    иначе «ядовитый» апдейт блокировал бы весь поток навсегда).
+    Возвращает True, если offset сдвинут (апдейт подтверждён, Telegram его больше
+    не отдаст), и False, если апдейт СОЗНАТЕЛЬНО оставлен неподтверждённым ради
+    переигрывания. На False вызывающий (`run_poll_loop`) ОБЯЗАН прервать разбор
+    пачки: offset у Telegram — единая «высшая отметка», подтверждение любого
+    СЛЕДУЮЩЕГО апдейта неявно подтвердило бы и этот, и переигрывания не было бы.
 
-    Различаем сбой БД (`SQLAlchemyError`) от прочих (Telegram API и т.п.):
-    сбой БД оставляет сессию в failed-transaction state — `rollback()` ОБЯЗАН
-    отработать ПЕРЕД `save_offset`, иначе тот сам кинет `PendingRollbackError`,
-    `process_update` вылетит без сохранения offset'а, следующая итерация
-    `run_poll_loop` получит СТАРЫЙ offset от `get_offset()` и переиграет тот же
-    апдейт заново — copyMessage задублирует зеркало клиента в топике на
-    каждый повтор поллинга (#3 review, воспроизведено).
+    Дедуп: update_id <= сохранённого offset — skip без side-effects.
+
+    Судьба offset'а по классам отказа:
+      - `TelegramNetworkError` (транзиентный: Telegram не ответил, бюджет ретраев
+        клиента исчерпан) — offset НЕ двигаем, `rollback()` частичных записей,
+        апдейт переигрывается на следующей итерации. Иначе ответ оператора
+        терялся НАВСЕГДА: copyMessage не дошёл, `record_message` не выполнился,
+        Telegram апдейт больше не отдаст, а оператор уверен, что ответил
+        (#tg-connection-resilience). Ограничено `_MAX_NETWORK_REPLAYS` — на
+        потолке offset всё-таки сдвигается с `logger.error`, иначе «вечно
+        недоставляемый» апдейт заклинил бы поток навсегда.
+      - `SQLAlchemyError` (сбой БД) — offset двигаем, но `rollback()` ОБЯЗАН
+        отработать ПЕРЕД `save_offset`: сбой БД оставляет сессию в
+        failed-transaction state, иначе `save_offset` сам кинет
+        `PendingRollbackError`, `process_update` вылетит без сохранения offset'а,
+        следующая итерация получит СТАРЫЙ offset от `get_offset()` и переиграет
+        тот же апдейт — copyMessage задублирует зеркало клиента в топике на
+        каждый повтор поллинга (#3 review, воспроизведено).
+      - любое прочее исключение (в т.ч. `TelegramApiError` — площадка ОТВЕТИЛА
+        отказом, повтор ничего не изменит) — offset двигаем, «ядовитый» апдейт
+        не блокирует поток.
     """
     update_id = update.get("update_id")
     if not isinstance(update_id, int):
         logger.warning("tgbot bridge: update без валидного update_id — игнор")
-        return
+        return True
 
     current_offset = storage.get_offset()
     if update_id <= current_offset:
@@ -663,7 +758,7 @@ async def process_update(
             update_id,
             current_offset,
         )
-        return
+        return True
 
     message = update.get("message")
     try:
@@ -677,6 +772,35 @@ async def process_update(
                 await _handle_group_reply(message, client, storage)
             # иначе — необрабатываемый тип чата/апдейта (edited_message, канал и
             # т.п.) — тихий игнор, но offset всё равно сдвигаем ниже.
+    except TelegramNetworkError:
+        attempts = _network_replay_attempts.get(update_id, 0) + 1
+        # Частичные записи этого апдейта не должны уехать в БД чужим commit'ом
+        # (сессия одна на всю пачку) — переигрывание начинается с чистого листа.
+        storage.rollback()
+        if attempts < _MAX_NETWORK_REPLAYS:
+            _network_replay_attempts[update_id] = attempts
+            logger.warning(
+                "tgbot bridge: Telegram недоступен на update_id=%d (отказ %d из %d) — "
+                "offset НЕ сдвигаем, апдейт переиграется на следующей итерации",
+                update_id,
+                attempts,
+                _MAX_NETWORK_REPLAYS,
+            )
+            return False
+        logger.error(
+            "tgbot bridge: update_id=%d исчерпал потолок переигрываний (%d сетевых "
+            "отказов подряд) — сдвигаем offset, содержимое апдейта ПОТЕРЯНО; поток не "
+            "блокируем, требуется ручной разбор support-топика "
+            "(chat_id=%s, message_id=%s)",
+            update_id,
+            _MAX_NETWORK_REPLAYS,
+            # Идентификаторы, а НЕ текст: это единственная строка, по которой
+            # человек найдёт потерянный ответ оператора в топике и перешлёт его
+            # руками. Без них в логе остаётся только update_id, которого в
+            # интерфейсе Telegram не видно. Текст сообщения — ПДн, в лог не идёт.
+            (message or {}).get("chat", {}).get("id") if isinstance(message, dict) else None,
+            (message or {}).get("message_id") if isinstance(message, dict) else None,
+        )
     except SQLAlchemyError:
         logger.exception(
             "tgbot bridge: DB-ошибка на update_id=%d — rollback перед сохранением "
@@ -690,9 +814,12 @@ async def process_update(
             "(не блокируем поток на 'ядовитом' апдейте)",
             update_id,
         )
-    finally:
-        storage.save_offset(update_id)
-        storage.commit()
+
+    # Апдейт подтверждён — счётчик переигрываний больше не нужен (словарь не растёт).
+    _network_replay_attempts.pop(update_id, None)
+    storage.save_offset(update_id)
+    storage.commit()
+    return True
 
 
 # ── Long-polling loop ─────────────────────────────────────────────────────────
@@ -717,8 +844,20 @@ async def run_poll_loop(
                 updates = await client.get_updates(
                     offset=offset + 1, timeout=poll_timeout_s, allowed_updates=["message"]
                 )
-                for update in updates:
-                    await process_update(update, client, storage)
+                for idx, update in enumerate(updates):
+                    if not await process_update(update, client, storage):
+                        # Апдейт намеренно не подтверждён (транзиентный сетевой
+                        # отказ). Обрабатывать остаток пачки НЕЛЬЗЯ: offset —
+                        # единая «высшая отметка», подтверждение следующего
+                        # апдейта неявно подтвердило бы и этот. Остаток Telegram
+                        # отдаст заново на следующей итерации.
+                        logger.warning(
+                            "tgbot bridge: update_id=%s не подтверждён — остаток пачки "
+                            "(%d апдейтов) разберём на следующей итерации",
+                            update.get("update_id"),
+                            len(updates) - idx - 1,
+                        )
+                        break
             consecutive_errors = 0
         except Exception:
             consecutive_errors += 1

tradein-mvp/backend/app/services/tgbot/client.py

@@ -54,6 +54,38 @@ _DEFAULT_RETRY_AFTER_S = 5.0
 _MAX_BACKOFF_S = 30.0
 _DEFAULT_MAX_RETRIES = 5
 
+# Раздельные таймауты вместо скаляра. httpx разворачивает скаляр в
+# connect=read=write=pool, поэтому long-poll `getUpdates` (read = 30с, которые
+# Telegram держит запрос, + 10с запаса = 40с) ставил 40 секунд и на УСТАНОВКУ
+# соединения. Живой connect до api.telegram.org из прод-контейнера занимает
+# 0.036с — 40-секундное ожидание коннекта было чистой слепотой: худший цикл
+# 4 попытки × 40с + backoff ≈ 174с, и всё это время бот не видит ответов
+# оператора (замер 12.09.2026: разрывы в логе 06:40:10 → 06:42:22 → 06:43:35,
+# 576 строк `network error` и 7 полных исчерпаний бюджета ретраев за сутки).
+# connect/write/pool к ожиданию ОТВЕТА Telegram отношения не имеют и коротки.
+_CONNECT_TIMEOUT_S = 5.0
+_WRITE_TIMEOUT_S = 10.0
+_POOL_TIMEOUT_S = 5.0
+
+# Пул keep-alive соединений на ОДИН экземпляр клиента. Параллелизма тут почти
+# нет (long-polling — один запрос за раз, интерактивные ручки — единицы в
+# минуту), так что смысл пула не в ширине, а в том, чтобы TCP+TLS-хендшейк не
+# повторялся на каждый запрос и каждый ретрай.
+_MAX_KEEPALIVE_CONNECTIONS = 5
+_MAX_CONNECTIONS = 10
+
+# Сколько держать простаивающее соединение. Задаём ЯВНО, потому что дефолт
+# httpx — 5 секунд, и с ним пул не давал бы ничего там, где он нужнее всего:
+# poll loop переиспользует соединение (следующий getUpdates уходит сразу), а
+# вот веб-поддержка шлёт сообщения раз в минуты — за 5с соединение протухает и
+# каждое зеркало снова платит полный TCP+TLS.
+#
+# Плата за длинный keep-alive — возросший шанс взять из пула соединение, которое
+# уже закрыла та сторона; httpx отдаёт это как `RemoteProtocolError` («Server
+# disconnected without sending a response»). Он ретраится с #3457, так что
+# сценарий закрыт: попытка на протухшем соединении стоит один повтор, а не отказ.
+_KEEPALIVE_EXPIRY_S = 90.0
+
 
 class TelegramError(Exception):
     """Общий предок отказов клиента: и «ответил ok: false», и «не ответил вовсе».
@@ -94,6 +126,21 @@ class TelegramNetworkError(TelegramError):
         super().__init__(f"Telegram {method} unreachable after {attempts} attempts: {reason}")
 
 
+def _request_timeout(read: float) -> httpx.Timeout:
+    """Разворачивает «сколько ждать ответа» (скаляр вызывающего) в таймауты httpx.
+
+    `read` — запрошенный бюджет ОТВЕТА (для long-poll это `poll_timeout + 10s`);
+    connect/write/pool фиксированы модульными константами и коротки: ждать
+    ответа Telegram — не то же самое, что ждать установки соединения.
+    """
+    return httpx.Timeout(
+        connect=_CONNECT_TIMEOUT_S,
+        read=read,
+        write=_WRITE_TIMEOUT_S,
+        pool=_POOL_TIMEOUT_S,
+    )
+
+
 def _extract_retry_after(
     response: httpx.Response, default: float = _DEFAULT_RETRY_AFTER_S
 ) -> float:
@@ -127,7 +174,24 @@ def _error_from_body(response: httpx.Response) -> tuple[int, str]:
 
 
 class TelegramClient:
-    """Bot API клиент на httpx.AsyncClient. Каждый вызов — отдельное короткоживущее соединение."""
+    """Bot API клиент поверх ОДНОГО долгоживущего `httpx.AsyncClient`.
+
+    Соединение переиспользуется всё время жизни экземпляра: `AsyncClient`
+    создаётся лениво при первом запросе и хранится в `self._http`. Раньше он
+    создавался ВНУТРИ цикла ретраев — то есть keep-alive не было вовсе: полный
+    TCP+TLS-хендшейк на каждый запрос и на каждую повторную попытку, и заново
+    кидался кубик «встанет ли коннект». Для long-polling'а, ходящего каждые
+    ~30с в бесконечном цикле, это была основная статья сетевых отказов.
+
+    Отсюда — требование к вызывающим: экземпляр НАДО переиспользовать (один на
+    процесс воркера, один на FastAPI-приложение — см.
+    `app.services.tgbot.shared`), а не создавать на каждый запрос, и закрывать
+    через `aclose()` или `async with`.
+
+    Таймаут теперь per-request: у `AsyncClient` он стоит дефолтом, а каждый
+    вызов `_request` передаёт свой `httpx.Timeout` (long-poll — свои 40с на
+    read, интерактивные ручки — свой узкий бюджет).
+    """
 
     def __init__(
         self,
@@ -137,6 +201,32 @@ class TelegramClient:
     ) -> None:
         self._base = f"{base_url}/bot{token}"
         self._timeout = timeout
+        self._http: httpx.AsyncClient | None = None
+
+    def _http_client(self) -> httpx.AsyncClient:
+        """Ленивое создание переиспользуемого AsyncClient (вне цикла ретраев)."""
+        if self._http is None:
+            self._http = httpx.AsyncClient(
+                timeout=_request_timeout(self._timeout),
+                limits=httpx.Limits(
+                    max_keepalive_connections=_MAX_KEEPALIVE_CONNECTIONS,
+                    max_connections=_MAX_CONNECTIONS,
+                    keepalive_expiry=_KEEPALIVE_EXPIRY_S,
+                ),
+            )
+        return self._http
+
+    async def aclose(self) -> None:
+        """Закрывает пул соединений. Идемпотентно; после — клиент снова ленив."""
+        http, self._http = self._http, None
+        if http is not None:
+            await http.aclose()
+
+    async def __aenter__(self) -> TelegramClient:
+        return self
+
+    async def __aexit__(self, *_exc: object) -> None:
+        await self.aclose()
 
     async def _request(
         self,
@@ -164,14 +254,19 @@ class TelegramClient:
         """
         url = f"{self._base}/{method}"
         effective_timeout = timeout if timeout is not None else self._timeout
+        # Раздельные таймауты считаем ОДИН раз и передаём per-request: у общего
+        # AsyncClient свой дефолт, а бюджет ответа у каждого вызова свой.
+        request_timeout = _request_timeout(effective_timeout)
         backoff_cap = _MAX_BACKOFF_S if max_backoff is None else max_backoff
         attempt = 0
+        # Клиент берём ДО цикла: пересоздавать его на каждую попытку значило бы
+        # заново платить за TCP+TLS ровно там, где сеть уже показала себя плохо.
+        client = self._http_client()
 
         while True:
             attempt += 1
             try:
-                async with httpx.AsyncClient(timeout=effective_timeout) as client:
-                    response = await client.post(url, json=payload)
+                response = await client.post(url, json=payload, timeout=request_timeout)
             except httpx.TransportError as exc:
                 # Ловим ВЕСЬ `TransportError`, а не узкий кортеж
                 # `(TimeoutException, NetworkError)`: `RemoteProtocolError`
@@ -306,8 +401,11 @@ class TelegramClient:
     ) -> list[dict[str, Any]]:
         """Long-polling getUpdates. `timeout` — сколько Telegram держит запрос открытым (сек).
 
-        HTTP-таймаут запроса берётся с запасом (`timeout + 10s`), чтобы не обрывать
-        соединение раньше, чем ответит сам Telegram long-poll.
+        Запас `+10s` относится к READ-таймауту (сколько ждём ответа), чтобы не
+        обрывать соединение раньше, чем ответит сам Telegram long-poll. На
+        connect/write/pool он НЕ распространяется — они короткие и фиксированы
+        (`_CONNECT_TIMEOUT_S` и соседи): установка соединения либо занимает
+        десятки миллисекунд, либо не состоится вовсе.
         """
         payload: dict[str, Any] = {"offset": offset, "timeout": timeout}
         if allowed_updates is not None:

tradein-mvp/backend/app/services/tgbot/shared.py

@@ -0,0 +1,50 @@
+"""Общий на приложение `TelegramClient` (#tg-connection-resilience).
+
+Зачем: до этого три HTTP-ручки (`api.v1.support` ×2, `api.v1.glitchtip`) делали
+`TelegramClient(token)` на КАЖДЫЙ входящий запрос, а клиент внутри пересоздавал
+`httpx.AsyncClient` на каждую попытку — то есть keep-alive не было ни на каком
+уровне и каждый запрос начинался с полного TCP+TLS-хендшейка до
+api.telegram.org. Здесь живёт один экземпляр на процесс: создаётся в lifespan
+(`app.main`), закрывается на shutdown там же.
+
+Воркер бота (`app.tgbot_main`) сюда НЕ ходит — у него свой процесс без ASGI и
+свой экземпляр на всё время жизни поллинга.
+"""
+
+from __future__ import annotations
+
+import logging
+
+from app.core.config import settings
+from app.services.tgbot.client import TelegramClient
+
+logger = logging.getLogger(__name__)
+
+_client: TelegramClient | None = None
+
+
+def get_telegram_client() -> TelegramClient:
+    """Общий клиент приложения. Ленив: создаётся при первом обращении.
+
+    Ленивость (а не «только из lifespan») нужна из-за kill-switch: при пустом
+    `TELEGRAM_BOT_TOKEN` в lifespan создавать нечего, а тесты ручек поднимают
+    приложение без прохода через startup.
+    """
+    global _client
+    if _client is None:
+        _client = TelegramClient(settings.telegram_bot_token)
+    return _client
+
+
+def init_telegram_client() -> TelegramClient:
+    """Явное создание на старте приложения (lifespan)."""
+    return get_telegram_client()
+
+
+async def close_telegram_client() -> None:
+    """Закрывает общий клиент на shutdown. Идемпотентно."""
+    global _client
+    client, _client = _client, None
+    if client is not None:
+        await client.aclose()
+        logger.info("tg shared client: пул соединений закрыт")

tradein-mvp/backend/app/tgbot_main.py

@@ -114,8 +114,12 @@ def _should_run() -> bool:
 
 
 async def _run_bridge() -> None:
-    client = TelegramClient(settings.telegram_bot_token)
-    await run_poll_loop(client, SessionLocal)
+    # `async with` — чтобы пул keep-alive соединений закрывался при любом выходе
+    # из поллинга (кооперативный drain по SIGTERM, hard-cancel, исключение).
+    # Клиент один на весь процесс: пересоздание на запрос убивало keep-alive и
+    # заставляло каждый long-poll начинаться с TCP+TLS-хендшейка.
+    async with TelegramClient(settings.telegram_bot_token) as client:
+        await run_poll_loop(client, SessionLocal)
 
 
 async def _await_bridge(task: asyncio.Task[None]) -> None:

tradein-mvp/backend/tests/services/tgbot/test_bridge.py

@@ -22,6 +22,12 @@ Coverage (per task spec + review follow-up):
 - сбой БД (SQLAlchemyError) во время обработки → rollback() ПЕРЕД save_offset,
   offset всё равно сдвигается — без этого следующий поллинг переиграл бы тот
   же апдейт и задублировал зеркало клиента в топике (#3 review)
+- ТРАНЗИЕНТНЫЙ сетевой отказ (TelegramNetworkError) на доставке ответа оператора
+  → offset НЕ сдвигается, апдейт переигрывается и доходит до клиента; на потолке
+  `_MAX_NETWORK_REPLAYS` offset всё-таки сдвигается (поток не заклинен); poll loop
+  прерывает разбор пачки на неподтверждённом апдейте (#tg-connection-resilience)
+- провал ВТОРИЧНОГО уведомления оператору в топик не отменяет основную ветку
+  (is_blocked остаётся, offset сдвигается) и логируется отдельной строкой
 - /start → приветствие без зеркалирования
 - Telegram 403 на доставку оператору → is_blocked + уведомление в топике
 - TELEGRAM_SUPPORT_CHAT_ID не задан → клиенту уходит "сервис недоступен"
@@ -45,7 +51,7 @@ from sqlalchemy.exc import SQLAlchemyError
 os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
 
 from app.services.tgbot import bridge
-from app.services.tgbot.client import TelegramClient
+from app.services.tgbot.client import TelegramClient, TelegramNetworkError
 
 SUPPORT_CHAT_ID = -100123456789
 SUPPORT_TOPIC_ID = 42
@@ -973,3 +979,219 @@ async def test_update_from_unrelated_chat_is_ignored_but_offset_advances() -> No
     assert calls == []
     assert storage.messages == []
     assert storage.get_offset() == 40
+
+
+# ── сетевая устойчивость (#tg-connection-resilience) ─────────────────────────
+
+
+@pytest.fixture(autouse=True)
+def _reset_network_replay_attempts() -> None:
+    """`bridge._network_replay_attempts` — module-level словарь, его состояние
+    иначе протекало бы между тестами (потолок переигрываний виден глобально)."""
+    bridge._network_replay_attempts.clear()
+
+
+def _network_boom(method: str = "copyMessage"):
+    """Асинхронная заглушка метода клиента, изображающая исчерпанный бюджет ретраев."""
+
+    async def _raise(*_args: object, **_kwargs: object) -> dict[str, Any]:
+        raise TelegramNetworkError(method, "ConnectTimeout", 4)
+
+    return _raise
+
+
+async def test_network_failure_on_operator_reply_keeps_offset_for_replay(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Главный сценарий потери: оператор ответил, Telegram в этот момент недоступен.
+
+    Раньше `finally: save_offset` подтверждал апдейт — ответ не доходил до клиента
+    НИКОГДА (Telegram апдейт больше не отдаёт, записи нет, оператор уверен, что
+    ответил). Теперь offset остаётся прежним, апдейт переигрывается и доходит.
+    """
+    calls: list[tuple[str, dict[str, Any]]] = []
+    client = _make_client({"copyMessage": {"message_id": 42}}, calls)
+    storage = FakeBridgeStorage()
+    storage.record_message(
+        chat_id=555,
+        direction="in",
+        tg_message_id=1,
+        topic_message_id=100,
+        kind="text",
+        text_body="вопрос клиента",
+        operator_tg_id=None,
+    )
+
+    update = {"update_id": 40, "message": _group_reply_message(reply_to_message_id=100)}
+
+    # Падает РОВНО первый copyMessage (monkeypatch.undo() тут не годится: он снял бы
+    # и autouse-патчи настроек support-группы, и ветка просто перестала бы работать).
+    real_copy = client.copy_message
+    failures = {"left": 1}
+
+    async def flaky_copy(**kwargs: Any) -> dict[str, Any]:
+        if failures["left"] > 0:
+            failures["left"] -= 1
+            raise TelegramNetworkError("copyMessage", "ConnectTimeout", 4)
+        return await real_copy(**kwargs)
+
+    monkeypatch.setattr(client, "copy_message", flaky_copy)
+    advanced = await bridge.process_update(update, client, storage)
+
+    assert advanced is False
+    assert storage.get_offset() == 0  # апдейт НЕ подтверждён — Telegram отдаст его снова
+    assert storage.commits == 0
+    assert storage.rollbacks == 1  # частичные записи не уедут чужим commit'ом
+    assert len(storage.messages) == 1  # фейковой записи 'out' не появилось
+
+    # Переигрывание: сеть починилась — тот же апдейт доставляется и подтверждается.
+    advanced = await bridge.process_update(update, client, storage)
+
+    assert advanced is True
+    assert storage.get_offset() == 40
+    assert [m for m, _ in calls] == ["copyMessage"]
+    out_rec = storage.messages[-1]
+    assert out_rec["direction"] == "out"
+    assert out_rec["chat_id"] == 555
+
+
+async def test_network_failure_stops_advancing_only_until_replay_cap(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Потолок переигрываний: «вечно недоставляемый» апдейт не должен заклинить
+    поток навсегда (ровно то, от чего защищал прежний безусловный `finally`)."""
+    calls: list[tuple[str, dict[str, Any]]] = []
+    client = _make_client({}, calls)
+    storage = FakeBridgeStorage()
+    storage.record_message(
+        chat_id=555,
+        direction="in",
+        tg_message_id=1,
+        topic_message_id=100,
+        kind="text",
+        text_body="вопрос клиента",
+        operator_tg_id=None,
+    )
+    monkeypatch.setattr(client, "copy_message", _network_boom())
+
+    update = {"update_id": 41, "message": _group_reply_message(reply_to_message_id=100)}
+
+    for _ in range(bridge._MAX_NETWORK_REPLAYS - 1):
+        assert await bridge.process_update(update, client, storage) is False
+        assert storage.get_offset() == 0
+
+    with caplog.at_level(logging.ERROR, logger="app.services.tgbot.bridge"):
+        advanced = await bridge.process_update(update, client, storage)
+
+    assert advanced is True
+    assert storage.get_offset() == 41  # поток разблокирован
+    assert "потолок переигрываний" in caplog.text
+    # Счётчик снят — словарь не растёт от апдейта к апдейту.
+    assert bridge._network_replay_attempts == {}
+
+
+async def test_poll_loop_stops_batch_on_unconfirmed_update(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Offset у Telegram — единая «высшая отметка»: подтвердив СЛЕДУЮЩИЙ апдейт
+    пачки, мы неявно подтвердили бы неудавшийся, и переигрывания не случилось бы.
+    Поэтому разбор пачки обрывается на первом неподтверждённом апдейте."""
+    processed: list[int] = []
+
+    async def fake_process(update: dict[str, Any], client: object, storage: object) -> bool:
+        processed.append(update["update_id"])
+        return update["update_id"] != 51  # 51 — сетевой отказ
+
+    monkeypatch.setattr(bridge, "process_update", fake_process)
+
+    iteration = {"n": 0}
+
+    def fake_shutdown() -> bool:
+        iteration["n"] += 1
+        return iteration["n"] > 1  # ровно одна итерация poll loop
+
+    monkeypatch.setattr(bridge, "shutdown_requested", fake_shutdown)
+    monkeypatch.setattr(bridge, "SqlBridgeStorage", lambda _db: FakeBridgeStorage())
+
+    class _FakeSession:
+        def __enter__(self) -> _FakeSession:
+            return self
+
+        def __exit__(self, *_exc: object) -> bool:
+            return False
+
+    class _FakeClient:
+        async def get_updates(self, **_kwargs: object) -> list[dict[str, Any]]:
+            return [{"update_id": 51}, {"update_id": 52}, {"update_id": 53}]
+
+    await bridge.run_poll_loop(_FakeClient(), _FakeSession, poll_timeout_s=1)
+
+    assert processed == [51]  # 52/53 придут заново следующим getUpdates
+
+
+async def test_topic_notification_failure_does_not_cancel_main_branch(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Уведомление в топик — вторичное действие: его сетевой отказ не отменяет
+    пометку is_blocked и не превращает апдейт в переигрываемый."""
+    calls: list[tuple[str, dict[str, Any]]] = []
+    client = _make_client({"copyMessage": 403}, calls)
+    storage = FakeBridgeStorage()
+    storage.record_message(
+        chat_id=555,
+        direction="in",
+        tg_message_id=1,
+        topic_message_id=100,
+        kind="text",
+        text_body="вопрос клиента",
+        operator_tg_id=None,
+    )
+    monkeypatch.setattr(client, "send_message", _network_boom("sendMessage"))
+
+    update = {"update_id": 42, "message": _group_reply_message(reply_to_message_id=100)}
+    with caplog.at_level(logging.WARNING, logger="app.services.tgbot.bridge"):
+        advanced = await bridge.process_update(update, client, storage)
+
+    assert advanced is True
+    assert 555 in storage.blocked  # основная ветка отработала
+    assert storage.get_offset() == 42
+    assert storage.commits == 1
+    assert len(storage.messages) == 1
+    assert "уведомление оператору в топик" in caplog.text
+    assert "chat_id=555" in caplog.text  # только идентификатор, без текста переписки
+
+
+async def test_topic_notification_uses_interactive_send_budget(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Уведомление шлётся с УЗКИМ бюджетом: воркерный дефолт (5 ретраев, backoff
+    до 30с, полный retry_after на 429) застопорил бы весь poll loop на минуты."""
+    calls: list[tuple[str, dict[str, Any]]] = []
+    client = _make_client({"copyMessage": 403}, calls)
+    storage = FakeBridgeStorage()
+    storage.record_message(
+        chat_id=555,
+        direction="in",
+        tg_message_id=1,
+        topic_message_id=100,
+        kind="text",
+        text_body="вопрос клиента",
+        operator_tg_id=None,
+    )
+
+    sent: list[dict[str, Any]] = []
+
+    async def recording_send(**kwargs: Any) -> dict[str, Any]:
+        sent.append(kwargs)
+        return {"message_id": 1}
+
+    monkeypatch.setattr(client, "send_message", recording_send)
+
+    update = {"update_id": 43, "message": _group_reply_message(reply_to_message_id=100)}
+    await bridge.process_update(update, client, storage)
+
+    assert len(sent) == 1
+    assert sent[0]["timeout"] == bridge._NOTIFY_SEND_TIMEOUT_S
+    assert sent[0]["max_retries"] == bridge._NOTIFY_SEND_MAX_RETRIES
+    assert sent[0]["max_backoff"] == bridge._NOTIFY_SEND_MAX_BACKOFF_S
+    assert sent[0]["chat_id"] == SUPPORT_CHAT_ID

tradein-mvp/backend/tests/services/tgbot/test_client.py

@@ -17,6 +17,9 @@ import pytest
 os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
 
 from app.services.tgbot.client import (
+    _CONNECT_TIMEOUT_S,
+    _POOL_TIMEOUT_S,
+    _WRITE_TIMEOUT_S,
     TelegramApiError,
     TelegramClient,
     TelegramNetworkError,
@@ -25,13 +28,19 @@ from app.services.tgbot.client import (
 _REAL_ASYNC_CLIENT = httpx.AsyncClient
 
 
-def _install_transport(handler) -> None:
+def _install_transport(handler) -> list[httpx.AsyncClient]:
+    """Подменяет транспорт. Возвращает список СОЗДАННЫХ AsyncClient — по нему
+    видно, переиспользуется ли один клиент или он плодится на каждый запрос."""
     transport = httpx.MockTransport(handler)
+    created: list[httpx.AsyncClient] = []
 
     def factory(*_: object, **__: object) -> httpx.AsyncClient:
-        return _REAL_ASYNC_CLIENT(transport=transport)
+        client = _REAL_ASYNC_CLIENT(transport=transport)
+        created.append(client)
+        return client
 
     mock.patch("app.services.tgbot.client.httpx.AsyncClient", factory).start()
+    return created
 
 
 @pytest.fixture(autouse=True)
@@ -304,3 +313,99 @@ async def test_network_error_is_not_api_error() -> None:
         await TelegramClient(token="t").send_message(chat_id=-1, text="x", max_retries=0)
 
     assert not isinstance(caught.value, TelegramApiError)
+
+
+# --- раздельные таймауты (#tg-connection-resilience) --------------------------
+
+
+async def _capture_request_timeout(call) -> dict[str, float]:
+    captured: dict[str, Any] = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        captured["timeout"] = request.extensions.get("timeout")
+        return httpx.Response(200, json={"ok": True, "result": []})
+
+    _install_transport(handler)
+    await call(TelegramClient(token="t"))
+    assert captured["timeout"] is not None, "таймаут не доехал до запроса"
+    return captured["timeout"]
+
+
+async def test_long_poll_timeout_applies_to_read_only_not_to_connect() -> None:
+    """40 секунд запаса long-poll'а — это бюджет ОТВЕТА, а не установки соединения.
+
+    Скаляр в httpx разворачивается в connect=read=write=pool, поэтому
+    `getUpdates` ждал 40с и коннекта тоже. Живой connect до api.telegram.org из
+    прод-контейнера — 0.036с; худший цикл из-за этого растягивался на ~174с
+    (4 попытки × 40с + backoff), и всё это время бот не видел ответов оператора.
+    """
+    timeout = await _capture_request_timeout(lambda tg: tg.get_updates(offset=0, timeout=30))
+
+    assert timeout["read"] == 40.0, "long-poll обязан сохранить свои 30+10с на ответ"
+    assert timeout["connect"] == _CONNECT_TIMEOUT_S, "connect не должен наследовать long-poll"
+    assert timeout["write"] == _WRITE_TIMEOUT_S
+    assert timeout["pool"] == _POOL_TIMEOUT_S
+
+
+async def test_interactive_call_keeps_its_own_narrow_read_budget() -> None:
+    """Узкий интерактивный бюджет ручки — тоже read, и он не подменяется дефолтом."""
+    timeout = await _capture_request_timeout(
+        lambda tg: tg.send_message(chat_id=1, text="x", timeout=6.0)
+    )
+
+    assert timeout["read"] == 6.0
+    assert timeout["connect"] == _CONNECT_TIMEOUT_S
+
+
+# --- переиспользование соединения --------------------------------------------
+
+
+async def test_two_calls_share_one_httpx_client_and_aclose_closes_it() -> None:
+    """Один `httpx.AsyncClient` на жизнь `TelegramClient`, а не на запрос.
+
+    Раньше клиент создавался ВНУТРИ цикла ретраев: нулевой keep-alive, полный
+    TCP+TLS-хендшейк на каждый запрос и на каждую попытку.
+    """
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json={"ok": True, "result": {"message_id": 1}})
+
+    created = _install_transport(handler)
+    tg = TelegramClient(token="t")
+    await tg.send_message(chat_id=1, text="раз")
+    await tg.send_message(chat_id=1, text="два")
+
+    assert len(created) == 1, f"клиент пересоздаётся на запрос: {len(created)} штук"
+    assert not created[0].is_closed
+
+    await tg.aclose()
+    assert created[0].is_closed, "aclose() обязан закрыть пул соединений"
+
+
+async def test_retries_reuse_the_same_http_client() -> None:
+    """Ретраи не пересоздают клиент — иначе повтор платит за хендшейк заново."""
+    calls = {"n": 0}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        calls["n"] += 1
+        if calls["n"] < 3:
+            return httpx.Response(502, json={"ok": False, "error_code": 502, "description": "gw"})
+        return httpx.Response(200, json={"ok": True, "result": {"message_id": 7}})
+
+    created = _install_transport(handler)
+    await TelegramClient(token="t").send_message(chat_id=1, text="x")
+
+    assert calls["n"] == 3, "бюджет ретраев изменился незаметно"
+    assert len(created) == 1, "на каждую попытку создаётся новый клиент"
+
+
+async def test_async_context_manager_closes_client_on_exit() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json={"ok": True, "result": {"message_id": 1}})
+
+    created = _install_transport(handler)
+    async with TelegramClient(token="t") as tg:
+        await tg.send_message(chat_id=1, text="x")
+
+    assert len(created) == 1
+    assert created[0].is_closed

tradein-mvp/backend/tests/services/tgbot/test_shared.py

@@ -0,0 +1,80 @@
+"""Жизненный цикл общего на приложение `TelegramClient` (`app.services.tgbot.shared`).
+
+Смысл модуля — ровно в том, что экземпляр ОДИН: до правки три HTTP-ручки создавали
+клиент на каждый входящий запрос, и keep-alive не было ни на каком уровне. Если
+синглтон однажды перестанет быть синглтоном, тесты ручек этого не заметят (они
+подменяют аксессор целиком), а на проде вернутся TLS-хендшейки на каждое сообщение.
+Поэтому проверяем идентичность и закрытие здесь, отдельно.
+
+Реальных сетевых вызовов нет: `TelegramClient` создаёт `httpx.AsyncClient` лениво,
+при первом запросе, а мы до запросов не доходим.
+"""
+
+from __future__ import annotations
+
+import os
+
+import pytest
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from app.services.tgbot import shared
+from app.services.tgbot.client import TelegramClient
+
+
+@pytest.fixture(autouse=True)
+def _reset_singleton():
+    """Глобал модуля не должен течь между тестами — иначе порядок решает исход."""
+    shared._client = None
+    yield
+    shared._client = None
+
+
+def test_get_returns_same_instance() -> None:
+    """Два обращения — один объект. Это и есть весь смысл модуля."""
+    first = shared.get_telegram_client()
+    second = shared.get_telegram_client()
+
+    assert isinstance(first, TelegramClient)
+    assert first is second, "аксессор создал второй клиент — пул перестал переиспользоваться"
+
+
+def test_init_returns_the_shared_instance() -> None:
+    """`init_telegram_client` в lifespan и `get_telegram_client` в ручке — один и тот же объект."""
+    created = shared.init_telegram_client()
+
+    assert created is shared.get_telegram_client()
+
+
+async def test_close_resets_and_next_get_builds_a_fresh_one() -> None:
+    """После shutdown глобал обнулён; повторный startup обязан получить рабочий клиент.
+
+    Ленивость после закрытия намеренна: держать «остановленный» флаг значило бы,
+    что повторный `init_telegram_client()` в том же процессе отдаёт закрытый пул.
+    """
+    first = shared.get_telegram_client()
+    await shared.close_telegram_client()
+
+    assert shared._client is None, "глобал не обнулён — следующий startup взял бы закрытый пул"
+
+    second = shared.get_telegram_client()
+    assert second is not first
+
+
+async def test_close_is_idempotent() -> None:
+    """Второй `close` не должен падать: lifespan зовёт его из `finally`."""
+    shared.get_telegram_client()
+
+    await shared.close_telegram_client()
+    await shared.close_telegram_client()
+
+    assert shared._client is None
+
+
+async def test_close_without_client_is_a_noop() -> None:
+    """Пустой токен — клиента в lifespan не создавали, закрывать нечего."""
+    assert shared._client is None
+
+    await shared.close_telegram_client()
+
+    assert shared._client is None

tradein-mvp/backend/tests/test_glitchtip_webhook.py

@@ -59,7 +59,11 @@ class _FakeTelegramClient:
 def _fake_telegram_client(monkeypatch: pytest.MonkeyPatch) -> Any:
     _FakeTelegramClient.calls = []
     _FakeTelegramClient._response = {"message_id": 1}
-    monkeypatch.setattr(glitchtip_module, "TelegramClient", _FakeTelegramClient)
+    # Ручка берёт ОБЩИЙ клиент приложения (#tg-connection-resilience), а не
+    # создаёт свой на запрос — подменяем аксессор, а не класс.
+    monkeypatch.setattr(
+        glitchtip_module, "get_telegram_client", lambda: _FakeTelegramClient("fake-token")
+    )
     return _FakeTelegramClient
 
 

tradein-mvp/backend/tests/test_support.py

@@ -69,7 +69,11 @@ class _FakeTelegramClient:
 def _fake_telegram_client(monkeypatch: pytest.MonkeyPatch) -> Any:
     _FakeTelegramClient.calls = []
     _FakeTelegramClient._response = {"message_id": 555}
-    monkeypatch.setattr(support_module, "TelegramClient", _FakeTelegramClient)
+    # Ручка берёт ОБЩИЙ клиент приложения (#tg-connection-resilience), а не
+    # создаёт свой на запрос — подменяем аксессор, а не класс.
+    monkeypatch.setattr(
+        support_module, "get_telegram_client", lambda: _FakeTelegramClient("fake-token")
+    )
     return _FakeTelegramClient