fix(relay): final DM reply honors thread-per-message mode — gate the reply_to strip on reply_in_thread (QA-6)

_resolve_reply_to_for_send dropped the triggering-ts reply_to on every
Slack DM with no metadata thread_id. But the final-reply lane (platforms/
base.py) builds metadata from source.thread_id only — None for a top-level
DM — so in thread-per-message mode that reply_to is the final reply's ONLY
threading signal, and stripping it exiled the final message to the DM root
while progress bubbles stayed threaded (sibling of the QA-5 prompt bug).

Mirror native _resolve_thread_ts: suppress the synthetic anchor only when
platforms.slack.extra.reply_in_thread=false. Flat mode behavior unchanged;
real threads and channels unchanged.

gateway/relay/adapter.py

@@ -814,7 +814,25 @@ class RelayAdapter(BasePlatformAdapter):
         if md.get("thread_id") or md.get("thread_ts"):
             # A real thread was resolved by run.py — honour it.
             return reply_to
-        # Synthetic DM self-anchor: post flat at the DM root (native parity).
+        # Mode gate (native _resolve_thread_ts parity). The final-reply lane
+        # (gateway/platforms/base.py) builds metadata from source.thread_id
+        # ONLY — for a top-level DM that is None, so in thread-per-message
+        # mode the triggering-ts reply_to here is the final reply's ONLY
+        # threading signal (run.py's synthetic root feeds just the
+        # progress/status lane). Dropping it unconditionally exiled the final
+        # 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
+        if reply_in_thread:
+            # Thread-per-message: the triggering ts is the thread anchor.
+            return reply_to
+        # Flat mode: synthetic DM self-anchor — post flat at the DM root.
         return None
 
     async def edit_message(

tests/gateway/relay/test_relay_slack_dm_streaming.py

@@ -73,18 +73,33 @@ def _wire(chat_id: str, chat_type: str, *, user_id="U1", scope_id=None):
 # The pure disambiguation contract (RelayAdapter.send)
 # ---------------------------------------------------------------------------
 @pytest.mark.asyncio
-async def test_slack_dm_reply_drops_synthetic_thread_anchor():
-    """A Slack DM reply with no real thread posts FLAT: reply_to is dropped so
-    the connector cannot thread it under the triggering message."""
+async def test_slack_dm_reply_keeps_anchor_in_thread_per_message_mode():
+    """Default mode (reply_in_thread=True, thread-per-message): the triggering
+    ts reply_to is the final reply's ONLY threading signal (base.py builds
+    metadata from source.thread_id, which is None for a top-level DM) — it
+    must be KEPT so the final message lands in the per-message thread with
+    the progress bubbles (2026-07-27 mixed-placement report)."""
     adapter, stub = _wire("D1", "dm")
     await adapter.send("D1", "the answer", reply_to="1700.0001")
     assert len(stub.sent) == 1
     frame = stub.sent[0]
     assert frame["op"] == "send"
-    # The synthetic self-anchor is suppressed on BOTH surfaces.
+    assert frame["reply_to"] == "1700.0001", (
+        "thread-per-message: the triggering ts anchors the final reply"
+    )
+
+
+@pytest.mark.asyncio
+async def test_slack_dm_reply_drops_synthetic_anchor_in_flat_mode():
+    """Flat mode (reply_in_thread=False): the synthetic self-anchor is dropped
+    so the reply posts flat at the DM root (native _resolve_thread_ts parity)
+    and no synthetic thread is invented (#18859)."""
+    adapter, stub = _wire("D1", "dm")
+    adapter.config.extra = {"reply_in_thread": False}
+    await adapter.send("D1", "the answer", reply_to="1700.0001")
+    frame = stub.sent[0]
     assert frame["reply_to"] is None
     assert "thread_id" not in (frame["metadata"] or {})
-    # And no synthetic thread_id was invented (the #18859 landmine).
     assert "thread_ts" not in (frame["metadata"] or {})
 
 
@@ -165,8 +180,13 @@ async def test_slack_dm_stream_consumer_edits_own_ts_not_flat():
 
     The connector returns a real message_id for the flat first send, so edit
     support must stay on and at least one edit op must be emitted (progressive
-    streaming), identical to a thread. No synthetic thread is created."""
+    streaming), identical to a thread. No synthetic thread is created.
+
+    Runs in EXPLICIT flat mode (reply_in_thread=False) — that is the mode this
+    contract belongs to; the default thread-per-message path is covered by
+    test_slack_dm_stream_consumer_threads_in_thread_per_message_mode."""
     adapter, stub = _wire("D1", "dm")
+    adapter.config.extra = {"reply_in_thread": False}
     consumer = await _drive_stream(
         adapter,
         "D1",
@@ -218,3 +238,24 @@ async def test_slack_thread_stream_consumer_still_threads_and_streams():
     # Thread preserved: the real thread_id rides along and reply_to is kept.
     assert first_send["metadata"]["thread_id"] == "1699.9000"
     assert first_send["reply_to"] == "1700.0002"
+
+
+@pytest.mark.asyncio
+async def test_slack_dm_stream_consumer_threads_in_thread_per_message_mode():
+    """Default mode: the DM stream's first send keeps the triggering-ts anchor
+    so the streamed final reply lands in the per-message thread; edits still
+    target the reply's own ts."""
+    adapter, stub = _wire("D1", "dm")
+    consumer = await _drive_stream(
+        adapter,
+        "D1",
+        metadata=None,
+        initial_reply_to_id="1700.0001",
+        chat_type="dm",
+    )
+    first_send = stub.sent[0]
+    assert first_send["op"] == "send"
+    assert first_send["reply_to"] == "1700.0001"
+    assert consumer.message_id and consumer.message_id != "__no_edit__"
+    edit_ids = {f["message_id"] for f in stub.sent if f["op"] == "edit"}
+    assert edit_ids <= {stub.next_send_result["message_id"]}