"""Regression tests for #31501 — prune stale Telegram DM topic bindings. When a Telegram user deletes a DM topic in the client, the Bot API responds to the gateway's next send with ``Thread not found``. The adapter falls back to a plain send (no ``message_thread_id``), but prior to this fix it left the corresponding row in ``telegram_dm_topic_bindings`` untouched. ``gateway.run._recover_telegram_topic_thread_id`` then walked the user's bindings newest-first on every later inbound message and cheerfully redirected them back to the deleted topic — tool progress, approvals and replies all silently landed in the wrong place until the operator manually ran ``DELETE`` on ``state.db``. The fix has three pieces — these tests pin all three: 1. ``SessionDB.delete_telegram_topic_binding`` — the targeted prune helper (new public API). 2. ``TelegramAdapter._prune_stale_dm_topic_binding`` — the adapter glue that calls the helper from a send-fallback hot path without raising on cleanup failure. 3. The two "Thread not found" call sites in the streaming send loop and the control-message helper now invoke (2) — we pin this with a source-level guard rather than spinning the full send pipeline. """ from __future__ import annotations import inspect from types import SimpleNamespace import pytest from hermes_state import SessionDB # --------------------------------------------------------------------------- # SessionDB.delete_telegram_topic_binding # --------------------------------------------------------------------------- def _seed_binding( db: SessionDB, *, chat_id: str = "5595856929", thread_id: str = "15287", user_id: str = "5595856929", session_id: str = "sess-target", ) -> None: db.create_session( session_id=session_id, source="telegram", user_id=user_id, ) db.bind_telegram_topic( chat_id=chat_id, thread_id=thread_id, user_id=user_id, session_key=f"agent:main:telegram:dm:{chat_id}:{thread_id}", session_id=session_id, ) class TestDeleteTelegramTopicBinding: def test_removes_matching_row_and_returns_count(self, tmp_path): db = SessionDB(db_path=tmp_path / "state.db") _seed_binding(db, thread_id="15287") # Sanity check — binding present before prune. assert db.get_telegram_topic_binding( chat_id="5595856929", thread_id="15287", ) is not None removed = db.delete_telegram_topic_binding( chat_id="5595856929", thread_id="15287", ) assert removed == 1 assert db.get_telegram_topic_binding( chat_id="5595856929", thread_id="15287", ) is None db.close() def test_does_not_touch_unrelated_bindings(self, tmp_path): # Critical for the fix: a chat with multiple topics must # only lose the one Telegram confirmed deleted, never the # rest. Otherwise the user's healthy topics also vanish # from recovery's view. db = SessionDB(db_path=tmp_path / "state.db") _seed_binding(db, thread_id="15287", session_id="sess-stale") _seed_binding(db, thread_id="15418", session_id="sess-fresh") removed = db.delete_telegram_topic_binding( chat_id="5595856929", thread_id="15287", ) assert removed == 1 # Stale binding is gone; the fresh one survives. assert db.get_telegram_topic_binding( chat_id="5595856929", thread_id="15287", ) is None assert db.get_telegram_topic_binding( chat_id="5595856929", thread_id="15418", ) is not None db.close() def test_missing_row_returns_zero_silently(self, tmp_path): db = SessionDB(db_path=tmp_path / "state.db") _seed_binding(db, thread_id="15287") # Different thread_id — must not raise, just report 0. removed = db.delete_telegram_topic_binding( chat_id="5595856929", thread_id="99999", ) assert removed == 0 # Original binding still intact. assert db.get_telegram_topic_binding( chat_id="5595856929", thread_id="15287", ) is not None db.close() def test_pristine_database_with_no_topic_tables_is_silent_noop(self, tmp_path): # Fresh profile that has never run /topic — the topic-mode # tables don't exist yet. The send-fallback hot path can # still hit this code, so we must not crash. db = SessionDB(db_path=tmp_path / "state.db") # Confirm precondition: tables really aren't there. tables = { row[0] for row in db._conn.execute( "SELECT name FROM sqlite_master WHERE type='table' " "AND name LIKE 'telegram_dm%'" ).fetchall() } assert "telegram_dm_topic_bindings" not in tables removed = db.delete_telegram_topic_binding( chat_id="any", thread_id="any", ) assert removed == 0 db.close() def test_idempotent_under_repeated_calls(self, tmp_path): db = SessionDB(db_path=tmp_path / "state.db") _seed_binding(db, thread_id="15287") first = db.delete_telegram_topic_binding( chat_id="5595856929", thread_id="15287", ) second = db.delete_telegram_topic_binding( chat_id="5595856929", thread_id="15287", ) assert first == 1 assert second == 0 # already gone, no spurious "1" db.close() class TestPruneClearsTopicModeWhenLastBindingGone: """Proactive cleanup (#31501 follow-up): pruning the chat's final binding must also flip ``telegram_dm_topic_mode.enabled`` to 0 so recovery fully stands down — covers the user who disabled topics in the Telegram client without ever running ``/topic off``.""" def test_clears_enabled_when_last_binding_pruned(self, tmp_path): db = SessionDB(db_path=tmp_path / "state.db") db.enable_telegram_topic_mode( chat_id="5595856929", user_id="5595856929", ) _seed_binding(db, thread_id="15287") assert db.is_telegram_topic_mode_enabled( chat_id="5595856929", user_id="5595856929", ) is True removed = db.delete_telegram_topic_binding( chat_id="5595856929", thread_id="15287", ) assert removed == 1 assert db.is_telegram_topic_mode_enabled( chat_id="5595856929", user_id="5595856929", ) is False db.close() def test_keeps_enabled_while_other_bindings_remain(self, tmp_path): # Deleting one of several topics must NOT disable topic mode — # the chat still has healthy lanes that recovery should serve. db = SessionDB(db_path=tmp_path / "state.db") db.enable_telegram_topic_mode( chat_id="5595856929", user_id="5595856929", ) _seed_binding(db, thread_id="15287", session_id="sess-stale") _seed_binding(db, thread_id="15418", session_id="sess-fresh") db.delete_telegram_topic_binding( chat_id="5595856929", thread_id="15287", ) assert db.is_telegram_topic_mode_enabled( chat_id="5595856929", user_id="5595856929", ) is True db.close() def test_noop_prune_leaves_enabled_untouched(self, tmp_path): # A prune that matches no row must not flip the flag — there's # still a live binding the (wrong) thread_id didn't match. db = SessionDB(db_path=tmp_path / "state.db") db.enable_telegram_topic_mode( chat_id="5595856929", user_id="5595856929", ) _seed_binding(db, thread_id="15287") removed = db.delete_telegram_topic_binding( chat_id="5595856929", thread_id="99999", ) assert removed == 0 assert db.is_telegram_topic_mode_enabled( chat_id="5595856929", user_id="5595856929", ) is True db.close() # --------------------------------------------------------------------------- # Adapter glue — _prune_stale_dm_topic_binding # --------------------------------------------------------------------------- def _bare_adapter(db: SessionDB | None = None): # The adapter accesses the SessionDB via # ``self._session_store._db`` (set by GatewayRunner via # ``set_session_store``). Build a minimal stand-in with just # the surface the prune helper touches; we don't need the # python-telegram-bot import-graph here. ``name`` is a # property that delegates to ``platform.value.title()``, so # we set ``platform`` rather than poking ``name`` directly. from gateway.config import Platform from plugins.platforms.telegram.adapter import TelegramAdapter adapter = object.__new__(TelegramAdapter) adapter.platform = Platform.TELEGRAM if db is not None: adapter._session_store = SimpleNamespace(_db=db) return adapter class TestPruneStaleDmTopicBindingHelper: def test_drops_binding_when_session_store_db_is_present(self, tmp_path): db = SessionDB(db_path=tmp_path / "state.db") _seed_binding(db, thread_id="15287") adapter = _bare_adapter(db) adapter._prune_stale_dm_topic_binding("5595856929", 15287) assert db.get_telegram_topic_binding( chat_id="5595856929", thread_id="15287", ) is None db.close() def test_silent_when_session_store_unavailable(self): # No ``_session_store`` attribute — the helper must not # explode (the streaming send path hits this in tests # that bypass the gateway runner). adapter = _bare_adapter() adapter._prune_stale_dm_topic_binding("123", "456") def test_silent_when_db_lacks_helper(self): # Old SessionDB without the new method (e.g. running # against an older state.db schema). Must be a no-op # rather than AttributeError. adapter = _bare_adapter() adapter._session_store = SimpleNamespace( _db=SimpleNamespace(), # no methods at all ) adapter._prune_stale_dm_topic_binding("123", "456") def test_swallows_db_exceptions_so_send_continues(self): class ExplodingDb: def delete_telegram_topic_binding(self, **_): raise RuntimeError("disk full or whatever") adapter = _bare_adapter() adapter._session_store = SimpleNamespace(_db=ExplodingDb()) # The point of the helper is that a failed cleanup must # NEVER turn into a failed user-facing send. No exception # should escape. adapter._prune_stale_dm_topic_binding("123", "456") def test_skips_when_chat_or_thread_missing(self, tmp_path): # Defensive — control-message paths sometimes call us # with chat_id=None when kwargs lack the key. We must # not produce a spurious DELETE that matches every row # with a NULL chat_id. db = SessionDB(db_path=tmp_path / "state.db") _seed_binding(db, thread_id="15287") adapter = _bare_adapter(db) adapter._prune_stale_dm_topic_binding(None, "15287") adapter._prune_stale_dm_topic_binding("5595856929", None) # Still there — neither call generated a DELETE. assert db.get_telegram_topic_binding( chat_id="5595856929", thread_id="15287", ) is not None db.close() # --------------------------------------------------------------------------- # Source-level wiring guards — both fallback sites must call the helper # --------------------------------------------------------------------------- class TestThreadNotFoundFallbackSitesPruneBinding: """Pin that the two ``Thread not found`` warning sites in the Telegram adapter actually invoke ``_prune_stale_dm_topic_binding``. These guards stop a future refactor from quietly losing the cleanup wire — re-opening #31501. """ def test_streaming_send_fallback_calls_prune(self): from plugins.platforms.telegram import adapter as telegram_mod src = inspect.getsource(telegram_mod.TelegramAdapter.send) # Locate the second-failure branch (the one that flips # ``used_thread_fallback``). It must invoke the prune # helper before flipping the flag. marker = "retrying without message_thread_id" idx = src.find(marker) assert idx != -1, ( "Streaming send must keep its 'thread not found' " "fallback log line — the prune wiring is anchored " "next to it." ) # 600 char window is enough to cover the warning, the # prune call, and the ``used_thread_fallback = True`` # assignment that follows. window = src[idx:idx + 600] assert "_prune_stale_dm_topic_binding" in window, ( "Streaming send 'Thread not found' fallback must call " "_prune_stale_dm_topic_binding so the stale row in " "telegram_dm_topic_bindings doesn't keep redirecting " "future inbound messages to the deleted topic (#31501)." ) def test_control_message_helper_calls_prune(self): from plugins.platforms.telegram import adapter as telegram_mod src = inspect.getsource( telegram_mod.TelegramAdapter._send_message_with_thread_fallback ) # The helper has a single retry path; the prune call # must sit inside it, not in dead code outside the # ``if message_thread_id is not None and …`` guard. assert "_prune_stale_dm_topic_binding" in src, ( "_send_message_with_thread_fallback must call " "_prune_stale_dm_topic_binding when Telegram returns " "BadRequest('Thread not found') for a control message " "(#31501)." ) # Belt-and-braces: the call must precede the retry # ``send_message`` so the prune happens whether or not # the retry itself succeeds. prune_idx = src.find("_prune_stale_dm_topic_binding") retry_idx = src.find("send_message(**retry_kwargs)") assert 0 <= prune_idx < retry_idx, ( "_prune_stale_dm_topic_binding must run before the " "fallback send_message retry." ) # --------------------------------------------------------------------------- # End-to-end semantic — prune + recovery returns None for deleted topic # --------------------------------------------------------------------------- class TestRecoveryAfterPrune: """The whole point of the fix: once a topic is pruned, the GatewayRunner's ``_recover_telegram_topic_thread_id`` must no longer steer future inbound messages to it. """ def test_recovery_no_longer_returns_pruned_topic(self, tmp_path): # Build the same fixture used elsewhere: two topic bindings # for the same user, then prune the most-recent one. # ``_recover_telegram_topic_thread_id`` walks bindings # newest-first, so without the prune it would pick the # one we just removed. from gateway.config import GatewayConfig, Platform, PlatformConfig from gateway.run import GatewayRunner from gateway.session import SessionSource, build_session_key db = SessionDB(db_path=tmp_path / "state.db") db.enable_telegram_topic_mode( chat_id="5595856929", user_id="5595856929", ) for sid, thread in (("sess-A", "111"), ("sess-B", "222")): db.create_session( session_id=sid, source="telegram", user_id="5595856929", ) db.bind_telegram_topic( chat_id="5595856929", thread_id=thread, user_id="5595856929", session_key=build_session_key(SessionSource( platform=Platform.TELEGRAM, user_id="5595856929", chat_id="5595856929", user_name="tester", chat_type="dm", thread_id=thread, )), session_id=sid, ) runner = object.__new__(GatewayRunner) runner.config = GatewayConfig( platforms={ Platform.TELEGRAM: PlatformConfig(enabled=True, token="***"), } ) runner._session_db = db runner._telegram_topic_mode_enabled = lambda _src: True # Sanity: before the prune, recovery picks "222" (newest). # Recovery only fires for a lobby-shaped inbound (omitted # message_thread_id or General topic "1"); a non-lobby # unknown thread is preserved as a brand-new topic. Use the # General topic id so the recovery walk actually runs. before = runner._recover_telegram_topic_thread_id(SessionSource( platform=Platform.TELEGRAM, user_id="5595856929", chat_id="5595856929", user_name="tester", chat_type="dm", thread_id="1", # General/stripped reply — triggers recovery )) assert before == "222" # User deletes topic 222 in Telegram → adapter prunes. db.delete_telegram_topic_binding( chat_id="5595856929", thread_id="222", ) # Now recovery falls back to topic 111 (the surviving # binding) instead of the dead one. This is the exact # behaviour change the bug report asks for. after = runner._recover_telegram_topic_thread_id(SessionSource( platform=Platform.TELEGRAM, user_id="5595856929", chat_id="5595856929", user_name="tester", chat_type="dm", thread_id="1", )) assert after == "111" db.close()