diff --git a/gateway/relay/adapter.py b/gateway/relay/adapter.py
index 4fcdb1a2ac5..c7eaa1a981c 100644
--- a/gateway/relay/adapter.py
+++ b/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
diff --git a/gateway/run.py b/gateway/run.py
index aac6a192555..5e7f159011e 100644
--- a/gateway/run.py
+++ b/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(
diff --git a/tests/gateway/relay/test_relay_slack_prompt_dm_root.py b/tests/gateway/relay/test_relay_slack_prompt_dm_root.py
index 2656cf32c03..098b1fcadde 100644
--- a/tests/gateway/relay/test_relay_slack_prompt_dm_root.py
+++ b/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