fix(gateway): don't treat group session user_id as thread_id in shutdown notifications (#10546)

_parse_session_key() blindly assigned parts[5] as thread_id for all
chat types. For group sessions with per-user isolation, parts[5] is
a user_id, not a thread_id. This could cause shutdown notifications
to route with incorrect thread metadata.

Only return thread_id for chat types where the 6th element is
unambiguous: dm and thread. For group/channel sessions, omit
thread_id since the suffix may be a user_id.

Based on the approach from PR #9938 by @Ruzzgar.

gateway/run.py

@@ -486,9 +486,14 @@ def _parse_session_key(session_key: str) -> "dict | None":
     """Parse a session key into its component parts.
 
     Session keys follow the format
-    ``agent:main:{platform}:{chat_type}:{chat_id}[:{thread_id}[:{user_id}]]``.
+    ``agent:main:{platform}:{chat_type}:{chat_id}[:{extra}...]``.
     Returns a dict with ``platform``, ``chat_type``, ``chat_id``, and
     optionally ``thread_id`` keys, or None if the key doesn't match.
+
+    The 6th element is only returned as ``thread_id`` for chat types where
+    it is unambiguous (``dm`` and ``thread``).  For group/channel sessions
+    the suffix may be a user_id (per-user isolation) rather than a
+    thread_id, so we leave ``thread_id`` out to avoid mis-routing.
     """
     parts = session_key.split(":")
     if len(parts) >= 5 and parts[0] == "agent" and parts[1] == "main":
@@ -497,7 +502,7 @@ def _parse_session_key(session_key: str) -> "dict | None":
             "chat_type": parts[3],
             "chat_id": parts[4],
         }
-        if len(parts) > 5:
+        if len(parts) > 5 and parts[3] in ("dm", "thread"):
             result["thread_id"] = parts[5]
         return result
     return None

tests/gateway/test_background_process_notifications.py

@@ -383,15 +383,27 @@ def test_parse_session_key_valid():
 
 
 def test_parse_session_key_with_extra_parts():
-    """Thread ID (6th part) is extracted; further parts are ignored."""
+    """6th part in a group key may be a user_id, not a thread_id — omit it."""
     result = _parse_session_key("agent:main:discord:group:chan123:thread456")
-    assert result == {"platform": "discord", "chat_type": "group", "chat_id": "chan123", "thread_id": "thread456"}
+    assert result == {"platform": "discord", "chat_type": "group", "chat_id": "chan123"}
 
 
 def test_parse_session_key_with_user_id_part():
-    """7th part (user_id) is ignored — only up to thread_id is extracted."""
-    result = _parse_session_key("agent:main:telegram:group:chat1:thread42:user99")
-    assert result == {"platform": "telegram", "chat_type": "group", "chat_id": "chat1", "thread_id": "thread42"}
+    """Group keys with per-user isolation have user_id as 6th part — don't return as thread_id."""
+    result = _parse_session_key("agent:main:telegram:group:chat1:user99")
+    assert result == {"platform": "telegram", "chat_type": "group", "chat_id": "chat1"}
+
+
+def test_parse_session_key_dm_with_thread():
+    """DM keys use parts[5] as thread_id unambiguously."""
+    result = _parse_session_key("agent:main:telegram:dm:chat1:topic42")
+    assert result == {"platform": "telegram", "chat_type": "dm", "chat_id": "chat1", "thread_id": "topic42"}
+
+
+def test_parse_session_key_thread_chat_type():
+    """Thread-typed keys use parts[5] as thread_id unambiguously."""
+    result = _parse_session_key("agent:main:discord:thread:chan1:thread99")
+    assert result == {"platform": "discord", "chat_type": "thread", "chat_id": "chan1", "thread_id": "thread99"}
 
 
 def test_parse_session_key_too_short():