test(telegram): cover allowlist + unauthorized_dm_behavior pair pass-through

Guard the early-auth pairing gap: unknown DMs must reach the gateway when
pairing is the effective unauthorized-DM behavior, while ignore mode and
unauthorized group senders stay rejected.

tests/gateway/test_telegram_auth_check.py

@@ -144,6 +144,131 @@ def test_is_user_authorized_from_message_allow_from():
     assert adapter._is_user_authorized_from_message(msg) is False
 
 
+def test_allowlist_dm_with_explicit_pair_behavior_reaches_gateway(monkeypatch):
+    """Allowlist + unauthorized_dm_behavior:pair must not early-drop unknown DMs.
+
+    Regression for the gap left by #40863: early intake rejection discarded
+    unauthorized DMs before gateway pairing could run, even when the operator
+    explicitly set telegram.unauthorized_dm_behavior: pair.
+    """
+    monkeypatch.setenv("TELEGRAM_ALLOWED_USERS", "111")
+
+    class Runner:
+        def _is_user_authorized(self, source):
+            return source.user_id == "111"
+
+        def _get_unauthorized_dm_behavior(self, platform, *, profile=None):
+            assert platform == Platform.TELEGRAM
+            return "pair"
+
+        async def handle(self, event):
+            return None
+
+    runner = Runner()
+    adapter = _make_adapter()
+    adapter._message_handler = runner.handle
+    msg = _make_message(from_user_id=999, chat_id=999, chat_type="private")
+
+    assert adapter._is_user_authorized_from_message(msg) is True
+
+
+def test_allowlist_dm_without_pair_behavior_still_early_rejects(monkeypatch):
+    """Allowlist without pairing opt-in keeps the #9337/#40863 silent drop."""
+    monkeypatch.setenv("TELEGRAM_ALLOWED_USERS", "111")
+
+    class Runner:
+        def _is_user_authorized(self, source):
+            return source.user_id == "111"
+
+        def _get_unauthorized_dm_behavior(self, platform, *, profile=None):
+            return "ignore"
+
+        async def handle(self, event):
+            return None
+
+    runner = Runner()
+    adapter = _make_adapter()
+    adapter._message_handler = runner.handle
+    msg = _make_message(from_user_id=999, chat_id=999, chat_type="private")
+
+    assert adapter._is_user_authorized_from_message(msg) is False
+
+
+def test_allow_from_dm_with_pair_override_reaches_gateway():
+    """Adapter allow_from + unauthorized_dm_behavior:pair still forwards DMs."""
+    adapter = _make_adapter(
+        allow_from=["111"],
+        unauthorized_dm_behavior="pair",
+    )
+    msg = _make_message(from_user_id=999, chat_id=999, chat_type="dm")
+    assert adapter._is_user_authorized_from_message(msg) is True
+
+
+def test_allowlist_group_with_pair_behavior_still_early_rejects(monkeypatch):
+    """Pairing is DM-only — unauthorized group senders stay blocked early."""
+    monkeypatch.setenv("TELEGRAM_ALLOWED_USERS", "111")
+
+    class Runner:
+        def _is_user_authorized(self, source):
+            return source.user_id == "111"
+
+        def _get_unauthorized_dm_behavior(self, platform, *, profile=None):
+            return "pair"
+
+        async def handle(self, event):
+            return None
+
+    runner = Runner()
+    adapter = _make_adapter(group_allow_from=["111"])
+    adapter._message_handler = runner.handle
+    msg = _make_message(from_user_id=999, chat_id=-100, chat_type="group")
+
+    assert adapter._is_user_authorized_from_message(msg) is False
+
+
+@pytest.mark.asyncio
+async def test_unauthorized_dm_with_pair_behavior_builds_event(monkeypatch):
+    """Unknown DM under pair behavior must reach event construction."""
+    monkeypatch.setenv("TELEGRAM_ALLOWED_USERS", "111")
+
+    class Runner:
+        def _is_user_authorized(self, source):
+            return source.user_id == "111"
+
+        def _get_unauthorized_dm_behavior(self, platform, *, profile=None):
+            return "pair"
+
+        async def handle(self, event):
+            return None
+
+    runner = Runner()
+    adapter = _make_adapter()
+    adapter._message_handler = runner.handle
+    build_called = False
+    original_build = adapter._build_message_event
+
+    def track_build(*a, **kw):
+        nonlocal build_called
+        build_called = True
+        return original_build(*a, **kw)
+
+    adapter._build_message_event = track_build
+    adapter._enqueue_text_event = lambda event: None
+    adapter._ensure_forum_commands = AsyncMock()
+    adapter._cache_replied_media = AsyncMock()
+    adapter._apply_telegram_group_observe_attribution = lambda event: event
+    adapter._clean_bot_trigger_text = lambda text: text
+    adapter._should_process_message = lambda *a, **kw: True
+
+    update = SimpleNamespace(
+        update_id=1,
+        message=_make_message(from_user_id=999, chat_id=999, chat_type="private"),
+        effective_message=None,
+    )
+    await adapter._handle_text_message(update, SimpleNamespace())
+    assert build_called is True
+
+
 def test_runner_auth_gets_group_user_allowlist_context(monkeypatch):
     """Group user allowlists need a group-shaped source, not a DM-shaped one."""
     monkeypatch.setenv("TELEGRAM_GROUP_ALLOWED_USERS", "111")