diff --git a/docs/relay-connector-contract.md b/docs/relay-connector-contract.md index 3698b55432a..184ce905e65 100644 --- a/docs/relay-connector-contract.md +++ b/docs/relay-connector-contract.md @@ -711,7 +711,56 @@ per-gateway secret and the same host as `/relay/provision`. --- -## 8. Versioning policy +## 8. Gateway-side platform behavior controls (enterprise) + +Enterprise deployments configure fronted-platform behavior on the GATEWAY +side, under `platforms.relay.extra.` — a supported subset of that +platform's native options. The native platform block (e.g. `platforms.slack`) +is not read on the relay lane; the connector receives the *outcome* of these +controls as frame metadata (§4) and executes mechanically — it holds no +platform behavior policy of its own. + +```yaml +platforms: + relay: + extra: + slack: + reply_in_thread: true # default +``` + +Resolution: nested `extra.` object wins → legacy flat key on +`extra` honored as fallback → default. Source of truth: +`RelayAdapter._effective_reply_in_thread` (`gateway/relay/adapter.py`). +Values coerce exactly as the native Slack adapter's do — `1/true/yes/on` +(case-insensitive, whitespace-trimmed) are ON, anything else is OFF — so a +YAML-quoted `"false"` turns a knob off rather than being read as a truthy +string. + +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`). `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 +thread-scoped, and in flat mode the send-side anchor strip guarantees the +status anchor can never leak into reply placement. Semantics of the native +key: see `website/docs/user-guide/messaging/slack.md`. + +Thread-anchor resolution applies to EVERY send lane — text (`send`) and media +(`send_media`) alike — through one choke point +(`RelayAdapter._apply_slack_thread_anchor`). Media frames egress via the same +connector-side Slack sender, which threads on `metadata.thread_id` only, so an +attachment resolves its anchor identically to a text reply: promoted into +metadata in thread-per-message mode, stripped in flat mode. + +Changes take effect on gateway restart; no connector involvement. + +--- + +## 9. Versioning policy - `contract_version` is an int; bump **only** for additive changes during the experimental phase (new optional fields, new `op`s). diff --git a/gateway/relay/adapter.py b/gateway/relay/adapter.py index 281602922fa..f4170d16444 100644 --- a/gateway/relay/adapter.py +++ b/gateway/relay/adapter.py @@ -72,6 +72,20 @@ class RelayAdapter(BasePlatformAdapter): # recipient's author binding; we re-attach this user_id as # metadata.user_id on the outbound action so it can. See _capture_scope. self._dm_user_by_chat: Dict[str, str] = {} + # chat_id -> chat_type (e.g. "dm", "channel", "group") learned from the + # inbound event. Used to reproduce native Slack's synthetic-DM-thread + # suppression on the relay lane: a DM streaming reply carries + # reply_to= as its edit anchor, but the connector + # maps a raw reply_to to a Slack thread_ts — so a plain DM reply would be + # threaded UNDER the user's message (and lose progressive edit streaming) + # instead of posting flat at the DM root. Native SlackAdapter drops that + # 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; + # 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 @@ -123,6 +137,27 @@ class RelayAdapter(BasePlatformAdapter): def message_len_fn(self) -> Callable[[str], int]: return _LEN_FNS.get(self.descriptor.len_unit, len) + @property + def supports_status_text(self) -> bool: # type: ignore[override] + """Whether the fronted platform renders a TEXT status line. + + Native parity (rich status text): Slack's typing surface is the + assistant status line ("Finding answers…" next to the bot name), a + text-rendering indicator. When the relay fronts Slack, advertise it so + run.py's live-status lane feeds per-tool phrases via + ``set_status_text()`` — exactly the wiring the native SlackAdapter + gets (``supports_status_text = True``). Other fronted platforms keep + textless typing bubbles and must NOT receive phrase traffic. + + Property (not class attr) because ONE RelayAdapter class fronts many + platforms; the answer depends on the handshaked descriptor. On a + multi-platform relay this scalar reflects the PRIMARY identity's + platform (same convention as the scalar ``descriptor``); per-chat + egress paths that need chat-accurate capabilities use + ``_descriptor_for_chat`` below. + """ + return self.descriptor.platform == Platform.SLACK.value + # ── per-chat capability resolution (Phase 1.5 multi-platform) ───────── def _descriptor_for_chat(self, chat_id: str) -> CapabilityDescriptor: """The capability descriptor governing a specific chat. @@ -279,6 +314,7 @@ class RelayAdapter(BasePlatformAdapter): async def _on_inbound(self, event) -> None: """Bridge a connector-delivered MessageEvent into the normal adapter path.""" self._capture_scope(event) + self._stamp_slack_session_thread(event) # Phase 3: a structured prompt answer resolves its waiting primitive # (approval/confirm/clarify) and is CONSUMED — it must not also # dispatch as a chat message. Unknown/expired prompt ids fall through @@ -288,6 +324,112 @@ class RelayAdapter(BasePlatformAdapter): await self._localize_inbound_media(event) await self.handle_message(event) + def _relay_slack_extra(self) -> Dict[str, Any]: + """The Slack-behavior subset of the RELAY platform config. + + Enterprise knob shape (Hermes-config directed, relay-namespaced): + + platforms: + relay: + extra: + slack: # supported subset of native Slack fields + reply_in_thread: true + + The native ``platforms.slack`` block keeps meaning "native adapter + settings"; relay-fronted Slack reads its subset here. Legacy fallback: + a flat key on the relay extra (``extra.reply_in_thread``) still wins + when no ``slack`` object exists, preserving current staging configs. + """ + extra = getattr(self.config, "extra", None) or {} + sub = extra.get("slack") + return sub if isinstance(sub, dict) else extra + + @staticmethod + def _coerce_flag(raw: Any, default: bool) -> bool: + """Coerce an operator-supplied boolean exactly as native Slack does. + + Native SlackAdapter reads its behavior flags with + ``str(raw).strip().lower() in {"1","true","yes","on"}``, so a + YAML-quoted ``"false"`` — a shape operators write routinely — turns + the flag OFF. A bare ``bool()`` would read that same string as True + (non-empty string), silently ignoring the off switch. These knobs are + documented as native-parity mirrors, so they must coerce identically + or the parity claim only holds for unquoted YAML booleans. + """ + if raw is None: + return default + if isinstance(raw, bool): + return raw + return str(raw).strip().lower() in {"1", "true", "yes", "on"} + + def _effective_reply_in_thread(self) -> bool: + """Resolve the thread-per-message vs flat-DM mode for fronted Slack.""" + try: + return self._coerce_flag( + self._relay_slack_extra().get("reply_in_thread"), True + ) + 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 self._coerce_flag( + 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. + + Native SlackAdapter's inbound handler stamps ``thread_ts = + event.thread_ts or ts`` — every TOP-LEVEL message carries its own ts + as ``source.thread_id``, so build_session_key appends it and each + top-level message gets a FRESH session (per-message threads ⇒ + per-message sessions; a 2nd message runs parallel instead of steering + the in-flight turn). The connector normalizes a top-level message + with thread_id=null, so without this stamp every top-level DM + collapses into ONE session key and message 2 pre-empts message 1 + ("Redirected current run", 2026-07-27 report). + + Only in thread-per-message mode: flat mode keeps the shared rolling + DM session on purpose (steer/queue there is the intended UX). Never + overwrites a real thread_id (an in-thread reply must keep resolving + to its thread's session). + """ + try: + src = getattr(event, "source", None) + if not src: + return + platform = getattr(src, "platform", None) + if getattr(platform, "value", platform) != Platform.SLACK.value: + return + if getattr(src, "thread_id", None): + return # real thread — its session key is already correct + message_id = getattr(event, "message_id", None) or getattr( + src, "message_id", None + ) + if not message_id: + 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) + async def _localize_inbound_media(self, event) -> None: """Download connector re-hosted attachments to local temp paths. @@ -380,10 +522,32 @@ class RelayAdapter(BasePlatformAdapter): scope = getattr(src, "scope_id", None) if scope: self._scope_by_chat[str(chat)] = str(scope) + # Remember the chat_type so send() can suppress the synthetic-DM + # thread anchor on Slack (native _resolve_thread_ts parity). send() + # only receives a chat_id, so it needs this per-chat cache to know a + # chat is a DM. + chat_type = getattr(src, "chat_type", None) + if chat_type: + self._chat_type_by_chat[str(chat)] = str(chat_type) + # Triggering message ts: 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 - def _with_scope(self, chat_id: str, metadata: Optional[Dict[str, Any]]) -> Dict[str, Any]: + def _with_scope( + self, chat_id: str, metadata: Optional[Dict[str, Any]] + ) -> Dict[str, Any]: """Ensure the outbound metadata carries the discriminator(s) the connector's egress guard needs to resolve the owning tenant. @@ -542,16 +706,26 @@ class RelayAdapter(BasePlatformAdapter): else: text = "" member = payload.get("member") or {} - user = (member.get("user") if isinstance(member, dict) else None) or payload.get("user") or {} + user = ( + (member.get("user") if isinstance(member, dict) else None) + or payload.get("user") + or {} + ) channel_id = str(payload.get("channel_id") or "") guild_id = payload.get("guild_id") # real Discord interaction wire field source = SessionSource( platform=Platform.RELAY, chat_id=channel_id, chat_type="channel" if guild_id else "dm", - user_id=str(user.get("id")) if isinstance(user, dict) and user.get("id") else None, - user_name=str(user.get("username")) if isinstance(user, dict) and user.get("username") else None, - scope_id=str(guild_id) if guild_id else None, # Discord guild → generic scope slot + user_id=str(user.get("id")) + if isinstance(user, dict) and user.get("id") + else None, + user_name=str(user.get("username")) + if isinstance(user, dict) and user.get("username") + else None, + scope_id=str(guild_id) + if guild_id + else None, # Discord guild → generic scope slot message_id=str(payload.get("id")) if payload.get("id") else None, ) event = MessageEvent(text=text, message_type=message_type, source=source) @@ -567,7 +741,9 @@ class RelayAdapter(BasePlatformAdapter): prompt_id, option_id = decoded msg = payload.get("message") or {} prompt_message_id = ( - str(msg.get("id")) if isinstance(msg, dict) and msg.get("id") else None + str(msg.get("id")) + if isinstance(msg, dict) and msg.get("id") + else None ) event.prompt_response = { "prompt_id": prompt_id, @@ -620,7 +796,9 @@ class RelayAdapter(BasePlatformAdapter): sub_name = str(opt.get("name") or "").strip() if sub_name: parts.append(sub_name) - parts.extend(RelayAdapter._render_interaction_options(opt.get("options"))) + parts.extend( + RelayAdapter._render_interaction_options(opt.get("options")) + ) else: value = opt.get("value") if value is not None and str(value).strip(): @@ -744,12 +922,19 @@ 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. 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 + ) result = await self._transport.send_outbound( { "op": "send", "chat_id": chat_id, "content": content, - "reply_to": reply_to, + "reply_to": effective_reply_to, "metadata": self._with_scope(chat_id, send_metadata), }, platform=self._platform_by_chat.get(str(chat_id)), @@ -760,6 +945,144 @@ class RelayAdapter(BasePlatformAdapter): error=result.get("error"), ) + def _resolve_reply_to_for_send( + self, + chat_id: str, + reply_to: Optional[str], + metadata: Optional[Dict[str, Any]], + ) -> Optional[str]: + """Suppress the synthetic-DM thread anchor for a Slack DM reply. + + A DM turn's streaming reply is sent with ``reply_to`` = the triggering + message's ts (the stream consumer's ``initial_reply_to_id``, used as the + edit anchor and, on threading platforms, the reply target). The + connector's slackRestSender maps a raw ``reply_to`` to a Slack + ``thread_ts``, so a plain DM reply would be posted THREADED under the + user's message instead of flat at the DM root — and a threaded first + send loses the progressive edit-streaming the user sees in a real + thread (the reported symptom: DM/home replies arrive flat, no + progressive edits). + + Native Slack Hermes already suppresses this synthetic DM thread anchor: + ``SlackAdapter._resolve_thread_ts`` returns ``None`` for a top-level / + DM message when ``reply_in_thread`` is off. The relay lane has no such + disambiguation, so we reproduce it here. run.py already encodes the + real-thread decision in ``metadata["thread_id"]`` (it is set only when + progress threading is active — a real thread, or channel autoThread); + for a DM with no real thread that key is absent. So the rule is: + + Slack DM + no real ``thread_id`` in metadata ⇒ drop ``reply_to``. + + This posts the reply flat at the DM root and lets the consumer edit its + own first-send ts — streaming works exactly as in a thread. It does NOT: + * reintroduce a synthetic DM thread_id (#18859 / the /sethome + landmine) — it removes an anchor, never adds one; + * regress real-thread streaming — a real thread carries a distinct + ``thread_id`` in metadata, so the guard leaves ``reply_to`` alone; + * regress channel autoThread — a channel/group top-level reply carries + ``thread_id`` (the message's own ts) in metadata when threading is + on, so it is left alone; and a non-DM chat is never matched here. + """ + if reply_to is None: + return None + if self._platform_by_chat.get(str(chat_id)) != Platform.SLACK.value: + return reply_to + if self._chat_type_by_chat.get(str(chat_id)) != "dm": + return reply_to + md = metadata or {} + if md.get("thread_id") or md.get("thread_ts"): + # A real thread was resolved by run.py — honour it. + return reply_to + # 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, same class as the prompt-placement bug). Native SlackAdapter only + # suppresses the anchor when reply_in_thread=false; mirror that. + reply_in_thread = self._effective_reply_in_thread() + 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 + + 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, @@ -817,13 +1140,39 @@ class RelayAdapter(BasePlatformAdapter): """ if self._transport is None: return + # Thread anchor for the status surface. 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 = 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 + # sender renders it on assistant.threads.setStatus — the same phrase + # the native adapter shows ("is running pytest…", "Finding answers…"). + # Absent (None/empty) => omit content; the connector falls back to its + # default "is typing…" heartbeat, preserving pre-phrase behaviour on + # every platform. Never send empty-string content here: on Slack that + # is the explicit CLEAR request reserved for stop_typing. + frame: Dict[str, Any] = { + "op": "typing", + "chat_id": chat_id, + "metadata": self._with_scope(chat_id, md), + } + phrase = getattr(self, "_status_text", {}).get(str(chat_id)) + if phrase: + frame["content"] = str(phrase) try: await self._transport.send_outbound( - { - "op": "typing", - "chat_id": chat_id, - "metadata": self._with_scope(chat_id, metadata), - }, + frame, platform=self._platform_by_chat.get(str(chat_id)), ) except Exception: # noqa: BLE001 - typing is cosmetic, never breaks a turn @@ -852,13 +1201,17 @@ 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, 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( { "op": "typing", "chat_id": chat_id, "content": "", - "metadata": self._with_scope(chat_id, metadata), + "metadata": self._with_scope(chat_id, md), }, platform=platform, ) @@ -983,14 +1336,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 @@ -1065,8 +1427,12 @@ class RelayAdapter(BasePlatformAdapter): if result is not None: return result return await super().send_image_file( - chat_id, image_path, caption=caption, reply_to=reply_to, - metadata=metadata, **kwargs, + chat_id, + image_path, + caption=caption, + reply_to=reply_to, + metadata=metadata, + **kwargs, ) async def send_voice( @@ -1091,8 +1457,12 @@ class RelayAdapter(BasePlatformAdapter): if result is not None: return result return await super().send_voice( - chat_id, audio_path, caption=caption, reply_to=reply_to, - metadata=metadata, **kwargs, + chat_id, + audio_path, + caption=caption, + reply_to=reply_to, + metadata=metadata, + **kwargs, ) async def send_video( @@ -1117,8 +1487,12 @@ class RelayAdapter(BasePlatformAdapter): if result is not None: return result return await super().send_video( - chat_id, video_path, caption=caption, reply_to=reply_to, - metadata=metadata, **kwargs, + chat_id, + video_path, + caption=caption, + reply_to=reply_to, + metadata=metadata, + **kwargs, ) async def send_document( @@ -1145,13 +1519,20 @@ class RelayAdapter(BasePlatformAdapter): if result is not None: return result return await super().send_document( - chat_id, file_path, caption=caption, file_name=file_name, - reply_to=reply_to, metadata=metadata, **kwargs, + chat_id, + file_path, + caption=caption, + file_name=file_name, + reply_to=reply_to, + metadata=metadata, + **kwargs, ) # ── Phase 3 interactive: prompt + react ────────────────────────────── - def _mint_prompt(self, kind: str, state: Dict[str, Any], timeout_s: float = 3600.0) -> str: + def _mint_prompt( + self, kind: str, state: Dict[str, Any], timeout_s: float = 3600.0 + ) -> str: """Register a pending prompt and return its 8-hex id. ``state`` carries what the resolver needs when the answer comes back @@ -1171,7 +1552,9 @@ class RelayAdapter(BasePlatformAdapter): # Opportunistic sweep so abandoned prompts can't accumulate: drop # anything already expired (cheap — dict is small by construction). now = time.time() - for stale in [k for k, v in self._pending_prompts.items() if v.get("expires_at", 0) < now]: + for stale in [ + k for k, v in self._pending_prompts.items() if v.get("expires_at", 0) < now + ]: self._pending_prompts.pop(stale, None) return prompt_id @@ -1207,6 +1590,20 @@ class RelayAdapter(BasePlatformAdapter): """ if self._transport is None or not self.descriptor.supports_op("prompt"): return None + # An interactive prompt (approval / clarify / slash-confirm) is emitted + # mid-turn in reply to the triggering inbound event, so `metadata` carries + # that event's thread context (run.py _thread_metadata_for_source stamps + # metadata.thread_id — for a Slack DM the triggering message's own ts, + # used only as a session-keying fallback). Forwarding it makes the + # connector thread the prompt card UNDER the triggering message instead + # 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 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, @@ -1214,8 +1611,10 @@ class RelayAdapter(BasePlatformAdapter): "prompt_kind": prompt_kind, "prompt_id": prompt_id, "options": options, - "reply_to": reply_to, - "metadata": self._with_scope(chat_id, metadata), + "reply_to": self._resolve_reply_to_for_send( + chat_id, reply_to, prompt_metadata + ), + "metadata": self._with_scope(chat_id, prompt_metadata), } if timeout_s is not None: action["timeout_s"] = int(timeout_s) @@ -1260,12 +1659,15 @@ class RelayAdapter(BasePlatformAdapter): button→text fallback takes over (same contract as a native adapter's failed button send). """ - options: list = [{"id": "once", "label": "✅ Allow Once", "style": "success"}] + options: list = [{"id": "once", "label": "Allow Once", "style": "primary"}] if not smart_denied and allow_session: - options.append({"id": "session", "label": "✅ Session", "style": "primary"}) + options.append({"id": "session", "label": "Allow Session"}) if allow_permanent: - options.append({"id": "always", "label": "✅ Always", "style": "primary"}) - options.append({"id": "deny", "label": "❌ Deny", "style": "danger"}) + options.append({ + "id": "always", + "label": "Always Allow", + }) + options.append({"id": "deny", "label": "Deny", "style": "danger"}) cmd_preview = command if len(command) <= 1500 else command[:1500] + "..." text = ( @@ -1274,7 +1676,9 @@ class RelayAdapter(BasePlatformAdapter): f"Reason: {description}" ) if smart_denied: - text += "\n\n**Smart DENY:** owner override applies to this one operation only." + text += ( + "\n\n**Smart DENY:** owner override applies to this one operation only." + ) prompt_id = self._mint_prompt( "exec_approval", @@ -1310,9 +1714,9 @@ class RelayAdapter(BasePlatformAdapter): gateway's text-intercept flow when the prompt lane is unavailable. """ options = [ - {"id": "once", "label": "✅ Approve Once", "style": "success"}, - {"id": "always", "label": "🔒 Always Approve", "style": "primary"}, - {"id": "cancel", "label": "❌ Cancel", "style": "danger"}, + {"id": "once", "label": "Approve Once", "style": "primary"}, + {"id": "always", "label": "Always Approve"}, + {"id": "cancel", "label": "Cancel", "style": "danger"}, ] text = f"**{title}**\n\n{message}" if title else message prompt_id = self._mint_prompt( @@ -1422,7 +1826,11 @@ class RelayAdapter(BasePlatformAdapter): if kind == "exec_approval": from tools.approval import resolve_gateway_approval - choice = option_id if option_id in {"once", "session", "always", "deny"} else "deny" + choice = ( + option_id + if option_id in {"once", "session", "always", "deny"} + else "deny" + ) count = resolve_gateway_approval(session_key, choice) label = { "once": "✅ Approved once", @@ -1435,13 +1843,17 @@ class RelayAdapter(BasePlatformAdapter): # Acknowledge in-channel (the connector's prompt message can't # be edited cross-platform yet — edit support varies; a short # confirmation preserves the audit trail the native edit gives). - await self.send(chat_id, label, metadata=self._prompt_reply_metadata(event)) + await self.send( + chat_id, label, metadata=self._prompt_reply_metadata(event) + ) if count: self.resume_typing_for_chat(chat_id) elif kind == "slash_confirm": from tools import slash_confirm as slash_confirm_mod - choice = option_id if option_id in {"once", "always", "cancel"} else "cancel" + choice = ( + option_id if option_id in {"once", "always", "cancel"} else "cancel" + ) result_text = await slash_confirm_mod.resolve( session_key, str(state.get("confirm_id") or ""), choice ) @@ -1450,13 +1862,20 @@ class RelayAdapter(BasePlatformAdapter): "always": "🔒 Always approve", "cancel": "❌ Cancelled", }.get(choice, "Resolved") - await self.send(chat_id, label, metadata=self._prompt_reply_metadata(event)) + await self.send( + chat_id, label, metadata=self._prompt_reply_metadata(event) + ) if result_text: await self.send( - chat_id, str(result_text), metadata=self._prompt_reply_metadata(event) + chat_id, + str(result_text), + metadata=self._prompt_reply_metadata(event), ) elif kind == "clarify": - from tools.clarify_gateway import mark_awaiting_text, resolve_gateway_clarify + from tools.clarify_gateway import ( + mark_awaiting_text, + resolve_gateway_clarify, + ) clarify_id = str(state.get("clarify_id") or "") if option_id == "other": diff --git a/gateway/run.py b/gateway/run.py index c93a7de1ea4..9c39dcb7bc1 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -21924,11 +21924,22 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew _slack_adapter_for_progress = self._adapter_for_source(source) if _slack_adapter_for_progress is not None: try: - _progress_reply_in_thread = bool( - _slack_adapter_for_progress.config.extra.get( - "reply_in_thread", True - ) + # Relay lane: the adapter owns mode resolution (nested + # platforms.relay.extra.slack subset with flat-key + # fallback). Native lane: read the flat extra as before. + _mode_fn = getattr( + _slack_adapter_for_progress, + "_effective_reply_in_thread", + None, ) + if callable(_mode_fn): + _progress_reply_in_thread = bool(_mode_fn()) + else: + _progress_reply_in_thread = bool( + _slack_adapter_for_progress.config.extra.get( + "reply_in_thread", True + ) + ) except Exception: _progress_reply_in_thread = True _progress_thread_id = _resolve_progress_thread_id( diff --git a/tests/gateway/relay/test_relay_slack_dm_streaming.py b/tests/gateway/relay/test_relay_slack_dm_streaming.py new file mode 100644 index 00000000000..9969b739575 --- /dev/null +++ b/tests/gateway/relay/test_relay_slack_dm_streaming.py @@ -0,0 +1,403 @@ +"""Slack relay: edit-based streaming of the reply must fire in a DM. + +Reported symptom (live): agent responses stream progressively (edit-based) in a +Slack THREAD but arrive FLAT (single message, no progressive edits) in a Slack +DM/home over the relay. + +Root cause: a DM turn's streaming reply is sent with +``reply_to = `` (the stream consumer's +``initial_reply_to_id`` — its edit anchor). The connector's slackRestSender maps +a raw ``reply_to`` to a Slack ``thread_ts``, so a plain DM reply gets threaded +UNDER the user's message instead of posting flat at the DM root, and a threaded +first send loses the progressive edit streaming the user sees in a real thread. +Native ``SlackAdapter._resolve_thread_ts`` already suppresses this synthetic DM +thread anchor; the relay lane had no such disambiguation. + +These are behaviour-contract tests: they assert how the outbound frame relates to +the chat type + thread metadata (the invariant the connector depends on), not a +snapshot. They drive the REAL ``RelayAdapter`` + ``GatewayStreamConsumer`` + +``StubConnector`` end to end. +""" + +from __future__ import annotations + +import pytest + +from gateway.config import Platform, PlatformConfig +from gateway.platforms.base import MessageEvent, MessageType +from gateway.relay.adapter import RelayAdapter +from gateway.relay.descriptor import CONTRACT_VERSION, CapabilityDescriptor +from gateway.session import SessionSource +from gateway.stream_consumer import GatewayStreamConsumer, StreamConsumerConfig + +from tests.gateway.relay.stub_connector import StubConnector + + +def _slack_desc(**kw) -> CapabilityDescriptor: + base = dict( + contract_version=CONTRACT_VERSION, + platform="slack", + label="Slack", + max_message_length=4000, + supports_draft_streaming=False, + supports_edit=True, + supports_threads=True, + markdown_dialect="mrkdwn", + len_unit="chars", + emoji="\U0001f4ac", + platform_hint="", + pii_safe=False, + ) + base.update(kw) + return CapabilityDescriptor(**base) + + +def _wire(chat_id: str, chat_type: str, *, user_id="U1", scope_id=None): + """A RelayAdapter fronting Slack, with inbound scope captured for chat_id.""" + stub = StubConnector(_slack_desc()) + adapter = RelayAdapter(PlatformConfig(), _slack_desc(), transport=stub) + src = SessionSource( + platform=Platform.SLACK, + chat_id=chat_id, + chat_type=chat_type, + user_id=user_id, + scope_id=scope_id, + ) + adapter._capture_scope( + MessageEvent(text="hi", source=src, message_type=MessageType.TEXT) + ) + return adapter, stub + + +# --------------------------------------------------------------------------- +# The pure disambiguation contract (RelayAdapter.send) +# --------------------------------------------------------------------------- +@pytest.mark.asyncio +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" + assert frame["reply_to"] == "1700.0001", ( + "thread-per-message: the triggering ts anchors the final reply" + ) + # 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" + + +@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 {}) + assert "thread_ts" not in (frame["metadata"] or {}) + + +@pytest.mark.asyncio +async def test_slack_dm_reply_with_real_thread_keeps_anchor(): + """A DM turn that IS inside a real thread (metadata carries a distinct + thread_id) must keep threading — the guard only drops the synthetic anchor.""" + adapter, stub = _wire("D1", "dm") + await adapter.send( + "D1", "in thread", reply_to="1700.0002", metadata={"thread_id": "1699.9000"} + ) + frame = stub.sent[0] + assert frame["reply_to"] == "1700.0002" + assert frame["metadata"]["thread_id"] == "1699.9000" + + +@pytest.mark.asyncio +async def test_slack_channel_top_level_reply_keeps_autothread_anchor(): + """A channel top-level reply carries thread_id (its own ts) in metadata when + autoThread is on; the DM-only guard must not touch it.""" + adapter, stub = _wire("C1", "channel", scope_id="T1") + await adapter.send( + "C1", "channel reply", reply_to="1700.0003", metadata={"thread_id": "1700.0003"} + ) + frame = stub.sent[0] + assert frame["reply_to"] == "1700.0003" + assert frame["metadata"]["thread_id"] == "1700.0003" + + +@pytest.mark.asyncio +async def test_non_slack_dm_reply_unchanged(): + """The disambiguation is Slack-scoped: a non-Slack relay chat keeps reply_to + (its connector owns its own threading semantics).""" + stub = StubConnector(_slack_desc(platform="discord")) + adapter = RelayAdapter( + PlatformConfig(), _slack_desc(platform="discord"), transport=stub + ) + src = SessionSource( + platform=Platform.DISCORD, chat_id="dc1", chat_type="dm", user_id="U1" + ) + adapter._capture_scope( + MessageEvent(text="hi", source=src, message_type=MessageType.TEXT) + ) + await adapter.send("dc1", "hi", reply_to="msg-9") + assert stub.sent[0]["reply_to"] == "msg-9" + + +# --------------------------------------------------------------------------- +# End-to-end: the stream consumer keeps edit-streaming in a DM +# --------------------------------------------------------------------------- +async def _drive_stream(adapter, chat_id, *, metadata, initial_reply_to_id, chat_type): + cfg = StreamConsumerConfig( + edit_interval=0.0, + buffer_threshold=1, + transport="edit", + chat_type=chat_type, + ) + consumer = GatewayStreamConsumer( + adapter=adapter, + chat_id=chat_id, + config=cfg, + metadata=metadata, + initial_reply_to_id=initial_reply_to_id, + ) + # Feed progressive deltas, then finalize — mirrors the live delta callback. + for chunk in ("Hel", "lo ", "world", ". Done."): + consumer.on_delta(chunk) + consumer.finish() + await consumer.run() + return consumer + + +@pytest.mark.asyncio +async def test_slack_dm_stream_consumer_edits_own_ts_not_flat(): + """A Slack DM turn (chat_type='dm', no thread, metadata None) still builds a + stream consumer that keeps edit support and emits progressive EDITs of the + reply message — the flat-DM regression contract. + + 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. + + 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", + metadata=None, # DM: _status_thread_metadata is None in run.py + initial_reply_to_id="1700.0001", # the triggering message ts + chat_type="dm", + ) + + ops = [f["op"] for f in stub.sent] + # First a flat send; edit support stays on so progressive edits CAN flow + # (exact intermediate-frame timing is covered by the stream_consumer unit + # suite — here we assert the DM regression contract: streaming is not + # self-disabled and every edit targets the reply's own ts). + assert ops[0] == "send" + # Edit support survived: message_id set, not the __no_edit__ sentinel. + assert consumer.message_id and consumer.message_id != "__no_edit__" + assert consumer._edit_supported is True + + first_send = stub.sent[0] + # The reply posts FLAT at the DM root — no synthetic thread anchor. + assert first_send["reply_to"] is None + assert "thread_id" not in (first_send["metadata"] or {}) + assert "thread_ts" not in (first_send["metadata"] or {}) + # reply_to_message_id (the mirrored self-anchor) is stripped too. + assert "reply_to_message_id" not in (first_send["metadata"] or {}) + + # Any edits that flowed target the same first-send ts (editing its own + # message), never a synthetic thread. + edit_ids = {f["message_id"] for f in stub.sent if f["op"] == "edit"} + assert edit_ids <= {stub.next_send_result["message_id"]} + + +@pytest.mark.asyncio +async def test_slack_thread_stream_consumer_still_threads_and_streams(): + """Regression guard: a Slack THREAD turn keeps its real thread_id AND streams + (the DM fix must not change the thread path).""" + adapter, stub = _wire("C1", "channel", scope_id="T1") + consumer = await _drive_stream( + adapter, + "C1", + metadata={"thread_id": "1699.9000"}, + initial_reply_to_id="1700.0002", + chat_type="channel", + ) + ops = [f["op"] for f in stub.sent] + assert ops[0] == "send" + assert consumer._edit_supported is True + first_send = stub.sent[0] + # 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"]} + + +# --------------------------------------------------------------------------- +# 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" + + +# --------------------------------------------------------------------------- +# Operator flags coerce exactly as the native Slack adapter's do. +# --------------------------------------------------------------------------- +@pytest.mark.parametrize( + "raw,expected", + [ + (False, False), + ("false", False), + ("False", False), + (" no ", False), + ("off", False), + ("0", False), + (True, True), + ("true", True), + ("yes", True), + ("on", True), + ("1", True), + ], +) +def test_relay_slack_flags_coerce_like_native(raw, expected): + """A YAML-quoted "false" must turn these knobs OFF, matching native's + str().strip().lower() predicate. A bare bool() would read any non-empty + string as True and silently ignore the operator's off switch.""" + adapter, _stub = _wire("D1", "dm") + adapter.config.extra = { + "slack": { + "reply_in_thread": raw, + "dm_top_level_threads_as_sessions": raw, + } + } + assert adapter._effective_reply_in_thread() is expected + assert adapter._dm_top_level_threads_as_sessions() is expected + + +def test_relay_slack_flags_default_true_when_absent(): + """Both knobs default ON when the operator sets nothing.""" + adapter, _stub = _wire("D1", "dm") + adapter.config.extra = {} + assert adapter._effective_reply_in_thread() is True + assert adapter._dm_top_level_threads_as_sessions() is True + + +# --------------------------------------------------------------------------- +# 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" + ) diff --git a/tests/gateway/relay/test_relay_slack_prompt_dm_root.py b/tests/gateway/relay/test_relay_slack_prompt_dm_root.py new file mode 100644 index 00000000000..466fd10071c --- /dev/null +++ b/tests/gateway/relay/test_relay_slack_prompt_dm_root.py @@ -0,0 +1,486 @@ +"""Slack relay: interactive prompts follow the turn's thread stamp. + +The threading MODE (flat DM vs thread-per-message) is decided in exactly ONE +place: run.py's ``_resolve_progress_thread_id``, which reads +``platforms.slack.extra.reply_in_thread`` and encodes the verdict into the +outbound ``metadata`` stamp: + + * flat mode -> the synthetic self-anchor is suppressed in run.py, so prompt + metadata arrives with NO ``thread_id`` and the card posts at the DM root; + * thread-per-message (default) -> ``metadata.thread_id`` is stamped for the + whole turn; on the FIRST turn it legitimately equals the triggering + message's ts (the synthetic root IS the thread). + +The prompt lane must TRUST that stamp, like ``_resolve_reply_to_for_send`` +does. Re-deriving the mode here (the old unconditional +``thread_id == message_id`` strip) exiled the approval card and its +resolved-state swap to the DM root while progress bubbles honoured the thread +(the 2026-07-27 mixed-placement report). + +These are behaviour-contract tests: they assert how the outbound ``prompt`` +frame relates to the inherited thread metadata (the invariant the connector +depends on), not a snapshot. They drive the REAL ``RelayAdapter`` + +``StubConnector`` end to end. +""" + +from __future__ import annotations + +import pytest + +from gateway.config import Platform, PlatformConfig +from gateway.platforms.base import MessageEvent, MessageType +from gateway.relay.adapter import RelayAdapter +from gateway.relay.descriptor import CONTRACT_VERSION, CapabilityDescriptor +from gateway.session import SessionSource + +from tests.gateway.relay.stub_connector import StubConnector + +FULL_OPS = ("send", "edit", "typing", "get_chat_info", "send_media", "prompt", "react") + + +def _slack_desc(**kw) -> CapabilityDescriptor: + base = dict( + contract_version=CONTRACT_VERSION, + platform="slack", + label="Slack", + max_message_length=4000, + supports_draft_streaming=False, + supports_edit=True, + supports_threads=True, + markdown_dialect="mrkdwn", + len_unit="chars", + supported_ops=FULL_OPS, + ) + base.update(kw) + return CapabilityDescriptor(**base) + + +def _wire( + chat_id: str, + chat_type: str, + *, + user_id="U1", + scope_id=None, + platform=Platform.SLACK, +): + """A RelayAdapter fronting Slack, with inbound scope + chat_type captured.""" + stub = StubConnector(_slack_desc()) + adapter = RelayAdapter(PlatformConfig(), _slack_desc(), transport=stub) + src = SessionSource( + platform=platform, + chat_id=chat_id, + chat_type=chat_type, + user_id=user_id, + scope_id=scope_id, + ) + adapter._capture_scope( + MessageEvent(text="hi", source=src, message_type=MessageType.TEXT) + ) + return adapter, stub + + +def _last_prompt(stub) -> dict: + prompts = [f for f in stub.sent if f["op"] == "prompt"] + assert prompts, "expected a prompt op on the wire" + return prompts[-1] + + +# --------------------------------------------------------------------------- +# Flat mode: run.py stamps NO thread_id -> the card posts at the DM root. +# --------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_exec_approval_flat_mode_posts_at_dm_root(): + """Flat-DM turn (reply_in_thread=false): run.py suppressed the synthetic + anchor upstream, so prompt metadata has no thread_id and none appears on + the wire — the card posts at the DM root.""" + adapter, stub = _wire("D1", "dm", scope_id="T1") + md = {"message_id": "1700000000.000100", "scope_id": "T1"} + result = await adapter.send_exec_approval( + "D1", "rm -rf /tmp/x", "sess:1", description="deletes files", metadata=md + ) + assert result.success is True + frame = _last_prompt(stub) + meta = frame["metadata"] or {} + assert "thread_id" not in meta + assert "thread_ts" not in meta + # reply_to on the outbound action stays unset — a root-level post. + assert frame["reply_to"] is None + # Tenant scope is preserved untouched (egress routing must not break). + assert meta.get("scope_id") == "T1" + + +# --------------------------------------------------------------------------- +# 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_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", + "message_id": "1700000000.000100", + "scope_id": "T1", + } + result = await adapter.send_exec_approval( + "D1", "rm -rf /tmp/x", "sess:1", description="deletes files", metadata=md + ) + assert result.success is True + frame = _last_prompt(stub) + meta = frame["metadata"] or {} + assert meta.get("thread_id") == "1700000000.000100", ( + "first-turn self-anchor is the thread root; the prompt must honour it" + ) + assert meta.get("scope_id") == "T1" + + +@pytest.mark.asyncio +async def test_clarify_forwards_run_py_thread_stamp_untouched(): + adapter, stub = _wire("D1", "dm", scope_id="T1") + md = { + "thread_id": "1700000000.000200", + "message_id": "1700000000.000200", + "scope_id": "T1", + } + result = await adapter.send_clarify( + "D1", "Which env?", ["prod", "staging"], "cl-1", "sess:1", metadata=md + ) + assert result.success is True + frame = _last_prompt(stub) + meta = frame["metadata"] or {} + assert meta.get("thread_id") == "1700000000.000200" + assert meta.get("scope_id") == "T1" + + +@pytest.mark.asyncio +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"} + await adapter.send_slash_confirm( + "D1", "Reload MCP", "invalidates cache", "s", "cf-1", metadata=md + ) + frame = _last_prompt(stub) + assert (frame["metadata"] or {}).get("thread_id") == "1700000000.000300" + + +# --------------------------------------------------------------------------- +# Regression guards: a REAL thread and non-DM / non-Slack chats are untouched +# --------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_exec_approval_in_real_thread_keeps_thread_id(): + """A DM prompt raised inside a REAL thread (thread_id distinct from the + triggering message ts) stays in that thread.""" + adapter, stub = _wire("D1", "dm", scope_id="T1") + md = { + "thread_id": "1699000000.999000", + "message_id": "1700000000.000100", + "scope_id": "T1", + } + await adapter.send_exec_approval("D1", "cmd", "s", metadata=md) + frame = _last_prompt(stub) + assert frame["metadata"]["thread_id"] == "1699000000.999000" + + +@pytest.mark.asyncio +async def test_channel_approval_keeps_thread_id(): + """A Slack CHANNEL prompt keeps its thread_id (autoThread / real thread).""" + adapter, stub = _wire("C1", "channel", scope_id="T1") + md = { + "thread_id": "1700000000.000400", + "message_id": "1700000000.000400", + "scope_id": "T1", + } + await adapter.send_exec_approval("C1", "cmd", "s", metadata=md) + frame = _last_prompt(stub) + assert frame["metadata"]["thread_id"] == "1700000000.000400" + + +@pytest.mark.asyncio +async def test_non_slack_dm_approval_keeps_thread_id(): + """A non-Slack relay DM keeps thread_id (its connector owns its own + threading semantics).""" + adapter, stub = _wire("dc1", "dm", platform=Platform.DISCORD) + md = {"thread_id": "9000", "message_id": "9000"} + await adapter.send_exec_approval("dc1", "cmd", "s", metadata=md) + frame = _last_prompt(stub) + assert frame["metadata"]["thread_id"] == "9000" + + +# --------------------------------------------------------------------------- +# Rich status: the relay advertises Slack's text status line and carries +# the live per-tool phrase on the typing frame (native set_status_text parity). +# --------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_slack_relay_advertises_status_text(): + adapter, _stub = _wire("D1", "dm") + assert adapter.supports_status_text is True + + +@pytest.mark.asyncio +async def test_non_slack_relay_does_not_advertise_status_text(): + stub = StubConnector(_slack_desc(platform="discord")) + adapter = RelayAdapter( + PlatformConfig(), _slack_desc(platform="discord"), transport=stub + ) + assert adapter.supports_status_text is False + + +@pytest.mark.asyncio +async def test_typing_carries_live_status_phrase(): + """set_status_text() -> the next typing frame carries the phrase as + content; clearing it (None) reverts to a content-less heartbeat frame + (never an empty string, which is Slack's explicit clear).""" + adapter, stub = _wire("D1", "dm", scope_id="T1") + adapter.set_status_text("D1", "is running pytest…") + await adapter.send_typing("D1", metadata={"scope_id": "T1"}) + typing = [f for f in stub.sent if f["op"] == "typing"] + assert typing and typing[-1].get("content") == "is running pytest…" + + adapter.set_status_text("D1", None) + await adapter.send_typing("D1", metadata={"scope_id": "T1"}) + typing = [f for f in stub.sent if f["op"] == "typing"] + assert "content" not in typing[-1], ( + "cleared phrase must omit content (empty string means CLEAR on Slack)" + ) + + +# --------------------------------------------------------------------------- +# 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_flat_mode_status_anchors_to_trigger_ts_by_default(): + """Flat-DM liveliness: the STATUS still anchors to the triggering ts + (renders in the footer space, no message artifact) while replies stay + flat — the send lane strips its anchors, so placement cannot inherit this.""" + 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 typing[-1]["metadata"].get("thread_id") == "1700.0042" + + +@pytest.mark.asyncio +async def test_typing_anchors_unconditionally_in_both_modes(): + """Liveliness is not a preference: the status anchors whenever an inbound + ts exists, regardless of reply_in_thread. Placement safety comes from the + send-side anchor strip, not from suppressing the status.""" + for extra in ({}, {"slack": {"reply_in_thread": False}}): + adapter, stub = _wire_with_ts("D1", "dm", "1700.0042") + adapter.config.extra = extra + 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_flat_mode_sends_stay_flat_with_status_anchor_active(): + """The liveliness anchor must NOT leak into reply placement: sends in + flat mode still strip the synthetic anchor (send-lane contract).""" + adapter, stub = _wire_with_ts("D1", "dm", "1700.0042") + adapter.config.extra = {"reply_in_thread": False} + await adapter.send_typing("D1", metadata=None) + await adapter.send("D1", "the answer", reply_to="1700.0042") + frame = [f for f in stub.sent if f["op"] == "send"][-1] + assert frame["reply_to"] is None + assert "thread_id" not in (frame["metadata"] or {}) + + +@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" + + +# --------------------------------------------------------------------------- +# Session keying: a top-level Slack DM message gets its own ts stamped as +# source.thread_id (native inbound parity) so each message keys a FRESH +# session in thread-per-message mode; flat mode and real threads untouched. +# --------------------------------------------------------------------------- +def _inbound_event(chat_id="D1", message_id="1700.0100", thread_id=None): + src = SessionSource( + platform=Platform.SLACK, chat_id=chat_id, chat_type="dm", + user_id="U1", scope_id="T1", thread_id=thread_id, + ) + return MessageEvent( + text="hi", source=src, message_type=MessageType.TEXT, + message_id=message_id, + ) + + +def test_top_level_dm_gets_session_thread_stamp(): + adapter, _ = _wire("D1", "dm") + ev = _inbound_event(message_id="1700.0100") + adapter._stamp_slack_session_thread(ev) + assert ev.source.thread_id == "1700.0100" + + +def test_two_top_level_messages_key_distinct_sessions(): + from gateway.session import build_session_key + adapter, _ = _wire("D1", "dm") + e1 = _inbound_event(message_id="1700.0100") + e2 = _inbound_event(message_id="1700.0200") + adapter._stamp_slack_session_thread(e1) + adapter._stamp_slack_session_thread(e2) + k1 = build_session_key(e1.source) + k2 = build_session_key(e2.source) + assert k1 != k2, "each top-level message must be its own session" + + +def test_real_thread_reply_keeps_its_thread_session(): + adapter, _ = _wire("D1", "dm") + ev = _inbound_event(message_id="1700.0300", thread_id="1700.0100") + adapter._stamp_slack_session_thread(ev) + assert ev.source.thread_id == "1700.0100", ( + "an in-thread reply must keep resolving to its thread's session" + ) + + +def test_flat_mode_keeps_shared_dm_session(): + adapter, _ = _wire("D1", "dm") + adapter.config.extra = {"reply_in_thread": False} + ev = _inbound_event(message_id="1700.0400") + adapter._stamp_slack_session_thread(ev) + assert ev.source.thread_id is None, ( + "flat mode: shared rolling DM session (steer/queue) is intended UX" + ) + + +def test_nested_relay_slack_config_subset_wins(): + """Enterprise knob shape: platforms.relay.extra.slack.reply_in_thread.""" + adapter, _ = _wire("D1", "dm") + adapter.config.extra = {"slack": {"reply_in_thread": False}} + assert adapter._effective_reply_in_thread() is False + adapter.config.extra = {"slack": {"reply_in_thread": True}} + assert adapter._effective_reply_in_thread() is True + # Legacy flat key still honoured when no nested object exists. + adapter.config.extra = {"reply_in_thread": False} + assert adapter._effective_reply_in_thread() is False + # 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" + ) + + +# --------------------------------------------------------------------------- +# 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" + )