fix(tg): out-строки писались с support_chat_id=NULL — вечный wildcard-матч

Deep review PR #3479 нашёл дефект в предыдущем фиксе (#3471 пункт 3): новые
direction='out' строки стали видимы резолверам (find_chat_by_topic_message,
find_thread_by_topic_message), но писались без support_chat_id. Резолверы
матчат support_chat_id IS NULL как лениентный wildcard "любой текущий чат"
(легаси-строки до 187/188) — то есть КАЖДАЯ out-строка становилась таким
wildcard. При ротации support-группы новый message_id мог бы случайно
совпасть со старой out-строкой: TG-путь увёл бы ответ ЧУЖОМУ клиенту через
copyMessage, веб-путь записал бы ответ в чужой тред. Ровно от этого
защищали миграции 187/188 (review M1).

- bridge.py: TG- и веб-ветка `_handle_group_reply` теперь передают
  support_chat_id=settings.telegram_support_chat_id в record_message /
  record_web_out_message (симметрично уже существующей in-ветке).
- web_support_storage.record_outbound: добавлен параметр support_chat_id,
  пишется в INSERT (колонка уже существовала, DDL не нужен).
- Тест test_group_reply_to_own_previous_tg_reply_resolves_target_chat сидел
  предыдущую out-строку с уже заполненным support_chat_id вручную, хотя код
  писал NULL — маскировал дефект. Добавлены прямые проверки на записанное
  support_chat_id (TG и веб), обе падают на прежней реализации (проверено
  локальным откатом изменения — 2 failed, restore — 41 passed).
- Комментарий про "апдейт частично применён в Telegram" в except-ветке
  веб-ответа был неверен для этого случая (на веб-пути ничего не уходит в
  Telegram до сбоя БД) — переписан на настоящую причину: сбой БД не
  переигрывается по общей политике process_update, а не из-за частичной
  доставки.

Refs #3471

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

@@ -229,6 +229,7 @@ class BridgeStorage(Protocol):
         text_body: str,
         operator_tg_id: int | None,
         topic_message_id: int | None = None,
+        support_chat_id: int | None = None,
     ) -> None: ...
 
 
@@ -430,6 +431,7 @@ class SqlBridgeStorage:
         text_body: str,
         operator_tg_id: int | None,
         topic_message_id: int | None = None,
+        support_chat_id: int | None = None,
     ) -> None:
         web_support_storage.record_outbound(
             self._db,
@@ -437,6 +439,7 @@ class SqlBridgeStorage:
             text_body=text_body,
             operator_tg_id=operator_tg_id,
             topic_message_id=topic_message_id,
+            support_chat_id=support_chat_id,
         )
 
 
@@ -684,6 +687,12 @@ async def _handle_group_reply(
             kind=_infer_kind(message),
             text_body=message.get("text") or message.get("caption"),
             operator_tg_id=operator_id,
+            # Deep review PR #3479: без этого out-строка была бы вечным
+            # wildcard для `find_chat_by_topic_message` (матчит support_chat_id
+            # IS NULL под ЛЮБЫМ текущим чатом) — при ротации support-группы
+            # (188) новый message_id мог бы совпасть со старой out-строкой и
+            # увести ответ ЧУЖОМУ клиенту. Симметрично in-ветке выше (строка ~601).
+            support_chat_id=settings.telegram_support_chat_id,
         )
         return
 
@@ -726,14 +735,29 @@ async def _handle_group_reply(
                 # реплай оператора на СВОЙ предыдущий веб-ответ не резолвится
                 # (см. `find_thread_by_topic_message`, direction-фильтр снят).
                 topic_message_id=message_id if isinstance(message_id, int) else None,
+                # Deep review PR #3479: БЕЗ этого out-строка писалась бы с
+                # support_chat_id=NULL — `find_thread_by_topic_message` матчит
+                # NULL под ЛЮБЫМ текущим чатом (лениентный wildcard для легаси
+                # строк до 187/188), т.е. каждая out-строка стала бы вечным
+                # wildcard. При ротации support-группы новый message_id мог бы
+                # совпасть со старой out-строкой и увести ответ в ЧУЖОЙ тред —
+                # ровно то, от чего защищала скоупинг-миграция 187/188.
+                support_chat_id=settings.telegram_support_chat_id,
             )
         except SQLAlchemyError:
             # #3471 P0: для веб-треда ЭТА запись — и есть доставка клиенту (веб-
             # фронт вычитывает ответ обычным polling'ом web_support_messages).
             # Откат без уведомления означал бы: оператор уверен, что ответил,
-            # клиент ждёт молча, а Telegram апдейт больше не переиграет (offset
-            # ниже всё равно сдвигается — апдейт частично применён в Telegram,
-            # переигрывать нельзя). rollback() ОБЯЗАН отработать ДО уведомления —
+            # клиент ждёт молча. Offset ниже всё равно сдвигается — НЕ потому,
+            # что апдейт "частично применён в Telegram" (в этой ветке до сбоя в
+            # Telegram ничего не уходило вообще: сам реплай оператора Telegram
+            # уже полностью доставил ДО того, как мы начали его разбирать,
+            # ретраить на стороне площадки нечего), а потому что действует общая
+            # политика `process_update` для `SQLAlchemyError` — сбой БД не
+            # переигрывается (в отличие от `TelegramNetworkError`), а
+            # сигнализируется громко; здесь это explicit-просьба оператору
+            # прислать ответ заново — human-in-the-loop retry вместо
+            # технического. rollback() ОБЯЗАН отработать ДО уведомления —
             # сессия в failed-transaction state, а `_notify_topic` шлёт через
             # `client`, не через `storage`, поэтому сам rollback тут не нужен для
             # отправки, но нужен, чтобы process_update дальше не упал на

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

@@ -146,6 +146,7 @@ def record_outbound(
     text_body: str,
     operator_tg_id: int | None,
     topic_message_id: int | None = None,
+    support_chat_id: int | None = None,
 ) -> int | None:
     """Записывает ответ оператора (реплай на веб-зеркало) как direction='out'.
 
@@ -154,15 +155,25 @@ def record_outbound(
     inbound-записи (конвенция 186/187), из-за чего реплай оператора на СВОЙ
     предыдущий ответ был нерезолвим — искать было нечего, а `reply_to_message_id`
     указывал на строку без ключа. `find_thread_by_topic_message` теперь матчит
-    обе стороны (direction-фильтр там снят)."""
+    обе стороны (direction-фильтр там снят).
+
+    `support_chat_id` (deep review PR #3479) — ОБЯЗАТЕЛЕН при заполненном
+    `topic_message_id`: `find_thread_by_topic_message` матчит `support_chat_id
+    IS NULL` как лениентный wildcard "под любым текущим чатом" (легаси-строки
+    до 187/188). Без этого поля КАЖДАЯ out-строка была бы таким wildcard — при
+    ротации support-группы новый message_id мог бы совпасть со старой
+    out-строкой и увести ответ в ЧУЖОЙ тред (ровно то, от чего защищала
+    скоупинг-миграция 187/188, см. review M1 там же)."""
     row = db.execute(
         text(
             """
             INSERT INTO web_support_messages
-                (thread_id, direction, text_body, topic_message_id, operator_tg_id, created_at)
+                (thread_id, direction, text_body, topic_message_id,
+                 support_chat_id, operator_tg_id, created_at)
             VALUES
                 (CAST(:thread_id AS bigint), 'out', :text_body,
                  CAST(:topic_message_id AS bigint),
+                 CAST(:support_chat_id AS bigint),
                  CAST(:operator_tg_id AS bigint), NOW())
             RETURNING id
             """
@@ -171,6 +182,7 @@ def record_outbound(
             "thread_id": thread_id,
             "text_body": text_body,
             "topic_message_id": topic_message_id,
+            "support_chat_id": support_chat_id,
             "operator_tg_id": operator_tg_id,
         },
     ).fetchone()

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

@@ -197,6 +197,7 @@ class FakeBridgeStorage:
         text_body: str,
         operator_tg_id: int | None,
         topic_message_id: int | None = None,
+        support_chat_id: int | None = None,
     ) -> None:
         if self.fail_next_record_web_out_message:
             self.fail_next_record_web_out_message = False
@@ -207,8 +208,15 @@ class FakeBridgeStorage:
                 "text_body": text_body,
                 "operator_tg_id": operator_tg_id,
                 "topic_message_id": topic_message_id,
+                "support_chat_id": support_chat_id,
             }
         )
+        # Зеркалим в `web_topic_to_thread` (deep review PR #3479) — реальный
+        # `record_outbound` пишет ту же строку в web_support_messages, которую
+        # потом читает `find_thread_by_topic_message`; без этого фейк не мог бы
+        # поймать баг "out-строка с support_chat_id=NULL — вечный wildcard".
+        if topic_message_id is not None:
+            self.web_topic_to_thread[topic_message_id] = (thread_id, support_chat_id)
 
 
 # ── httpx mocking helpers (mirrors tests/services/test_dadata.py) ───────────
@@ -875,9 +883,10 @@ async def test_group_reply_to_web_mirror_db_failure_notifies_operator_and_advanc
     ЕСТЬ доставка клиенту (веб-фронт читает её polling'ом), поэтому тихий откат
     означал бы навсегда потерянный ответ оператора (воспроизведено на проде
     31.08.2026 — клиент kopylov). Теперь: rollback → уведомление оператору
-    реплаем в топик, что ответ НЕ доставлен → offset всё равно сдвигается
-    (апдейт уже частично применён в Telegram, переигрывать нельзя) → исключение
-    наружу НЕ улетает."""
+    реплаем в топик, что ответ НЕ доставлен → offset всё равно сдвигается (та же
+    политика, что у любого другого `SQLAlchemyError` в `process_update` — сбой
+    БД не переигрывается, human-in-the-loop retry заменяет технический) →
+    исключение наружу НЕ улетает."""
     calls: list[tuple[str, dict[str, Any]]] = []
     client = _make_client({}, calls)
     storage = FakeBridgeStorage()
@@ -899,7 +908,7 @@ async def test_group_reply_to_web_mirror_db_failure_notifies_operator_and_advanc
     assert notice["chat_id"] == SUPPORT_CHAT_ID
     assert notice["reply_to_message_id"] == 210
     assert "НЕ доставлен" in notice["text"]
-    # Апдейт частично применён в Telegram — не переигрываем, offset сдвинут и закоммичен.
+    # Сбой БД не переигрывается (общая политика SQLAlchemyError) — offset сдвинут и закоммичен.
     assert storage.get_offset() == 70
     assert storage.commits == 1
 
@@ -931,7 +940,7 @@ async def test_group_reply_to_web_mirror_db_failure_and_notify_failure_logs_erro
     assert storage.web_out_messages == []
     assert "потерян молча" in caplog.text
     assert "51" in caplog.text  # thread_id узнаваем в логе
-    assert storage.get_offset() == 71  # апдейт частично применён — не переигрываем
+    assert storage.get_offset() == 71  # сбой БД не переигрывается — offset сдвинут
     assert storage.commits == 1
 
 
@@ -939,7 +948,14 @@ async def test_group_reply_to_own_previous_web_reply_resolves_thread() -> None:
     """#3471 P0 (пункт 3): реплай оператора на СВОЙ предыдущий веб-ответ (не на
     исходное зеркало клиента) теперь тоже резолвится — `record_outbound`
     сохраняет topic_message_id исходящей записи, `find_thread_by_topic_message`
-    больше не фильтрует по direction."""
+    больше не фильтрует по direction.
+
+    Deep review PR #3479: новая out-строка ОБЯЗАНА писаться с ТЕКУЩИМ
+    `support_chat_id`, а не NULL — NULL матчится `find_thread_by_topic_message`
+    как лениентный wildcard "любой чат" (легаси до 187/188), т.е. NULL сделал бы
+    КАЖДУЮ out-строку вечным wildcard-совпадением при ротации support-группы.
+    Эта проверка падает на дефектной реализации (support_chat_id не передавался
+    в `record_web_out_message`), даже когда resolve выше внешне "работает"."""
     calls: list[tuple[str, dict[str, Any]]] = []
     client = _make_client({}, calls)
     storage = FakeBridgeStorage()
@@ -958,12 +974,19 @@ async def test_group_reply_to_own_previous_web_reply_resolves_thread() -> None:
     assert len(storage.web_out_messages) == 1
     assert storage.web_out_messages[0]["thread_id"] == 52
     assert storage.web_out_messages[0]["topic_message_id"] == 220
+    # Deep review PR #3479: НЕ NULL — иначе эта строка стала бы вечным wildcard.
+    assert storage.web_out_messages[0]["support_chat_id"] == SUPPORT_CHAT_ID
 
 
 async def test_group_reply_to_own_previous_tg_reply_resolves_target_chat() -> None:
     """#3471 P0 (пункт 3): та же история для Telegram-пути — реплай оператора на
     СВОЙ предыдущий ответ клиенту (direction='out', topic_message_id теперь
-    заполнен) резолвится в chat_id, а не проваливается в orphan-check."""
+    заполнен) резолвится в chat_id, а не проваливается в orphan-check.
+
+    Deep review PR #3479: новая out-строка ОБЯЗАНА писаться с ТЕКУЩИМ
+    `support_chat_id` (симметрично in-ветке) — иначе она стала бы вечным
+    wildcard в `find_chat_by_topic_message` при ротации support-группы, и
+    reply мог бы увести ответ ЧУЖОМУ клиенту."""
     calls: list[tuple[str, dict[str, Any]]] = []
     client = _make_client({"copyMessage": {"message_id": 601}}, calls)
     storage = FakeBridgeStorage()
@@ -991,6 +1014,8 @@ async def test_group_reply_to_own_previous_tg_reply_resolves_target_chat() -> No
     assert calls[0][1]["chat_id"] == 555
     assert len(storage.messages) == 2  # исходный 'out' + новый 'out'
     assert storage.messages[-1]["topic_message_id"] == 230
+    # Deep review PR #3479: НЕ NULL — иначе эта строка стала бы вечным wildcard.
+    assert storage.messages[-1]["support_chat_id"] == SUPPORT_CHAT_ID
 
 
 # ── C) дедуп ──────────────────────────────────────────────────────────────────