fix(relay): per-message sessions for fronted Slack DMs — stamp the inbound ts as session thread (QA-3)

A 2nd top-level DM while a turn was in flight resolved to the SAME
session key and steered the running turn ('Redirected current run')
instead of starting its own. Native SlackAdapter stamps thread_ts =
event.thread_ts or ts on EVERY inbound, so build_session_key isolates
each top-level message; the connector normalizes top-level messages with
thread_id=null and the relay lane never reproduced the stamp.

_stamp_slack_session_thread applies native parity on the inbound bridge:
top-level Slack message + thread-per-message mode => source.thread_id =
its own ts (fresh session, parallel turns). Real thread replies and flat
mode untouched (flat keeps the shared rolling DM session on purpose).

Also introduces the enterprise config shape for relay-fronted Slack:
platforms.relay.extra.slack.<subset of native Slack fields> (nested
object wins; legacy flat extra.reply_in_thread still honoured). All
reply_in_thread reads (send/typing/stop_typing/run.py progress) now
route through one resolver.

gateway/relay/adapter.py

@@ -274,6 +274,7 @@ class RelayAdapter(BasePlatformAdapter):
     async def _on_inbound(self, event) -> None:
         """Bridge a connector-delivered MessageEvent into the normal adapter path."""
         self._capture_scope(event)
+        self._stamp_slack_session_thread(event)
         # Phase 3: a structured prompt answer resolves its waiting primitive
         # (approval/confirm/clarify) and is CONSUMED — it must not also
         # dispatch as a chat message. Unknown/expired prompt ids fall through
@@ -283,6 +284,71 @@ class RelayAdapter(BasePlatformAdapter):
         await self._localize_inbound_media(event)
         await self.handle_message(event)
 
+    def _relay_slack_extra(self) -> Dict[str, Any]:
+        """The Slack-behavior subset of the RELAY platform config.
+
+        Enterprise knob shape (Hermes-config directed, relay-namespaced):
+
+            platforms:
+              relay:
+                extra:
+                  slack:              # supported subset of native Slack fields
+                    reply_in_thread: true
+
+        The native ``platforms.slack`` block keeps meaning "native adapter
+        settings"; relay-fronted Slack reads its subset here. Legacy fallback:
+        a flat key on the relay extra (``extra.reply_in_thread``) still wins
+        when no ``slack`` object exists, preserving current staging configs.
+        """
+        extra = getattr(self.config, "extra", None) or {}
+        sub = extra.get("slack")
+        return sub if isinstance(sub, dict) else extra
+
+    def _effective_reply_in_thread(self) -> bool:
+        """Resolve the thread-per-message vs flat-DM mode for fronted Slack."""
+        try:
+            return bool(self._relay_slack_extra().get("reply_in_thread", True))
+        except Exception:  # noqa: BLE001 - config shape is operator-owned
+            return True
+
+    def _stamp_slack_session_thread(self, event) -> None:
+        """Native session-keying parity for fronted Slack (QA-3).
+
+        Native SlackAdapter's inbound handler stamps ``thread_ts =
+        event.thread_ts or ts`` — every TOP-LEVEL message carries its own ts
+        as ``source.thread_id``, so build_session_key appends it and each
+        top-level message gets a FRESH session (per-message threads ⇒
+        per-message sessions; a 2nd message runs parallel instead of steering
+        the in-flight turn). The connector normalizes a top-level message
+        with thread_id=null, so without this stamp every top-level DM
+        collapses into ONE session key and message 2 pre-empts message 1
+        ("Redirected current run", 2026-07-27 report).
+
+        Only in thread-per-message mode: flat mode keeps the shared rolling
+        DM session on purpose (steer/queue there is the intended UX). Never
+        overwrites a real thread_id (an in-thread reply must keep resolving
+        to its thread's session).
+        """
+        try:
+            src = getattr(event, "source", None)
+            if not src:
+                return
+            platform = getattr(src, "platform", None)
+            if getattr(platform, "value", platform) != Platform.SLACK.value:
+                return
+            if getattr(src, "thread_id", None):
+                return  # real thread — its session key is already correct
+            message_id = getattr(event, "message_id", None) or getattr(
+                src, "message_id", None
+            )
+            if not message_id:
+                return
+            if not self._effective_reply_in_thread():
+                return
+            src.thread_id = str(message_id)
+        except Exception:  # noqa: BLE001 - session stamping must never break inbound
+            logger.debug("slack session-thread stamp failed", exc_info=True)
+
     async def _localize_inbound_media(self, event) -> None:
         """Download connector re-hosted attachments to local temp paths.
 
@@ -875,12 +941,7 @@ class RelayAdapter(BasePlatformAdapter):
         # message to the DM root while progress stayed threaded (2026-07-27
         # report, sibling of the QA-5 prompt bug). Native SlackAdapter only
         # suppresses the anchor when reply_in_thread=false; mirror that.
-        try:
-            reply_in_thread = bool(
-                (self.config.extra or {}).get("reply_in_thread", True)
-            )
-        except Exception:  # noqa: BLE001 - config shape is adapter-owned
-            reply_in_thread = True
+        reply_in_thread = self._effective_reply_in_thread()
         if reply_in_thread:
             # Thread-per-message: the triggering ts is the thread anchor.
             return reply_to
@@ -962,12 +1023,7 @@ class RelayAdapter(BasePlatformAdapter):
             and self._platform_by_chat.get(str(chat_id)) == Platform.SLACK.value
             and self._chat_type_by_chat.get(str(chat_id)) == "dm"
         ):
-            try:
-                reply_in_thread = bool(
-                    (self.config.extra or {}).get("reply_in_thread", True)
-                )
-            except Exception:  # noqa: BLE001 - config shape is adapter-owned
-                reply_in_thread = True
+            reply_in_thread = self._effective_reply_in_thread()
             anchor = self._last_inbound_ts_by_chat.get(str(chat_id))
             if reply_in_thread and anchor:
                 md["thread_id"] = anchor
@@ -1028,12 +1084,7 @@ class RelayAdapter(BasePlatformAdapter):
             not (md.get("thread_id") or md.get("thread_ts"))
             and self._chat_type_by_chat.get(str(chat_id)) == "dm"
         ):
-            try:
-                reply_in_thread = bool(
-                    (self.config.extra or {}).get("reply_in_thread", True)
-                )
-            except Exception:  # noqa: BLE001 - config shape is adapter-owned
-                reply_in_thread = True
+            reply_in_thread = self._effective_reply_in_thread()
             anchor = self._last_inbound_ts_by_chat.get(str(chat_id))
             if reply_in_thread and anchor:
                 md["thread_id"] = anchor

gateway/run.py

@@ -20688,11 +20688,22 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
             _slack_adapter_for_progress = self._adapter_for_source(source)
             if _slack_adapter_for_progress is not None:
                 try:
-                    _progress_reply_in_thread = bool(
-                        _slack_adapter_for_progress.config.extra.get(
-                            "reply_in_thread", True
-                        )
+                    # Relay lane: the adapter owns mode resolution (nested
+                    # platforms.relay.extra.slack subset with flat-key
+                    # fallback). Native lane: read the flat extra as before.
+                    _mode_fn = getattr(
+                        _slack_adapter_for_progress,
+                        "_effective_reply_in_thread",
+                        None,
                     )
+                    if callable(_mode_fn):
+                        _progress_reply_in_thread = bool(_mode_fn())
+                    else:
+                        _progress_reply_in_thread = bool(
+                            _slack_adapter_for_progress.config.extra.get(
+                                "reply_in_thread", True
+                            )
+                        )
                 except Exception:
                     _progress_reply_in_thread = True
         _progress_thread_id = _resolve_progress_thread_id(

tests/gateway/relay/test_relay_slack_prompt_dm_root.py

@@ -306,3 +306,72 @@ async def test_stop_typing_clear_targets_same_synthesized_thread():
         f for f in stub.sent if f["op"] == "typing" and f.get("content") == ""
     ]
     assert clears and clears[-1]["metadata"].get("thread_id") == "1700.0042"
+
+
+# ---------------------------------------------------------------------------
+# QA-3 session keying: a top-level Slack DM message gets its own ts stamped as
+# source.thread_id (native inbound parity) so each message keys a FRESH
+# session in thread-per-message mode; flat mode and real threads untouched.
+# ---------------------------------------------------------------------------
+def _inbound_event(chat_id="D1", message_id="1700.0100", thread_id=None):
+    src = SessionSource(
+        platform=Platform.SLACK, chat_id=chat_id, chat_type="dm",
+        user_id="U1", scope_id="T1", thread_id=thread_id,
+    )
+    return MessageEvent(
+        text="hi", source=src, message_type=MessageType.TEXT,
+        message_id=message_id,
+    )
+
+
+def test_top_level_dm_gets_session_thread_stamp():
+    adapter, _ = _wire("D1", "dm")
+    ev = _inbound_event(message_id="1700.0100")
+    adapter._stamp_slack_session_thread(ev)
+    assert ev.source.thread_id == "1700.0100"
+
+
+def test_two_top_level_messages_key_distinct_sessions():
+    from gateway.session import build_session_key
+    adapter, _ = _wire("D1", "dm")
+    e1 = _inbound_event(message_id="1700.0100")
+    e2 = _inbound_event(message_id="1700.0200")
+    adapter._stamp_slack_session_thread(e1)
+    adapter._stamp_slack_session_thread(e2)
+    k1 = build_session_key(e1.source)
+    k2 = build_session_key(e2.source)
+    assert k1 != k2, "each top-level message must be its own session (QA-3)"
+
+
+def test_real_thread_reply_keeps_its_thread_session():
+    adapter, _ = _wire("D1", "dm")
+    ev = _inbound_event(message_id="1700.0300", thread_id="1700.0100")
+    adapter._stamp_slack_session_thread(ev)
+    assert ev.source.thread_id == "1700.0100", (
+        "an in-thread reply must keep resolving to its thread's session"
+    )
+
+
+def test_flat_mode_keeps_shared_dm_session():
+    adapter, _ = _wire("D1", "dm")
+    adapter.config.extra = {"reply_in_thread": False}
+    ev = _inbound_event(message_id="1700.0400")
+    adapter._stamp_slack_session_thread(ev)
+    assert ev.source.thread_id is None, (
+        "flat mode: shared rolling DM session (steer/queue) is intended UX"
+    )
+
+
+def test_nested_relay_slack_config_subset_wins():
+    """Enterprise knob shape: platforms.relay.extra.slack.reply_in_thread."""
+    adapter, _ = _wire("D1", "dm")
+    adapter.config.extra = {"slack": {"reply_in_thread": False}}
+    assert adapter._effective_reply_in_thread() is False
+    adapter.config.extra = {"slack": {"reply_in_thread": True}}
+    assert adapter._effective_reply_in_thread() is True
+    # Legacy flat key still honoured when no nested object exists.
+    adapter.config.extra = {"reply_in_thread": False}
+    assert adapter._effective_reply_in_thread() is False
+    # Default: thread-per-message.
+    adapter.config.extra = {}
+    assert adapter._effective_reply_in_thread() is True