feat(relay): dm_top_level_threads_as_sessions escape hatch — native session-keying parity

Review finding: native gates per-message DM sessions behind
platforms.slack.extra.dm_top_level_threads_as_sessions; the relay lane
coupled session keying to reply_in_thread alone, so 'threaded replies +
one rolling session' was expressible on native but not here.

Adds the same knob to the relay subset (platforms.relay.extra.slack.
dm_top_level_threads_as_sessions, default true = per-message sessions,
unchanged behavior). false keeps thread-per-message reply placement but
skips the session stamp — one rolling DM session, legacy steer posture.
TDD: opt-out + default-unchanged tests written first.

docs/relay-connector-contract.md

@@ -736,7 +736,8 @@ Current controls (Slack):
 
 | Key | Default | Effect |
 | --- | --- | --- |
-| `reply_in_thread` | `true` | `true`: thread-per-message — each top-level DM message anchors its own thread (status, progress, prompts, final reply all carry that `metadata.thread_id`) and keys its own session, so concurrent messages run in parallel. `false`: flat rolling DM — send-lane frames carry NO thread anchor (stripped, not omitted), one shared session per DM. |
+| `reply_in_thread` | `true` | `true`: thread-per-message — each top-level DM message anchors its own thread (status, progress, prompts, final reply all carry that `metadata.thread_id`). `false`: flat rolling DM — send-lane frames carry NO thread anchor (stripped, not omitted), one shared session per DM. |
+| `dm_top_level_threads_as_sessions` | `true` | Native-parity escape hatch (mirrors `platforms.slack.extra.dm_top_level_threads_as_sessions`). `true`: in thread-per-message mode each top-level DM message keys its own session, so concurrent messages run in parallel. `false`: threaded reply placement is kept but the session stamp is skipped — one rolling DM session (legacy steer/queue posture). No effect in flat mode, which always keeps the single rolling session. |
 
 Typing/status frames always carry the triggering-ts anchor when one is known
 (liveliness is unconditional, both modes): Slack's status line is

gateway/relay/adapter.py

@@ -311,6 +311,26 @@ class RelayAdapter(BasePlatformAdapter):
         except Exception:  # noqa: BLE001 - config shape is operator-owned
             return True
 
+    def _dm_top_level_threads_as_sessions(self) -> bool:
+        """Native-parity escape hatch: per-message DM sessions on/off.
+
+        Mirrors native SlackAdapter._dm_top_level_threads_as_sessions
+        (platforms.slack.extra.dm_top_level_threads_as_sessions). Default
+        True: in thread-per-message mode each top-level DM message keys its
+        own session (parallel turns). Set
+        platforms.relay.extra.slack.dm_top_level_threads_as_sessions: false
+        to keep threaded reply PLACEMENT but ONE rolling DM session — the
+        legacy steer/queue posture, decoupled from reply_in_thread.
+        """
+        try:
+            return bool(
+                self._relay_slack_extra().get(
+                    "dm_top_level_threads_as_sessions", 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 DMs.
 
@@ -345,6 +365,8 @@ class RelayAdapter(BasePlatformAdapter):
                 return
             if not self._effective_reply_in_thread():
                 return
+            if not self._dm_top_level_threads_as_sessions():
+                return  # opt-out: threaded replies, one rolling session
             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)

tests/gateway/relay/test_relay_slack_prompt_dm_root.py

@@ -450,3 +450,37 @@ def test_run_py_keeps_self_anchor_in_thread_mode():
         _resolve_progress_thread_id("slack", None, "1700.001", reply_in_thread=True)
         == "1700.001"
     )
+
+
+# ---------------------------------------------------------------------------
+# Native parity escape hatch: platforms.relay.extra.slack.
+# dm_top_level_threads_as_sessions=false keeps threaded replies but ONE
+# rolling DM session (mirrors native SlackAdapter._dm_top_level_threads_as_sessions).
+# Without the knob, reply_in_thread alone couples placement AND session
+# keying — a posture native operators can express and relay ones could not.
+# ---------------------------------------------------------------------------
+@pytest.mark.asyncio
+async def test_session_stamp_opt_out_keeps_rolling_dm_session():
+    adapter, stub = _wire("D1", "dm")
+    adapter.config.extra = {
+        "slack": {
+            "reply_in_thread": True,
+            "dm_top_level_threads_as_sessions": False,
+        }
+    }
+    event = _inbound_event("D1", message_id="1700.0001", thread_id=None)
+    adapter._stamp_slack_session_thread(event)
+    assert getattr(event.source, "thread_id", None) is None, (
+        "opt-out: top-level DM must NOT be stamped — one rolling session"
+    )
+
+
+@pytest.mark.asyncio
+async def test_session_stamp_default_remains_per_message():
+    adapter, stub = _wire("D1", "dm")
+    adapter.config.extra = {"slack": {"reply_in_thread": True}}
+    event = _inbound_event("D1", message_id="1700.0002", thread_id=None)
+    adapter._stamp_slack_session_thread(event)
+    assert getattr(event.source, "thread_id", None) == "1700.0002", (
+        "default (native parity): per-message sessions stay on"
+    )