From 33833e232e5195ce55722bafa50701c52aa24365 Mon Sep 17 00:00:00 2001 From: Ben Barclay Date: Wed, 29 Jul 2026 12:03:00 -0700 Subject: [PATCH] fix(relay): apply the Slack thread anchor on the media lane too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The DM thread-anchor contract was resolved only in send(). _send_media() — backing send_image, send_image_file, send_voice, send_video and send_document — passed reply_to straight to the frame and never touched metadata, so attachments egressing through the same connector-side Slack sender got both failure shapes this branch set out to remove: flat mode → reply_to survives, the image threads UNDER the user's DM message (the original reported symptom) thread mode → no metadata.thread_id, and threadTs() never reads reply_to, so the image lands in the home channel instead of the per-message thread Both are reachable: gateway/run.py delivers agent artifacts through send_voice/send_document. Extract the three steps that must always happen together (mode gate, mirrored reply_to_message_id strip, metadata promotion) into _apply_slack_thread_anchor and route BOTH lanes through it, so text and media cannot drift again. The media lane copies caller metadata rather than mutating it — these helpers are called in loops with a shared mapping. Also fold send_typing/stop_typing's duplicated status-anchor blocks into _with_status_thread_anchor. They had already drifted (stop_typing omitted the platform check) and the clear must target the thread the heartbeat set or the status line sticks until Slack's own timeout. Tests: media lane pinned in both modes plus the channel and no-caller-mutation cases; verified as real by reverting the fix and watching them fail. --- gateway/relay/adapter.py | 148 +++++++++++------- .../relay/test_relay_slack_dm_streaming.py | 98 +++++++++++- 2 files changed, 191 insertions(+), 55 deletions(-) diff --git a/gateway/relay/adapter.py b/gateway/relay/adapter.py index d88815d676a..7c61ff4763d 100644 --- a/gateway/relay/adapter.py +++ b/gateway/relay/adapter.py @@ -904,32 +904,12 @@ class RelayAdapter(BasePlatformAdapter): if self._transport is None: return SendResult(success=False, error="no transport") # Native _resolve_thread_ts parity: a Slack DM reply must post flat at - # the DM root, not threaded under the triggering message. Drop the - # synthetic self-anchor from BOTH the top-level reply_to and the mirrored - # metadata.reply_to_message_id so the connector can't thread on either. - effective_reply_to = self._resolve_reply_to_for_send( + # the DM root, not threaded under the triggering message. One shared + # helper resolves the anchor for EVERY egress lane (see + # _apply_slack_thread_anchor) so the text and media lanes cannot drift. + effective_reply_to = self._apply_slack_thread_anchor( chat_id, reply_to, send_metadata ) - if effective_reply_to is None and reply_to is not None: - send_metadata.pop("reply_to_message_id", None) - # The connector's Slack sender THREADS ON METADATA ONLY — - # threadTs() reads metadata.thread_id/thread_ts and never looks at - # the frame's reply_to. A send whose only threading signal is - # reply_to (base.py's final-reply and fallback lanes build metadata - # from source.thread_id = None for a top-level DM) would post to the - # home channel even though _resolve_reply_to_for_send kept the - # anchor. Promote the surviving anchor into metadata.thread_id so - # the wire carries it where the connector actually reads it. Only - # when the mode gate kept the anchor (thread-per-message / real - # thread) — flat mode already nulled effective_reply_to above. - if ( - effective_reply_to is not None - and self._platform_by_chat.get(str(chat_id)) == Platform.SLACK.value - and not ( - send_metadata.get("thread_id") or send_metadata.get("thread_ts") - ) - ): - send_metadata["thread_id"] = str(effective_reply_to) result = await self._transport.send_outbound( { "op": "send", @@ -1010,6 +990,80 @@ class RelayAdapter(BasePlatformAdapter): # Flat mode: synthetic DM self-anchor — post flat at the DM root. return None + def _apply_slack_thread_anchor( + self, + chat_id: str, + reply_to: Optional[str], + metadata: Dict[str, Any], + *, + mirror_key: str = "reply_to_message_id", + ) -> Optional[str]: + """Resolve the outbound Slack thread anchor for ONE egress frame. + + The single choke point every send lane goes through — text (``send``) + and media (``send_media``) alike. It does three things that must always + happen together, and previously only happened on the text lane: + + 1. Mode gate: ``_resolve_reply_to_for_send`` drops the synthetic DM + self-anchor in flat mode, keeps it in thread-per-message mode. + 2. Mirror strip: when the anchor is dropped, remove the mirrored + ``metadata.reply_to_message_id`` too, so the connector cannot + thread on the copy we forgot about. + 3. Anchor promotion: the connector's Slack sender THREADS ON METADATA + ONLY — ``threadTs()`` reads ``metadata.thread_id``/``thread_ts`` + and never looks at the frame's ``reply_to``. A surviving anchor is + promoted into ``metadata.thread_id`` or the message silently lands + in the home channel instead of the per-message thread. + + ``metadata`` is mutated in place; the effective ``reply_to`` is + returned. Non-Slack and non-DM chats are untouched by (1), and (3) is + Slack-only, so other fronted platforms keep their existing behaviour. + """ + effective_reply_to = self._resolve_reply_to_for_send( + chat_id, reply_to, metadata + ) + if effective_reply_to is None and reply_to is not None: + metadata.pop(mirror_key, None) + if ( + effective_reply_to is not None + and self._platform_by_chat.get(str(chat_id)) == Platform.SLACK.value + and not (metadata.get("thread_id") or metadata.get("thread_ts")) + ): + metadata["thread_id"] = str(effective_reply_to) + return effective_reply_to + + def _with_status_thread_anchor( + self, chat_id: str, metadata: Optional[Dict[str, Any]] + ) -> Dict[str, Any]: + """Copy ``metadata`` with the typing/status thread anchor applied. + + Slack's status line is THREAD-scoped: the connector's typing case + no-ops without a thread anchor, and the typing lane's metadata + (base.py ``_thread_metadata_for_source``) carries none for a top-level + DM (``source.thread_id`` is None). Synthesize it from the per-chat + inbound-ts cache, exactly as native ``send_typing`` resolves + ``thread_ts`` from ``metadata.message_id``. + + Unconditional across both modes: in flat mode the send lane strips its + own anchors (see ``_apply_slack_thread_anchor``), so the status anchor + cannot leak into reply placement, and ``setStatus`` clears without + leaving a message artifact. + + Shared by ``send_typing`` and ``stop_typing`` — the clear MUST target + the same thread the heartbeat set, or the status line sticks until + Slack's own timeout. Keeping one implementation is what guarantees it. + """ + md = dict(metadata or {}) + if ( + not (md.get("thread_id") or md.get("thread_ts")) + and self._platform_by_chat.get(str(chat_id)) == Platform.SLACK.value + and self._chat_type_by_chat.get(str(chat_id)) == "dm" + ): + anchor = self._last_inbound_ts_by_chat.get(str(chat_id)) + if anchor: + md["thread_id"] = anchor + return md + async def edit_message( self, chat_id: str, @@ -1079,22 +1133,7 @@ class RelayAdapter(BasePlatformAdapter): # like native send_typing resolves thread_ts from metadata.message_id. # Flat mode (reply_in_thread=false) keeps the no-anchor no-op: there # is no thread and must not be one (#18859). - md = dict(metadata or {}) - if ( - not (md.get("thread_id") or md.get("thread_ts")) - and self._platform_by_chat.get(str(chat_id)) == Platform.SLACK.value - and self._chat_type_by_chat.get(str(chat_id)) == "dm" - ): - # Thread mode: status targets the per-message thread. - # Flat mode: the status STILL anchors to the triggering ts — - # setStatus renders in the footer space and clears without a - # message artifact, and flat sends strip their anchors - # so reply placement cannot inherit it. Unconditional: liveliness - # is not a preference, it ships in whatever form the mode - # supports (no speculative opt-out knob). - anchor = self._last_inbound_ts_by_chat.get(str(chat_id)) - if anchor: - md["thread_id"] = anchor + md = self._with_status_thread_anchor(chat_id, metadata) # Rich status parity: run.py's live-status lane stashes the # current per-tool phrase via set_status_text() (base class store). # Carry it as the typing frame's content so the connector's Slack @@ -1143,18 +1182,10 @@ class RelayAdapter(BasePlatformAdapter): platform = self._platform_by_chat.get(str(chat_id)) if platform != Platform.SLACK.value: return - # Clear must target the SAME thread the heartbeat set: apply - # the identical synthetic-anchor rule as send_typing, or the clear - # frame no-ops threadless and the status line sticks until Slack's - # own timeout. - md = dict(metadata or {}) - if ( - not (md.get("thread_id") or md.get("thread_ts")) - and self._chat_type_by_chat.get(str(chat_id)) == "dm" - ): - anchor = self._last_inbound_ts_by_chat.get(str(chat_id)) - if anchor: - md["thread_id"] = anchor + # Clear must target the SAME thread the heartbeat set, or the clear + # frame no-ops threadless and the status line sticks until Slack's own + # timeout. Shared helper with send_typing so the two guards cannot drift. + md = self._with_status_thread_anchor(chat_id, metadata) try: await self._transport.send_outbound( { @@ -1286,14 +1317,23 @@ class RelayAdapter(BasePlatformAdapter): if not uploaded: return None source_url = uploaded + # Same Slack thread-anchor contract as the text lane (send). Media + # frames egress through the connector's Slack sender too, so an + # unresolved anchor threads an image under the user's DM message in + # flat mode, and loses the per-message thread entirely in thread mode + # (threadTs() reads metadata only). Route through the shared helper. + media_metadata: Dict[str, Any] = dict(metadata or {}) + effective_reply_to = self._apply_slack_thread_anchor( + chat_id, reply_to, media_metadata + ) action: Dict[str, Any] = { "op": "send_media", "chat_id": chat_id, "media_kind": media_kind, "source_url": source_url, "content": caption or "", - "reply_to": reply_to, - "metadata": self._with_scope(chat_id, metadata), + "reply_to": effective_reply_to, + "metadata": self._with_scope(chat_id, media_metadata), } if filename: action["filename"] = filename diff --git a/tests/gateway/relay/test_relay_slack_dm_streaming.py b/tests/gateway/relay/test_relay_slack_dm_streaming.py index 2438609d89c..a812f358ec3 100644 --- a/tests/gateway/relay/test_relay_slack_dm_streaming.py +++ b/tests/gateway/relay/test_relay_slack_dm_streaming.py @@ -87,7 +87,7 @@ async def test_slack_dm_reply_keeps_anchor_in_thread_per_message_mode(): assert frame["reply_to"] == "1700.0001", ( "thread-per-message: the triggering ts anchors the final reply" ) - # QA-7: the connector's Slack sender threads on metadata.thread_id ONLY + # The connector's Slack sender threads on metadata.thread_id ONLY # (threadTs() never reads the frame's reply_to), so the surviving anchor # must be promoted into metadata for the send to actually thread. assert (frame["metadata"] or {}).get("thread_id") == "1700.0001" @@ -263,3 +263,99 @@ async def test_slack_dm_stream_consumer_threads_in_thread_per_message_mode(): 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"]} + + +# --------------------------------------------------------------------------- +# The media lane obeys the SAME thread-anchor contract as the text lane. +# +# send() and _send_media() both egress through the connector's Slack sender, +# so an anchor resolved on only one of them threads an image under the user's +# DM message in flat mode, and loses the per-message thread in thread mode +# (threadTs() reads metadata only). Both lanes route through +# _apply_slack_thread_anchor; these pin that they stay in agreement. +# --------------------------------------------------------------------------- +def _media_wire(chat_id: str, chat_type: str): + """Like _wire, but the descriptor advertises the send_media op.""" + desc = _slack_desc(supported_ops=("send", "edit", "typing", "send_media")) + stub = StubConnector(desc) + adapter = RelayAdapter(PlatformConfig(), desc, transport=stub) + src = SessionSource( + platform=Platform.SLACK, + chat_id=chat_id, + chat_type=chat_type, + user_id="U1", + ) + adapter._capture_scope( + MessageEvent(text="hi", source=src, message_type=MessageType.TEXT) + ) + return adapter, stub + + +@pytest.mark.asyncio +async def test_slack_dm_media_keeps_and_promotes_anchor_in_thread_mode(): + """Thread-per-message: an image must land in the per-message thread. The + connector threads on metadata.thread_id only, so the surviving anchor has + to be promoted there — a bare reply_to would post to the home channel.""" + adapter, stub = _media_wire("D1", "dm") + await adapter.send_image("D1", "https://example.com/x.png", reply_to="1700.0001") + frame = [f for f in stub.sent if f["op"] == "send_media"][-1] + assert frame["reply_to"] == "1700.0001" + assert (frame["metadata"] or {}).get("thread_id") == "1700.0001", ( + "media frame must carry the anchor where the connector reads it" + ) + + +@pytest.mark.asyncio +async def test_slack_dm_media_drops_synthetic_anchor_in_flat_mode(): + """Flat mode: the media frame drops the synthetic self-anchor exactly as + the text lane does, so the image posts flat at the DM root instead of + threading under the user's message, and invents no thread (#18859).""" + adapter, stub = _media_wire("D1", "dm") + adapter.config.extra = {"reply_in_thread": False} + await adapter.send_image("D1", "https://example.com/x.png", reply_to="1700.0001") + frame = [f for f in stub.sent if f["op"] == "send_media"][-1] + assert frame["reply_to"] is None + assert "thread_id" not in (frame["metadata"] or {}) + assert "thread_ts" not in (frame["metadata"] or {}) + + +@pytest.mark.asyncio +async def test_slack_channel_media_anchor_untouched(): + """Non-DM chats are outside the synthetic-anchor rule: a channel media + send keeps its reply_to unchanged.""" + adapter, stub = _media_wire("C1", "channel") + await adapter.send_image("C1", "https://example.com/x.png", reply_to="1700.0009") + frame = [f for f in stub.sent if f["op"] == "send_media"][-1] + assert frame["reply_to"] == "1700.0009" + + +@pytest.mark.asyncio +async def test_media_caller_metadata_not_mutated(): + """The anchor promotion must not leak into the caller's dict — media + helpers are called in loops with a shared metadata mapping.""" + adapter, stub = _media_wire("D1", "dm") + caller_md = {"user_id": "U1"} + await adapter.send_image( + "D1", "https://example.com/x.png", reply_to="1700.0001", metadata=caller_md + ) + assert caller_md == {"user_id": "U1"}, "caller metadata was mutated in place" + + +# --------------------------------------------------------------------------- +# The status clear targets the same thread the heartbeat set. +# --------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_typing_and_clear_share_one_status_anchor(): + """send_typing and stop_typing resolve the anchor through one helper: a + clear that no-ops threadless leaves the status line stuck until Slack's + own timeout.""" + adapter, stub = _wire("D1", "dm") + adapter._last_inbound_ts_by_chat["D1"] = "1700.0001" + await adapter.send_typing("D1") + await adapter.stop_typing("D1") + typing_frames = [f for f in stub.sent if f["op"] == "typing"] + assert len(typing_frames) == 2 + anchors = [(f["metadata"] or {}).get("thread_id") for f in typing_frames] + assert anchors == ["1700.0001", "1700.0001"], ( + "the clear must target the thread the heartbeat set" + )