fix(gateway): match the profile-namespaced session key in the clarify bypass lookup

gateway/platforms/base.py

@@ -5952,10 +5952,12 @@ async def handle_message(self, event: MessageEvent) -> None:
         if needs_topic_recovery:
             await asyncio.to_thread(self._apply_topic_recovery, event)
 
+        _sk_store = getattr(self, "_session_store", None)
         session_key = build_session_key(
             event.source,
             group_sessions_per_user=self.config.extra.get("group_sessions_per_user", True),
             thread_sessions_per_user=self.config.extra.get("thread_sessions_per_user", False),
+            profile=_sk_store._resolve_profile_for_key(event.source) if _sk_store else None,
         )
 
         # On-entry self-heal: if the adapter still has an _active_sessions

tests/gateway/test_clarify_active_session_bypass.py

@@ -1,7 +1,7 @@
 """Regression tests for clarify replies while a gateway session is busy."""
 
 import asyncio
-from unittest.mock import AsyncMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -87,3 +87,56 @@ async def test_active_session_routes_typed_choice_clarify_reply_to_runner_not_bu
     assert adapter._pending_messages == {}
 
 
+@pytest.mark.asyncio
+async def test_active_session_bypass_uses_profile_namespaced_key_under_multiplex():
+    """Regression for issue #82975: under a named-profile multiplex, the
+    adapter's clarify bypass lookup must use the SAME profile-namespaced
+    session key that the runner registers pending clarifies under
+    (SessionStore._generate_session_key() includes
+    profile=self._resolve_profile_for_key(source)), not the legacy
+    unnamespaced key. Otherwise the lookup misses, and a user's answer to
+    a pending clarify is routed to the busy-session queue instead of
+    resolving it -- the turn then hangs until the clarify's 3600s timeout."""
+    _clear_clarify_state()
+    from tools import clarify_gateway as cm
+
+    adapter = _ClarifyBypassAdapter()
+    adapter._message_handler = AsyncMock(return_value="")
+    adapter._busy_session_handler = AsyncMock(return_value=True)
+    event = _event("None of those are valid options")
+
+    # A session_store configured for profile multiplexing, matching what
+    # the runner's SessionStore._generate_session_key() actually produces.
+    session_store = MagicMock()
+    session_store._resolve_profile_for_key.return_value = "ops"
+    adapter._session_store = session_store
+
+    profile_namespaced_key = build_session_key(
+        event.source,
+        group_sessions_per_user=adapter.config.extra.get("group_sessions_per_user", True),
+        thread_sessions_per_user=adapter.config.extra.get("thread_sessions_per_user", False),
+        profile="ops",
+    )
+    # Sanity: the profile-namespaced key really is different from the
+    # legacy unnamespaced one -- otherwise this test wouldn't distinguish
+    # the fixed behavior from the bug.
+    legacy_key = build_session_key(
+        event.source,
+        group_sessions_per_user=adapter.config.extra.get("group_sessions_per_user", True),
+        thread_sessions_per_user=adapter.config.extra.get("thread_sessions_per_user", False),
+    )
+    assert profile_namespaced_key != legacy_key
+
+    adapter._active_sessions[profile_namespaced_key] = asyncio.Event()
+    # The runner registers the pending clarify under its own
+    # profile-namespaced key, exactly as it would in a real multiplexed
+    # deployment.
+    cm.register("clarify-1", profile_namespaced_key, "Pick one", ["A", "B"])
+
+    await adapter.handle_message(event)
+
+    adapter._message_handler.assert_awaited_once_with(event)
+    adapter._busy_session_handler.assert_not_awaited()
+    assert adapter._pending_messages == {}
+
+