fix(gateway): isolate DM sessions on user_id when chat_id is absent (#41764)

build_session_key collapsed every DM that arrived without a chat_id into
one shared 'agent:main:<platform>:dm' key. A single cached AIAgent then
served multiple users' conversations, bleeding history across senders.

DMs now fall back to the sender's user_id_alt/user_id (mirroring the
group-path participant precedence and the telegram auth-path fallback)
before the bare per-platform sink. Telegram's normal event path always
sets chat_id, so this hardens the synthetic-source / non-standard-adapter
paths that don't.

tests/gateway/test_session.py

@@ -784,6 +784,53 @@ class TestWhatsAppSessionKeyConsistency:
         assert build_session_key(second) == "agent:main:telegram:dm:100"
         assert build_session_key(first) != build_session_key(second)
 
+    def test_dm_without_chat_id_falls_back_to_user_id(self):
+        """A DM source missing chat_id must isolate on the sender's user_id
+        rather than collapsing into the shared per-platform sink."""
+        source = SessionSource(
+            platform=Platform.TELEGRAM,
+            chat_id="",
+            chat_type="dm",
+            user_id="jordan",
+        )
+        assert build_session_key(source) == "agent:main:telegram:dm:jordan"
+
+    def test_dm_without_chat_id_distinct_users_do_not_collide(self):
+        """Two different DM senders without chat_id must not share one
+        session (the cross-user history-bleed footgun)."""
+        first = SessionSource(
+            platform=Platform.TELEGRAM, chat_id="", chat_type="dm", user_id="jordan"
+        )
+        second = SessionSource(
+            platform=Platform.TELEGRAM, chat_id="", chat_type="dm", user_id="dima"
+        )
+        assert build_session_key(first) != build_session_key(second)
+        assert build_session_key(first) == "agent:main:telegram:dm:jordan"
+        assert build_session_key(second) == "agent:main:telegram:dm:dima"
+
+    def test_dm_without_chat_id_prefers_user_id_alt(self):
+        """user_id_alt wins over user_id for the DM fallback, matching the
+        group-path participant precedence."""
+        source = SessionSource(
+            platform=Platform.TELEGRAM,
+            chat_id="",
+            chat_type="dm",
+            user_id="primary",
+            user_id_alt="alt",
+        )
+        assert build_session_key(source) == "agent:main:telegram:dm:alt"
+
+    def test_dm_without_chat_id_or_user_id_falls_back_to_thread_then_sink(self):
+        """With neither chat_id nor user identifiers, thread_id is the next
+        discriminator; only a completely identifier-less DM hits the sink."""
+        threaded = SessionSource(
+            platform=Platform.TELEGRAM, chat_id="", chat_type="dm", thread_id="7"
+        )
+        assert build_session_key(threaded) == "agent:main:telegram:dm:7"
+
+        bare = SessionSource(platform=Platform.TELEGRAM, chat_id="", chat_type="dm")
+        assert build_session_key(bare) == "agent:main:telegram:dm"
+
     def test_discord_group_includes_chat_id(self):
         """Group/channel keys include chat_type and chat_id."""
         source = SessionSource(