refactor(relay): remove dead _strip_synthetic_dm_thread; pin the run.py anchor-suppression boundary

Review finding (2026-07-28): every path through _strip_synthetic_dm_thread
returned metadata unmodified — the actual strip was removed when prompts
switched to trusting the run.py thread stamp, leaving a 50-line no-op and
four tests that passed against it (verified by reviewer's negative control).

- delete the function + its _send_prompt call site (verbatim pass-through
  with a pointer comment to the single mode authority)
- rewrite the three pass-through tests as end-to-end placement contracts
  (forward run.py's stamp untouched)
- NEW boundary tests pinning run.py._resolve_progress_thread_id itself:
  flat mode suppresses the synthetic self-anchor / preserves real threads;
  thread mode keeps the first-turn self-anchor. This is the cross-module
  coupling the review flagged as unpinned — if the upstream suppression
  regresses, these fail instead of prompts silently threading.

gateway/relay/adapter.py

@@ -1448,61 +1448,6 @@ class RelayAdapter(BasePlatformAdapter):
             return None
         return state
 
-    def _strip_synthetic_dm_thread(
-        self, chat_id: str, metadata: Optional[Dict[str, Any]]
-    ) -> Optional[Dict[str, Any]]:
-        """Drop the synthetic DM thread anchor from an interactive prompt's metadata.
-
-        A clarify/approval/confirm prompt is emitted mid-turn in reply to the
-        triggering inbound event, so ``metadata`` carries that event's thread
-        context — run.py's ``_thread_metadata_for_source`` stamps
-        ``metadata["thread_id"]`` (and, for Slack, ``metadata["message_id"]`` =
-        the triggering message ts). For a Slack DM with no REAL thread, that
-        ``thread_id`` is the message's own synthetic self-anchor (a session-keying
-        fallback), and forwarding it makes the connector's slackRestSender thread
-        the prompt card UNDER the user's message instead of posting it flat at the
-        DM root — the reported bug ("approval block was put in a thread").
-
-        Native Slack Hermes already suppresses this synthetic DM thread anchor
-        (``SlackAdapter._resolve_thread_ts`` returns ``None`` for a top-level / DM
-        message). We reproduce it here with the same discipline used on the
-        streaming path (``_resolve_reply_to_for_send``):
-
-          Slack DM + thread_id is the synthetic self-anchor  ⇒  strip thread_id.
-
-        A REAL thread (``thread_id`` distinct from the triggering message ts) is
-        left untouched so a prompt raised inside a thread stays in that thread;
-        non-DM / non-Slack chats are never matched. Only the threading keys are
-        removed — tenant scope (``scope_id`` / ``slack_team_id``) and everything
-        else survive so egress routing is unaffected.
-        """
-        if not metadata:
-            return metadata
-        if self._platform_by_chat.get(str(chat_id)) != Platform.SLACK.value:
-            return metadata
-        if self._chat_type_by_chat.get(str(chat_id)) != "dm":
-            return metadata
-        thread_id = metadata.get("thread_id")
-        if not thread_id:
-            return metadata
-        # Trust the run.py stamp (QA-5). The threading MODE is decided in ONE
-        # place — run.py's _resolve_progress_thread_id, which reads
-        # platforms.slack.extra.reply_in_thread:
-        #   * flat mode (reply_in_thread=false): the synthetic self-anchor is
-        #     suppressed THERE, so prompt metadata arrives with NO thread_id and
-        #     this helper is a no-op — the card posts flat at the DM root;
-        #   * thread-per-message mode (default): metadata.thread_id is stamped
-        #     for the whole turn, and on the FIRST turn it legitimately equals
-        #     the triggering message's ts (the synthetic root IS the thread).
-        # The previous unconditional thread_id == message_id strip re-derived
-        # the mode here and got it wrong for thread-per-message: the approval
-        # card (and its resolved-state swap) was exiled to the DM root while
-        # progress bubbles honoured the thread (2026-07-27 mixed-placement
-        # screenshot). Mirror native SlackAdapter._resolve_thread_ts, which
-        # only performs the self-anchor strip when reply_in_thread=false — a
-        # state this lane never sees with an anchor present, per the above.
-        return metadata
-
     async def _send_prompt(
         self,
         chat_id: str,
@@ -1533,7 +1478,11 @@ class RelayAdapter(BasePlatformAdapter):
         # of posting it flat at the DM root (the reported bug). Native Slack
         # Hermes suppresses this synthetic DM thread anchor; drop it here for the
         # same Slack-DM-with-no-real-thread case, matching _resolve_reply_to_for_send.
-        prompt_metadata = self._strip_synthetic_dm_thread(chat_id, metadata)
+        # Prompt metadata is forwarded VERBATIM. The threading mode is decided
+        # in exactly one place — run.py's _resolve_progress_thread_id (flat mode
+        # suppresses the synthetic self-anchor there; thread mode stamps the
+        # turn's thread). Boundary pinned by test_run_py_suppresses_self_anchor*.
+        prompt_metadata = metadata
         action: Dict[str, Any] = {
             "op": "prompt",
             "chat_id": chat_id,

tests/gateway/relay/test_relay_slack_prompt_dm_root.py

@@ -110,14 +110,16 @@ async def test_exec_approval_flat_mode_posts_at_dm_root():
 
 
 # ---------------------------------------------------------------------------
-# Thread-per-message mode: the first-turn self-anchor (thread_id == message_id)
-# IS the thread root — the prompt must stay in the thread (QA-5 regression).
+# Thread-per-message mode, end-to-end placement contract: run.py stamps the
+# turn's thread (first turn: the triggering message's own ts) and the adapter
+# forwards prompt metadata UNTOUCHED — no re-derivation, no strip. Mixed
+# placement (progress threaded, card at root) was the 2026-07-27 regression.
 # ---------------------------------------------------------------------------
 @pytest.mark.asyncio
-async def test_exec_approval_first_turn_self_anchor_stays_in_thread():
-    """Thread-per-message first turn: run.py stamps thread_id = the triggering
-    message's own ts. The approval card must post INTO that thread — stripping
-    it exiled the card to the home channel (2026-07-27 report)."""
+async def test_exec_approval_forwards_run_py_thread_stamp_untouched():
+    """The adapter must forward run.py's thread stamp verbatim: the approval
+    card posts INTO the stamped thread. Any adapter-side re-derivation or
+    strip exiled the card to the home channel (2026-07-27 report)."""
     adapter, stub = _wire("D1", "dm", scope_id="T1")
     md = {
         "thread_id": "1700000000.000100",
@@ -137,7 +139,7 @@ async def test_exec_approval_first_turn_self_anchor_stays_in_thread():
 
 
 @pytest.mark.asyncio
-async def test_clarify_first_turn_self_anchor_stays_in_thread():
+async def test_clarify_forwards_run_py_thread_stamp_untouched():
     adapter, stub = _wire("D1", "dm", scope_id="T1")
     md = {
         "thread_id": "1700000000.000200",
@@ -155,8 +157,8 @@ async def test_clarify_first_turn_self_anchor_stays_in_thread():
 
 
 @pytest.mark.asyncio
-async def test_slash_confirm_first_turn_self_anchor_stays_in_thread():
-    """The stamp-trusting rule covers every prompt surface (single
+async def test_slash_confirm_forwards_run_py_thread_stamp_untouched():
+    """The forward-untouched rule covers every prompt surface (single
     _send_prompt choke point)."""
     adapter, stub = _wire("D1", "dm")
     md = {"thread_id": "1700000000.000300", "message_id": "1700000000.000300"}
@@ -403,3 +405,48 @@ def test_nested_relay_slack_config_subset_wins():
     # Default: thread-per-message.
     adapter.config.extra = {}
     assert adapter._effective_reply_in_thread() is True
+
+
+# ---------------------------------------------------------------------------
+# Cross-module boundary pin (review 2026-07-28): the adapter deliberately has
+# NO prompt-side strip — flat-mode placement depends entirely on run.py's
+# _resolve_progress_thread_id suppressing the synthetic self-anchor upstream.
+# If that suppression regresses, prompt cards silently thread again. These
+# tests pin the boundary in BOTH modes so the coupling is load-bearing.
+# ---------------------------------------------------------------------------
+def test_run_py_suppresses_self_anchor_in_flat_mode():
+    from gateway.run import _resolve_progress_thread_id
+
+    # Flat mode + synthetic self-anchor (thread_id == own message id) => None:
+    # prompt/progress metadata arrives at the adapter with NO thread anchor.
+    assert (
+        _resolve_progress_thread_id(
+            "slack", "1700.001", "1700.001", reply_in_thread=False
+        )
+        is None
+    )
+    # Flat mode + REAL thread (ids differ) => the real thread survives.
+    assert (
+        _resolve_progress_thread_id(
+            "slack", "1699.000", "1700.001", reply_in_thread=False
+        )
+        == "1699.000"
+    )
+
+
+def test_run_py_keeps_self_anchor_in_thread_mode():
+    from gateway.run import _resolve_progress_thread_id
+
+    # Thread-per-message mode: the first-turn self-anchor IS the thread root
+    # and must flow through to the adapter unchanged.
+    assert (
+        _resolve_progress_thread_id(
+            "slack", "1700.001", "1700.001", reply_in_thread=True
+        )
+        == "1700.001"
+    )
+    # No source thread at all: Slack synthesizes the root from the message id.
+    assert (
+        _resolve_progress_thread_id("slack", None, "1700.001", reply_in_thread=True)
+        == "1700.001"
+    )