From 467534b43ed87e55861d0ef87ebd7822f313bc3d Mon Sep 17 00:00:00 2001 From: Victor Kyriazakos Date: Mon, 27 Jul 2026 15:03:51 +0000 Subject: [PATCH] =?UTF-8?q?fix(relay):=20typing/status=20targets=20the=20p?= =?UTF-8?q?er-message=20thread=20=E2=80=94=20synthesize=20the=20anchor=20f?= =?UTF-8?q?rom=20the=20inbound=20ts=20(QA-1)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Slack's thinking-status line (thread replies footer, plain chat:write — no assistant scopes needed) is thread-only: the connector's typing case no-ops without thread_ts. The typing lane's metadata has no anchor for a top-level DM (base.py builds from source.thread_id = None), so every status heartbeat was silently dropped — the trace showed typing frames with meta_keys=['user_id'] only. Cache the triggering message ts per chat on inbound (_capture_scope) and synthesize metadata.thread_id on send_typing/stop_typing in thread-per-message mode, mirroring native send_typing's _resolve_thread_ts(metadata.message_id). Flat mode unchanged (#18859); real-thread metadata wins over the cache; the clear frame targets the same synthesized thread so the status never sticks. --- gateway/relay/adapter.py | 66 ++++++++++++++++++- .../relay/test_relay_slack_prompt_dm_root.py | 60 +++++++++++++++++ 2 files changed, 124 insertions(+), 2 deletions(-) diff --git a/gateway/relay/adapter.py b/gateway/relay/adapter.py index fff21bb6a94..8b125a22adc 100644 --- a/gateway/relay/adapter.py +++ b/gateway/relay/adapter.py @@ -82,6 +82,10 @@ class RelayAdapter(BasePlatformAdapter): # synthetic reply_to in _resolve_thread_ts; the relay lane needs the same # disambiguation, and it needs the chat_type to know a chat is a DM. self._chat_type_by_chat: Dict[str, str] = {} + # chat_id -> last triggering message ts (Slack). The typing/status + # lane's synthetic thread anchor in thread-per-message mode (QA-1); + # see _capture_scope and send_typing. + self._last_inbound_ts_by_chat: Dict[str, str] = {} # chat_id -> the UNDERLYING platform (e.g. "discord", "telegram") this # chat belongs to (Phase 1.5 multi-platform-per-agent). One relay adapter # fronts N platforms on one WS; an outbound reply must egress through the @@ -378,6 +382,19 @@ class RelayAdapter(BasePlatformAdapter): chat_type = getattr(src, "chat_type", None) if chat_type: self._chat_type_by_chat[str(chat)] = str(chat_type) + # Triggering message ts (QA-1): the typing/status lane's metadata + # (base.py _thread_metadata_for_source) carries NO thread anchor + # for a top-level DM, but in thread-per-message mode the status + # must target the per-message thread (its root = this ts). Cache + # it per chat so send_typing can synthesize the anchor, mirroring + # native send_typing's _resolve_thread_ts(metadata.message_id). + # NOTE: message_id lives on the EVENT (MessageEvent), not the + # source — fall back to source for defensive coverage. + message_id = getattr(event, "message_id", None) or getattr( + src, "message_id", None + ) + if message_id: + self._last_inbound_ts_by_chat[str(chat)] = str(message_id) except Exception: # noqa: BLE001 - scope tracking must never break inbound pass @@ -909,6 +926,33 @@ class RelayAdapter(BasePlatformAdapter): """ if self._transport is None: return + # Thread anchor for the status surface (QA-1). Slack's status line + # ("is thinking…" in the thread's replies footer — works with plain + # chat:write, confirmed on native no-assistant bots) is THREAD-only: + # the connector's typing case no-ops without a thread_ts. But the + # typing lane's metadata (base.py _thread_metadata_for_source) has no + # anchor for a top-level DM — source.thread_id is None — so every + # heartbeat was silently dropped. In thread-per-message mode the + # turn's thread root IS the triggering message ts (run.py's synthetic + # root); synthesize it here from the per-chat inbound cache, exactly + # 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" + ): + 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 + anchor = self._last_inbound_ts_by_chat.get(str(chat_id)) + if reply_in_thread and anchor: + md["thread_id"] = anchor # Rich status parity (QA-1): 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 @@ -921,7 +965,7 @@ class RelayAdapter(BasePlatformAdapter): frame: Dict[str, Any] = { "op": "typing", "chat_id": chat_id, - "metadata": self._with_scope(chat_id, metadata), + "metadata": self._with_scope(chat_id, md), } phrase = getattr(self, "_status_text", {}).get(str(chat_id)) if phrase: @@ -957,13 +1001,31 @@ 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 (QA-1): 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" + ): + 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 + anchor = self._last_inbound_ts_by_chat.get(str(chat_id)) + if reply_in_thread and anchor: + md["thread_id"] = anchor try: await self._transport.send_outbound( { "op": "typing", "chat_id": chat_id, "content": "", - "metadata": self._with_scope(chat_id, metadata), + "metadata": self._with_scope(chat_id, md), }, platform=platform, ) 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 5e48e972b8d..2656cf32c03 100644 --- a/tests/gateway/relay/test_relay_slack_prompt_dm_root.py +++ b/tests/gateway/relay/test_relay_slack_prompt_dm_root.py @@ -246,3 +246,63 @@ async def test_typing_carries_live_status_phrase(): assert "content" not in typing[-1], ( "cleared phrase must omit content (empty string means CLEAR on Slack)" ) + + +# --------------------------------------------------------------------------- +# QA-1 status thread anchor: typing frames synthesize the per-message thread +# root in thread-per-message mode (the status line is thread-only on Slack). +# --------------------------------------------------------------------------- +def _wire_with_ts(chat_id, chat_type, message_id, **kw): + adapter, stub = _wire(chat_id, chat_type, **kw) + src = SessionSource( + platform=Platform.SLACK, chat_id=chat_id, chat_type=chat_type, + user_id="U1", scope_id=kw.get("scope_id"), + ) + ev = MessageEvent( + text="hi", source=src, message_type=MessageType.TEXT, message_id=message_id + ) + adapter._capture_scope(ev) + return adapter, stub + + +@pytest.mark.asyncio +async def test_typing_synthesizes_thread_anchor_in_thread_mode(): + """Top-level DM turn, thread-per-message mode: the typing frame gains the + triggering ts as thread_id so the connector's setStatus targets the + per-message thread instead of no-oping threadless.""" + adapter, stub = _wire_with_ts("D1", "dm", "1700.0042") + await adapter.send_typing("D1", metadata=None) + typing = [f for f in stub.sent if f["op"] == "typing"] + assert typing and typing[-1]["metadata"].get("thread_id") == "1700.0042" + + +@pytest.mark.asyncio +async def test_typing_keeps_no_anchor_in_flat_mode(): + """Flat mode: no synthetic thread for the status either (#18859).""" + adapter, stub = _wire_with_ts("D1", "dm", "1700.0042") + adapter.config.extra = {"reply_in_thread": False} + await adapter.send_typing("D1", metadata=None) + typing = [f for f in stub.sent if f["op"] == "typing"] + assert typing and "thread_id" not in typing[-1]["metadata"] + + +@pytest.mark.asyncio +async def test_typing_honours_real_thread_anchor(): + """Metadata that already names a thread wins over the synthetic cache.""" + adapter, stub = _wire_with_ts("D1", "dm", "1700.0042") + await adapter.send_typing("D1", metadata={"thread_id": "1699.9000"}) + typing = [f for f in stub.sent if f["op"] == "typing"] + assert typing[-1]["metadata"]["thread_id"] == "1699.9000" + + +@pytest.mark.asyncio +async def test_stop_typing_clear_targets_same_synthesized_thread(): + """The clear frame targets the same synthesized thread as the heartbeat + (else the status line sticks).""" + adapter, stub = _wire_with_ts("D1", "dm", "1700.0042") + await adapter.send_typing("D1", metadata=None) + await adapter.stop_typing("D1", metadata=None) + clears = [ + 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"